From 6fbdd19ddb8cc5c9bb38fb1cd910c232a3999d28 Mon Sep 17 00:00:00 2001 From: Ben Konyi Date: Wed, 12 Apr 2017 14:20:43 -0700 Subject: [PATCH] Various fixes for sync socket implementation. Revert "Revert "Added sync_socket.h include in sync_socket_win.cc"" This reverts commit 2bd6168e1ebf35f6dbd9b2724339362c493377fc. Revert "Revert "Added synchronous socket implementation to dart:io."" This reverts commit a47bce03f5502dac7fb12e777e87b5aa1cfc830b. BUG= R=zra@google.com Review-Url: https://codereview.chromium.org/2814773004 . --- .../tool/input_sdk/patch/io_patch.dart | 8 + runtime/bin/io_impl_sources.gypi | 7 + runtime/bin/io_natives.cc | 11 + runtime/bin/io_sources.gypi | 1 + runtime/bin/socket_android.cc | 11 +- runtime/bin/socket_fuchsia.cc | 15 +- runtime/bin/socket_linux.cc | 14 +- runtime/bin/socket_macos.cc | 13 +- runtime/bin/socket_unsupported.cc | 1 + runtime/bin/sync_socket.cc | 365 +++++++++++++ runtime/bin/sync_socket.h | 49 ++ runtime/bin/sync_socket_android.cc | 80 +++ runtime/bin/sync_socket_fuchsia.cc | 80 +++ runtime/bin/sync_socket_linux.cc | 80 +++ runtime/bin/sync_socket_macos.cc | 77 +++ runtime/bin/sync_socket_patch.dart | 347 ++++++++++++ runtime/bin/sync_socket_win.cc | 78 +++ .../_internal/js_runtime/lib/io_patch.dart | 8 + sdk/lib/io/io.dart | 1 + sdk/lib/io/io_sources.gypi | 1 + sdk/lib/io/sync_socket.dart | 108 ++++ .../io/raw_synchronous_socket_test.dart | 493 ++++++++++++++++++ tests/standalone/standalone.status | 1 + 23 files changed, 1804 insertions(+), 45 deletions(-) create mode 100644 runtime/bin/sync_socket.cc create mode 100644 runtime/bin/sync_socket.h create mode 100644 runtime/bin/sync_socket_android.cc create mode 100644 runtime/bin/sync_socket_fuchsia.cc create mode 100644 runtime/bin/sync_socket_linux.cc create mode 100644 runtime/bin/sync_socket_macos.cc create mode 100644 runtime/bin/sync_socket_patch.dart create mode 100644 runtime/bin/sync_socket_win.cc create mode 100644 sdk/lib/io/sync_socket.dart create mode 100644 tests/standalone/io/raw_synchronous_socket_test.dart diff --git a/pkg/dev_compiler/tool/input_sdk/patch/io_patch.dart b/pkg/dev_compiler/tool/input_sdk/patch/io_patch.dart index e7ed5bef5c6..e3342f108f3 100644 --- a/pkg/dev_compiler/tool/input_sdk/patch/io_patch.dart +++ b/pkg/dev_compiler/tool/input_sdk/patch/io_patch.dart @@ -427,6 +427,14 @@ class SecureSocket { } } +@patch +class RawSynchronousSocket { + @patch + static RawSynchronousSocket connectSync(host, int port) { + throw new UnsupportedError("RawSynchronousSocket.connectSync"); + } +} + @patch class SecurityContext { @patch diff --git a/runtime/bin/io_impl_sources.gypi b/runtime/bin/io_impl_sources.gypi index a932f631973..e0e4ae0f259 100644 --- a/runtime/bin/io_impl_sources.gypi +++ b/runtime/bin/io_impl_sources.gypi @@ -90,5 +90,12 @@ 'stdio_macos.cc', 'stdio_unsupported.cc', 'stdio_win.cc', + 'sync_socket.cc', + 'sync_socket.h', + 'sync_socket_android.cc', + 'sync_socket_fuchsia.cc', + 'sync_socket_linux.cc', + 'sync_socket_macos.cc', + 'sync_socket_win.cc', ], } diff --git a/runtime/bin/io_natives.cc b/runtime/bin/io_natives.cc index 32279340cf1..b3a14e134f7 100644 --- a/runtime/bin/io_natives.cc +++ b/runtime/bin/io_natives.cc @@ -151,6 +151,17 @@ namespace bin { V(Stdout_GetTerminalSize, 1) \ V(Stdout_AnsiSupported, 1) \ V(StringToSystemEncoding, 1) \ + V(SynchronousSocket_Available, 1) \ + V(SynchronousSocket_CloseSync, 1) \ + V(SynchronousSocket_CreateConnectSync, 3) \ + V(SynchronousSocket_GetPort, 1) \ + V(SynchronousSocket_GetRemotePeer, 1) \ + V(SynchronousSocket_LookupRequest, 2) \ + V(SynchronousSocket_ShutdownRead, 1) \ + V(SynchronousSocket_ShutdownWrite, 1) \ + V(SynchronousSocket_Read, 2) \ + V(SynchronousSocket_ReadList, 4) \ + V(SynchronousSocket_WriteList, 4) \ V(SystemEncodingToString, 1) \ V(X509_Subject, 1) \ V(X509_Issuer, 1) \ diff --git a/runtime/bin/io_sources.gypi b/runtime/bin/io_sources.gypi index e5bdbe3cbbc..24bc72e7c35 100644 --- a/runtime/bin/io_sources.gypi +++ b/runtime/bin/io_sources.gypi @@ -17,5 +17,6 @@ 'socket_patch.dart', 'stdio_patch.dart', 'secure_socket_patch.dart', + 'sync_socket_patch.dart', ], } diff --git a/runtime/bin/socket_android.cc b/runtime/bin/socket_android.cc index 130c7f6828e..d4b54002b3d 100644 --- a/runtime/bin/socket_android.cc +++ b/runtime/bin/socket_android.cc @@ -7,18 +7,11 @@ #include "platform/globals.h" #if defined(HOST_OS_ANDROID) +#include "bin/socket.h" + #include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT #include "bin/fdutils.h" -#include "bin/file.h" -#include "bin/socket.h" -#include "bin/socket_base_android.h" #include "platform/signal_blocker.h" namespace dart { diff --git a/runtime/bin/socket_fuchsia.cc b/runtime/bin/socket_fuchsia.cc index 2936bc6992f..ba6f51f0c3b 100644 --- a/runtime/bin/socket_fuchsia.cc +++ b/runtime/bin/socket_fuchsia.cc @@ -7,22 +7,11 @@ #include "platform/globals.h" #if defined(HOST_OS_FUCHSIA) +#include "bin/socket.h" + #include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT #include "bin/fdutils.h" -#include "bin/file.h" -#include "bin/socket.h" -#include "bin/socket_base_fuchsia.h" #include "platform/signal_blocker.h" // #define SOCKET_LOG_INFO 1 diff --git a/runtime/bin/socket_linux.cc b/runtime/bin/socket_linux.cc index 92fa7697e44..52221d5c996 100644 --- a/runtime/bin/socket_linux.cc +++ b/runtime/bin/socket_linux.cc @@ -7,21 +7,11 @@ #include "platform/globals.h" #if defined(HOST_OS_LINUX) +#include "bin/socket.h" + #include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT #include "bin/fdutils.h" -#include "bin/file.h" -#include "bin/socket.h" -#include "bin/socket_base_linux.h" -#include "bin/thread.h" #include "platform/signal_blocker.h" namespace dart { diff --git a/runtime/bin/socket_macos.cc b/runtime/bin/socket_macos.cc index b0990c44a88..1f01ab315fc 100644 --- a/runtime/bin/socket_macos.cc +++ b/runtime/bin/socket_macos.cc @@ -7,20 +7,11 @@ #include "platform/globals.h" #if defined(HOST_OS_MACOS) +#include "bin/socket.h" + #include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT -#include // NOLINT #include "bin/fdutils.h" -#include "bin/file.h" -#include "bin/socket.h" -#include "bin/socket_base_macos.h" #include "platform/signal_blocker.h" namespace dart { diff --git a/runtime/bin/socket_unsupported.cc b/runtime/bin/socket_unsupported.cc index 1641bcfe5a8..bf9ff200673 100644 --- a/runtime/bin/socket_unsupported.cc +++ b/runtime/bin/socket_unsupported.cc @@ -11,6 +11,7 @@ namespace dart { namespace bin { + void FUNCTION_NAME(Socket_CreateConnect)(Dart_NativeArguments args) { Dart_ThrowException( DartUtils::NewDartArgumentError("Sockets unsupported on this platform")); diff --git a/runtime/bin/sync_socket.cc b/runtime/bin/sync_socket.cc new file mode 100644 index 00000000000..9dde966cfa2 --- /dev/null +++ b/runtime/bin/sync_socket.cc @@ -0,0 +1,365 @@ +// Copyright (c) 2017, 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. + +#if !defined(DART_IO_DISABLED) + +#include "bin/sync_socket.h" + +#include "bin/dartutils.h" +#include "bin/io_buffer.h" +#include "bin/isolate_data.h" +#include "bin/lockers.h" +#include "bin/thread.h" +#include "bin/utils.h" + +#include "include/dart_api.h" + +#include "platform/globals.h" +#include "platform/utils.h" + +#define DART_CHECK_ERROR_AND_CLEANUP(handle, ptr) \ + do { \ + if (Dart_IsError((handle))) { \ + delete (ptr); \ + Dart_SetReturnValue(args, (handle)); \ + return; \ + } \ + } while (0) + +#define DART_CHECK_ERROR(handle) \ + do { \ + if (Dart_IsError((handle))) { \ + Dart_SetReturnValue(args, (handle)); \ + return; \ + } \ + } while (0) + +namespace dart { +namespace bin { + +static const int kSocketIdNativeField = 0; + +void FUNCTION_NAME(SynchronousSocket_LookupRequest)(Dart_NativeArguments args) { + if (Dart_GetNativeArgumentCount(args) != 2) { + Dart_SetReturnValue( + args, DartUtils::NewDartArgumentError("Invalid argument count.")); + return; + } + + char* peer = NULL; + Dart_Handle host_arg = + Dart_GetNativeStringArgument(args, 0, reinterpret_cast(&peer)); + DART_CHECK_ERROR(host_arg); + + char* host = NULL; + host_arg = Dart_StringToCString(host_arg, const_cast(&host)); + DART_CHECK_ERROR(host_arg); + + int64_t type = 0; + Dart_Handle port_error = Dart_GetNativeIntegerArgument(args, 1, &type); + DART_CHECK_ERROR(port_error); + + OSError* os_error = NULL; + AddressList* addresses = + SocketBase::LookupAddress(host, type, &os_error); + if (addresses == NULL) { + Dart_SetReturnValue(args, DartUtils::NewDartOSError(os_error)); + return; + } + + Dart_Handle array = Dart_NewList(addresses->count()); + DART_CHECK_ERROR_AND_CLEANUP(array, addresses); + + for (intptr_t i = 0; i < addresses->count(); i++) { + SocketAddress* addr = addresses->GetAt(i); + Dart_Handle entry = Dart_NewList(3); + DART_CHECK_ERROR_AND_CLEANUP(entry, addresses); + + Dart_Handle type = Dart_NewInteger(addr->GetType()); + DART_CHECK_ERROR_AND_CLEANUP(type, addresses); + Dart_Handle error = Dart_ListSetAt(entry, 0, type); + DART_CHECK_ERROR_AND_CLEANUP(error, addresses); + + Dart_Handle as_string = Dart_NewStringFromCString(addr->as_string()); + DART_CHECK_ERROR_AND_CLEANUP(as_string, addresses); + error = Dart_ListSetAt(entry, 1, as_string); + DART_CHECK_ERROR_AND_CLEANUP(error, addresses); + + RawAddr raw = addr->addr(); + Dart_Handle data = SocketAddress::ToTypedData(raw); + DART_CHECK_ERROR_AND_CLEANUP(data, addresses); + + error = Dart_ListSetAt(entry, 2, data); + DART_CHECK_ERROR_AND_CLEANUP(error, addresses); + error = Dart_ListSetAt(array, i, entry); + DART_CHECK_ERROR_AND_CLEANUP(error, addresses); + } + delete addresses; + Dart_SetReturnValue(args, array); + return; +} + + +void FUNCTION_NAME(SynchronousSocket_CreateConnectSync)( + Dart_NativeArguments args) { + RawAddr addr; + SocketAddress::GetSockAddr(Dart_GetNativeArgument(args, 1), &addr); + Dart_Handle port_arg = Dart_GetNativeArgument(args, 2); + DART_CHECK_ERROR(port_arg); + int64_t port = DartUtils::GetInt64ValueCheckRange(port_arg, 0, 65535); + SocketAddress::SetAddrPort(&addr, static_cast(port)); + intptr_t socket = SynchronousSocket::CreateConnect(addr); + if (socket >= 0) { + Dart_Handle error = SynchronousSocket::SetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), new SynchronousSocket(socket)); + DART_CHECK_ERROR(error); + Dart_SetBooleanReturnValue(args, true); + } else { + Dart_SetReturnValue(args, DartUtils::NewDartOSError()); + } +} + + +void FUNCTION_NAME(SynchronousSocket_WriteList)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + Dart_Handle buffer_obj = Dart_GetNativeArgument(args, 1); + if (!Dart_IsList(buffer_obj)) { + Dart_SetReturnValue(args, DartUtils::NewDartArgumentError( + "First parameter must be a List")); + return; + } + intptr_t offset = DartUtils::GetIntptrValue(Dart_GetNativeArgument(args, 2)); + intptr_t length = DartUtils::GetIntptrValue(Dart_GetNativeArgument(args, 3)); + Dart_TypedData_Type type; + uint8_t* buffer = NULL; + intptr_t len; + result = Dart_TypedDataAcquireData(buffer_obj, &type, + reinterpret_cast(&buffer), &len); + DART_CHECK_ERROR(result); + ASSERT((offset + length) <= len); + buffer += offset; + intptr_t bytes_written = SocketBase::Write(socket->fd(), buffer, length); + if (bytes_written >= 0) { + Dart_SetIntegerReturnValue(args, bytes_written); + } else { + OSError os_error; + Dart_SetReturnValue(args, DartUtils::NewDartOSError(&os_error)); + } + Dart_TypedDataReleaseData(buffer_obj); +} + + +void FUNCTION_NAME(SynchronousSocket_ReadList)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + Dart_Handle buffer_obj = Dart_GetNativeArgument(args, 1); + if (!Dart_IsList(buffer_obj)) { + Dart_SetReturnValue(args, DartUtils::NewDartArgumentError( + "First parameter must be a List")); + return; + } + intptr_t offset = DartUtils::GetIntptrValue(Dart_GetNativeArgument(args, 2)); + intptr_t bytes = DartUtils::GetIntptrValue(Dart_GetNativeArgument(args, 3)); + intptr_t array_len = 0; + + result = Dart_ListLength(buffer_obj, &array_len); + DART_CHECK_ERROR(result); + + uint8_t* buffer = Dart_ScopeAllocate(bytes); + intptr_t bytes_read = SocketBase::Read(socket->fd(), buffer, bytes); + if (bytes_read < 0) { + Dart_SetReturnValue(args, DartUtils::NewDartOSError()); + return; + } + if (bytes_read > 0) { + result = Dart_ListSetAsBytes(buffer_obj, offset, buffer, bytes_read); + DART_CHECK_ERROR(result); + } + Dart_SetIntegerReturnValue(args, bytes_read); +} + + +void FUNCTION_NAME(SynchronousSocket_Available)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + intptr_t available = SocketBase::Available(socket->fd()); + if (available >= 0) { + Dart_SetIntegerReturnValue(args, available); + } else { + Dart_SetReturnValue(args, DartUtils::NewDartOSError()); + } +} + + +void FUNCTION_NAME(SynchronousSocket_CloseSync)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + SocketBase::Close(socket->fd()); + socket->SetClosedFd(); +} + + +void FUNCTION_NAME(SynchronousSocket_Read)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + int64_t length = 0; + if (!DartUtils::GetInt64Value(Dart_GetNativeArgument(args, 1), &length)) { + Dart_SetReturnValue(args, DartUtils::NewDartArgumentError( + "First parameter must be an integer.")); + return; + } + uint8_t* buffer = NULL; + result = IOBuffer::Allocate(length, &buffer); + ASSERT(buffer != NULL); + intptr_t bytes_read = SocketBase::Read(socket->fd(), buffer, length); + if (bytes_read == length) { + Dart_SetReturnValue(args, result); + } else if (bytes_read > 0) { + uint8_t* new_buffer = NULL; + Dart_Handle new_result = IOBuffer::Allocate(bytes_read, &new_buffer); + ASSERT(new_buffer != NULL); + memmove(new_buffer, buffer, bytes_read); + Dart_SetReturnValue(args, new_result); + } else if (bytes_read == -1) { + Dart_SetReturnValue(args, DartUtils::NewDartOSError()); + } +} + + +void FUNCTION_NAME(SynchronousSocket_ShutdownRead)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + SynchronousSocket::ShutdownRead(socket->fd()); +} + + +void FUNCTION_NAME(SynchronousSocket_ShutdownWrite)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + SynchronousSocket::ShutdownWrite(socket->fd()); +} + + +void FUNCTION_NAME(SynchronousSocket_GetPort)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + intptr_t port = SocketBase::GetPort(socket->fd()); + if (port > 0) { + Dart_SetReturnValue(args, Dart_NewInteger(port)); + } else { + Dart_SetReturnValue(args, DartUtils::NewDartOSError()); + } +} + + +void FUNCTION_NAME(SynchronousSocket_GetRemotePeer)(Dart_NativeArguments args) { + SynchronousSocket* socket = NULL; + Dart_Handle result = SynchronousSocket::GetSocketIdNativeField( + Dart_GetNativeArgument(args, 0), &socket); + DART_CHECK_ERROR(result); + + intptr_t port = 0; + SocketAddress* addr = SocketBase::GetRemotePeer(socket->fd(), &port); + if (addr == NULL) { + Dart_SetReturnValue(args, DartUtils::NewDartOSError()); + return; + } + Dart_Handle list = Dart_NewList(2); + DART_CHECK_ERROR_AND_CLEANUP(list, addr); + + Dart_Handle entry = Dart_NewList(3); + DART_CHECK_ERROR_AND_CLEANUP(entry, addr); + + Dart_Handle error = + Dart_ListSetAt(entry, 0, Dart_NewInteger(addr->GetType())); + DART_CHECK_ERROR_AND_CLEANUP(error, addr); + error = + Dart_ListSetAt(entry, 1, Dart_NewStringFromCString(addr->as_string())); + DART_CHECK_ERROR_AND_CLEANUP(error, addr); + + RawAddr raw = addr->addr(); + error = Dart_ListSetAt(entry, 2, SocketAddress::ToTypedData(raw)); + DART_CHECK_ERROR_AND_CLEANUP(error, addr); + + error = Dart_ListSetAt(list, 0, entry); + DART_CHECK_ERROR_AND_CLEANUP(error, addr); + error = Dart_ListSetAt(list, 1, Dart_NewInteger(port)); + DART_CHECK_ERROR_AND_CLEANUP(error, addr); + Dart_SetReturnValue(args, list); + delete addr; +} + + +static void SynchronousSocketFinalizer(void* isolate_data, + Dart_WeakPersistentHandle handle, + void* data) { + SynchronousSocket* socket = reinterpret_cast(data); + if (socket->fd() >= 0) { + SocketBase::Close(socket->fd()); + socket->SetClosedFd(); + } + delete socket; +} + + +Dart_Handle SynchronousSocket::SetSocketIdNativeField( + Dart_Handle handle, + SynchronousSocket* socket) { + Dart_Handle error = Dart_SetNativeInstanceField( + handle, kSocketIdNativeField, reinterpret_cast(socket)); + if (Dart_IsError(error)) { + delete socket; + return error; + } + + Dart_NewWeakPersistentHandle(handle, reinterpret_cast(socket), + sizeof(SynchronousSocket), + SynchronousSocketFinalizer); + return error; +} + + +Dart_Handle SynchronousSocket::GetSocketIdNativeField( + Dart_Handle socket_obj, + SynchronousSocket** socket) { + ASSERT(socket != NULL); + intptr_t id; + Dart_Handle result = + Dart_GetNativeInstanceField(socket_obj, kSocketIdNativeField, &id); + if (Dart_IsError(result)) { + return result; + } + *socket = reinterpret_cast(id); + return result; +} + +} // namespace bin +} // namespace dart + +#endif // !defined(DART_IO_DISABLED) diff --git a/runtime/bin/sync_socket.h b/runtime/bin/sync_socket.h new file mode 100644 index 00000000000..b6eb784bc0b --- /dev/null +++ b/runtime/bin/sync_socket.h @@ -0,0 +1,49 @@ +// Copyright (c) 2017, 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. + +#ifndef RUNTIME_BIN_SYNC_SOCKET_H_ +#define RUNTIME_BIN_SYNC_SOCKET_H_ + +#if defined(DART_IO_DISABLED) +#error "sync_socket.h can only be included on builds with IO enabled" +#endif + +#include "bin/socket_base.h" +#include "platform/globals.h" + +namespace dart { +namespace bin { + +class SynchronousSocket { + public: + explicit SynchronousSocket(intptr_t fd); + ~SynchronousSocket() { ASSERT(fd_ == kClosedFd); } + + intptr_t fd() const { return fd_; } + void SetClosedFd(); + + static bool Initialize(); + + static intptr_t CreateConnect(const RawAddr& addr); + + static Dart_Handle SetSocketIdNativeField(Dart_Handle handle, + SynchronousSocket* socket); + static Dart_Handle GetSocketIdNativeField(Dart_Handle socket_obj, + SynchronousSocket** socket); + + static void ShutdownRead(intptr_t fd); + static void ShutdownWrite(intptr_t fd); + + private: + static const int kClosedFd = -1; + + intptr_t fd_; + + DISALLOW_COPY_AND_ASSIGN(SynchronousSocket); +}; + +} // namespace bin +} // namespace dart + +#endif // RUNTIME_BIN_SYNC_SOCKET_H_ diff --git a/runtime/bin/sync_socket_android.cc b/runtime/bin/sync_socket_android.cc new file mode 100644 index 00000000000..24911887ec5 --- /dev/null +++ b/runtime/bin/sync_socket_android.cc @@ -0,0 +1,80 @@ +// Copyright (c) 2017, 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. + +#if !defined(DART_IO_DISABLED) + +#include "platform/globals.h" +#if defined(HOST_OS_ANDROID) + +#include "bin/sync_socket.h" + +#include // NOLINT + +#include "bin/fdutils.h" +#include "platform/signal_blocker.h" + +namespace dart { +namespace bin { + +SynchronousSocket::SynchronousSocket(intptr_t fd) : fd_(fd) {} + + +void SynchronousSocket::SetClosedFd() { + fd_ = kClosedFd; +} + + +bool SynchronousSocket::Initialize() { + // Nothing to do on Android. + return true; +} + + +static intptr_t Create(const RawAddr& addr) { + intptr_t fd; + intptr_t type = SOCK_STREAM; + fd = NO_RETRY_EXPECTED(socket(addr.ss.ss_family, type, 0)); + if (fd < 0) { + return -1; + } + return fd; +} + + +static intptr_t Connect(intptr_t fd, const RawAddr& addr) { + intptr_t result = TEMP_FAILURE_RETRY( + connect(fd, &addr.addr, SocketAddress::GetAddrLength(addr))); + if (result == 0) { + return fd; + } + ASSERT(errno != EINPROGRESS); + FDUtils::FDUtils::SaveErrorAndClose(fd); + return -1; +} + + +intptr_t SynchronousSocket::CreateConnect(const RawAddr& addr) { + intptr_t fd = Create(addr); + if (fd < 0) { + return fd; + } + return Connect(fd, addr); +} + + +void SynchronousSocket::ShutdownRead(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_RD)); +} + + +void SynchronousSocket::ShutdownWrite(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_WR)); +} + +} // namespace bin +} // namespace dart + +#endif // defined(HOST_OS_ANDROID) + +#endif // !defined(DART_IO_DISABLED) diff --git a/runtime/bin/sync_socket_fuchsia.cc b/runtime/bin/sync_socket_fuchsia.cc new file mode 100644 index 00000000000..ec55e2c7908 --- /dev/null +++ b/runtime/bin/sync_socket_fuchsia.cc @@ -0,0 +1,80 @@ +// Copyright (c) 2017, 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. + +#if !defined(DART_IO_DISABLED) + +#include "platform/globals.h" +#if defined(HOST_OS_FUCHSIA) + +#include "bin/sync_socket.h" + +#include // NOLINT + +#include "bin/fdutils.h" +#include "platform/signal_blocker.h" + +namespace dart { +namespace bin { + +SynchronousSocket::SynchronousSocket(intptr_t fd) : fd_(fd) {} + + +void SynchronousSocket::SetClosedFd() { + fd_ = kClosedFd; +} + + +bool SynchronousSocket::Initialize() { + // Nothing to do on Fuchsia. + return true; +} + + +static intptr_t Create(const RawAddr& addr) { + intptr_t fd; + intptr_t type = SOCK_STREAM | SOCK_CLOEXEC; + fd = NO_RETRY_EXPECTED(socket(addr.ss.ss_family, type, 0)); + if (fd < 0) { + return -1; + } + return fd; +} + + +static intptr_t Connect(intptr_t fd, const RawAddr& addr) { + intptr_t result = TEMP_FAILURE_RETRY( + connect(fd, &addr.addr, SocketAddress::GetAddrLength(addr))); + if (result == 0) { + return fd; + } + ASSERT(errno != EINPROGRESS); + FDUtils::FDUtils::SaveErrorAndClose(fd); + return -1; +} + + +intptr_t SynchronousSocket::CreateConnect(const RawAddr& addr) { + intptr_t fd = Create(addr); + if (fd < 0) { + return fd; + } + return Connect(fd, addr); +} + + +void SynchronousSocket::ShutdownRead(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_RD)); +} + + +void SynchronousSocket::ShutdownWrite(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_WR)); +} + +} // namespace bin +} // namespace dart + +#endif // defined(HOST_OS_FUCHSIA) + +#endif // !defined(DART_IO_DISABLED) diff --git a/runtime/bin/sync_socket_linux.cc b/runtime/bin/sync_socket_linux.cc new file mode 100644 index 00000000000..122661593f6 --- /dev/null +++ b/runtime/bin/sync_socket_linux.cc @@ -0,0 +1,80 @@ +// Copyright (c) 2017, 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. + +#if !defined(DART_IO_DISABLED) + +#include "platform/globals.h" +#if defined(HOST_OS_LINUX) + +#include "bin/sync_socket.h" + +#include // NOLINT + +#include "bin/fdutils.h" +#include "platform/signal_blocker.h" + +namespace dart { +namespace bin { + +SynchronousSocket::SynchronousSocket(intptr_t fd) : fd_(fd) {} + + +void SynchronousSocket::SetClosedFd() { + fd_ = kClosedFd; +} + + +bool SynchronousSocket::Initialize() { + // Nothing to do on Linux. + return true; +} + + +static intptr_t Create(const RawAddr& addr) { + intptr_t fd; + intptr_t type = SOCK_STREAM | SOCK_CLOEXEC; + fd = NO_RETRY_EXPECTED(socket(addr.ss.ss_family, type, 0)); + if (fd < 0) { + return -1; + } + return fd; +} + + +static intptr_t Connect(intptr_t fd, const RawAddr& addr) { + intptr_t result = TEMP_FAILURE_RETRY( + connect(fd, &addr.addr, SocketAddress::GetAddrLength(addr))); + if (result == 0) { + return fd; + } + ASSERT(errno != EINPROGRESS); + FDUtils::FDUtils::SaveErrorAndClose(fd); + return -1; +} + + +intptr_t SynchronousSocket::CreateConnect(const RawAddr& addr) { + intptr_t fd = Create(addr); + if (fd < 0) { + return fd; + } + return Connect(fd, addr); +} + + +void SynchronousSocket::ShutdownRead(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_RD)); +} + + +void SynchronousSocket::ShutdownWrite(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_WR)); +} + +} // namespace bin +} // namespace dart + +#endif // defined(HOST_OS_LINUX) + +#endif // !defined(DART_IO_DISABLED) diff --git a/runtime/bin/sync_socket_macos.cc b/runtime/bin/sync_socket_macos.cc new file mode 100644 index 00000000000..9b7e3bf89f7 --- /dev/null +++ b/runtime/bin/sync_socket_macos.cc @@ -0,0 +1,77 @@ +// Copyright (c) 2017, 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. + +#if !defined(DART_IO_DISABLED) + +#include "platform/globals.h" +#if defined(HOST_OS_MACOS) + +#include "bin/sync_socket.h" + +#include // NOLINT + +#include "bin/fdutils.h" +#include "platform/signal_blocker.h" + +namespace dart { +namespace bin { + +SynchronousSocket::SynchronousSocket(intptr_t fd) : fd_(fd) {} + + +void SynchronousSocket::SetClosedFd() { + fd_ = kClosedFd; +} + + +static intptr_t Create(const RawAddr& addr) { + intptr_t fd; + fd = NO_RETRY_EXPECTED(socket(addr.ss.ss_family, SOCK_STREAM, 0)); + if (fd < 0) { + return -1; + } + if (!FDUtils::SetCloseOnExec(fd)) { + FDUtils::SaveErrorAndClose(fd); + return -1; + } + return fd; +} + + +static intptr_t Connect(intptr_t fd, const RawAddr& addr) { + intptr_t result = TEMP_FAILURE_RETRY( + connect(fd, &addr.addr, SocketAddress::GetAddrLength(addr))); + if (result == 0) { + return fd; + } + ASSERT(errno != EINPROGRESS); + FDUtils::FDUtils::SaveErrorAndClose(fd); + return -1; +} + + +intptr_t SynchronousSocket::CreateConnect(const RawAddr& addr) { + intptr_t fd = Create(addr); + if (fd < 0) { + return fd; + } + return Connect(fd, addr); +} + + +void SynchronousSocket::ShutdownRead(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_RD)); +} + + +void SynchronousSocket::ShutdownWrite(intptr_t fd) { + VOID_NO_RETRY_EXPECTED(shutdown(fd, SHUT_WR)); +} + +} // namespace bin +} // namespace dart + +#endif // defined(HOST_OS_MACOS) + +#endif // !defined(DART_IO_DISABLED) diff --git a/runtime/bin/sync_socket_patch.dart b/runtime/bin/sync_socket_patch.dart new file mode 100644 index 00000000000..4bd735a5225 --- /dev/null +++ b/runtime/bin/sync_socket_patch.dart @@ -0,0 +1,347 @@ +// Copyright (c) 2017, 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. + +@patch +class RawSynchronousSocket { + @patch + static RawSynchronousSocket connectSync(host, int port) { + return _RawSynchronousSocket.connectSync(host, port); + } +} + +class _RawSynchronousSocket implements RawSynchronousSocket { + final _NativeSynchronousSocket _socket; + + _RawSynchronousSocket(this._socket); + + static RawSynchronousSocket connectSync(host, int port) { + _throwOnBadPort(port); + return new _RawSynchronousSocket( + _NativeSynchronousSocket.connectSync(host, port)); + } + + InternetAddress get address => _socket.address; + int get port => _socket.port; + InternetAddress get remoteAddress => _socket.remoteAddress; + int get remotePort => _socket.remotePort; + + int available() => _socket.available; + + void closeSync() => _socket.closeSync(); + + int readIntoSync(List buffer, [int start = 0, int end]) => + _socket.readIntoSync(buffer, start, end); + + List readSync(int bytes) => _socket.readSync(bytes); + + void shutdown(SocketDirection direction) => _socket.shutdown(direction); + + void writeFromSync(List buffer, [int start = 0, int end]) => + _socket.writeFromSync(buffer, start, end); +} + +// The NativeFieldWrapperClass1 can not be used with a mixin, due to missing +// implicit constructor. +class _NativeSynchronousSocketNativeWrapper extends NativeFieldWrapperClass1 {} + +// The _NativeSynchronousSocket class encapsulates a synchronous OS socket. +class _NativeSynchronousSocket extends _NativeSynchronousSocketNativeWrapper { + // Socket close state. + bool isClosed = false; + bool isClosedRead = false; + bool isClosedWrite = false; + + // Holds the address used to connect the socket. + InternetAddress localAddress; + + // Holds the port of the socket, 0 if not known. + int localPort = 0; + + _ReadWriteResourceInfo resourceInfo; + + static _NativeSynchronousSocket connectSync(host, int port) { + if (host == null) { + throw ArgumentError("Parameter host cannot be null"); + } + List<_InternetAddress> addresses = null; + var error = null; + if (host is _InternetAddress) { + addresses = [host]; + } else { + try { + addresses = lookup(host); + } catch (e) { + error = e; + } + if (error != null || addresses == null || addresses.isEmpty) { + throw createError(error, "Failed host lookup: '$host'"); + } + } + assert(addresses is List); + var it = addresses.iterator; + _NativeSynchronousSocket connectNext() { + if (!it.moveNext()) { + // Could not connect. Throw the first connection error we encountered. + assert(error != null); + throw error; + } + var address = it.current; + var socket = new _NativeSynchronousSocket(); + socket.localAddress = address; + var result = socket.nativeCreateConnectSync(address._in_addr, port); + if (result is OSError) { + // Keep first error, if present. + if (error == null) { + error = createError(result, "Connection failed", address, port); + } + return connectNext(); + } else { + // Query the local port, for error messages. + try { + socket.port; + } catch (e) { + if (error == null) { + error = createError(e, "Connection failed", address, port); + } + return connectNext(); + } + setupResourceInfo(socket); + } + return socket; + } + + return connectNext(); + } + + InternetAddress get address => localAddress; + int get available => nativeAvailable(); + + int get port { + if (localPort != 0) { + return localPort; + } + if (isClosed) { + throw const SocketException.closed(); + } + var result = nativeGetPort(); + if (result is OSError) { + throw result; + } + return localPort = result; + } + + InternetAddress get remoteAddress { + if (isClosed) { + throw const SocketException.closed(); + } + var result = nativeGetRemotePeer(); + if (result is OSError) { + throw result; + } + var addr = result[0]; + return new _InternetAddress(addr[1], null, addr[2]); + } + + int get remotePort { + if (isClosed) { + throw const SocketException.closed(); + } + var result = nativeGetRemotePeer(); + if (result is OSError) { + throw result; + } + return result[1]; + } + + void closeSync() { + if (!isClosed) { + nativeCloseSync(); + _SocketResourceInfo.SocketClosed(resourceInfo); + isClosed = true; + } + } + + // Create the appropriate error/exception from different returned + // error objects. + static createError(error, String message, + [InternetAddress address, int port]) { + if (error is OSError) { + return new SocketException(message, + osError: error, address: address, port: port); + } else { + return new SocketException(message, address: address, port: port); + } + } + + static List<_InternetAddress> lookup(String host, + {InternetAddressType type: InternetAddressType.ANY}) { + var response = _nativeLookupRequest(host, type._value); + if (response is OSError) { + throw response; + } + List<_InternetAddress> addresses = + new List<_InternetAddress>(response.length); + for (int i = 0; i < response.length; ++i) { + var result = response[i]; + addresses[i] = new _InternetAddress(result[1], host, result[2]); + } + return addresses; + } + + int readIntoSync(List buffer, int start, int end) { + _checkAvailable(); + if (isClosedRead) { + throw new SocketException("Socket is closed for reading"); + } + + if ((buffer is! List) || + ((start != null) && (start is! int)) || + ((end != null) && (end is! int))) { + throw new ArgumentError("Invalid arguments to readIntoSync"); + } + if (start == null) { + throw new ArgumentError("start cannot be null"); + } + end = RangeError.checkValidRange(start, end, buffer.length); + if (end == start) { + return 0; + } + var result = nativeReadInto(buffer, start, (end - start)); + if (result is OSError) { + throw new SocketException("readIntoSync failed", osError: result); + } + resourceInfo.addRead(result); + return result; + } + + List readSync(int len) { + _checkAvailable(); + if (isClosedRead) { + throw new SocketException("Socket is closed for reading"); + } + + if ((len != null) && (len < 0)) { + throw new ArgumentError("Illegal length $len"); + } + if (len == 0) { + return null; + } + var result = nativeRead(len); + if (result is OSError) { + throw result; + } + assert(resourceInfo != null); + if (result != null) { + if (resourceInfo != null) { + resourceInfo.totalRead += result.length; + } + } + if (resourceInfo != null) { + resourceInfo.didRead(); + } + return result; + } + + static void setupResourceInfo(_NativeSynchronousSocket socket) { + socket.resourceInfo = new _SocketResourceInfo(socket); + } + + void shutdown(SocketDirection direction) { + if (isClosed) { + return; + } + switch (direction) { + case SocketDirection.RECEIVE: + shutdownRead(); + break; + case SocketDirection.SEND: + shutdownWrite(); + break; + case SocketDirection.BOTH: + closeSync(); + break; + default: + throw new ArgumentError(direction); + } + } + + void shutdownRead() { + if (isClosed || isClosedRead) { + return; + } + if (isClosedWrite) { + closeSync(); + } else { + nativeShutdownRead(); + } + isClosedRead = true; + } + + void shutdownWrite() { + if (isClosed || isClosedWrite) { + return; + } + if (isClosedRead) { + closeSync(); + } else { + nativeShutdownWrite(); + } + isClosedWrite = true; + } + + void writeFromSync(List buffer, int start, int end) { + _checkAvailable(); + if (isClosedWrite) { + throw new SocketException("Socket is closed for writing"); + } + if ((buffer is! List) || + ((start != null) && (start is! int)) || + ((end != null) && (end is! int))) { + throw new ArgumentError("Invalid arguments to writeFromSync"); + } + if (start == null) { + throw new ArgumentError("start cannot be equal to null"); + } + + end = RangeError.checkValidRange(start, end, buffer.length); + if (end == start) { + return; + } + + _BufferAndStart bufferAndStart = + _ensureFastAndSerializableByteData(buffer, start, end); + var result = nativeWrite(bufferAndStart.buffer, bufferAndStart.start, + end - (start - bufferAndStart.start)); + if (result is OSError) { + throw new SocketException("writeFromSync failed", osError: result); + } + assert(resourceInfo != null); + if (resourceInfo != null) { + resourceInfo.addWrite(result); + } + } + + void _checkAvailable() { + if (isClosed) { + throw const SocketException.closed(); + } + } + + // Native method declarations. + static _nativeLookupRequest(host, int type) + native "SynchronousSocket_LookupRequest"; + nativeCreateConnectSync(host, int port) + native "SynchronousSocket_CreateConnectSync"; + nativeAvailable() native "SynchronousSocket_Available"; + nativeCloseSync() native "SynchronousSocket_CloseSync"; + int nativeGetPort() native "SynchronousSocket_GetPort"; + List nativeGetRemotePeer() native "SynchronousSocket_GetRemotePeer"; + nativeRead(int len) native "SynchronousSocket_Read"; + nativeReadInto(List buffer, int offset, int bytes) + native "SynchronousSocket_ReadList"; + nativeShutdownRead() native "SynchronousSocket_ShutdownRead"; + nativeShutdownWrite() native "SynchronousSocket_ShutdownWrite"; + nativeWrite(List buffer, int offset, int bytes) + native "SynchronousSocket_WriteList"; +} diff --git a/runtime/bin/sync_socket_win.cc b/runtime/bin/sync_socket_win.cc new file mode 100644 index 00000000000..3d5338b5f6d --- /dev/null +++ b/runtime/bin/sync_socket_win.cc @@ -0,0 +1,78 @@ +// Copyright (c) 2017, 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. + +#if !defined(DART_IO_DISABLED) + +#include "platform/globals.h" +#if defined(HOST_OS_WINDOWS) + +#include "bin/sync_socket.h" + +#include "bin/builtin.h" +#include "bin/log.h" +#include "bin/utils.h" +#include "bin/utils_win.h" + +// #define SOCKET_LOG_ERROR 1 + +// define SOCKET_LOG_ERROR to get log messages only for errors. +#if defined(SOCKET_LOG_ERROR) +#define LOG_ERR(msg, ...) \ + { \ + int err = errno; \ + Log::PrintErr("Dart Socket ERROR: %s:%d: " msg, __FILE__, __LINE__, \ + ##__VA_ARGS__); \ + errno = err; \ + } +#else +#define LOG_ERR(msg, ...) +#endif // defined(SOCKET_LOG_ERROR) + +namespace dart { +namespace bin { + +SynchronousSocket::SynchronousSocket(intptr_t fd) { + LOG_ERR("SynchronousSocket is unimplemented\n"); + UNIMPLEMENTED(); +} + + +bool SynchronousSocket::Initialize() { + LOG_ERR("SynchronousSocket::Initialize is unimplemented\n"); + UNIMPLEMENTED(); + return false; +} + + +void SynchronousSocket::SetClosedFd() { + LOG_ERR("SynchronousSocket::SetClosedFd is unimplemented\n"); + UNIMPLEMENTED(); +} + + +intptr_t SynchronousSocket::CreateConnect(const RawAddr& addr) { + LOG_ERR("SynchronousSocket::CreateConnect is unimplemented\n"); + UNIMPLEMENTED(); + return -1; +} + + +void SynchronousSocket::ShutdownRead(intptr_t fd) { + LOG_ERR("SynchronousSocket::ShutdownRead is unimplemented\n"); + UNIMPLEMENTED(); +} + + +void SynchronousSocket::ShutdownWrite(intptr_t fd) { + LOG_ERR("SynchronousSocket::ShutdownWrite is unimplemented\n"); + UNIMPLEMENTED(); +} + + +} // namespace bin +} // namespace dart + +#endif // defined(HOST_OS_WINDOWS) + +#endif // !defined(DART_IO_DISABLED) diff --git a/sdk/lib/_internal/js_runtime/lib/io_patch.dart b/sdk/lib/_internal/js_runtime/lib/io_patch.dart index e7ed5bef5c6..e3342f108f3 100644 --- a/sdk/lib/_internal/js_runtime/lib/io_patch.dart +++ b/sdk/lib/_internal/js_runtime/lib/io_patch.dart @@ -427,6 +427,14 @@ class SecureSocket { } } +@patch +class RawSynchronousSocket { + @patch + static RawSynchronousSocket connectSync(host, int port) { + throw new UnsupportedError("RawSynchronousSocket.connectSync"); + } +} + @patch class SecurityContext { @patch diff --git a/sdk/lib/io/io.dart b/sdk/lib/io/io.dart index 333b79cc895..9f98149cb19 100644 --- a/sdk/lib/io/io.dart +++ b/sdk/lib/io/io.dart @@ -242,5 +242,6 @@ part 'service_object.dart'; part 'socket.dart'; part 'stdio.dart'; part 'string_transformer.dart'; +part 'sync_socket.dart'; part 'websocket.dart'; part 'websocket_impl.dart'; diff --git a/sdk/lib/io/io_sources.gypi b/sdk/lib/io/io_sources.gypi index 10a6bf48024..c49b9237075 100644 --- a/sdk/lib/io/io_sources.gypi +++ b/sdk/lib/io/io_sources.gypi @@ -36,6 +36,7 @@ 'socket.dart', 'stdio.dart', 'string_transformer.dart', + 'sync_socket.dart', 'websocket.dart', 'websocket_impl.dart', ], diff --git a/sdk/lib/io/sync_socket.dart b/sdk/lib/io/sync_socket.dart new file mode 100644 index 00000000000..cc055a7f9d6 --- /dev/null +++ b/sdk/lib/io/sync_socket.dart @@ -0,0 +1,108 @@ +// Copyright (c) 2017, 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. + +part of dart.io; + +/** + * A low-level class for communicating synchronously over a TCP socket. + * + * Warning: [RawSynchronousSocket] should probably only be used to connect to + * 'localhost'. The operations below will block the calling thread to wait for + * a response from the network. The thread can process no other events while + * waiting for these operations to complete. [RawSynchronousSocket] is not + * suitable for applications that require high performance or asynchronous I/O + * such as a server. Instead such applications should use the non-blocking + * sockets and asynchronous operations in the Socket or RawSocket classes. + */ +abstract class RawSynchronousSocket { + /** + * Creates a new socket connection and returns a [RawSynchronousSocket]. + * + * [host] can either be a [String] or an [InternetAddress]. If [host] is a + * [String], [connectSync] will perform a [InternetAddress.lookup] and try + * all returned [InternetAddress]es, until connected. Unless a + * connection was established, the error from the first failing connection is + * returned. + */ + external static RawSynchronousSocket connectSync(host, int port); + + /** + * Returns the number of received and unread bytes in the socket that can be + * read. + */ + int available(); + + /** + * Closes the [RawSynchronousSocket]. + * + * Once [closeSync] has been called, attempting to call [readSync], + * [readIntoSync], [writeFromSync], [remoteAddress], and [remotePort] will + * cause a [SocketException] to be thrown. + */ + void closeSync(); + + /** + * Reads into an existing [List] from the socket into the range: + * [[start],[end]). + * + * Reads into an existing [List] from the socket. If [start] is present, + * the bytes will be filled into [buffer] from index [start], otherwise index + * 0. If [end] is present, [end] - [start] bytes will be read into [buffer], + * otherwise up to [buffer.length]. If [end] == [start], no bytes are read. + * Returns the number of bytes read. + */ + int readIntoSync(List buffer, [int start = 0, int end]); + + /** + * Reads up to [bytes] bytes from the socket. + * + * Blocks and waits for a response of up to a specified number of bytes + * sent by the socket. [bytes] specifies the maximum number of bytes to + * be read. Returns the list of bytes read, which could be less than the + * value specified by [bytes]. + */ + List readSync(int bytes); + + /** + * Shutdown a socket in the provided direction. + * + * Calling shutdown will never throw an exception and calling it several times + * is supported. If both [RECEIVE] and [SEND] directions are closed, the + * socket is closed completely, the same as if [closeSync] has been called. + */ + void shutdown(SocketDirection direction); + + /** + * Writes data from a specified range in a [List] to the socket. + * + * Writes into the socket from a [List]. If [start] is present, the bytes + * will be written to the socket starting from index [start]. If [start] is + * not present, the bytes will be written starting from index 0. If [end] is + * present, the [end] - [start] bytes will be written into the socket starting + * at index [start]. If [end] is not provided, [buffer.length] elements will + * be written to the socket starting from index [start]. If [end] == [start], + * nothing happens. + */ + void writeFromSync(List buffer, [int start = 0, int end]); + + /** + * The port used by this socket. + */ + int get port; + + /** + * The remote port connected to by this socket. + */ + int get remotePort; + + /** + * The [InternetAddress] used to connect this socket. + */ + InternetAddress get address; + + /** + * The remote [InternetAddress] connected to by this socket. + */ + InternetAddress get remoteAddress; +} diff --git a/tests/standalone/io/raw_synchronous_socket_test.dart b/tests/standalone/io/raw_synchronous_socket_test.dart new file mode 100644 index 00000000000..28236b98718 --- /dev/null +++ b/tests/standalone/io/raw_synchronous_socket_test.dart @@ -0,0 +1,493 @@ +// Copyright (c) 2017, 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:isolate"; +import "dart:math"; + +import "package:async_helper/async_helper.dart"; +import "package:expect/expect.dart"; + +const String LOOPBACK_IP_V4_STRING = "127.0.0.1"; + +void testArguments() { + Expect.throws(() => RawSynchronousSocket.connectSync(null, 0)); + Expect.throws( + () => RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, null)); + Expect.throws( + () => RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, 65536)); + Expect.throws( + () => RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, -1)); + Expect.throws(() => + RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, 0, backlog: -1)); +} + +// The connection attempt happens on the main Dart thread and the OS timeout can +// be arbitrarily long, causing timeout issues on the build bots. This isn't an +// issue with the async sockets since the lookup for a connect call happens on +// the IO service thread. +/* +void testInvalidConnect() { + // Connect to an unknown DNS name. + try { + var socket = RawSynchronousSocket.connectSync("ko.faar.__hest__", 0); + Expect.fail("Failure expected"); + } catch (e) { + Expect.isTrue(e is SocketException); + } + + // Connect to an unavaliable IP-address. + try { + var socket = RawSynchronousSocket.connectSync("1.2.3.4", 0); + Expect.fail("Failure expected"); + } catch (e) { + Expect.isTrue(e is SocketException); + } +} +*/ + +void testSimpleConnect() { + asyncStart(); + RawServerSocket.bind(InternetAddress.LOOPBACK_IP_V4, 0).then((server) { + var socket = + RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, server.port); + server.listen((serverSocket) { + Expect.equals(socket.address, serverSocket.remoteAddress); + Expect.equals(socket.port, serverSocket.remotePort); + Expect.equals(socket.remoteAddress, server.address); + Expect.equals(socket.remotePort, server.port); + socket.closeSync(); + server.close(); + asyncEnd(); + }); + }); +} + +void testServerListenAfterConnect() { + asyncStart(); + RawServerSocket.bind(InternetAddress.LOOPBACK_IP_V4, 0).then((server) { + Expect.isTrue(server.port > 0); + var client = + RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, server.port); + server.listen((socket) { + client.closeSync(); + server.close(); + socket.close(); + asyncEnd(); + }); + }); +} + +const messageSize = 1000; +// Configuration fields for the EchoServer. +enum EchoServerTypes { + // Max accumulated connections to server before close. Defaults to 1. + CONNECTION_COUNT, + // Sets the range of the fields to check in the list generated by + // createTestData(). + OFFSET_END, + OFFSET_START, + // The port used to communicate with an isolate. + ISOLATE_SEND_PORT, + // The port of the newly created echo server. + SERVER_PORT +} + +List createTestData() { + return new List.generate(messageSize, (index) => index & 0xff); +} + +// Consumes data generated by a test and compares it against the original test +// data. The optional fields, start and end, are used to compare against +// segments of the original test data list. In other words, data.length == (end +// - start). +void verifyTestData(List data, [int start = 0, int end]) { + assert(data != null); + List expected = createTestData(); + if (end == null) { + end = data.length; + } + end = min(messageSize, end); + Expect.equals(end - start, data.length); + for (int i = 0; i < (end - start); i++) { + Expect.equals(expected[start + i], data[i]); + } +} + +// The echo server is spawned in a new isolate and is used to test various +// synchronous read/write operations by echoing any data received back to the +// sender. The server should shutdown automatically after a specified number of +// socket disconnections (default: 1). +Future echoServer(var sendPort) async { + RawServerSocket.bind(InternetAddress.LOOPBACK_IP_V4, 0).then((server) async { + ReceivePort receivePort = new ReceivePort(); + Map response = { + EchoServerTypes.ISOLATE_SEND_PORT: receivePort.sendPort, + EchoServerTypes.SERVER_PORT: server.port + }; + sendPort.send(response); + Map limits = await receivePort.first; + int start = limits[EchoServerTypes.OFFSET_START]; + int end = limits[EchoServerTypes.OFFSET_END]; + int length = end - start; + int connection_count = limits[EchoServerTypes.CONNECTION_COUNT] ?? 1; + int connections = 0; + sendPort = limits[EchoServerTypes.ISOLATE_SEND_PORT]; + server.listen((client) { + int bytesRead = 0; + int bytesWritten = 0; + bool closedEventReceived = false; + List data = new List(length); + client.writeEventsEnabled = false; + client.listen((event) { + switch (event) { + case RawSocketEvent.READ: + Expect.isTrue(bytesWritten == 0); + Expect.isTrue(client.available() > 0); + var buffer = client.read(client.available()); + data.setRange(bytesRead, bytesRead + buffer.length, buffer); + bytesRead += buffer.length; + // Once we've read all the data, we can echo it back. Otherwise, + // keep waiting for more bytes. + if (bytesRead >= length) { + verifyTestData(data, start, end); + client.writeEventsEnabled = true; + } + break; + case RawSocketEvent.WRITE: + Expect.isFalse(client.writeEventsEnabled); + bytesWritten += + client.write(data, bytesWritten, data.length - bytesWritten); + if (bytesWritten < length) { + client.writeEventsEnabled = true; + } else if (bytesWritten == length) { + // Close the socket for writing from the server since we're done + // writing to this socket. The connection is closed completely + // after the client closes the socket for reading from the server. + client.shutdown(SocketDirection.SEND); + } + break; + case RawSocketEvent.READ_CLOSED: + client.close(); + break; + case RawSocketEvent.CLOSED: + Expect.isFalse(closedEventReceived); + closedEventReceived = true; + break; + default: + throw "Unexpected event $event"; + } + }, onDone: () { + Expect.isTrue(closedEventReceived); + connections++; + if (connections >= connection_count) { + server.close(); + } + }); + }, onDone: () { + // Let the client know we're shutting down then kill the isolate. + sendPort.send(null); + kill(); + }); + }); +} + +Future testSimpleReadWrite({bool dropReads}) async { + asyncStart(); + // This test creates a server and a client connects. The client writes data + // to the socket and the server echos it back. The client confirms the data it + // reads is the same as the data sent, then closes the socket, resulting in + // the closing of the server, which responds on receivePort with null to + // specify the echo server isolate is about to be killed. If an error occurs + // in the echo server, the exception and stack trace are sent to receivePort, + // which prints the exception and stack trace before eventually throwing an + // error. + ReceivePort receivePort = new ReceivePort(); + Isolate echo = await Isolate.spawn(echoServer, receivePort.sendPort); + + Map response = await receivePort.first; + SendPort sendPort = response[EchoServerTypes.ISOLATE_SEND_PORT]; + int serverInternetPort = response[EchoServerTypes.SERVER_PORT]; + + receivePort = new ReceivePort(); + echo.addErrorListener(receivePort.sendPort); + + Map limits = { + EchoServerTypes.OFFSET_START: 0, + EchoServerTypes.OFFSET_END: messageSize, + EchoServerTypes.ISOLATE_SEND_PORT: receivePort.sendPort + }; + sendPort.send(limits); + + try { + var socket = RawSynchronousSocket.connectSync( + LOOPBACK_IP_V4_STRING, serverInternetPort); + List data = createTestData(); + socket.writeFromSync(data); + List result = socket.readSync(data.length); + verifyTestData(result); + socket.shutdown(SocketDirection.SEND); + socket.closeSync(); + } catch (e, stack) { + print("Echo test failed in the client"); + rethrow; + } + // Wait for the server to shutdown before finishing the test. + var result = await receivePort.first; + if (result != null) { + throw "Echo test failed in server!\nError: ${result[0]}\nStack trace:" + + " ${result[1]}"; + } + asyncEnd(); +} + +Future testPartialRead() async { + asyncStart(); + // This test is based on testSimpleReadWrite, but instead of reading the + // entire echoed message at once, it reads it in two calls to readIntoSync. + ReceivePort receivePort = new ReceivePort(); + Isolate echo = await Isolate.spawn(echoServer, receivePort.sendPort); + + Map response = await receivePort.first; + SendPort sendPort = response[EchoServerTypes.ISOLATE_SEND_PORT]; + int serverInternetPort = response[EchoServerTypes.SERVER_PORT]; + List data = createTestData(); + + receivePort = new ReceivePort(); + echo.addErrorListener(receivePort.sendPort); + + Map limits = { + EchoServerTypes.OFFSET_START: 0, + EchoServerTypes.OFFSET_END: 1000, + EchoServerTypes.ISOLATE_SEND_PORT: receivePort.sendPort + }; + sendPort.send(limits); + + try { + var socket = RawSynchronousSocket.connectSync( + LOOPBACK_IP_V4_STRING, serverInternetPort); + int half_length = (data.length / 2).toInt(); + + // Send the full data list to the server. + socket.writeFromSync(data); + List result = new List(data.length); + + // Read half at a time and check that there's still more bytes available. + socket.readIntoSync(result, 0, half_length); + verifyTestData(result.sublist(0, half_length), 0, half_length); + Expect.isTrue(socket.available() == (data.length - half_length)); + + // Read the second half and verify again. + socket.readIntoSync(result, half_length); + verifyTestData(result); + Expect.isTrue(socket.available() == 0); + + socket.closeSync(); + } catch (e, stack) { + print("Echo test failed in the client."); + rethrow; + } + // Wait for the server to shutdown before finishing the test. + var result = await receivePort.first; + if (result != null) { + throw "Echo test failed in server!\nError: ${result[0]}\nStack trace:" + + " ${result[1]}"; + } + asyncEnd(); +} + +Future testPartialWrite() async { + asyncStart(); + // This test is based on testSimpleReadWrite, but instead of writing the + // entire data buffer at once, it writes different parts of the buffer over + // multiple calls to writeFromSync. + ReceivePort receivePort = new ReceivePort(); + Isolate echo = await Isolate.spawn(echoServer, receivePort.sendPort); + + Map response = await receivePort.first; + List data = createTestData(); + SendPort sendPort = response[EchoServerTypes.ISOLATE_SEND_PORT]; + int startOffset = 32; + int endOffset = (data.length / 2).toInt(); + int serverInternetPort = response[EchoServerTypes.SERVER_PORT]; + + receivePort = new ReceivePort(); + echo.addErrorListener(receivePort.sendPort); + + Map limits = { + EchoServerTypes.OFFSET_START: startOffset, + EchoServerTypes.OFFSET_END: endOffset, + EchoServerTypes.ISOLATE_SEND_PORT: receivePort.sendPort + }; + sendPort.send(limits); + try { + var socket = RawSynchronousSocket.connectSync( + LOOPBACK_IP_V4_STRING, serverInternetPort); + List data = createTestData(); + + // Write a subset of data to the server. + socket.writeFromSync(data, startOffset, endOffset); + + // Grab the response and verify it's correct. + List result = new List(endOffset - startOffset); + socket.readIntoSync(result); + + Expect.equals(result.length, endOffset - startOffset); + verifyTestData(result, startOffset, endOffset); + socket.closeSync(); + } catch (e, stack) { + print("Echo test failed in the client."); + rethrow; + } + + // Wait for the server to shutdown before finishing the test. + var result = await receivePort.first; + if (result != null) { + throw "Echo test failed in server!\nError: ${result[0]}\nStack trace:" + + " ${result[1]}"; + } + asyncEnd(); +} + +Future testShutdown() async { + asyncStart(); + // This test creates a server and a client connects. The client then tries to + // perform various operations after being shutdown in a specific direction, to + // ensure reads or writes cannot be performed if the socket has been shutdown + // for reading or writing. + ReceivePort receivePort = new ReceivePort(); + Isolate echo = await Isolate.spawn(echoServer, receivePort.sendPort); + + Map response = await receivePort.first; + SendPort sendPort = response[EchoServerTypes.ISOLATE_SEND_PORT]; + int serverInternetPort = response[EchoServerTypes.SERVER_PORT]; + List data = createTestData(); + + receivePort = new ReceivePort(); + echo.addErrorListener(receivePort.sendPort); + + Map limits = { + EchoServerTypes.OFFSET_START: 0, + EchoServerTypes.OFFSET_END: data.length, + EchoServerTypes.ISOLATE_SEND_PORT: receivePort.sendPort, + // Tell the server to shutdown after 3 sockets disconnect. + EchoServerTypes.CONNECTION_COUNT: 3 + }; + sendPort.send(limits); + + try { + var socket = RawSynchronousSocket.connectSync( + LOOPBACK_IP_V4_STRING, serverInternetPort); + + // Close from both directions. Shouldn't be able to read/write to the + // socket. + socket.shutdown(SocketDirection.BOTH); + Expect.throws( + () => socket.writeFromSync(data), (e) => e is SocketException); + Expect.throws( + () => socket.readSync(data.length), (e) => e is SocketException); + socket.closeSync(); + + // Close the socket for reading, do a write, and see if we can get any + // response from the server (we shouldn't be able to). + socket = RawSynchronousSocket.connectSync( + LOOPBACK_IP_V4_STRING, serverInternetPort); + socket.shutdown(SocketDirection.RECEIVE); + socket.writeFromSync(data); + // Throws exception when the socket is closed for RECEIVE. + Expect.throws( + () => socket.readSync(data.length), (e) => e is SocketException); + Expect.isTrue(socket.available() == 0); + socket.closeSync(); + + // Close the socket for writing and try to do a write. This should cause an + // OSError to be throw as the pipe is closed for writing. + socket = RawSynchronousSocket.connectSync( + LOOPBACK_IP_V4_STRING, serverInternetPort); + socket.shutdown(SocketDirection.SEND); + Expect.throws( + () => socket.writeFromSync(data), (e) => e is SocketException); + socket.closeSync(); + } catch (e, stack) { + print("Echo test failed in client."); + rethrow; + } + // Wait for the server to shutdown before finishing the test. + var result = await receivePort.first; + if (result != null) { + throw "Echo test failed in server!\nError: ${result[0]}\nStack trace:" + + " ${result[1]}"; + } + asyncEnd(); +} + +Future testInvalidReadWriteOperations() { + asyncStart(); + RawServerSocket.bind(InternetAddress.LOOPBACK_IP_V4, 0).then((server) { + server.listen((socket) {}); + List data = createTestData(); + var socket = + RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, server.port); + + // Invalid writeFromSync invocations + Expect.throws(() => socket.writeFromSync(data, data.length + 1), + (e) => e is RangeError); + Expect.throws(() => socket.writeFromSync(data, 0, data.length + 1), + (e) => e is RangeError); + Expect.throws( + () => socket.writeFromSync(data, 1, 0), (e) => e is RangeError); + Expect.throws( + () => socket.writeFromSync(data, null), (e) => e is ArgumentError); + + // Invalid readIntoSync invocations + List buffer = new List(10); + Expect.throws(() => socket.readIntoSync(buffer, buffer.length + 1), + (e) => e is RangeError); + Expect.throws(() => socket.readIntoSync(buffer, 0, buffer.length + 1), + (e) => e is RangeError); + Expect.throws( + () => socket.readIntoSync(buffer, 1, 0), (e) => e is RangeError); + Expect.throws( + () => socket.readIntoSync(buffer, null), (e) => e is ArgumentError); + + // Invalid readSync invocation + Expect.throws(() => socket.readSync(-1), (e) => e is ArgumentError); + + server.close(); + socket.closeSync(); + asyncEnd(); + }); +} + +void testClosedError() { + asyncStart(); + RawServerSocket.bind(InternetAddress.LOOPBACK_IP_V4, 0).then((server) { + server.listen((socket) { + socket.close(); + }); + var socket = + RawSynchronousSocket.connectSync(LOOPBACK_IP_V4_STRING, server.port); + server.close(); + socket.closeSync(); + Expect.throws(() => socket.remotePort, (e) => e is SocketException); + Expect.throws(() => socket.remoteAddress, (e) => e is SocketException); + asyncEnd(); + }); +} + +main() async { + asyncStart(); + testArguments(); + // testInvalidConnect(); Long timeout for bad lookups, so disable for bots. + await testShutdown(); + testSimpleConnect(); + testServerListenAfterConnect(); + await testSimpleReadWrite(); + await testPartialRead(); + await testPartialWrite(); + testInvalidReadWriteOperations(); + testClosedError(); + asyncEnd(); +} diff --git a/tests/standalone/standalone.status b/tests/standalone/standalone.status index 05251c5a48e..9fc1a5f5594 100644 --- a/tests/standalone/standalone.status +++ b/tests/standalone/standalone.status @@ -225,6 +225,7 @@ io/process_sync_test: Pass, Timeout # Issue 24596 io/sleep_test: Pass, Fail # Issue 25757 io/http_server_early_client_close2_test: Pass, Crash # Issue 28197 io/process_stdin_transform_unsubscribe_test: Pass, Timeout # Issue #28558 +io/raw_synchronous_socket_test: Crash # Issue 29333, currently unsupported [ $system == windows && $compiler != dart2analyzer ] io/platform_resolved_executable_test/06: RuntimeError # Issue 23641