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 func allocateTag(r *fcallRequest, m map[Tag]*fcallRequest) (Tag, error) {
100 	// Tversion can only use NOTAG, so check if we're sending a Tversion.
101 	if r.message.Type() == Tversion {
102 		if _, exists := m[NOTAG]; exists {
103 			return 0, errors.New("NOTAG already in use")
104 		}
105 		return NOTAG, nil
106 	}
107 
108 	// The tag pool is depleted if all 65536 tags are taken, or if 65535 tags
109 	// are taken and NOTAG is available.
110 	if len(m) > 0xFFFF {
111 		return 0, errors.New("tag pool depleted")
112 	} else if len(m) == 0xFFFF {
113 		if _, exists := m[NOTAG]; !exists {
114 			return 0, errors.New("tag pool depleted")
115 		}
116 	}
117 
118 	// Look for the first tag that doesn't exist in the map and return it.
119 	for selected := Tag(0); selected < NOTAG; selected++ {
120 		if _, exists := m[selected]; !exists {
121 			return selected, nil
122 		}
123 	}
124 
125 	return 0, errors.New("allocateTag: unexpected error")
126 }
127 
128 // handle takes messages off the wire and wakes up the waiting tag call.
129 func (t *transport) handle() {
130 	defer func() {
131 		log.Println("exited handle loop")
132 		t.Close()
133 	}()
134 	// the following variable block are protected components owned by this thread.
135 	var (
136 		responses = make(chan *Fcall)
137 		// outstanding provides a map of tags to outstanding requests.
138 		outstanding = map[Tag]*fcallRequest{}
139 	)
140 
141 	// loop to read messages off of the connection
142 	go func() {
143 		defer func() {
144 			log.Println("exited read loop")
145 			t.Close()
146 		}()
147 	loop:
148 		for {
149 			fcall := new(Fcall)
150 			if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
151 				switch err := err.(type) {
152 				case net.Error:
153 					if err.Timeout() || err.Temporary() {
154 						// BUG(stevvooe): There may be partial reads under
155 						// timeout errors where this is actually fatal.
156 
157 						// can only retry if we haven't offset the frame.
158 						continue loop
159 					}
160 				}
161 
162 				log.Println("fatal error reading msg:", err)
163 				t.Close()
164 				return
165 			}
166 
167 			select {
168 			case <-t.ctx.Done():
169 				log.Println("ctx done")
170 				return
171 			case <-t.closed:
172 				log.Println("transport closed")
173 				return
174 			case responses <- fcall:
175 			}
176 		}
177 	}()
178 
179 	for {
180 		select {
181 		case req := <-t.requests:
182 			selected, err := allocateTag(req, outstanding)
183 			if err != nil {
184 				req.err <- err
185 				continue
186 			}
187 
188 			outstanding[selected] = req
189 			fcall := newFcall(selected, req.message)
190 
191 			// TODO(stevvooe): Consider the case of requests that never
192 			// receive a response. We need to remove the fcall context from
193 			// the tag map and dealloc the tag. We may also want to send a
194 			// flush for the tag.
195 			if err := t.ch.WriteFcall(req.ctx, fcall); err != nil {
196 				delete(outstanding, fcall.Tag)
197 				req.err <- err
198 			}
199 		case b := <-responses:
200 			req, ok := outstanding[b.Tag]
201 			if !ok {
202 				// BUG(stevvooe): The exact handling of an unknown tag is
203 				// unclear at this point. These may not necessarily fatal to
204 				// the session, since they could be messages that the client no
205 				// longer cares for. When we figure this out, replace this
206 				// panic with something more sensible.
207 				panic(fmt.Sprintf("unknown tag received: %v", b))
208 			}
209 
210 			// BUG(stevvooe): Must detect duplicate tag and ensure that we are
211 			// waking up the right caller. If a duplicate is received, the
212 			// entry should not be deleted.
213 			delete(outstanding, b.Tag)
214 
215 			req.response <- b
216 
217 			// TODO(stevvooe): Reclaim tag id.
218 		case <-t.ctx.Done():
219 			return
220 		case <-t.closed:
221 			return
222 		}
223 	}
224 }
225 
226 func (t *transport) flush(ctx context.Context, tag Tag) error {
227 	// TODO(stevvooe): We need to fire and forget flush messages when a call
228 	// context gets cancelled.
229 	panic("not implemented")
230 }
231 
232 func (t *transport) Close() error {
233 	select {
234 	case <-t.closed:
235 		return ErrClosed
236 	case <-t.ctx.Done():
237 		return t.ctx.Err()
238 	default:
239 		close(t.closed)
240 	}
241 
242 	return nil
243 }