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 		dequeue(mux, r);
53 		puttag(mux, r);
54 		qunlock(&mux->lk);
55 		return nil;
56 	}
57 
58 	qlock(&mux->lk);
59 	/* wait for our packet */
60 	while(mux->muxer && !r->p){
61 		rsleep(&r->r);
62 	}
63 
64 	/* if not done, there's no muxer: start muxing */
65 	if(!r->p){
66 		if(mux->muxer)
67 			abort();
68 		mux->muxer = 1;
69 		while(!r->p){
70 			qunlock(&mux->lk);
71 			p = _muxrecv(mux);
72 			if(p)
73 				tag = mux->gettag(mux, p) - mux->mintag;
74 			else
75 				tag = ~0;
76 //print("mux tag %d\n", tag);
77 			qlock(&mux->lk);
78 			if(p == nil){	/* eof -- just give up and pass the buck */
79 				dequeue(mux, r);
80 				break;
81 			}
82 			/* hand packet to correct sleeper */
83 			if(tag < 0 || tag >= mux->mwait){
84 				fprint(2, "%s: bad rpc tag %ux\n", argv0, tag);
85 				/* must leak packet! don't know how to free it! */
86 				continue;
87 			}
88 			r2 = mux->wait[tag];
89 			if(r2 == nil || r2->prev == nil){
90 				fprint(2, "%s: bad rpc tag %ux (no one waiting on that tag)\n", argv0, tag);
91 				/* must leak packet! don't know how to free it! */
92 				continue;
93 			}	
94 			r2->p = p;
95 			dequeue(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 	puttag(mux, r);
107 	qunlock(&mux->lk);
108 	return p;
109 }
110 
111 static void
112 enqueue(Mux *mux, Muxrpc *r)
113 {
114 	r->next = mux->sleep.next;
115 	r->prev = &mux->sleep;
116 	r->next->prev = r;
117 	r->prev->next = r;
118 }
119 
120 static void
121 dequeue(Mux *mux, Muxrpc *r)
122 {
123 	r->next->prev = r->prev;
124 	r->prev->next = r->next;
125 	r->prev = nil;
126 	r->next = nil;
127 }
128 
129 static int 
130 gettag(Mux *mux, Muxrpc *r)
131 {
132 	int i, mw;
133 	Muxrpc **w;
134 
135 	for(;;){
136 		/* wait for a free tag */
137 		while(mux->nwait == mux->mwait){
138 			if(mux->mwait < mux->maxtag-mux->mintag){
139 				mw = mux->mwait;
140 				if(mw == 0)
141 					mw = 1;
142 				else
143 					mw <<= 1;
144 				w = realloc(mux->wait, mw*sizeof(w[0]));
145 				if(w == nil)
146 					return -1;
147 				memset(w+mux->mwait, 0, (mw-mux->mwait)*sizeof(w[0]));
148 				mux->wait = w;
149 				mux->freetag = mux->mwait;
150 				mux->mwait = mw;
151 				break;
152 			}
153 			rsleep(&mux->tagrend);
154 		}
155 
156 		i=mux->freetag;
157 		if(mux->wait[i] == 0)
158 			goto Found;
159 		for(; i<mux->mwait; i++)
160 			if(mux->wait[i] == 0)
161 				goto Found;
162 		for(i=0; i<mux->freetag; i++)
163 			if(mux->wait[i] == 0)
164 				goto Found;
165 		/* should not fall out of while without free tag */
166 		fprint(2, "libfs: nwait botch\n");
167 		abort();
168 	}
169 
170 Found:
171 	mux->nwait++;
172 	mux->wait[i] = r;
173 	r->tag = i+mux->mintag;
174 	return r->tag;
175 }
176 
177 static void
178 puttag(Mux *mux, Muxrpc *r)
179 {
180 	int i;
181 
182 	i = r->tag - mux->mintag;
183 	assert(mux->wait[i] == r);
184 	mux->wait[i] = nil;
185 	mux->nwait--;
186 	mux->freetag = i;
187 	rwakeup(&mux->tagrend);
188 	free(r);
189 }