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 // fcallRequest encompasses the request to send a message via fcall.
47 type fcallRequest struct {
48 	ctx      context.Context
49 	message  Message
50 	response chan *Fcall
51 	err      chan error
52 }
53 
54 func newFcallRequest(ctx context.Context, msg Message) *fcallRequest {
55 	return &fcallRequest{
56 		ctx:      ctx,
57 		message:  msg,
58 		response: make(chan *Fcall, 1),
59 		err:      make(chan error, 1),
60 	}
61 }
62 
63 func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
64 	req := newFcallRequest(ctx, msg)
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 		select {
152 		case req := <-t.requests:
153 			// BUG(stevvooe): This is an awful tag allocation procedure.
154 			// Replace this with something that let's us allocate tags and
155 			// associate data with them, returning to them to a pool when
156 			// complete. Such a system would provide a lot of information
157 			// about outstanding requests.
158 			tags++
159 			fcall := newFcall(tags, req.message)
160 			outstanding[fcall.Tag] = req
161 
162 			// TODO(stevvooe): Consider the case of requests that never
163 			// receive a response. We need to remove the fcall context from
164 			// the tag map and dealloc the tag. We may also want to send a
165 			// flush for the tag.
166 			if err := t.ch.WriteFcall(req.ctx, fcall); err != nil {
167 				delete(outstanding, fcall.Tag)
168 				req.err <- err
169 			}
170 		case b := <-responses:
171 			req, ok := outstanding[b.Tag]
172 			if !ok {
173 				panic("unknown tag received")
174 			}
175 
176 			// BUG(stevvooe): Must detect duplicate tag and ensure that we are
177 			// waking up the right caller. If a duplicate is received, the
178 			// entry should not be deleted.
179 			delete(outstanding, b.Tag)
180 
181 			req.response <- b
182 
183 			// TODO(stevvooe): Reclaim tag id.
184 		case <-t.ctx.Done():
185 			return
186 		case <-t.closed:
187 			return
188 		}
189 	}
190 }
191 
192 func (t *transport) flush(ctx context.Context, tag Tag) error {
193 	// TODO(stevvooe): We need to fire and forget flush messages when a call
194 	// context gets cancelled.
195 	panic("not implemented")
196 }
197 
198 func (t *transport) Close() error {
199 	select {
200 	case <-t.closed:
201 		return ErrClosed
202 	case <-t.ctx.Done():
203 		return t.ctx.Err()
204 	default:
205 		close(t.closed)
206 	}
207 
208 	return nil
209 }