16 // Channel defines the operations necessary to implement a 9p message channel
17 // interface. Typically, message channels do no protocol processing except to
18 // send and receive message frames.
19 type Channel interface {
20 // ReadFcall reads one fcall frame into the provided fcall structure. The
21 // Fcall may be cleared whether there is an error or not. If the operation
22 // is successful, the contents of the fcall will be populated in the
23 // argument. ReadFcall cannot be called concurrently with other calls to
24 // ReadFcall. This both to preserve message ordering and to allow lockless
26 ReadFcall(ctx context.Context, fcall *Fcall) error
28 // WriteFcall writes the provided fcall to the channel. WriteFcall cannot
29 // be called concurrently with other calls to WriteFcall.
30 WriteFcall(ctx context.Context, fcall *Fcall) error
32 // MSize returns the current msize for the channel.
35 // SetMSize sets the maximum message size for the channel. This must never
36 // be called currently with ReadFcall or WriteFcall.
40 // NewChannel returns a new channel to read and write Fcalls with the provided
41 // connection and message size.
42 func NewChannel(conn net.Conn, msize int) Channel {
43 return newChannel(conn, codec9p{}, msize)
47 defaultRWTimeout = 30 * time.Second // default read/write timeout if not set in context
50 // channel provides bidirectional protocol framing for 9p over net.Conn.
51 // Operations are not thread-safe but reads and writes may be carried out
52 // concurrently, supporting separate read and write loops.
56 // A connection, or message channel abstraction, has a lifecycle delineated by
57 // Tversion/Rversion request response cycles. For now, this is part of the
58 // channel itself but doesn't necessarily influence the channels state, except
59 // the msize. Visually, it might look something like this:
61 // [Established] -> [Version] -> [Session] -> [Version]---+
63 // |_________________________________|
65 // The connection is established, then we negotiate a version, run a session,
66 // then negotiate a version and so on. For most purposes, we are likely going
67 // to terminate the connection after the session but we may want to support
68 // connection pooling. Pooling may result in possible security leaks if the
69 // connections are shared among contexts, since the version is negotiated at
70 // the start of the session. To avoid this, we can actually use a "tombstone"
71 // version message which clears the server's session state without starting a
72 // new session. The next version message would then prepare the session
73 // without leaking any Fid's.
84 func newChannel(conn net.Conn, codec Codec, msize int) *channel {
88 brd: bufio.NewReaderSize(conn, msize), // msize may not be optimal buffer size
89 bwr: bufio.NewWriterSize(conn, msize),
90 closed: make(chan struct{}),
92 rdbuf: make([]byte, msize),
96 func (ch *channel) MSize() int {
100 // setmsize resizes the buffers for use with a separate msize. This call must
101 // be protected by a mutex or made before passing to other goroutines.
102 func (ch *channel) SetMSize(msize int) {
103 // NOTE(stevvooe): We cannot safely resize the buffered reader and writer.
104 // Proceed assuming that original size is sufficient.
107 if msize < len(ch.rdbuf) {
108 // just change the cap
109 ch.rdbuf = ch.rdbuf[:msize]
113 ch.rdbuf = make([]byte, msize)
116 // ReadFcall reads the next message from the channel into fcall.
117 func (ch *channel) ReadFcall(ctx context.Context, fcall *Fcall) error {
126 deadline, ok := ctx.Deadline()
128 deadline = time.Now().Add(defaultRWTimeout)
131 if err := ch.conn.SetReadDeadline(deadline); err != nil {
132 log.Printf("transport: error setting read deadline on %v: %v", ch.conn.RemoteAddr(), err)
135 n, err := readmsg(ch.brd, ch.rdbuf)
137 // TODO(stevvooe): There may be more we can do here to detect partial
138 // reads. For now, we just propagate the error untouched.
142 if n > len(ch.rdbuf) {
143 // TODO(stevvooe): Make this error detectable and respond with error
145 return fmt.Errorf("message too large for buffer: %v > %v ", n, len(ch.rdbuf))
148 // clear out the fcall
150 if err := ch.codec.Unmarshal(ch.rdbuf[:n], fcall); err != nil {
157 func (ch *channel) WriteFcall(ctx context.Context, fcall *Fcall) error {
166 deadline, ok := ctx.Deadline()
168 deadline = time.Now().Add(defaultRWTimeout)
171 if err := ch.conn.SetWriteDeadline(deadline); err != nil {
172 log.Printf("transport: error setting read deadline on %v: %v", ch.conn.RemoteAddr(), err)
175 p, err := ch.codec.Marshal(fcall)
180 if err := sendmsg(ch.bwr, p); err != nil {
184 return ch.bwr.Flush()
187 // readmsg reads a 9p message into p from rd, ensuring that all bytes are
188 // consumed from the size header. If the size header indicates the message is
189 // larger than p, the entire message will be discarded, leaving a truncated
190 // portion in p. Any error should be treated as a framing error unless n is
191 // zero. The caller must check that n is less than or equal to len(p) to
192 // ensure that a valid message has been read.
193 func readmsg(rd io.Reader, p []byte) (n int, err error) {
196 if err := binary.Read(rd, binary.LittleEndian, &msize); err != nil {
200 n += binary.Size(msize)
201 mbody := int(msize) - 4
207 np, err := io.ReadFull(rd, p)
214 // message has been read up to len(p) but we must consume the entire
215 // message. This is an error condition but is non-fatal if we can
216 // consume msize bytes.
217 nn, err := io.CopyN(ioutil.Discard, rd, int64(mbody-len(p)))
227 // sendmsg writes a message of len(p) to wr with a 9p size header. All errors
228 // should be considered terminal.
229 func sendmsg(wr io.Writer, p []byte) error {
230 size := uint32(len(p) + 4) // message size plus 4-bytes for size.
231 if err := binary.Write(wr, binary.LittleEndian, size); err != nil {
235 // This assume partial writes to wr aren't possible. Not sure if this
236 // valid. Matters during timeout retries.
237 if n, err := wr.Write(p); err != nil {
239 } else if n < len(p) {
240 return io.ErrShortWrite