1 4b33cdd0 2015-11-30 stephen.d package p9p
3 4b33cdd0 2015-11-30 stephen.d import (
10 529e2b2e 2016-11-14 noreply "context"
13 4b33cdd0 2015-11-30 stephen.d // roundTripper manages the request and response from the client-side. A
14 4b33cdd0 2015-11-30 stephen.d // roundTripper must abide by similar rules to the http.RoundTripper.
15 4b33cdd0 2015-11-30 stephen.d // Typically, the roundTripper will manage tag assignment and message
16 4b33cdd0 2015-11-30 stephen.d // serialization.
17 4b33cdd0 2015-11-30 stephen.d type roundTripper interface {
18 4b33cdd0 2015-11-30 stephen.d send(ctx context.Context, msg Message) (Message, error)
21 4b33cdd0 2015-11-30 stephen.d // transport plays the role of being a client channel manager. It multiplexes
22 4b33cdd0 2015-11-30 stephen.d // function calls onto the wire and dispatches responses to blocking calls to
23 4b33cdd0 2015-11-30 stephen.d // send. On the whole, transport is thread-safe for calling send
24 4b33cdd0 2015-11-30 stephen.d type transport struct {
25 4b33cdd0 2015-11-30 stephen.d ctx context.Context
26 4b33cdd0 2015-11-30 stephen.d ch Channel
27 4b33cdd0 2015-11-30 stephen.d requests chan *fcallRequest
29 b9021eb7 2016-09-19 noreply shutdown chan struct{}
30 b9021eb7 2016-09-19 noreply once sync.Once // protect closure of shutdown
31 4b33cdd0 2015-11-30 stephen.d closed chan struct{}
33 4b33cdd0 2015-11-30 stephen.d tags uint16
36 4b33cdd0 2015-11-30 stephen.d var _ roundTripper = &transport{}
38 20882a7b 2016-05-25 stevvooe func newTransport(ctx context.Context, ch Channel) roundTripper {
39 4b33cdd0 2015-11-30 stephen.d t := &transport{
40 4b33cdd0 2015-11-30 stephen.d ctx: ctx,
41 4b33cdd0 2015-11-30 stephen.d ch: ch,
42 4b33cdd0 2015-11-30 stephen.d requests: make(chan *fcallRequest),
43 b9021eb7 2016-09-19 noreply shutdown: make(chan struct{}),
44 4b33cdd0 2015-11-30 stephen.d closed: make(chan struct{}),
47 4b33cdd0 2015-11-30 stephen.d go t.handle()
49 4b33cdd0 2015-11-30 stephen.d return t
52 4b33cdd0 2015-11-30 stephen.d // fcallRequest encompasses the request to send a message via fcall.
53 4b33cdd0 2015-11-30 stephen.d type fcallRequest struct {
54 4b33cdd0 2015-11-30 stephen.d ctx context.Context
55 4b33cdd0 2015-11-30 stephen.d message Message
56 4b33cdd0 2015-11-30 stephen.d response chan *Fcall
57 4b33cdd0 2015-11-30 stephen.d err chan error
60 4b33cdd0 2015-11-30 stephen.d func newFcallRequest(ctx context.Context, msg Message) *fcallRequest {
61 4b33cdd0 2015-11-30 stephen.d return &fcallRequest{
62 4b33cdd0 2015-11-30 stephen.d ctx: ctx,
63 4b33cdd0 2015-11-30 stephen.d message: msg,
64 4b33cdd0 2015-11-30 stephen.d response: make(chan *Fcall, 1),
65 4b33cdd0 2015-11-30 stephen.d err: make(chan error, 1),
69 4b33cdd0 2015-11-30 stephen.d func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
70 4b33cdd0 2015-11-30 stephen.d req := newFcallRequest(ctx, msg)
72 4b33cdd0 2015-11-30 stephen.d // dispatch the request.
73 4b33cdd0 2015-11-30 stephen.d select {
74 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
75 4b33cdd0 2015-11-30 stephen.d return nil, ErrClosed
76 4b33cdd0 2015-11-30 stephen.d case <-ctx.Done():
77 4b33cdd0 2015-11-30 stephen.d return nil, ctx.Err()
78 4b33cdd0 2015-11-30 stephen.d case t.requests <- req:
81 4b33cdd0 2015-11-30 stephen.d // wait for the response.
82 4b33cdd0 2015-11-30 stephen.d select {
83 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
84 4b33cdd0 2015-11-30 stephen.d return nil, ErrClosed
85 4b33cdd0 2015-11-30 stephen.d case <-ctx.Done():
86 4b33cdd0 2015-11-30 stephen.d return nil, ctx.Err()
87 4b33cdd0 2015-11-30 stephen.d case err := <-req.err:
88 4b33cdd0 2015-11-30 stephen.d return nil, err
89 4b33cdd0 2015-11-30 stephen.d case resp := <-req.response:
90 4b33cdd0 2015-11-30 stephen.d if resp.Type == Rerror {
91 4b33cdd0 2015-11-30 stephen.d // pack the error into something useful
92 4b33cdd0 2015-11-30 stephen.d respmesg, ok := resp.Message.(MessageRerror)
93 4b33cdd0 2015-11-30 stephen.d if !ok {
94 4b33cdd0 2015-11-30 stephen.d return nil, fmt.Errorf("invalid error response: %v", resp)
97 4b33cdd0 2015-11-30 stephen.d return nil, respmesg
100 4b33cdd0 2015-11-30 stephen.d return resp.Message, nil
104 11b5e5c7 2016-05-25 stevvooe // allocateTag returns a valid tag given a tag pool map. It receives a hint as
105 11b5e5c7 2016-05-25 stevvooe // to where to start the tag search. It returns an error if the allocation is
106 11b5e5c7 2016-05-25 stevvooe // not possible. The provided map must not contain NOTAG as a key.
107 11b5e5c7 2016-05-25 stevvooe func allocateTag(r *fcallRequest, m map[Tag]*fcallRequest, hint Tag) (Tag, error) {
108 11b5e5c7 2016-05-25 stevvooe // The tag pool is depleted if 65535 (0xFFFF) tags are taken.
109 11b5e5c7 2016-05-25 stevvooe if len(m) >= 0xFFFF {
110 11b5e5c7 2016-05-25 stevvooe return 0, errors.New("tag pool depleted")
113 11b5e5c7 2016-05-25 stevvooe // Look for the first tag that doesn't exist in the map and return it.
114 11b5e5c7 2016-05-25 stevvooe for i := 0; i < 0xFFFF; i++ {
116 11b5e5c7 2016-05-25 stevvooe if hint == NOTAG {
117 11b5e5c7 2016-05-25 stevvooe hint = 0
120 11b5e5c7 2016-05-25 stevvooe if _, exists := m[hint]; !exists {
121 11b5e5c7 2016-05-25 stevvooe return hint, nil
125 11b5e5c7 2016-05-25 stevvooe return 0, errors.New("allocateTag: unexpected error")
128 4b33cdd0 2015-11-30 stephen.d // handle takes messages off the wire and wakes up the waiting tag call.
129 4b33cdd0 2015-11-30 stephen.d func (t *transport) handle() {
130 4b33cdd0 2015-11-30 stephen.d defer func() {
131 b9021eb7 2016-09-19 noreply close(t.closed)
134 4b33cdd0 2015-11-30 stephen.d // the following variable block are protected components owned by this thread.
136 4b33cdd0 2015-11-30 stephen.d responses = make(chan *Fcall)
137 4b33cdd0 2015-11-30 stephen.d // outstanding provides a map of tags to outstanding requests.
138 4b33cdd0 2015-11-30 stephen.d outstanding = map[Tag]*fcallRequest{}
139 11b5e5c7 2016-05-25 stevvooe selected Tag
142 4b33cdd0 2015-11-30 stephen.d // loop to read messages off of the connection
143 4b33cdd0 2015-11-30 stephen.d go func() {
144 4b33cdd0 2015-11-30 stephen.d defer func() {
145 b9021eb7 2016-09-19 noreply t.close() // single main loop
149 4b33cdd0 2015-11-30 stephen.d fcall := new(Fcall)
150 4b33cdd0 2015-11-30 stephen.d if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
151 4b33cdd0 2015-11-30 stephen.d switch err := err.(type) {
152 4b33cdd0 2015-11-30 stephen.d case net.Error:
153 4b33cdd0 2015-11-30 stephen.d if err.Timeout() || err.Temporary() {
154 4b33cdd0 2015-11-30 stephen.d // BUG(stevvooe): There may be partial reads under
155 4b33cdd0 2015-11-30 stephen.d // timeout errors where this is actually fatal.
157 4b33cdd0 2015-11-30 stephen.d // can only retry if we haven't offset the frame.
158 4b33cdd0 2015-11-30 stephen.d continue loop
162 2da28ffa 2017-11-10 noreply log.Println("p9p: fatal error reading msg:", err)
163 4b33cdd0 2015-11-30 stephen.d return
166 4b33cdd0 2015-11-30 stephen.d select {
167 4b33cdd0 2015-11-30 stephen.d case <-t.ctx.Done():
168 4b33cdd0 2015-11-30 stephen.d return
169 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
170 4b33cdd0 2015-11-30 stephen.d return
171 4b33cdd0 2015-11-30 stephen.d case responses <- fcall:
177 4b33cdd0 2015-11-30 stephen.d select {
178 4b33cdd0 2015-11-30 stephen.d case req := <-t.requests:
179 11b5e5c7 2016-05-25 stevvooe var err error
181 11b5e5c7 2016-05-25 stevvooe selected, err = allocateTag(req, outstanding, selected)
182 11b5e5c7 2016-05-25 stevvooe if err != nil {
183 11b5e5c7 2016-05-25 stevvooe req.err <- err
184 11b5e5c7 2016-05-25 stevvooe continue
187 11b5e5c7 2016-05-25 stevvooe outstanding[selected] = req
188 11b5e5c7 2016-05-25 stevvooe fcall := newFcall(selected, req.message)
190 4b33cdd0 2015-11-30 stephen.d // TODO(stevvooe): Consider the case of requests that never
191 4b33cdd0 2015-11-30 stephen.d // receive a response. We need to remove the fcall context from
192 4b33cdd0 2015-11-30 stephen.d // the tag map and dealloc the tag. We may also want to send a
193 4b33cdd0 2015-11-30 stephen.d // flush for the tag.
194 4b33cdd0 2015-11-30 stephen.d if err := t.ch.WriteFcall(req.ctx, fcall); err != nil {
195 4b33cdd0 2015-11-30 stephen.d delete(outstanding, fcall.Tag)
196 4b33cdd0 2015-11-30 stephen.d req.err <- err
198 4b33cdd0 2015-11-30 stephen.d case b := <-responses:
199 4b33cdd0 2015-11-30 stephen.d req, ok := outstanding[b.Tag]
200 4b33cdd0 2015-11-30 stephen.d if !ok {
201 f52b8701 2016-05-19 stevvooe // BUG(stevvooe): The exact handling of an unknown tag is
202 f52b8701 2016-05-19 stevvooe // unclear at this point. These may not necessarily fatal to
203 f52b8701 2016-05-19 stevvooe // the session, since they could be messages that the client no
204 f52b8701 2016-05-19 stevvooe // longer cares for. When we figure this out, replace this
205 f52b8701 2016-05-19 stevvooe // panic with something more sensible.
206 f52b8701 2016-05-19 stevvooe panic(fmt.Sprintf("unknown tag received: %v", b))
209 4b33cdd0 2015-11-30 stephen.d // BUG(stevvooe): Must detect duplicate tag and ensure that we are
210 4b33cdd0 2015-11-30 stephen.d // waking up the right caller. If a duplicate is received, the
211 4b33cdd0 2015-11-30 stephen.d // entry should not be deleted.
212 4b33cdd0 2015-11-30 stephen.d delete(outstanding, b.Tag)
214 4b33cdd0 2015-11-30 stephen.d req.response <- b
216 4b33cdd0 2015-11-30 stephen.d // TODO(stevvooe): Reclaim tag id.
217 b9021eb7 2016-09-19 noreply case <-t.shutdown:
219 4b33cdd0 2015-11-30 stephen.d case <-t.ctx.Done():
220 4b33cdd0 2015-11-30 stephen.d return
225 4b33cdd0 2015-11-30 stephen.d func (t *transport) flush(ctx context.Context, tag Tag) error {
226 4b33cdd0 2015-11-30 stephen.d // TODO(stevvooe): We need to fire and forget flush messages when a call
227 4b33cdd0 2015-11-30 stephen.d // context gets cancelled.
228 4b33cdd0 2015-11-30 stephen.d panic("not implemented")
231 4b33cdd0 2015-11-30 stephen.d func (t *transport) Close() error {
232 b9021eb7 2016-09-19 noreply t.close()
234 4b33cdd0 2015-11-30 stephen.d select {
235 4b33cdd0 2015-11-30 stephen.d case <-t.closed:
236 b9021eb7 2016-09-19 noreply return nil
237 4b33cdd0 2015-11-30 stephen.d case <-t.ctx.Done():
238 4b33cdd0 2015-11-30 stephen.d return t.ctx.Err()
242 b9021eb7 2016-09-19 noreply // close starts the shutdown process.
243 b9021eb7 2016-09-19 noreply func (t *transport) close() {
244 b9021eb7 2016-09-19 noreply t.once.Do(func() {
245 b9021eb7 2016-09-19 noreply close(t.shutdown)
