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