1 e6bcde66 2015-10-29 stephen.d package p9pnew
3 e6bcde66 2015-10-29 stephen.d import (
8 e6bcde66 2015-10-29 stephen.d "golang.org/x/net/context"
11 e6bcde66 2015-10-29 stephen.d // roundTripper manages the request and response from the client-side. A
12 e6bcde66 2015-10-29 stephen.d // roundTripper must abide by many of the rules of http.RoundTripper.
13 e6bcde66 2015-10-29 stephen.d // Typically, the roundTripper will manage tag assignment and message
14 e6bcde66 2015-10-29 stephen.d // serialization.
15 e6bcde66 2015-10-29 stephen.d type roundTripper interface {
16 e6bcde66 2015-10-29 stephen.d send(ctx context.Context, fc *Fcall) (*Fcall, error)
19 40d4a02d 2015-11-03 stephen.d // transport plays the role of being a client channel manager. It multiplexes
20 40d4a02d 2015-11-03 stephen.d // function calls onto the wire and dispatches responses to blocking calls to
21 40d4a02d 2015-11-03 stephen.d // send. On the whole, transport is thread-safe for calling send
22 e6bcde66 2015-10-29 stephen.d type transport struct {
23 e6bcde66 2015-10-29 stephen.d ctx context.Context
24 3bf22e58 2015-11-04 stephen.d ch Channel
25 e6bcde66 2015-10-29 stephen.d requests chan *fcallRequest
26 e6bcde66 2015-10-29 stephen.d closed chan struct{}
28 e6bcde66 2015-10-29 stephen.d tags uint16
31 fb37ce2a 2015-10-30 stephen.d func newTransport(ctx context.Context, ch *channel) roundTripper {
32 e6bcde66 2015-10-29 stephen.d t := &transport{
33 e6bcde66 2015-10-29 stephen.d ctx: ctx,
34 fb37ce2a 2015-10-30 stephen.d ch: ch,
35 e6bcde66 2015-10-29 stephen.d requests: make(chan *fcallRequest),
36 e6bcde66 2015-10-29 stephen.d closed: make(chan struct{}),
39 e6bcde66 2015-10-29 stephen.d go t.handle()
41 e6bcde66 2015-10-29 stephen.d return t
44 e6bcde66 2015-10-29 stephen.d type fcallRequest struct {
45 e6bcde66 2015-10-29 stephen.d ctx context.Context
46 e6bcde66 2015-10-29 stephen.d fcall *Fcall
47 e6bcde66 2015-10-29 stephen.d response chan *Fcall
48 e6bcde66 2015-10-29 stephen.d err chan error
51 e6bcde66 2015-10-29 stephen.d func newFcallRequest(ctx context.Context, fcall *Fcall) *fcallRequest {
52 e6bcde66 2015-10-29 stephen.d return &fcallRequest{
53 e6bcde66 2015-10-29 stephen.d ctx: ctx,
54 e6bcde66 2015-10-29 stephen.d fcall: fcall,
55 e6bcde66 2015-10-29 stephen.d response: make(chan *Fcall, 1),
56 e6bcde66 2015-10-29 stephen.d err: make(chan error, 1),
60 e6bcde66 2015-10-29 stephen.d func (t *transport) send(ctx context.Context, fcall *Fcall) (*Fcall, error) {
61 e6bcde66 2015-10-29 stephen.d req := newFcallRequest(ctx, fcall)
63 e6bcde66 2015-10-29 stephen.d log.Println("dispatch", fcall)
64 e6bcde66 2015-10-29 stephen.d // dispatch the request.
65 e6bcde66 2015-10-29 stephen.d select {
66 e6bcde66 2015-10-29 stephen.d case <-t.closed:
67 e6bcde66 2015-10-29 stephen.d return nil, ErrClosed
68 e6bcde66 2015-10-29 stephen.d case <-ctx.Done():
69 e6bcde66 2015-10-29 stephen.d return nil, ctx.Err()
70 e6bcde66 2015-10-29 stephen.d case t.requests <- req:
73 e6bcde66 2015-10-29 stephen.d log.Println("wait", fcall)
74 e6bcde66 2015-10-29 stephen.d // wait for the response.
75 e6bcde66 2015-10-29 stephen.d select {
76 e6bcde66 2015-10-29 stephen.d case <-t.closed:
77 e6bcde66 2015-10-29 stephen.d return nil, ErrClosed
78 e6bcde66 2015-10-29 stephen.d case <-ctx.Done():
79 e6bcde66 2015-10-29 stephen.d return nil, ctx.Err()
80 a8abc687 2015-10-30 stephen.d case err := <-req.err:
81 a8abc687 2015-10-30 stephen.d return nil, err
82 e6bcde66 2015-10-29 stephen.d case resp := <-req.response:
83 fb37ce2a 2015-10-30 stephen.d log.Println("resp", resp)
84 97423e8b 2015-10-29 stephen.d if resp.Type == Rerror {
85 97423e8b 2015-10-29 stephen.d // pack the error into something useful
86 97423e8b 2015-10-29 stephen.d respmesg, ok := resp.Message.(*MessageRerror)
87 97423e8b 2015-10-29 stephen.d if !ok {
88 97423e8b 2015-10-29 stephen.d return nil, fmt.Errorf("invalid error response: %v", resp)
91 97423e8b 2015-10-29 stephen.d return nil, new9pError(respmesg.Ename)
94 e6bcde66 2015-10-29 stephen.d return resp, nil
98 e6bcde66 2015-10-29 stephen.d // handle takes messages off the wire and wakes up the waiting tag call.
99 e6bcde66 2015-10-29 stephen.d func (t *transport) handle() {
100 a8abc687 2015-10-30 stephen.d defer func() {
101 a8abc687 2015-10-30 stephen.d log.Println("exited handle loop")
102 a8abc687 2015-10-30 stephen.d close(t.closed)
104 e6bcde66 2015-10-29 stephen.d // the following variable block are protected components owned by this thread.
106 e6bcde66 2015-10-29 stephen.d responses = make(chan *Fcall)
107 e6bcde66 2015-10-29 stephen.d tags Tag
108 e6bcde66 2015-10-29 stephen.d // outstanding provides a map of tags to outstanding requests.
109 e6bcde66 2015-10-29 stephen.d outstanding = map[Tag]*fcallRequest{}
112 e6bcde66 2015-10-29 stephen.d // loop to read messages off of the connection
113 e6bcde66 2015-10-29 stephen.d go func() {
114 fb37ce2a 2015-10-30 stephen.d defer func() {
115 fb37ce2a 2015-10-30 stephen.d log.Println("exited read loop")
116 fb37ce2a 2015-10-30 stephen.d close(t.closed)
120 fb37ce2a 2015-10-30 stephen.d fcall := new(Fcall)
121 3bf22e58 2015-11-04 stephen.d if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
122 e6bcde66 2015-10-29 stephen.d switch err := err.(type) {
123 e6bcde66 2015-10-29 stephen.d case net.Error:
124 e6bcde66 2015-10-29 stephen.d if err.Timeout() || err.Temporary() {
125 fb37ce2a 2015-10-30 stephen.d // BUG(stevvooe): There may be partial reads under
126 fb37ce2a 2015-10-30 stephen.d // timeout errors where this is actually fatal.
128 fb37ce2a 2015-10-30 stephen.d // can only retry if we haven't offset the frame.
129 fb37ce2a 2015-10-30 stephen.d continue loop
133 fb37ce2a 2015-10-30 stephen.d log.Println("fatal error reading msg:", err)
134 fb37ce2a 2015-10-30 stephen.d t.Close()
135 fb37ce2a 2015-10-30 stephen.d return
138 e6bcde66 2015-10-29 stephen.d select {
139 e6bcde66 2015-10-29 stephen.d case <-t.ctx.Done():
140 fb37ce2a 2015-10-30 stephen.d log.Println("ctx done")
141 e6bcde66 2015-10-29 stephen.d return
142 e6bcde66 2015-10-29 stephen.d case <-t.closed:
143 fb37ce2a 2015-10-30 stephen.d log.Println("transport closed")
144 e6bcde66 2015-10-29 stephen.d return
145 fb37ce2a 2015-10-30 stephen.d case responses <- fcall:
151 a8abc687 2015-10-30 stephen.d log.Println("wait...")
152 e6bcde66 2015-10-29 stephen.d select {
153 e6bcde66 2015-10-29 stephen.d case req := <-t.requests:
154 74ec7ac9 2015-10-30 stephen.d if req.fcall.Tag == NOTAG {
155 74ec7ac9 2015-10-30 stephen.d // NOTE(stevvooe): We disallow fcalls with NOTAG to come
156 74ec7ac9 2015-10-30 stephen.d // through this path since we can't join the tagged response
157 74ec7ac9 2015-10-30 stephen.d // with the waiting caller. This is typically used for the
158 74ec7ac9 2015-10-30 stephen.d // Tversion/Rversion round trip to setup a session.
160 74ec7ac9 2015-10-30 stephen.d // It may be better to allow these through but block all
161 74ec7ac9 2015-10-30 stephen.d // requests until a notag message has a response.
163 74ec7ac9 2015-10-30 stephen.d req.err <- fmt.Errorf("disallowed tag through transport")
164 74ec7ac9 2015-10-30 stephen.d continue
167 74ec7ac9 2015-10-30 stephen.d // BUG(stevvooe): This is an awful tag allocation procedure.
168 74ec7ac9 2015-10-30 stephen.d // Replace this with something that let's us allocate tags and
169 74ec7ac9 2015-10-30 stephen.d // associate data with them, returning to them to a pool when
170 74ec7ac9 2015-10-30 stephen.d // complete. Such a system would provide a lot of information
171 74ec7ac9 2015-10-30 stephen.d // about outstanding requests.
172 74ec7ac9 2015-10-30 stephen.d tags++
173 74ec7ac9 2015-10-30 stephen.d req.fcall.Tag = tags
174 74ec7ac9 2015-10-30 stephen.d outstanding[req.fcall.Tag] = req
176 e6bcde66 2015-10-29 stephen.d // TODO(stevvooe): Consider the case of requests that never
177 e6bcde66 2015-10-29 stephen.d // receive a response. We need to remove the fcall context from
178 e6bcde66 2015-10-29 stephen.d // the tag map and dealloc the tag. We may also want to send a
179 e6bcde66 2015-10-29 stephen.d // flush for the tag.
180 3bf22e58 2015-11-04 stephen.d if err := t.ch.WriteFcall(req.ctx, req.fcall); err != nil {
181 a8abc687 2015-10-30 stephen.d log.Println("error writing fcall", err, req.fcall)
182 e6bcde66 2015-10-29 stephen.d delete(outstanding, req.fcall.Tag)
183 e6bcde66 2015-10-29 stephen.d req.err <- err
186 e6bcde66 2015-10-29 stephen.d log.Println("sent", req.fcall)
187 e6bcde66 2015-10-29 stephen.d case b := <-responses:
188 fb37ce2a 2015-10-30 stephen.d log.Println("recv", b)
189 e6bcde66 2015-10-29 stephen.d req, ok := outstanding[b.Tag]
190 e6bcde66 2015-10-29 stephen.d if !ok {
191 e6bcde66 2015-10-29 stephen.d panic("unknown tag received")
193 e6bcde66 2015-10-29 stephen.d delete(outstanding, req.fcall.Tag)
195 e6bcde66 2015-10-29 stephen.d req.response <- b
197 e6bcde66 2015-10-29 stephen.d // TODO(stevvooe): Reclaim tag id.
198 e6bcde66 2015-10-29 stephen.d case <-t.ctx.Done():
199 e6bcde66 2015-10-29 stephen.d return
200 e6bcde66 2015-10-29 stephen.d case <-t.closed:
201 e6bcde66 2015-10-29 stephen.d return
206 e6bcde66 2015-10-29 stephen.d func (t *transport) Close() error {
207 e6bcde66 2015-10-29 stephen.d select {
208 e6bcde66 2015-10-29 stephen.d case <-t.closed:
209 e6bcde66 2015-10-29 stephen.d return ErrClosed
210 e6bcde66 2015-10-29 stephen.d case <-t.ctx.Done():
211 e6bcde66 2015-10-29 stephen.d return t.ctx.Err()
212 e6bcde66 2015-10-29 stephen.d default:
213 e6bcde66 2015-10-29 stephen.d close(t.closed)
216 e6bcde66 2015-10-29 stephen.d return nil
