Add UDP support to dart:io

runtime/bin/eventhandler_win.cc

Index: runtime/bin/eventhandler_win.cc
diff --git a/runtime/bin/eventhandler_win.cc b/runtime/bin/eventhandler_win.cc
index 334d496ee677915af4ca78d7be0103fec852e6fb..28add7ffd4c9c3a515a17a19d42a1100d94dd732 100644
--- a/runtime/bin/eventhandler_win.cc
+++ b/runtime/bin/eventhandler_win.cc
@@ -33,8 +33,14 @@ static const int kShutdownId = -2;
 
 OverlappedBuffer* OverlappedBuffer::AllocateBuffer(int buffer_size,
                                                    Operation operation) {
+  int actual_size = buffer_size;
+  // For calling recvfrom additional buffer space is needed for the source
+  // address information.
+  if (operation == kRecvFrom) {

[Anders Johnsen, 2013/12/12 12:19:58]

Move this to AllocateRecvFromBuffer?

[Søren Gjesse, 2013/12/12 15:44:39]

Done.

+    actual_size += sizeof(socklen_t) + sizeof(struct sockaddr_storage);

[Anders Johnsen, 2013/12/12 12:19:58]

Make the size a constant in OverlappedBuffer.

[Søren Gjesse, 2013/12/12 15:44:39]

Done.

+  }
   OverlappedBuffer* buffer =
-      new(buffer_size) OverlappedBuffer(buffer_size, operation);
+      new(actual_size) OverlappedBuffer(actual_size, operation);
   return buffer;
 }
 
@@ -50,11 +56,21 @@ OverlappedBuffer* OverlappedBuffer::AllocateReadBuffer(int buffer_size) {
 }
 
 
+OverlappedBuffer* OverlappedBuffer::AllocateRecvFromBuffer(int buffer_size) {
+  return AllocateBuffer(buffer_size, kRecvFrom);
+}
+
+
 OverlappedBuffer* OverlappedBuffer::AllocateWriteBuffer(int buffer_size) {
   return AllocateBuffer(buffer_size, kWrite);
 }
 
 
+OverlappedBuffer* OverlappedBuffer::AllocateSendToBuffer(int buffer_size) {
+  return AllocateBuffer(buffer_size, kSendTo);
+}
+
+
 OverlappedBuffer* OverlappedBuffer::AllocateDisconnectBuffer() {
   return AllocateBuffer(0, kDisconnect);
 }
