Add UDP support to dart:io

runtime/bin/eventhandler_win.cc

 // Copyright (c) 2013, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE file.
 
 #include "platform/globals.h"
 #if defined(TARGET_OS_WINDOWS)
 
 #include "bin/eventhandler.h"
 
 #include <winsock2.h>  // NOLINT
@@ skipping 15 matching lines @@
 
 static const int kBufferSize = 64 * 1024;
 static const int kStdOverlappedBufferSize = 16 * 1024;
 
 static const int kInfinityTimeout = -1;
 static const int kTimeoutId = -1;
 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;
 }
 
 
 OverlappedBuffer* OverlappedBuffer::AllocateAcceptBuffer(int buffer_size) {
   OverlappedBuffer* buffer = AllocateBuffer(buffer_size, kAccept);
   return buffer;
 }
 
 
 OverlappedBuffer* OverlappedBuffer::AllocateReadBuffer(int buffer_size) {
   return AllocateBuffer(buffer_size, kRead);
 }
 
 
+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);
 }
 
 
 void OverlappedBuffer::DisposeBuffer(OverlappedBuffer* buffer) {
   delete buffer;
 }
 
 
@@ skipping 16 matching lines @@
 
 int OverlappedBuffer::Write(const void* buffer, int num_bytes) {
   ASSERT(num_bytes == buflen_);
   memmove(GetBufferStart(), buffer, num_bytes);
   data_length_ = num_bytes;
   return num_bytes;
 }
 
 
 int OverlappedBuffer::GetRemainingLength() {
-  ASSERT(operation_ == kRead);
+  ASSERT(operation_ == kRead || operation_ == kRecvFrom);
   return data_length_ - index_;
 }
 
 
 Handle::Handle(HANDLE handle)
     : handle_(reinterpret_cast<HANDLE>(handle)),
       port_(0),
       mask_(0),
       completion_port_(INVALID_HANDLE_VALUE),
       event_handler_(NULL),
@@ skipping 87 matching lines @@
   ASSERT(data_ready_ == NULL);
   if (!IsClosing() && !buffer->IsEmpty()) {
     data_ready_ = pending_read_;
   } else {
     OverlappedBuffer::DisposeBuffer(buffer);
   }
   pending_read_ = NULL;
 }
 
 
+void Handle::RecvFromComplete(OverlappedBuffer* buffer) {
+  ReadComplete(buffer);
+}
+
+
 void Handle::WriteComplete(OverlappedBuffer* buffer) {
   ScopedLock lock(this);
   // Currently only one outstanding write at the time.
   ASSERT(pending_write_ == buffer);
   OverlappedBuffer::DisposeBuffer(buffer);
   pending_write_ = NULL;
 }
 
 
 static void ReadFileThread(uword args) {
@@ skipping 60 matching lines @@
     int result = dart::Thread::Start(ReadFileThread,
                                      reinterpret_cast<uword>(this));
     if (result != 0) {
       FATAL1("Failed to start read file thread %d", result);
     }
     return true;
   }
 }
 
 
+bool Handle::IssueRecvFrom() {
+  return false;
+}
+
+
 bool Handle::IssueWrite() {
   ScopedLock lock(this);
   ASSERT(type_ != kListenSocket);
   ASSERT(completion_port_ != INVALID_HANDLE_VALUE);
   ASSERT(pending_write_ != NULL);
   ASSERT(pending_write_->operation() == OverlappedBuffer::kWrite);
 
   OverlappedBuffer* buffer = pending_write_;
   BOOL ok = WriteFile(handle_,
                       buffer->GetBufferStart(),
                       buffer->GetBufferSize(),
                       NULL,
                       buffer->GetCleanOverlapped());
   if (ok || GetLastError() == ERROR_IO_PENDING) {
     // Completing asynchronously.
     pending_write_ = buffer;
     return true;
   }
   OverlappedBuffer::DisposeBuffer(buffer);
   HandleIssueError();
   return false;
 }
 
 
+bool Handle::IssueSendTo(struct sockaddr* sa, socklen_t sa_len) {
+  return false;
+}
+
+
 void Handle::HandleIssueError() {
   DWORD error = GetLastError();
   if (error == ERROR_BROKEN_PIPE) {
     event_handler_->HandleClosed(this);
   } else {
     event_handler_->HandleError(this);
   }
   SetLastError(error);
 }
 
