2 056fe1ba 2003-11-23 devnull * Multiplexed Venti client. It would be nice if we
3 056fe1ba 2003-11-23 devnull * could turn this into a generic library routine rather
4 056fe1ba 2003-11-23 devnull * than keep it Venti specific. A user-level 9P client
5 056fe1ba 2003-11-23 devnull * could use something like this too.
7 a09e80f9 2004-05-23 devnull * (Actually it does - this should be replaced with libmux,
8 a09e80f9 2004-05-23 devnull * which should be renamed librpcmux.)
10 056fe1ba 2003-11-23 devnull * This is a little more complicated than it might be
11 056fe1ba 2003-11-23 devnull * because we want it to work well within and without libthread.
13 056fe1ba 2003-11-23 devnull * The mux code is inspired by tra's, which is inspired by the Plan 9 kernel.
16 056fe1ba 2003-11-23 devnull #include <u.h>
17 056fe1ba 2003-11-23 devnull #include <libc.h>
18 056fe1ba 2003-11-23 devnull #include <venti.h>
20 056fe1ba 2003-11-23 devnull typedef struct Rwait Rwait;
21 056fe1ba 2003-11-23 devnull struct Rwait
23 056fe1ba 2003-11-23 devnull Rendez r;
24 056fe1ba 2003-11-23 devnull Packet *p;
25 056fe1ba 2003-11-23 devnull int done;
26 056fe1ba 2003-11-23 devnull int sleeping;
29 056fe1ba 2003-11-23 devnull static int gettag(VtConn*, Rwait*);
30 056fe1ba 2003-11-23 devnull static void puttag(VtConn*, Rwait*, int);
31 056fe1ba 2003-11-23 devnull static void muxrpc(VtConn*, Packet*);
34 7252036f 2005-11-02 devnull _vtrpc(VtConn *z, Packet *p, VtFcall *tx)
37 056fe1ba 2003-11-23 devnull uchar tag, buf[2], *top;
38 5ddc97fc 2005-02-14 devnull Rwait *r, *rr;
40 056fe1ba 2003-11-23 devnull /* must malloc because stack could be private */
41 056fe1ba 2003-11-23 devnull r = vtmallocz(sizeof(Rwait));
43 056fe1ba 2003-11-23 devnull qlock(&z->lk);
44 056fe1ba 2003-11-23 devnull r->r.l = &z->lk;
45 056fe1ba 2003-11-23 devnull tag = gettag(z, r);
47 7252036f 2005-11-02 devnull /* vtfcallrpc can't print packet because it doesn't have tag */
48 7252036f 2005-11-02 devnull tx->tag = tag;
49 7252036f 2005-11-02 devnull if(chattyventi)
50 7252036f 2005-11-02 devnull fprint(2, "%s -> %F\n", argv0, tx);
53 056fe1ba 2003-11-23 devnull /* slam tag into packet */
54 056fe1ba 2003-11-23 devnull top = packetpeek(p, buf, 0, 2);
55 056fe1ba 2003-11-23 devnull if(top == nil){
56 056fe1ba 2003-11-23 devnull packetfree(p);
57 056fe1ba 2003-11-23 devnull return nil;
59 056fe1ba 2003-11-23 devnull if(top == buf){
60 056fe1ba 2003-11-23 devnull werrstr("first two bytes must be in same packet fragment");
61 056fe1ba 2003-11-23 devnull packetfree(p);
62 056fe1ba 2003-11-23 devnull return nil;
64 056fe1ba 2003-11-23 devnull top[1] = tag;
65 056fe1ba 2003-11-23 devnull qunlock(&z->lk);
66 056fe1ba 2003-11-23 devnull if(vtsend(z, p) < 0)
67 056fe1ba 2003-11-23 devnull return nil;
69 056fe1ba 2003-11-23 devnull qlock(&z->lk);
70 056fe1ba 2003-11-23 devnull /* wait for the muxer to give us our packet */
71 056fe1ba 2003-11-23 devnull r->sleeping = 1;
72 056fe1ba 2003-11-23 devnull z->nsleep++;
73 056fe1ba 2003-11-23 devnull while(z->muxer && !r->done)
74 056fe1ba 2003-11-23 devnull rsleep(&r->r);
75 056fe1ba 2003-11-23 devnull z->nsleep--;
76 056fe1ba 2003-11-23 devnull r->sleeping = 0;
78 056fe1ba 2003-11-23 devnull /* if not done, there's no muxer: start muxing */
79 056fe1ba 2003-11-23 devnull if(!r->done){
80 056fe1ba 2003-11-23 devnull if(z->muxer)
82 056fe1ba 2003-11-23 devnull z->muxer = 1;
83 056fe1ba 2003-11-23 devnull while(!r->done){
84 056fe1ba 2003-11-23 devnull qunlock(&z->lk);
85 056fe1ba 2003-11-23 devnull if((p = vtrecv(z)) == nil){
86 2e965b33 2004-05-05 devnull werrstr("unexpected eof on venti connection");
87 056fe1ba 2003-11-23 devnull z->muxer = 0;
88 056fe1ba 2003-11-23 devnull return nil;
90 056fe1ba 2003-11-23 devnull qlock(&z->lk);
91 056fe1ba 2003-11-23 devnull muxrpc(z, p);
93 056fe1ba 2003-11-23 devnull z->muxer = 0;
94 5ddc97fc 2005-02-14 devnull /* if there is anyone else sleeping, wake first unfinished to mux */
95 5ddc97fc 2005-02-14 devnull if(z->nsleep)
96 5ddc97fc 2005-02-14 devnull for(i=0; i<256; i++){
97 5ddc97fc 2005-02-14 devnull rr = z->wait[i];
98 5ddc97fc 2005-02-14 devnull if(rr && rr->sleeping && !rr->done){
99 5ddc97fc 2005-02-14 devnull rwakeup(&rr->r);
105 056fe1ba 2003-11-23 devnull p = r->p;
106 056fe1ba 2003-11-23 devnull puttag(z, r, tag);
107 056fe1ba 2003-11-23 devnull vtfree(r);
108 056fe1ba 2003-11-23 devnull qunlock(&z->lk);
109 056fe1ba 2003-11-23 devnull return p;
113 7252036f 2005-11-02 devnull vtrpc(VtConn *z, Packet *p)
115 7252036f 2005-11-02 devnull return _vtrpc(z, p, nil);
118 056fe1ba 2003-11-23 devnull static int
119 056fe1ba 2003-11-23 devnull gettag(VtConn *z, Rwait *r)
124 056fe1ba 2003-11-23 devnull while(z->ntag == 256)
125 056fe1ba 2003-11-23 devnull rsleep(&z->tagrend);
126 056fe1ba 2003-11-23 devnull for(i=0; i<256; i++)
127 056fe1ba 2003-11-23 devnull if(z->wait[i] == 0){
128 056fe1ba 2003-11-23 devnull z->ntag++;
129 056fe1ba 2003-11-23 devnull z->wait[i] = r;
130 056fe1ba 2003-11-23 devnull return i;
132 056fe1ba 2003-11-23 devnull fprint(2, "libventi: ntag botch\n");
133 056fe1ba 2003-11-23 devnull goto Again;
136 056fe1ba 2003-11-23 devnull static void
137 056fe1ba 2003-11-23 devnull puttag(VtConn *z, Rwait *r, int tag)
139 056fe1ba 2003-11-23 devnull assert(z->wait[tag] == r);
140 056fe1ba 2003-11-23 devnull z->wait[tag] = nil;
141 056fe1ba 2003-11-23 devnull z->ntag--;
142 056fe1ba 2003-11-23 devnull rwakeup(&z->tagrend);
145 056fe1ba 2003-11-23 devnull static void
146 056fe1ba 2003-11-23 devnull muxrpc(VtConn *z, Packet *p)
148 056fe1ba 2003-11-23 devnull uchar tag, buf[2], *top;
149 056fe1ba 2003-11-23 devnull Rwait *r;
151 056fe1ba 2003-11-23 devnull if((top = packetpeek(p, buf, 0, 2)) == nil){
152 056fe1ba 2003-11-23 devnull fprint(2, "libventi: short packet in vtrpc\n");
153 056fe1ba 2003-11-23 devnull packetfree(p);
157 056fe1ba 2003-11-23 devnull tag = top[1];
158 056fe1ba 2003-11-23 devnull if((r = z->wait[tag]) == nil){
159 056fe1ba 2003-11-23 devnull fprint(2, "libventi: unexpected packet tag %d in vtrpc\n", tag);
160 056fe1ba 2003-11-23 devnull abort();
161 056fe1ba 2003-11-23 devnull packetfree(p);
165 056fe1ba 2003-11-23 devnull r->p = p;
166 056fe1ba 2003-11-23 devnull r->done = 1;
167 056fe1ba 2003-11-23 devnull rwakeup(&r->r);
