Skip to content

G
go
Commit 8e496f1d authored by Keith Randall's avatar Keith Randall
Browse files
Options

runtime: simplify buffered channels. 

This change removes the retry mechanism we use for buffered channels.
Instead, any sender waking up a receiver or vice versa completes the
full protocol with its counterpart.  This means the counterpart does
not need to relock the channel when it wakes up.  (Currently
buffered channels need to relock on wakeup.)

For sends on a channel with waiting receivers, this change replaces
two copies (sender->queue, queue->receiver) with one (sender->receiver).
For receives on channels with a waiting sender, two copies are still required.

This change unifies to a large degree the algorithm for buffered
and unbuffered channels, simplifying the overall implementation.

Fixes #11506

benchmark                        old ns/op     new ns/op     delta
BenchmarkChanProdCons10          125           110           -12.00%
BenchmarkChanProdCons0           303           284           -6.27%
BenchmarkChanProdCons100         75.5          71.3          -5.56%
BenchmarkChanContended           6452          6125          -5.07%
BenchmarkChanNonblocking         11.5          11.0          -4.35%
BenchmarkChanCreation            149           143           -4.03%
BenchmarkChanSem                 63.6          61.6          -3.14%
BenchmarkChanUncontended         6390          6212          -2.79%
BenchmarkChanSync                282           276           -2.13%
BenchmarkChanProdConsWork10      516           506           -1.94%
BenchmarkChanProdConsWork0       696           685           -1.58%
BenchmarkChanProdConsWork100     470           469           -0.21%
BenchmarkChanPopular             660427        660012        -0.06%