@@ skipping 220 matching lines @@
   if (data_ready_ == NULL) return 0;
   num_bytes = data_ready_->Read(buffer, num_bytes);
   if (data_ready_->IsEmpty()) {
     OverlappedBuffer::DisposeBuffer(data_ready_);
     data_ready_ = NULL;
   }
   return 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;
   if (num_bytes > kBufferSize) num_bytes = kBufferSize;
   ASSERT(SupportsOverlappedIO());
   if (completion_port_ == INVALID_HANDLE_VALUE) return 0;
   pending_write_ = OverlappedBuffer::AllocateWriteBuffer(num_bytes);
   pending_write_->Write(buffer, num_bytes);
   if (!IssueWrite()) return -1;
   return 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();
 }
 
 
 void StdHandle::RunWriteLoop() {
   write_monitor_->Enter();
   write_thread_running_ = true;
   // Notify we have started.
@@ skipping 131 matching lines @@
   shutdown(socket(), SD_BOTH);
   IssueDisconnect();
 }
 
 
 bool ClientSocket::IssueRead() {
   ScopedLock lock(this);
   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;
   int rc = WSARecv(socket(),
                    buffer->GetWASBUF(),
                    1,
                    NULL,
                    &flags,
                    buffer->GetCleanOverlapped(),
                    NULL);
@@ skipping 63 matching lines @@
     CreateCompletionPort(event_handler_->completion_port());
   }
 }
 
 
 bool ClientSocket::IsClosed() {
   return false;
 }
 
 
+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();
     delete handle;
     DartUtils::PostInt32(port, 1 << kDestroyedEvent);
   }
 }
 
 
 void EventHandlerImplementation::HandleInterrupt(InterruptMessage* msg) {
@@ skipping 47 matching lines @@
           // for read.
           if ((msg->data & (1 << kInEvent)) != 0) {
             if (handle->Available() > 0) {
               int event_mask = (1 << kInEvent);
               handle->set_mask(handle->mask() & ~event_mask);
               DartUtils::PostInt32(handle->port(), event_mask);
             } else if (handle->IsClosedRead()) {
               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();
+              }
             }
           }
 
           // If out events (can write events) have been requested, and there
           // are no pending writes, post an out event immediately.
           if ((msg->data & (1 << kOutEvent)) != 0) {
             if (!handle->HasPendingWrite()) {
               int event_mask = (1 << kOutEvent);
               handle->set_mask(handle->mask() & ~event_mask);
               DartUtils::PostInt32(handle->port(), event_mask);
@@ skipping 74 matching lines @@
       HandleClosed(handle);
     } else {
       HandleError(handle);
     }
   }
 
   DeleteIfClosed(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) {
   handle->WriteComplete(buffer);
 
   if (bytes > 0) {
     if (!handle->IsError() && !handle->IsClosing()) {
       int event_mask = 1 << kOutEvent;
       if ((handle->mask() & event_mask) != 0) {
         DartUtils::PostInt32(handle->port(), event_mask);
@@ skipping 31 matching lines @@
     case OverlappedBuffer::kAccept: {
       ListenSocket* listen_socket = reinterpret_cast<ListenSocket*>(key);
       HandleAccept(listen_socket, buffer);
       break;
     }
     case OverlappedBuffer::kRead: {
       Handle* handle = reinterpret_cast<Handle*>(key);
       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;
     }
     case OverlappedBuffer::kDisconnect: {
       ClientSocket* client_socket = reinterpret_cast<ClientSocket*>(key);
       HandleDisconnect(client_socket, bytes, buffer);
       break;
     }
     default:
@@ skipping 76 matching lines @@
       // ERROR_NETNAME_DELETED occurs when the client closes
       // the socket it is reading from.
       DWORD last_error = GetLastError();
       if (last_error == ERROR_CONNECTION_ABORTED ||
           last_error == ERROR_OPERATION_ABORTED ||
           last_error == ERROR_NETNAME_DELETED ||
           last_error == ERROR_BROKEN_PIPE) {
         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);
       }
     } else if (key == NULL) {
       // A key of NULL signals an interrupt message.
       InterruptMessage* msg = reinterpret_cast<InterruptMessage*>(overlapped);
       handler_impl->HandleInterrupt(msg);
       delete msg;
     } else {
       handler_impl->HandleIOCompletion(bytes, key, overlapped);
@@ skipping 18 matching lines @@
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)