1 /* Copyright (C) 2003 Russ Cox, Massachusetts Institute of Technology */
5 * Generic RPC packet multiplexor. Inspired by but not derived from
6 * Plan 9 kernel. Originally developed as part of Tra, later used in
7 * libnventi, and then finally split out into a generic library.
14 static int gettag(Mux*, Muxrpc*);
15 static void puttag(Mux*, Muxrpc*);
16 static void enqueue(Mux*, Muxrpc*);
17 static void dequeue(Mux*, Muxrpc*);
22 memset(&mux->lk, 0, sizeof(Mux)-offsetof(Mux, lk));
23 mux->tagrend.l = &mux->lk;
24 mux->rpcfork.l = &mux->lk;
25 mux->sleep.next = &mux->sleep;
26 mux->sleep.prev = &mux->sleep;
30 muxrpc(Mux *mux, void *tx)
36 /* must malloc because stack could be private */
37 r = mallocz(sizeof(Muxrpc), 1);
46 if(tag < 0 || mux->settag(mux, tx, tag) < 0 || _muxsend(mux, tx) < 0){
53 /* add ourselves to sleep queue */
57 /* wait for our packet */
58 while(mux->muxer && !r->p){
62 /* if not done, there's no muxer: start muxing */
71 tag = mux->gettag(mux, p);
75 if(p == nil){ /* eof -- just give up and pass the buck */
79 /* hand packet to correct sleeper */
80 if(tag < 0 || tag >= mux->mwait){
81 fprint(2, "%s: bad rpc tag %ux\n", argv0, tag);
82 /* must leak packet! don't know how to free it! */
92 /* if there is anyone else sleeping, wake them to mux */
93 if(mux->sleep.next != &mux->sleep)
94 rwakeup(&mux->sleep.next->r);
103 enqueue(Mux *mux, Muxrpc *r)
105 r->next = mux->sleep.next;
106 r->prev = &mux->sleep;
112 dequeue(Mux *mux, Muxrpc *r)
114 r->next->prev = r->prev;
115 r->prev->next = r->next;
121 gettag(Mux *mux, Muxrpc *r)
127 /* wait for a free tag */
128 while(mux->nwait == mux->mwait){
129 if(mux->mwait < mux->maxtag-mux->mintag){
135 w = realloc(mux->wait, mw*sizeof(w[0]));
138 memset(w+mux->mwait, 0, (mw-mux->mwait)*sizeof(w[0]));
140 mux->freetag = mux->mwait;
144 rsleep(&mux->tagrend);
148 if(mux->wait[i] == 0)
150 for(; i<mux->mwait; i++)
151 if(mux->wait[i] == 0)
153 for(i=0; i<mux->freetag; i++)
154 if(mux->wait[i] == 0)
156 /* should not fall out of while without free tag */
157 fprint(2, "libfs: nwait botch\n");
164 r->tag = i+mux->mintag;
169 puttag(Mux *mux, Muxrpc *r)
173 i = r->tag - mux->mintag;
174 assert(mux->wait[i] == r);
178 rwakeup(&mux->tagrend);