From 1d8fa7fa5db01cde6e764af29d4c2fb80bad9d8c Mon Sep 17 00:00:00 2001 From: Keith Randall Date: Tue, 2 Sep 2014 14:13:29 -0700 Subject: [PATCH] runtime: convert select implementation to Go. LGTM=rsc R=golang-codereviews, bradfitz, iant, khr, rsc CC=golang-codereviews https://golang.org/cl/139020043 --- src/cmd/api/goapi.go | 1 + src/cmd/gc/select.c | 3 +- src/pkg/reflect/value.go | 4 +- src/pkg/runtime/chan.go | 3 +- src/pkg/runtime/chan.goc | 664 ---------------------------- src/pkg/runtime/chan.h | 5 +- src/pkg/runtime/pprof/pprof_test.go | 4 +- src/pkg/runtime/race.go | 13 + src/pkg/runtime/race0.go | 2 + src/pkg/runtime/select.go | 645 +++++++++++++++++++++++++++ src/pkg/runtime/stack.c | 3 + src/pkg/runtime/thunk.s | 3 + 12 files changed, 677 insertions(+), 673 deletions(-) delete mode 100644 src/pkg/runtime/chan.goc create mode 100644 src/pkg/runtime/select.go diff --git a/src/cmd/api/goapi.go b/src/cmd/api/goapi.go index 715e9f6ad8..8dec9e2cc7 100644 --- a/src/cmd/api/goapi.go +++ b/src/cmd/api/goapi.go @@ -403,6 +403,7 @@ func (w *Walker) parseFile(dir, file string) (*ast.File, error) { " sudog struct{};" + " waitq struct{};" + " wincallbackcontext struct{};" + + " _select struct{}; " + "); " + "const ( cb_max = 2000 )" f, err = parser.ParseFile(fset, filename, src, 0) diff --git a/src/cmd/gc/select.c b/src/cmd/gc/select.c index 8cf9926c17..ed23e4318b 100644 --- a/src/cmd/gc/select.c +++ b/src/cmd/gc/select.c @@ -347,12 +347,13 @@ selecttype(int32 size) sudog->type->local = 1; scase = nod(OTSTRUCT, N, N); - scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("sg")), sudog)); + scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("elem")), typenod(ptrto(types[TUINT8])))); scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("chan")), typenod(ptrto(types[TUINT8])))); scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("pc")), typenod(types[TUINTPTR]))); scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("kind")), typenod(types[TUINT16]))); scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("so")), typenod(types[TUINT16]))); scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("receivedp")), typenod(ptrto(types[TUINT8])))); + scase->list = list(scase->list, nod(ODCLFIELD, newname(lookup("releasetime")), typenod(types[TUINT64]))); typecheck(&scase, Etype); scase->type->noalg = 1; scase->type->local = 1; diff --git a/src/pkg/reflect/value.go b/src/pkg/reflect/value.go index 4394ed0739..76086c561b 100644 --- a/src/pkg/reflect/value.go +++ b/src/pkg/reflect/value.go @@ -2073,7 +2073,7 @@ func Copy(dst, src Value) int { } // A runtimeSelect is a single case passed to rselect. -// This must match ../runtime/chan.c:/runtimeSelect +// This must match ../runtime/select.go:/runtimeSelect type runtimeSelect struct { dir uintptr // 0, SendDir, or RecvDir typ *rtype // channel type @@ -2091,7 +2091,7 @@ func rselect([]runtimeSelect) (chosen int, recvOK bool) // A SelectDir describes the communication direction of a select case. type SelectDir int -// NOTE: These values must match ../runtime/chan.c:/SelectDir. +// NOTE: These values must match ../runtime/select.go:/selectDir. const ( _ SelectDir = iota diff --git a/src/pkg/runtime/chan.go b/src/pkg/runtime/chan.go index 5e972983c6..77df169399 100644 --- a/src/pkg/runtime/chan.go +++ b/src/pkg/runtime/chan.go @@ -4,8 +4,7 @@ package runtime -// This file contains the implementation of Go channels -// and select statements. +// This file contains the implementation of Go channels. import "unsafe" diff --git a/src/pkg/runtime/chan.goc b/src/pkg/runtime/chan.goc deleted file mode 100644 index f14b58d52a..0000000000 --- a/src/pkg/runtime/chan.goc +++ /dev/null @@ -1,664 +0,0 @@ -// 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*); - -// TODO(khr): temporary placeholders until the rest of this code is moved to Go. -extern byte runtime·chansend; -extern byte runtime·chanrecv; - -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; incase; 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; - extern uint64 runtime·blockprofilerate; - - sel = *selp; - - if(debug) - runtime·printf("select: sel=%p\n", sel); - - t0 = 0; - if(runtime·blockprofilerate > 0) { - t0 = runtime·cputicks(); - for(i=0; incase; 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; incase; i++) - sel->pollorder[i] = i; - for(i=1; incase; 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; incase; 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+1ncase; 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; incase; 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, &runtime·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; incase; 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; incase; 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, &runtime·chanrecv); - else if(cas->kind == CaseSend) - runtime·racereadobjectpc(cas->sg.elem, c->elemtype, cas->pc, &runtime·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, &runtime·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) - runtime·memmove(cas->sg.elem, chanbuf(c, c->recvx), c->elemsize); - runtime·memclr(chanbuf(c, c->recvx), c->elemsize); - 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, &runtime·chansend); - } - runtime·memmove(chanbuf(c, c->sendx), cas->sg.elem, c->elemsize); - 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, &runtime·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) - runtime·memmove(cas->sg.elem, sg->elem, c->elemsize); - 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) - runtime·memclr(cas->sg.elem, c->elemsize); - 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, &runtime·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) - runtime·memmove(sg->elem, cas->sg.elem, c->elemsize); - 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; idir) { - 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)); -} diff --git a/src/pkg/runtime/chan.h b/src/pkg/runtime/chan.h index a439fa7c9a..c34ff1533d 100644 --- a/src/pkg/runtime/chan.h +++ b/src/pkg/runtime/chan.h @@ -47,12 +47,13 @@ enum // Changes here must also be made in src/cmd/gc/select.c's selecttype. struct Scase { - SudoG sg; // must be first member (cast to Scase) + void* elem; // data element Hchan* chan; // chan - byte* pc; // return pc + uintptr pc; // return pc uint16 kind; uint16 so; // vararg of selected bool bool* receivedp; // pointer to received bool (recv2) + int64 releasetime; }; // Known to compiler. diff --git a/src/pkg/runtime/pprof/pprof_test.go b/src/pkg/runtime/pprof/pprof_test.go index dd8f2d0529..df271273ce 100644 --- a/src/pkg/runtime/pprof/pprof_test.go +++ b/src/pkg/runtime/pprof/pprof_test.go @@ -328,13 +328,13 @@ func TestBlockProfile(t *testing.T) { `}, {"select recv async", blockSelectRecvAsync, ` [0-9]+ [0-9]+ @ 0x[0-9,a-f]+ 0x[0-9,a-f]+ 0x[0-9,a-f]+ 0x[0-9,a-f]+ 0x[0-9,a-f]+ -# 0x[0-9,a-f]+ runtime\.selectgo\+0x[0-9,a-f]+ .*/src/pkg/runtime/chan.goc:[0-9]+ +# 0x[0-9,a-f]+ runtime\.selectgo\+0x[0-9,a-f]+ .*/src/pkg/runtime/select.go:[0-9]+ # 0x[0-9,a-f]+ runtime/pprof_test\.blockSelectRecvAsync\+0x[0-9,a-f]+ .*/src/pkg/runtime/pprof/pprof_test.go:[0-9]+ # 0x[0-9,a-f]+ runtime/pprof_test\.TestBlockProfile\+0x[0-9,a-f]+ .*/src/pkg/runtime/pprof/pprof_test.go:[0-9]+ `}, {"select send sync", blockSelectSendSync, ` [0-9]+ [0-9]+ @ 0x[0-9,a-f]+ 0x[0-9,a-f]+ 0x[0-9,a-f]+ 0x[0-9,a-f]+ 0x[0-9,a-f]+ -# 0x[0-9,a-f]+ runtime\.selectgo\+0x[0-9,a-f]+ .*/src/pkg/runtime/chan.goc:[0-9]+ +# 0x[0-9,a-f]+ runtime\.selectgo\+0x[0-9,a-f]+ .*/src/pkg/runtime/select.go:[0-9]+ # 0x[0-9,a-f]+ runtime/pprof_test\.blockSelectSendSync\+0x[0-9,a-f]+ .*/src/pkg/runtime/pprof/pprof_test.go:[0-9]+ # 0x[0-9,a-f]+ runtime/pprof_test\.TestBlockProfile\+0x[0-9,a-f]+ .*/src/pkg/runtime/pprof/pprof_test.go:[0-9]+ `}, diff --git a/src/pkg/runtime/race.go b/src/pkg/runtime/race.go index 3707549a3f..df8493e35c 100644 --- a/src/pkg/runtime/race.go +++ b/src/pkg/runtime/race.go @@ -45,3 +45,16 @@ func raceReadObjectPC(t *_type, addr unsafe.Pointer, callerpc, pc uintptr) { racereadpc(addr, callerpc, pc) } } + +func raceWriteObjectPC(t *_type, addr unsafe.Pointer, callerpc, pc uintptr) { + kind := t.kind & kindMask + if kind == kindArray || kind == kindStruct { + // for composite objects we have to write every address + // because a write might happen to any subobject. + racewriterangepc(addr, int(t.size), callerpc, pc) + } else { + // for non-composite objects we can write just the start + // address, as any write must write the first byte. + racewritepc(addr, callerpc, pc) + } +} diff --git a/src/pkg/runtime/race0.go b/src/pkg/runtime/race0.go index f228c6d7b4..2e67ae6a50 100644 --- a/src/pkg/runtime/race0.go +++ b/src/pkg/runtime/race0.go @@ -16,3 +16,5 @@ const raceenabled = false func raceReadObjectPC(t *_type, addr unsafe.Pointer, callerpc, pc uintptr) { } +func raceWriteObjectPC(t *_type, addr unsafe.Pointer, callerpc, pc uintptr) { +} diff --git a/src/pkg/runtime/select.go b/src/pkg/runtime/select.go new file mode 100644 index 0000000000..31976cd6f8 --- /dev/null +++ b/src/pkg/runtime/select.go @@ -0,0 +1,645 @@ +// 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 + +// This file contains the implementation of Go select statements. + +import "unsafe" + +const ( + debugSelect = false +) + +var ( + chansendpc uintptr + chanrecvpc uintptr +) + +func init() { + f := chansend + chansendpc = **(**uintptr)(unsafe.Pointer(&f)) + g := chanrecv + chanrecvpc = **(**uintptr)(unsafe.Pointer(&g)) +} + +func selectsize(size uintptr) uintptr { + selsize := unsafe.Sizeof(_select{}) + + (size-1)*unsafe.Sizeof(_select{}.scase[0]) + + size*unsafe.Sizeof(*_select{}.lockorder) + + size*unsafe.Sizeof(*_select{}.pollorder) + return round(selsize, _Int64Align) +} + +func newselect(sel *_select, selsize int64, size int32) { + if selsize != int64(selectsize(uintptr(size))) { + print("runtime: bad select size ", selsize, ", want ", selectsize(uintptr(size)), "\n") + gothrow("bad select size") + } + sel.tcase = uint16(size) + sel.ncase = 0 + sel.lockorder = (**hchan)(add(unsafe.Pointer(&sel.scase), uintptr(size)*unsafe.Sizeof(_select{}.scase[0]))) + sel.pollorder = (*uint16)(add(unsafe.Pointer(sel.lockorder), uintptr(size)*unsafe.Sizeof(*_select{}.lockorder))) + + if debugSelect { + print("newselect s=", sel, " size=", size, "\n") + } +} + +//go:nosplit +func selectsend(sel *_select, c *hchan, elem unsafe.Pointer) (selected bool) { + // nil cases do not compete + if c != nil { + selectsendImpl(sel, c, getcallerpc(unsafe.Pointer(&sel)), elem, uintptr(unsafe.Pointer(&selected))-uintptr(unsafe.Pointer(&sel))) + } + return +} + +// cut in half to give stack a chance to split +func selectsendImpl(sel *_select, c *hchan, pc uintptr, elem unsafe.Pointer, so uintptr) { + i := sel.ncase + if i >= sel.tcase { + gothrow("selectsend: too many cases") + } + sel.ncase = i + 1 + cas := (*scase)(add(unsafe.Pointer(&sel.scase), uintptr(i)*unsafe.Sizeof(sel.scase[0]))) + + cas.pc = pc + cas._chan = c + cas.so = uint16(so) + cas.kind = _CaseSend + cas.elem = elem + + if debugSelect { + print("selectsend s=", sel, " pc=", hex(cas.pc), " chan=", cas._chan, " so=", cas.so, "\n") + } +} + +//go:nosplit +func selectrecv(sel *_select, c *hchan, elem unsafe.Pointer) (selected bool) { + // nil cases do not compete + if c != nil { + selectrecvImpl(sel, c, getcallerpc(unsafe.Pointer(&sel)), elem, nil, uintptr(unsafe.Pointer(&selected))-uintptr(unsafe.Pointer(&sel))) + } + return +} + +//go:nosplit +func selectrecv2(sel *_select, c *hchan, elem unsafe.Pointer, received *bool) (selected bool) { + // nil cases do not compete + if c != nil { + selectrecvImpl(sel, c, getcallerpc(unsafe.Pointer(&sel)), elem, received, uintptr(unsafe.Pointer(&selected))-uintptr(unsafe.Pointer(&sel))) + } + return +} + +func selectrecvImpl(sel *_select, c *hchan, pc uintptr, elem unsafe.Pointer, received *bool, so uintptr) { + i := sel.ncase + if i >= sel.tcase { + gothrow("selectrecv: too many cases") + } + sel.ncase = i + 1 + cas := (*scase)(add(unsafe.Pointer(&sel.scase), uintptr(i)*unsafe.Sizeof(sel.scase[0]))) + cas.pc = pc + cas._chan = c + cas.so = uint16(so) + cas.kind = _CaseRecv + cas.elem = elem + cas.receivedp = received + + if debugSelect { + print("selectrecv s=", sel, " pc=", hex(cas.pc), " chan=", cas._chan, " so=", cas.so, "\n") + } +} + +//go:nosplit +func selectdefault(sel *_select) (selected bool) { + selectdefaultImpl(sel, getcallerpc(unsafe.Pointer(&sel)), uintptr(unsafe.Pointer(&selected))-uintptr(unsafe.Pointer(&sel))) + return +} + +func selectdefaultImpl(sel *_select, callerpc uintptr, so uintptr) { + i := sel.ncase + if i >= sel.tcase { + gothrow("selectdefault: too many cases") + } + sel.ncase = i + 1 + cas := (*scase)(add(unsafe.Pointer(&sel.scase), uintptr(i)*unsafe.Sizeof(sel.scase[0]))) + cas.pc = callerpc + cas._chan = nil + cas.so = uint16(so) + cas.kind = _CaseDefault + + if debugSelect { + print("selectdefault s=", sel, " pc=", hex(cas.pc), " so=", cas.so, "\n") + } +} + +func sellock(sel *_select) { + lockslice := sliceStruct{unsafe.Pointer(sel.lockorder), int(sel.ncase), int(sel.ncase)} + lockorder := *(*[]*hchan)(unsafe.Pointer(&lockslice)) + var c *hchan + for _, c0 := range lockorder { + if c0 != nil && c0 != c { + c = c0 + lock(&c.lock) + } + } +} + +func selunlock(sel *_select) { + // 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 := int(sel.ncase) + r := 0 + lockslice := sliceStruct{unsafe.Pointer(sel.lockorder), n, n} + lockorder := *(*[]*hchan)(unsafe.Pointer(&lockslice)) + // skip the default case + if n > 0 && lockorder[0] == nil { + r = 1 + } + for i := n - 1; i >= r; i-- { + c := lockorder[i] + if i > 0 && c == lockorder[i-1] { + continue // will unlock it on the next iteration + } + unlock(&c.lock) + } +} + +func selparkcommit(gp *g, sel *_select) bool { + selunlock(sel) + return true +} + +func block() { + gopark(nil, nil, "select (no cases)") // forever +} + +// overwrites return pc on stack to signal which case of the select +// to run, so cannot appear at the top of a split stack. +//go:nosplit +func selectgo(sel *_select) { + pc, offset := selectgoImpl(sel) + *(*bool)(add(unsafe.Pointer(&sel), uintptr(offset))) = true + setcallerpc(unsafe.Pointer(&sel), pc) +} + +// selectgoImpl returns scase.pc and scase.so for the select +// case which fired. +func selectgoImpl(sel *_select) (uintptr, uint16) { + if debugSelect { + print("select: sel=", sel, "\n") + } + + scaseslice := sliceStruct{unsafe.Pointer(&sel.scase), int(sel.ncase), int(sel.ncase)} + scases := *(*[]scase)(unsafe.Pointer(&scaseslice)) + + var t0 int64 + if blockprofilerate > 0 { + t0 = cputicks() + for i := 0; i < int(sel.ncase); i++ { + scases[i].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 + pollslice := sliceStruct{unsafe.Pointer(sel.pollorder), int(sel.ncase), int(sel.ncase)} + pollorder := *(*[]uint16)(unsafe.Pointer(&pollslice)) + for i := 0; i < int(sel.ncase); i++ { + pollorder[i] = uint16(i) + } + for i := 1; i < int(sel.ncase); i++ { + o := pollorder[i] + j := int(fastrand2()) % (i + 1) + pollorder[i] = pollorder[j] + 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. + lockslice := sliceStruct{unsafe.Pointer(sel.lockorder), int(sel.ncase), int(sel.ncase)} + lockorder := *(*[]*hchan)(unsafe.Pointer(&lockslice)) + for i := 0; i < int(sel.ncase); i++ { + j := i + c := scases[j]._chan + for j > 0 && lockorder[(j-1)/2].sortkey() < c.sortkey() { + k := (j - 1) / 2 + lockorder[j] = lockorder[k] + j = k + } + lockorder[j] = c + } + for i := int(sel.ncase) - 1; i >= 0; i-- { + c := lockorder[i] + lockorder[i] = lockorder[0] + j := 0 + for { + k := j*2 + 1 + if k >= i { + break + } + if k+1 < i && lockorder[k].sortkey() < lockorder[k+1].sortkey() { + k++ + } + if c.sortkey() < lockorder[k].sortkey() { + lockorder[j] = lockorder[k] + j = k + continue + } + break + } + lockorder[j] = c + } + /* + for i := 0; i+1 < int(sel.ncase); i++ { + if lockorder[i].sortkey() > lockorder[i+1].sortkey() { + print("i=", i, " x=", lockorder[i], " y=", lockorder[i+1], "\n") + gothrow("select: broken sort") + } + } + */ + + // lock all the channels involved in the select + sellock(sel) + + var ( + gp *g + done uint32 + sg *sudog + c *hchan + k *scase + sglist *sudog + sgnext *sudog + fn func(*g, *_select) bool + ) + +loop: + // pass 1 - look for something already waiting + var dfl *scase + var cas *scase + for i := 0; i < int(sel.ncase); i++ { + cas = &scases[pollorder[i]] + c = cas._chan + + switch cas.kind { + case _CaseRecv: + if c.dataqsiz > 0 { + if c.qcount > 0 { + goto asyncrecv + } + } else { + sg = c.sendq.dequeue() + if sg != nil { + goto syncrecv + } + } + if c.closed != 0 { + goto rclose + } + + case _CaseSend: + if raceenabled { + racereadpc(unsafe.Pointer(c), cas.pc, chansendpc) + } + if c.closed != 0 { + goto sclose + } + if c.dataqsiz > 0 { + if c.qcount < c.dataqsiz { + goto asyncsend + } + } else { + sg = c.recvq.dequeue() + if sg != nil { + goto syncsend + } + } + + case _CaseDefault: + dfl = cas + } + } + + if dfl != nil { + selunlock(sel) + cas = dfl + goto retc + } + + // pass 2 - enqueue on all chans + gp = getg() + done = 0 + for i := 0; i < int(sel.ncase); i++ { + cas = &scases[pollorder[i]] + c = cas._chan + sg := acquireSudog() + sg.g = gp + // Note: selectdone is adjusted for stack copies in stack.c:adjustsudogs + sg.selectdone = (*uint32)(noescape(unsafe.Pointer(&done))) + sg.elem = cas.elem + sg.releasetime = 0 + if t0 != 0 { + sg.releasetime = -1 + } + sg.waitlink = gp.waiting + gp.waiting = sg + + switch cas.kind { + case _CaseRecv: + c.recvq.enqueue(sg) + + case _CaseSend: + c.sendq.enqueue(sg) + } + } + + // wait for someone to wake us up + gp.param = nil + fn = selparkcommit + gopark(**(**unsafe.Pointer)(unsafe.Pointer(&fn)), unsafe.Pointer(sel), "select") + + // someone woke us up + sellock(sel) + sg = (*sudog)(gp.param) + + // pass 3 - dequeue from unsuccessful chans + // otherwise they stack up on quiet channels + // record the successful case, if any. + // We singly-linked up the SudoGs in case order, so when + // iterating through the linked list they are in reverse order. + cas = nil + sglist = gp.waiting + gp.waiting = nil + for i := int(sel.ncase) - 1; i >= 0; i-- { + k = &scases[pollorder[i]] + if sglist.releasetime > 0 { + k.releasetime = sglist.releasetime + } + if sg == sglist { + cas = k + } else { + c = k._chan + if k.kind == _CaseSend { + c.sendq.dequeueg(gp) + } else { + c.recvq.dequeueg(gp) + } + } + sgnext = sglist.waitlink + releaseSudog(sglist) + sglist = sgnext + } + + if cas == nil { + goto loop + } + + c = cas._chan + + if c.dataqsiz > 0 { + gothrow("selectgo: shouldn't happen") + } + + if debugSelect { + print("wait-return: sel=", sel, " c=", c, " cas=", cas, " kind=", cas.kind, "\n") + } + + if cas.kind == _CaseRecv { + if cas.receivedp != nil { + *cas.receivedp = true + } + } + + if raceenabled { + if cas.kind == _CaseRecv && cas.elem != nil { + raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc) + } else if cas.kind == _CaseSend { + raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) + } + } + + selunlock(sel) + goto retc + +asyncrecv: + // can receive from buffer + if raceenabled { + if cas.elem != nil { + raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc) + } + raceacquire(chanbuf(c, c.recvx)) + racerelease(chanbuf(c, c.recvx)) + } + if cas.receivedp != nil { + *cas.receivedp = true + } + if cas.elem != nil { + memmove(cas.elem, chanbuf(c, c.recvx), uintptr(c.elemsize)) + } + memclr(chanbuf(c, c.recvx), uintptr(c.elemsize)) + c.recvx++ + if c.recvx == c.dataqsiz { + c.recvx = 0 + } + c.qcount-- + sg = c.sendq.dequeue() + if sg != nil { + gp = sg.g + selunlock(sel) + if sg.releasetime != 0 { + sg.releasetime = cputicks() + } + goready(gp) + } else { + selunlock(sel) + } + goto retc + +asyncsend: + // can send to buffer + if raceenabled { + raceacquire(chanbuf(c, c.sendx)) + racerelease(chanbuf(c, c.sendx)) + raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) + } + memmove(chanbuf(c, c.sendx), cas.elem, uintptr(c.elemsize)) + c.sendx++ + if c.sendx == c.dataqsiz { + c.sendx = 0 + } + c.qcount++ + sg = c.recvq.dequeue() + if sg != nil { + gp = sg.g + selunlock(sel) + if sg.releasetime != 0 { + sg.releasetime = cputicks() + } + goready(gp) + } else { + selunlock(sel) + } + goto retc + +syncrecv: + // can receive from sleeping sender (sg) + if raceenabled { + if cas.elem != nil { + raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc) + } + racesync(c, sg) + } + selunlock(sel) + if debugSelect { + print("syncrecv: sel=", sel, " c=", c, "\n") + } + if cas.receivedp != nil { + *cas.receivedp = true + } + if cas.elem != nil { + memmove(cas.elem, sg.elem, uintptr(c.elemsize)) + } + gp = sg.g + gp.param = unsafe.Pointer(sg) + if sg.releasetime != 0 { + sg.releasetime = cputicks() + } + goready(gp) + goto retc + +rclose: + // read at end of closed channel + selunlock(sel) + if cas.receivedp != nil { + *cas.receivedp = false + } + if cas.elem != nil { + memclr(cas.elem, uintptr(c.elemsize)) + } + if raceenabled { + raceacquire(unsafe.Pointer(c)) + } + goto retc + +syncsend: + // can send to sleeping receiver (sg) + if raceenabled { + raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) + racesync(c, sg) + } + selunlock(sel) + if debugSelect { + print("syncsend: sel=", sel, " c=", c, "\n") + } + if sg.elem != nil { + memmove(sg.elem, cas.elem, uintptr(c.elemsize)) + } + gp = sg.g + gp.param = unsafe.Pointer(sg) + if sg.releasetime != 0 { + sg.releasetime = cputicks() + } + goready(gp) + +retc: + if cas.releasetime > 0 { + blockevent(cas.releasetime-t0, 2) + } + return cas.pc, cas.so + +sclose: + // send on closed channel + selunlock(sel) + panic("send on closed channel") +} + +func (c *hchan) sortkey() uintptr { + // TODO(khr): if we have a moving garbage collector, we'll need to + // change this function. + return uintptr(unsafe.Pointer(c)) +} + +// A runtimeSelect is a single case passed to rselect. +// This must match ../reflect/value.go:/runtimeSelect +type runtimeSelect struct { + dir selectDir + typ unsafe.Pointer // channel type (not used here) + ch *hchan // channel + val unsafe.Pointer // ptr to data (SendDir) or ptr to receive buffer (RecvDir) +} + +// These values must match ../reflect/value.go:/SelectDir. +type selectDir int + +const ( + _ selectDir = iota + selectSend // case Chan <- Send + selectRecv // case <-Chan: + selectDefault // default +) + +func reflect_rselect(cases []runtimeSelect) (chosen int, recvOK bool) { + // flagNoScan is safe here, because all objects are also referenced from cases. + size := selectsize(uintptr(len(cases))) + sel := (*_select)(gomallocgc(size, nil, flagNoScan)) + newselect(sel, int64(size), int32(len(cases))) + r := new(bool) + for i := range cases { + rc := &cases[i] + switch rc.dir { + case selectDefault: + selectdefaultImpl(sel, uintptr(i), 0) + case selectSend: + if rc.ch == nil { + break + } + selectsendImpl(sel, rc.ch, uintptr(i), rc.val, 0) + case selectRecv: + if rc.ch == nil { + break + } + selectrecvImpl(sel, rc.ch, uintptr(i), rc.val, r, 0) + } + } + + pc, _ := selectgoImpl(sel) + chosen = int(pc) + recvOK = *r + return +} + +func (q *waitq) dequeueg(gp *g) { + var prevsgp *sudog + l := &q.first + for { + sgp := *l + if sgp == nil { + return + } + if sgp.g == gp { + *l = sgp.next + if q.last == sgp { + q.last = prevsgp + } + return + } + l = &sgp.next + prevsgp = sgp + } +} diff --git a/src/pkg/runtime/stack.c b/src/pkg/runtime/stack.c index 1f51e667f2..62ec5993a8 100644 --- a/src/pkg/runtime/stack.c +++ b/src/pkg/runtime/stack.c @@ -764,6 +764,9 @@ adjustsudogs(G *gp, AdjustInfo *adjinfo) e = s->elem; if(adjinfo->oldstk <= e && e < adjinfo->oldbase) s->elem = e + adjinfo->delta; + e = (byte*)s->selectdone; + if(adjinfo->oldstk <= e && e < adjinfo->oldbase) + s->selectdone = (uint32*)(e + adjinfo->delta); } } diff --git a/src/pkg/runtime/thunk.s b/src/pkg/runtime/thunk.s index eaba5e1489..048b7a7236 100644 --- a/src/pkg/runtime/thunk.s +++ b/src/pkg/runtime/thunk.s @@ -115,3 +115,6 @@ TEXT reflect·unsafe_NewArray(SB),NOSPLIT,$0-0 TEXT reflect·makechan(SB),NOSPLIT,$0-0 JMP runtime·makechan(SB) + +TEXT reflect·rselect(SB), NOSPLIT, $0-0 + JMP runtime·reflect_rselect(SB) -- 2.48.1