1 d3df3087 2003-12-06 devnull /* Copyright (C) 2003 Russ Cox, Massachusetts Institute of Technology */
2 d3df3087 2003-12-06 devnull /* See COPYRIGHT */
5 d3df3087 2003-12-06 devnull * Generic RPC packet multiplexor. Inspired by but not derived from
6 d3df3087 2003-12-06 devnull * Plan 9 kernel. Originally developed as part of Tra, later used in
7 d3df3087 2003-12-06 devnull * libnventi, and then finally split out into a generic library.
10 d3df3087 2003-12-06 devnull #include <u.h>
11 d3df3087 2003-12-06 devnull #include <libc.h>
12 d3df3087 2003-12-06 devnull #include <mux.h>
14 d3df3087 2003-12-06 devnull static int gettag(Mux*, Muxrpc*);
15 d3df3087 2003-12-06 devnull static void puttag(Mux*, Muxrpc*);
16 d3df3087 2003-12-06 devnull static void enqueue(Mux*, Muxrpc*);
17 d3df3087 2003-12-06 devnull static void dequeue(Mux*, Muxrpc*);
20 d3df3087 2003-12-06 devnull muxinit(Mux *mux)
22 b214663d 2004-01-09 devnull memset(&mux->lk, 0, sizeof(Mux)-offsetof(Mux, lk));
23 d3df3087 2003-12-06 devnull mux->tagrend.l = &mux->lk;
24 b214663d 2004-01-09 devnull mux->rpcfork.l = &mux->lk;
25 d3df3087 2003-12-06 devnull mux->sleep.next = &mux->sleep;
26 d3df3087 2003-12-06 devnull mux->sleep.prev = &mux->sleep;
29 2d2e5c71 2006-06-25 devnull static Muxrpc*
30 2d2e5c71 2006-06-25 devnull allocmuxrpc(Mux *mux)
32 2d2e5c71 2006-06-25 devnull Muxrpc *r;
34 d3df3087 2003-12-06 devnull /* must malloc because stack could be private */
35 d3df3087 2003-12-06 devnull r = mallocz(sizeof(Muxrpc), 1);
36 a28c0548 2005-07-13 devnull if(r == nil){
37 a28c0548 2005-07-13 devnull werrstr("mallocz: %r");
38 d3df3087 2003-12-06 devnull return nil;
40 2d2e5c71 2006-06-25 devnull r->mux = mux;
41 d3df3087 2003-12-06 devnull r->r.l = &mux->lk;
42 2d2e5c71 2006-06-25 devnull r->waiting = 1;
44 2d2e5c71 2006-06-25 devnull return r;
47 2d2e5c71 2006-06-25 devnull static int
48 2d2e5c71 2006-06-25 devnull tagmuxrpc(Muxrpc *r, void *tx)
51 2d2e5c71 2006-06-25 devnull Mux *mux;
53 2d2e5c71 2006-06-25 devnull mux = r->mux;
54 43db87f1 2004-12-27 devnull /* assign the tag, add selves to response queue */
55 ceb04770 2003-12-09 devnull qlock(&mux->lk);
56 d3df3087 2003-12-06 devnull tag = gettag(mux, r);
57 cbeb0b26 2006-04-01 devnull /*print("gettag %p %d\n", r, tag); */
58 43db87f1 2004-12-27 devnull enqueue(mux, r);
59 ceb04770 2003-12-09 devnull qunlock(&mux->lk);
61 43db87f1 2004-12-27 devnull /* actually send the packet */
62 ceb04770 2003-12-09 devnull if(tag < 0 || mux->settag(mux, tx, tag) < 0 || _muxsend(mux, tx) < 0){
63 a28c0548 2005-07-13 devnull werrstr("settag/send tag %d: %r", tag);
64 a28c0548 2005-07-13 devnull fprint(2, "%r\n");
65 ceb04770 2003-12-09 devnull qlock(&mux->lk);
66 d165fa68 2005-01-04 devnull dequeue(mux, r);
67 d3df3087 2003-12-06 devnull puttag(mux, r);
68 ceb04770 2003-12-09 devnull qunlock(&mux->lk);
69 2d2e5c71 2006-06-25 devnull return -1;
71 2d2e5c71 2006-06-25 devnull return 0;
75 2d2e5c71 2006-06-25 devnull muxmsgandqlock(Mux *mux, void *p)
78 2d2e5c71 2006-06-25 devnull Muxrpc *r2;
80 2d2e5c71 2006-06-25 devnull tag = mux->gettag(mux, p) - mux->mintag;
81 2d2e5c71 2006-06-25 devnull /*print("mux tag %d\n", tag); */
82 d3df3087 2003-12-06 devnull qlock(&mux->lk);
83 2d2e5c71 2006-06-25 devnull /* hand packet to correct sleeper */
84 2d2e5c71 2006-06-25 devnull if(tag < 0 || tag >= mux->mwait){
85 2d2e5c71 2006-06-25 devnull fprint(2, "%s: bad rpc tag %ux\n", argv0, tag);
86 2d2e5c71 2006-06-25 devnull /* must leak packet! don't know how to free it! */
89 2d2e5c71 2006-06-25 devnull r2 = mux->wait[tag];
90 2d2e5c71 2006-06-25 devnull if(r2 == nil || r2->prev == nil){
91 2d2e5c71 2006-06-25 devnull fprint(2, "%s: bad rpc tag %ux (no one waiting on that tag)\n", argv0, tag);
92 2d2e5c71 2006-06-25 devnull /* must leak packet! don't know how to free it! */
95 2d2e5c71 2006-06-25 devnull r2->p = p;
96 2d2e5c71 2006-06-25 devnull dequeue(mux, r2);
97 2d2e5c71 2006-06-25 devnull rwakeup(&r2->r);
101 2d2e5c71 2006-06-25 devnull electmuxer(Mux *mux)
103 2d2e5c71 2006-06-25 devnull /* if there is anyone else sleeping, wake them to mux */
104 2d2e5c71 2006-06-25 devnull if(mux->sleep.next != &mux->sleep){
105 2d2e5c71 2006-06-25 devnull mux->muxer = mux->sleep.next;
106 2d2e5c71 2006-06-25 devnull rwakeup(&mux->muxer->r);
108 2d2e5c71 2006-06-25 devnull mux->muxer = nil;
112 2d2e5c71 2006-06-25 devnull muxrpc(Mux *mux, void *tx)
114 2d2e5c71 2006-06-25 devnull int tag;
115 2d2e5c71 2006-06-25 devnull Muxrpc *r;
116 2d2e5c71 2006-06-25 devnull void *p;
118 2d2e5c71 2006-06-25 devnull if((r = allocmuxrpc(mux)) == nil)
119 2d2e5c71 2006-06-25 devnull return nil;
121 2d2e5c71 2006-06-25 devnull if((tag = tagmuxrpc(r, tx)) < 0)
122 2d2e5c71 2006-06-25 devnull return nil;
124 2d2e5c71 2006-06-25 devnull qlock(&mux->lk);
125 d3df3087 2003-12-06 devnull /* wait for our packet */
126 2d2e5c71 2006-06-25 devnull while(mux->muxer && mux->muxer != r && !r->p)
127 d3df3087 2003-12-06 devnull rsleep(&r->r);
129 d3df3087 2003-12-06 devnull /* if not done, there's no muxer: start muxing */
130 d3df3087 2003-12-06 devnull if(!r->p){
131 2d2e5c71 2006-06-25 devnull if(mux->muxer != nil && mux->muxer != r)
132 d3df3087 2003-12-06 devnull abort();
133 2d2e5c71 2006-06-25 devnull mux->muxer = r;
134 d3df3087 2003-12-06 devnull while(!r->p){
135 d3df3087 2003-12-06 devnull qunlock(&mux->lk);
136 2d2e5c71 2006-06-25 devnull p = _muxrecv(mux, 1);
137 2d2e5c71 2006-06-25 devnull if(p == nil){
138 2d2e5c71 2006-06-25 devnull /* eof -- just give up and pass the buck */
139 2d2e5c71 2006-06-25 devnull qlock(&mux->lk);
140 d3df3087 2003-12-06 devnull dequeue(mux, r);
143 2d2e5c71 2006-06-25 devnull muxmsgandqlock(mux, p);
145 2d2e5c71 2006-06-25 devnull electmuxer(mux);
147 cbeb0b26 2006-04-01 devnull /*print("finished %p\n", r); */
148 d3df3087 2003-12-06 devnull p = r->p;
149 d165fa68 2005-01-04 devnull puttag(mux, r);
150 d3df3087 2003-12-06 devnull qunlock(&mux->lk);
151 a28c0548 2005-07-13 devnull if(p == nil)
152 a28c0548 2005-07-13 devnull werrstr("unexpected eof");
153 d3df3087 2003-12-06 devnull return p;
157 2d2e5c71 2006-06-25 devnull muxrpcstart(Mux *mux, void *tx)
159 2d2e5c71 2006-06-25 devnull int tag;
160 2d2e5c71 2006-06-25 devnull Muxrpc *r;
162 2d2e5c71 2006-06-25 devnull if((r = allocmuxrpc(mux)) == nil)
163 2d2e5c71 2006-06-25 devnull return nil;
164 2d2e5c71 2006-06-25 devnull if((tag = tagmuxrpc(r, tx)) < 0)
165 2d2e5c71 2006-06-25 devnull return nil;
166 2d2e5c71 2006-06-25 devnull return r;
170 2d2e5c71 2006-06-25 devnull muxrpccanfinish(Muxrpc *r)
172 2d2e5c71 2006-06-25 devnull char *p;
173 2d2e5c71 2006-06-25 devnull Mux *mux;
175 2d2e5c71 2006-06-25 devnull mux = r->mux;
176 2d2e5c71 2006-06-25 devnull qlock(&mux->lk);
177 2d2e5c71 2006-06-25 devnull if(!r->p && !mux->muxer){
178 2d2e5c71 2006-06-25 devnull mux->muxer = r;
179 2d2e5c71 2006-06-25 devnull while(!r->p){
180 2d2e5c71 2006-06-25 devnull qunlock(&mux->lk);
181 2d2e5c71 2006-06-25 devnull p = _muxrecv(mux, 0);
182 2d2e5c71 2006-06-25 devnull if(p == nil){
183 2d2e5c71 2006-06-25 devnull qlock(&mux->lk);
186 2d2e5c71 2006-06-25 devnull muxmsgandqlock(mux, p);
188 2d2e5c71 2006-06-25 devnull electmuxer(mux);
190 2d2e5c71 2006-06-25 devnull p = r->p;
192 2d2e5c71 2006-06-25 devnull puttag(mux, r);
193 2d2e5c71 2006-06-25 devnull qunlock(&mux->lk);
194 2d2e5c71 2006-06-25 devnull return p;
197 d3df3087 2003-12-06 devnull static void
198 d3df3087 2003-12-06 devnull enqueue(Mux *mux, Muxrpc *r)
200 d3df3087 2003-12-06 devnull r->next = mux->sleep.next;
201 d3df3087 2003-12-06 devnull r->prev = &mux->sleep;
202 d3df3087 2003-12-06 devnull r->next->prev = r;
203 d3df3087 2003-12-06 devnull r->prev->next = r;
206 d3df3087 2003-12-06 devnull static void
207 d3df3087 2003-12-06 devnull dequeue(Mux *mux, Muxrpc *r)
209 d3df3087 2003-12-06 devnull r->next->prev = r->prev;
210 d3df3087 2003-12-06 devnull r->prev->next = r->next;
211 d3df3087 2003-12-06 devnull r->prev = nil;
212 d3df3087 2003-12-06 devnull r->next = nil;
215 d3df3087 2003-12-06 devnull static int
216 d3df3087 2003-12-06 devnull gettag(Mux *mux, Muxrpc *r)
218 ceb04770 2003-12-09 devnull int i, mw;
219 ceb04770 2003-12-09 devnull Muxrpc **w;
221 ceb04770 2003-12-09 devnull for(;;){
222 ceb04770 2003-12-09 devnull /* wait for a free tag */
223 ceb04770 2003-12-09 devnull while(mux->nwait == mux->mwait){
224 ceb04770 2003-12-09 devnull if(mux->mwait < mux->maxtag-mux->mintag){
225 ceb04770 2003-12-09 devnull mw = mux->mwait;
226 ceb04770 2003-12-09 devnull if(mw == 0)
229 ceb04770 2003-12-09 devnull mw <<= 1;
230 ceb04770 2003-12-09 devnull w = realloc(mux->wait, mw*sizeof(w[0]));
231 ceb04770 2003-12-09 devnull if(w == nil)
232 ceb04770 2003-12-09 devnull return -1;
233 e95a7088 2003-12-09 devnull memset(w+mux->mwait, 0, (mw-mux->mwait)*sizeof(w[0]));
234 ceb04770 2003-12-09 devnull mux->wait = w;
235 ceb04770 2003-12-09 devnull mux->freetag = mux->mwait;
236 ceb04770 2003-12-09 devnull mux->mwait = mw;
239 ceb04770 2003-12-09 devnull rsleep(&mux->tagrend);
242 ceb04770 2003-12-09 devnull i=mux->freetag;
243 ceb04770 2003-12-09 devnull if(mux->wait[i] == 0)
244 ceb04770 2003-12-09 devnull goto Found;
245 ceb04770 2003-12-09 devnull for(; i<mux->mwait; i++)
246 ceb04770 2003-12-09 devnull if(mux->wait[i] == 0)
247 ceb04770 2003-12-09 devnull goto Found;
248 ceb04770 2003-12-09 devnull for(i=0; i<mux->freetag; i++)
249 ceb04770 2003-12-09 devnull if(mux->wait[i] == 0)
250 ceb04770 2003-12-09 devnull goto Found;
251 ceb04770 2003-12-09 devnull /* should not fall out of while without free tag */
252 ceb04770 2003-12-09 devnull fprint(2, "libfs: nwait botch\n");
253 ceb04770 2003-12-09 devnull abort();
257 ceb04770 2003-12-09 devnull mux->nwait++;
258 ceb04770 2003-12-09 devnull mux->wait[i] = r;
259 ceb04770 2003-12-09 devnull r->tag = i+mux->mintag;
260 0b22e9bd 2005-01-30 devnull return r->tag;
263 d3df3087 2003-12-06 devnull static void
264 d3df3087 2003-12-06 devnull puttag(Mux *mux, Muxrpc *r)
268 ceb04770 2003-12-09 devnull i = r->tag - mux->mintag;
269 ceb04770 2003-12-09 devnull assert(mux->wait[i] == r);
270 ceb04770 2003-12-09 devnull mux->wait[i] = nil;
271 d3df3087 2003-12-06 devnull mux->nwait--;
272 ceb04770 2003-12-09 devnull mux->freetag = i;
273 d3df3087 2003-12-06 devnull rwakeup(&mux->tagrend);
274 ceb04770 2003-12-09 devnull free(r);
