1 package p9p
2 
3 import (
4 	"fmt"
5 	"log"
6 	"net"
7 
8 	"golang.org/x/net/context"
9 )
10 
11 // roundTripper manages the request and response from the client-side. A
12 // roundTripper must abide by similar rules to the http.RoundTripper.
13 // Typically, the roundTripper will manage tag assignment and message
14 // serialization.
15 type roundTripper interface {
16 	send(ctx context.Context, msg Message) (Message, error)
17 }
18 
19 // transport plays the role of being a client channel manager. It multiplexes
20 // function calls onto the wire and dispatches responses to blocking calls to
21 // send. On the whole, transport is thread-safe for calling send
22 type transport struct {
23 	ctx      context.Context
24 	ch       Channel
25 	requests chan *fcallRequest
26 	closed   chan struct{}
27 
28 	tags uint16
29 }
30 
31 var _ roundTripper = &transport{}
32 
33 func newTransport(ctx context.Context, ch *channel) roundTripper {
34 	t := &transport{
35 		ctx:      ctx,
36 		ch:       ch,
37 		requests: make(chan *fcallRequest),
38 		closed:   make(chan struct{}),
39 	}
40 
41 	go t.handle()
42 
43 	return t
44 }
45 
46 type fcallRequest struct {
47 	ctx      context.Context
48 	fcall    *Fcall
49 	response chan *Fcall
50 	err      chan error
51 }
52 
53 func newFcallRequest(ctx context.Context, fcall *Fcall) *fcallRequest {
54 	return &fcallRequest{
55 		ctx:      ctx,
56 		fcall:    fcall,
57 		response: make(chan *Fcall, 1),
58 		err:      make(chan error, 1),
59 	}
60 }
61 
62 func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
63 	fcall := newFcall(msg)
64 	req := newFcallRequest(ctx, fcall)
65 
66 	// dispatch the request.
67 	select {
68 	case <-t.closed:
69 		return nil, ErrClosed
70 	case <-ctx.Done():
71 		return nil, ctx.Err()
72 	case t.requests <- req:
73 	}
74 
75 	// wait for the response.
76 	select {
77 	case <-t.closed:
78 		return nil, ErrClosed
79 	case <-ctx.Done():
80 		return nil, ctx.Err()
81 	case err := <-req.err:
82 		return nil, err
83 	case resp := <-req.response:
84 		if resp.Type == Rerror {
85 			// pack the error into something useful
86 			respmesg, ok := resp.Message.(MessageRerror)
87 			if !ok {
88 				return nil, fmt.Errorf("invalid error response: %v", resp)
89 			}
90 
91 			return nil, respmesg
92 		}
93 
94 		return resp.Message, nil
95 	}
96 }
97 
98 // handle takes messages off the wire and wakes up the waiting tag call.
99 func (t *transport) handle() {
100 	defer func() {
101 		log.Println("exited handle loop")
102 		t.Close()
103 	}()
104 	// the following variable block are protected components owned by this thread.
105 	var (
106 		responses = make(chan *Fcall)
107 		tags      Tag
108 		// outstanding provides a map of tags to outstanding requests.
109 		outstanding = map[Tag]*fcallRequest{}
110 	)
111 
112 	// loop to read messages off of the connection
113 	go func() {
114 		defer func() {
115 			log.Println("exited read loop")
116 			t.Close()
117 		}()
118 	loop:
119 		for {
120 			fcall := new(Fcall)
121 			if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
122 				switch err := err.(type) {
123 				case net.Error:
124 					if err.Timeout() || err.Temporary() {
125 						// BUG(stevvooe): There may be partial reads under
126 						// timeout errors where this is actually fatal.
127 
128 						// can only retry if we haven't offset the frame.
129 						continue loop
130 					}
131 				}
132 
133 				log.Println("fatal error reading msg:", err)
134 				t.Close()
135 				return
136 			}
137 
138 			select {
139 			case <-t.ctx.Done():
140 				log.Println("ctx done")
141 				return
142 			case <-t.closed:
143 				log.Println("transport closed")
144 				return
145 			case responses <- fcall:
146 			}
147 		}
148 	}()
149 
150 	for {
151 		log.Println("wait...")
152 		select {
153 		case req := <-t.requests:
154 			if req.fcall.Tag == NOTAG {
155 				// NOTE(stevvooe): We disallow fcalls with NOTAG to come
156 				// through this path since we can't join the tagged response
157 				// with the waiting caller. This is typically used for the
158 				// Tversion/Rversion round trip to setup a session.
159 				//
160 				// It may be better to allow these through but block all
161 				// requests until a notag message has a response.
162 
163 				req.err <- fmt.Errorf("disallowed tag through transport")
164 				continue
165 			}
166 
167 			// BUG(stevvooe): This is an awful tag allocation procedure.
168 			// Replace this with something that let's us allocate tags and
169 			// associate data with them, returning to them to a pool when
170 			// complete. Such a system would provide a lot of information
171 			// about outstanding requests.
172 			tags++
173 			req.fcall.Tag = tags
174 			outstanding[req.fcall.Tag] = req
175 
176 			// TODO(stevvooe): Consider the case of requests that never
177 			// receive a response. We need to remove the fcall context from
178 			// the tag map and dealloc the tag. We may also want to send a
179 			// flush for the tag.
180 			if err := t.ch.WriteFcall(req.ctx, req.fcall); err != nil {
181 				delete(outstanding, req.fcall.Tag)
182 				req.err <- err
183 			}
184 		case b := <-responses:
185 			req, ok := outstanding[b.Tag]
186 			if !ok {
187 				panic("unknown tag received")
188 			}
189 			delete(outstanding, req.fcall.Tag)
190 
191 			req.response <- b
192 
193 			// TODO(stevvooe): Reclaim tag id.
194 		case <-t.ctx.Done():
195 			return
196 		case <-t.closed:
197 			return
198 		}
199 	}
200 }
201 
202 func (t *transport) flush(ctx context.Context, tag Tag) error {
203 	// TODO(stevvooe): We need to fire and forget flush messages when a call
204 	// context gets cancelled.
205 	panic("not implemented")
206 }
207 
208 func (t *transport) Close() error {
209 	select {
210 	case <-t.closed:
211 		return ErrClosed
212 	case <-t.ctx.Done():
213 		return t.ctx.Err()
214 	default:
215 		close(t.closed)
216 	}
217 
218 	return nil
219 }