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