1 /*
2 * Multiplexed Venti client. It would be nice if we
3 * could turn this into a generic library routine rather
4 * than keep it Venti specific. A user-level 9P client
5 * could use something like this too.
6 *
7 * (Actually it does - this should be replaced with libmux,
8 * which should be renamed librpcmux.)
9 *
10 * This is a little more complicated than it might be
11 * because we want it to work well within and without libthread.
12 *
13 * The mux code is inspired by tra's, which is inspired by the Plan 9 kernel.
14 */
15
16 #include <u.h>
17 #include <libc.h>
18 #include <venti.h>
19
20 typedef struct Rwait Rwait;
21 struct Rwait
22 {
23 Rendez r;
24 Packet *p;
25 int done;
26 int sleeping;
27 };
28
29 static int gettag(VtConn*, Rwait*);
30 static void puttag(VtConn*, Rwait*, int);
31 static void muxrpc(VtConn*, Packet*);
32
33 Packet*
_vtrpc(VtConn * z,Packet * p,VtFcall * tx)34 _vtrpc(VtConn *z, Packet *p, VtFcall *tx)
35 {
36 int i;
37 uchar tag, buf[2], *top;
38 Rwait *r, *rr;
39
40 if(z == nil){
41 werrstr("not connected");
42 packetfree(p);
43 return nil;
44 }
45
46 /* must malloc because stack could be private */
47 r = vtmallocz(sizeof(Rwait));
48
49 qlock(&z->lk);
50 r->r.l = &z->lk;
51 tag = gettag(z, r);
52 if(tx){
53 /* vtfcallrpc can't print packet because it doesn't have tag */
54 tx->tag = tag;
55 if(chattyventi)
56 fprint(2, "%s -> %F\n", argv0, tx);
57 }
58
59 /* slam tag into packet */
60 top = packetpeek(p, buf, 0, 2);
61 if(top == nil){
62 packetfree(p);
63 return nil;
64 }
65 if(top == buf){
66 werrstr("first two bytes must be in same packet fragment");
67 packetfree(p);
68 vtfree(r);
69 return nil;
70 }
71 top[1] = tag;
72 qunlock(&z->lk);
73 if(vtsend(z, p) < 0){
74 vtfree(r);
75 return nil;
76 }
77
78 qlock(&z->lk);
79 /* wait for the muxer to give us our packet */
80 r->sleeping = 1;
81 z->nsleep++;
82 while(z->muxer && !r->done)
83 rsleep(&r->r);
84 z->nsleep--;
85 r->sleeping = 0;
86
87 /* if not done, there's no muxer: start muxing */
88 if(!r->done){
89 if(z->muxer)
90 abort();
91 z->muxer = 1;
92 while(!r->done){
93 qunlock(&z->lk);
94 if((p = vtrecv(z)) == nil){
95 werrstr("unexpected eof on venti connection");
96 z->muxer = 0;
97 vtfree(r);
98 return nil;
99 }
100 qlock(&z->lk);
101 muxrpc(z, p);
102 }
103 z->muxer = 0;
104 /* if there is anyone else sleeping, wake first unfinished to mux */
105 if(z->nsleep)
106 for(i=0; i<256; i++){
107 rr = z->wait[i];
108 if(rr && rr->sleeping && !rr->done){
109 rwakeup(&rr->r);
110 break;
111 }
112 }
113 }
114
115 p = r->p;
116 puttag(z, r, tag);
117 vtfree(r);
118 qunlock(&z->lk);
119 return p;
120 }
121
122 Packet*
vtrpc(VtConn * z,Packet * p)123 vtrpc(VtConn *z, Packet *p)
124 {
125 return _vtrpc(z, p, nil);
126 }
127
128 static int
gettag(VtConn * z,Rwait * r)129 gettag(VtConn *z, Rwait *r)
130 {
131 int i;
132
133 Again:
134 while(z->ntag == 256)
135 rsleep(&z->tagrend);
136 for(i=0; i<256; i++)
137 if(z->wait[i] == 0){
138 z->ntag++;
139 z->wait[i] = r;
140 return i;
141 }
142 fprint(2, "libventi: ntag botch\n");
143 goto Again;
144 }
145
146 static void
puttag(VtConn * z,Rwait * r,int tag)147 puttag(VtConn *z, Rwait *r, int tag)
148 {
149 assert(z->wait[tag] == r);
150 z->wait[tag] = nil;
151 z->ntag--;
152 rwakeup(&z->tagrend);
153 }
154
155 static void
muxrpc(VtConn * z,Packet * p)156 muxrpc(VtConn *z, Packet *p)
157 {
158 uchar tag, buf[2], *top;
159 Rwait *r;
160
161 if((top = packetpeek(p, buf, 0, 2)) == nil){
162 fprint(2, "libventi: short packet in vtrpc\n");
163 packetfree(p);
164 return;
165 }
166
167 tag = top[1];
168 if((r = z->wait[tag]) == nil){
169 fprint(2, "libventi: unexpected packet tag %d in vtrpc\n", tag);
170 abort();
171 packetfree(p);
172 return;
173 }
174
175 r->p = p;
176 r->done = 1;
177 rwakeup(&r->r);
178 }
179
180