diff --git a/runtime/bin/eventhandler.h b/runtime/bin/eventhandler.h index d35fc539ab1..7da269cadc4 100644 --- a/runtime/bin/eventhandler.h +++ b/runtime/bin/eventhandler.h @@ -259,22 +259,53 @@ class DescriptorInfoBase { DISALLOW_COPY_AND_ASSIGN(DescriptorInfoBase); }; +#if !defined(DART_HOST_OS_WINDOWS) +// On POSIX systems we use tokens to balance work between isolates listening to +// the same socket. +template +class TokenCounter { + public: + void Return(intptr_t amount) { + ASSERT(amount >= 0); + tokens_ += amount; + ASSERT(tokens_ <= kInitialAmount); + } + + void TakeOne() { tokens_--; } + + bool IsEmpty() const { return tokens_ <= 0; } + + private: + intptr_t tokens_ = kInitialAmount; +}; +#else +// On Windows we don't use this mechanism. +template +class TokenCounter { + public: + void Return(intptr_t amount) { + // Do nothing. + } + + void TakeOne() { + // Do nothing. + } + + bool IsEmpty() const { return false; } +}; + +#endif + // Describes a OS descriptor (e.g. file descriptor on linux or HANDLE on // windows) which is connected to a single Dart_Port. // // Subclasses of this class can be e.g. connected tcp sockets. template class DescriptorInfoSingleMixin : public DI { - private: - static constexpr int kTokenCount = 16; - public: - DescriptorInfoSingleMixin(intptr_t fd, bool disable_tokens) - : DI(fd), - port_(0), - tokens_(kTokenCount), - mask_(0), - disable_tokens_(disable_tokens) {} + template + DescriptorInfoSingleMixin(intptr_t fd, Args... args) + : DI(fd, args...), port_(0), mask_(0) {} virtual ~DescriptorInfoSingleMixin() {} @@ -302,9 +333,7 @@ class DescriptorInfoSingleMixin : public DI { virtual Dart_Port NextNotifyDartPort(intptr_t events_ready) { ASSERT(IS_IO_EVENT(events_ready) || IS_EVENT(events_ready, kDestroyedEvent)); - if (!disable_tokens_) { - tokens_--; - } + tokens_.TakeOne(); return port_; } @@ -317,21 +346,16 @@ class DescriptorInfoSingleMixin : public DI { if (port_ != 0) { DartUtils::PostInt32(port_, events); } - if (!disable_tokens_) { - tokens_--; - } + tokens_.TakeOne(); } virtual void ReturnTokens(Dart_Port port, int count) { ASSERT(port_ == port); - if (!disable_tokens_) { - tokens_ += count; - } - ASSERT(tokens_ <= kTokenCount); + tokens_.Return(count); } virtual intptr_t Mask() { - if (tokens_ <= 0) { + if (tokens_.IsEmpty()) { return 0; } return mask_; @@ -341,9 +365,8 @@ class DescriptorInfoSingleMixin : public DI { private: Dart_Port port_; - int tokens_; + TokenCounter tokens_; intptr_t mask_; - bool disable_tokens_; DISALLOW_COPY_AND_ASSIGN(DescriptorInfoSingleMixin); }; @@ -356,8 +379,6 @@ class DescriptorInfoSingleMixin : public DI { template class DescriptorInfoMultipleMixin : public DI { private: - static constexpr int kTokenCount = 4; - static bool SamePortValue(void* key1, void* key2) { return reinterpret_cast(key1) == reinterpret_cast(key2); @@ -383,16 +404,15 @@ class DescriptorInfoMultipleMixin : public DI { struct PortEntry { Dart_Port dart_port; intptr_t is_reading; - intptr_t token_count; + TokenCounter tokens; - bool IsReady() { return token_count > 0 && is_reading != 0; } + bool IsReady() { return !tokens.IsEmpty() && is_reading != 0; } }; public: - DescriptorInfoMultipleMixin(intptr_t fd, bool disable_tokens) - : DI(fd), - tokens_map_(&SamePortValue, kTokenCount), - disable_tokens_(disable_tokens) {} + template + DescriptorInfoMultipleMixin(intptr_t fd, Args... args) + : DI(fd, args...), tokens_map_(&SamePortValue, 4) {} virtual ~DescriptorInfoMultipleMixin() { RemoveAllPorts(); } @@ -405,7 +425,6 @@ class DescriptorInfoMultipleMixin : public DI { if (entry->value == nullptr) { pentry = new PortEntry(); pentry->dart_port = port; - pentry->token_count = kTokenCount; pentry->is_reading = IsReadingMask(mask); entry->value = reinterpret_cast(pentry); @@ -498,10 +517,8 @@ class DescriptorInfoMultipleMixin : public DI { PortEntry* pentry = reinterpret_cast(active_readers_.head()); // Update token count. - if (!disable_tokens_) { - pentry->token_count--; - } - if (pentry->token_count <= 0) { + pentry->tokens.TakeOne(); + if (pentry->tokens.IsEmpty()) { active_readers_.RemoveHead(); } else { active_readers_.Rotate(); @@ -525,11 +542,8 @@ class DescriptorInfoMultipleMixin : public DI { // Update token count. bool was_ready = pentry->IsReady(); - if (!disable_tokens_) { - pentry->token_count--; - } - - if (was_ready && (pentry->token_count <= 0)) { + pentry->tokens.TakeOne(); + if (was_ready && pentry->tokens.IsEmpty()) { active_readers_.Remove(pentry); } } @@ -542,10 +556,7 @@ class DescriptorInfoMultipleMixin : public DI { PortEntry* pentry = reinterpret_cast(entry->value); bool was_ready = pentry->IsReady(); - if (!disable_tokens_) { - pentry->token_count += count; - } - ASSERT(pentry->token_count <= kTokenCount); + pentry->tokens.Return(count); bool is_ready = pentry->IsReady(); if (!was_ready && is_ready) { active_readers_.Add(pentry); @@ -573,11 +584,9 @@ class DescriptorInfoMultipleMixin : public DI { CircularLinkedList active_readers_; // A convenience mapping: - // Dart_Port -> struct PortEntry { dart_port, mask, token_count } + // Dart_Port -> struct PortEntry { dart_port, mask, tokens } SimpleHashMap tokens_map_; - bool disable_tokens_; - DISALLOW_COPY_AND_ASSIGN(DescriptorInfoMultipleMixin); }; diff --git a/runtime/bin/eventhandler_fuchsia.h b/runtime/bin/eventhandler_fuchsia.h index 76589b5a260..d9a381a34c0 100644 --- a/runtime/bin/eventhandler_fuchsia.h +++ b/runtime/bin/eventhandler_fuchsia.h @@ -116,8 +116,7 @@ class DescriptorInfo : public DescriptorInfoBase { class DescriptorInfoSingle : public DescriptorInfoSingleMixin { public: - explicit DescriptorInfoSingle(intptr_t fd) - : DescriptorInfoSingleMixin(fd, false) {} + explicit DescriptorInfoSingle(intptr_t fd) : DescriptorInfoSingleMixin(fd) {} virtual ~DescriptorInfoSingle() {} private: @@ -128,7 +127,7 @@ class DescriptorInfoMultiple : public DescriptorInfoMultipleMixin { public: explicit DescriptorInfoMultiple(intptr_t fd) - : DescriptorInfoMultipleMixin(fd, false) {} + : DescriptorInfoMultipleMixin(fd) {} virtual ~DescriptorInfoMultiple() {} private: diff --git a/runtime/bin/eventhandler_linux.h b/runtime/bin/eventhandler_linux.h index 31cb3ec5e8b..405ec30d706 100644 --- a/runtime/bin/eventhandler_linux.h +++ b/runtime/bin/eventhandler_linux.h @@ -39,8 +39,7 @@ class DescriptorInfo : public DescriptorInfoBase { class DescriptorInfoSingle : public DescriptorInfoSingleMixin { public: - explicit DescriptorInfoSingle(intptr_t fd) - : DescriptorInfoSingleMixin(fd, false) {} + explicit DescriptorInfoSingle(intptr_t fd) : DescriptorInfoSingleMixin(fd) {} virtual ~DescriptorInfoSingle() {} private: @@ -51,7 +50,7 @@ class DescriptorInfoMultiple : public DescriptorInfoMultipleMixin { public: explicit DescriptorInfoMultiple(intptr_t fd) - : DescriptorInfoMultipleMixin(fd, false) {} + : DescriptorInfoMultipleMixin(fd) {} virtual ~DescriptorInfoMultiple() {} private: diff --git a/runtime/bin/eventhandler_macos.h b/runtime/bin/eventhandler_macos.h index a7684f21e12..7469f5461a0 100644 --- a/runtime/bin/eventhandler_macos.h +++ b/runtime/bin/eventhandler_macos.h @@ -51,8 +51,7 @@ class DescriptorInfo : public DescriptorInfoBase { class DescriptorInfoSingle : public DescriptorInfoSingleMixin { public: - explicit DescriptorInfoSingle(intptr_t fd) - : DescriptorInfoSingleMixin(fd, false) {} + explicit DescriptorInfoSingle(intptr_t fd) : DescriptorInfoSingleMixin(fd) {} virtual ~DescriptorInfoSingle() {} private: @@ -63,7 +62,7 @@ class DescriptorInfoMultiple : public DescriptorInfoMultipleMixin { public: explicit DescriptorInfoMultiple(intptr_t fd) - : DescriptorInfoMultipleMixin(fd, false) {} + : DescriptorInfoMultipleMixin(fd) {} virtual ~DescriptorInfoMultiple() {} private: diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc index 1683ac7a376..a747c3d31c3 100644 --- a/runtime/bin/eventhandler_win.cc +++ b/runtime/bin/eventhandler_win.cc @@ -43,43 +43,87 @@ static constexpr int kAcceptExAddressAdditionalBytes = 16; static constexpr int kAcceptExAddressStorageSize = sizeof(SOCKADDR_STORAGE) + kAcceptExAddressAdditionalBytes; -OverlappedBuffer* OverlappedBuffer::AllocateBuffer(int buffer_size, +OverlappedBuffer::OverlappedBuffer(Handle* handle, + int buffer_size, + Operation operation) + : buflen_(buffer_size), operation_(operation), handle_(handle) { + memset(GetBufferStart(), 0, GetBufferSize()); + if (operation == kRecvFrom) { + // Reserve part of the buffer for the length of source sockaddr + // and source sockaddr. + const int kAdditionalSize = + sizeof(struct sockaddr_storage) + sizeof(socklen_t); + ASSERT(buflen_ > kAdditionalSize); + buflen_ -= kAdditionalSize; + from_len_addr_ = + reinterpret_cast(GetBufferStart() + GetBufferSize()); + *from_len_addr_ = sizeof(struct sockaddr_storage); + from_ = reinterpret_cast(from_len_addr_ + 1); + } else { + from_len_addr_ = nullptr; + from_ = nullptr; + } + index_ = 0; + data_length_ = 0; + if (operation_ == kAccept) { + client_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + } + + // Retain handle for the duration of the operation. + handle->Retain(); +} + +OverlappedBuffer::~OverlappedBuffer() { + // If handle was not detached from the buffer release the reference + // we were holding to it. + if (handle_ != nullptr) { + handle_->Release(); + } +} + +OverlappedBuffer* OverlappedBuffer::AllocateBuffer(Handle* handle, + int buffer_size, Operation operation) { OverlappedBuffer* buffer = - new (buffer_size) OverlappedBuffer(buffer_size, operation); + new (buffer_size) OverlappedBuffer(handle, buffer_size, operation); return buffer; } -OverlappedBuffer* OverlappedBuffer::AllocateAcceptBuffer(int buffer_size) { - OverlappedBuffer* buffer = AllocateBuffer(buffer_size, kAccept); +OverlappedBuffer* OverlappedBuffer::AllocateAcceptBuffer(Handle* handle, + int buffer_size) { + OverlappedBuffer* buffer = AllocateBuffer(handle, buffer_size, kAccept); return buffer; } -OverlappedBuffer* OverlappedBuffer::AllocateReadBuffer(int buffer_size) { - return AllocateBuffer(buffer_size, kRead); +OverlappedBuffer* OverlappedBuffer::AllocateReadBuffer(Handle* handle, + int buffer_size) { + return AllocateBuffer(handle, buffer_size, kRead); } -OverlappedBuffer* OverlappedBuffer::AllocateRecvFromBuffer(int buffer_size) { +OverlappedBuffer* OverlappedBuffer::AllocateRecvFromBuffer(Handle* handle, + int buffer_size) { // For calling recvfrom additional buffer space is needed for the source // address information. buffer_size += sizeof(socklen_t) + sizeof(struct sockaddr_storage); - return AllocateBuffer(buffer_size, kRecvFrom); + return AllocateBuffer(handle, buffer_size, kRecvFrom); } -OverlappedBuffer* OverlappedBuffer::AllocateWriteBuffer(int buffer_size) { - return AllocateBuffer(buffer_size, kWrite); +OverlappedBuffer* OverlappedBuffer::AllocateWriteBuffer(Handle* handle, + int buffer_size) { + return AllocateBuffer(handle, buffer_size, kWrite); } -OverlappedBuffer* OverlappedBuffer::AllocateSendToBuffer(int buffer_size) { - return AllocateBuffer(buffer_size, kSendTo); +OverlappedBuffer* OverlappedBuffer::AllocateSendToBuffer(Handle* handle, + int buffer_size) { + return AllocateBuffer(handle, buffer_size, kSendTo); } -OverlappedBuffer* OverlappedBuffer::AllocateDisconnectBuffer() { - return AllocateBuffer(0, kDisconnect); +OverlappedBuffer* OverlappedBuffer::AllocateDisconnectBuffer(Handle* handle) { + return AllocateBuffer(handle, 0, kDisconnect); } -OverlappedBuffer* OverlappedBuffer::AllocateConnectBuffer() { - return AllocateBuffer(0, kConnect); +OverlappedBuffer* OverlappedBuffer::AllocateConnectBuffer(Handle* handle) { + return AllocateBuffer(handle, 0, kConnect); } void OverlappedBuffer::DisposeBuffer(OverlappedBuffer* buffer) { @@ -113,13 +157,14 @@ int OverlappedBuffer::GetRemainingLength() { return data_length_ - index_; } -Handle::Handle(intptr_t handle) +Handle::Handle(intptr_t handle, + Type type, + Handle::SupportsOverlappedIO supports_overlapped_io /* = kYes */) : ReferenceCounted(), DescriptorInfoBase(handle), monitor_(), + type_(type), handle_(reinterpret_cast(handle)), - completion_port_(INVALID_HANDLE_VALUE), - event_handler_(nullptr), data_ready_(), pending_read_(nullptr), pending_write_(nullptr), @@ -128,23 +173,19 @@ Handle::Handle(intptr_t handle) read_thread_handle_(nullptr), read_thread_starting_(false), read_thread_finished_(false), - flags_(0) {} + flags_(0) { + if (supports_overlapped_io == SupportsOverlappedIO::kYes) { + EventHandler::delegate()->AssociateWithCompletionPort(this); + } else { + flags_ |= 1 << kDoesNotSupportOverlappedIO; + } +} Handle::~Handle() {} -bool Handle::CreateCompletionPort(HANDLE completion_port) { - ASSERT(completion_port_ == INVALID_HANDLE_VALUE); - // A reference to the Handle is Retained by the IO completion port. - // It is Released by DeleteIfClosed. - Retain(); - completion_port_ = CreateIoCompletionPort( - handle(), completion_port, reinterpret_cast(this), 0); - return (completion_port_ != nullptr); -} - void Handle::Close() { MonitorLocker ml(&monitor_); - if (!SupportsOverlappedIO()) { + if (!supports_overlapped_io()) { // If the handle uses synchronous I/O (e.g. stdin), cancel any pending // operation before closing the handle, so the read thread is not blocked. BOOL result = CancelIoEx(handle_, nullptr); @@ -233,11 +274,6 @@ void Handle::WriteComplete(OverlappedBuffer* buffer) { pending_write_ = nullptr; } -static void ReadFileThread(uword args) { - Handle* handle = reinterpret_cast(args); - handle->ReadSyncCompleteAsync(); -} - void Handle::NotifyReadThreadStarted() { MonitorLocker ml(&monitor_); ASSERT(read_thread_starting_); @@ -272,9 +308,9 @@ void Handle::ReadSyncCompleteAsync() { bytes_read = 0; } OVERLAPPED* overlapped = pending_read_->GetCleanOverlapped(); - ok = - PostQueuedCompletionStatus(event_handler_->completion_port(), bytes_read, - reinterpret_cast(this), overlapped); + ok = PostQueuedCompletionStatus(EventHandler::delegate()->completion_port(), + bytes_read, reinterpret_cast(this), + overlapped); if (!ok) { FATAL("PostQueuedCompletionStatus failed"); } @@ -284,10 +320,9 @@ void Handle::ReadSyncCompleteAsync() { bool Handle::IssueRead() { ASSERT(type_ != kListenSocket); ASSERT(!HasPendingRead()); - OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize); - if (SupportsOverlappedIO()) { - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); - + OverlappedBuffer* buffer = + OverlappedBuffer::AllocateReadBuffer(this, kBufferSize); + if (supports_overlapped_io()) { BOOL ok = ReadFile(handle_, buffer->GetBufferStart(), buffer->GetBufferSize(), nullptr, buffer->GetCleanOverlapped()); @@ -301,10 +336,17 @@ bool Handle::IssueRead() { return false; } else { // Completing asynchronously through thread. + Retain(); pending_read_ = buffer; read_thread_starting_ = true; - int result = Thread::Start("dart:io ReadFile", ReadFileThread, - reinterpret_cast(this)); + int result = Thread::Start( + "dart:io ReadFile", + [](uword args) { + auto handle = reinterpret_cast(args); + handle->ReadSyncCompleteAsync(); + handle->Release(); + }, + reinterpret_cast(this)); if (result != 0) { FATAL("Failed to start read file thread %d", result); } @@ -319,7 +361,6 @@ bool Handle::IssueRecvFrom() { bool Handle::IssueWrite() { MonitorLocker ml(&monitor_); ASSERT(type_ != kListenSocket); - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); ASSERT(HasPendingWrite()); ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite); @@ -366,35 +407,10 @@ void Handle::HandleIssueError() { SetLastError(error); } -void FileHandle::EnsureInitialized(EventHandlerImplementation* event_handler) { - MonitorLocker ml(&monitor_); - event_handler_ = event_handler; - if (completion_port_ == INVALID_HANDLE_VALUE) { - if (SupportsOverlappedIO()) { - CreateCompletionPort(event_handler_->completion_port()); - } else { - // We need to retain the Handle even if overlapped IO is not supported. - // It is Released by DeleteIfClosed after ReadSyncCompleteAsync - // manually puts an event on the IO completion port. - Retain(); - completion_port_ = event_handler_->completion_port(); - } - } -} - bool FileHandle::IsClosed() { return IsClosing() && !HasPendingRead() && !HasPendingWrite(); } -void DirectoryWatchHandle::EnsureInitialized( - EventHandlerImplementation* event_handler) { - MonitorLocker ml(&monitor_); - event_handler_ = event_handler; - if (completion_port_ == INVALID_HANDLE_VALUE) { - CreateCompletionPort(event_handler_->completion_port()); - } -} - bool DirectoryWatchHandle::IsClosed() { return IsClosing() && !HasPendingRead(); } @@ -405,12 +421,12 @@ bool DirectoryWatchHandle::IssueRead() { if (HasPendingRead() || (data_ready_ != nullptr)) { return true; } - OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(kBufferSize); + OverlappedBuffer* buffer = + OverlappedBuffer::AllocateReadBuffer(this, kBufferSize); // Set up pending_read_ before ReadDirectoryChangesW because it might be // needed in ReadComplete invoked on event loop thread right away if data is // also ready right away. pending_read_ = buffer; - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); BOOL ok = ReadDirectoryChangesW( handle_, buffer->GetBufferStart(), buffer->GetBufferSize(), recursive_, events_, nullptr, buffer->GetCleanOverlapped(), nullptr); @@ -445,56 +461,31 @@ void SocketHandle::HandleIssueError() { WSASetLastError(error); } -bool ListenSocket::LoadAcceptEx() { - // Load the AcceptEx function into memory using WSAIoctl. - GUID guid_accept_ex = WSAID_ACCEPTEX; - DWORD bytes; - int status = WSAIoctl(socket(), SIO_GET_EXTENSION_FUNCTION_POINTER, - &guid_accept_ex, sizeof(guid_accept_ex), &AcceptEx_, - sizeof(AcceptEx_), &bytes, nullptr, nullptr); - return (status != SOCKET_ERROR); -} - -bool ListenSocket::LoadGetAcceptExSockaddrs() { - // Load the GetAcceptExSockaddrs function into memory using WSAIoctl. - GUID guid_get_accept_ex_sockaddrs = WSAID_GETACCEPTEXSOCKADDRS; - DWORD bytes; - int status = - WSAIoctl(socket(), SIO_GET_EXTENSION_FUNCTION_POINTER, - &guid_get_accept_ex_sockaddrs, - sizeof(guid_get_accept_ex_sockaddrs), &GetAcceptExSockaddrs_, - sizeof(GetAcceptExSockaddrs_), &bytes, nullptr, nullptr); - return (status != SOCKET_ERROR); -} - bool ListenSocket::IssueAccept() { MonitorLocker ml(&monitor_); - OverlappedBuffer* buffer = - OverlappedBuffer::AllocateAcceptBuffer(2 * kAcceptExAddressStorageSize); + OverlappedBuffer* buffer = OverlappedBuffer::AllocateAcceptBuffer( + this, 2 * kAcceptExAddressStorageSize); DWORD received; BOOL ok; - ok = AcceptEx_(socket(), buffer->client(), buffer->GetBufferStart(), - 0, // For now don't receive data with accept. - kAcceptExAddressStorageSize, kAcceptExAddressStorageSize, - &received, buffer->GetCleanOverlapped()); - if (!ok) { - if (WSAGetLastError() != WSA_IO_PENDING) { - int error = WSAGetLastError(); - closesocket(buffer->client()); - OverlappedBuffer::DisposeBuffer(buffer); - WSASetLastError(error); - return false; - } + ok = EventHandler::delegate()->accept_ex()( + socket(), buffer->client(), buffer->GetBufferStart(), + 0, // For now don't receive data with accept. + kAcceptExAddressStorageSize, kAcceptExAddressStorageSize, &received, + buffer->GetCleanOverlapped()); + if (ok || WSAGetLastError() == WSA_IO_PENDING) { + pending_accept_count_++; + return true; } - pending_accept_count_++; - - return true; + int error = WSAGetLastError(); + closesocket(buffer->client()); + OverlappedBuffer::DisposeBuffer(buffer); + WSASetLastError(error); + return false; } -void ListenSocket::AcceptComplete(OverlappedBuffer* buffer, - HANDLE completion_port) { +void ListenSocket::AcceptComplete(OverlappedBuffer* buffer) { MonitorLocker ml(&monitor_); if (!IsClosing()) { // Update the accepted socket to support the full range of API calls. @@ -510,7 +501,7 @@ void ListenSocket::AcceptComplete(OverlappedBuffer* buffer, int local_addr_length; LPSOCKADDR remote_addr; int remote_addr_length; - GetAcceptExSockaddrs_( + EventHandler::delegate()->get_accept_ex_sockaddrs()( buffer->GetBufferStart(), 0, kAcceptExAddressStorageSize, kAcceptExAddressStorageSize, &local_addr, &local_addr_length, &remote_addr, &remote_addr_length); @@ -521,7 +512,6 @@ void ListenSocket::AcceptComplete(OverlappedBuffer* buffer, ClientSocket* client_socket = new ClientSocket( buffer->client(), std::unique_ptr(raw_remote_addr)); client_socket->mark_connected(); - client_socket->CreateCompletionPort(completion_port); if (accepted_head_ == nullptr) { accepted_head_ = client_socket; accepted_tail_ = client_socket; @@ -543,15 +533,10 @@ void ListenSocket::AcceptComplete(OverlappedBuffer* buffer, OverlappedBuffer::DisposeBuffer(buffer); } -static void DeleteIfClosed(Handle* handle) { +static void NotifyDestroyedIfClosed(Handle* handle) { if (handle->IsClosed()) { - handle->set_completion_port(INVALID_HANDLE_VALUE); - handle->set_event_handler(nullptr); handle->NotifyAllDartPorts(1 << kDestroyedEvent); handle->RemoveAllPorts(); - // Once the Handle is closed, no further events on the IO completion port - // will mention it. Thus, we can drop the reference here. - handle->Release(); } } @@ -563,23 +548,12 @@ void ListenSocket::DoClose() { ClientSocket* client = Accept(); if (client != nullptr) { client->Close(); - // Release the reference from the list. - // When an accept completes, we make a new ClientSocket (1 reference), - // and add it to the IO completion port (1 more reference). If an - // accepted connection is never requested by the Dart code, then - // this list owns a reference (first Release), and the IO completion - // port owns a reference, (second Release in DeleteIfClosed). + NotifyDestroyedIfClosed(client); client->Release(); - DeleteIfClosed(client); } else { break; } } - // To finish resetting the state of the ListenSocket back to what it was - // before EnsureInitialized was called, we have to reset the AcceptEx_ - // and GetAcceptExSockaddrs_ function pointers. - AcceptEx_ = nullptr; - GetAcceptExSockaddrs_ = nullptr; } bool ListenSocket::CanAccept() { @@ -614,23 +588,6 @@ ClientSocket* ListenSocket::Accept() { return result; } -void ListenSocket::EnsureInitialized( - EventHandlerImplementation* event_handler) { - MonitorLocker ml(&monitor_); - if (AcceptEx_ == nullptr) { - ASSERT(completion_port_ == INVALID_HANDLE_VALUE); - ASSERT(event_handler_ == nullptr); - event_handler_ = event_handler; - CreateCompletionPort(event_handler_->completion_port()); - bool isLoaded = LoadAcceptEx(); - ASSERT(isLoaded); - } - if (GetAcceptExSockaddrs_ == nullptr) { - bool isLoaded = LoadGetAcceptExSockaddrs(); - ASSERT(isLoaded); - } -} - bool ListenSocket::IsClosed() { return IsClosing() && !HasPendingAccept(); } @@ -692,18 +649,15 @@ intptr_t Handle::RecvFrom(void* buffer, intptr_t Handle::Write(const void* buffer, intptr_t num_bytes) { MonitorLocker ml(&monitor_); - if (HasPendingWrite()) { + if (HasPendingWrite() || IsClosed()) { return 0; } if (num_bytes > kBufferSize) { num_bytes = kBufferSize; } - ASSERT(SupportsOverlappedIO()); - if (completion_port_ == INVALID_HANDLE_VALUE) { - return 0; - } + ASSERT(supports_overlapped_io()); int truncated_bytes = Utils::Minimum(num_bytes, INT_MAX); - pending_write_ = OverlappedBuffer::AllocateWriteBuffer(truncated_bytes); + pending_write_ = OverlappedBuffer::AllocateWriteBuffer(this, truncated_bytes); pending_write_->Write(buffer, truncated_bytes); if (!IssueWrite()) { return -1; @@ -716,7 +670,7 @@ intptr_t Handle::SendTo(const void* buffer, struct sockaddr* sa, socklen_t sa_len) { MonitorLocker ml(&monitor_); - if (HasPendingWrite()) { + if (HasPendingWrite() || IsClosed()) { return 0; } if (num_bytes > kBufferSize) { @@ -728,13 +682,13 @@ intptr_t Handle::SendTo(const void* buffer, SetLastError(ERROR_INVALID_USER_BUFFER); return -1; } - ASSERT(SupportsOverlappedIO()); - if (completion_port_ == INVALID_HANDLE_VALUE) { - return 0; - } - pending_write_ = OverlappedBuffer::AllocateSendToBuffer(num_bytes); + pending_write_ = OverlappedBuffer::AllocateSendToBuffer(this, num_bytes); pending_write_->Write(buffer, num_bytes); if (!IssueSendTo(sa, sa_len)) { + if (pending_write_ != nullptr) { + OverlappedBuffer::DisposeBuffer(pending_write_); + pending_write_ = nullptr; + } return -1; } return num_bytes; @@ -751,11 +705,6 @@ StdHandle* StdHandle::Stdin(HANDLE handle) { return stdin_; } -static void WriteFileThread(uword args) { - StdHandle* handle = reinterpret_cast(args); - handle->RunWriteLoop(); -} - void StdHandle::RunWriteLoop() { MonitorLocker ml(&monitor_); write_thread_running_ = true; @@ -788,7 +737,7 @@ void StdHandle::WriteSyncCompleteAsync() { thread_wrote_ += bytes_written; OVERLAPPED* overlapped = pending_write_->GetCleanOverlapped(); ok = PostQueuedCompletionStatus( - event_handler_->completion_port(), bytes_written, + EventHandler::delegate()->completion_port(), bytes_written, reinterpret_cast(this), overlapped); if (!ok) { FATAL("PostQueuedCompletionStatus failed"); @@ -818,11 +767,16 @@ intptr_t StdHandle::Write(const void* buffer, intptr_t num_bytes) { if (!write_thread_exists_) { write_thread_exists_ = true; // The write thread gets a reference to the Handle, which it places in - // the events it puts on the IO completion port. The reference is - // Released by DeleteIfClosed. + // the events it puts on the IO completion port. Retain(); - int result = Thread::Start("dart:io WriteFile", WriteFileThread, - reinterpret_cast(this)); + int result = Thread::Start( + "dart:io WriteFile", + [](uword args) { + auto handle = reinterpret_cast(args); + handle->RunWriteLoop(); + handle->Release(); + }, + reinterpret_cast(this)); if (result != 0) { FATAL("Failed to start write file thread %d", result); } @@ -834,7 +788,7 @@ intptr_t StdHandle::Write(const void* buffer, intptr_t num_bytes) { // Only queue up to INT_MAX bytes. int truncated_bytes = Utils::Minimum(num_bytes, INT_MAX); // Create buffer and notify thread about the new handle. - pending_write_ = OverlappedBuffer::AllocateWriteBuffer(truncated_bytes); + pending_write_ = OverlappedBuffer::AllocateWriteBuffer(this, truncated_bytes); pending_write_->Write(buffer, truncated_bytes); ml.Notify(); return 0; @@ -865,17 +819,6 @@ void StdHandle::DoClose() { intptr_t ClientSocket::disconnecting_ = 0; #endif -bool ClientSocket::LoadDisconnectEx() { - // Load the DisconnectEx function into memory using WSAIoctl. - GUID guid_disconnect_ex = WSAID_DISCONNECTEX; - DWORD bytes; - int status = - WSAIoctl(socket(), SIO_GET_EXTENSION_FUNCTION_POINTER, - &guid_disconnect_ex, sizeof(guid_disconnect_ex), &DisconnectEx_, - sizeof(DisconnectEx_), &bytes, nullptr, nullptr); - return (status != SOCKET_ERROR); -} - void ClientSocket::Shutdown(int how) { int rc = shutdown(socket(), how); if (how == SD_RECEIVE) { @@ -899,12 +842,11 @@ void ClientSocket::DoClose() { bool ClientSocket::IssueRead() { MonitorLocker ml(&monitor_); - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); ASSERT(!HasPendingRead()); // TODO(sgjesse): Use a MTU value here. Only the loopback adapter can // handle 64k datagrams. - OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(65536); + OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(this, 65536); DWORD flags; flags = 0; @@ -922,7 +864,6 @@ bool ClientSocket::IssueRead() { bool ClientSocket::IssueWrite() { MonitorLocker ml(&monitor_); - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); ASSERT(HasPendingWrite()); ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite); @@ -938,9 +879,9 @@ bool ClientSocket::IssueWrite() { } void ClientSocket::IssueDisconnect() { - OverlappedBuffer* buffer = OverlappedBuffer::AllocateDisconnectBuffer(); - BOOL ok = - DisconnectEx_(socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0); + OverlappedBuffer* buffer = OverlappedBuffer::AllocateDisconnectBuffer(this); + BOOL ok = EventHandler::delegate()->disconnect_ex()( + socket(), buffer->GetCleanOverlapped(), TF_REUSE_SOCKET, 0); // DisconnectEx works like other OverlappedIO APIs, where we can get either an // immediate success or delayed operation by WSA_IO_PENDING being set. if (ok || (WSAGetLastError() != WSA_IO_PENDING)) { @@ -985,16 +926,6 @@ void ClientSocket::ConnectComplete(OverlappedBuffer* buffer) { } } -void ClientSocket::EnsureInitialized( - EventHandlerImplementation* event_handler) { - MonitorLocker ml(&monitor_); - if (completion_port_ == INVALID_HANDLE_VALUE) { - ASSERT(event_handler_ == nullptr); - event_handler_ = event_handler; - CreateCompletionPort(event_handler_->completion_port()); - } -} - bool ClientSocket::IsClosed() { return connected_ && closed_ && !HasPendingRead() && !HasPendingWrite(); } @@ -1009,7 +940,6 @@ bool ClientSocket::PopulateRemoteAddr(RawAddr& addr) { bool DatagramSocket::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) { MonitorLocker ml(&monitor_); - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); ASSERT(HasPendingWrite()); ASSERT(pending_write_->operation() == OverlappedBuffer::kSendTo); @@ -1026,11 +956,10 @@ bool DatagramSocket::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) { bool DatagramSocket::IssueRecvFrom() { MonitorLocker ml(&monitor_); - ASSERT(completion_port_ != INVALID_HANDLE_VALUE); ASSERT(!HasPendingRead()); OverlappedBuffer* buffer = - OverlappedBuffer::AllocateRecvFromBuffer(kMaxUDPPackageLength); + OverlappedBuffer::AllocateRecvFromBuffer(this, kMaxUDPPackageLength); DWORD flags; flags = 0; @@ -1047,16 +976,6 @@ bool DatagramSocket::IssueRecvFrom() { return false; } -void DatagramSocket::EnsureInitialized( - EventHandlerImplementation* event_handler) { - MonitorLocker ml(&monitor_); - if (completion_port_ == INVALID_HANDLE_VALUE) { - ASSERT(event_handler_ == nullptr); - event_handler_ = event_handler; - CreateCompletionPort(event_handler_->completion_port()); - } -} - bool DatagramSocket::IsClosed() { return IsClosing() && !HasPendingRead() && !HasPendingWrite(); } @@ -1085,9 +1004,11 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) { Handle* handle = reinterpret_cast(socket->fd()); ASSERT(handle != nullptr); + handle->Retain(); + RefCntReleaseScope rh(handle); + if (handle->is_listen_socket()) { ListenSocket* listen_socket = reinterpret_cast(handle); - listen_socket->EnsureInitialized(this); MonitorLocker ml(&listen_socket->monitor_); @@ -1120,7 +1041,6 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) { UNREACHABLE(); } } else { - handle->EnsureInitialized(this); MonitorLocker ml(&handle->monitor_); if (IS_COMMAND(msg->data, kReturnTokenCommand)) { @@ -1201,20 +1121,18 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) { } } - DeleteIfClosed(handle); + NotifyDestroyedIfClosed(handle); } } void EventHandlerImplementation::HandleAccept(ListenSocket* listen_socket, OverlappedBuffer* buffer) { - listen_socket->AcceptComplete(buffer, completion_port_); + listen_socket->AcceptComplete(buffer); { MonitorLocker ml(&listen_socket->monitor_); TryDispatchingPendingAccepts(listen_socket); } - - DeleteIfClosed(listen_socket); } void EventHandlerImplementation::TryDispatchingPendingAccepts( @@ -1251,8 +1169,6 @@ void EventHandlerImplementation::HandleRead(Handle* handle, HandleError(handle); } } - - DeleteIfClosed(handle); } void EventHandlerImplementation::HandleRecvFrom(Handle* handle, @@ -1272,8 +1188,6 @@ void EventHandlerImplementation::HandleRecvFrom(Handle* handle, } else { HandleError(handle); } - - DeleteIfClosed(handle); } void EventHandlerImplementation::HandleWrite(Handle* handle, @@ -1294,15 +1208,12 @@ void EventHandlerImplementation::HandleWrite(Handle* handle, } else { HandleError(handle); } - - DeleteIfClosed(handle); } void EventHandlerImplementation::HandleDisconnect(ClientSocket* client_socket, int bytes, OverlappedBuffer* buffer) { client_socket->DisconnectComplete(buffer); - DeleteIfClosed(client_socket); } void EventHandlerImplementation::HandleConnect(ClientSocket* client_socket, @@ -1315,7 +1226,6 @@ void EventHandlerImplementation::HandleConnect(ClientSocket* client_socket, client_socket->ConnectComplete(buffer); } client_socket->mark_connected(); - DeleteIfClosed(client_socket); } void EventHandlerImplementation::HandleTimeout() { @@ -1326,45 +1236,62 @@ void EventHandlerImplementation::HandleTimeout() { timeout_queue_.RemoveCurrent(); } +static const char* OperationName(OverlappedBuffer::Operation op) { + switch (op) { + case OverlappedBuffer::kAccept: + return "Accept"; + case OverlappedBuffer::kRead: + return "Read"; + case OverlappedBuffer::kRecvFrom: + return "RecvFrom"; + case OverlappedBuffer::kWrite: + return "Write"; + case OverlappedBuffer::kSendTo: + return "SendTo"; + case OverlappedBuffer::kDisconnect: + return "Disconnect"; + case OverlappedBuffer::kConnect: + return "Connect"; + } + return "?"; +} + void EventHandlerImplementation::HandleIOCompletion(DWORD bytes, ULONG_PTR key, OVERLAPPED* overlapped) { OverlappedBuffer* buffer = OverlappedBuffer::GetFromOverlapped(overlapped); + Handle* handle = reinterpret_cast(key); + RefCntReleaseScope release(buffer->StealHandle()); switch (buffer->operation()) { case OverlappedBuffer::kAccept: { - ListenSocket* listen_socket = reinterpret_cast(key); - HandleAccept(listen_socket, buffer); + HandleAccept(static_cast(handle), buffer); break; } case OverlappedBuffer::kRead: { - Handle* handle = reinterpret_cast(key); HandleRead(handle, bytes, buffer); break; } case OverlappedBuffer::kRecvFrom: { - Handle* handle = reinterpret_cast(key); HandleRecvFrom(handle, bytes, buffer); break; } case OverlappedBuffer::kWrite: case OverlappedBuffer::kSendTo: { - Handle* handle = reinterpret_cast(key); HandleWrite(handle, bytes, buffer); break; } case OverlappedBuffer::kDisconnect: { - ClientSocket* client_socket = reinterpret_cast(key); - HandleDisconnect(client_socket, bytes, buffer); + HandleDisconnect(static_cast(handle), bytes, buffer); break; } case OverlappedBuffer::kConnect: { - ClientSocket* client_socket = reinterpret_cast(key); - HandleConnect(client_socket, bytes, buffer); + HandleConnect(static_cast(handle), bytes, buffer); break; } default: UNREACHABLE(); } + NotifyDestroyedIfClosed(handle); } void EventHandlerImplementation::HandleCompletionOrInterrupt( @@ -1415,6 +1342,35 @@ EventHandlerImplementation::EventHandlerImplementation() { shutdown_ = false; } +namespace { +template +void GetSocketExtensionFunction(SOCKET socket, GUID guid, F* result) { + DWORD bytes; + int status = + WSAIoctl(socket, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), + result, sizeof(F), &bytes, nullptr, nullptr); + if (status == SOCKET_ERROR) { + FATAL("Failed to get a pointer to the extension function."); + } +} +} // namespace + +void EventHandlerImplementation::InitializeSocketExtensions() { + if (socket_extensions_initialized_.load()) { + return; + } + + MonitorLocker ml(&monitor_); + SOCKET dummy = socket(AF_INET, SOCK_STREAM, 0); + GetSocketExtensionFunction(dummy, WSAID_ACCEPTEX, &accept_ex_); + GetSocketExtensionFunction(dummy, WSAID_CONNECTEX, &connect_ex_); + GetSocketExtensionFunction(dummy, WSAID_DISCONNECTEX, &disconnect_ex_); + GetSocketExtensionFunction(dummy, WSAID_GETACCEPTEXSOCKADDRS, + &get_accept_ex_sockaddrs_); + socket_extensions_initialized_.store(true); + closesocket(dummy); +} + EventHandlerImplementation::~EventHandlerImplementation() { // Join the handler thread. DWORD res = WaitForSingleObject(handler_thread_handle_, INFINITE); @@ -1423,6 +1379,15 @@ EventHandlerImplementation::~EventHandlerImplementation() { CloseHandle(completion_port_); } +void EventHandlerImplementation::AssociateWithCompletionPort(Handle* handle) { + HANDLE result = + CreateIoCompletionPort(handle->handle(), completion_port_, + reinterpret_cast(handle), 0); + if (result == nullptr) { + FATAL("Failed to associate handle with completion port"); + } +} + int64_t EventHandlerImplementation::GetTimeout() { if (!timeout_queue_.HasTimeout()) { return kInfinityTimeout; @@ -1452,7 +1417,7 @@ void EventHandlerImplementation::EventHandlerEntry(uword args) { ASSERT(handler_impl != nullptr); { - MonitorLocker ml(&handler_impl->startup_monitor_); + MonitorLocker ml(&handler_impl->monitor_); handler_impl->handler_thread_id_ = Thread::GetCurrentThreadId(); handler_impl->handler_thread_handle_ = OpenThread(SYNCHRONIZE, false, handler_impl->handler_thread_id_); @@ -1526,7 +1491,7 @@ void EventHandlerImplementation::Start(EventHandler* handler) { } { - MonitorLocker ml(&startup_monitor_); + MonitorLocker ml(&monitor_); while (handler_thread_id_ == Thread::kInvalidThreadId) { ml.Wait(); } diff --git a/runtime/bin/eventhandler_win.h b/runtime/bin/eventhandler_win.h index 90515770102..f370f13cb1d 100644 --- a/runtime/bin/eventhandler_win.h +++ b/runtime/bin/eventhandler_win.h @@ -46,13 +46,16 @@ class OverlappedBuffer { kConnect }; - static OverlappedBuffer* AllocateAcceptBuffer(int buffer_size); - static OverlappedBuffer* AllocateReadBuffer(int buffer_size); - static OverlappedBuffer* AllocateRecvFromBuffer(int buffer_size); - static OverlappedBuffer* AllocateWriteBuffer(int buffer_size); - static OverlappedBuffer* AllocateSendToBuffer(int buffer_size); - static OverlappedBuffer* AllocateDisconnectBuffer(); - static OverlappedBuffer* AllocateConnectBuffer(); + static OverlappedBuffer* AllocateAcceptBuffer(Handle* handle, + int buffer_size); + static OverlappedBuffer* AllocateReadBuffer(Handle* handle, int buffer_size); + static OverlappedBuffer* AllocateRecvFromBuffer(Handle* handle, + int buffer_size); + static OverlappedBuffer* AllocateWriteBuffer(Handle* handle, int buffer_size); + static OverlappedBuffer* AllocateSendToBuffer(Handle* handle, + int buffer_size); + static OverlappedBuffer* AllocateDisconnectBuffer(Handle* handle); + static OverlappedBuffer* AllocateConnectBuffer(Handle* handle); static void DisposeBuffer(OverlappedBuffer* buffer); // Find the IO buffer from the OVERLAPPED address. @@ -100,32 +103,17 @@ class OverlappedBuffer { void operator delete(void* buffer) { free(buffer); } - private: - OverlappedBuffer(int buffer_size, Operation operation) - : operation_(operation), buflen_(buffer_size) { - memset(GetBufferStart(), 0, GetBufferSize()); - if (operation == kRecvFrom) { - // Reserve part of the buffer for the length of source sockaddr - // and source sockaddr. - const int kAdditionalSize = - sizeof(struct sockaddr_storage) + sizeof(socklen_t); - ASSERT(buflen_ > kAdditionalSize); - buflen_ -= kAdditionalSize; - from_len_addr_ = - reinterpret_cast(GetBufferStart() + GetBufferSize()); - *from_len_addr_ = sizeof(struct sockaddr_storage); - from_ = reinterpret_cast(from_len_addr_ + 1); - } else { - from_len_addr_ = nullptr; - from_ = nullptr; - } - index_ = 0; - data_length_ = 0; - if (operation_ == kAccept) { - client_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); - } + Handle* StealHandle() { + auto handle = handle_; + handle_ = nullptr; + return handle; } + ~OverlappedBuffer(); + + private: + OverlappedBuffer(Handle* handle, int buffer_size, Operation operation); + void* operator new(size_t size, int buffer_size) { return malloc(size + buffer_size); } @@ -133,12 +121,15 @@ class OverlappedBuffer { // Allocate an overlapped buffer for thse specified amount of data and // operation. Some operations need additional buffer space, which is // handled by this method. - static OverlappedBuffer* AllocateBuffer(int buffer_size, Operation operation); + static OverlappedBuffer* AllocateBuffer(Handle* handle, + int buffer_size, + Operation operation); OVERLAPPED overlapped_; // OVERLAPPED structure for overlapped IO. SOCKET client_; // Used for AcceptEx client socket. int buflen_; // Length of the buffer. Operation operation_; // Type of operation issued. + Handle* handle_ = nullptr; int index_; // Index for next read from read buffer. int data_length_; // Length of the actual data in the buffer. @@ -171,6 +162,8 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { kDatagramSocket }; + enum class SupportsOverlappedIO { kYes, kNo }; + // Socket interface exposing normal socket operations. intptr_t Available(); bool DataReady(); @@ -205,12 +198,8 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { void MarkClosedWrite() { flags_ |= (1 << kCloseWrite); } void MarkError() { flags_ |= (1 << kError); } - virtual void EnsureInitialized(EventHandlerImplementation* event_handler) = 0; - HANDLE handle() { return handle_; } - bool CreateCompletionPort(HANDLE completion_port); - void Close(); virtual void DoClose(); virtual bool IsClosed() = 0; @@ -227,10 +216,7 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { bool is_client_socket() { return type_ == kClientSocket; } bool is_datagram_socket() { return type_ == kDatagramSocket; } - void MarkDoesNotSupportOverlappedIO() { - flags_ |= (1 << kDoesNotSupportOverlappedIO); - } - bool SupportsOverlappedIO() { + bool supports_overlapped_io() { return (flags_ & (1 << kDoesNotSupportOverlappedIO)) == 0; } @@ -239,14 +225,6 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { DWORD last_error() { return last_error_; } void set_last_error(DWORD last_error) { last_error_ = last_error; } - void set_completion_port(HANDLE completion_port) { - completion_port_ = completion_port; - } - - void set_event_handler(EventHandlerImplementation* event_handler) { - event_handler_ = event_handler; - } - protected: // For access to monitor_; friend class EventHandlerImplementation; @@ -259,16 +237,16 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { kError = 4 }; - explicit Handle(intptr_t handle); + Handle(intptr_t handle, + Type type_, + SupportsOverlappedIO supports_overlapped_io); virtual ~Handle(); virtual void HandleIssueError(); Monitor monitor_; - Type type_; + const Type type_; HANDLE handle_; - HANDLE completion_port_; - EventHandlerImplementation* event_handler_; std::unique_ptr data_ready_; // Buffer for data ready to be read. @@ -297,13 +275,20 @@ class Handle : public ReferenceCounted, public DescriptorInfoBase { class FileHandle : public DescriptorInfoSingleMixin { public: explicit FileHandle(HANDLE handle) - : DescriptorInfoSingleMixin(reinterpret_cast(handle), true) { - type_ = kFile; - } + : DescriptorInfoSingleMixin(reinterpret_cast(handle), + kFile, + SupportsOverlappedIO::kYes) {} - virtual void EnsureInitialized(EventHandlerImplementation* event_handler); virtual bool IsClosed(); + protected: + FileHandle(HANDLE handle, + Type type, + SupportsOverlappedIO supports_overlapped_io) + : DescriptorInfoSingleMixin(reinterpret_cast(handle), + type, + supports_overlapped_io) {} + private: DISALLOW_COPY_AND_ASSIGN(FileHandle); }; @@ -327,14 +312,12 @@ class StdHandle : public FileHandle { static StdHandle* stdin_; explicit StdHandle(HANDLE handle) - : FileHandle(handle), + : FileHandle(handle, kStd, SupportsOverlappedIO::kNo), thread_id_(Thread::kInvalidThreadId), thread_handle_(nullptr), thread_wrote_(0), write_thread_exists_(false), - write_thread_running_(false) { - type_ = kStd; - } + write_thread_running_(false) {} ThreadId thread_id_; HANDLE thread_handle_; @@ -348,13 +331,12 @@ class StdHandle : public FileHandle { class DirectoryWatchHandle : public DescriptorInfoSingleMixin { public: DirectoryWatchHandle(HANDLE handle, int events, bool recursive) - : DescriptorInfoSingleMixin(reinterpret_cast(handle), true), + : DescriptorInfoSingleMixin(reinterpret_cast(handle), + kDirectoryWatch, + SupportsOverlappedIO::kYes), events_(events), - recursive_(recursive) { - type_ = kDirectoryWatch; - } + recursive_(recursive) {} - virtual void EnsureInitialized(EventHandlerImplementation* event_handler); virtual bool IsClosed(); virtual bool IssueRead(); @@ -373,7 +355,8 @@ class SocketHandle : public Handle { SOCKET socket() const { return socket_; } protected: - explicit SocketHandle(intptr_t s) : Handle(s), socket_(s) {} + explicit SocketHandle(intptr_t s, Type type) + : Handle(s, type, SupportsOverlappedIO::kYes), socket_(s) {} virtual void HandleIssueError(); @@ -387,15 +370,11 @@ class SocketHandle : public Handle { class ListenSocket : public DescriptorInfoMultipleMixin { public: explicit ListenSocket(intptr_t s) - : DescriptorInfoMultipleMixin(s, true), - AcceptEx_(nullptr), - GetAcceptExSockaddrs_(nullptr), + : DescriptorInfoMultipleMixin(s, kListenSocket), pending_accept_count_(0), accepted_head_(nullptr), accepted_tail_(nullptr), - accepted_count_(0) { - type_ = kListenSocket; - } + accepted_count_(0) {} virtual ~ListenSocket() { ASSERT(!HasPendingAccept()); ASSERT(accepted_head_ == nullptr); @@ -409,9 +388,8 @@ class ListenSocket : public DescriptorInfoMultipleMixin { // Internal interface used by the event handler. bool HasPendingAccept() { return pending_accept_count_ > 0; } bool IssueAccept(); - void AcceptComplete(OverlappedBuffer* buffer, HANDLE completion_port); + void AcceptComplete(OverlappedBuffer* buffer); - virtual void EnsureInitialized(EventHandlerImplementation* event_handler); virtual void DoClose(); virtual bool IsClosed(); @@ -420,12 +398,6 @@ class ListenSocket : public DescriptorInfoMultipleMixin { int accepted_count() { return accepted_count_; } private: - bool LoadAcceptEx(); - bool LoadGetAcceptExSockaddrs(); - - LPFN_ACCEPTEX AcceptEx_; - LPFN_GETACCEPTEXSOCKADDRS GetAcceptExSockaddrs_; - // The number of asynchronous `IssueAccept` operations which haven't completed // yet. int pending_accept_count_; @@ -447,15 +419,11 @@ class ClientSocket : public DescriptorInfoSingleMixin { public: explicit ClientSocket(intptr_t s, std::unique_ptr remote_addr = nullptr) - : DescriptorInfoSingleMixin(s, true), - DisconnectEx_(nullptr), + : DescriptorInfoSingleMixin(s, kClientSocket), next_(nullptr), connected_(false), closed_(false), - remote_addr_(std::move(remote_addr)) { - LoadDisconnectEx(); - type_ = kClientSocket; - } + remote_addr_(std::move(remote_addr)) {} virtual ~ClientSocket() { // Don't delete this object until all pending requests have been handled. @@ -474,7 +442,6 @@ class ClientSocket : public DescriptorInfoSingleMixin { void DisconnectComplete(OverlappedBuffer* buffer); void ConnectComplete(OverlappedBuffer* buffer); - virtual void EnsureInitialized(EventHandlerImplementation* event_handler); virtual void DoClose(); virtual bool IsClosed(); @@ -496,9 +463,6 @@ class ClientSocket : public DescriptorInfoSingleMixin { #endif private: - bool LoadDisconnectEx(); - - LPFN_DISCONNECTEX DisconnectEx_; ClientSocket* next_; bool connected_; bool closed_; @@ -513,9 +477,8 @@ class ClientSocket : public DescriptorInfoSingleMixin { class DatagramSocket : public DescriptorInfoSingleMixin { public: - explicit DatagramSocket(intptr_t s) : DescriptorInfoSingleMixin(s, true) { - type_ = kDatagramSocket; - } + explicit DatagramSocket(intptr_t s) + : DescriptorInfoSingleMixin(s, kDatagramSocket) {} virtual ~DatagramSocket() { // Don't delete this object until all pending requests have been handled. @@ -527,7 +490,6 @@ class DatagramSocket : public DescriptorInfoSingleMixin { virtual bool IssueRecvFrom(); virtual bool IssueSendTo(sockaddr* sa, socklen_t sa_len); - virtual void EnsureInitialized(EventHandlerImplementation* event_handler); virtual void DoClose(); virtual bool IsClosed(); @@ -541,6 +503,8 @@ class EventHandlerImplementation { EventHandlerImplementation(); virtual ~EventHandlerImplementation(); + void AssociateWithCompletionPort(Handle* handle); + void SendData(intptr_t id, Dart_Port dart_port, int64_t data); void Start(EventHandler* handler); void Shutdown(); @@ -570,8 +534,30 @@ class EventHandlerImplementation { HANDLE completion_port() { return completion_port_; } + LPFN_ACCEPTEX accept_ex() { + InitializeSocketExtensions(); + return accept_ex_; + } + + LPFN_CONNECTEX connect_ex() { + InitializeSocketExtensions(); + return connect_ex_; + } + + LPFN_DISCONNECTEX disconnect_ex() { + InitializeSocketExtensions(); + return disconnect_ex_; + } + + LPFN_GETACCEPTEXSOCKADDRS get_accept_ex_sockaddrs() { + InitializeSocketExtensions(); + return get_accept_ex_sockaddrs_; + } + private: - Monitor startup_monitor_; + void InitializeSocketExtensions(); + + Monitor monitor_; ThreadId handler_thread_id_; HANDLE handler_thread_handle_; @@ -579,6 +565,12 @@ class EventHandlerImplementation { bool shutdown_; HANDLE completion_port_; + std::atomic socket_extensions_initialized_{false}; + LPFN_ACCEPTEX accept_ex_ = nullptr; + LPFN_CONNECTEX connect_ex_ = nullptr; + LPFN_DISCONNECTEX disconnect_ex_ = nullptr; + LPFN_GETACCEPTEXSOCKADDRS get_accept_ex_sockaddrs_ = nullptr; + DISALLOW_COPY_AND_ASSIGN(EventHandlerImplementation); }; diff --git a/runtime/bin/file_system_watcher_win.cc b/runtime/bin/file_system_watcher_win.cc index 9ba9f687c54..56925b263c3 100644 --- a/runtime/bin/file_system_watcher_win.cc +++ b/runtime/bin/file_system_watcher_win.cc @@ -59,7 +59,6 @@ intptr_t FileSystemWatcher::WatchPath(intptr_t id, new DirectoryWatchHandle(dir, list_events, recursive); // Issue a read directly, to be sure events are tracked from now on. This is // okay, since in Dart, we create the socket and start reading immediately. - handle->EnsureInitialized(EventHandler::delegate()); handle->IssueRead(); return reinterpret_cast(handle); } diff --git a/runtime/bin/process_win.cc b/runtime/bin/process_win.cc index f1fec905518..44b61e0412c 100644 --- a/runtime/bin/process_win.cc +++ b/runtime/bin/process_win.cc @@ -774,7 +774,15 @@ class OverlappedHandle { private: void ClearOverlapped() { memset(&overlapped_, 0, sizeof(overlapped_)); - overlapped_.hEvent = event_; + // |FileHandle| constructor eagerly associates the given handle with + // |EventHandler|'s completion port. However we don't want to notify + // that completion port when |ReadFile| operation completes because + // we are manually draining the pipe here instead of using |EventHandler|. + // Setting LSB of |hEvent| to 1 prevents completion packets from being + // enqueued. See documentation for |GetQueuedCompletionStatus| (specifically + // notes for |lpOverlapped| argument). + overlapped_.hEvent = + reinterpret_cast(reinterpret_cast(event_) | 0x1); } OVERLAPPED overlapped_; @@ -797,11 +805,11 @@ bool Process::Wait(intptr_t pid, // All pipes created to the sub-process support overlapped IO. FileHandle* stdout_handle = reinterpret_cast(out); - ASSERT(stdout_handle->SupportsOverlappedIO()); + ASSERT(stdout_handle->supports_overlapped_io()); FileHandle* stderr_handle = reinterpret_cast(err); - ASSERT(stderr_handle->SupportsOverlappedIO()); + ASSERT(stderr_handle->supports_overlapped_io()); FileHandle* exit_handle = reinterpret_cast(exit_event); - ASSERT(exit_handle->SupportsOverlappedIO()); + ASSERT(exit_handle->supports_overlapped_io()); // Create three events for overlapped IO. These are created as already // signalled to ensure they have read called at least once. @@ -866,6 +874,7 @@ bool Process::Wait(intptr_t pid, exit_code = -exit_code; } result->set_exit_code(exit_code); + return true; } @@ -984,16 +993,12 @@ intptr_t Process::SetSignalHandler(intptr_t signal) { } MutexLocker lock(signal_mutex); FileHandle* write_handle = new FileHandle(fds[kWriteHandle]); - write_handle->EnsureInitialized(EventHandler::delegate()); intptr_t write_fd = reinterpret_cast(write_handle); if (signal_handlers == nullptr) { if (SetConsoleCtrlHandler(SignalHandler, true) == 0) { int error_code = GetLastError(); - // Since SetConsoleCtrlHandler failed, the IO completion port will - // never receive an event for this handle, and will therefore never - // release the reference Retained by EnsureInitialized(). So, we - // have to do a second Release() here. - write_handle->Release(); + // Since SetConsoleCtrlHandler failed, there will be no subsequent IO + // operation on this handle. Release() it. write_handle->Release(); CloseProcessPipe(fds); SetLastError(error_code); @@ -1020,8 +1025,6 @@ void Process::ClearSignalHandler(intptr_t signal, Dart_Port port) { signal_handlers = handler->next(); } handler->Unlink(); - FileHandle* file_handle = reinterpret_cast(handler->fd()); - file_handle->Release(); remove = true; } } diff --git a/runtime/bin/socket_base_win.cc b/runtime/bin/socket_base_win.cc index a46a2c024e3..966dada4de8 100644 --- a/runtime/bin/socket_base_win.cc +++ b/runtime/bin/socket_base_win.cc @@ -238,8 +238,6 @@ intptr_t SocketBase::GetStdioHandle(intptr_t num) { } StdHandle* std_handle = StdHandle::Stdin(handle); std_handle->Retain(); - std_handle->MarkDoesNotSupportOverlappedIO(); - std_handle->EnsureInitialized(EventHandler::delegate()); return reinterpret_cast(std_handle); } diff --git a/runtime/bin/socket_win.cc b/runtime/bin/socket_win.cc index 52674b51f08..f81e0951983 100644 --- a/runtime/bin/socket_win.cc +++ b/runtime/bin/socket_win.cc @@ -71,7 +71,7 @@ static intptr_t Connect(intptr_t fd, int status = bind(s, &bind_addr.addr, SocketAddress::GetAddrLength(bind_addr)); if (status != NO_ERROR) { - int rc = WSAGetLastError(); + const int rc = WSAGetLastError(); handle->mark_closed(); // Destructor asserts that socket is marked closed. handle->Release(); closesocket(s); @@ -79,34 +79,20 @@ static intptr_t Connect(intptr_t fd, return -1; } - LPFN_CONNECTEX connectEx = nullptr; - GUID guid_connect_ex = WSAID_CONNECTEX; - DWORD bytes; - status = WSAIoctl(s, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid_connect_ex, - sizeof(guid_connect_ex), &connectEx, sizeof(connectEx), - &bytes, nullptr, nullptr); - DWORD rc; - if (status != SOCKET_ERROR) { - handle->EnsureInitialized(EventHandler::delegate()); - - OverlappedBuffer* overlapped = OverlappedBuffer::AllocateConnectBuffer(); - - status = connectEx(s, &addr.addr, SocketAddress::GetAddrLength(addr), - nullptr, 0, nullptr, overlapped->GetCleanOverlapped()); - - if (status == TRUE) { - handle->ConnectComplete(overlapped); - return fd; - } else if (WSAGetLastError() == ERROR_IO_PENDING) { - return fd; - } - rc = WSAGetLastError(); - // Cleanup in case of error. - OverlappedBuffer::DisposeBuffer(overlapped); - handle->Release(); - } else { - rc = WSAGetLastError(); + OverlappedBuffer* overlapped = + OverlappedBuffer::AllocateConnectBuffer(handle); + status = EventHandler::delegate()->connect_ex()( + s, &addr.addr, SocketAddress::GetAddrLength(addr), nullptr, 0, nullptr, + overlapped->GetCleanOverlapped()); + if (status == TRUE) { + handle->ConnectComplete(overlapped); + return fd; + } else if (WSAGetLastError() == ERROR_IO_PENDING) { + return fd; } + const int rc = WSAGetLastError(); + // Cleanup in case of error. + OverlappedBuffer::DisposeBuffer(overlapped); handle->Close(); handle->Release(); SetLastError(rc); @@ -218,8 +204,6 @@ intptr_t Socket::CreateBindDatagram(const RawAddr& addr, } DatagramSocket* datagram_socket = new DatagramSocket(s); - datagram_socket->EnsureInitialized(EventHandler::delegate()); - return reinterpret_cast(datagram_socket); } @@ -294,7 +278,6 @@ intptr_t ServerSocket::CreateUnixDomainBindListen(const RawAddr& addr, bool ServerSocket::StartAccept(intptr_t fd) { ListenSocket* listen_socket = reinterpret_cast(fd); - listen_socket->EnsureInitialized(EventHandler::delegate()); // Always keep 5 outstanding accepts going, to enhance performance. for (int i = 0; i < 5; i++) { if (!listen_socket->IssueAccept()) {