Implement EventLoopForIO on Windows.

Author: dvander

amio.ambuild

@@ -66,8 +66,10 @@ def Configure(binary):
     ]
   elif builder.target_platform == 'windows':
     binary.sources += [
+      'windows/windows-base-poller.cc',
       'windows/windows-context.cc',
       'windows/windows-errors.cc',
+      'windows/windows-event-loop.cc',
       'windows/windows-file.cc',
       'windows/windows-iocp.cc',
       'windows/windows-net.cc',

include/amio-eventloop.h

@@ -97,7 +97,7 @@ class AMIO_LINK TaskQueue
 };
 
 // An event loop encapsulates a TaskQueue and optionally other polling systems.
-class AMIO_LINK EventLoop : public TaskQueue::Delegate
+class AMIO_LINK EventLoop
 {
  public:
   // Event loops should be freed with |delete|.
@@ -118,15 +118,22 @@ class AMIO_LINK EventLoop : public TaskQueue::Delegate
   // Polls for new events in a loop. The only way to exit the loop is to post
   // a ShouldQuit() message.
   virtual void Loop() = 0;
+
+  // Shuts down the event loop; this is usually called after Loop() exits. It
+  // must not be called from within Loop().
+  virtual void Shutdown() = 0;
 };
 
+#if defined(KE_POSIX)
 // An EventQueue is a wrapper around a poller. Instead of immediately
 // delivering events, it buffers status changes and can deliver them
 // incrementally. This is useful for constructing event loops that
 // prioritize some tasks over others.
 //
-// To use a status queue, simply Poll() and then ask the queue to process
-// events.
+// To use an EventQueue, simply Poll() and then ask the queue to process events.
+// EventQueues are not implemented on Windows. A Poller is effectively an
+// EventQueue already, and users who wish to have multiple may simply use
+// multiple pollers.
 //
 // EventQueues are not thread-safe. All operations must occur on the same
 // thread.
@@ -153,6 +160,7 @@ class EventQueue : public IODispatcher
   // not shutdown the underlying poller.
   virtual void Shutdown() = 0;
 };
+#endif
 
 // An event loop for I/O multiplexing. This is essentially a wrapper around
 // a Poller and a single EventQueue. Tasks are prioritized over events.

include/amio-windows.h

@@ -275,7 +275,8 @@ class AMIO_LINK Transport : public ke::RefcountedThreadsafe<Transport>
   // discard any events received on closed transports.
   // 
   // If you have allocated data associated with a context, it is best to use
-  // IUserData so it is always freed.
+  // IUserData so it is always freed, since any pending IO operations will
+  // not post back their context.
   virtual void Close() = 0;
 
   // Returns true if the handle has been closed.

tests/AMBuild.tests

@@ -26,7 +26,6 @@ runner.compiler.linkflags += [KE.static_lib.binary]
 
 runner.sources += [
   'main.cc',
-  'common/test-event-queues.cc',
   'common/test-event-loops.cc',
   'common/test-network.cc',
   'common/test-server-client.cc',
@@ -39,6 +38,7 @@ if builder.target_platform == 'windows':
   ]
 else:
   runner.sources += [
+    'posix/test-event-queues.cc',
     'posix/test-pipes.cc',
     'posix/test-threading.cc',
   ]

tests/common/test-event-loops.cc

@@ -28,8 +28,8 @@ class PokeThread : public IRunnable
 
   void Run() override {
     me_->Loop();
-    other_->PostQuit();
     Ended = true;
+    other_->PostQuit();
   }
 
   bool Ended;

tests/common/test-server-client.cc

@@ -147,13 +147,19 @@ TestServerClient::Run()
     return false;
 
   server->Close();
+  server = nullptr;
 
   // Try to connect again. We should get an error.
   cli_helper = new ClientHelper();
   if (!Client::Create(&client, poller_, address, Protocol::TCP, cli_helper)) {
     if (!check(client.connection == nullptr, "should not have connected"))
       return false;
 
+#if defined(KE_WINDOWS)
+    // Sleep for one second to give the network stack time to catch up.
+    Sleep(1000);
+#endif
+
     if (!check_error(poller_->Poll(), "poll for error"))
       return false;
 

tests/common/test-event-queues.cc renamed to tests/posix/test-event-queues.cc

@@ -113,4 +113,11 @@ WinBasePoller::unlink_impl(WinContext *context)
   // Undo everything we did in link().
   context->detach();
   context->Release();
-}
+}
+
+AlreadyRefed<WinContext>
+WinBasePoller::take(WinContext *context)
+{
+  context->detach();
+  return AlreadyRefed<WinContext>(context);
+}

