1 4b33cdd0 2015-11-30 stephen.d package p9p
3 4b33cdd0 2015-11-30 stephen.d import (
9 4b33cdd0 2015-11-30 stephen.d "golang.org/x/net/context"
12 4b33cdd0 2015-11-30 stephen.d // roundTripper manages the request and response from the client-side. A
13 4b33cdd0 2015-11-30 stephen.d // roundTripper must abide by similar rules to the http.RoundTripper.
14 4b33cdd0 2015-11-30 stephen.d // Typically, the roundTripper will manage tag assignment and message
15 4b33cdd0 2015-11-30 stephen.d // serialization.
16 4b33cdd0 2015-11-30 stephen.d type roundTripper interface {
17 4b33cdd0 2015-11-30 stephen.d send(ctx context.Context, msg Message) (Message, error)
20 4b33cdd0 2015-11-30 stephen.d // transport plays the role of being a client channel manager. It multiplexes
21 4b33cdd0 2015-11-30 stephen.d // function calls onto the wire and dispatches responses to blocking calls to
22 4b33cdd0 2015-11-30 stephen.d // send. On the whole, transport is thread-safe for calling send
23 4b33cdd0 2015-11-30 stephen.d type transport struct {
24 4b33cdd0 2015-11-30 stephen.d ctx context.Context
25 4b33cdd0 2015-11-30 stephen.d ch Channel
26 4b33cdd0 2015-11-30 stephen.d requests chan *fcallRequest
27 4b33cdd0 2015-11-30 stephen.d closed chan struct{}
29 4b33cdd0 2015-11-30 stephen.d tags uint16
32 4b33cdd0 2015-11-30 stephen.d var _ roundTripper = &transport{}
34 20882a7b 2016-05-25 stevvooe func newTransport(ctx context.Context, ch Channel) roundTripper {
35 4b33cdd0 2015-11-30 stephen.d t := &transport{
36 4b33cdd0 2015-11-30 stephen.d ctx: ctx,
37 4b33cdd0 2015-11-30 stephen.d ch: ch,
38 4b33cdd0 2015-11-30 stephen.d requests: make(chan *fcallRequest),
39 4b33cdd0 2015-11-30 stephen.d closed: make(chan struct{}),
42 4b33cdd0 2015-11-30 stephen.d go t.handle()
44 4b33cdd0 2015-11-30 stephen.d return t
47 4b33cdd0 2015-11-30 stephen.d // fcallRequest encompasses the request to send a message via fcall.
48 4b33cdd0 2015-11-30 stephen.d type fcallRequest struct {
49 4b33cdd0 2015-11-30 stephen.d ctx context.Context
50 4b33cdd0 2015-11-30 stephen.d message Message
51 4b33cdd0 2015-11-30 stephen.d response chan *Fcall
52 4b33cdd0 2015-11-30 stephen.d err chan error
55 4b33cdd0 2015-11-30 stephen.d func newFcallRequest(ctx context.Context, msg Message) *fcallRequest {
56 4b33cdd0 2015-11-30 stephen.d return &fcallRequest{
57 4b33cdd0 2015-11-30 stephen.d ctx: ctx,
58 4b33cdd0 2015-11-30 stephen.d message: msg,
59 4b33cdd0 2015-11-30 stephen.d response: make(chan *Fcall, 1),
60 4b33cdd0 2015-11-30 stephen.d err: make(chan error, 1),
64 4b33cdd0 2015-11-30 stephen.d func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
65 4b33cdd0 2015-11-30 stephen.d req := newFcallRequest(ctx, msg)
67 4b33cdd0 2015-11-30 stephen.d // dispatch the request.
68 4b33cdd0 2015-11-30 stephen.d select {
69 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
70 4b33cdd0 2015-11-30 stephen.d return nil, ErrClosed
71 4b33cdd0 2015-11-30 stephen.d case <-ctx.Done():
72 4b33cdd0 2015-11-30 stephen.d return nil, ctx.Err()
73 4b33cdd0 2015-11-30 stephen.d case t.requests <- req:
76 4b33cdd0 2015-11-30 stephen.d // wait for the response.
77 4b33cdd0 2015-11-30 stephen.d select {
78 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
79 4b33cdd0 2015-11-30 stephen.d return nil, ErrClosed
80 4b33cdd0 2015-11-30 stephen.d case <-ctx.Done():
81 4b33cdd0 2015-11-30 stephen.d return nil, ctx.Err()
82 4b33cdd0 2015-11-30 stephen.d case err := <-req.err:
83 4b33cdd0 2015-11-30 stephen.d return nil, err
84 4b33cdd0 2015-11-30 stephen.d case resp := <-req.response:
85 4b33cdd0 2015-11-30 stephen.d if resp.Type == Rerror {
86 4b33cdd0 2015-11-30 stephen.d // pack the error into something useful
87 4b33cdd0 2015-11-30 stephen.d respmesg, ok := resp.Message.(MessageRerror)
88 4b33cdd0 2015-11-30 stephen.d if !ok {
89 4b33cdd0 2015-11-30 stephen.d return nil, fmt.Errorf("invalid error response: %v", resp)
92 4b33cdd0 2015-11-30 stephen.d return nil, respmesg
95 4b33cdd0 2015-11-30 stephen.d return resp.Message, nil
99 11b5e5c7 2016-05-25 stevvooe // allocateTag returns a valid tag given a tag pool map. It receives a hint as
100 11b5e5c7 2016-05-25 stevvooe // to where to start the tag search. It returns an error if the allocation is
101 11b5e5c7 2016-05-25 stevvooe // not possible. The provided map must not contain NOTAG as a key.
102 11b5e5c7 2016-05-25 stevvooe func allocateTag(r *fcallRequest, m map[Tag]*fcallRequest, hint Tag) (Tag, error) {
103 11b5e5c7 2016-05-25 stevvooe // The tag pool is depleted if 65535 (0xFFFF) tags are taken.
104 11b5e5c7 2016-05-25 stevvooe if len(m) >= 0xFFFF {
105 11b5e5c7 2016-05-25 stevvooe return 0, errors.New("tag pool depleted")
108 11b5e5c7 2016-05-25 stevvooe // Look for the first tag that doesn't exist in the map and return it.
109 11b5e5c7 2016-05-25 stevvooe for i := 0; i < 0xFFFF; i++ {
111 11b5e5c7 2016-05-25 stevvooe if hint == NOTAG {
112 11b5e5c7 2016-05-25 stevvooe hint = 0
115 11b5e5c7 2016-05-25 stevvooe if _, exists := m[hint]; !exists {
116 11b5e5c7 2016-05-25 stevvooe return hint, nil
120 11b5e5c7 2016-05-25 stevvooe return 0, errors.New("allocateTag: unexpected error")
123 4b33cdd0 2015-11-30 stephen.d // handle takes messages off the wire and wakes up the waiting tag call.
124 4b33cdd0 2015-11-30 stephen.d func (t *transport) handle() {
125 4b33cdd0 2015-11-30 stephen.d defer func() {
126 4b33cdd0 2015-11-30 stephen.d log.Println("exited handle loop")
127 4b33cdd0 2015-11-30 stephen.d t.Close()
129 4b33cdd0 2015-11-30 stephen.d // the following variable block are protected components owned by this thread.
131 4b33cdd0 2015-11-30 stephen.d responses = make(chan *Fcall)
132 4b33cdd0 2015-11-30 stephen.d // outstanding provides a map of tags to outstanding requests.
133 4b33cdd0 2015-11-30 stephen.d outstanding = map[Tag]*fcallRequest{}
134 11b5e5c7 2016-05-25 stevvooe selected Tag
137 4b33cdd0 2015-11-30 stephen.d // loop to read messages off of the connection
138 4b33cdd0 2015-11-30 stephen.d go func() {
139 4b33cdd0 2015-11-30 stephen.d defer func() {
140 4b33cdd0 2015-11-30 stephen.d log.Println("exited read loop")
141 4b33cdd0 2015-11-30 stephen.d t.Close()
145 4b33cdd0 2015-11-30 stephen.d fcall := new(Fcall)
146 4b33cdd0 2015-11-30 stephen.d if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
147 4b33cdd0 2015-11-30 stephen.d switch err := err.(type) {
148 4b33cdd0 2015-11-30 stephen.d case net.Error:
149 4b33cdd0 2015-11-30 stephen.d if err.Timeout() || err.Temporary() {
150 4b33cdd0 2015-11-30 stephen.d // BUG(stevvooe): There may be partial reads under
151 4b33cdd0 2015-11-30 stephen.d // timeout errors where this is actually fatal.
153 4b33cdd0 2015-11-30 stephen.d // can only retry if we haven't offset the frame.
154 4b33cdd0 2015-11-30 stephen.d continue loop
158 4b33cdd0 2015-11-30 stephen.d log.Println("fatal error reading msg:", err)
159 4b33cdd0 2015-11-30 stephen.d t.Close()
160 4b33cdd0 2015-11-30 stephen.d return
163 4b33cdd0 2015-11-30 stephen.d select {
164 4b33cdd0 2015-11-30 stephen.d case <-t.ctx.Done():
165 4b33cdd0 2015-11-30 stephen.d log.Println("ctx done")
166 4b33cdd0 2015-11-30 stephen.d return
167 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
168 4b33cdd0 2015-11-30 stephen.d log.Println("transport closed")
169 4b33cdd0 2015-11-30 stephen.d return
170 4b33cdd0 2015-11-30 stephen.d case responses <- fcall:
176 4b33cdd0 2015-11-30 stephen.d select {
177 4b33cdd0 2015-11-30 stephen.d case req := <-t.requests:
178 11b5e5c7 2016-05-25 stevvooe var err error
180 11b5e5c7 2016-05-25 stevvooe selected, err = allocateTag(req, outstanding, selected)
181 11b5e5c7 2016-05-25 stevvooe if err != nil {
182 11b5e5c7 2016-05-25 stevvooe req.err <- err
183 11b5e5c7 2016-05-25 stevvooe continue
186 11b5e5c7 2016-05-25 stevvooe outstanding[selected] = req
187 11b5e5c7 2016-05-25 stevvooe fcall := newFcall(selected, req.message)
189 4b33cdd0 2015-11-30 stephen.d // TODO(stevvooe): Consider the case of requests that never
190 4b33cdd0 2015-11-30 stephen.d // receive a response. We need to remove the fcall context from
191 4b33cdd0 2015-11-30 stephen.d // the tag map and dealloc the tag. We may also want to send a
192 4b33cdd0 2015-11-30 stephen.d // flush for the tag.
193 4b33cdd0 2015-11-30 stephen.d if err := t.ch.WriteFcall(req.ctx, fcall); err != nil {
194 4b33cdd0 2015-11-30 stephen.d delete(outstanding, fcall.Tag)
195 4b33cdd0 2015-11-30 stephen.d req.err <- err
197 4b33cdd0 2015-11-30 stephen.d case b := <-responses:
198 4b33cdd0 2015-11-30 stephen.d req, ok := outstanding[b.Tag]
199 4b33cdd0 2015-11-30 stephen.d if !ok {
200 f52b8701 2016-05-19 stevvooe // BUG(stevvooe): The exact handling of an unknown tag is
201 f52b8701 2016-05-19 stevvooe // unclear at this point. These may not necessarily fatal to
202 f52b8701 2016-05-19 stevvooe // the session, since they could be messages that the client no
203 f52b8701 2016-05-19 stevvooe // longer cares for. When we figure this out, replace this
204 f52b8701 2016-05-19 stevvooe // panic with something more sensible.
205 f52b8701 2016-05-19 stevvooe panic(fmt.Sprintf("unknown tag received: %v", b))
208 4b33cdd0 2015-11-30 stephen.d // BUG(stevvooe): Must detect duplicate tag and ensure that we are
209 4b33cdd0 2015-11-30 stephen.d // waking up the right caller. If a duplicate is received, the
210 4b33cdd0 2015-11-30 stephen.d // entry should not be deleted.
211 4b33cdd0 2015-11-30 stephen.d delete(outstanding, b.Tag)
213 4b33cdd0 2015-11-30 stephen.d req.response <- b
215 4b33cdd0 2015-11-30 stephen.d // TODO(stevvooe): Reclaim tag id.
216 4b33cdd0 2015-11-30 stephen.d case <-t.ctx.Done():
217 4b33cdd0 2015-11-30 stephen.d return
218 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
219 4b33cdd0 2015-11-30 stephen.d return
224 4b33cdd0 2015-11-30 stephen.d func (t *transport) flush(ctx context.Context, tag Tag) error {
225 4b33cdd0 2015-11-30 stephen.d // TODO(stevvooe): We need to fire and forget flush messages when a call
226 4b33cdd0 2015-11-30 stephen.d // context gets cancelled.
227 4b33cdd0 2015-11-30 stephen.d panic("not implemented")
230 4b33cdd0 2015-11-30 stephen.d func (t *transport) Close() error {
231 4b33cdd0 2015-11-30 stephen.d select {
232 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
233 4b33cdd0 2015-11-30 stephen.d return ErrClosed
234 4b33cdd0 2015-11-30 stephen.d case <-t.ctx.Done():
235 4b33cdd0 2015-11-30 stephen.d return t.ctx.Err()
236 4b33cdd0 2015-11-30 stephen.d default:
237 4b33cdd0 2015-11-30 stephen.d close(t.closed)
240 4b33cdd0 2015-11-30 stephen.d return nil
