1 package p9p
2 
3 import (
4 	"fmt"
5 	"log"
6 	"net"
7 	"time"
8 
9 	"golang.org/x/net/context"
10 )
11 
12 // TODO(stevvooe): Add net/http.Server-like type here to manage connections.
13 // Coupled with Handler mux, we can get a very http-like experience for 9p
14 // servers.
15 
16 // ServeConn the 9p handler over the provided network connection.
17 func ServeConn(ctx context.Context, cn net.Conn, handler Handler) error {
18 
19 	// TODO(stevvooe): It would be nice if the handler could declare the
20 	// supported version. Before we had handler, we used the session to get
21 	// the version (msize, version := session.Version()). We must decided if
22 	// we want to proxy version and message size decisions all the back to the
23 	// origin server or make those decisions at each link of a proxy chain.
24 
25 	ch := newChannel(cn, codec9p{}, DefaultMSize)
26 	negctx, cancel := context.WithTimeout(ctx, 1*time.Second)
27 	defer cancel()
28 
29 	// TODO(stevvooe): For now, we negotiate here. It probably makes sense to
30 	// do this outside of this function and then pass in a ready made channel.
31 	// We are not really ready to export the channel type yet.
32 
33 	if err := servernegotiate(negctx, ch, DefaultVersion); err != nil {
34 		// TODO(stevvooe): Need better error handling and retry support here.
35 		return fmt.Errorf("error negotiating version:", err)
36 	}
37 
38 	ctx = withVersion(ctx, DefaultVersion)
39 
40 	c := &conn{
41 		ctx:     ctx,
42 		ch:      ch,
43 		handler: handler,
44 		closed:  make(chan struct{}),
45 	}
46 
47 	return c.serve()
48 }
49 
50 // conn plays role of session dispatch for handler in a server.
51 type conn struct {
52 	ctx     context.Context
53 	session Session
54 	ch      Channel
55 	handler Handler
56 	closed  chan struct{}
57 	err     error // terminal error for the conn
58 }
59 
60 // activeRequest includes information about the active request.
61 type activeRequest struct {
62 	ctx     context.Context
63 	request *Fcall
64 	cancel  context.CancelFunc
65 }
66 
67 // serve messages on the connection until an error is encountered.
68 func (c *conn) serve() error {
69 	tags := map[Tag]*activeRequest{} // active requests
70 
71 	requests := make(chan *Fcall) // sync, read-limited
72 	responses := make(chan *Fcall)
73 	completed := make(chan *Fcall, 1)
74 
75 	// read loop
76 	go c.read(requests)
77 	go c.write(responses)
78 
79 	log.Println("server.run()")
80 	for {
81 		select {
82 		case req := <-requests:
83 			if _, ok := tags[req.Tag]; ok {
84 				select {
85 				case responses <- newErrorFcall(req.Tag, ErrDuptag):
86 					// Send to responses, bypass tag management.
87 				case <-c.ctx.Done():
88 					return c.ctx.Err()
89 				case <-c.closed:
90 					return c.err
91 				}
92 				continue
93 			}
94 
95 			switch msg := req.Message.(type) {
96 			case MessageTflush:
97 				log.Println("server: flushing message", msg.Oldtag)
98 
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() // cancel the context of oldtag
104 					resp = newFcall(req.Tag, MessageRflush{})
105 				} else {
106 					resp = newErrorFcall(req.Tag, ErrUnknownTag)
107 				}
108 
109 				select {
110 				case responses <- resp:
111 					// bypass tag management in completed.
112 				case <-c.ctx.Done():
113 					return c.ctx.Err()
114 				case <-c.closed:
115 					return c.err
116 				}
117 			default:
118 				// Allows us to session handlers to cancel processing of the fcall
119 				// through context.
120 				ctx, cancel := context.WithCancel(c.ctx)
121 
122 				// The contents of these instances are only writable in the main
123 				// server loop. The value of tag will not change.
124 				tags[req.Tag] = &activeRequest{
125 					ctx:     ctx,
126 					request: req,
127 					cancel:  cancel,
128 				}
129 
130 				go func(ctx context.Context, req *Fcall) {
131 					var resp *Fcall
132 					msg, err := c.handler.Handle(ctx, req.Message)
133 					if err != nil {
134 						// all handler errors are forwarded as protocol errors.
135 						resp = newErrorFcall(req.Tag, err)
136 					} else {
137 						resp = newFcall(req.Tag, msg)
138 					}
139 
140 					select {
141 					case completed <- resp:
142 					case <-ctx.Done():
143 						return
144 					case <-c.closed:
145 						return
146 					}
147 				}(ctx, req)
148 			}
149 		case resp := <-completed:
150 			// only responses that flip the tag state traverse this section.
151 			active, ok := tags[resp.Tag]
152 			if !ok {
153 				panic("BUG: unbalanced tag")
154 			}
155 
156 			select {
157 			case responses <- resp:
158 			case <-active.ctx.Done():
159 				// the context was canceled for some reason, perhaps timeout or
160 				// due to a flush call. We treat this as a condition where a
161 				// response should not be sent.
162 				log.Println("canceled", resp, active.ctx.Err())
163 			}
164 			delete(tags, resp.Tag)
165 		case <-c.ctx.Done():
166 			return c.ctx.Err()
167 		case <-c.closed:
168 			return c.err
169 		}
170 	}
171 }
172 
173 // read takes requests off the channel and sends them on requests.
174 func (c *conn) read(requests chan *Fcall) {
175 	for {
176 		req := new(Fcall)
177 		if err := c.ch.ReadFcall(c.ctx, req); err != nil {
178 			if err, ok := err.(net.Error); ok {
179 				if err.Timeout() || err.Temporary() {
180 					// TODO(stevvooe): A full idle timeout on the connection
181 					// should be enforced here. No logging because it is quite
182 					// chatty.
183 					continue
184 				}
185 			}
186 
187 			c.CloseWithError(fmt.Errorf("error reading fcall: %v", err))
188 			return
189 		}
190 
191 		select {
192 		case requests <- req:
193 		case <-c.ctx.Done():
194 			c.CloseWithError(c.ctx.Err())
195 			return
196 		case <-c.closed:
197 			return
198 		}
199 	}
200 }
201 
202 func (c *conn) write(responses chan *Fcall) {
203 	for {
204 		select {
205 		case resp := <-responses:
206 			if err := c.ch.WriteFcall(c.ctx, resp); err != nil {
207 				if err, ok := err.(net.Error); ok {
208 					if err.Timeout() || err.Temporary() {
209 						// TODO(stevvooe): A full idle timeout on the
210 						// connection should be enforced here. We log here,
211 						// since this is less common.
212 						log.Println("9p server: temporary error writing fcall: %v", err)
213 						continue
214 					}
215 				}
216 
217 				c.CloseWithError(fmt.Errorf("error writing fcall: %v", err))
218 				return
219 			}
220 		case <-c.ctx.Done():
221 			c.CloseWithError(c.ctx.Err())
222 			return
223 		case <-c.closed:
224 			return
225 		}
226 	}
227 }
228 
229 func (c *conn) Close() error {
230 	return c.CloseWithError(nil)
231 }
232 
233 func (c *conn) CloseWithError(err error) error {
234 	select {
235 	case <-c.closed:
236 		return c.err
237 	default:
238 		close(c.closed)
239 		if err == nil {
240 			c.err = err
241 		} else {
242 			c.err = ErrClosed
243 		}
244 
245 		return c.err
246 	}
247 }