| // Copyright 2009 The Go Authors. All rights reserved. |
| // Use of this source code is governed by a BSD-style |
| // license that can be found in the LICENSE file. |
| |
| package runtime |
| #include "runtime.h" |
| #include "arch_GOARCH.h" |
| #include "type.h" |
| #include "race.h" |
| #include "malloc.h" |
| #include "chan.h" |
| #include "mgc0.h" |
| #include "typekind.h" |
| #include "../../cmd/ld/textflag.h" |
| |
| static void dequeueg(WaitQ*); |
| static SudoG* dequeue(WaitQ*); |
| static void enqueue(WaitQ*, SudoG*); |
| static void racesync(Hchan*, SudoG*); |
| |
| /* |
| * generic single channel send/recv |
| * if the bool pointer is nil, |
| * then the full exchange will |
| * occur. if pres is not nil, |
| * then the protocol will not |
| * sleep but return if it could |
| * not complete. |
| * |
| * sleep can wake up with g->param == nil |
| * when a channel involved in the sleep has |
| * been closed. it is easiest to loop and re-run |
| * the operation; we'll see that it's now closed. |
| */ |
| static bool |
| chansend(ChanType *t, Hchan *c, byte *ep, bool block, void *pc) |
| { |
| SudoG *sg; |
| SudoG mysg; |
| G* gp; |
| int64 t0; |
| |
| if(raceenabled) |
| runtime·racereadobjectpc(ep, t->elem, runtime·getcallerpc(&t), chansend); |
| |
| if(c == nil) { |
| USED(t); |
| if(!block) |
| return false; |
| runtime·park(nil, nil, runtime·gostringnocopy((byte*)"chan send (nil chan)")); |
| return false; // not reached |
| } |
| |
| if(debug) { |
| runtime·printf("chansend: chan=%p; elem=", c); |
| c->elemtype->alg->print(c->elemsize, ep); |
| runtime·prints("\n"); |
| } |
| |
| if(raceenabled) |
| runtime·racereadpc(c, pc, chansend); |
| |
| // Fast path: check for failed non-blocking operation without acquiring the lock. |
| // |
| // After observing that the channel is not closed, we observe that the channel is |
| // not ready for sending. Each of these observations is a single word-sized read |
| // (first c.closed and second c.recvq.first or c.qcount depending on kind of channel). |
| // Because a closed channel cannot transition from 'ready for sending' to |
| // 'not ready for sending', even if the channel is closed between the two observations, |
| // they imply a moment between the two when the channel was both not yet closed |
| // and not ready for sending. We behave as if we observed the channel at that moment, |
| // and report that the send cannot proceed. |
| // |
| // It is okay if the reads are reordered here: if we observe that the channel is not |
| // ready for sending and then observe that it is not closed, that implies that the |
| // channel wasn't closed during the first observation. |
| if(!block && !c->closed && ((c->dataqsiz == 0 && c->recvq.first == nil) || |
| (c->dataqsiz > 0 && c->qcount == c->dataqsiz))) |
| return false; |
| |
| t0 = 0; |
| mysg.releasetime = 0; |
| if(runtime·blockprofilerate > 0) { |
| t0 = runtime·cputicks(); |
| mysg.releasetime = -1; |
| } |
| |
| runtime·lock(&c->lock); |
| if(c->closed) |
| goto closed; |
| |
| if(c->dataqsiz > 0) |
| goto asynch; |
| |
| sg = dequeue(&c->recvq); |
| if(sg != nil) { |
| if(raceenabled) |
| racesync(c, sg); |
| runtime·unlock(&c->lock); |
| |
| gp = sg->g; |
| gp->param = sg; |
| if(sg->elem != nil) |
| c->elemtype->alg->copy(c->elemsize, sg->elem, ep); |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| return true; |
| } |
| |
| if(!block) { |
| runtime·unlock(&c->lock); |
| return false; |
| } |
| |
| mysg.elem = ep; |
| mysg.g = g; |
| mysg.selectdone = nil; |
| g->param = nil; |
| enqueue(&c->sendq, &mysg); |
| runtime·parkunlock(&c->lock, runtime·gostringnocopy((byte*)"chan send")); |
| |
| if(g->param == nil) { |
| runtime·lock(&c->lock); |
| if(!c->closed) |
| runtime·throw("chansend: spurious wakeup"); |
| goto closed; |
| } |
| |
| if(mysg.releasetime > 0) |
| runtime·blockevent(mysg.releasetime - t0, 2); |
| |
| return true; |
| |
| asynch: |
| if(c->closed) |
| goto closed; |
| |
| if(c->qcount >= c->dataqsiz) { |
| if(!block) { |
| runtime·unlock(&c->lock); |
| return false; |
| } |
| mysg.g = g; |
| mysg.elem = nil; |
| mysg.selectdone = nil; |
| enqueue(&c->sendq, &mysg); |
| runtime·parkunlock(&c->lock, runtime·gostringnocopy((byte*)"chan send")); |
| |
| runtime·lock(&c->lock); |
| goto asynch; |
| } |
| |
| if(raceenabled) { |
| runtime·raceacquire(chanbuf(c, c->sendx)); |
| runtime·racerelease(chanbuf(c, c->sendx)); |
| } |
| |
| c->elemtype->alg->copy(c->elemsize, chanbuf(c, c->sendx), ep); |
| if(++c->sendx == c->dataqsiz) |
| c->sendx = 0; |
| c->qcount++; |
| |
| sg = dequeue(&c->recvq); |
| if(sg != nil) { |
| gp = sg->g; |
| runtime·unlock(&c->lock); |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| } else |
| runtime·unlock(&c->lock); |
| if(mysg.releasetime > 0) |
| runtime·blockevent(mysg.releasetime - t0, 2); |
| return true; |
| |
| closed: |
| runtime·unlock(&c->lock); |
| runtime·panicstring("send on closed channel"); |
| return false; // not reached |
| } |
| |
| |
| static bool |
| chanrecv(ChanType *t, Hchan* c, byte *ep, bool block, bool *received) |
| { |
| SudoG *sg; |
| SudoG mysg; |
| G *gp; |
| int64 t0; |
| |
| // raceenabled: don't need to check ep, as it is always on the stack. |
| |
| if(debug) |
| runtime·printf("chanrecv: chan=%p\n", c); |
| |
| if(c == nil) { |
| USED(t); |
| if(!block) |
| return false; |
| runtime·park(nil, nil, runtime·gostringnocopy((byte*)"chan receive (nil chan)")); |
| return false; // not reached |
| } |
| |
| // Fast path: check for failed non-blocking operation without acquiring the lock. |
| // |
| // After observing that the channel is not ready for receiving, we observe that the |
| // channel is not closed. Each of these observations is a single word-sized read |
| // (first c.sendq.first or c.qcount, and second c.closed). |
| // Because a channel cannot be reopened, the later observation of the channel |
| // being not closed implies that it was also not closed at the moment of the |
| // first observation. We behave as if we observed the channel at that moment |
| // and report that the receive cannot proceed. |
| // |
| // The order of operations is important here: reversing the operations can lead to |
| // incorrect behavior when racing with a close. |
| if(!block && ((c->dataqsiz == 0 && c->sendq.first == nil) || |
| (c->dataqsiz > 0 && runtime·atomicloadp((void**)&c->qcount) == 0)) && |
| !runtime·atomicload(&c->closed)) |
| return false; |
| |
| t0 = 0; |
| mysg.releasetime = 0; |
| if(runtime·blockprofilerate > 0) { |
| t0 = runtime·cputicks(); |
| mysg.releasetime = -1; |
| } |
| |
| runtime·lock(&c->lock); |
| if(c->dataqsiz > 0) |
| goto asynch; |
| |
| if(c->closed) |
| goto closed; |
| |
| sg = dequeue(&c->sendq); |
| if(sg != nil) { |
| if(raceenabled) |
| racesync(c, sg); |
| runtime·unlock(&c->lock); |
| |
| if(ep != nil) |
| c->elemtype->alg->copy(c->elemsize, ep, sg->elem); |
| gp = sg->g; |
| gp->param = sg; |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| |
| if(received != nil) |
| *received = true; |
| return true; |
| } |
| |
| if(!block) { |
| runtime·unlock(&c->lock); |
| return false; |
| } |
| |
| mysg.elem = ep; |
| mysg.g = g; |
| mysg.selectdone = nil; |
| g->param = nil; |
| enqueue(&c->recvq, &mysg); |
| runtime·parkunlock(&c->lock, runtime·gostringnocopy((byte*)"chan receive")); |
| |
| if(g->param == nil) { |
| runtime·lock(&c->lock); |
| if(!c->closed) |
| runtime·throw("chanrecv: spurious wakeup"); |
| goto closed; |
| } |
| |
| if(received != nil) |
| *received = true; |
| if(mysg.releasetime > 0) |
| runtime·blockevent(mysg.releasetime - t0, 2); |
| return true; |
| |
| asynch: |
| if(c->qcount <= 0) { |
| if(c->closed) |
| goto closed; |
| |
| if(!block) { |
| runtime·unlock(&c->lock); |
| if(received != nil) |
| *received = false; |
| return false; |
| } |
| mysg.g = g; |
| mysg.elem = nil; |
| mysg.selectdone = nil; |
| enqueue(&c->recvq, &mysg); |
| runtime·parkunlock(&c->lock, runtime·gostringnocopy((byte*)"chan receive")); |
| |
| runtime·lock(&c->lock); |
| goto asynch; |
| } |
| |
| if(raceenabled) { |
| runtime·raceacquire(chanbuf(c, c->recvx)); |
| runtime·racerelease(chanbuf(c, c->recvx)); |
| } |
| |
| if(ep != nil) |
| c->elemtype->alg->copy(c->elemsize, ep, chanbuf(c, c->recvx)); |
| c->elemtype->alg->copy(c->elemsize, chanbuf(c, c->recvx), nil); |
| if(++c->recvx == c->dataqsiz) |
| c->recvx = 0; |
| c->qcount--; |
| |
| sg = dequeue(&c->sendq); |
| if(sg != nil) { |
| gp = sg->g; |
| runtime·unlock(&c->lock); |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| } else |
| runtime·unlock(&c->lock); |
| |
| if(received != nil) |
| *received = true; |
| if(mysg.releasetime > 0) |
| runtime·blockevent(mysg.releasetime - t0, 2); |
| return true; |
| |
| closed: |
| if(ep != nil) |
| c->elemtype->alg->copy(c->elemsize, ep, nil); |
| if(received != nil) |
| *received = false; |
| if(raceenabled) |
| runtime·raceacquire(c); |
| runtime·unlock(&c->lock); |
| if(mysg.releasetime > 0) |
| runtime·blockevent(mysg.releasetime - t0, 2); |
| return true; |
| } |
| |
| #pragma textflag NOSPLIT |
| func chanrecv1(t *ChanType, c *Hchan, elem *byte) { |
| chanrecv(t, c, elem, true, nil); |
| } |
| |
| // chanrecv2(hchan *chan any, elem *any) (received bool); |
| #pragma textflag NOSPLIT |
| func chanrecv2(t *ChanType, c *Hchan, elem *byte) (received bool) { |
| chanrecv(t, c, elem, true, &received); |
| } |
| |
| // compiler implements |
| // |
| // select { |
| // case c <- v: |
| // ... foo |
| // default: |
| // ... bar |
| // } |
| // |
| // as |
| // |
| // if selectnbsend(c, v) { |
| // ... foo |
| // } else { |
| // ... bar |
| // } |
| // |
| #pragma textflag NOSPLIT |
| func selectnbsend(t *ChanType, c *Hchan, elem *byte) (selected bool) { |
| selected = chansend(t, c, elem, false, runtime·getcallerpc(&t)); |
| } |
| |
| // compiler implements |
| // |
| // select { |
| // case v = <-c: |
| // ... foo |
| // default: |
| // ... bar |
| // } |
| // |
| // as |
| // |
| // if selectnbrecv(&v, c) { |
| // ... foo |
| // } else { |
| // ... bar |
| // } |
| // |
| #pragma textflag NOSPLIT |
| func selectnbrecv(t *ChanType, elem *byte, c *Hchan) (selected bool) { |
| selected = chanrecv(t, c, elem, false, nil); |
| } |
| |
| // compiler implements |
| // |
| // select { |
| // case v, ok = <-c: |
| // ... foo |
| // default: |
| // ... bar |
| // } |
| // |
| // as |
| // |
| // if c != nil && selectnbrecv2(&v, &ok, c) { |
| // ... foo |
| // } else { |
| // ... bar |
| // } |
| // |
| #pragma textflag NOSPLIT |
| func selectnbrecv2(t *ChanType, elem *byte, received *bool, c *Hchan) (selected bool) { |
| selected = chanrecv(t, c, elem, false, received); |
| } |
| |
| #pragma textflag NOSPLIT |
| func reflect·chansend(t *ChanType, c *Hchan, elem *byte, nb bool) (selected bool) { |
| selected = chansend(t, c, elem, !nb, runtime·getcallerpc(&t)); |
| } |
| |
| func reflect·chanrecv(t *ChanType, c *Hchan, nb bool, elem *byte) (selected bool, received bool) { |
| received = false; |
| selected = chanrecv(t, c, elem, !nb, &received); |
| } |
| |
| static int64 |
| selectsize(int32 size) |
| { |
| Select *sel; |
| int64 selsize; |
| |
| selsize = sizeof(*sel) + |
| (size-1)*sizeof(sel->scase[0]) + |
| size*sizeof(sel->lockorder[0]) + |
| size*sizeof(sel->pollorder[0]); |
| return ROUND(selsize, Int64Align); |
| } |
| |
| #pragma textflag NOSPLIT |
| func newselect(sel *Select, selsize int64, size int32) { |
| if(selsize != selectsize(size)) { |
| runtime·printf("runtime: bad select size %D, want %D\n", selsize, selectsize(size)); |
| runtime·throw("bad select size"); |
| } |
| sel->tcase = size; |
| sel->ncase = 0; |
| sel->lockorder = (void*)(sel->scase + size); |
| sel->pollorder = (void*)(sel->lockorder + size); |
| |
| if(debug) |
| runtime·printf("newselect s=%p size=%d\n", sel, size); |
| } |
| |
| // cut in half to give stack a chance to split |
| static void selectsend(Select *sel, Hchan *c, void *pc, void *elem, int32 so); |
| |
| #pragma textflag NOSPLIT |
| func selectsend(sel *Select, c *Hchan, elem *byte) (selected bool) { |
| selected = false; |
| |
| // nil cases do not compete |
| if(c != nil) |
| selectsend(sel, c, runtime·getcallerpc(&sel), elem, (byte*)&selected - (byte*)&sel); |
| } |
| |
| static void |
| selectsend(Select *sel, Hchan *c, void *pc, void *elem, int32 so) |
| { |
| int32 i; |
| Scase *cas; |
| |
| i = sel->ncase; |
| if(i >= sel->tcase) |
| runtime·throw("selectsend: too many cases"); |
| sel->ncase = i+1; |
| cas = &sel->scase[i]; |
| |
| cas->pc = pc; |
| cas->chan = c; |
| cas->so = so; |
| cas->kind = CaseSend; |
| cas->sg.elem = elem; |
| |
| if(debug) |
| runtime·printf("selectsend s=%p pc=%p chan=%p so=%d\n", |
| sel, cas->pc, cas->chan, cas->so); |
| } |
| |
| // cut in half to give stack a chance to split |
| static void selectrecv(Select *sel, Hchan *c, void *pc, void *elem, bool*, int32 so); |
| |
| #pragma textflag NOSPLIT |
| func selectrecv(sel *Select, c *Hchan, elem *byte) (selected bool) { |
| selected = false; |
| |
| // nil cases do not compete |
| if(c != nil) |
| selectrecv(sel, c, runtime·getcallerpc(&sel), elem, nil, (byte*)&selected - (byte*)&sel); |
| } |
| |
| #pragma textflag NOSPLIT |
| func selectrecv2(sel *Select, c *Hchan, elem *byte, received *bool) (selected bool) { |
| selected = false; |
| |
| // nil cases do not compete |
| if(c != nil) |
| selectrecv(sel, c, runtime·getcallerpc(&sel), elem, received, (byte*)&selected - (byte*)&sel); |
| } |
| |
| static void |
| selectrecv(Select *sel, Hchan *c, void *pc, void *elem, bool *received, int32 so) |
| { |
| int32 i; |
| Scase *cas; |
| |
| i = sel->ncase; |
| if(i >= sel->tcase) |
| runtime·throw("selectrecv: too many cases"); |
| sel->ncase = i+1; |
| cas = &sel->scase[i]; |
| cas->pc = pc; |
| cas->chan = c; |
| |
| cas->so = so; |
| cas->kind = CaseRecv; |
| cas->sg.elem = elem; |
| cas->receivedp = received; |
| |
| if(debug) |
| runtime·printf("selectrecv s=%p pc=%p chan=%p so=%d\n", |
| sel, cas->pc, cas->chan, cas->so); |
| } |
| |
| // cut in half to give stack a chance to split |
| static void selectdefault(Select*, void*, int32); |
| |
| #pragma textflag NOSPLIT |
| func selectdefault(sel *Select) (selected bool) { |
| selected = false; |
| selectdefault(sel, runtime·getcallerpc(&sel), (byte*)&selected - (byte*)&sel); |
| } |
| |
| static void |
| selectdefault(Select *sel, void *callerpc, int32 so) |
| { |
| int32 i; |
| Scase *cas; |
| |
| i = sel->ncase; |
| if(i >= sel->tcase) |
| runtime·throw("selectdefault: too many cases"); |
| sel->ncase = i+1; |
| cas = &sel->scase[i]; |
| cas->pc = callerpc; |
| cas->chan = nil; |
| |
| cas->so = so; |
| cas->kind = CaseDefault; |
| |
| if(debug) |
| runtime·printf("selectdefault s=%p pc=%p so=%d\n", |
| sel, cas->pc, cas->so); |
| } |
| |
| static void |
| sellock(Select *sel) |
| { |
| uint32 i; |
| Hchan *c, *c0; |
| |
| c = nil; |
| for(i=0; i<sel->ncase; i++) { |
| c0 = sel->lockorder[i]; |
| if(c0 && c0 != c) { |
| c = sel->lockorder[i]; |
| runtime·lock(&c->lock); |
| } |
| } |
| } |
| |
| static void |
| selunlock(Select *sel) |
| { |
| int32 i, n, r; |
| Hchan *c; |
| |
| // We must be very careful here to not touch sel after we have unlocked |
| // the last lock, because sel can be freed right after the last unlock. |
| // Consider the following situation. |
| // First M calls runtime·park() in runtime·selectgo() passing the sel. |
| // Once runtime·park() has unlocked the last lock, another M makes |
| // the G that calls select runnable again and schedules it for execution. |
| // When the G runs on another M, it locks all the locks and frees sel. |
| // Now if the first M touches sel, it will access freed memory. |
| n = (int32)sel->ncase; |
| r = 0; |
| // skip the default case |
| if(n>0 && sel->lockorder[0] == nil) |
| r = 1; |
| for(i = n-1; i >= r; i--) { |
| c = sel->lockorder[i]; |
| if(i>0 && sel->lockorder[i-1] == c) |
| continue; // will unlock it on the next iteration |
| runtime·unlock(&c->lock); |
| } |
| } |
| |
| static bool |
| selparkcommit(G *gp, void *sel) |
| { |
| USED(gp); |
| selunlock(sel); |
| return true; |
| } |
| |
| func block() { |
| runtime·park(nil, nil, runtime·gostringnocopy((byte*)"select (no cases)")); // forever |
| } |
| |
| static void* selectgo(Select**); |
| |
| // selectgo(sel *byte); |
| // |
| // overwrites return pc on stack to signal which case of the select |
| // to run, so cannot appear at the top of a split stack. |
| #pragma textflag NOSPLIT |
| func selectgo(sel *Select) { |
| runtime·setcallerpc(&sel, selectgo(&sel)); |
| } |
| |
| static void* |
| selectgo(Select **selp) |
| { |
| Select *sel; |
| uint32 o, i, j, k, done; |
| int64 t0; |
| Scase *cas, *dfl; |
| Hchan *c; |
| SudoG *sg; |
| G *gp; |
| byte *as; |
| void *pc; |
| |
| sel = *selp; |
| |
| if(debug) |
| runtime·printf("select: sel=%p\n", sel); |
| |
| t0 = 0; |
| if(runtime·blockprofilerate > 0) { |
| t0 = runtime·cputicks(); |
| for(i=0; i<sel->ncase; i++) |
| sel->scase[i].sg.releasetime = -1; |
| } |
| |
| // The compiler rewrites selects that statically have |
| // only 0 or 1 cases plus default into simpler constructs. |
| // The only way we can end up with such small sel->ncase |
| // values here is for a larger select in which most channels |
| // have been nilled out. The general code handles those |
| // cases correctly, and they are rare enough not to bother |
| // optimizing (and needing to test). |
| |
| // generate permuted order |
| for(i=0; i<sel->ncase; i++) |
| sel->pollorder[i] = i; |
| for(i=1; i<sel->ncase; i++) { |
| o = sel->pollorder[i]; |
| j = runtime·fastrand1()%(i+1); |
| sel->pollorder[i] = sel->pollorder[j]; |
| sel->pollorder[j] = o; |
| } |
| |
| // sort the cases by Hchan address to get the locking order. |
| // simple heap sort, to guarantee n log n time and constant stack footprint. |
| for(i=0; i<sel->ncase; i++) { |
| j = i; |
| c = sel->scase[j].chan; |
| while(j > 0 && sel->lockorder[k=(j-1)/2] < c) { |
| sel->lockorder[j] = sel->lockorder[k]; |
| j = k; |
| } |
| sel->lockorder[j] = c; |
| } |
| for(i=sel->ncase; i-->0; ) { |
| c = sel->lockorder[i]; |
| sel->lockorder[i] = sel->lockorder[0]; |
| j = 0; |
| for(;;) { |
| k = j*2+1; |
| if(k >= i) |
| break; |
| if(k+1 < i && sel->lockorder[k] < sel->lockorder[k+1]) |
| k++; |
| if(c < sel->lockorder[k]) { |
| sel->lockorder[j] = sel->lockorder[k]; |
| j = k; |
| continue; |
| } |
| break; |
| } |
| sel->lockorder[j] = c; |
| } |
| /* |
| for(i=0; i+1<sel->ncase; i++) |
| if(sel->lockorder[i] > sel->lockorder[i+1]) { |
| runtime·printf("i=%d %p %p\n", i, sel->lockorder[i], sel->lockorder[i+1]); |
| runtime·throw("select: broken sort"); |
| } |
| */ |
| sellock(sel); |
| |
| loop: |
| // pass 1 - look for something already waiting |
| dfl = nil; |
| for(i=0; i<sel->ncase; i++) { |
| o = sel->pollorder[i]; |
| cas = &sel->scase[o]; |
| c = cas->chan; |
| |
| switch(cas->kind) { |
| case CaseRecv: |
| if(c->dataqsiz > 0) { |
| if(c->qcount > 0) |
| goto asyncrecv; |
| } else { |
| sg = dequeue(&c->sendq); |
| if(sg != nil) |
| goto syncrecv; |
| } |
| if(c->closed) |
| goto rclose; |
| break; |
| |
| case CaseSend: |
| if(raceenabled) |
| runtime·racereadpc(c, cas->pc, chansend); |
| if(c->closed) |
| goto sclose; |
| if(c->dataqsiz > 0) { |
| if(c->qcount < c->dataqsiz) |
| goto asyncsend; |
| } else { |
| sg = dequeue(&c->recvq); |
| if(sg != nil) |
| goto syncsend; |
| } |
| break; |
| |
| case CaseDefault: |
| dfl = cas; |
| break; |
| } |
| } |
| |
| if(dfl != nil) { |
| selunlock(sel); |
| cas = dfl; |
| goto retc; |
| } |
| |
| |
| // pass 2 - enqueue on all chans |
| done = 0; |
| for(i=0; i<sel->ncase; i++) { |
| o = sel->pollorder[i]; |
| cas = &sel->scase[o]; |
| c = cas->chan; |
| sg = &cas->sg; |
| sg->g = g; |
| sg->selectdone = &done; |
| |
| switch(cas->kind) { |
| case CaseRecv: |
| enqueue(&c->recvq, sg); |
| break; |
| |
| case CaseSend: |
| enqueue(&c->sendq, sg); |
| break; |
| } |
| } |
| |
| g->param = nil; |
| runtime·park(selparkcommit, sel, runtime·gostringnocopy((byte*)"select")); |
| |
| sellock(sel); |
| sg = g->param; |
| |
| // pass 3 - dequeue from unsuccessful chans |
| // otherwise they stack up on quiet channels |
| for(i=0; i<sel->ncase; i++) { |
| cas = &sel->scase[i]; |
| if(cas != (Scase*)sg) { |
| c = cas->chan; |
| if(cas->kind == CaseSend) |
| dequeueg(&c->sendq); |
| else |
| dequeueg(&c->recvq); |
| } |
| } |
| |
| if(sg == nil) |
| goto loop; |
| |
| cas = (Scase*)sg; |
| c = cas->chan; |
| |
| if(c->dataqsiz > 0) |
| runtime·throw("selectgo: shouldn't happen"); |
| |
| if(debug) |
| runtime·printf("wait-return: sel=%p c=%p cas=%p kind=%d\n", |
| sel, c, cas, cas->kind); |
| |
| if(cas->kind == CaseRecv) { |
| if(cas->receivedp != nil) |
| *cas->receivedp = true; |
| } |
| |
| if(raceenabled) { |
| if(cas->kind == CaseRecv && cas->sg.elem != nil) |
| runtime·racewriteobjectpc(cas->sg.elem, c->elemtype, cas->pc, chanrecv); |
| else if(cas->kind == CaseSend) |
| runtime·racereadobjectpc(cas->sg.elem, c->elemtype, cas->pc, chansend); |
| } |
| |
| selunlock(sel); |
| goto retc; |
| |
| asyncrecv: |
| // can receive from buffer |
| if(raceenabled) { |
| if(cas->sg.elem != nil) |
| runtime·racewriteobjectpc(cas->sg.elem, c->elemtype, cas->pc, chanrecv); |
| runtime·raceacquire(chanbuf(c, c->recvx)); |
| runtime·racerelease(chanbuf(c, c->recvx)); |
| } |
| if(cas->receivedp != nil) |
| *cas->receivedp = true; |
| if(cas->sg.elem != nil) |
| c->elemtype->alg->copy(c->elemsize, cas->sg.elem, chanbuf(c, c->recvx)); |
| c->elemtype->alg->copy(c->elemsize, chanbuf(c, c->recvx), nil); |
| if(++c->recvx == c->dataqsiz) |
| c->recvx = 0; |
| c->qcount--; |
| sg = dequeue(&c->sendq); |
| if(sg != nil) { |
| gp = sg->g; |
| selunlock(sel); |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| } else { |
| selunlock(sel); |
| } |
| goto retc; |
| |
| asyncsend: |
| // can send to buffer |
| if(raceenabled) { |
| runtime·raceacquire(chanbuf(c, c->sendx)); |
| runtime·racerelease(chanbuf(c, c->sendx)); |
| runtime·racereadobjectpc(cas->sg.elem, c->elemtype, cas->pc, chansend); |
| } |
| c->elemtype->alg->copy(c->elemsize, chanbuf(c, c->sendx), cas->sg.elem); |
| if(++c->sendx == c->dataqsiz) |
| c->sendx = 0; |
| c->qcount++; |
| sg = dequeue(&c->recvq); |
| if(sg != nil) { |
| gp = sg->g; |
| selunlock(sel); |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| } else { |
| selunlock(sel); |
| } |
| goto retc; |
| |
| syncrecv: |
| // can receive from sleeping sender (sg) |
| if(raceenabled) { |
| if(cas->sg.elem != nil) |
| runtime·racewriteobjectpc(cas->sg.elem, c->elemtype, cas->pc, chanrecv); |
| racesync(c, sg); |
| } |
| selunlock(sel); |
| if(debug) |
| runtime·printf("syncrecv: sel=%p c=%p o=%d\n", sel, c, o); |
| if(cas->receivedp != nil) |
| *cas->receivedp = true; |
| if(cas->sg.elem != nil) |
| c->elemtype->alg->copy(c->elemsize, cas->sg.elem, sg->elem); |
| gp = sg->g; |
| gp->param = sg; |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| goto retc; |
| |
| rclose: |
| // read at end of closed channel |
| selunlock(sel); |
| if(cas->receivedp != nil) |
| *cas->receivedp = false; |
| if(cas->sg.elem != nil) |
| c->elemtype->alg->copy(c->elemsize, cas->sg.elem, nil); |
| if(raceenabled) |
| runtime·raceacquire(c); |
| goto retc; |
| |
| syncsend: |
| // can send to sleeping receiver (sg) |
| if(raceenabled) { |
| runtime·racereadobjectpc(cas->sg.elem, c->elemtype, cas->pc, chansend); |
| racesync(c, sg); |
| } |
| selunlock(sel); |
| if(debug) |
| runtime·printf("syncsend: sel=%p c=%p o=%d\n", sel, c, o); |
| if(sg->elem != nil) |
| c->elemtype->alg->copy(c->elemsize, sg->elem, cas->sg.elem); |
| gp = sg->g; |
| gp->param = sg; |
| if(sg->releasetime) |
| sg->releasetime = runtime·cputicks(); |
| runtime·ready(gp); |
| |
| retc: |
| // return pc corresponding to chosen case. |
| // Set boolean passed during select creation |
| // (at offset selp + cas->so) to true. |
| // If cas->so == 0, this is a reflect-driven select and we |
| // don't need to update the boolean. |
| pc = cas->pc; |
| if(cas->so > 0) { |
| as = (byte*)selp + cas->so; |
| *as = true; |
| } |
| if(cas->sg.releasetime > 0) |
| runtime·blockevent(cas->sg.releasetime - t0, 2); |
| return pc; |
| |
| sclose: |
| // send on closed channel |
| selunlock(sel); |
| runtime·panicstring("send on closed channel"); |
| return nil; // not reached |
| } |
| |
| // This struct must match ../reflect/value.go:/runtimeSelect. |
| typedef struct runtimeSelect runtimeSelect; |
| struct runtimeSelect |
| { |
| uintptr dir; |
| ChanType *typ; |
| Hchan *ch; |
| byte *val; |
| }; |
| |
| // This enum must match ../reflect/value.go:/SelectDir. |
| enum SelectDir { |
| SelectSend = 1, |
| SelectRecv, |
| SelectDefault, |
| }; |
| |
| func reflect·rselect(cases Slice) (chosen int, recvOK bool) { |
| int32 i; |
| Select *sel; |
| runtimeSelect* rcase, *rc; |
| |
| chosen = -1; |
| recvOK = false; |
| |
| rcase = (runtimeSelect*)cases.array; |
| |
| // FlagNoScan is safe here, because all objects are also referenced from cases. |
| sel = runtime·mallocgc(selectsize(cases.len), 0, FlagNoScan); |
| runtime·newselect(sel, selectsize(cases.len), cases.len); |
| for(i=0; i<cases.len; i++) { |
| rc = &rcase[i]; |
| switch(rc->dir) { |
| case SelectDefault: |
| selectdefault(sel, (void*)i, 0); |
| break; |
| case SelectSend: |
| if(rc->ch == nil) |
| break; |
| selectsend(sel, rc->ch, (void*)i, rc->val, 0); |
| break; |
| case SelectRecv: |
| if(rc->ch == nil) |
| break; |
| selectrecv(sel, rc->ch, (void*)i, rc->val, &recvOK, 0); |
| break; |
| } |
| } |
| |
| chosen = (intgo)(uintptr)selectgo(&sel); |
| } |
| |
| static SudoG* |
| dequeue(WaitQ *q) |
| { |
| SudoG *sgp; |
| |
| loop: |
| sgp = q->first; |
| if(sgp == nil) |
| return nil; |
| q->first = sgp->next; |
| if(q->last == sgp) |
| q->last = nil; |
| |
| // if sgp participates in a select and is already signaled, ignore it |
| if(sgp->selectdone != nil) { |
| // claim the right to signal |
| if(*sgp->selectdone != 0 || !runtime·cas(sgp->selectdone, 0, 1)) |
| goto loop; |
| } |
| |
| return sgp; |
| } |
| |
| static void |
| dequeueg(WaitQ *q) |
| { |
| SudoG **l, *sgp, *prevsgp; |
| |
| prevsgp = nil; |
| for(l=&q->first; (sgp=*l) != nil; l=&sgp->next, prevsgp=sgp) { |
| if(sgp->g == g) { |
| *l = sgp->next; |
| if(q->last == sgp) |
| q->last = prevsgp; |
| break; |
| } |
| } |
| } |
| |
| static void |
| enqueue(WaitQ *q, SudoG *sgp) |
| { |
| sgp->next = nil; |
| if(q->first == nil) { |
| q->first = sgp; |
| q->last = sgp; |
| return; |
| } |
| q->last->next = sgp; |
| q->last = sgp; |
| } |
| |
| static void |
| racesync(Hchan *c, SudoG *sg) |
| { |
| runtime·racerelease(chanbuf(c, 0)); |
| runtime·raceacquireg(sg->g, chanbuf(c, 0)); |
| runtime·racereleaseg(sg->g, chanbuf(c, 0)); |
| runtime·raceacquire(chanbuf(c, 0)); |
| } |