Change-Id: I164113a56432fbc7cace0786e49c5a6e6a708ea4
Reviewed-on: https://go-review.googlesource.com/9345
Run-TryBot: Keith Randall <khr@golang.org>
Reviewed-by: default avatarAustin Clements <austin@google.com>
Reviewed-by: default avatarDmitry Vyukov <dvyukov@google.com>
parent 7bb2a7d6
Hide whitespace changes
Inline Side-by-side
Showing with 264 additions and 349 deletions
src/runtime/chan.go
View file @ 8e496f1d
...@@ -6,6 +6,11 @@ package runtime ...@@ -6,6 +6,11 @@ package runtime
// This file contains the implementation of Go channels. // This file contains the implementation of Go channels.
// Invariants:
// At least one of c.sendq and c.recvq is empty.
// For buffered channels, also:
// c.qcount > 0 implies that c.recvq is empty.
// c.qcount < c.dataqsiz implies that c.sendq is empty.
import "unsafe" import "unsafe"
const ( const (
...@@ -153,135 +158,117 @@ func chansend(t *chantype, c *hchan, ep unsafe.Pointer, block bool, callerpc uin ...@@ -153,135 +158,117 @@ func chansend(t *chantype, c *hchan, ep unsafe.Pointer, block bool, callerpc uin
} }
lock(&c.lock) lock(&c.lock)
if c.closed != 0 { if c.closed != 0 {
unlock(&c.lock) unlock(&c.lock)
panic("send on closed channel") panic("send on closed channel")
} }
if c.dataqsiz == 0 { // synchronous channel if sg := c.recvq.dequeue(); sg != nil {
sg := c.recvq.dequeue() // Found a waiting receiver. We pass the value we want to send
if sg != nil { // found a waiting receiver // directly to the receiver, bypassing the channel buffer (if any).
if raceenabled { send(c, sg, ep, func() { unlock(&c.lock) })
racesync(c, sg) return true
} }
unlock(&c.lock)
recvg := sg.g
if sg.elem != nil {
syncsend(c, sg, ep)
}
recvg.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(recvg, 3)
return true
}
if !block {
unlock(&c.lock)
return false
}
// no receiver available: block on this channel.
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
gp.waiting = mysg
mysg.g = gp
mysg.selectdone = nil
gp.param = nil
c.sendq.enqueue(mysg)
goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)
// someone woke us up. if c.qcount < c.dataqsiz {
if mysg != gp.waiting { // Space is available in the channel buffer. Enqueue the element to send.
throw("G waiting list is corrupted!") qp := chanbuf(c, c.sendx)
} if raceenabled {
gp.waiting = nil raceacquire(qp)
if gp.param == nil { racerelease(qp)
if c.closed == 0 {
throw("chansend: spurious wakeup")
}
panic("send on closed channel")
} }
gp.param = nil typedmemmove(c.elemtype, qp, ep)
if mysg.releasetime > 0 { c.sendx++
blockevent(int64(mysg.releasetime)-t0, 2) if c.sendx == c.dataqsiz {
c.sendx = 0
} }
releaseSudog(mysg) c.qcount++
unlock(&c.lock)
return true return true
} }
// asynchronous channel if !block {
// wait for some space to write our data unlock(&c.lock)
var t1 int64 return false
for futile := byte(0); c.qcount >= c.dataqsiz; futile = traceFutileWakeup {
if !block {
unlock(&c.lock)
return false
}
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
mysg.g = gp
mysg.elem = nil
mysg.selectdone = nil
c.sendq.enqueue(mysg)
goparkunlock(&c.lock, "chan send", traceEvGoBlockSend|futile, 3)
// someone woke us up - try again
if mysg.releasetime > 0 {
t1 = mysg.releasetime
}
releaseSudog(mysg)
lock(&c.lock)
if c.closed != 0 {
unlock(&c.lock)
panic("send on closed channel")
}
} }
// write our data into the channel buffer // Block on the channel. Some receiver will complete our operation for us.
if raceenabled { gp := getg()
raceacquire(chanbuf(c, c.sendx)) mysg := acquireSudog()
racerelease(chanbuf(c, c.sendx)) mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
mysg.g = gp
mysg.selectdone = nil
gp.waiting = mysg
gp.param = nil
c.sendq.enqueue(mysg)
goparkunlock(&c.lock, "chan send", traceEvGoBlockSend, 3)
// someone woke us up.
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
if gp.param == nil {
if c.closed == 0 {
throw("chansend: spurious wakeup")
}
panic("send on closed channel")
} }
typedmemmove(c.elemtype, chanbuf(c, c.sendx), ep) gp.param = nil
c.sendx++ if mysg.releasetime > 0 {
if c.sendx == c.dataqsiz { blockevent(int64(mysg.releasetime)-t0, 2)
c.sendx = 0
} }
c.qcount++ releaseSudog(mysg)
return true
}
// wake up a waiting receiver // send processes a send operation on an empty channel c.
sg := c.recvq.dequeue() // The value ep sent by the sender is copied to the receiver sg.
if sg != nil { // The receiver is then woken up to go on its merry way.
recvg := sg.g // Channel c must be empty and locked. send unlocks c with unlockf.
unlock(&c.lock) // sg must already be dequeued from c.
if sg.releasetime != 0 { // ep must be non-nil and point to the heap or the caller's stack.
sg.releasetime = cputicks() func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func()) {
if raceenabled {
if c.dataqsiz == 0 {
racesync(c, sg)
} else {
// Pretend we go through the buffer, even though
// we copy directly. Note that we need to increment
// the head/tail locations only when raceenabled.
qp := chanbuf(c, c.recvx)
raceacquire(qp)
racerelease(qp)
raceacquireg(sg.g, qp)
racereleaseg(sg.g, qp)
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
} }
goready(recvg, 3)
} else {
unlock(&c.lock)
} }
if t1 > 0 { unlockf()
blockevent(t1-t0, 2) if sg.elem != nil {
sendDirect(c.elemtype, sg.elem, ep)
sg.elem = nil
} }
return true gp := sg.g
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, 4)
} }
func syncsend(c *hchan, sg *sudog, elem unsafe.Pointer) { func sendDirect(t *_type, dst, src unsafe.Pointer) {
// Send on unbuffered channel is the only operation // Send on an unbuffered or empty-buffered channel is the only operation
// in the entire runtime where one goroutine // in the entire runtime where one goroutine
// writes to the stack of another goroutine. The GC assumes that // writes to the stack of another goroutine. The GC assumes that
// stack writes only happen when the goroutine is running and are // stack writes only happen when the goroutine is running and are
...@@ -290,9 +277,8 @@ func syncsend(c *hchan, sg *sudog, elem unsafe.Pointer) { ...@@ -290,9 +277,8 @@ func syncsend(c *hchan, sg *sudog, elem unsafe.Pointer) {
// typedmemmove will call heapBitsBulkBarrier, but the target bytes // typedmemmove will call heapBitsBulkBarrier, but the target bytes
// are not in the heap, so that will not help. We arrange to call // are not in the heap, so that will not help. We arrange to call
// memmove and typeBitsBulkBarrier instead. // memmove and typeBitsBulkBarrier instead.
memmove(sg.elem, elem, c.elemtype.size) memmove(dst, src, t.size)
typeBitsBulkBarrier(c.elemtype, uintptr(sg.elem), c.elemtype.size) typeBitsBulkBarrier(t, uintptr(dst), t.size)
sg.elem = nil
} }
func closechan(c *hchan) { func closechan(c *hchan) {
...@@ -320,27 +306,36 @@ func closechan(c *hchan) { ...@@ -320,27 +306,36 @@ func closechan(c *hchan) {
if sg == nil { if sg == nil {
break break
} }
gp := sg.g if sg.elem != nil {
sg.elem = nil memclr(sg.elem, uintptr(c.elemsize))
gp.param = nil sg.elem = nil
}
if sg.releasetime != 0 { if sg.releasetime != 0 {
sg.releasetime = cputicks() sg.releasetime = cputicks()
} }
gp := sg.g
gp.param = nil
if raceenabled {
raceacquireg(gp, unsafe.Pointer(c))
}
goready(gp, 3) goready(gp, 3)
} }
// release all writers // release all writers (they will panic)
for { for {
sg := c.sendq.dequeue() sg := c.sendq.dequeue()
if sg == nil { if sg == nil {
break break
} }
gp := sg.g
sg.elem = nil sg.elem = nil
gp.param = nil
if sg.releasetime != 0 { if sg.releasetime != 0 {
sg.releasetime = cputicks() sg.releasetime = cputicks()
} }
gp := sg.g
gp.param = nil
if raceenabled {
raceacquireg(gp, unsafe.Pointer(c))
}
goready(gp, 3) goready(gp, 3)
} }
unlock(&c.lock) unlock(&c.lock)
...@@ -363,8 +358,10 @@ func chanrecv2(t *chantype, c *hchan, elem unsafe.Pointer) (received bool) { ...@@ -363,8 +358,10 @@ func chanrecv2(t *chantype, c *hchan, elem unsafe.Pointer) (received bool) {
// If block == false and no elements are available, returns (false, false). // If block == false and no elements are available, returns (false, false).
// Otherwise, if c is closed, zeros *ep and returns (true, false). // Otherwise, if c is closed, zeros *ep and returns (true, false).
// Otherwise, fills in *ep with an element and returns (true, true). // Otherwise, fills in *ep with an element and returns (true, true).
// A non-nil ep must point to the heap or the caller's stack.
func chanrecv(t *chantype, c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) { func chanrecv(t *chantype, c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
// raceenabled: don't need to check ep, as it is always on the stack. // raceenabled: don't need to check ep, as it is always on the stack
// or is new memory allocated by reflect.
if debugChan { if debugChan {
print("chanrecv: chan=", c, "\n") print("chanrecv: chan=", c, "\n")
...@@ -402,167 +399,139 @@ func chanrecv(t *chantype, c *hchan, ep unsafe.Pointer, block bool) (selected, r ...@@ -402,167 +399,139 @@ func chanrecv(t *chantype, c *hchan, ep unsafe.Pointer, block bool) (selected, r
} }
lock(&c.lock) lock(&c.lock)
if c.dataqsiz == 0 { // synchronous channel
if c.closed != 0 {
return recvclosed(c, ep)
}
sg := c.sendq.dequeue()
if sg != nil {
if raceenabled {
racesync(c, sg)
}
unlock(&c.lock)
if ep != nil {
typedmemmove(c.elemtype, ep, sg.elem)
}
sg.elem = nil
gp := sg.g
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, 3)
selected = true
received = true
return
}
if !block {
unlock(&c.lock)
return
}
// no sender available: block on this channel. if c.closed != 0 && c.qcount == 0 {
gp := getg() if raceenabled {
mysg := acquireSudog() raceacquire(unsafe.Pointer(c))
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
} }
mysg.elem = ep unlock(&c.lock)
mysg.waitlink = nil if ep != nil {
gp.waiting = mysg memclr(ep, uintptr(c.elemsize))
mysg.g = gp
mysg.selectdone = nil
gp.param = nil
c.recvq.enqueue(mysg)
goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)
// someone woke us up
if mysg != gp.waiting {
throw("G waiting list is corrupted!")
}
gp.waiting = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
haveData := gp.param != nil
gp.param = nil
releaseSudog(mysg)
if haveData {
// a sender sent us some data. It already wrote to ep.
selected = true
received = true
return
}
lock(&c.lock)
if c.closed == 0 {
throw("chanrecv: spurious wakeup")
} }
return recvclosed(c, ep) return true, false
} }
// asynchronous channel if sg := c.sendq.dequeue(); sg != nil {
// wait for some data to appear // Found a waiting sender. If buffer is size 0, receive value
var t1 int64 // directly from sender. Otherwise, recieve from head of queue
for futile := byte(0); c.qcount <= 0; futile = traceFutileWakeup { // and add sender's value to the tail of the queue (both map to
if c.closed != 0 { // the same buffer slot because the queue is full).
selected, received = recvclosed(c, ep) recv(c, sg, ep, func() { unlock(&c.lock) })
if t1 > 0 { return true, true
blockevent(t1-t0, 2) }
}
return
}
if !block {
unlock(&c.lock)
return
}
// wait for someone to send an element if c.qcount > 0 {
gp := getg() // Receive directly from queue
mysg := acquireSudog() qp := chanbuf(c, c.recvx)
mysg.releasetime = 0 if raceenabled {
if t0 != 0 { raceacquire(qp)
mysg.releasetime = -1 racerelease(qp)
} }
mysg.elem = nil if ep != nil {
mysg.g = gp typedmemmove(c.elemtype, ep, qp)
mysg.selectdone = nil
c.recvq.enqueue(mysg)
goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv|futile, 3)
// someone woke us up - try again
if mysg.releasetime > 0 {
t1 = mysg.releasetime
} }
releaseSudog(mysg) memclr(qp, uintptr(c.elemsize))
lock(&c.lock) c.recvx++
} if c.recvx == c.dataqsiz {
c.recvx = 0
if raceenabled {
raceacquire(chanbuf(c, c.recvx))
racerelease(chanbuf(c, c.recvx))
}
if ep != nil {
typedmemmove(c.elemtype, ep, chanbuf(c, c.recvx))
}
memclr(chanbuf(c, c.recvx), uintptr(c.elemsize))
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
c.qcount--
// ping a sender now that there is space
sg := c.sendq.dequeue()
if sg != nil {
gp := sg.g
unlock(&c.lock)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
} }
goready(gp, 3) c.qcount--
} else {
unlock(&c.lock) unlock(&c.lock)
return true, true
} }
if t1 > 0 { if !block {
blockevent(t1-t0, 2) unlock(&c.lock)
} return false, false
selected = true }
received = true
return // no sender available: block on this channel.
gp := getg()
mysg := acquireSudog()
mysg.releasetime = 0
if t0 != 0 {
mysg.releasetime = -1
}
mysg.elem = ep
mysg.waitlink = nil
gp.waiting = mysg
mysg.g = gp
mysg.selectdone = nil
gp.param = nil
c.recvq.enqueue(mysg)
goparkunlock(&c.lock, "chan receive", traceEvGoBlockRecv, 3)
// someone woke us up
if mysg != gp.waiting {
throw("G waiting list is corrupted")
}
gp.waiting = nil
if mysg.releasetime > 0 {
blockevent(mysg.releasetime-t0, 2)
}
closed := gp.param == nil
gp.param = nil
releaseSudog(mysg)
return true, !closed
} }
// recvclosed is a helper function for chanrecv. Handles cleanup // recv processes a receive operation on a full channel c.
// when the receiver encounters a closed channel. // There are 2 parts:
// Caller must hold c.lock, recvclosed will release the lock. // 1) The value sent by the sender sg is put into the channel
func recvclosed(c *hchan, ep unsafe.Pointer) (selected, recevied bool) { // and the sender is woken up to go on its merry way.
if raceenabled { // 2) The value received by the receiver (the current G) is
raceacquire(unsafe.Pointer(c)) // written to ep.
// For synchronous channels, both values are the same.
// For asynchronous channels, the receiver gets its data from
// the channel buffer and the sender's data is put in the
// channel buffer.
// Channel c must be full and locked. recv unlocks c with unlockf.
// sg must already be dequeued from c.
// A non-nil ep must point to the heap or the caller's stack.
func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func()) {
if c.dataqsiz == 0 {
if raceenabled {
racesync(c, sg)
}
unlockf()
if ep != nil {
// copy data from sender
// ep points to our own stack or heap, so nothing
// special (ala sendDirect) needed here.
typedmemmove(c.elemtype, ep, sg.elem)
}
} else {
// Queue is full. Take the item at the
// head of the queue. Make the sender enqueue
// its item at the tail of the queue. Since the
// queue is full, those are both the same slot.
qp := chanbuf(c, c.recvx)
if raceenabled {
raceacquire(qp)
racerelease(qp)
raceacquireg(sg.g, qp)
racereleaseg(sg.g, qp)
}
// copy data from queue to receiver
if ep != nil {
typedmemmove(c.elemtype, ep, qp)
}
// copy data from sender to queue
typedmemmove(c.elemtype, qp, sg.elem)
c.recvx++
if c.recvx == c.dataqsiz {
c.recvx = 0
}
c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz
unlockf()
} }
unlock(&c.lock) sg.elem = nil
if ep != nil { gp := sg.g
memclr(ep, uintptr(c.elemsize)) gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
} }
return true, false goready(gp, 4)
} }
// compiler implements // compiler implements
......
src/runtime/select.go
View file @ 8e496f1d
...@@ -304,7 +304,7 @@ func selectgoImpl(sel *hselect) (uintptr, uint16) { ...@@ -304,7 +304,7 @@ func selectgoImpl(sel *hselect) (uintptr, uint16) {
k *scase k *scase
sglist *sudog sglist *sudog
sgnext *sudog sgnext *sudog
futile byte qp unsafe.Pointer
) )
loop: loop:
...@@ -317,15 +317,12 @@ loop: ...@@ -317,15 +317,12 @@ loop:
switch cas.kind { switch cas.kind {
case caseRecv: case caseRecv:
if c.dataqsiz > 0 { sg = c.sendq.dequeue()
if c.qcount > 0 { if sg != nil {
goto asyncrecv goto recv
} }
} else { if c.qcount > 0 {
sg = c.sendq.dequeue() goto bufrecv
if sg != nil {
goto syncrecv
}
} }
if c.closed != 0 { if c.closed != 0 {
goto rclose goto rclose
...@@ -338,15 +335,12 @@ loop: ...@@ -338,15 +335,12 @@ loop:
if c.closed != 0 { if c.closed != 0 {
goto sclose goto sclose
} }
if c.dataqsiz > 0 { sg = c.recvq.dequeue()
if c.qcount < c.dataqsiz { if sg != nil {
goto asyncsend goto send
} }
} else { if c.qcount < c.dataqsiz {
sg = c.recvq.dequeue() goto bufsend
if sg != nil {
goto syncsend
}
} }
case caseDefault: case caseDefault:
...@@ -363,6 +357,9 @@ loop: ...@@ -363,6 +357,9 @@ loop:
// pass 2 - enqueue on all chans // pass 2 - enqueue on all chans
gp = getg() gp = getg()
done = 0 done = 0
if gp.waiting != nil {
throw("gp.waiting != nil")
}
for i := 0; i < int(sel.ncase); i++ { for i := 0; i < int(sel.ncase); i++ {
cas = &scases[pollorder[i]] cas = &scases[pollorder[i]]
c = cas.c c = cas.c
...@@ -389,7 +386,7 @@ loop: ...@@ -389,7 +386,7 @@ loop:
// wait for someone to wake us up // wait for someone to wake us up
gp.param = nil gp.param = nil
gopark(selparkcommit, unsafe.Pointer(sel), "select", traceEvGoBlockSelect|futile, 2) gopark(selparkcommit, unsafe.Pointer(sel), "select", traceEvGoBlockSelect, 2)
// someone woke us up // someone woke us up
sellock(sel) sellock(sel)
...@@ -432,16 +429,13 @@ loop: ...@@ -432,16 +429,13 @@ loop:
} }
if cas == nil { if cas == nil {
futile = traceFutileWakeup // This can happen if we were woken up by a close().
// TODO: figure that out explicitly so we don't need this loop.
goto loop goto loop
} }
c = cas.c c = cas.c
if c.dataqsiz > 0 {
throw("selectgo: shouldn't happen")
}
if debugSelect { if debugSelect {
print("wait-return: sel=", sel, " c=", c, " cas=", cas, " kind=", cas.kind, "\n") print("wait-return: sel=", sel, " c=", c, " cas=", cas, " kind=", cas.kind, "\n")
} }
...@@ -470,7 +464,7 @@ loop: ...@@ -470,7 +464,7 @@ loop:
selunlock(sel) selunlock(sel)
goto retc goto retc
asyncrecv: bufrecv:
// can receive from buffer // can receive from buffer
if raceenabled { if raceenabled {
if cas.elem != nil { if cas.elem != nil {
...@@ -485,29 +479,20 @@ asyncrecv: ...@@ -485,29 +479,20 @@ asyncrecv:
if cas.receivedp != nil { if cas.receivedp != nil {
*cas.receivedp = true *cas.receivedp = true
} }
qp = chanbuf(c, c.recvx)
if cas.elem != nil { if cas.elem != nil {
typedmemmove(c.elemtype, cas.elem, chanbuf(c, c.recvx)) typedmemmove(c.elemtype, cas.elem, qp)
} }
memclr(chanbuf(c, c.recvx), uintptr(c.elemsize)) memclr(qp, uintptr(c.elemsize))
c.recvx++ c.recvx++
if c.recvx == c.dataqsiz { if c.recvx == c.dataqsiz {
c.recvx = 0 c.recvx = 0
} }
c.qcount-- c.qcount--
sg = c.sendq.dequeue() selunlock(sel)
if sg != nil {
gp = sg.g
selunlock(sel)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, 3)
} else {
selunlock(sel)
}
goto retc goto retc
asyncsend: bufsend:
// can send to buffer // can send to buffer
if raceenabled { if raceenabled {
raceacquire(chanbuf(c, c.sendx)) raceacquire(chanbuf(c, c.sendx))
...@@ -523,47 +508,18 @@ asyncsend: ...@@ -523,47 +508,18 @@ asyncsend:
c.sendx = 0 c.sendx = 0
} }
c.qcount++ c.qcount++
sg = c.recvq.dequeue() selunlock(sel)
if sg != nil {
gp = sg.g
selunlock(sel)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, 3)
} else {
selunlock(sel)
}
goto retc goto retc
syncrecv: recv:
// can receive from sleeping sender (sg) // can receive from sleeping sender (sg)
if raceenabled { recv(c, sg, cas.elem, func() { selunlock(sel) })
if cas.elem != nil {
raceWriteObjectPC(c.elemtype, cas.elem, cas.pc, chanrecvpc)
}
racesync(c, sg)
}
if msanenabled && cas.elem != nil {
msanwrite(cas.elem, c.elemtype.size)
}
selunlock(sel)
if debugSelect { if debugSelect {
print("syncrecv: sel=", sel, " c=", c, "\n") print("syncrecv: sel=", sel, " c=", c, "\n")
} }
if cas.receivedp != nil { if cas.receivedp != nil {
*cas.receivedp = true *cas.receivedp = true
} }
if cas.elem != nil {
typedmemmove(c.elemtype, cas.elem, sg.elem)
}
sg.elem = nil
gp = sg.g
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, 3)
goto retc goto retc
rclose: rclose:
...@@ -580,29 +536,19 @@ rclose: ...@@ -580,29 +536,19 @@ rclose:
} }
goto retc goto retc
syncsend: send:
// can send to sleeping receiver (sg) // can send to a sleeping receiver (sg)
if raceenabled { if raceenabled {
raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc) raceReadObjectPC(c.elemtype, cas.elem, cas.pc, chansendpc)
racesync(c, sg)
} }
if msanenabled { if msanenabled {
msanread(cas.elem, c.elemtype.size) msanread(cas.elem, c.elemtype.size)
} }
selunlock(sel) send(c, sg, cas.elem, func() { selunlock(sel) })
if debugSelect { if debugSelect {
print("syncsend: sel=", sel, " c=", c, "\n") print("syncsend: sel=", sel, " c=", c, "\n")
} }
if sg.elem != nil { goto retc
syncsend(c, sg, cas.elem)
}
sg.elem = nil
gp = sg.g
gp.param = unsafe.Pointer(sg)
if sg.releasetime != 0 {
sg.releasetime = cputicks()
}
goready(gp, 3)
retc: retc:
if cas.releasetime > 0 { if cas.releasetime > 0 {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7