1 e342de7d 2015-11-11 adrien package p9p
8 e342de7d 2015-11-11 adrien "golang.org/x/net/context"
11 e342de7d 2015-11-11 adrien // roundTripper manages the request and response from the client-side. A
12 e342de7d 2015-11-11 adrien // roundTripper must abide by similar rules to the http.RoundTripper.
13 e342de7d 2015-11-11 adrien // Typically, the roundTripper will manage tag assignment and message
14 e342de7d 2015-11-11 adrien // serialization.
15 e342de7d 2015-11-11 adrien type roundTripper interface {
16 e342de7d 2015-11-11 adrien send(ctx context.Context, msg Message) (Message, error)
19 e342de7d 2015-11-11 adrien // transport plays the role of being a client channel manager. It multiplexes
20 e342de7d 2015-11-11 adrien // function calls onto the wire and dispatches responses to blocking calls to
21 e342de7d 2015-11-11 adrien // send. On the whole, transport is thread-safe for calling send
22 e342de7d 2015-11-11 adrien type transport struct {
23 e342de7d 2015-11-11 adrien ctx context.Context
24 e342de7d 2015-11-11 adrien ch Channel
25 e342de7d 2015-11-11 adrien requests chan *fcallRequest
26 e342de7d 2015-11-11 adrien closed chan struct{}
28 e342de7d 2015-11-11 adrien tags uint16
31 e342de7d 2015-11-11 adrien var _ roundTripper = &transport{}
33 e342de7d 2015-11-11 adrien func newTransport(ctx context.Context, ch *channel) roundTripper {
34 e342de7d 2015-11-11 adrien t := &transport{
37 e342de7d 2015-11-11 adrien requests: make(chan *fcallRequest),
38 e342de7d 2015-11-11 adrien closed: make(chan struct{}),
41 e342de7d 2015-11-11 adrien go t.handle()
46 e342de7d 2015-11-11 adrien type fcallRequest struct {
47 e342de7d 2015-11-11 adrien ctx context.Context
48 e342de7d 2015-11-11 adrien fcall *Fcall
49 e342de7d 2015-11-11 adrien response chan *Fcall
50 e342de7d 2015-11-11 adrien err chan error
53 e342de7d 2015-11-11 adrien func newFcallRequest(ctx context.Context, fcall *Fcall) *fcallRequest {
54 e342de7d 2015-11-11 adrien return &fcallRequest{
56 e342de7d 2015-11-11 adrien fcall: fcall,
57 e342de7d 2015-11-11 adrien response: make(chan *Fcall, 1),
58 e342de7d 2015-11-11 adrien err: make(chan error, 1),
62 e342de7d 2015-11-11 adrien func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
63 e342de7d 2015-11-11 adrien fcall := newFcall(msg)
64 e342de7d 2015-11-11 adrien req := newFcallRequest(ctx, fcall)
66 e342de7d 2015-11-11 adrien // dispatch the request.
68 e342de7d 2015-11-11 adrien case <-t.closed:
69 e342de7d 2015-11-11 adrien return nil, ErrClosed
70 e342de7d 2015-11-11 adrien case <-ctx.Done():
71 e342de7d 2015-11-11 adrien return nil, ctx.Err()
72 e342de7d 2015-11-11 adrien case t.requests <- req:
75 e342de7d 2015-11-11 adrien // wait for the response.
77 e342de7d 2015-11-11 adrien case <-t.closed:
78 e342de7d 2015-11-11 adrien return nil, ErrClosed
79 e342de7d 2015-11-11 adrien case <-ctx.Done():
80 e342de7d 2015-11-11 adrien return nil, ctx.Err()
81 e342de7d 2015-11-11 adrien case err := <-req.err:
82 e342de7d 2015-11-11 adrien return nil, err
83 e342de7d 2015-11-11 adrien case resp := <-req.response:
84 e342de7d 2015-11-11 adrien if resp.Type == Rerror {
85 e342de7d 2015-11-11 adrien // pack the error into something useful
86 e342de7d 2015-11-11 adrien respmesg, ok := resp.Message.(MessageRerror)
88 e342de7d 2015-11-11 adrien return nil, fmt.Errorf("invalid error response: %v", resp)
91 e342de7d 2015-11-11 adrien return nil, respmesg
94 e342de7d 2015-11-11 adrien return resp.Message, nil
98 e342de7d 2015-11-11 adrien // handle takes messages off the wire and wakes up the waiting tag call.
99 e342de7d 2015-11-11 adrien func (t *transport) handle() {
100 e342de7d 2015-11-11 adrien defer func() {
101 e342de7d 2015-11-11 adrien log.Println("exited handle loop")
102 e342de7d 2015-11-11 adrien t.Close()
104 e342de7d 2015-11-11 adrien // the following variable block are protected components owned by this thread.
106 e342de7d 2015-11-11 adrien responses = make(chan *Fcall)
108 e342de7d 2015-11-11 adrien // outstanding provides a map of tags to outstanding requests.
109 e342de7d 2015-11-11 adrien outstanding = map[Tag]*fcallRequest{}
112 e342de7d 2015-11-11 adrien // loop to read messages off of the connection
113 e342de7d 2015-11-11 adrien go func() {
114 e342de7d 2015-11-11 adrien defer func() {
115 e342de7d 2015-11-11 adrien log.Println("exited read loop")
116 e342de7d 2015-11-11 adrien t.Close()
120 e342de7d 2015-11-11 adrien fcall := new(Fcall)
121 e342de7d 2015-11-11 adrien if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
122 e342de7d 2015-11-11 adrien switch err := err.(type) {
123 e342de7d 2015-11-11 adrien case net.Error:
124 e342de7d 2015-11-11 adrien if err.Timeout() || err.Temporary() {
125 e342de7d 2015-11-11 adrien // BUG(stevvooe): There may be partial reads under
126 e342de7d 2015-11-11 adrien // timeout errors where this is actually fatal.
128 e342de7d 2015-11-11 adrien // can only retry if we haven't offset the frame.
129 e342de7d 2015-11-11 adrien continue loop
133 e342de7d 2015-11-11 adrien log.Println("fatal error reading msg:", err)
134 e342de7d 2015-11-11 adrien t.Close()
139 e342de7d 2015-11-11 adrien case <-t.ctx.Done():
140 e342de7d 2015-11-11 adrien log.Println("ctx done")
142 e342de7d 2015-11-11 adrien case <-t.closed:
143 e342de7d 2015-11-11 adrien log.Println("transport closed")
145 e342de7d 2015-11-11 adrien case responses <- fcall:
151 e342de7d 2015-11-11 adrien log.Println("wait...")
153 e342de7d 2015-11-11 adrien case req := <-t.requests:
154 e342de7d 2015-11-11 adrien if req.fcall.Tag == NOTAG {
155 e342de7d 2015-11-11 adrien // NOTE(stevvooe): We disallow fcalls with NOTAG to come
156 e342de7d 2015-11-11 adrien // through this path since we can't join the tagged response
157 e342de7d 2015-11-11 adrien // with the waiting caller. This is typically used for the
158 e342de7d 2015-11-11 adrien // Tversion/Rversion round trip to setup a session.
160 e342de7d 2015-11-11 adrien // It may be better to allow these through but block all
161 e342de7d 2015-11-11 adrien // requests until a notag message has a response.
163 e342de7d 2015-11-11 adrien req.err <- fmt.Errorf("disallowed tag through transport")
167 e342de7d 2015-11-11 adrien // BUG(stevvooe): This is an awful tag allocation procedure.
168 e342de7d 2015-11-11 adrien // Replace this with something that let's us allocate tags and
169 e342de7d 2015-11-11 adrien // associate data with them, returning to them to a pool when
170 e342de7d 2015-11-11 adrien // complete. Such a system would provide a lot of information
171 e342de7d 2015-11-11 adrien // about outstanding requests.
173 e342de7d 2015-11-11 adrien req.fcall.Tag = tags
174 e342de7d 2015-11-11 adrien outstanding[req.fcall.Tag] = req
176 e342de7d 2015-11-11 adrien // TODO(stevvooe): Consider the case of requests that never
177 e342de7d 2015-11-11 adrien // receive a response. We need to remove the fcall context from
178 e342de7d 2015-11-11 adrien // the tag map and dealloc the tag. We may also want to send a
179 e342de7d 2015-11-11 adrien // flush for the tag.
180 e342de7d 2015-11-11 adrien if err := t.ch.WriteFcall(req.ctx, req.fcall); err != nil {
181 e342de7d 2015-11-11 adrien delete(outstanding, req.fcall.Tag)
182 e342de7d 2015-11-11 adrien req.err <- err
184 e342de7d 2015-11-11 adrien case b := <-responses:
185 e342de7d 2015-11-11 adrien req, ok := outstanding[b.Tag]
187 e342de7d 2015-11-11 adrien panic("unknown tag received")
189 e342de7d 2015-11-11 adrien delete(outstanding, req.fcall.Tag)
191 e342de7d 2015-11-11 adrien req.response <- b
193 e342de7d 2015-11-11 adrien // TODO(stevvooe): Reclaim tag id.
194 e342de7d 2015-11-11 adrien case <-t.ctx.Done():
196 e342de7d 2015-11-11 adrien case <-t.closed:
202 e342de7d 2015-11-11 adrien func (t *transport) flush(ctx context.Context, tag Tag) error {
203 e342de7d 2015-11-11 adrien // TODO(stevvooe): We need to fire and forget flush messages when a call
204 e342de7d 2015-11-11 adrien // context gets cancelled.
205 e342de7d 2015-11-11 adrien panic("not implemented")
208 e342de7d 2015-11-11 adrien func (t *transport) Close() error {
210 e342de7d 2015-11-11 adrien case <-t.closed:
211 e342de7d 2015-11-11 adrien return ErrClosed
212 e342de7d 2015-11-11 adrien case <-t.ctx.Done():
213 e342de7d 2015-11-11 adrien return t.ctx.Err()
215 e342de7d 2015-11-11 adrien close(t.closed)
218 e342de7d 2015-11-11 adrien return nil
