1 #include "u.h"
2 #include "libc.h"
3 #include "thread.h"
4 #include "threadimpl.h"
5 
6 /*
7  * One can go through a lot of effort to avoid this global lock.
8  * You have to put locks in all the channels and all the Alt
9  * structures.  At the beginning of an alt you have to lock all
10  * the channels, but then to try to actually exec an op you
11  * have to lock the other guy's alt structure, so that other 
12  * people aren't trying to use him in some other op at the
13  * same time.
14  * 
15  * For Plan 9 apps, it's just not worth the extra effort.
16  */
17 static QLock chanlock;
18 
19 Channel*
20 chancreate(int elemsize, int bufsize)
21 {
22 	Channel *c;
23 
24 	c = malloc(sizeof *c+bufsize*elemsize);
25 	if(c == nil)
26 		sysfatal("chancreate malloc: %r");
27 	memset(c, 0, sizeof *c);
28 	c->elemsize = elemsize;
29 	c->bufsize = bufsize;
30 	c->nbuf = 0;
31 	c->buf = (uchar*)(c+1);
32 	return c;
33 }
34 
35 void
36 chansetname(Channel *c, char *fmt, ...)
37 {
38 	char *name;
39 	va_list arg;
40 
41 	va_start(arg, fmt);
42 	name = vsmprint(fmt, arg);
43 	va_end(arg);
44 	free(c->name);
45 	c->name = name;
46 }
47 
48 /* bug - work out races */
49 void
50 chanfree(Channel *c)
51 {
52 	if(c == nil)
53 		return;
54 	free(c->name);
55 	free(c->arecv.a);
56 	free(c->asend.a);
57 	free(c);
58 }
59 
60 static void
61 addarray(_Altarray *a, Alt *alt)
62 {
63 	if(a->n == a->m){
64 		a->m += 16;
65 		a->a = realloc(a->a, a->m*sizeof a->a[0]);
66 	}
67 	a->a[a->n++] = alt;
68 }
69 
70 static void
71 delarray(_Altarray *a, int i)
72 {
73 	--a->n;
74 	a->a[i] = a->a[a->n];
75 }
76 
77 /*
78  * doesn't really work for things other than CHANSND and CHANRCV
79  * but is only used as arg to chanarray, which can handle it
80  */
81 #define otherop(op)	(CHANSND+CHANRCV-(op))
82 
83 static _Altarray*
84 chanarray(Channel *c, uint op)
85 {
86 	switch(op){
87 	default:
88 		return nil;
89 	case CHANSND:
90 		return &c->asend;
91 	case CHANRCV:
92 		return &c->arecv;
93 	}
94 }
95 
96 static int
97 altcanexec(Alt *a)
98 {
99 	_Altarray *ar;
100 	Channel *c;
101 
102 	if(a->op == CHANNOP)
103 		return 0;
104 	c = a->c;
105 	if(c->bufsize == 0){
106 		ar = chanarray(c, otherop(a->op));
107 		return ar && ar->n;
108 	}else{
109 		switch(a->op){
110 		default:
111 			return 0;
112 		case CHANSND:
113 			return c->nbuf < c->bufsize;
114 		case CHANRCV:
115 			return c->nbuf > 0;
116 		}
117 	}
118 }
119 
120 static void
121 altqueue(Alt *a)
122 {
123 	_Altarray *ar;
124 
125 	ar = chanarray(a->c, a->op);
126 	addarray(ar, a);
127 }
128 
129 static void
130 altdequeue(Alt *a)
131 {
132 	int i;
133 	_Altarray *ar;
134 
135 	ar = chanarray(a->c, a->op);
136 	if(ar == nil){
137 		fprint(2, "bad use of altdequeue op=%d\n", a->op);
138 		abort();
139 	}
140 
141 	for(i=0; i<ar->n; i++)
142 		if(ar->a[i] == a){
143 			delarray(ar, i);
144 			return;
145 		}
146 	fprint(2, "cannot find self in altdq\n");
147 	abort();
148 }
149 
150 static void
151 altalldequeue(Alt *a)
152 {
153 	int i;
154 
155 	for(i=0; a[i].op!=CHANEND && a[i].op!=CHANNOBLK; i++)
156 		if(a[i].op != CHANNOP)
157 			altdequeue(&a[i]);
158 }
159 
160 static void
161 amove(void *dst, void *src, uint n)
162 {
163 	if(dst){
164 		if(src == nil)
165 			memset(dst, 0, n);
166 		else
167 			memmove(dst, src, n);
168 	}
169 }
170 
171 /*
172  * Actually move the data around.  There are up to three
173  * players: the sender, the receiver, and the channel itself.
174  * If the channel is unbuffered or the buffer is empty,
175  * data goes from sender to receiver.  If the channel is full,
176  * the receiver removes some from the channel and the sender
177  * gets to put some in.
178  */
179 static void
180 altcopy(Alt *s, Alt *r)
181 {
182 	Alt *t;
183 	Channel *c;
184 	uchar *cp;
185 
186 	/*
187 	 * Work out who is sender and who is receiver
188 	 */
189 	if(s == nil && r == nil)
190 		return;
191 	assert(s != nil);
192 	c = s->c;
193 	if(s->op == CHANRCV){
194 		t = s;
195 		s = r;
196 		r = t;
197 	}
198 	assert(s==nil || s->op == CHANSND);
199 	assert(r==nil || r->op == CHANRCV);
200 
201 	/*
202 	 * Channel is empty (or unbuffered) - copy directly.
203 	 */
204 	if(s && r && c->nbuf == 0){
205 		amove(r->v, s->v, c->elemsize);
206 		return;
207 	}
208 
209 	/*
210 	 * Otherwise it's always okay to receive and then send.
211 	 */
212 	if(r){
213 		cp = c->buf + c->off*c->elemsize;
214 		amove(r->v, cp, c->elemsize);
215 		--c->nbuf;
216 		if(++c->off == c->bufsize)
217 			c->off = 0;
218 	}
219 	if(s){
220 		cp = c->buf + (c->off+c->nbuf)%c->bufsize*c->elemsize;
221 		amove(cp, s->v, c->elemsize);
222 		++c->nbuf;
223 	}
224 }
225 
226 static void
227 altexec(Alt *a)
228 {
229 	int i;
230 	_Altarray *ar;
231 	Alt *other;
232 	Channel *c;
233 
234 	c = a->c;
235 	ar = chanarray(c, otherop(a->op));
236 	if(ar && ar->n){
237 		i = rand()%ar->n;
238 		other = ar->a[i];
239 		altcopy(a, other);
240 		altalldequeue(other->xalt);
241 		other->xalt[0].xalt = other;
242 		_threadready(other->thread);
243 	}else
244 		altcopy(a, nil);
245 }
246 
247 #define dbgalt 0
248 int
249 chanalt(Alt *a)
250 {
251 	int i, j, ncan, n, canblock;
252 	Channel *c;
253 	_Thread *t;
254 
255 	for(i=0; a[i].op != CHANEND && a[i].op != CHANNOBLK; i++)
256 		;
257 	n = i;
258 	canblock = a[i].op == CHANEND;
259 
260 	t = proc()->thread;
261 	for(i=0; i<n; i++){
262 		a[i].thread = t;
263 		a[i].xalt = a;
264 	}
265 	qlock(&chanlock);
266 if(dbgalt) print("alt ");
267 	ncan = 0;
268 	for(i=0; i<n; i++){
269 		c = a[i].c;
270 if(dbgalt) print(" %c:", "esrnb"[a[i].op]);
271 if(dbgalt) if(c->name) print("%s", c->name); else print("%p", c);
272 		if(altcanexec(&a[i])){
273 if(dbgalt) print("*");
274 			ncan++;
275 		}
276 	}
277 	if(ncan){
278 		j = rand()%ncan;
279 		for(i=0; i<n; i++){
280 			if(altcanexec(&a[i])){
281 				if(j-- == 0){
282 if(dbgalt){
283 c = a[i].c;
284 print(" => %c:", "esrnb"[a[i].op]);
285 if(c->name) print("%s", c->name); else print("%p", c);
286 print("\n");
287 }
288 					altexec(&a[i]);
289 					qunlock(&chanlock);
290 					return i;
291 				}
292 			}
293 		}
294 	}
295 if(dbgalt)print("\n");
296 
297 	if(!canblock){
298 		qunlock(&chanlock);
299 		return -1;
300 	}
301 
302 	for(i=0; i<n; i++){
303 		if(a[i].op != CHANNOP)
304 			altqueue(&a[i]);
305 	}
306 	qunlock(&chanlock);
307 
308 	_threadswitch();
309 
310 	/*
311 	 * the guy who ran the op took care of dequeueing us
312 	 * and then set a[0].alt to the one that was executed.
313 	 */
314 	return a[0].xalt - a;
315 }
316 
317 static int
318 _chanop(Channel *c, int op, void *p, int canblock)
319 {
320 	Alt a[2];
321 
322 	a[0].c = c;
323 	a[0].op = op;
324 	a[0].v = p;
325 	a[1].op = canblock ? CHANEND : CHANNOBLK;
326 	if(chanalt(a) < 0)
327 		return -1;
328 	return 1;
329 }
330 
331 int
332 chansend(Channel *c, void *v)
333 {
334 	return _chanop(c, CHANSND, v, 1);
335 }
336 
337 int
338 channbsend(Channel *c, void *v)
339 {
340 	return _chanop(c, CHANSND, v, 0);
341 }
342 
343 int
344 chanrecv(Channel *c, void *v)
345 {
346 	return _chanop(c, CHANRCV, v, 1);
347 }
348 
349 int
350 channbrecv(Channel *c, void *v)
351 {
352 	return _chanop(c, CHANRCV, v, 0);
353 }
354 
355 int
356 chansendp(Channel *c, void *v)
357 {
358 	return _chanop(c, CHANSND, (void*)&v, 1);
359 }
360 
361 void*
362 chanrecvp(Channel *c)
363 {
364 	void *v;
365 
366 	_chanop(c, CHANRCV, (void*)&v, 1);
367 	return v;
368 }
369 
370 int
371 channbsendp(Channel *c, void *v)
372 {
373 	return _chanop(c, CHANSND, (void*)&v, 0);
374 }
375 
376 void*
377 channbrecvp(Channel *c)
378 {
379 	void *v;
380 
381 	_chanop(c, CHANRCV, (void*)&v, 0);
382 	return v;
383 }
384 
385 int
386 chansendul(Channel *c, ulong val)
387 {
388 	return _chanop(c, CHANSND, &val, 1);
389 }
390 
391 ulong
392 chanrecvul(Channel *c)
393 {
394 	ulong val;
395 
396 	_chanop(c, CHANRCV, &val, 1);
397 	return val;
398 }
399 
400 int
401 channbsendul(Channel *c, ulong val)
402 {
403 	return _chanop(c, CHANSND, &val, 0);
404 }
405 
406 ulong
407 channbrecvul(Channel *c)
408 {
409 	ulong val;
410 
411 	_chanop(c, CHANRCV, &val, 0);
412 	return val;
413 }
414