Blob


1 package p9p
3 import (
4 "fmt"
5 "log"
6 "net"
7 "sync"
8 "time"
10 "context"
11 )
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
15 // servers.
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)
28 defer cancel()
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)
37 }
39 ctx = withVersion(ctx, DefaultVersion)
41 c := &conn{
42 ctx: ctx,
43 ch: ch,
44 handler: handler,
45 closed: make(chan struct{}),
46 }
48 return c.serve()
49 }
51 // conn plays role of session dispatch for handler in a server.
52 type conn struct {
53 ctx context.Context
54 session Session
55 ch Channel
56 handler Handler
58 once sync.Once
59 closed chan struct{}
60 err error // terminal error for the conn
61 }
63 // activeRequest includes information about the active request.
64 type activeRequest struct {
65 ctx context.Context
66 request *Fcall
67 cancel context.CancelFunc
68 }
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
78 // read loop
79 go c.read(requests)
80 go c.write(responses)
82 for {
83 select {
84 case req := <-requests:
85 if _, ok := tags[req.Tag]; ok {
86 select {
87 case responses <- newErrorFcall(req.Tag, ErrDuptag):
88 // Send to responses, bypass tag management.
89 case <-c.ctx.Done():
90 return c.ctx.Err()
91 case <-c.closed:
92 return c.err
93 }
94 continue
95 }
97 switch msg := req.Message.(type) {
98 case MessageTflush:
99 var resp *Fcall
100 // check if we have actually know about the requested flush
101 active, ok := tags[msg.Oldtag]
102 if ok {
103 active.cancel() // propagate cancellation to callees
104 delete(tags, msg.Oldtag)
105 resp = newFcall(req.Tag, MessageRflush{})
106 } else {
107 resp = newErrorFcall(req.Tag, ErrUnknownTag)
110 select {
111 case responses <- resp:
112 // bypass tag management in completed.
113 case <-c.ctx.Done():
114 return c.ctx.Err()
115 case <-c.closed:
116 return c.err
118 default:
119 // Allows us to session handlers to cancel processing of the fcall
120 // through context.
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{
126 ctx: ctx,
127 request: req,
128 cancel: cancel,
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.
135 var resp *Fcall
136 msg, err := c.handler.Handle(ctx, req.Message)
137 if err != nil {
138 // all handler errors are forwarded as protocol errors.
139 resp = newErrorFcall(req.Tag, err)
140 } else {
141 resp = newFcall(req.Tag, msg)
144 select {
145 case completed <- resp:
146 case <-ctx.Done():
147 return
148 case <-c.closed:
149 return
151 }(ctx, req)
153 case resp := <-completed:
154 // only responses that flip the tag state traverse this section.
155 active, ok := tags[resp.Tag]
156 if !ok {
157 // The tag is no longer active. Likely a flushed message.
158 continue
161 select {
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)
169 case <-c.ctx.Done():
170 return c.ctx.Err()
171 case <-c.closed:
172 return c.err
177 // read takes requests off the channel and sends them on requests.
178 func (c *conn) read(requests chan *Fcall) {
179 for {
180 req := new(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
186 // chatty.
187 continue
191 c.CloseWithError(fmt.Errorf("error reading fcall: %v", err))
192 return
195 select {
196 case requests <- req:
197 case <-c.ctx.Done():
198 c.CloseWithError(c.ctx.Err())
199 return
200 case <-c.closed:
201 return
206 func (c *conn) write(responses chan *Fcall) {
207 for {
208 select {
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)
222 continue
226 c.CloseWithError(fmt.Errorf("error writing fcall: %v", err))
227 return
229 case <-c.ctx.Done():
230 c.CloseWithError(c.ctx.Err())
231 return
232 case <-c.closed:
233 return
238 func (c *conn) Close() error {
239 return c.CloseWithError(nil)
242 func (c *conn) CloseWithError(err error) error {
243 c.once.Do(func() {
244 if err == nil {
245 err = ErrClosed
248 c.err = err
249 close(c.closed)
250 })
252 return c.err