diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc index 0de689a3c37..418740524a9 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()) { + if (!IsClosing() && !buffer->IsEmpty()) { data_ready_ = pending_read_; } else { OverlappedBuffer::DisposeBuffer(buffer); @@ -611,13 +611,10 @@ 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 67c8c5b21eb..bbd2179c98e 100644 --- a/runtime/bin/eventhandler_win.h +++ b/runtime/bin/eventhandler_win.h @@ -172,7 +172,6 @@ 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 8c8ef806208..f34671c80a8 100644 --- a/runtime/bin/io_natives.cc +++ b/runtime/bin/io_natives.cc @@ -132,7 +132,6 @@ namespace bin { V(ServerSocket_CreateBindListen, 7) \ V(SocketBase_IsBindError, 2) \ V(Socket_Available, 1) \ - V(Socket_AvailableDatagram, 1) \ V(Socket_CreateBindConnect, 5) \ V(Socket_CreateBindDatagram, 6) \ V(Socket_CreateConnect, 4) \ diff --git a/runtime/bin/socket.cc b/runtime/bin/socket.cc index 89c88830434..c96530ba7ae 100644 --- a/runtime/bin/socket.cc +++ b/runtime/bin/socket.cc @@ -399,12 +399,17 @@ void FUNCTION_NAME(Socket_RecvFrom)(Dart_NativeArguments args) { RawAddr addr; const intptr_t bytes_read = SocketBase::RecvFrom( socket->fd(), recv_buffer, kReceiveBufferLen, &addr, SocketBase::kAsync); + if (bytes_read == 0) { + Dart_SetReturnValue(args, Dart_Null()); + return; + } if (bytes_read < 0) { + ASSERT(bytes_read == -1); Dart_ThrowException(DartUtils::NewDartOSError()); } // 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)) { @@ -416,11 +421,6 @@ 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); if (addr.addr.sa_family == AF_INET) { @@ -1015,18 +1015,6 @@ 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.cc b/runtime/bin/socket_base.cc index 8d48e3c9dfd..8789c433825 100644 --- a/runtime/bin/socket_base.cc +++ b/runtime/bin/socket_base.cc @@ -102,7 +102,7 @@ void SocketAddress::SetAddrPort(RawAddr* addr, intptr_t port) { } intptr_t SocketAddress::GetAddrPort(const RawAddr& addr) { - if (addr.addr.sa_family == AF_INET) { + if (addr.ss.ss_family == AF_INET) { return ntohs(addr.in.sin_port); } else { return ntohs(addr.in6.sin6_port); diff --git a/runtime/bin/socket_base.h b/runtime/bin/socket_base.h index f90acfc27cf..404357b7320 100644 --- a/runtime/bin/socket_base.h +++ b/runtime/bin/socket_base.h @@ -167,7 +167,6 @@ 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 5df1bcafd0f..c72c276a2ac 100644 --- a/runtime/bin/socket_base_android.cc +++ b/runtime/bin/socket_base_android.cc @@ -82,18 +82,14 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, socklen_t addr_len = sizeof(addr->ss); ssize_t read_bytes = TEMP_FAILURE_RETRY( recvfrom(fd, buffer, num_bytes, 0, &addr->addr, &addr_len)); + if ((sync == kAsync) && (read_bytes == -1) && (errno == EWOULDBLOCK)) { + // If the read would block we need to retry and therefore return 0 + // as the number of bytes written. + read_bytes = 0; + } 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 7ab16d489ad..9a52d96fb88 100644 --- a/runtime/bin/socket_base_linux.cc +++ b/runtime/bin/socket_base_linux.cc @@ -82,18 +82,14 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, socklen_t addr_len = sizeof(addr->ss); ssize_t read_bytes = TEMP_FAILURE_RETRY( recvfrom(fd, buffer, num_bytes, 0, &addr->addr, &addr_len)); + if ((sync == kAsync) && (read_bytes == -1) && (errno == EWOULDBLOCK)) { + // If the read would block we need to retry and therefore return 0 + // as the number of bytes written. + read_bytes = 0; + } 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 d91c83573de..a44891166a9 100644 --- a/runtime/bin/socket_base_macos.cc +++ b/runtime/bin/socket_base_macos.cc @@ -81,18 +81,14 @@ intptr_t SocketBase::RecvFrom(intptr_t fd, socklen_t addr_len = sizeof(addr->ss); ssize_t read_bytes = TEMP_FAILURE_RETRY( recvfrom(fd, buffer, num_bytes, 0, &addr->addr, &addr_len)); + if ((sync == kAsync) && (read_bytes == -1) && (errno == EWOULDBLOCK)) { + // If the read would block we need to retry and therefore return 0 + // as the number of bytes written. + read_bytes = 0; + } 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 0167dadea4c..b24dc054bc1 100644 --- a/runtime/bin/socket_base_win.cc +++ b/runtime/bin/socket_base_win.cc @@ -88,13 +88,6 @@ 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 ff350507e51..8d5bbe37b77 100644 --- a/sdk/lib/_internal/vm/bin/socket_patch.dart +++ b/sdk/lib/_internal/vm/bin/socket_patch.dart @@ -364,18 +364,12 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { // Holds the address used to connect or bind the socket. InternetAddress localAddress; - // 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. + // The number of available bytes to read. 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; @@ -765,19 +759,26 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { } Datagram receive() { - if (isClosing || isClosed || !_availableDatagram) return null; + if (isClosing || isClosed) return null; try { Datagram result = nativeRecvFrom(); - assert(result != null); + 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; + } + } if (resourceInfo != null) { - resourceInfo.totalRead += result.data.length; resourceInfo.didRead(); } if (!const bool.fromEnvironment("dart.vm.product")) { _SocketProfile.collectStatistic(nativeGetSocketId(), _SocketProfileType.readBytes, result?.data?.length); } - _availableDatagram = nativeAvailableDatagram(); return result; } catch (e) { reportError(e, StackTrace.current, "Receive failed"); @@ -912,7 +913,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { readEventIssued = false; if (isClosing) return; if (!sendReadEvents) return; - if (stopRead()) { + if (available == 0) { if (isClosedRead && !closedReadEventSent) { if (isClosedWrite) close(); var handler = eventHandlers[closedEvent]; @@ -932,14 +933,6 @@ 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; @@ -987,11 +980,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { if (isListening) { connections++; } else { - if (isUdp) { - _availableDatagram = nativeAvailableDatagram(); - } else { - available = nativeAvailable(); - } + available = nativeAvailable(); issueReadEvent(); continue; } @@ -1250,7 +1239,6 @@ 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 89bc1280a8e..8b5638d8582 100644 --- a/sdk_nnbd/lib/_internal/vm/bin/socket_patch.dart +++ b/sdk_nnbd/lib/_internal/vm/bin/socket_patch.dart @@ -364,18 +364,12 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { // Holds the address used to connect or bind the socket. late InternetAddress localAddress; - // 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. + // The number of available bytes to read. 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; @@ -770,21 +764,32 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { } Datagram? receive() { - if (isClosing || isClosed || !_availableDatagram) return null; + if (isClosing || isClosed) return null; try { - Datagram? result = nativeRecvFrom(); - assert(result != null); + 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(); + if (resourceInformation != null) { + resourceInformation.totalRead += datagram.data.length; + } + } if (resourceInformation != null) { - resourceInformation.totalRead += result?.data?.length!; resourceInformation.didRead(); } if (!const bool.fromEnvironment("dart.vm.product")) { _SocketProfile.collectStatistic(nativeGetSocketId(), - _SocketProfileType.readBytes, result?.data.length); + _SocketProfileType.readBytes, datagram?.data.length); } - _availableDatagram = nativeAvailableDatagram(); - return result; + return datagram; } catch (e) { reportError(e, StackTrace.current, "Receive failed"); return null; @@ -928,7 +933,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { readEventIssued = false; if (isClosing) return; if (!sendReadEvents) return; - if (stopRead()) { + if (available == 0) { if (isClosedRead && !closedReadEventSent) { if (isClosedWrite) close(); var handler = eventHandlers[closedEvent]; @@ -948,14 +953,6 @@ 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; @@ -1003,11 +1000,7 @@ class _NativeSocket extends _NativeSocketNativeWrapper with _ServiceObject { if (isListening) { connections++; } else { - if (isUdp) { - _availableDatagram = nativeAvailableDatagram(); - } else { - available = nativeAvailable(); - } + available = nativeAvailable(); issueReadEvent(); continue; } @@ -1273,7 +1266,6 @@ 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 deleted file mode 100644 index 6226c2a4fd2..00000000000 --- a/tests/standalone/io/raw_datagram_zero_length_test.dart +++ /dev/null @@ -1,31 +0,0 @@ -// 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) return; - var datagram = receiver.receive(); - Expect.equals(0, datagram?.data?.length); - 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 deleted file mode 100644 index 357802c6788..00000000000 --- a/tests/standalone_2/io/raw_datagram_zero_length_test.dart +++ /dev/null @@ -1,31 +0,0 @@ -// 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) return; - var datagram = receiver.receive(); - Expect.equals(0, datagram.data.length); - Expect.isNull(receiver.receive()); - receiver.close(); - sub.cancel(); - asyncEnd(); - }); - - sender.send(Uint8List(0), address, receiver.port); - sender.close(); -}