From fdf4762c9946bf0b239b6ca90c6c8ff8f4cdeaeb Mon Sep 17 00:00:00 2001 From: Zichang Guo Date: Fri, 27 Mar 2020 15:28:23 +0000 Subject: [PATCH] Reland "[dart:io] Fix hanging on zero-length datagram" This is a reland of c326c587c5e34db92bc55b454f82cd02121b9288 The fix is to remove "_availableDatagram" check at the start of receive(). Since receive() can be called without receiver being listening, this check will block the receive(). Original change's description: > [dart:io] Fix hanging on zero-length datagram > > This is observed on Win and Linux. > Here is a doc to explain the problem: https://docs.google.com/document/d/1ZzyBUMrDHLU6vNryjgSMJfUruOdpmTImtE_s2E0J8IQ/edit?usp=sharing. > > Bug: https://github.com/dart-lang/sdk/issues/39910 > Change-Id: Ia961239f45615f14108bcd66043ac33d9a0a4abe > Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/137425 > Commit-Queue: Zichang Guo > Reviewed-by: Siva Annamalai Bug: https://github.com/dart-lang/sdk/issues/39910 Change-Id: Iefc0af96ed4e1b433396bb5fe11276e8a04c00d5 Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/140164 Reviewed-by: Siva Annamalai Commit-Queue: Zichang Guo --- runtime/bin/eventhandler_win.cc | 7 ++- runtime/bin/eventhandler_win.h | 1 + runtime/bin/io_natives.cc | 1 + runtime/bin/socket.cc | 19 +++++++- runtime/bin/socket_base.h | 1 + runtime/bin/socket_base_android.cc | 9 ++++ runtime/bin/socket_base_linux.cc | 9 ++++ runtime/bin/socket_base_macos.cc | 9 ++++ runtime/bin/socket_base_win.cc | 7 +++ sdk/lib/_internal/vm/bin/socket_patch.dart | 31 ++++++++---- .../lib/_internal/vm/bin/socket_patch.dart | 48 ++++++++++++------- .../io/raw_datagram_zero_length_test.dart | 30 ++++++++++++ .../io/raw_datagram_zero_length_test.dart | 30 ++++++++++++ 13 files changed, 173 insertions(+), 29 deletions(-) create mode 100644 tests/standalone/io/raw_datagram_zero_length_test.dart create mode 100644 tests/standalone_2/io/raw_datagram_zero_length_test.dart diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc index 418740524a9..0de689a3c37 100644 --- a/runtime/bin/eventhandler_win.cc +++ b/runtime/bin/eventhandler_win.cc @@ -208,7 +208,7 @@ void Handle::ReadComplete(OverlappedBuffer* buffer) { // Currently only one outstanding read at the time. ASSERT(pending_read_ == buffer); ASSERT(data_ready_ == NULL); - if (!IsClosing() && !buffer->IsEmpty()) { + if (!IsClosing()) { data_ready_ = pending_read_; } else { OverlappedBuffer::DisposeBuffer(buffer); @@ -611,10 +611,13 @@ intptr_t Handle::Available() { if (data_ready_ == NULL) { return 0; } - ASSERT(!data_ready_->IsEmpty()); return data_ready_->GetRemainingLength(); } +bool Handle::DataReady() { + return data_ready_ != NULL; +} + intptr_t Handle::Read(void* buffer, intptr_t num_bytes) { MonitorLocker ml(&monitor_); if (data_ready_ == NULL) { diff --git a/runtime/bin/eventhandler_win.h b/runtime/bin/eventhandler_win.h index bbd2179c98e..67c8c5b21eb 100644 --- a/runtime/bin/eventhandler_win.h +++ b/runtime/bin/eventhandler_win.h @@ -172,6 +172,7 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { // Socket interface exposing normal socket operations. intptr_t Available(); + bool DataReady(); intptr_t Read(void* buffer, intptr_t num_bytes); intptr_t RecvFrom(void* buffer, intptr_t num_bytes, diff --git a/runtime/bin/io_natives.cc b/runtime/bin/io_natives.cc index 155d415f240..6ee986d260a 100644 --- a/runtime/bin/io_natives.cc +++ b/runtime/bin/io_natives.cc @@ -134,6 +134,7 @@ namespace bin { V(ServerSocket_CreateUnixDomainBindListen, 5) \ V(SocketBase_IsBindError, 2) \ V(Socket_Available, 1) \ + V(Socket_AvailableDatagram, 1) \ V(Socket_CreateBindConnect, 5) \ V(Socket_CreateUnixDomainBindConnect, 4) \ V(Socket_CreateBindDatagram, 6) \ diff --git a/runtime/bin/socket.cc b/runtime/bin/socket.cc index 036d9f3e6a8..46bcef0e959 100644 --- a/runtime/bin/socket.cc +++ b/runtime/bin/socket.cc @@ -557,7 +557,7 @@ void FUNCTION_NAME(Socket_RecvFrom)(Dart_NativeArguments args) { } // Datagram data read. Copy into buffer of the exact size, - ASSERT(bytes_read > 0); + ASSERT(bytes_read >= 0); uint8_t* data_buffer = NULL; Dart_Handle data = IOBuffer::Allocate(bytes_read, &data_buffer); if (Dart_IsNull(data)) { @@ -569,6 +569,11 @@ void FUNCTION_NAME(Socket_RecvFrom)(Dart_NativeArguments args) { ASSERT(data_buffer != NULL); memmove(data_buffer, recv_buffer, bytes_read); + // Memory Sanitizer complains addr not being initialized, which is done + // through RecvFrom(). + // Issue: https://github.com/google/sanitizers/issues/1201 + MSAN_UNPOISON(&addr, sizeof(RawAddr)); + // Get the port and clear it in the sockaddr structure. int port = SocketAddress::GetAddrPort(addr); // TODO(21403): Add checks for AF_UNIX, if unix domain sockets @@ -1200,6 +1205,18 @@ void FUNCTION_NAME(Socket_LeaveMulticast)(Dart_NativeArguments args) { } } +void FUNCTION_NAME(Socket_AvailableDatagram)(Dart_NativeArguments args) { + const int kReceiveBufferLen = 1; + Socket* socket = + Socket::GetSocketIdNativeField(Dart_GetNativeArgument(args, 0)); + ASSERT(socket != NULL); + // Ensure that a receive buffer for peeking the UDP socket exists. + uint8_t recv_buffer[kReceiveBufferLen]; + bool available = SocketBase::AvailableDatagram(socket->fd(), recv_buffer, + kReceiveBufferLen); + Dart_SetBooleanReturnValue(args, available); +} + static void NormalSocketFinalizer(void* isolate_data, Dart_WeakPersistentHandle handle, void* data) { diff --git a/runtime/bin/socket_base.h b/runtime/bin/socket_base.h index 8af4c593c19..bf9405ef001 100644 --- a/runtime/bin/socket_base.h +++ b/runtime/bin/socket_base.h @@ -185,6 +185,7 @@ class SocketBase : public AllStatic { intptr_t num_bytes, RawAddr* addr, SocketOpKind sync); + static bool AvailableDatagram(intptr_t fd, void* buffer, intptr_t num_bytes); // Returns true if the given error-number is because the system was not able // to bind the socket to a specific IP. static bool IsBindError(intptr_t error_number); diff --git a/runtime/bin/socket_base_android.cc b/runtime/bin/socket_base_android.cc index 6cea30d3883..9085f1448da 100644 --- a/runtime/bin/socket_base_android.cc +++ b/runtime/bin/socket_base_android.cc @@ -98,6 +98,15 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, return read_bytes; } +bool SocketBase::AvailableDatagram(intptr_t fd, + void* buffer, + intptr_t num_bytes) { + ASSERT(fd >= 0); + ssize_t read_bytes = + TEMP_FAILURE_RETRY(recvfrom(fd, buffer, num_bytes, MSG_PEEK, NULL, NULL)); + return read_bytes >= 0; +} + intptr_t SocketBase::Write(intptr_t fd, const void* buffer, intptr_t num_bytes, diff --git a/runtime/bin/socket_base_linux.cc b/runtime/bin/socket_base_linux.cc index fb1e5550535..8734aa68bbf 100644 --- a/runtime/bin/socket_base_linux.cc +++ b/runtime/bin/socket_base_linux.cc @@ -98,6 +98,15 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, return read_bytes; } +bool SocketBase::AvailableDatagram(intptr_t fd, + void* buffer, + intptr_t num_bytes) { + ASSERT(fd >= 0); + ssize_t read_bytes = + TEMP_FAILURE_RETRY(recvfrom(fd, buffer, num_bytes, MSG_PEEK, NULL, NULL)); + return read_bytes >= 0; +} + intptr_t SocketBase::Write(intptr_t fd, const void* buffer, intptr_t num_bytes, diff --git a/runtime/bin/socket_base_macos.cc b/runtime/bin/socket_base_macos.cc index e8bea831e2c..74bf924420c 100644 --- a/runtime/bin/socket_base_macos.cc +++ b/runtime/bin/socket_base_macos.cc @@ -97,6 +97,15 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, return read_bytes; } +bool SocketBase::AvailableDatagram(intptr_t fd, + void* buffer, + intptr_t num_bytes) { + ASSERT(fd >= 0); + ssize_t read_bytes = + TEMP_FAILURE_RETRY(recvfrom(fd, buffer, num_bytes, MSG_PEEK, NULL, NULL)); + return read_bytes >= 0; +} + intptr_t SocketBase::Write(intptr_t fd, const void* buffer, intptr_t num_bytes, diff --git a/runtime/bin/socket_base_win.cc b/runtime/bin/socket_base_win.cc index 4ce3fc053d9..f3a27562c6b 100644 --- a/runtime/bin/socket_base_win.cc +++ b/runtime/bin/socket_base_win.cc @@ -91,6 +91,13 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, return handle->RecvFrom(buffer, num_bytes, &addr->addr, addr_len); } +bool SocketBase::AvailableDatagram(intptr_t fd, + void* buffer, + intptr_t num_bytes) { + ClientSocket* client_socket = reinterpret_cast(fd); + return client_socket->DataReady(); +} + intptr_t SocketBase::Write(intptr_t fd, const void* buffer, intptr_t num_bytes, diff --git a/sdk/lib/_internal/vm/bin/socket_patch.dart b/sdk/lib/_internal/vm/bin/socket_patch.dart index f0abd96be83..e50d82123c6 100644 --- a/sdk/lib/_internal/vm/bin/socket_patch.dart +++ b/sdk/lib/_internal/vm/bin/socket_patch.dart @@ -425,12 +425,18 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { // Holds the address used to connect or bind the socket. InternetAddress localAddress; - // The number of available bytes to read. + // The size of data that is ready to be read, for TCP sockets. + // This might be out-of-date when Read is called. + // The number of pending connections, for Listening sockets. int available = 0; + // Only used for UDP sockets. + bool _availableDatagram = false; + // The number of incoming connnections for Listening socket. int connections = 0; + // The count of received event from eventhandler. int tokens = 0; bool sendReadEvents = false; @@ -845,11 +851,6 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { try { Datagram result = nativeRecvFrom(); if (result != null) { - // Read the next available. Available is only for the next datagram, not - // the sum of all datagrams pending, so we need to call after each - // receive. If available becomes > 0, the _NativeSocket will continue to - // emit read events. - available = nativeAvailable(); if (resourceInfo != null) { resourceInfo.totalRead += result.data.length; } @@ -861,6 +862,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { _SocketProfile.collectStatistic(nativeGetSocketId(), _SocketProfileType.readBytes, result?.data?.length); } + _availableDatagram = nativeAvailableDatagram(); return result; } catch (e) { reportError(e, StackTrace.current, "Receive failed"); @@ -1001,7 +1003,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { readEventIssued = false; if (isClosing) return; if (!sendReadEvents) return; - if (available == 0) { + if (stopRead()) { if (isClosedRead && !closedReadEventSent) { if (isClosedWrite) close(); var handler = eventHandlers[closedEvent]; @@ -1021,6 +1023,14 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { scheduleMicrotask(issue); } + bool stopRead() { + if (isUdp) { + return !_availableDatagram; + } else { + return available == 0; + } + } + void issueWriteEvent({bool delayed: true}) { if (writeEventIssued) return; if (!writeAvailable) return; @@ -1068,7 +1078,11 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { if (isListening) { connections++; } else { - available = nativeAvailable(); + if (isUdp) { + _availableDatagram = nativeAvailableDatagram(); + } else { + available = nativeAvailable(); + } issueReadEvent(); continue; } @@ -1327,6 +1341,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { void nativeSetSocketId(int id, int typeFlags) native "Socket_SetSocketId"; int nativeAvailable() native "Socket_Available"; + bool nativeAvailableDatagram() native "Socket_AvailableDatagram"; Uint8List nativeRead(int len) native "Socket_Read"; Datagram nativeRecvFrom() native "Socket_RecvFrom"; int nativeWrite(List buffer, int offset, int bytes) diff --git a/sdk_nnbd/lib/_internal/vm/bin/socket_patch.dart b/sdk_nnbd/lib/_internal/vm/bin/socket_patch.dart index 8138f873dbc..8ef06ed3091 100644 --- a/sdk_nnbd/lib/_internal/vm/bin/socket_patch.dart +++ b/sdk_nnbd/lib/_internal/vm/bin/socket_patch.dart @@ -427,12 +427,18 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { // Holds the address used to connect or bind the socket. late InternetAddress localAddress; - // The number of available bytes to read. + // The size of data that is ready to be read, for TCP sockets. + // This might be out-of-date when Read is called. + // The number of pending connections, for Listening sockets. int available = 0; + // Only used for UDP sockets. + bool _availableDatagram = false; + // The number of incoming connnections for Listening socket. int connections = 0; + // The count of received event from eventhandler. int tokens = 0; bool sendReadEvents = false; @@ -850,30 +856,23 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { Datagram? receive() { if (isClosing || isClosed) return null; try { - Datagram? datagram = nativeRecvFrom(); - final resourceInformation = resourceInfo; - assert(resourceInformation != null || - isPipe || - isInternal || - isInternalSignal); - if (datagram != null) { - // Read the next available. Available is only for the next datagram, not - // the sum of all datagrams pending, so we need to call after each - // receive. If available becomes > 0, the _NativeSocket will continue to - // emit read events. - available = nativeAvailable(); + Datagram? result = nativeRecvFrom(); + if (result != null) { + final resourceInformation = resourceInfo; if (resourceInformation != null) { - resourceInformation.totalRead += datagram.data.length; + resourceInformation.totalRead += result.data.length; } } + final resourceInformation = resourceInfo; if (resourceInformation != null) { resourceInformation.didRead(); } if (!const bool.fromEnvironment("dart.vm.product")) { _SocketProfile.collectStatistic(nativeGetSocketId(), - _SocketProfileType.readBytes, datagram?.data.length); + _SocketProfileType.readBytes, result?.data.length); } - return datagram; + _availableDatagram = nativeAvailableDatagram(); + return result; } catch (e) { reportError(e, StackTrace.current, "Receive failed"); return null; @@ -1023,7 +1022,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { readEventIssued = false; if (isClosing) return; if (!sendReadEvents) return; - if (available == 0) { + if (stopRead()) { if (isClosedRead && !closedReadEventSent) { if (isClosedWrite) close(); var handler = eventHandlers[closedEvent]; @@ -1043,6 +1042,14 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { scheduleMicrotask(issue); } + bool stopRead() { + if (isUdp) { + return !_availableDatagram; + } else { + return available == 0; + } + } + void issueWriteEvent({bool delayed: true}) { if (writeEventIssued) return; if (!writeAvailable) return; @@ -1090,7 +1097,11 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { if (isListening) { connections++; } else { - available = nativeAvailable(); + if (isUdp) { + _availableDatagram = nativeAvailableDatagram(); + } else { + available = nativeAvailable(); + } issueReadEvent(); continue; } @@ -1356,6 +1367,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { void nativeSetSocketId(int id, int typeFlags) native "Socket_SetSocketId"; int nativeAvailable() native "Socket_Available"; + bool nativeAvailableDatagram() native "Socket_AvailableDatagram"; Uint8List? nativeRead(int len) native "Socket_Read"; Datagram? nativeRecvFrom() native "Socket_RecvFrom"; int nativeWrite(List buffer, int offset, int bytes) diff --git a/tests/standalone/io/raw_datagram_zero_length_test.dart b/tests/standalone/io/raw_datagram_zero_length_test.dart new file mode 100644 index 00000000000..29936352cc1 --- /dev/null +++ b/tests/standalone/io/raw_datagram_zero_length_test.dart @@ -0,0 +1,30 @@ +// Copyright (c) 2020, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import "dart:async"; +import "dart:io"; +import "dart:typed_data"; + +import "package:async_helper/async_helper.dart"; +import "package:expect/expect.dart"; + +main() async { + asyncStart(); + var address = InternetAddress.loopbackIPv4; + var sender = await RawDatagramSocket.bind(address, 0); + var receiver = await RawDatagramSocket.bind(address, 0); + + var sub; + sub = receiver.listen((event) { + if (event == RawSocketEvent.read) { + Expect.isNull(receiver.receive()); + } + receiver.close(); + sub.cancel(); + asyncEnd(); + }); + + sender.send(Uint8List(0), address, receiver.port); + sender.close(); +} diff --git a/tests/standalone_2/io/raw_datagram_zero_length_test.dart b/tests/standalone_2/io/raw_datagram_zero_length_test.dart new file mode 100644 index 00000000000..29936352cc1 --- /dev/null +++ b/tests/standalone_2/io/raw_datagram_zero_length_test.dart @@ -0,0 +1,30 @@ +// Copyright (c) 2020, the Dart project authors. Please see the AUTHORS file +// for details. All rights reserved. Use of this source code is governed by a +// BSD-style license that can be found in the LICENSE file. + +import "dart:async"; +import "dart:io"; +import "dart:typed_data"; + +import "package:async_helper/async_helper.dart"; +import "package:expect/expect.dart"; + +main() async { + asyncStart(); + var address = InternetAddress.loopbackIPv4; + var sender = await RawDatagramSocket.bind(address, 0); + var receiver = await RawDatagramSocket.bind(address, 0); + + var sub; + sub = receiver.listen((event) { + if (event == RawSocketEvent.read) { + Expect.isNull(receiver.receive()); + } + receiver.close(); + sub.cancel(); + asyncEnd(); + }); + + sender.send(Uint8List(0), address, receiver.port); + sender.close(); +}