@@ -91,7 +107,7 @@ int OverlappedBuffer::Write(const void* buffer, int num_bytes) {
 
 
 int OverlappedBuffer::GetRemainingLength() {
-  ASSERT(operation_ == kRead);
+  ASSERT(operation_ == kRead || operation_ == kRecvFrom);
   return data_length_ - index_;
 }
 
@@ -199,6 +215,11 @@ void Handle::ReadComplete(OverlappedBuffer* buffer) {
 }
 
 
+void Handle::RecvFromComplete(OverlappedBuffer* buffer) {
+  ReadComplete(buffer);
+}
+
+
 void Handle::WriteComplete(OverlappedBuffer* buffer) {
   ScopedLock lock(this);
   // Currently only one outstanding write at the time.
@@ -279,6 +300,11 @@ bool Handle::IssueRead() {
 }
 
 
+bool Handle::IssueRecvFrom() {
+  return false;
+}
+
+
 bool Handle::IssueWrite() {
   ScopedLock lock(this);
   ASSERT(type_ != kListenSocket);
@@ -303,6 +329,11 @@ bool Handle::IssueWrite() {
 }
 
 
+bool Handle::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
+  return false;
+}
+
+
 void Handle::HandleIssueError() {
   DWORD error = GetLastError();
   if (error == ERROR_BROKEN_PIPE) {
@@ -543,6 +574,27 @@ int Handle::Read(void* buffer, int num_bytes) {
 }
 
 
+int Handle::RecvFrom(
+    void* buffer, int num_bytes, struct sockaddr* sa, socklen_t sa_len) {
+  ScopedLock lock(this);
+  if (data_ready_ == NULL) return 0;
+  num_bytes = data_ready_->Read(buffer, num_bytes);

[Anders Johnsen, 2013/12/12 12:19:58]

Will this allow us to read only a part of a UDP packet?

[Søren Gjesse, 2013/12/12 15:44:39]

It will, I changed it to dispose of the rest, so that the next call to RecvFrom
should always get the next datagram or nothing. This is how recvfrom on Linux
and Mac OS works if the buffer is smaller than the datagram.

+  if (data_ready_->from()->sa_family == AF_INET) {
+    ASSERT(sa_len >= sizeof(struct sockaddr_in));
+    memmove(sa, data_ready_->from(), sizeof(struct sockaddr_in));
+  } else {
+    ASSERT(data_ready_->from()->sa_family == AF_INET6);
+    ASSERT(sa_len >= sizeof(struct sockaddr_in6));
+    memmove(sa, data_ready_->from(), sizeof(struct sockaddr_in6));
+  }
+  if (data_ready_->IsEmpty()) {
+    OverlappedBuffer::DisposeBuffer(data_ready_);
+    data_ready_ = NULL;
+  }
+  return num_bytes;
+}
+
+
 int Handle::Write(const void* buffer, int num_bytes) {
   ScopedLock lock(this);
   if (pending_write_ != NULL) return 0;
@@ -556,6 +608,20 @@ int Handle::Write(const void* buffer, int num_bytes) {
 }
 
 
+int Handle::SendTo(
+    const void* buffer, int num_bytes, struct sockaddr* sa, socklen_t sa_len) {
+  ScopedLock lock(this);
+  if (pending_write_ != NULL) return 0;
+  if (num_bytes > kBufferSize) num_bytes = kBufferSize;
+  ASSERT(SupportsOverlappedIO());
+  if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
+  pending_write_ = OverlappedBuffer::AllocateSendToBuffer(num_bytes);
+  pending_write_->Write(buffer, num_bytes);
+  if (!IssueSendTo(sa, sa_len)) return -1;

[Anders Johnsen, 2013/12/12 12:19:58]

Can this leak the pending_write?

[Søren Gjesse, 2013/12/12 15:44:39]

No, IssueSendTo will dispose of it before returning false (same for Write
above).

+  return num_bytes;
+}
+
+
 static void WriteFileThread(uword args) {
   StdHandle* handle = reinterpret_cast<StdHandle*>(args);
   handle->RunWriteLoop();
@@ -707,7 +773,9 @@ bool ClientSocket::IssueRead() {
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_read_ == NULL);
 
-  OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(1024);
+  // TODO(sgjesse): Use a MTU value here. Only the loopback adapter can
+  // handle 64k datagrams.
+  OverlappedBuffer* buffer = OverlappedBuffer::AllocateReadBuffer(65536);
 
   DWORD flags;
   flags = 0;
@@ -791,6 +859,85 @@ bool ClientSocket::IsClosed() {
 }
 
 
+bool DatagramSocket::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
+  ScopedLock lock(this);
+  ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
+  ASSERT(pending_write_ != NULL);
+  ASSERT(pending_write_->operation() == OverlappedBuffer::kSendTo);
+
+  int rc = WSASendTo(socket(),
+                     pending_write_->GetWASBUF(),
+                     1,
+                     NULL,
+                     0,
+                     sa,
+                     sa_len,
+                     pending_write_->GetCleanOverlapped(),
+                     NULL);
+  if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
+    return true;
+  }
+  OverlappedBuffer::DisposeBuffer(pending_write_);
+  pending_write_ = NULL;
+  HandleIssueError();
+  return false;
+}
+
+
+bool DatagramSocket::IssueRecvFrom() {
+  ScopedLock lock(this);
+  ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
+  ASSERT(pending_read_ == NULL);
+
+  OverlappedBuffer* buffer = OverlappedBuffer::AllocateRecvFromBuffer(1024);
+
+  DWORD flags;
+  flags = 0;
+  int len;
+  int rc = WSARecvFrom(socket(),
+                       buffer->GetWASBUF(),
+                       1,
+                       NULL,
+                       &flags,
+                       buffer->from(),
+                       buffer->from_len_addr(),
+                       buffer->GetCleanOverlapped(),
+                       NULL);
+  if (rc == NO_ERROR || WSAGetLastError() == WSA_IO_PENDING) {
+    pending_read_ = buffer;
+    return true;
+  }
+  OverlappedBuffer::DisposeBuffer(buffer);
+  pending_read_ = NULL;
+  HandleIssueError();
+  return false;
+}
+
+
+void DatagramSocket::EnsureInitialized(
+    EventHandlerImplementation* event_handler) {
+  ScopedLock lock(this);
+  if (completion_port_ == INVALID_HANDLE_VALUE) {
+    ASSERT(event_handler_ == NULL);
+    event_handler_ = event_handler;
+    CreateCompletionPort(event_handler_->completion_port());
+  }
+}
+
+
+bool DatagramSocket::IsClosed() {
+  return IsClosing() && !HasPendingRead() && !HasPendingWrite();
+}
+
+
+void DatagramSocket::DoClose() {
+  // Just close the socket. This will cause any queued requests to be aborted.
+  closesocket(socket());
+  MarkClosedRead();
+  MarkClosedWrite();
+}
+
+
 static void DeleteIfClosed(Handle* handle) {
   if (handle->IsClosed()) {
     Dart_Port port = handle->port();
@@ -858,7 +1005,11 @@ void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
               int event_mask = (1 << kCloseEvent);
               DartUtils::PostInt32(handle->port(), event_mask);
             } else if (!handle->HasPendingRead()) {
-              handle->IssueRead();
+              if (handle->is_datagram_socket()) {
+                handle->IssueRecvFrom();
+              } else {
+                handle->IssueRead();
+              }
             }
           }
 
@@ -953,6 +1104,23 @@ void EventHandlerImplementation::HandleRead(Handle* handle,
 }
 
 
+void EventHandlerImplementation::HandleRecvFrom(Handle* handle,
+                                                int bytes,
+                                                OverlappedBuffer* buffer) {
+  ASSERT(handle->is_datagram_socket());
+  buffer->set_data_length(bytes);
+  handle->ReadComplete(buffer);
+  if (!handle->IsClosing()) {
+    int event_mask = 1 << kInEvent;
+    if ((handle->mask() & event_mask) != 0) {
+      DartUtils::PostInt32(handle->port(), event_mask);
+    }
+  }
+
+  DeleteIfClosed(handle);
+}
+
+
 void EventHandlerImplementation::HandleWrite(Handle* handle,
                                              int bytes,
                                              OverlappedBuffer* buffer) {
@@ -1004,7 +1172,13 @@ void EventHandlerImplementation::HandleIOCompletion(DWORD bytes,
       HandleRead(handle, bytes, buffer);
       break;
     }
-    case OverlappedBuffer::kWrite: {
+    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;
@@ -1101,6 +1275,8 @@ void EventHandlerImplementation::EventHandlerEntry(uword args) {
         ASSERT(bytes == 0);
         handler_impl->HandleIOCompletion(bytes, key, overlapped);
       } else {
+        OverlappedBuffer* buffer =

[Anders Johnsen, 2013/12/12 12:19:58]

Unused?

[Søren Gjesse, 2013/12/12 15:44:39]

Yes, removed.

+            OverlappedBuffer::GetFromOverlapped(overlapped);
         ASSERT(bytes == 0);
         handler_impl->HandleIOCompletion(-1, key, overlapped);
       }