1 e342de7d 2015-11-11 adrien package p9p
9 e342de7d 2015-11-11 adrien "golang.org/x/net/context"
12 641595f8 2015-11-14 stephen.d // TODO(stevvooe): Add net/http.Server-like type here to manage connections.
13 641595f8 2015-11-14 stephen.d // Coupled with Handler mux, we can get a very http-like experience for 9p
14 641595f8 2015-11-14 stephen.d // servers.
16 641595f8 2015-11-14 stephen.d // ServeConn the 9p handler over the provided network connection.
17 641595f8 2015-11-14 stephen.d func ServeConn(ctx context.Context, cn net.Conn, handler Handler) error {
19 e342de7d 2015-11-11 adrien // TODO(stevvooe): It would be nice if the handler could declare the
20 e342de7d 2015-11-11 adrien // supported version. Before we had handler, we used the session to get
21 e342de7d 2015-11-11 adrien // the version (msize, version := session.Version()). We must decided if
22 e342de7d 2015-11-11 adrien // we want to proxy version and message size decisions all the back to the
23 e342de7d 2015-11-11 adrien // origin server or make those decisions at each link of a proxy chain.
25 641595f8 2015-11-14 stephen.d ch := newChannel(cn, codec9p{}, DefaultMSize)
26 e342de7d 2015-11-11 adrien negctx, cancel := context.WithTimeout(ctx, 1*time.Second)
27 e342de7d 2015-11-11 adrien defer cancel()
29 e342de7d 2015-11-11 adrien // TODO(stevvooe): For now, we negotiate here. It probably makes sense to
30 e342de7d 2015-11-11 adrien // do this outside of this function and then pass in a ready made channel.
31 e342de7d 2015-11-11 adrien // We are not really ready to export the channel type yet.
33 e342de7d 2015-11-11 adrien if err := servernegotiate(negctx, ch, DefaultVersion); err != nil {
34 e342de7d 2015-11-11 adrien // TODO(stevvooe): Need better error handling and retry support here.
35 641595f8 2015-11-14 stephen.d return fmt.Errorf("error negotiating version:", err)
38 e342de7d 2015-11-11 adrien ctx = withVersion(ctx, DefaultVersion)
40 641595f8 2015-11-14 stephen.d c := &conn{
43 e342de7d 2015-11-11 adrien handler: handler,
44 e342de7d 2015-11-11 adrien closed: make(chan struct{}),
47 641595f8 2015-11-14 stephen.d return c.serve()
50 641595f8 2015-11-14 stephen.d // conn plays role of session dispatch for handler in a server.
51 641595f8 2015-11-14 stephen.d type conn struct {
52 e342de7d 2015-11-11 adrien ctx context.Context
53 e342de7d 2015-11-11 adrien session Session
54 e342de7d 2015-11-11 adrien ch Channel
55 e342de7d 2015-11-11 adrien handler Handler
56 e342de7d 2015-11-11 adrien closed chan struct{}
57 641595f8 2015-11-14 stephen.d err error // terminal error for the conn
60 ceb907e4 2015-11-11 adrien // activeRequest includes information about the active request.
61 ceb907e4 2015-11-11 adrien type activeRequest struct {
62 ceb907e4 2015-11-11 adrien ctx context.Context
63 ceb907e4 2015-11-11 adrien request *Fcall
64 ceb907e4 2015-11-11 adrien cancel context.CancelFunc
67 641595f8 2015-11-14 stephen.d // serve messages on the connection until an error is encountered.
68 641595f8 2015-11-14 stephen.d func (c *conn) serve() error {
69 ceb907e4 2015-11-11 adrien tags := map[Tag]*activeRequest{} // active requests
71 ceb907e4 2015-11-11 adrien requests := make(chan *Fcall) // sync, read-limited
72 ceb907e4 2015-11-11 adrien responses := make(chan *Fcall)
73 ceb907e4 2015-11-11 adrien completed := make(chan *Fcall, 1)
75 ceb907e4 2015-11-11 adrien // read loop
76 641595f8 2015-11-14 stephen.d go c.read(requests)
77 641595f8 2015-11-14 stephen.d go c.write(responses)
79 ceb907e4 2015-11-11 adrien log.Println("server.run()")
82 ceb907e4 2015-11-11 adrien case req := <-requests:
83 ceb907e4 2015-11-11 adrien if _, ok := tags[req.Tag]; ok {
85 ceb907e4 2015-11-11 adrien case responses <- newErrorFcall(req.Tag, ErrDuptag):
86 ceb907e4 2015-11-11 adrien // Send to responses, bypass tag management.
87 641595f8 2015-11-14 stephen.d case <-c.ctx.Done():
88 641595f8 2015-11-14 stephen.d return c.ctx.Err()
89 641595f8 2015-11-14 stephen.d case <-c.closed:
90 641595f8 2015-11-14 stephen.d return c.err
95 ceb907e4 2015-11-11 adrien switch msg := req.Message.(type) {
96 ceb907e4 2015-11-11 adrien case MessageTflush:
97 ceb907e4 2015-11-11 adrien log.Println("server: flushing message", msg.Oldtag)
99 ceb907e4 2015-11-11 adrien var resp *Fcall
100 ceb907e4 2015-11-11 adrien // check if we have actually know about the requested flush
101 ceb907e4 2015-11-11 adrien active, ok := tags[msg.Oldtag]
103 ceb907e4 2015-11-11 adrien active.cancel() // cancel the context of oldtag
104 ceb907e4 2015-11-11 adrien resp = newFcall(req.Tag, MessageRflush{})
106 ceb907e4 2015-11-11 adrien resp = newErrorFcall(req.Tag, ErrUnknownTag)
110 ceb907e4 2015-11-11 adrien case responses <- resp:
111 ceb907e4 2015-11-11 adrien // bypass tag management in completed.
112 641595f8 2015-11-14 stephen.d case <-c.ctx.Done():
113 641595f8 2015-11-14 stephen.d return c.ctx.Err()
114 641595f8 2015-11-14 stephen.d case <-c.closed:
115 641595f8 2015-11-14 stephen.d return c.err
118 ceb907e4 2015-11-11 adrien // Allows us to session handlers to cancel processing of the fcall
119 ceb907e4 2015-11-11 adrien // through context.
120 641595f8 2015-11-14 stephen.d ctx, cancel := context.WithCancel(c.ctx)
122 ceb907e4 2015-11-11 adrien // The contents of these instances are only writable in the main
123 ceb907e4 2015-11-11 adrien // server loop. The value of tag will not change.
124 ceb907e4 2015-11-11 adrien tags[req.Tag] = &activeRequest{
125 ceb907e4 2015-11-11 adrien ctx: ctx,
126 ceb907e4 2015-11-11 adrien request: req,
127 ceb907e4 2015-11-11 adrien cancel: cancel,
130 ceb907e4 2015-11-11 adrien go func(ctx context.Context, req *Fcall) {
131 ceb907e4 2015-11-11 adrien var resp *Fcall
132 641595f8 2015-11-14 stephen.d msg, err := c.handler.Handle(ctx, req.Message)
133 ceb907e4 2015-11-11 adrien if err != nil {
134 ceb907e4 2015-11-11 adrien // all handler errors are forwarded as protocol errors.
135 ceb907e4 2015-11-11 adrien resp = newErrorFcall(req.Tag, err)
137 ceb907e4 2015-11-11 adrien resp = newFcall(req.Tag, msg)
141 ceb907e4 2015-11-11 adrien case completed <- resp:
142 ceb907e4 2015-11-11 adrien case <-ctx.Done():
144 641595f8 2015-11-14 stephen.d case <-c.closed:
147 ceb907e4 2015-11-11 adrien }(ctx, req)
149 ceb907e4 2015-11-11 adrien case resp := <-completed:
150 ceb907e4 2015-11-11 adrien // only responses that flip the tag state traverse this section.
151 ceb907e4 2015-11-11 adrien active, ok := tags[resp.Tag]
153 ceb907e4 2015-11-11 adrien panic("BUG: unbalanced tag")
157 ceb907e4 2015-11-11 adrien case responses <- resp:
158 ceb907e4 2015-11-11 adrien case <-active.ctx.Done():
159 ceb907e4 2015-11-11 adrien // the context was canceled for some reason, perhaps timeout or
160 ceb907e4 2015-11-11 adrien // due to a flush call. We treat this as a condition where a
161 ceb907e4 2015-11-11 adrien // response should not be sent.
162 ceb907e4 2015-11-11 adrien log.Println("canceled", resp, active.ctx.Err())
164 ceb907e4 2015-11-11 adrien delete(tags, resp.Tag)
165 641595f8 2015-11-14 stephen.d case <-c.ctx.Done():
166 641595f8 2015-11-14 stephen.d return c.ctx.Err()
167 641595f8 2015-11-14 stephen.d case <-c.closed:
168 641595f8 2015-11-14 stephen.d return c.err
173 641595f8 2015-11-14 stephen.d // read takes requests off the channel and sends them on requests.
174 641595f8 2015-11-14 stephen.d func (c *conn) read(requests chan *Fcall) {
176 641595f8 2015-11-14 stephen.d req := new(Fcall)
177 641595f8 2015-11-14 stephen.d if err := c.ch.ReadFcall(c.ctx, req); err != nil {
178 641595f8 2015-11-14 stephen.d if err, ok := err.(net.Error); ok {
179 641595f8 2015-11-14 stephen.d if err.Timeout() || err.Temporary() {
180 641595f8 2015-11-14 stephen.d // TODO(stevvooe): A full idle timeout on the connection
181 641595f8 2015-11-14 stephen.d // should be enforced here. No logging because it is quite
182 641595f8 2015-11-14 stephen.d // chatty.
183 641595f8 2015-11-14 stephen.d continue
187 641595f8 2015-11-14 stephen.d c.CloseWithError(fmt.Errorf("error reading fcall: %v", err))
191 641595f8 2015-11-14 stephen.d select {
192 641595f8 2015-11-14 stephen.d case requests <- req:
193 641595f8 2015-11-14 stephen.d case <-c.ctx.Done():
194 641595f8 2015-11-14 stephen.d c.CloseWithError(c.ctx.Err())
196 641595f8 2015-11-14 stephen.d case <-c.closed:
197 641595f8 2015-11-14 stephen.d return
202 641595f8 2015-11-14 stephen.d func (c *conn) write(responses chan *Fcall) {
204 641595f8 2015-11-14 stephen.d select {
205 641595f8 2015-11-14 stephen.d case resp := <-responses:
206 641595f8 2015-11-14 stephen.d if err := c.ch.WriteFcall(c.ctx, resp); err != nil {
207 641595f8 2015-11-14 stephen.d if err, ok := err.(net.Error); ok {
208 641595f8 2015-11-14 stephen.d if err.Timeout() || err.Temporary() {
209 641595f8 2015-11-14 stephen.d // TODO(stevvooe): A full idle timeout on the
210 641595f8 2015-11-14 stephen.d // connection should be enforced here. We log here,
211 641595f8 2015-11-14 stephen.d // since this is less common.
212 641595f8 2015-11-14 stephen.d log.Println("9p server: temporary error writing fcall: %v", err)
213 641595f8 2015-11-14 stephen.d continue
217 641595f8 2015-11-14 stephen.d c.CloseWithError(fmt.Errorf("error writing fcall: %v", err))
218 641595f8 2015-11-14 stephen.d return
220 641595f8 2015-11-14 stephen.d case <-c.ctx.Done():
221 641595f8 2015-11-14 stephen.d c.CloseWithError(c.ctx.Err())
222 641595f8 2015-11-14 stephen.d return
223 641595f8 2015-11-14 stephen.d case <-c.closed:
224 641595f8 2015-11-14 stephen.d return
229 641595f8 2015-11-14 stephen.d func (c *conn) Close() error {
230 641595f8 2015-11-14 stephen.d return c.CloseWithError(nil)
233 641595f8 2015-11-14 stephen.d func (c *conn) CloseWithError(err error) error {
234 641595f8 2015-11-14 stephen.d select {
235 641595f8 2015-11-14 stephen.d case <-c.closed:
236 641595f8 2015-11-14 stephen.d return c.err
237 641595f8 2015-11-14 stephen.d default:
238 641595f8 2015-11-14 stephen.d close(c.closed)
239 641595f8 2015-11-14 stephen.d if err == nil {
240 641595f8 2015-11-14 stephen.d c.err = err
241 641595f8 2015-11-14 stephen.d } else {
242 641595f8 2015-11-14 stephen.d c.err = ErrClosed
245 641595f8 2015-11-14 stephen.d return c.err
