13 // TODO(stevvooe): Add net/http.Server-like type here to manage connections.
14 // Coupled with Handler mux, we can get a very http-like experience for 9p
17 // ServeConn the 9p handler over the provided network connection.
18 func ServeConn(ctx context.Context, cn net.Conn, handler Handler) error {
20 // TODO(stevvooe): It would be nice if the handler could declare the
21 // supported version. Before we had handler, we used the session to get
22 // the version (msize, version := session.Version()). We must decided if
23 // we want to proxy version and message size decisions all the back to the
24 // origin server or make those decisions at each link of a proxy chain.
26 ch := newChannel(cn, codec9p{}, DefaultMSize)
27 negctx, cancel := context.WithTimeout(ctx, 1*time.Second)
30 // TODO(stevvooe): For now, we negotiate here. It probably makes sense to
31 // do this outside of this function and then pass in a ready made channel.
32 // We are not really ready to export the channel type yet.
34 if err := servernegotiate(negctx, ch, DefaultVersion); err != nil {
35 // TODO(stevvooe): Need better error handling and retry support here.
36 return fmt.Errorf("error negotiating version: %s", err)
39 ctx = withVersion(ctx, DefaultVersion)
45 closed: make(chan struct{}),
51 // conn plays role of session dispatch for handler in a server.
60 err error // terminal error for the conn
63 // activeRequest includes information about the active request.
64 type activeRequest struct {
67 cancel context.CancelFunc
70 // serve messages on the connection until an error is encountered.
71 func (c *conn) serve() error {
72 tags := map[Tag]*activeRequest{} // active requests
74 requests := make(chan *Fcall) // sync, read-limited
75 responses := make(chan *Fcall) // sync, goroutine consumed
76 completed := make(chan *Fcall) // sync, send in goroutine per request
84 case req := <-requests:
85 if _, ok := tags[req.Tag]; ok {
87 case responses <- newErrorFcall(req.Tag, ErrDuptag):
88 // Send to responses, bypass tag management.
97 switch msg := req.Message.(type) {
100 // check if we have actually know about the requested flush
101 active, ok := tags[msg.Oldtag]
103 active.cancel() // propagate cancellation to callees
104 delete(tags, msg.Oldtag)
105 resp = newFcall(req.Tag, MessageRflush{})
107 resp = newErrorFcall(req.Tag, ErrUnknownTag)
111 case responses <- resp:
112 // bypass tag management in completed.
119 // Allows us to session handlers to cancel processing of the fcall
121 ctx, cancel := context.WithCancel(c.ctx)
123 // The contents of these instances are only writable in the main
124 // server loop. The value of tag will not change.
125 tags[req.Tag] = &activeRequest{
131 go func(ctx context.Context, req *Fcall) {
132 // TODO(stevvooe): Re-write incoming Treads so that handler
133 // can always respond with a message of the correct msize.
136 msg, err := c.handler.Handle(ctx, req.Message)
138 // all handler errors are forwarded as protocol errors.
139 resp = newErrorFcall(req.Tag, err)
141 resp = newFcall(req.Tag, msg)
145 case completed <- resp:
153 case resp := <-completed:
154 // only responses that flip the tag state traverse this section.
155 active, ok := tags[resp.Tag]
157 // The tag is no longer active. Likely a flushed message.
162 case responses <- resp:
163 case <-active.ctx.Done():
164 // the context was canceled for some reason, perhaps timeout or
165 // due to a flush call. We treat this as a condition where a
166 // response should not be sent.
168 delete(tags, resp.Tag)
177 // read takes requests off the channel and sends them on requests.
178 func (c *conn) read(requests chan *Fcall) {
181 if err := c.ch.ReadFcall(c.ctx, req); err != nil {
182 if err, ok := err.(net.Error); ok {
183 if err.Timeout() || err.Temporary() {
184 // TODO(stevvooe): A full idle timeout on the connection
185 // should be enforced here. No logging because it is quite
191 c.CloseWithError(fmt.Errorf("error reading fcall: %v", err))
196 case requests <- req:
198 c.CloseWithError(c.ctx.Err())
206 func (c *conn) write(responses chan *Fcall) {
209 case resp := <-responses:
210 // TODO(stevvooe): Correctly protect againt overflowing msize from
211 // handler. This can be done above, in the main message handler
212 // loop, by adjusting incoming Tread calls to have a Count that
213 // won't overflow the msize.
215 if err := c.ch.WriteFcall(c.ctx, resp); err != nil {
216 if err, ok := err.(net.Error); ok {
217 if err.Timeout() || err.Temporary() {
218 // TODO(stevvooe): A full idle timeout on the
219 // connection should be enforced here. We log here,
220 // since this is less common.
221 log.Printf("p9p: temporary error writing fcall: %v", err)
226 c.CloseWithError(fmt.Errorf("error writing fcall: %v", err))
230 c.CloseWithError(c.ctx.Err())
238 func (c *conn) Close() error {
239 return c.CloseWithError(nil)
242 func (c *conn) CloseWithError(err error) error {