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;
30 d3df3087 2003-12-06 devnull muxrpc(Mux *mux, void *tx)
33 d3df3087 2003-12-06 devnull Muxrpc *r, *r2;
36 d3df3087 2003-12-06 devnull /* must malloc because stack could be private */
37 d3df3087 2003-12-06 devnull r = mallocz(sizeof(Muxrpc), 1);
38 d3df3087 2003-12-06 devnull if(r == nil)
39 d3df3087 2003-12-06 devnull return nil;
40 d3df3087 2003-12-06 devnull r->r.l = &mux->lk;
42 43db87f1 2004-12-27 devnull /* assign the tag, add selves to response queue */
43 ceb04770 2003-12-09 devnull qlock(&mux->lk);
44 d3df3087 2003-12-06 devnull tag = gettag(mux, r);
45 43db87f1 2004-12-27 devnull //print("gettag %p %d\n", r, tag);
46 43db87f1 2004-12-27 devnull enqueue(mux, r);
47 ceb04770 2003-12-09 devnull qunlock(&mux->lk);
49 43db87f1 2004-12-27 devnull /* actually send the packet */
50 ceb04770 2003-12-09 devnull if(tag < 0 || mux->settag(mux, tx, tag) < 0 || _muxsend(mux, tx) < 0){
51 ceb04770 2003-12-09 devnull qlock(&mux->lk);
52 d3df3087 2003-12-06 devnull puttag(mux, r);
53 ceb04770 2003-12-09 devnull qunlock(&mux->lk);
54 d3df3087 2003-12-06 devnull return nil;
57 d3df3087 2003-12-06 devnull qlock(&mux->lk);
58 d3df3087 2003-12-06 devnull /* wait for our packet */
59 c6687d45 2004-09-21 devnull while(mux->muxer && !r->p){
60 d3df3087 2003-12-06 devnull rsleep(&r->r);
63 d3df3087 2003-12-06 devnull /* if not done, there's no muxer: start muxing */
64 d3df3087 2003-12-06 devnull if(!r->p){
65 d3df3087 2003-12-06 devnull if(mux->muxer)
67 d3df3087 2003-12-06 devnull mux->muxer = 1;
68 d3df3087 2003-12-06 devnull while(!r->p){
69 d3df3087 2003-12-06 devnull qunlock(&mux->lk);
70 d3df3087 2003-12-06 devnull p = _muxrecv(mux);
72 d3df3087 2003-12-06 devnull tag = mux->gettag(mux, p);
74 d3df3087 2003-12-06 devnull tag = ~0;
75 43db87f1 2004-12-27 devnull //print("mux tag %d\n", tag);
76 d3df3087 2003-12-06 devnull qlock(&mux->lk);
77 d3df3087 2003-12-06 devnull if(p == nil){ /* eof -- just give up and pass the buck */
78 d3df3087 2003-12-06 devnull dequeue(mux, r);
81 d3df3087 2003-12-06 devnull /* hand packet to correct sleeper */
82 d3df3087 2003-12-06 devnull if(tag < 0 || tag >= mux->mwait){
83 d3df3087 2003-12-06 devnull fprint(2, "%s: bad rpc tag %ux\n", argv0, tag);
84 d3df3087 2003-12-06 devnull /* must leak packet! don't know how to free it! */
85 d3df3087 2003-12-06 devnull continue;
87 d3df3087 2003-12-06 devnull r2 = mux->wait[tag];
88 43db87f1 2004-12-27 devnull if(r2 == nil){
89 43db87f1 2004-12-27 devnull fprint(2, "%s: bad rpc tag %ux (no one waiting on that tag)\n", argv0, tag);
90 43db87f1 2004-12-27 devnull /* must leak packet! don't know how to free it! */
91 43db87f1 2004-12-27 devnull continue;
93 d3df3087 2003-12-06 devnull r2->p = p;
94 e95a7088 2003-12-09 devnull dequeue(mux, r2);
95 43db87f1 2004-12-27 devnull puttag(mux, r2);
96 d3df3087 2003-12-06 devnull rwakeup(&r2->r);
98 d3df3087 2003-12-06 devnull mux->muxer = 0;
100 d3df3087 2003-12-06 devnull /* if there is anyone else sleeping, wake them to mux */
101 d3df3087 2003-12-06 devnull if(mux->sleep.next != &mux->sleep)
102 d3df3087 2003-12-06 devnull rwakeup(&mux->sleep.next->r);
104 43db87f1 2004-12-27 devnull //print("finished %p\n", r);
105 d3df3087 2003-12-06 devnull p = r->p;
106 d3df3087 2003-12-06 devnull qunlock(&mux->lk);
107 d3df3087 2003-12-06 devnull return p;
110 d3df3087 2003-12-06 devnull static void
111 d3df3087 2003-12-06 devnull enqueue(Mux *mux, Muxrpc *r)
113 d3df3087 2003-12-06 devnull r->next = mux->sleep.next;
114 d3df3087 2003-12-06 devnull r->prev = &mux->sleep;
115 d3df3087 2003-12-06 devnull r->next->prev = r;
116 d3df3087 2003-12-06 devnull r->prev->next = r;
119 d3df3087 2003-12-06 devnull static void
120 d3df3087 2003-12-06 devnull dequeue(Mux *mux, Muxrpc *r)
122 d3df3087 2003-12-06 devnull r->next->prev = r->prev;
123 d3df3087 2003-12-06 devnull r->prev->next = r->next;
124 d3df3087 2003-12-06 devnull r->prev = nil;
125 d3df3087 2003-12-06 devnull r->next = nil;
128 d3df3087 2003-12-06 devnull static int
129 d3df3087 2003-12-06 devnull gettag(Mux *mux, Muxrpc *r)
131 ceb04770 2003-12-09 devnull int i, mw;
132 ceb04770 2003-12-09 devnull Muxrpc **w;
134 ceb04770 2003-12-09 devnull for(;;){
135 ceb04770 2003-12-09 devnull /* wait for a free tag */
136 ceb04770 2003-12-09 devnull while(mux->nwait == mux->mwait){
137 ceb04770 2003-12-09 devnull if(mux->mwait < mux->maxtag-mux->mintag){
138 ceb04770 2003-12-09 devnull mw = mux->mwait;
139 ceb04770 2003-12-09 devnull if(mw == 0)
142 ceb04770 2003-12-09 devnull mw <<= 1;
143 ceb04770 2003-12-09 devnull w = realloc(mux->wait, mw*sizeof(w[0]));
144 ceb04770 2003-12-09 devnull if(w == nil)
145 ceb04770 2003-12-09 devnull return -1;
146 e95a7088 2003-12-09 devnull memset(w+mux->mwait, 0, (mw-mux->mwait)*sizeof(w[0]));
147 ceb04770 2003-12-09 devnull mux->wait = w;
148 ceb04770 2003-12-09 devnull mux->freetag = mux->mwait;
149 ceb04770 2003-12-09 devnull mux->mwait = mw;
152 ceb04770 2003-12-09 devnull rsleep(&mux->tagrend);
155 ceb04770 2003-12-09 devnull i=mux->freetag;
156 ceb04770 2003-12-09 devnull if(mux->wait[i] == 0)
157 ceb04770 2003-12-09 devnull goto Found;
158 ceb04770 2003-12-09 devnull for(; i<mux->mwait; i++)
159 ceb04770 2003-12-09 devnull if(mux->wait[i] == 0)
160 ceb04770 2003-12-09 devnull goto Found;
161 ceb04770 2003-12-09 devnull for(i=0; i<mux->freetag; i++)
162 ceb04770 2003-12-09 devnull if(mux->wait[i] == 0)
163 ceb04770 2003-12-09 devnull goto Found;
164 ceb04770 2003-12-09 devnull /* should not fall out of while without free tag */
165 ceb04770 2003-12-09 devnull fprint(2, "libfs: nwait botch\n");
166 ceb04770 2003-12-09 devnull abort();
170 ceb04770 2003-12-09 devnull mux->nwait++;
171 ceb04770 2003-12-09 devnull mux->wait[i] = r;
172 ceb04770 2003-12-09 devnull r->tag = i+mux->mintag;
173 ceb04770 2003-12-09 devnull return i;
176 d3df3087 2003-12-06 devnull static void
177 d3df3087 2003-12-06 devnull puttag(Mux *mux, Muxrpc *r)
181 ceb04770 2003-12-09 devnull i = r->tag - mux->mintag;
182 ceb04770 2003-12-09 devnull assert(mux->wait[i] == r);
183 ceb04770 2003-12-09 devnull mux->wait[i] = nil;
184 d3df3087 2003-12-06 devnull mux->nwait--;
185 ceb04770 2003-12-09 devnull mux->freetag = i;
186 d3df3087 2003-12-06 devnull rwakeup(&mux->tagrend);
187 ceb04770 2003-12-09 devnull free(r);