1 // Copyright 2009 The Go Authors. All rights reserved.
2 // Use of this source code is governed by a BSD-style
3 // license that can be found in the LICENSE file.
10 static int32 debug = 0;
12 typedef struct WaitQ WaitQ;
13 typedef struct SudoG SudoG;
14 typedef struct Select Select;
15 typedef struct Scase Scase;
17 typedef struct __go_type_descriptor Type;
18 typedef struct __go_channel_type ChanType;
22 G* g; // g and selgen constitute
23 uint32 selgen; // a weak pointer to g
25 byte* elem; // data element
36 uint32 qcount; // total data in the q
37 uint32 dataqsiz; // size of the circular q
41 uint32 sendx; // send index
42 uint32 recvx; // receive index
43 WaitQ recvq; // list of recv waiters
44 WaitQ sendq; // list of send waiters
48 // Buffer follows Hchan immediately in memory.
49 // chanbuf(c, i) is pointer to the i'th slot in the buffer.
50 #define chanbuf(c, i) ((byte*)((c)+1)+(uintptr)(c)->elemsize*(i))
62 SudoG sg; // must be first member (cast to Scase)
65 uint16 index; // index to return
66 bool* receivedp; // pointer to received bool (recv2)
71 uint16 tcase; // total count of scase[]
72 uint16 ncase; // currently filled scase[]
73 uint16* pollorder; // case poll order
74 Hchan** lockorder; // channel lock order
75 Scase scase[1]; // one per case (in order of appearance)
78 static void dequeueg(WaitQ*);
79 static SudoG* dequeue(WaitQ*);
80 static void enqueue(WaitQ*, SudoG*);
83 runtime_makechan_c(ChanType *t, int64 hint)
89 elem = t->__element_type;
91 if(hint < 0 || (int32)hint != hint || (elem->__size > 0 && (uintptr)hint > ((uintptr)-1) / elem->__size))
92 runtime_panicstring("makechan: size out of range");
96 // allocate memory in one call
97 c = (Hchan*)runtime_mal(n + hint*elem->__size);
98 c->elemsize = elem->__size;
99 c->elemalign = elem->__align;
103 runtime_printf("makechan: chan=%p; elemsize=%lld; elemalign=%d; dataqsiz=%d\n",
104 c, (long long)elem->__size, elem->__align, c->dataqsiz);
110 // func makechan(typ *ChanType, size uint32) (chan)
111 uintptr reflect_makechan(ChanType *, uint32)
112 asm ("libgo_reflect.reflect.makechan");
115 reflect_makechan(ChanType *t, uint32 size)
120 c = runtime_makechan_c(t, size);
121 ret = runtime_mal(sizeof(void*));
122 __builtin_memcpy(ret, &c, sizeof(void*));
126 // makechan(t *ChanType, hint int64) (hchan *chan any);
128 __go_new_channel(ChanType *t, uintptr hint)
130 return runtime_makechan_c(t, hint);
134 __go_new_channel_big(ChanType *t, uint64 hint)
136 return runtime_makechan_c(t, hint);
140 * generic single channel send/recv
141 * if the bool pointer is nil,
142 * then the full exchange will
143 * occur. if pres is not nil,
144 * then the protocol will not
145 * sleep but return if it could
148 * sleep can wake up with g->param == nil
149 * when a channel involved in the sleep has
150 * been closed. it is easiest to loop and re-run
151 * the operation; we'll see that it's now closed.
154 runtime_chansend(ChanType *t, Hchan *c, byte *ep, bool *pres)
169 g->status = Gwaiting;
170 g->waitreason = "chan send (nil chan)";
172 return; // not reached
175 if(runtime_gcwaiting)
179 runtime_printf("chansend: chan=%p\n", c);
189 sg = dequeue(&c->recvq);
196 runtime_memmove(sg->elem, ep, c->elemsize);
212 mysg.selgen = NOSELGEN;
214 g->status = Gwaiting;
215 g->waitreason = "chan send";
216 enqueue(&c->sendq, &mysg);
220 if(g->param == nil) {
223 runtime_throw("chansend: spurious wakeup");
233 if(c->qcount >= c->dataqsiz) {
241 mysg.selgen = NOSELGEN;
242 g->status = Gwaiting;
243 g->waitreason = "chan send";
244 enqueue(&c->sendq, &mysg);
251 runtime_memmove(chanbuf(c, c->sendx), ep, c->elemsize);
252 if(++c->sendx == c->dataqsiz)
256 sg = dequeue(&c->recvq);
269 runtime_panicstring("send on closed channel");
274 runtime_chanrecv(ChanType *t, Hchan* c, byte *ep, bool *selected, bool *received)
281 if(runtime_gcwaiting)
285 runtime_printf("chanrecv: chan=%p\n", c);
291 if(selected != nil) {
295 g->status = Gwaiting;
296 g->waitreason = "chan receive (nil chan)";
298 return; // not reached
308 sg = dequeue(&c->sendq);
313 runtime_memmove(ep, sg->elem, c->elemsize);
325 if(selected != nil) {
333 mysg.selgen = NOSELGEN;
335 g->status = Gwaiting;
336 g->waitreason = "chan receive";
337 enqueue(&c->recvq, &mysg);
341 if(g->param == nil) {
344 runtime_throw("chanrecv: spurious wakeup");
357 if(selected != nil) {
366 mysg.selgen = NOSELGEN;
367 g->status = Gwaiting;
368 g->waitreason = "chan receive";
369 enqueue(&c->recvq, &mysg);
377 runtime_memmove(ep, chanbuf(c, c->recvx), c->elemsize);
378 runtime_memclr(chanbuf(c, c->recvx), c->elemsize);
379 if(++c->recvx == c->dataqsiz)
383 sg = dequeue(&c->sendq);
399 runtime_memclr(ep, c->elemsize);
407 // The compiler generates a call to __go_send_small to send a value 8
410 __go_send_small(ChanType *t, Hchan* c, uint64 val)
412 byte b[sizeof(uint64)];
414 runtime_memclr(b, sizeof(uint64));
415 __builtin_memcpy(b, &val, t->__element_type->__size);
416 runtime_chansend(t, c, b, nil);
419 // The compiler generates a call to __go_send_big to send a value
420 // larger than 8 bytes or smaller.
422 __go_send_big(ChanType *t, Hchan* c, byte* p)
424 runtime_chansend(t, c, p, nil);
427 // The compiler generates a call to __go_receive_small to receive a
428 // value 8 bytes or smaller.
430 __go_receive_small(ChanType *t, Hchan* c)
433 byte b[sizeof(uint64)];
438 runtime_chanrecv(t, c, u.b, nil, nil);
442 // The compiler generates a call to __go_receive_big to receive a
443 // value larger than 8 bytes.
445 __go_receive_big(ChanType *t, Hchan* c, byte* p)
447 runtime_chanrecv(t, c, p, nil, nil);
450 _Bool runtime_chanrecv2(ChanType *t, Hchan* c, byte* p)
451 __asm__("runtime.chanrecv2");
454 runtime_chanrecv2(ChanType *t, Hchan* c, byte* p)
458 runtime_chanrecv(t, c, p, nil, &received);
462 // func selectnbsend(c chan any, elem any) bool
464 // compiler implements
475 // if selectnbsend(c, v) {
482 runtime_selectnbsend(ChanType *t, Hchan *c, byte *p)
486 runtime_chansend(t, c, p, &res);
490 // func selectnbrecv(elem *any, c chan any) bool
492 // compiler implements
503 // if selectnbrecv(&v, c) {
510 runtime_selectnbrecv(ChanType *t, byte *v, Hchan *c)
514 runtime_chanrecv(t, c, v, &selected, nil);
518 // func selectnbrecv2(elem *any, ok *bool, c chan any) bool
520 // compiler implements
531 // if c != nil && selectnbrecv2(&v, &ok, c) {
538 runtime_selectnbrecv2(ChanType *t, byte *v, _Bool *received, Hchan *c)
544 runtime_chanrecv(t, c, v, &selected, received == nil ? nil : &r);
551 // func chansend(c chan, val iword, nb bool) (selected bool)
552 // where an iword is the same word an interface value would use:
553 // the actual data if it fits, or else a pointer to the data.
555 _Bool reflect_chansend(ChanType *, Hchan *, uintptr, _Bool)
556 __asm__("libgo_reflect.reflect.chansend");
559 reflect_chansend(ChanType *t, Hchan *c, uintptr val, _Bool nb)
567 sp = (bool*)&selected;
572 if(__go_is_pointer_type(t->__element_type))
576 runtime_chansend(t, c, vp, sp);
581 // func chanrecv(c chan, nb bool) (val iword, selected, received bool)
582 // where an iword is the same word an interface value would use:
583 // the actual data if it fits, or else a pointer to the data.
592 struct chanrecv_ret reflect_chanrecv(ChanType *, Hchan *, _Bool)
593 __asm__("libgo_reflect.reflect.chanrecv");
596 reflect_chanrecv(ChanType *t, Hchan *c, _Bool nb)
598 struct chanrecv_ret ret;
612 if(__go_is_pointer_type(t->__element_type)) {
613 vp = (byte*)&ret.val;
615 vp = runtime_mal(t->__element_type->__size);
616 ret.val = (uintptr)vp;
618 runtime_chanrecv(t, c, vp, sp, &received);
620 ret.selected = selected;
621 ret.received = received;
625 static void newselect(int32, Select**);
627 // newselect(size uint32) (sel *byte);
629 void* runtime_newselect(int) __asm__("runtime.newselect");
632 runtime_newselect(int size)
636 newselect(size, &sel);
641 newselect(int32 size, Select **selp)
650 sel = runtime_mal(sizeof(*sel) +
651 n*sizeof(sel->scase[0]) +
652 size*sizeof(sel->lockorder[0]) +
653 size*sizeof(sel->pollorder[0]));
657 sel->pollorder = (void*)(sel->scase + size);
658 sel->lockorder = (void*)(sel->pollorder + size);
662 runtime_printf("newselect s=%p size=%d\n", sel, size);
665 // cut in half to give stack a chance to split
666 static void selectsend(Select *sel, Hchan *c, int index, void *elem);
668 // selectsend(sel *byte, hchan *chan any, elem *any) (selected bool);
670 void runtime_selectsend(Select *, Hchan *, void *, int)
671 __asm__("runtime.selectsend");
674 runtime_selectsend(Select *sel, Hchan *c, void *elem, int index)
676 // nil cases do not compete
680 selectsend(sel, c, index, elem);
684 selectsend(Select *sel, Hchan *c, int index, void *elem)
691 runtime_throw("selectsend: too many cases");
693 cas = &sel->scase[i];
697 cas->kind = CaseSend;
701 runtime_printf("selectsend s=%p index=%d chan=%p\n",
702 sel, cas->index, cas->chan);
705 // cut in half to give stack a chance to split
706 static void selectrecv(Select *sel, Hchan *c, int index, void *elem, bool*);
708 // selectrecv(sel *byte, hchan *chan any, elem *any) (selected bool);
710 void runtime_selectrecv(Select *, Hchan *, void *, int)
711 __asm__("runtime.selectrecv");
714 runtime_selectrecv(Select *sel, Hchan *c, void *elem, int index)
716 // nil cases do not compete
720 selectrecv(sel, c, index, elem, nil);
723 // selectrecv2(sel *byte, hchan *chan any, elem *any, received *bool) (selected bool);
725 void runtime_selectrecv2(Select *, Hchan *, void *, bool *, int)
726 __asm__("runtime.selectrecv2");
729 runtime_selectrecv2(Select *sel, Hchan *c, void *elem, bool *received, int index)
731 // nil cases do not compete
735 selectrecv(sel, c, index, elem, received);
739 selectrecv(Select *sel, Hchan *c, int index, void *elem, bool *received)
746 runtime_throw("selectrecv: too many cases");
748 cas = &sel->scase[i];
752 cas->kind = CaseRecv;
754 cas->receivedp = received;
757 runtime_printf("selectrecv s=%p index=%d chan=%p\n",
758 sel, cas->index, cas->chan);
761 // cut in half to give stack a chance to split
762 static void selectdefault(Select*, int);
764 // selectdefault(sel *byte) (selected bool);
766 void runtime_selectdefault(Select *, int) __asm__("runtime.selectdefault");
769 runtime_selectdefault(Select *sel, int index)
771 selectdefault(sel, index);
775 selectdefault(Select *sel, int index)
782 runtime_throw("selectdefault: too many cases");
784 cas = &sel->scase[i];
788 cas->kind = CaseDefault;
791 runtime_printf("selectdefault s=%p index=%d\n",
802 for(i=0; i<sel->ncase; i++) {
803 c0 = sel->lockorder[i];
805 c = sel->lockorder[i];
812 selunlock(Select *sel)
818 for(i=sel->ncase; i-->0;) {
819 c0 = sel->lockorder[i];
833 g->status = Gwaiting; // forever
834 g->waitreason = "select (no cases)";
838 static int selectgo(Select**);
840 // selectgo(sel *byte);
842 int runtime_selectgo(Select *) __asm__("runtime.selectgo");
845 runtime_selectgo(Select *sel)
847 return selectgo(&sel);
851 selectgo(Select **selp)
863 if(runtime_gcwaiting)
867 runtime_printf("select: sel=%p\n", sel);
871 // The compiler rewrites selects that statically have
872 // only 0 or 1 cases plus default into simpler constructs.
873 // The only way we can end up with such small sel->ncase
874 // values here is for a larger select in which most channels
875 // have been nilled out. The general code handles those
876 // cases correctly, and they are rare enough not to bother
877 // optimizing (and needing to test).
879 // generate permuted order
880 for(i=0; i<sel->ncase; i++)
881 sel->pollorder[i] = i;
882 for(i=1; i<sel->ncase; i++) {
883 o = sel->pollorder[i];
884 j = runtime_fastrand1()%(i+1);
885 sel->pollorder[i] = sel->pollorder[j];
886 sel->pollorder[j] = o;
889 // sort the cases by Hchan address to get the locking order.
890 for(i=0; i<sel->ncase; i++) {
891 c = sel->scase[i].chan;
892 for(j=i; j>0 && sel->lockorder[j-1] >= c; j--)
893 sel->lockorder[j] = sel->lockorder[j-1];
894 sel->lockorder[j] = c;
899 // pass 1 - look for something already waiting
901 for(i=0; i<sel->ncase; i++) {
902 o = sel->pollorder[i];
903 cas = &sel->scase[o];
908 if(c->dataqsiz > 0) {
912 sg = dequeue(&c->sendq);
923 if(c->dataqsiz > 0) {
924 if(c->qcount < c->dataqsiz)
927 sg = dequeue(&c->recvq);
946 // pass 2 - enqueue on all chans
947 for(i=0; i<sel->ncase; i++) {
948 o = sel->pollorder[i];
949 cas = &sel->scase[o];
953 sg->selgen = g->selgen;
957 enqueue(&c->recvq, sg);
961 enqueue(&c->sendq, sg);
967 g->status = Gwaiting;
968 g->waitreason = "select";
975 // pass 3 - dequeue from unsuccessful chans
976 // otherwise they stack up on quiet channels
977 for(i=0; i<sel->ncase; i++) {
978 cas = &sel->scase[i];
979 if(cas != (Scase*)sg) {
981 if(cas->kind == CaseSend)
995 runtime_throw("selectgo: shouldnt happen");
998 runtime_printf("wait-return: sel=%p c=%p cas=%p kind=%d\n",
999 sel, c, cas, cas->kind);
1001 if(cas->kind == CaseRecv) {
1002 if(cas->receivedp != nil)
1003 *cas->receivedp = true;
1010 // can receive from buffer
1011 if(cas->receivedp != nil)
1012 *cas->receivedp = true;
1013 if(cas->sg.elem != nil)
1014 runtime_memmove(cas->sg.elem, chanbuf(c, c->recvx), c->elemsize);
1015 runtime_memclr(chanbuf(c, c->recvx), c->elemsize);
1016 if(++c->recvx == c->dataqsiz)
1019 sg = dequeue(&c->sendq);
1030 // can send to buffer
1031 runtime_memmove(chanbuf(c, c->sendx), cas->sg.elem, c->elemsize);
1032 if(++c->sendx == c->dataqsiz)
1035 sg = dequeue(&c->recvq);
1046 // can receive from sleeping sender (sg)
1049 runtime_printf("syncrecv: sel=%p c=%p o=%d\n", sel, c, o);
1050 if(cas->receivedp != nil)
1051 *cas->receivedp = true;
1052 if(cas->sg.elem != nil)
1053 runtime_memmove(cas->sg.elem, sg->elem, c->elemsize);
1060 // read at end of closed channel
1062 if(cas->receivedp != nil)
1063 *cas->receivedp = false;
1064 if(cas->sg.elem != nil)
1065 runtime_memclr(cas->sg.elem, c->elemsize);
1069 // can send to sleeping receiver (sg)
1072 runtime_printf("syncsend: sel=%p c=%p o=%d\n", sel, c, o);
1074 runtime_memmove(sg->elem, cas->sg.elem, c->elemsize);
1080 // return index corresponding to chosen case
1086 // send on closed channel
1088 runtime_panicstring("send on closed channel");
1089 return 0; // not reached
1092 // closechan(sel *byte);
1094 runtime_closechan(Hchan *c)
1100 runtime_panicstring("close of nil channel");
1102 if(runtime_gcwaiting)
1108 runtime_panicstring("close of closed channel");
1113 // release all readers
1115 sg = dequeue(&c->recvq);
1123 // release all writers
1125 sg = dequeue(&c->sendq);
1137 __go_builtin_close(Hchan *c)
1139 runtime_closechan(c);
1143 // func chanclose(c chan)
1145 void reflect_chanclose(uintptr) __asm__("libgo_reflect.reflect.chanclose");
1148 reflect_chanclose(uintptr c)
1150 runtime_closechan((Hchan*)c);
1154 // func chanlen(c chan) (len int32)
1156 int32 reflect_chanlen(uintptr) __asm__("libgo_reflect.reflect.chanlen");
1159 reflect_chanlen(uintptr ca)
1173 __go_chan_len(Hchan *c)
1175 return reflect_chanlen((uintptr)c);
1179 // func chancap(c chan) (cap int32)
1181 int32 reflect_chancap(uintptr) __asm__("libgo_reflect.reflect.chancap");
1184 reflect_chancap(uintptr ca)
1198 __go_chan_cap(Hchan *c)
1200 return reflect_chancap((uintptr)c);
1212 q->first = sgp->link;
1214 // if sgp is stale, ignore it
1215 if(sgp->selgen != NOSELGEN &&
1216 (sgp->selgen != sgp->g->selgen ||
1217 !runtime_cas(&sgp->g->selgen, sgp->selgen, sgp->selgen + 2))) {
1218 //prints("INVALID PSEUDOG POINTER\n");
1228 SudoG **l, *sgp, *prevsgp;
1233 for(l=&q->first; (sgp=*l) != nil; l=&sgp->link, prevsgp=sgp) {
1244 enqueue(WaitQ *q, SudoG *sgp)
1247 if(q->first == nil) {
1252 q->last->link = sgp;