diff options
| -rw-r--r-- | pkg/flipcall/BUILD | 1 | ||||
| -rw-r--r-- | pkg/flipcall/ctrl_futex.go | 1 | ||||
| -rw-r--r-- | pkg/flipcall/flipcall.go | 10 | ||||
| -rw-r--r-- | pkg/flipcall/flipcall_test.go | 19 | ||||
| -rw-r--r-- | pkg/flipcall/flipcall_unsafe.go | 18 | ||||
| -rw-r--r-- | third_party/gvsync/downgradable_rwmutex_unsafe.go | 3 |
6 files changed, 48 insertions, 4 deletions
diff --git a/pkg/flipcall/BUILD b/pkg/flipcall/BUILD index bd1d614b6..c1e078c7c 100644 --- a/pkg/flipcall/BUILD +++ b/pkg/flipcall/BUILD @@ -18,6 +18,7 @@ go_library( "//pkg/abi/linux", "//pkg/log", "//pkg/memutil", + "//third_party/gvsync", ], ) diff --git a/pkg/flipcall/ctrl_futex.go b/pkg/flipcall/ctrl_futex.go index d59159912..8390915a2 100644 --- a/pkg/flipcall/ctrl_futex.go +++ b/pkg/flipcall/ctrl_futex.go @@ -82,6 +82,7 @@ func (ep *Endpoint) ctrlWaitFirst() error { *ep.dataLen() = w.Len() // Return control to the client. + raceBecomeInactive() if err := ep.futexSwitchToPeer(); err != nil { return err } diff --git a/pkg/flipcall/flipcall.go b/pkg/flipcall/flipcall.go index 991018684..386cee42c 100644 --- a/pkg/flipcall/flipcall.go +++ b/pkg/flipcall/flipcall.go @@ -180,7 +180,11 @@ const ( // Preconditions: ep is a client Endpoint. ep.Connect(), ep.RecvFirst(), // ep.SendRecv(), and ep.SendLast() have never been called. func (ep *Endpoint) Connect() error { - return ep.ctrlConnect() + err := ep.ctrlConnect() + if err == nil { + raceBecomeActive() + } + return err } // RecvFirst blocks until the peer Endpoint calls Endpoint.SendRecv(), then @@ -192,6 +196,7 @@ func (ep *Endpoint) RecvFirst() (uint32, error) { if err := ep.ctrlWaitFirst(); err != nil { return 0, err } + raceBecomeActive() recvDataLen := atomic.LoadUint32(ep.dataLen()) if recvDataLen > ep.dataCap { return 0, fmt.Errorf("received packet with invalid datagram length %d (maximum %d)", recvDataLen, ep.dataCap) @@ -218,9 +223,11 @@ func (ep *Endpoint) SendRecv(dataLen uint32) (uint32, error) { // after ep.ctrlRoundTrip(), so if the peer is mutating it concurrently then // they can only shoot themselves in the foot. *ep.dataLen() = dataLen + raceBecomeInactive() if err := ep.ctrlRoundTrip(); err != nil { return 0, err } + raceBecomeActive() recvDataLen := atomic.LoadUint32(ep.dataLen()) if recvDataLen > ep.dataCap { return 0, fmt.Errorf("received packet with invalid datagram length %d (maximum %d)", recvDataLen, ep.dataCap) @@ -240,6 +247,7 @@ func (ep *Endpoint) SendLast(dataLen uint32) error { panic(fmt.Sprintf("attempting to send packet with datagram length %d (maximum %d)", dataLen, ep.dataCap)) } *ep.dataLen() = dataLen + raceBecomeInactive() if err := ep.ctrlWakeLast(); err != nil { return err } diff --git a/pkg/flipcall/flipcall_test.go b/pkg/flipcall/flipcall_test.go index 435e4eeae..168a487ec 100644 --- a/pkg/flipcall/flipcall_test.go +++ b/pkg/flipcall/flipcall_test.go @@ -62,6 +62,9 @@ func (c *testConnection) destroy() { } func testSendRecv(t *testing.T, c *testConnection) { + // This shared variable is used to confirm that synchronization between + // flipcall endpoints is visible to the Go race detector. + state := 0 var serverRun sync.WaitGroup serverRun.Add(1) go func() { @@ -71,11 +74,19 @@ func testSendRecv(t *testing.T, c *testConnection) { t.Errorf("server Endpoint.RecvFirst() failed: %v", err) return } + state++ + if state != 2 { + t.Errorf("shared state counter: got %d, wanted 2", state) + } t.Logf("server Endpoint got packet 1, sending packet 2 and waiting for packet 3") if _, err := c.serverEP.SendRecv(0); err != nil { t.Errorf("server Endpoint.SendRecv() failed: %v", err) return } + state++ + if state != 4 { + t.Errorf("shared state counter: got %d, wanted 4", state) + } t.Logf("server Endpoint got packet 3") }() defer func() { @@ -89,10 +100,18 @@ func testSendRecv(t *testing.T, c *testConnection) { if err := c.clientEP.Connect(); err != nil { t.Fatalf("client Endpoint.Connect() failed: %v", err) } + state++ + if state != 1 { + t.Errorf("shared state counter: got %d, wanted 1", state) + } t.Logf("client Endpoint sending packet 1 and waiting for packet 2") if _, err := c.clientEP.SendRecv(0); err != nil { t.Fatalf("client Endpoint.SendRecv() failed: %v", err) } + state++ + if state != 3 { + t.Errorf("shared state counter: got %d, wanted 3", state) + } t.Logf("client Endpoint got packet 2, sending packet 3") if err := c.clientEP.SendLast(0); err != nil { t.Fatalf("client Endpoint.SendLast() failed: %v", err) diff --git a/pkg/flipcall/flipcall_unsafe.go b/pkg/flipcall/flipcall_unsafe.go index 73e6eef29..a37952637 100644 --- a/pkg/flipcall/flipcall_unsafe.go +++ b/pkg/flipcall/flipcall_unsafe.go @@ -17,6 +17,8 @@ package flipcall import ( "reflect" "unsafe" + + "gvisor.dev/gvisor/third_party/gvsync" ) // Packets consist of a 16-byte header followed by an arbitrarily-sized @@ -67,3 +69,19 @@ func (ep *Endpoint) Data() []byte { bsReflect.Cap = int(ep.dataCap) return bs } + +// ioSync is a dummy variable used to indicate synchronization to the Go race +// detector. Compare syscall.ioSync. +var ioSync int64 + +func raceBecomeActive() { + if gvsync.RaceEnabled { + gvsync.RaceAcquire((unsafe.Pointer)(&ioSync)) + } +} + +func raceBecomeInactive() { + if gvsync.RaceEnabled { + gvsync.RaceReleaseMerge((unsafe.Pointer)(&ioSync)) + } +} diff --git a/third_party/gvsync/downgradable_rwmutex_unsafe.go b/third_party/gvsync/downgradable_rwmutex_unsafe.go index 069939033..1f6007aa1 100644 --- a/third_party/gvsync/downgradable_rwmutex_unsafe.go +++ b/third_party/gvsync/downgradable_rwmutex_unsafe.go @@ -57,9 +57,6 @@ func (rw *DowngradableRWMutex) RLock() { // RUnlock undoes a single RLock call. func (rw *DowngradableRWMutex) RUnlock() { if RaceEnabled { - // TODO(jamieliu): Why does this need to be ReleaseMerge instead of - // Release? IIUC this establishes Unlock happens-before RUnlock, which - // seems unnecessary. RaceReleaseMerge(unsafe.Pointer(&rw.writerSem)) RaceDisable() } |