1 #include "threadimpl.h"
2 
3 int	_threaddebuglevel;
4 
5 static	uint		threadnproc;
6 static	uint		threadnsysproc;
7 static	Lock		threadnproclock;
8 static	Ref		threadidref;
9 static	Proc		*threadmainproc;
10 
11 static	void		addproc(Proc*);
12 static	void		delproc(Proc*);
13 static	void		addthread(_Threadlist*, _Thread*);
14 static	void		delthread(_Threadlist*, _Thread*);
15 static	void		addthreadinproc(Proc*, _Thread*);
16 static	void		delthreadinproc(Proc*, _Thread*);
17 static	void		contextswitch(Context *from, Context *to);
18 static	void		procscheduler(Proc*);
19 static	int		threadinfo(void*, char*);
20 
21 static void
22 _threaddebug(char *fmt, ...)
23 {
24 	va_list arg;
25 	char buf[128];
26 	_Thread *t;
27 	char *p;
28 	static int fd = -1;
29 
30 	if(_threaddebuglevel == 0)
31 		return;
32 
33 	if(fd < 0){
34 		p = strrchr(argv0, '/');
35 		if(p)
36 			p++;
37 		else
38 			p = argv0;
39 		snprint(buf, sizeof buf, "/tmp/%s.tlog", p);
40 		if((fd = create(buf, OWRITE, 0666)) < 0)
41 			fd = open("/dev/null", OWRITE);
42 	}
43 
44 	va_start(arg, fmt);
45 	vsnprint(buf, sizeof buf, fmt, arg);
46 	va_end(arg);
47 	t = proc()->thread;
48 	if(t)
49 		fprint(fd, "%d.%d: %s\n", getpid(), t->id, buf);
50 	else
51 		fprint(fd, "%d._: %s\n", getpid(), buf);
52 }
53 
54 static _Thread*
55 getthreadnow(void)
56 {
57 	return proc()->thread;
58 }
59 _Thread	*(*threadnow)(void) = getthreadnow;
60 
61 static Proc*
62 procalloc(void)
63 {
64 	Proc *p;
65 
66 	p = malloc(sizeof *p);
67 	if(p == nil)
68 		sysfatal("procalloc malloc: %r");
69 	memset(p, 0, sizeof *p);
70 	addproc(p);
71 	lock(&threadnproclock);
72 	threadnproc++;
73 	unlock(&threadnproclock);
74 	return p;
75 }
76 
77 static void
78 threadstart(uint y, uint x)
79 {
80 	_Thread *t;
81 	ulong z;
82 
83 	z = x<<16;	/* hide undefined 32-bit shift from 32-bit compilers */
84 	z <<= 16;
85 	z |= y;
86 	t = (_Thread*)z;
87 
88 //print("threadstart %p\n", v);
89 	t->startfn(t->startarg);
90 //print("threadexits %p\n", v);
91 	threadexits(nil);
92 //print("not reacehd\n");
93 }
94 
95 static _Thread*
96 threadalloc(void (*fn)(void*), void *arg, uint stack)
97 {
98 	_Thread *t;
99 	sigset_t zero;
100 	uint x, y;
101 	ulong z;
102 
103 	/* allocate the task and stack together */
104 	t = malloc(sizeof *t+stack);
105 	if(t == nil)
106 		sysfatal("threadalloc malloc: %r");
107 	memset(t, 0, sizeof *t);
108 	t->stk = (uchar*)(t+1);
109 	t->stksize = stack;
110 	t->id = incref(&threadidref);
111 	t->startfn = fn;
112 	t->startarg = arg;
113 
114 	/* do a reasonable initialization */
115 	memset(&t->context.uc, 0, sizeof t->context.uc);
116 	sigemptyset(&zero);
117 	sigprocmask(SIG_BLOCK, &zero, &t->context.uc.uc_sigmask);
118 
119 	/* must initialize with current context */
120 	if(getcontext(&t->context.uc) < 0)
121 		sysfatal("threadalloc getcontext: %r");
122 
123 	/* call makecontext to do the real work. */
124 	/* leave a few words open on both ends */
125 	t->context.uc.uc_stack.ss_sp = t->stk+8;
126 	t->context.uc.uc_stack.ss_size = t->stksize-64;
127 #if defined(__sun__) && !defined(__MAKECONTEXT_V2_SOURCE)		/* sigh */
128 	/* can avoid this with __MAKECONTEXT_V2_SOURCE but only on SunOS 5.9 */
129 	t->context.uc.uc_stack.ss_sp = 
130 		(char*)t->context.uc.uc_stack.ss_sp
131 		+t->context.uc.uc_stack.ss_size;
132 #endif
133 	/*
134 	 * All this magic is because you have to pass makecontext a
135 	 * function that takes some number of word-sized variables,
136 	 * and on 64-bit machines pointers are bigger than words.
137 	 */
138 	z = (ulong)t;
139 	y = z;
140 	z >>= 16;	/* hide undefined 32-bit shift from 32-bit compilers */
141 	x = z>>16;
142 	makecontext(&t->context.uc, (void(*)(void))threadstart, 2, y, x);
143 
144 	return t;
145 }
146 
147 _Thread*
148 _threadcreate(Proc *p, void (*fn)(void*), void *arg, uint stack)
149 {
150 	_Thread *t;
151 
152 	t = threadalloc(fn, arg, stack);
153 	t->proc = p;
154 	addthreadinproc(p, t);
155 	p->nthread++;
156 	_threadready(t);
157 	return t;
158 }
159 
160 int
161 threadcreate(void (*fn)(void*), void *arg, uint stack)
162 {
163 	_Thread *t;
164 
165 	t = _threadcreate(proc(), fn, arg, stack);
166 	return t->id;
167 }
168 
169 int
170 proccreate(void (*fn)(void*), void *arg, uint stack)
171 {
172 	int id;
173 	_Thread *t;
174 	Proc *p;
175 
176 	p = procalloc();
177 	t = _threadcreate(p, fn, arg, stack);
178 	id = t->id;	/* t might be freed after _procstart */
179 	_procstart(p, procscheduler);
180 	return id;
181 }
182 
183 void
184 _threadswitch(void)
185 {
186 	Proc *p;
187 
188 	needstack(0);
189 	p = proc();
190 //print("threadswtch %p\n", p);
191 	contextswitch(&p->thread->context, &p->schedcontext);
192 }
193 
194 void
195 _threadready(_Thread *t)
196 {
197 	Proc *p;
198 
199 	p = t->proc;
200 	lock(&p->lock);
201 	p->runrend.l = &p->lock;
202 	addthread(&p->runqueue, t);
203 //print("%d wake for job %d->%d\n", time(0), getpid(), p->osprocid);
204 	if(p != proc())
205 		_procwakeupandunlock(&p->runrend);
206 	else
207 		unlock(&p->lock);
208 }
209 
210 int
211 threadidle(void)
212 {
213 	int n;
214 	Proc *p;
215 	
216 	p = proc();
217 	n = p->nswitch;
218 	lock(&p->lock);
219 	p->runrend.l = &p->lock;
220 	addthread(&p->idlequeue, p->thread);
221 	unlock(&p->lock);
222 	_threadswitch();
223 	return p->nswitch - n;
224 }
225 
226 int
227 threadyield(void)
228 {
229 	int n;
230 	Proc *p;
231 
232 	p = proc();
233 	n = p->nswitch;
234 	_threadready(p->thread);
235 	_threadswitch();
236 	return p->nswitch - n;
237 }
238 
239 void
240 threadexits(char *msg)
241 {
242 	Proc *p;
243 
244 	p = proc();
245 	if(msg == nil)
246 		msg = "";
247 	utfecpy(p->msg, p->msg+sizeof p->msg, msg);
248 	proc()->thread->exiting = 1;
249 	_threadswitch();
250 }
251 
252 static void
253 contextswitch(Context *from, Context *to)
254 {
255 	if(swapcontext(&from->uc, &to->uc) < 0){
256 		fprint(2, "swapcontext failed: %r\n");
257 		assert(0);
258 	}
259 }
260 
261 static void
262 procscheduler(Proc *p)
263 {
264 	_Thread *t;
265 
266 	setproc(p);
267 	_threaddebug("scheduler enter");
268 //	print("s %p\n", p);
269 	lock(&p->lock);
270 	for(;;){
271 		while((t = p->runqueue.head) == nil){
272 			if(p->nthread == 0)
273 				goto Out;
274 			if((t = p->idlequeue.head) != nil){
275 				/*
276 				 * Run all the idling threads once.
277 				 */
278 				while((t = p->idlequeue.head) != nil){
279 					delthread(&p->idlequeue, t);
280 					addthread(&p->runqueue, t);
281 				}
282 				continue;
283 			}
284 			p->runrend.l = &p->lock;
285 			_threaddebug("scheduler sleep");
286 			_procsleep(&p->runrend);
287 			_threaddebug("scheduler wake");
288 		}
289 		delthread(&p->runqueue, t);
290 		unlock(&p->lock);
291 		p->thread = t;
292 		p->nswitch++;
293 		_threaddebug("run %d (%s)", t->id, t->name);
294 		contextswitch(&p->schedcontext, &t->context);
295 //print("back in scheduler\n");
296 		p->thread = nil;
297 		lock(&p->lock);
298 		if(t->exiting){
299 			delthreadinproc(p, t);
300 			p->nthread--;
301 //print("nthread %d\n", p->nthread);
302 			free(t);
303 		}
304 	}
305 
306 Out:
307 	_threaddebug("scheduler exit");
308 	if(p->mainproc){
309 		/*
310 		 * Stupid bug - on Linux 2.6 and maybe elsewhere,
311 		 * if the main thread exits then the others keep running
312 		 * but the process shows up as a zombie in ps and is not
313 		 * attachable with ptrace.  We'll just sit around pretending
314 		 * to be a system proc instead of exiting.
315 		 */
316 		_threaddaemonize();
317 		lock(&threadnproclock);
318 		if(++threadnsysproc == threadnproc)
319 			threadexitsall(p->msg);
320 		p->sysproc = 1;
321 		unlock(&threadnproclock);
322 		for(;;)
323 		 	sleep(1000);
324 	}
325 
326 	delproc(p);
327 	lock(&threadnproclock);
328 	if(p->sysproc)
329 		--threadnsysproc;
330 	if(--threadnproc == threadnsysproc)
331 		threadexitsall(p->msg);
332 	unlock(&threadnproclock);
333 	unlock(&p->lock);
334 	free(p);
335 }
336 
337 void
338 _threadsetsysproc(void)
339 {
340 	lock(&threadnproclock);
341 	if(++threadnsysproc == threadnproc)
342 		threadexitsall(nil);
343 	unlock(&threadnproclock);
344 	proc()->sysproc = 1;
345 }
346 
347 void**
348 procdata(void)
349 {
350 	return &proc()->udata;
351 }
352 
353 void**
354 threaddata(void)
355 {
356 	return &proc()->thread->udata;
357 }
358 
359 extern Jmp *(*_notejmpbuf)(void);
360 static Jmp*
361 threadnotejmp(void)
362 {
363 	return &proc()->sigjmp;
364 }
365 
366 /*
367  * debugging
368  */
369 void
370 threadsetname(char *fmt, ...)
371 {
372 	va_list arg;
373 	_Thread *t;
374 
375 	t = proc()->thread;
376 	va_start(arg, fmt);
377 	vsnprint(t->name, sizeof t->name, fmt, arg);
378 	va_end(arg);
379 }
380 
381 char*
382 threadgetname(void)
383 {
384 	return proc()->thread->name;
385 }
386 
387 void
388 threadsetstate(char *fmt, ...)
389 {
390 	va_list arg;
391 	_Thread *t;
392 
393 	t = proc()->thread;
394 	va_start(arg, fmt);
395 	vsnprint(t->state, sizeof t->name, fmt, arg);
396 	va_end(arg);
397 }
398 
399 int
400 threadid(void)
401 {
402 	_Thread *t;
403 	
404 	t = proc()->thread;
405 	return t->id;
406 }
407 
408 void
409 needstack(int n)
410 {
411 	_Thread *t;
412 
413 	t = proc()->thread;
414 
415 	if((char*)&t <= (char*)t->stk
416 	|| (char*)&t - (char*)t->stk < 256+n){
417 		fprint(2, "thread stack overflow: &t=%p tstk=%p n=%d\n", &t, t->stk, 256+n);
418 		abort();
419 	}
420 }
421 
422 /*
423  * locking
424  */
425 static int
426 threadqlock(QLock *l, int block, ulong pc)
427 {
428 //print("threadqlock %p\n", l);
429 	lock(&l->l);
430 	if(l->owner == nil){
431 		l->owner = (*threadnow)();
432 //print("qlock %p @%#x by %p\n", l, pc, l->owner);
433 		unlock(&l->l);
434 		return 1;
435 	}
436 	if(!block){
437 		unlock(&l->l);
438 		return 0;
439 	}
440 //print("qsleep %p @%#x by %p\n", l, pc, (*threadnow)());
441 	addthread(&l->waiting, (*threadnow)());
442 	unlock(&l->l);
443 
444 	_threadswitch();
445 
446 	if(l->owner != (*threadnow)()){
447 		fprint(2, "%s: qlock pc=0x%lux owner=%p self=%p oops\n",
448 			argv0, pc, l->owner, (*threadnow)());
449 		abort();
450 	}
451 //print("qlock wakeup %p @%#x by %p\n", l, pc, (*threadnow)());
452 	return 1;
453 }
454 
455 static void
456 threadqunlock(QLock *l, ulong pc)
457 {
458 	_Thread *ready;
459 	
460 	lock(&l->l);
461 //print("qlock unlock %p @%#x by %p (owner %p)\n", l, pc, (*threadnow)(), l->owner);
462 	if(l->owner == 0){
463 		fprint(2, "%s: qunlock pc=0x%lux owner=%p self=%p oops\n",
464 			argv0, pc, l->owner, (*threadnow)());
465 		abort();
466 	}
467 	if((l->owner = ready = l->waiting.head) != nil)
468 		delthread(&l->waiting, l->owner);
469 	/*
470 	 * N.B. Cannot call _threadready() before unlocking l->l,
471 	 * because the thread we are readying might:
472 	 *	- be in another proc
473 	 *	- start running immediately
474 	 *	- and free l before we get a chance to run again
475 	 */
476 	unlock(&l->l);
477 	if(ready)
478 		_threadready(l->owner);
479 }
480 
481 static int
482 threadrlock(RWLock *l, int block, ulong pc)
483 {
484 	USED(pc);
485 
486 	lock(&l->l);
487 	if(l->writer == nil && l->wwaiting.head == nil){
488 		l->readers++;
489 		unlock(&l->l);
490 		return 1;
491 	}
492 	if(!block){
493 		unlock(&l->l);
494 		return 0;
495 	}
496 	addthread(&l->rwaiting, (*threadnow)());
497 	unlock(&l->l);
498 	_threadswitch();
499 	return 1;	
500 }
501 
502 static int
503 threadwlock(RWLock *l, int block, ulong pc)
504 {
505 	USED(pc);
506 
507 	lock(&l->l);
508 	if(l->writer == nil && l->readers == 0){
509 		l->writer = (*threadnow)();
510 		unlock(&l->l);
511 		return 1;
512 	}
513 	if(!block){
514 		unlock(&l->l);
515 		return 0;
516 	}
517 	addthread(&l->wwaiting, (*threadnow)());
518 	unlock(&l->l);
519 	_threadswitch();
520 	return 1;
521 }
522 
523 static void
524 threadrunlock(RWLock *l, ulong pc)
525 {
526 	_Thread *t;
527 
528 	USED(pc);
529 	t = nil;
530 	lock(&l->l);
531 	--l->readers;
532 	if(l->readers == 0 && (t = l->wwaiting.head) != nil){
533 		delthread(&l->wwaiting, t);
534 		l->writer = t;
535 	}
536 	unlock(&l->l);
537 	if(t)
538 		_threadready(t);
539 
540 }
541 
542 static void
543 threadwunlock(RWLock *l, ulong pc)
544 {
545 	_Thread *t;
546 
547 	USED(pc);
548 	lock(&l->l);
549 	l->writer = nil;
550 	assert(l->readers == 0);
551 	while((t = l->rwaiting.head) != nil){
552 		delthread(&l->rwaiting, t);
553 		l->readers++;
554 		_threadready(t);
555 	}
556 	t = nil;
557 	if(l->readers == 0 && (t = l->wwaiting.head) != nil){
558 		delthread(&l->wwaiting, t);
559 		l->writer = t;
560 	}
561 	unlock(&l->l);
562 	if(t)
563 		_threadready(t);
564 }
565 
566 /*
567  * sleep and wakeup
568  */
569 static void
570 threadrsleep(Rendez *r, ulong pc)
571 {
572 	addthread(&r->waiting, proc()->thread);
573 	qunlock(r->l);
574 	_threadswitch();
575 	qlock(r->l);
576 }
577 
578 static int
579 threadrwakeup(Rendez *r, int all, ulong pc)
580 {
581 	int i;
582 	_Thread *t;
583 
584 	for(i=0;; i++){
585 		if(i==1 && !all)
586 			break;
587 		if((t = r->waiting.head) == nil)
588 			break;
589 		delthread(&r->waiting, t);
590 		_threadready(t);	
591 	}
592 	return i;
593 }
594 
595 /*
596  * startup
597  */
598 
599 static int threadargc;
600 static char **threadargv;
601 int mainstacksize;
602 
603 static void
604 threadmainstart(void *v)
605 {
606 	USED(v);
607 
608 	/*
609 	 * N.B. This call to proc() is a program's first call (indirectly) to a
610 	 * pthreads function while executing on a non-pthreads-allocated
611 	 * stack.  If the pthreads implementation is using the stack pointer
612 	 * to locate the per-thread data, then this call will blow up.
613 	 * This means the pthread implementation is not suitable for
614 	 * running under libthread.  Time to write your own.  Sorry.
615 	 */
616 	threadmainproc = proc();
617 	threadmain(threadargc, threadargv);
618 }
619 
620 int
621 main(int argc, char **argv)
622 {
623 	Proc *p;
624 
625 	argv0 = argv[0];
626 
627 	if(getenv("NOLIBTHREADDAEMONIZE") == nil)
628 		_threadsetupdaemonize();
629 
630 	threadargc = argc;
631 	threadargv = argv;
632 
633 	/*
634 	 * Install locking routines into C library.
635 	 */
636 	_lock = _threadlock;
637 	_unlock = _threadunlock;
638 	_qlock = threadqlock;
639 	_qunlock = threadqunlock;
640 	_rlock = threadrlock;
641 	_runlock = threadrunlock;
642 	_wlock = threadwlock;
643 	_wunlock = threadwunlock;
644 	_rsleep = threadrsleep;
645 	_rwakeup = threadrwakeup;
646 	_notejmpbuf = threadnotejmp;
647 
648 	_pthreadinit();
649 	p = procalloc();
650 	p->mainproc = 1;
651 	_threadsetproc(p);
652 	if(mainstacksize == 0)
653 		mainstacksize = 256*1024;
654 	atnotify(threadinfo, 1);
655 	_threadcreate(p, threadmainstart, nil, mainstacksize);
656 	procscheduler(p);
657 	sysfatal("procscheduler returned in threadmain!");
658 	/* does not return */
659 	return 0;
660 }
661 
662 /*
663  * hooray for linked lists
664  */
665 static void
666 addthread(_Threadlist *l, _Thread *t)
667 {
668 	if(l->tail){
669 		l->tail->next = t;
670 		t->prev = l->tail;
671 	}else{
672 		l->head = t;
673 		t->prev = nil;
674 	}
675 	l->tail = t;
676 	t->next = nil;
677 }
678 
679 static void
680 delthread(_Threadlist *l, _Thread *t)
681 {
682 	if(t->prev)
683 		t->prev->next = t->next;
684 	else
685 		l->head = t->next;
686 	if(t->next)
687 		t->next->prev = t->prev;
688 	else
689 		l->tail = t->prev;
690 }
691 
692 static void
693 addthreadinproc(Proc *p, _Thread *t)
694 {
695 	_Threadlist *l;
696 
697 	l = &p->allthreads;
698 	if(l->tail){
699 		l->tail->allnext = t;
700 		t->allprev = l->tail;
701 	}else{
702 		l->head = t;
703 		t->allprev = nil;
704 	}
705 	l->tail = t;
706 	t->allnext = nil;
707 }
708 
709 static void
710 delthreadinproc(Proc *p, _Thread *t)
711 {
712 	_Threadlist *l;
713 
714 	l = &p->allthreads;
715 	if(t->allprev)
716 		t->allprev->allnext = t->allnext;
717 	else
718 		l->head = t->allnext;
719 	if(t->allnext)
720 		t->allnext->allprev = t->allprev;
721 	else
722 		l->tail = t->allprev;
723 }
724 
725 Proc *_threadprocs;
726 Lock _threadprocslock;
727 static Proc *_threadprocstail;
728 
729 static void
730 addproc(Proc *p)
731 {
732 	lock(&_threadprocslock);
733 	if(_threadprocstail){
734 		_threadprocstail->next = p;
735 		p->prev = _threadprocstail;
736 	}else{
737 		_threadprocs = p;
738 		p->prev = nil;
739 	}
740 	_threadprocstail = p;
741 	p->next = nil;
742 	unlock(&_threadprocslock);
743 }
744 
745 static void
746 delproc(Proc *p)
747 {
748 	lock(&_threadprocslock);
749 	if(p->prev)
750 		p->prev->next = p->next;
751 	else
752 		_threadprocs = p->next;
753 	if(p->next)
754 		p->next->prev = p->prev;
755 	else
756 		_threadprocstail = p->prev;
757 	unlock(&_threadprocslock);
758 }
759 
760 /* 
761  * notify - for now just use the usual mechanisms
762  */
763 void
764 threadnotify(int (*f)(void*, char*), int in)
765 {
766 	atnotify(f, in);
767 }
768 
769 static int
770 onrunqueue(Proc *p, _Thread *t)
771 {
772 	_Thread *tt;
773 
774 	for(tt=p->runqueue.head; tt; tt=tt->next)
775 		if(tt == t)
776 			return 1;
777 	return 0;
778 }
779 
780 /*
781  * print state - called from SIGINFO
782  */
783 static int
784 threadinfo(void *v, char *s)
785 {
786 	Proc *p;
787 	_Thread *t;
788 
789 	if(strcmp(s, "quit") != 0 && strcmp(s, "sys: status request") != 0)
790 		return 0;
791 
792 	for(p=_threadprocs; p; p=p->next){
793 		fprint(2, "proc %p %s%s\n", (void*)p->osprocid, p->msg,
794 			p->sysproc ? " (sysproc)": "");
795 		for(t=p->allthreads.head; t; t=t->allnext){
796 			fprint(2, "\tthread %d %s: %s %s\n", 
797 				t->id, 
798 				t == p->thread ? "Running" : 
799 				onrunqueue(p, t) ? "Ready" : "Sleeping",
800 				t->state, t->name);
801 		}
802 	}
803 	return 1;
804 }
805 
806