dart-lang/sdk
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

[vm/io,win] Remove Handle::EnsureInitialized

Browse Source 
`Handle` objects which used by Windows specific implementation of
`EventHandler` have complicated life cycle and haphazard approach
to locking. This CL is the first in a series of CLs which attempts
to clean this up a bit - with a final goal to avoid recursively
locking `Handle::monitor_`.

In this CL we remove `Handle::EnsureInitialized` method which
was primarily responsible for associating `HANDLE` with
`EventHandler`'s completion port. Instead we move this logic
into `Handle` constructor.

We also cleanup reference counting logic applied to handles which
participate in asynchronous operations: instead of eagerly
incrementing reference count when `HANDLE` is associated with
the completion port we increment it whenever asynchronous operation
is in progress and decrement it when it completes. This makes the
code much simpler to reason about and avoids necessity to write
code like:

    handle->Release();
    handle->Release();

accompanied with comments explaining why we need to drop
two references due to eager increment.

Finally we cleanup the handling of provider-specific extensions
accessed through `WSAIoctl`'s `SIO_GET_EXTENSION_FUNCTION_POINTER`.
We were fetching them for every individual socket instead of
fetching them once and caching them globally.

TEST=ci

Cq-Include-Trybots: luci.dart.try:vm-win-debug-x64-try,pkg-win-release-try,analyzer-win-release-try,vm-win-release-x64-try
Change-Id: Ia8b391b713b1a1afb9df9e931eada6abf97f7854
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/378140
Reviewed-by: Martin Kustermann <kustermann@google.com>
This commit is contained in:
Slava Egorov
2024-08-06 12:26:19 +00:00
committed by Commit Queue
parent b678794932
commit bbd407725e
10 changed files with 382 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/bin/eventhandler.h
+56 -47
View File
@@ -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 <intptr_t kInitialAmount>
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 <intptr_t kInitialAmount>
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 <typename DI>
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 <typename... Args>
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</*kInitialAmount=*/16> tokens_;
intptr_t mask_;
bool disable_tokens_;
DISALLOW_COPY_AND_ASSIGN(DescriptorInfoSingleMixin);
};
@@ -356,8 +379,6 @@ class DescriptorInfoSingleMixin : public DI {
template <typename DI>
class DescriptorInfoMultipleMixin : public DI {
private:
static constexpr int kTokenCount = 4;
static bool SamePortValue(void* key1, void* key2) {
return reinterpret_cast<Dart_Port>(key1) ==
reinterpret_cast<Dart_Port>(key2);
@@ -383,16 +404,15 @@ class DescriptorInfoMultipleMixin : public DI {
struct PortEntry {
Dart_Port dart_port;
intptr_t is_reading;
intptr_t token_count;
TokenCounter</*kInitialAmount=*/4> 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 <typename... Args>
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<void*>(pentry);
@@ -498,10 +517,8 @@ class DescriptorInfoMultipleMixin : public DI {
PortEntry* pentry = reinterpret_cast<PortEntry*>(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<PortEntry*>(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<PortEntry*> 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);
};
runtime/bin/eventhandler_fuchsia.h
+2 -3
View File
@@ -116,8 +116,7 @@ class DescriptorInfo : public DescriptorInfoBase {
class DescriptorInfoSingle : public DescriptorInfoSingleMixin<DescriptorInfo> {
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<DescriptorInfo> {
public:
explicit DescriptorInfoMultiple(intptr_t fd)
: DescriptorInfoMultipleMixin(fd, false) {}
: DescriptorInfoMultipleMixin(fd) {}
virtual ~DescriptorInfoMultiple() {}
private:
runtime/bin/eventhandler_linux.h
+2 -3
View File
@@ -39,8 +39,7 @@ class DescriptorInfo : public DescriptorInfoBase {
class DescriptorInfoSingle : public DescriptorInfoSingleMixin<DescriptorInfo> {
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<DescriptorInfo> {
public:
explicit DescriptorInfoMultiple(intptr_t fd)
: DescriptorInfoMultipleMixin(fd, false) {}
: DescriptorInfoMultipleMixin(fd) {}
virtual ~DescriptorInfoMultiple() {}
private:
runtime/bin/eventhandler_macos.h
+2 -3
View File
@@ -51,8 +51,7 @@ class DescriptorInfo : public DescriptorInfoBase {
class DescriptorInfoSingle : public DescriptorInfoSingleMixin<DescriptorInfo> {
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<DescriptorInfo> {
public:
explicit DescriptorInfoMultiple(intptr_t fd)
: DescriptorInfoMultipleMixin(fd, false) {}
: DescriptorInfoMultipleMixin(fd) {}
virtual ~DescriptorInfoMultiple() {}
private:
runtime/bin/eventhandler_win.cc
+204 -239
View File
@@ -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<socklen_t*>(GetBufferStart() + GetBufferSize());
*from_len_addr_ = sizeof(struct sockaddr_storage);
from_ = reinterpret_cast<struct sockaddr*>(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>(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<ULONG_PTR>(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<Handle*>(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<ULONG_PTR>(this), overlapped);
ok = PostQueuedCompletionStatus(EventHandler::delegate()->completion_port(),
bytes_read, reinterpret_cast<ULONG_PTR>(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<uword>(this));
int result = Thread::Start(
"dart:io ReadFile",
[](uword args) {
auto handle = reinterpret_cast<Handle*>(args);
handle->ReadSyncCompleteAsync();
handle->Release();
},
reinterpret_cast<uword>(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<RawAddr>(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<intptr_t>(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<StdHandle*>(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<ULONG_PTR>(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<uword>(this));
int result = Thread::Start(
"dart:io WriteFile",
[](uword args) {
auto handle = reinterpret_cast<StdHandle*>(args);
handle->RunWriteLoop();
handle->Release();
},
reinterpret_cast<uword>(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<intptr_t>(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<Handle*>(socket->fd());
ASSERT(handle != nullptr);
handle->Retain();
RefCntReleaseScope<Handle> rh(handle);
if (handle->is_listen_socket()) {
ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(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<Handle*>(key);
RefCntReleaseScope<Handle> release(buffer->StealHandle());
switch (buffer->operation()) {
case OverlappedBuffer::kAccept: {
ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(key);
HandleAccept(listen_socket, buffer);
HandleAccept(static_cast<ListenSocket*>(handle), buffer);
break;
}
case OverlappedBuffer::kRead: {
Handle* handle = reinterpret_cast<Handle*>(key);
HandleRead(handle, bytes, buffer);
break;
}
case OverlappedBuffer::kRecvFrom: {
Handle* handle = reinterpret_cast<Handle*>(key);
HandleRecvFrom(handle, bytes, buffer);
break;
}
case OverlappedBuffer::kWrite:
case OverlappedBuffer::kSendTo: {
Handle* handle = reinterpret_cast<Handle*>(key);
HandleWrite(handle, bytes, buffer);
break;
}
case OverlappedBuffer::kDisconnect: {
ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
HandleDisconnect(client_socket, bytes, buffer);
HandleDisconnect(static_cast<ClientSocket*>(handle), bytes, buffer);
break;
}
case OverlappedBuffer::kConnect: {
ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
HandleConnect(client_socket, bytes, buffer);
HandleConnect(static_cast<ClientSocket*>(handle), bytes, buffer);
break;
}
default:
UNREACHABLE();
}
NotifyDestroyedIfClosed(handle);
}
void EventHandlerImplementation::HandleCompletionOrInterrupt(
@@ -1415,6 +1342,35 @@ EventHandlerImplementation::EventHandlerImplementation() {
shutdown_ = false;
}
namespace {
template <typename F>
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<ULONG_PTR>(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();
}
runtime/bin/eventhandler_win.h
+87 -95
View File
@@ -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<socklen_t*>(GetBufferStart() + GetBufferSize());
*from_len_addr_ = sizeof(struct sockaddr_storage);
from_ = reinterpret_cast<struct sockaddr*>(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<Handle>, 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<Handle>, 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<Handle>, 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<Handle>, 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<Handle>, 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<OverlappedBuffer>
data_ready_; // Buffer for data ready to be read.
@@ -297,13 +275,20 @@ class Handle : public ReferenceCounted<Handle>, public DescriptorInfoBase {
class FileHandle : public DescriptorInfoSingleMixin<Handle> {
public:
explicit FileHandle(HANDLE handle)
: DescriptorInfoSingleMixin(reinterpret_cast<intptr_t>(handle), true) {
type_ = kFile;
}
: DescriptorInfoSingleMixin(reinterpret_cast<intptr_t>(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<intptr_t>(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<Handle> {
public:
DirectoryWatchHandle(HANDLE handle, int events, bool recursive)
: DescriptorInfoSingleMixin(reinterpret_cast<intptr_t>(handle), true),
: DescriptorInfoSingleMixin(reinterpret_cast<intptr_t>(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<SocketHandle> {
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<SocketHandle> {
// 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<SocketHandle> {
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<SocketHandle> {
public:
explicit ClientSocket(intptr_t s,
std::unique_ptr<RawAddr> 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<SocketHandle> {
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<SocketHandle> {
#endif
private:
bool LoadDisconnectEx();
LPFN_DISCONNECTEX DisconnectEx_;
ClientSocket* next_;
bool connected_;
bool closed_;
@@ -513,9 +477,8 @@ class ClientSocket : public DescriptorInfoSingleMixin<SocketHandle> {
class DatagramSocket : public DescriptorInfoSingleMixin<SocketHandle> {
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<SocketHandle> {
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<bool> 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);
};
runtime/bin/file_system_watcher_win.cc
-1
View File
@@ -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<intptr_t>(handle);
}
runtime/bin/process_win.cc
+15 -12
View File
@@ -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<HANDLE>(reinterpret_cast<uintptr_t>(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<FileHandle*>(out);
ASSERT(stdout_handle->SupportsOverlappedIO());
ASSERT(stdout_handle->supports_overlapped_io());
FileHandle* stderr_handle = reinterpret_cast<FileHandle*>(err);
ASSERT(stderr_handle->SupportsOverlappedIO());
ASSERT(stderr_handle->supports_overlapped_io());
FileHandle* exit_handle = reinterpret_cast<FileHandle*>(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<intptr_t>(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<FileHandle*>(handler->fd());
file_handle->Release();
remove = true;
}
}
runtime/bin/socket_base_win.cc
-2
View File
@@ -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<intptr_t>(std_handle);
}
runtime/bin/socket_win.cc
+14 -31
View File
@@ -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<intptr_t>(datagram_socket);
}
@@ -294,7 +278,6 @@ intptr_t ServerSocket::CreateUnixDomainBindListen(const RawAddr& addr,
bool ServerSocket::StartAccept(intptr_t fd) {
ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(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()) {