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 32 matching lines @@
   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) {
+  // 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);
+}
+
+
 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);
+  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));
+  }
+  // Always dispose of the buffer, as UDP messages must be read in their
+  // entirety to match how recvfrom works in a socket.
+  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;
+  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 114 matching lines @@
 
 
 void EventHandlerImplementation::Shutdown() {
   SendData(kShutdownId, 0, 0);
 }
 
 }  // namespace bin
 }  // namespace dart
 
 #endif  // defined(TARGET_OS_WINDOWS)