13 // roundTripper manages the request and response from the client-side. A
14 // roundTripper must abide by similar rules to the http.RoundTripper.
15 // Typically, the roundTripper will manage tag assignment and message
17 type roundTripper interface {
18 send(ctx context.Context, msg Message) (Message, error)
21 // transport plays the role of being a client channel manager. It multiplexes
22 // function calls onto the wire and dispatches responses to blocking calls to
23 // send. On the whole, transport is thread-safe for calling send
24 type transport struct {
27 requests chan *fcallRequest
29 shutdown chan struct{}
30 once sync.Once // protect closure of shutdown
36 var _ roundTripper = &transport{}
38 func newTransport(ctx context.Context, ch Channel) roundTripper {
42 requests: make(chan *fcallRequest),
43 shutdown: make(chan struct{}),
44 closed: make(chan struct{}),
52 // fcallRequest encompasses the request to send a message via fcall.
53 type fcallRequest struct {
60 func newFcallRequest(ctx context.Context, msg Message) *fcallRequest {
64 response: make(chan *Fcall, 1),
65 err: make(chan error, 1),
69 func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
70 req := newFcallRequest(ctx, msg)
72 // dispatch the request.
78 case t.requests <- req:
81 // wait for the response.
87 case err := <-req.err:
89 case resp := <-req.response:
90 if resp.Type == Rerror {
91 // pack the error into something useful
92 respmesg, ok := resp.Message.(MessageRerror)
94 return nil, fmt.Errorf("invalid error response: %v", resp)
100 return resp.Message, nil
104 // allocateTag returns a valid tag given a tag pool map. It receives a hint as
105 // to where to start the tag search. It returns an error if the allocation is
106 // not possible. The provided map must not contain NOTAG as a key.
107 func allocateTag(r *fcallRequest, m map[Tag]*fcallRequest, hint Tag) (Tag, error) {
108 // The tag pool is depleted if 65535 (0xFFFF) tags are taken.
109 if len(m) >= 0xFFFF {
110 return 0, errors.New("tag pool depleted")
113 // Look for the first tag that doesn't exist in the map and return it.
114 for i := 0; i < 0xFFFF; i++ {
120 if _, exists := m[hint]; !exists {
125 return 0, errors.New("allocateTag: unexpected error")
128 // handle takes messages off the wire and wakes up the waiting tag call.
129 func (t *transport) handle() {
131 log.Println("exited handle loop")
135 // the following variable block are protected components owned by this thread.
137 responses = make(chan *Fcall)
138 // outstanding provides a map of tags to outstanding requests.
139 outstanding = map[Tag]*fcallRequest{}
143 // loop to read messages off of the connection
146 log.Println("exited read loop")
147 t.close() // single main loop
152 if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
153 switch err := err.(type) {
155 if err.Timeout() || err.Temporary() {
156 // BUG(stevvooe): There may be partial reads under
157 // timeout errors where this is actually fatal.
159 // can only retry if we haven't offset the frame.
164 log.Println("fatal error reading msg:", err)
172 log.Println("transport closed")
174 case responses <- fcall:
181 case req := <-t.requests:
184 selected, err = allocateTag(req, outstanding, selected)
190 outstanding[selected] = req
191 fcall := newFcall(selected, req.message)
193 // TODO(stevvooe): Consider the case of requests that never
194 // receive a response. We need to remove the fcall context from
195 // the tag map and dealloc the tag. We may also want to send a
196 // flush for the tag.
197 if err := t.ch.WriteFcall(req.ctx, fcall); err != nil {
198 delete(outstanding, fcall.Tag)
201 case b := <-responses:
202 req, ok := outstanding[b.Tag]
204 // BUG(stevvooe): The exact handling of an unknown tag is
205 // unclear at this point. These may not necessarily fatal to
206 // the session, since they could be messages that the client no
207 // longer cares for. When we figure this out, replace this
208 // panic with something more sensible.
209 panic(fmt.Sprintf("unknown tag received: %v", b))
212 // BUG(stevvooe): Must detect duplicate tag and ensure that we are
213 // waking up the right caller. If a duplicate is received, the
214 // entry should not be deleted.
215 delete(outstanding, b.Tag)
219 // TODO(stevvooe): Reclaim tag id.
228 func (t *transport) flush(ctx context.Context, tag Tag) error {
229 // TODO(stevvooe): We need to fire and forget flush messages when a call
230 // context gets cancelled.
231 panic("not implemented")
234 func (t *transport) Close() error {
245 // close starts the shutdown process.
246 func (t *transport) close() {