1 /* Copyright (C) 2003 Russ Cox, Massachusetts Institute of Technology */
2 /* See COPYRIGHT */
3 
4 /*
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.
8  */
9 
10 #include <u.h>
11 #include <libc.h>
12 #include <mux.h>
13 
14 static int gettag(Mux*, Muxrpc*);
15 static void puttag(Mux*, Muxrpc*);
16 static void enqueue(Mux*, Muxrpc*);
17 static void dequeue(Mux*, Muxrpc*);
18 
19 void
20 muxinit(Mux *mux)
21 {
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;
27 }
28 
29 void*
30 muxrpc(Mux *mux, void *tx)
31 {
32 	int tag;
33 	Muxrpc *r, *r2;
34 	void *p;
35 
36 	/* must malloc because stack could be private */
37 	r = mallocz(sizeof(Muxrpc), 1);
38 	if(r == nil)
39 		return nil;
40 	r->r.l = &mux->lk;
41 
42 	/* assign the tag, add selves to response queue */
43 	qlock(&mux->lk);
44 	tag = gettag(mux, r);
45 //print("gettag %p %d\n", r, tag);
46 	enqueue(mux, r);
47 	qunlock(&mux->lk);
48 
49 	/* actually send the packet */
50 	if(tag < 0 || mux->settag(mux, tx, tag) < 0 || _muxsend(mux, tx) < 0){
51 		qlock(&mux->lk);
52 		puttag(mux, r);
53 		qunlock(&mux->lk);
54 		return nil;
55 	}
56 
57 	qlock(&mux->lk);
58 	/* wait for our packet */
59 	while(mux->muxer && !r->p){
60 		rsleep(&r->r);
61 	}
62 
63 	/* if not done, there's no muxer: start muxing */
64 	if(!r->p){
65 		if(mux->muxer)
66 			abort();
67 		mux->muxer = 1;
68 		while(!r->p){
69 			qunlock(&mux->lk);
70 			p = _muxrecv(mux);
71 			if(p)
72 				tag = mux->gettag(mux, p);
73 			else
74 				tag = ~0;
75 //print("mux tag %d\n", tag);
76 			qlock(&mux->lk);
77 			if(p == nil){	/* eof -- just give up and pass the buck */
78 				dequeue(mux, r);
79 				break;
80 			}
81 			/* hand packet to correct sleeper */
82 			if(tag < 0 || tag >= mux->mwait){
83 				fprint(2, "%s: bad rpc tag %ux\n", argv0, tag);
84 				/* must leak packet! don't know how to free it! */
85 				continue;
86 			}
87 			r2 = mux->wait[tag];
88 			if(r2 == nil){
89 				fprint(2, "%s: bad rpc tag %ux (no one waiting on that tag)\n", argv0, tag);
90 				/* must leak packet! don't know how to free it! */
91 				continue;
92 			}	
93 			r2->p = p;
94 			dequeue(mux, r2);
95 			puttag(mux, r2);
96 			rwakeup(&r2->r);
97 		}
98 		mux->muxer = 0;
99 
100 		/* if there is anyone else sleeping, wake them to mux */
101 		if(mux->sleep.next != &mux->sleep)
102 			rwakeup(&mux->sleep.next->r);
103 	}
104 //print("finished %p\n", r);
105 	p = r->p;
106 	qunlock(&mux->lk);
107 	return p;
108 }
109 
110 static void
111 enqueue(Mux *mux, Muxrpc *r)
112 {
113 	r->next = mux->sleep.next;
114 	r->prev = &mux->sleep;
115 	r->next->prev = r;
116 	r->prev->next = r;
117 }
118 
119 static void
120 dequeue(Mux *mux, Muxrpc *r)
121 {
122 	r->next->prev = r->prev;
123 	r->prev->next = r->next;
124 	r->prev = nil;
125 	r->next = nil;
126 }
127 
128 static int 
129 gettag(Mux *mux, Muxrpc *r)
130 {
131 	int i, mw;
132 	Muxrpc **w;
133 
134 	for(;;){
135 		/* wait for a free tag */
136 		while(mux->nwait == mux->mwait){
137 			if(mux->mwait < mux->maxtag-mux->mintag){
138 				mw = mux->mwait;
139 				if(mw == 0)
140 					mw = 1;
141 				else
142 					mw <<= 1;
143 				w = realloc(mux->wait, mw*sizeof(w[0]));
144 				if(w == nil)
145 					return -1;
146 				memset(w+mux->mwait, 0, (mw-mux->mwait)*sizeof(w[0]));
147 				mux->wait = w;
148 				mux->freetag = mux->mwait;
149 				mux->mwait = mw;
150 				break;
151 			}
152 			rsleep(&mux->tagrend);
153 		}
154 
155 		i=mux->freetag;
156 		if(mux->wait[i] == 0)
157 			goto Found;
158 		for(; i<mux->mwait; i++)
159 			if(mux->wait[i] == 0)
160 				goto Found;
161 		for(i=0; i<mux->freetag; i++)
162 			if(mux->wait[i] == 0)
163 				goto Found;
164 		/* should not fall out of while without free tag */
165 		fprint(2, "libfs: nwait botch\n");
166 		abort();
167 	}
168 
169 Found:
170 	mux->nwait++;
171 	mux->wait[i] = r;
172 	r->tag = i+mux->mintag;
173 	return i;
174 }
175 
176 static void
177 puttag(Mux *mux, Muxrpc *r)
178 {
179 	int i;
180 
181 	i = r->tag - mux->mintag;
182 	assert(mux->wait[i] == r);
183 	mux->wait[i] = nil;
184 	mux->nwait--;
185 	mux->freetag = i;
186 	rwakeup(&mux->tagrend);
187 	free(r);
188 }