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*);
32 056fe1ba 2003-11-23 devnull Packet *vtrpc(VtConn*, Packet*);
35 056fe1ba 2003-11-23 devnull vtrpc(VtConn *z, Packet *p)
38 056fe1ba 2003-11-23 devnull uchar tag, buf[2], *top;
39 056fe1ba 2003-11-23 devnull Rwait *r;
41 056fe1ba 2003-11-23 devnull /* must malloc because stack could be private */
42 056fe1ba 2003-11-23 devnull r = vtmallocz(sizeof(Rwait));
44 056fe1ba 2003-11-23 devnull qlock(&z->lk);
45 056fe1ba 2003-11-23 devnull r->r.l = &z->lk;
46 056fe1ba 2003-11-23 devnull tag = gettag(z, r);
48 056fe1ba 2003-11-23 devnull /* slam tag into packet */
49 056fe1ba 2003-11-23 devnull top = packetpeek(p, buf, 0, 2);
50 056fe1ba 2003-11-23 devnull if(top == nil){
51 056fe1ba 2003-11-23 devnull packetfree(p);
52 056fe1ba 2003-11-23 devnull return nil;
54 056fe1ba 2003-11-23 devnull if(top == buf){
55 056fe1ba 2003-11-23 devnull werrstr("first two bytes must be in same packet fragment");
56 056fe1ba 2003-11-23 devnull packetfree(p);
57 056fe1ba 2003-11-23 devnull return nil;
59 056fe1ba 2003-11-23 devnull top[1] = tag;
60 056fe1ba 2003-11-23 devnull qunlock(&z->lk);
61 056fe1ba 2003-11-23 devnull if(vtsend(z, p) < 0)
62 056fe1ba 2003-11-23 devnull return nil;
64 056fe1ba 2003-11-23 devnull qlock(&z->lk);
65 056fe1ba 2003-11-23 devnull /* wait for the muxer to give us our packet */
66 056fe1ba 2003-11-23 devnull r->sleeping = 1;
67 056fe1ba 2003-11-23 devnull z->nsleep++;
68 056fe1ba 2003-11-23 devnull while(z->muxer && !r->done)
69 056fe1ba 2003-11-23 devnull rsleep(&r->r);
70 056fe1ba 2003-11-23 devnull z->nsleep--;
71 056fe1ba 2003-11-23 devnull r->sleeping = 0;
73 056fe1ba 2003-11-23 devnull /* if not done, there's no muxer: start muxing */
74 056fe1ba 2003-11-23 devnull if(!r->done){
75 056fe1ba 2003-11-23 devnull if(z->muxer)
77 056fe1ba 2003-11-23 devnull z->muxer = 1;
78 056fe1ba 2003-11-23 devnull while(!r->done){
79 056fe1ba 2003-11-23 devnull qunlock(&z->lk);
80 056fe1ba 2003-11-23 devnull if((p = vtrecv(z)) == nil){
81 2e965b33 2004-05-05 devnull werrstr("unexpected eof on venti connection");
82 056fe1ba 2003-11-23 devnull z->muxer = 0;
83 056fe1ba 2003-11-23 devnull return nil;
85 056fe1ba 2003-11-23 devnull qlock(&z->lk);
86 056fe1ba 2003-11-23 devnull muxrpc(z, p);
88 056fe1ba 2003-11-23 devnull z->muxer = 0;
89 056fe1ba 2003-11-23 devnull /* if there is anyone else sleeping, wake them to mux */
90 056fe1ba 2003-11-23 devnull if(z->nsleep){
91 056fe1ba 2003-11-23 devnull for(i=0; i<256; i++)
92 056fe1ba 2003-11-23 devnull if(z->wait[i] != nil && ((Rwait*)z->wait[i])->sleeping)
94 056fe1ba 2003-11-23 devnull if(i==256)
95 056fe1ba 2003-11-23 devnull fprint(2, "libventi: nsleep botch\n");
97 056fe1ba 2003-11-23 devnull rwakeup(&((Rwait*)z->wait[i])->r);
101 056fe1ba 2003-11-23 devnull p = r->p;
102 056fe1ba 2003-11-23 devnull puttag(z, r, tag);
103 056fe1ba 2003-11-23 devnull vtfree(r);
104 056fe1ba 2003-11-23 devnull qunlock(&z->lk);
105 056fe1ba 2003-11-23 devnull return p;
108 056fe1ba 2003-11-23 devnull static int
109 056fe1ba 2003-11-23 devnull gettag(VtConn *z, Rwait *r)
114 056fe1ba 2003-11-23 devnull while(z->ntag == 256)
115 056fe1ba 2003-11-23 devnull rsleep(&z->tagrend);
116 056fe1ba 2003-11-23 devnull for(i=0; i<256; i++)
117 056fe1ba 2003-11-23 devnull if(z->wait[i] == 0){
118 056fe1ba 2003-11-23 devnull z->ntag++;
119 056fe1ba 2003-11-23 devnull z->wait[i] = r;
120 056fe1ba 2003-11-23 devnull return i;
122 056fe1ba 2003-11-23 devnull fprint(2, "libventi: ntag botch\n");
123 056fe1ba 2003-11-23 devnull goto Again;
126 056fe1ba 2003-11-23 devnull static void
127 056fe1ba 2003-11-23 devnull puttag(VtConn *z, Rwait *r, int tag)
129 056fe1ba 2003-11-23 devnull assert(z->wait[tag] == r);
130 056fe1ba 2003-11-23 devnull z->wait[tag] = nil;
131 056fe1ba 2003-11-23 devnull z->ntag--;
132 056fe1ba 2003-11-23 devnull rwakeup(&z->tagrend);
135 056fe1ba 2003-11-23 devnull static void
136 056fe1ba 2003-11-23 devnull muxrpc(VtConn *z, Packet *p)
138 056fe1ba 2003-11-23 devnull uchar tag, buf[2], *top;
139 056fe1ba 2003-11-23 devnull Rwait *r;
141 056fe1ba 2003-11-23 devnull if((top = packetpeek(p, buf, 0, 2)) == nil){
142 056fe1ba 2003-11-23 devnull fprint(2, "libventi: short packet in vtrpc\n");
143 056fe1ba 2003-11-23 devnull packetfree(p);
147 056fe1ba 2003-11-23 devnull tag = top[1];
148 056fe1ba 2003-11-23 devnull if((r = z->wait[tag]) == nil){
149 056fe1ba 2003-11-23 devnull fprint(2, "libventi: unexpected packet tag %d in vtrpc\n", tag);
150 056fe1ba 2003-11-23 devnull abort();
151 056fe1ba 2003-11-23 devnull packetfree(p);
155 056fe1ba 2003-11-23 devnull r->p = p;
156 056fe1ba 2003-11-23 devnull r->done = 1;
157 056fe1ba 2003-11-23 devnull rwakeup(&r->r);