windows/windows-base-poller.cc

@@ -50,17 +50,18 @@ class WinBasePoller
   void link(WinContext *context, const T &object, RequestType type) {
     if (object)
       object->AddRef();
-
     this->link_impl(context, type);
   }
   template <typename T>
   void unlink(WinContext *context, const T &object) {
     this->unlink_impl(context);
-
     if (object)
       object->Release();
   }
 
+  // Special version just for Poll().
+  AlreadyRefed<WinContext> take(WinContext *context);
+
  private:
   void link_impl(WinContext *context, RequestType type);
   void unlink_impl(WinContext *context);

windows/windows-base-poller.h

@@ -0,0 +1,130 @@
+// vim: set ts=2 sw=2 tw=99 et:
+// 
+// Copyright (C) 2014 David Anderson
+// 
+// This file is part of the AlliedModders I/O Library.
+// 
+// The AlliedModders I/O library is licensed under the GNU General Public
+// License, version 3 or higher. For more information, see LICENSE.txt
+//
+#include "windows-event-loop.h"
+#include <stdio.h>
+
+using namespace ke;
+using namespace amio;
+
+PassRef<IOError>
+EventLoopForIO::Create(Ref<EventLoopForIO> *outp, Ref<Poller> poller)
+{
+  if (!poller) {
+    if (Ref<IOError> error = PollerFactory::Create(&poller))
+      return error;
+  }
+
+  *outp = new WindowsEventLoopForIO(poller);
+  return nullptr;
+}
+
+WindowsEventLoopForIO::WindowsEventLoopForIO(Ref<Poller> poller)
+ : poller_(poller),
+   tasks_(this),
+   wakeup_(new Wakeup()),
+   received_wakeup_(false)
+{
+}
+
+void
+WindowsEventLoopForIO::Shutdown()
+{
+  wakeup_ = nullptr;
+  poller_ = nullptr;
+}
+
+WindowsEventLoopForIO::Wakeup::Wakeup()
+{
+  context_ = IOContext::New();
+}
+
+void
+WindowsEventLoopForIO::Wakeup::OnCompleted(IOResult &r)
+{
+  // Just take the context back.
+  context_ = r.context;
+}
+
+void
+WindowsEventLoopForIO::Wakeup::Signal(Ref<Poller> poller)
+{
+  // If the context is already in the completion port, then we'll always be
+  // able to dequeue something, so we don't need to post a wakeup message.
+  if (!context_)
+    return;
+
+  if (Ref<IOError> error = poller->Post(context_, this)) {
+    fprintf(stderr, "could not post wakeup: %s\n", error->Message());
+    return;
+  }
+
+  context_ = nullptr;
+}
+
+PassRef<Poller>
+WindowsEventLoopForIO::GetPoller()
+{
+  return poller_;
+}
+
+PassRef<IOError>
+WindowsEventLoopForIO::Attach(Ref<Transport> transport, Ref<IOListener> listener)
+{
+  return poller_->Attach(transport, listener);
+}
+
+void
+WindowsEventLoopForIO::PostTask(Task *task)
+{
+  tasks_.PostTask(task);
+}
+
+void
+WindowsEventLoopForIO::PostQuit()
+{
+  tasks_.PostQuit();
+}
+
+bool
+WindowsEventLoopForIO::ShouldQuit()
+{
+  return tasks_.ShouldQuit();
+}
+
+void
+WindowsEventLoopForIO::Loop()
+{
+  while (!ShouldQuit()) {
+    if (tasks_.ProcessNextTask())
+      continue;
+
+    received_wakeup_ = false;
+
+    if (Ref<IOError> error = poller_->PollOne()) {
+      fprintf(stderr, "Could not poll: %s\n", error->Message());
+      continue;
+    }
+
+    if (received_wakeup_)
+      continue;
+  }
+}
+
+void
+WindowsEventLoopForIO::NotifyTask()
+{
+  wakeup_->Signal(poller_);
+}
+
+void
+WindowsEventLoopForIO::NotifyQuit()
+{
+  wakeup_->Signal(poller_);
+}

