runtime: converge duplicate calls to netpollBreak into one
There might be some concurrent (maybe not concurrent, just sequential but in a short time window) and duplicate calls to `netpollBreak`, trying to wake up a net-poller. If one has called `netpollBreak` and that waking event hasn't been received by epollwait/kevent/..., then the subsequent calls of `netpollBreak` ought to be ignored or in other words, these calls should be converged into one.
Benchmarks go1.13.5 darwin/amd64:
benchmark-func time/op (old) time/op (new) delta
BenchmarkNetpollBreak-4 29668ns ±1% 3131ns ±2% -89.45%
mem/B (old) mem/B (new) delta
154B ±13% 0B ±0% -100%
Change-Id: I3cf757a5d6edc5a99adad7aea3baee4b7f2a8f5c
GitHub-Last-Rev: 15bcfbab8a5db51f65da01315a5880a5dbf9e028
GitHub-Pull-Request: golang/go#36294
Reviewed-on: https://go-review.googlesource.com/c/go/+/212737
Run-TryBot: Emmanuel Odeke <emm.odeke@gmail.com>
TryBot-Result: Gobot Gobot <gobot@golang.org>
Reviewed-by: Ian Lance Taylor <iant@golang.org>
diff --git a/src/runtime/netpoll_epoll.go b/src/runtime/netpoll_epoll.go
index b9dc18c..cc4c36f 100644
--- a/src/runtime/netpoll_epoll.go
+++ b/src/runtime/netpoll_epoll.go
@@ -6,7 +6,10 @@
package runtime
-import "unsafe"
+import (
+ "runtime/internal/atomic"
+ "unsafe"
+)
func epollcreate(size int32) int32
func epollcreate1(flags int32) int32
@@ -22,6 +25,8 @@
epfd int32 = -1 // epoll descriptor
netpollBreakRd, netpollBreakWr uintptr // for netpollBreak
+
+ netpollWakeSig uintptr // used to avoid duplicate calls of netpollBreak
)
func netpollinit() {
@@ -74,20 +79,22 @@
// netpollBreak interrupts an epollwait.
func netpollBreak() {
- for {
- var b byte
- n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
- if n == 1 {
- break
+ if atomic.Casuintptr(&netpollWakeSig, 0, 1) {
+ for {
+ var b byte
+ n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
+ if n == 1 {
+ break
+ }
+ if n == -_EINTR {
+ continue
+ }
+ if n == -_EAGAIN {
+ return
+ }
+ println("runtime: netpollBreak write failed with", -n)
+ throw("runtime: netpollBreak write failed")
}
- if n == -_EINTR {
- continue
- }
- if n == -_EAGAIN {
- return
- }
- println("runtime: netpollBreak write failed with", -n)
- throw("runtime: netpollBreak write failed")
}
}
@@ -147,6 +154,7 @@
// if blocking.
var tmp [16]byte
read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
+ atomic.Storeuintptr(&netpollWakeSig, 0)
}
continue
}
diff --git a/src/runtime/netpoll_kqueue.go b/src/runtime/netpoll_kqueue.go
index 39d4022..2ff21d8 100644
--- a/src/runtime/netpoll_kqueue.go
+++ b/src/runtime/netpoll_kqueue.go
@@ -8,12 +8,17 @@
// Integrated network poller (kqueue-based implementation).
-import "unsafe"
+import (
+ "runtime/internal/atomic"
+ "unsafe"
+)
var (
kq int32 = -1
netpollBreakRd, netpollBreakWr uintptr // for netpollBreak
+
+ netpollWakeSig uintptr // used to avoid duplicate calls of netpollBreak
)
func netpollinit() {
@@ -78,17 +83,19 @@
// netpollBreak interrupts a kevent.
func netpollBreak() {
- for {
- var b byte
- n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
- if n == 1 || n == -_EAGAIN {
- break
+ if atomic.Casuintptr(&netpollWakeSig, 0, 1) {
+ for {
+ var b byte
+ n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
+ if n == 1 || n == -_EAGAIN {
+ break
+ }
+ if n == -_EINTR {
+ continue
+ }
+ println("runtime: netpollBreak write failed with", -n)
+ throw("runtime: netpollBreak write failed")
}
- if n == -_EINTR {
- continue
- }
- println("runtime: netpollBreak write failed with", -n)
- throw("runtime: netpollBreak write failed")
}
}
@@ -145,6 +152,7 @@
// if blocking.
var tmp [16]byte
read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
+ atomic.Storeuintptr(&netpollWakeSig, 0)
}
continue
}
diff --git a/src/runtime/netpoll_os_test.go b/src/runtime/netpoll_os_test.go
new file mode 100644
index 0000000..b96b9f3
--- /dev/null
+++ b/src/runtime/netpoll_os_test.go
@@ -0,0 +1,28 @@
+package runtime_test
+
+import (
+ "runtime"
+ "sync"
+ "testing"
+)
+
+var wg sync.WaitGroup
+
+func init() {
+ runtime.NetpollGenericInit()
+}
+
+func BenchmarkNetpollBreak(b *testing.B) {
+ b.StartTimer()
+ for i := 0; i < b.N; i++ {
+ for j := 0; j < 10; j++ {
+ wg.Add(1)
+ go func() {
+ runtime.NetpollBreak()
+ wg.Done()
+ }()
+ }
+ }
+ wg.Wait()
+ b.StopTimer()
+}
diff --git a/src/runtime/netpoll_solaris.go b/src/runtime/netpoll_solaris.go
index 15818cb..34b3ee9 100644
--- a/src/runtime/netpoll_solaris.go
+++ b/src/runtime/netpoll_solaris.go
@@ -4,7 +4,10 @@
package runtime
-import "unsafe"
+import (
+ "runtime/internal/atomic"
+ "unsafe"
+)
// Solaris runtime-integrated network poller.
//
@@ -85,6 +88,7 @@
libc_port_dissociate,
libc_port_getn,
libc_port_alert libcFunc
+ netpollWakeSig uintptr // used to avoid duplicate calls of netpollBreak
)
func errno() int32 {
@@ -187,15 +191,17 @@
// netpollBreak interrupts a port_getn wait.
func netpollBreak() {
- // Use port_alert to put portfd into alert mode.
- // This will wake up all threads sleeping in port_getn on portfd,
- // and cause their calls to port_getn to return immediately.
- // Further, until portfd is taken out of alert mode,
- // all calls to port_getn will return immediately.
- if port_alert(portfd, _PORT_ALERT_UPDATE, _POLLHUP, uintptr(unsafe.Pointer(&portfd))) < 0 {
- if e := errno(); e != _EBUSY {
- println("runtime: port_alert failed with", e)
- throw("runtime: netpoll: port_alert failed")
+ if atomic.Casuintptr(&netpollWakeSig, 0, 1) {
+ // Use port_alert to put portfd into alert mode.
+ // This will wake up all threads sleeping in port_getn on portfd,
+ // and cause their calls to port_getn to return immediately.
+ // Further, until portfd is taken out of alert mode,
+ // all calls to port_getn will return immediately.
+ if port_alert(portfd, _PORT_ALERT_UPDATE, _POLLHUP, uintptr(unsafe.Pointer(&portfd))) < 0 {
+ if e := errno(); e != _EBUSY {
+ println("runtime: port_alert failed with", e)
+ throw("runtime: netpoll: port_alert failed")
+ }
}
}
}
@@ -268,6 +274,7 @@
println("runtime: port_alert failed with", e)
throw("runtime: netpoll: port_alert failed")
}
+ atomic.Storeuintptr(&netpollWakeSig, 0)
}
continue
}
diff --git a/src/runtime/netpoll_windows.go b/src/runtime/netpoll_windows.go
index 28b2f6e..56a0798 100644
--- a/src/runtime/netpoll_windows.go
+++ b/src/runtime/netpoll_windows.go
@@ -5,6 +5,7 @@
package runtime
import (
+ "runtime/internal/atomic"
"unsafe"
)
@@ -31,7 +32,11 @@
qty uint32
}
-var iocphandle uintptr = _INVALID_HANDLE_VALUE // completion port io handle
+var (
+ iocphandle uintptr = _INVALID_HANDLE_VALUE // completion port io handle
+
+ netpollWakeSig uintptr // used to avoid duplicate calls of netpollBreak
+)
func netpollinit() {
iocphandle = stdcall4(_CreateIoCompletionPort, _INVALID_HANDLE_VALUE, 0, 0, _DWORD_MAX)
@@ -62,9 +67,11 @@
}
func netpollBreak() {
- if stdcall4(_PostQueuedCompletionStatus, iocphandle, 0, 0, 0) == 0 {
- println("runtime: netpoll: PostQueuedCompletionStatus failed (errno=", getlasterror(), ")")
- throw("runtime: netpoll: PostQueuedCompletionStatus failed")
+ if atomic.Casuintptr(&netpollWakeSig, 0, 1) {
+ if stdcall4(_PostQueuedCompletionStatus, iocphandle, 0, 0, 0) == 0 {
+ println("runtime: netpoll: PostQueuedCompletionStatus failed (errno=", getlasterror(), ")")
+ throw("runtime: netpoll: PostQueuedCompletionStatus failed")
+ }
}
}
@@ -126,6 +133,7 @@
}
handlecompletion(&toRun, op, errno, qty)
} else {
+ atomic.Storeuintptr(&netpollWakeSig, 0)
if delay == 0 {
// Forward the notification to the
// blocked poller.