windows/windows-event-loop.cc

@@ -0,0 +1,69 @@
+// vim: set ts=2 sw=2 tw=99 et:
+// 
+// Copyright (C) 2014 David Anderson
+// 
+// This file is part of the AlliedModders I/O Library.
+// 
+// The AlliedModders I/O library is licensed under the GNU General Public
+// License, version 3 or higher. For more information, see LICENSE.txt
+//
+#ifndef _include_amio_windows_event_loop_h_
+#define _include_amio_windows_event_loop_h_
+
+#include <amio.h>
+#include <amio-eventloop.h>
+#include "../shared/shared-task-queue.h"
+
+namespace amio {
+
+using namespace ke;
+
+class WindowsEventLoopForIO
+ : public EventLoopForIO,
+   public TaskQueue::Delegate,
+   public RefcountedThreadsafe<WindowsEventLoopForIO>
+{
+ public:
+  WindowsEventLoopForIO(Ref<Poller> poller);
+
+  KE_IMPL_REFCOUNTING_TS(WindowsEventLoopForIO);
+
+  PassRef<Poller> GetPoller() override;
+
+  PassRef<IOError> Attach(Ref<Transport> transport, Ref<IOListener> listener) override;
+
+  void PostTask(Task *task) override;
+  void PostQuit() override;
+  bool ShouldQuit() override;
+  void Loop() override;
+  void Shutdown() override;
+
+  void NotifyTask() override;
+  void NotifyQuit() override;
+
+ private:
+  class Wakeup
+   : public IOListener,
+     public RefcountedThreadsafe<Wakeup>
+  {
+   public:
+    Wakeup();
+    KE_IMPL_REFCOUNTING_TS(Wakeup);
+
+    void OnCompleted(IOResult &r);
+    void Signal(Ref<Poller> poller);
+
+   private:
+    Ref<IOContext> context_;
+  };
+
+ private:
+  Ref<Poller> poller_;
+  TaskQueueImpl tasks_;
+  Ref<Wakeup> wakeup_;
+  bool received_wakeup_;
+};
+
+} // namespace amio
+
+#endif // _include_amio_windows_event_loop_h_

windows/windows-event-loop.h

@@ -144,6 +144,10 @@ CompletionPort::Dispatch(WinContext *context, OVERLAPPED_ENTRY &entry, DWORD err
 {
   Ref<IOListener> listener;
 
+  IOResult result;
+  result.bytes = entry.dwNumberOfBytesTransferred;
+  result.completed = true;
+
   // We take the lock at this point - since we could race with changeListener
   // or WinContext::cancel().
   RequestType request;
@@ -160,8 +164,8 @@ CompletionPort::Dispatch(WinContext *context, OVERLAPPED_ENTRY &entry, DWORD err
 
         // If we have a transport and it's closed - or the IO operation was
         // cancelled - just leave now. Note we always grab the transport even
-        // if the state was Cancelled, since we have to free the ref.
-        if (request == RequestType::Cancelled)
+        // if the state was Cancelled, since we have to acquire its ref.
+        if (transport->Closed() || request == RequestType::Cancelled)
           return false;
 
         // Would be much nicer if we could use RtlNtStatusToDosError() here, but
@@ -181,15 +185,11 @@ CompletionPort::Dispatch(WinContext *context, OVERLAPPED_ENTRY &entry, DWORD err
     }
 
     // Detach the context. After this point, we can return whenever we want.
-    // Note that the object has already been refed above, so we always pass
-    // nullptr here.
-    WinBasePoller::unlink(context, static_cast<IRefcounted *>(nullptr));
+    // Note that the object may have exactly one ref at this point, so it's
+    // important that it goes right into a Refed location.
+    result.context = WinBasePoller::take(context);
   }
 
-  IOResult result;
-  result.bytes = entry.dwNumberOfBytesTransferred;
-  result.completed = true;
-  result.context = context;
   if (error) {
     if (error == ERROR_HANDLE_EOF)
       result.ended = true;