POSIX: Get IPCSyncChannel unittests working

chrome/common/ipc_sync_channel_unittest.cc

 // Copyright (c) 2006-2008 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 //
 // Unit test for SyncChannel.
 
-#include <windows.h>
 #include <string>
 #include <vector>
 
 #include "base/basictypes.h"
 #include "base/logging.h"
 #include "base/message_loop.h"
+#include "base/platform_thread.h"
 #include "base/string_util.h"
 #include "base/thread.h"
+#include "base/waitable_event.h"
 #include "chrome/common/child_process.h"
 #include "chrome/common/ipc_message.h"
 #include "chrome/common/ipc_sync_channel.h"
 #include "chrome/common/stl_util-inl.h"
 #include "testing/gtest/include/gtest/gtest.h"
 
 #define IPC_MESSAGE_MACROS_ENUMS
 #include "chrome/common/ipc_sync_channel_unittest.h"
 
 // define the classes
 #define IPC_MESSAGE_MACROS_CLASSES
 #include "chrome/common/ipc_sync_channel_unittest.h"
 
 using namespace IPC;
+using base::WaitableEvent;
 
 namespace {
 
 class IPCSyncChannelTest : public testing::Test {
  private:
   MessageLoop message_loop_;
 };
 
 // SyncChannel should only be used in child processes as we don't want to hang
 // the browser.  So in the unit test we need to have a ChildProcess object.
 class TestProcess : public ChildProcess {
  public:
   explicit TestProcess(const std::wstring& channel_name) {}
   static void GlobalInit() {
     ChildProcessFactory<TestProcess> factory;
     ChildProcess::GlobalInit(L"blah", &factory);
   }
 };
 
-// Wrapper around an event handle.
-class Event {
- public:
-  Event() : handle_(CreateEvent(NULL, FALSE, FALSE, NULL)) { }
-  ~Event() { CloseHandle(handle_); }
-  void Set() { SetEvent(handle_); }
-  void Wait() { WaitForSingleObject(handle_, INFINITE); }
-  HANDLE handle() { return handle_; }
-
- private:
-  HANDLE handle_;
-
-  DISALLOW_EVIL_CONSTRUCTORS(Event);
-};
-
 // Base class for a "process" with listener and IPC threads.
 class Worker : public Channel::Listener, public Message::Sender {
  public:
   // Will create a channel without a name.
   Worker(Channel::Mode mode, const std::string& thread_name)
-      : channel_name_(),
+      : done_(new WaitableEvent(false, false)),
+        channel_created_(new WaitableEvent(false, false)),
         mode_(mode),
         ipc_thread_((thread_name + "_ipc").c_str()),
         listener_thread_((thread_name + "_listener").c_str()),
         overrided_thread_(NULL) { }
 
   // Will create a named channel and use this name for the threads' name.
   Worker(const std::wstring& channel_name, Channel::Mode mode)
-      : channel_name_(channel_name),
+      : done_(new WaitableEvent(false, false)),
+        channel_created_(new WaitableEvent(false, false)),
+        channel_name_(channel_name),
         mode_(mode),
         ipc_thread_((WideToUTF8(channel_name) + "_ipc").c_str()),
         listener_thread_((WideToUTF8(channel_name) + "_listener").c_str()),
         overrided_thread_(NULL) { }
 
   // The IPC thread needs to outlive SyncChannel, so force the correct order of
   // destruction.
   virtual ~Worker() {
-    Event listener_done, ipc_done;
+    WaitableEvent listener_done(false, false), ipc_done(false, false);
     ListenerThread()->message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
-        this, &Worker::OnListenerThreadShutdown, listener_done.handle(),
-        ipc_done.handle()));
-    HANDLE handles[] = { listener_done.handle(), ipc_done.handle() };
-    WaitForMultipleObjects(2, handles, TRUE, INFINITE);
+        this, &Worker::OnListenerThreadShutdown, &listener_done,
+        &ipc_done));
+    listener_done.Wait();
+    ipc_done.Wait();
     ipc_thread_.Stop();
     listener_thread_.Stop();
   }
   void AddRef() { }
   void Release() { }
   bool Send(Message* msg) { return channel_->Send(msg); }
   bool SendWithTimeout(Message* msg, int timeout_ms) {
     return channel_->SendWithTimeout(msg, timeout_ms);
   }
-  void WaitForChannelCreation() { channel_created_.Wait(); }
+  void WaitForChannelCreation() { channel_created_->Wait(); }
   void CloseChannel() {
     DCHECK(MessageLoop::current() == ListenerThread()->message_loop());
     channel_->Close();
   }
   void Start() {
     StartThread(&listener_thread_, MessageLoop::TYPE_DEFAULT);
     ListenerThread()->message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
         this, &Worker::OnStart));
   }
   void OverrideThread(base::Thread* overrided_thread) {
@@ skipping 14 matching lines @@
     int answer = 0;
     SyncMessage* msg = new SyncChannelTestMsg_Double(5, &answer);
     if (pump)
       msg->EnableMessagePumping();
     bool result = Send(msg);
     DCHECK(result == succeed);
     DCHECK(answer == (succeed ? 10 : 0));
     return result;
   }
   Channel::Mode mode() { return mode_; }
-  HANDLE done_event() { return done_.handle(); }
+  WaitableEvent* done_event() { return done_.get(); }
 
  protected:
   // Derived classes need to call this when they've completed their part of
   // the test.
-  void Done() { done_.Set(); }
+  void Done() { done_->Signal(); }
   // Functions for dervied classes to implement if they wish.
   virtual void Run() { }
   virtual void OnAnswer(int* answer) { NOTREACHED(); }
   virtual void OnAnswerDelay(Message* reply_msg) {
     // The message handler map below can only take one entry for
     // SyncChannelTestMsg_AnswerToLife, so since some classes want
     // the normal version while other want the delayed reply, we
     // call the normal version if the derived class didn't override
     // this function.
     int answer;
@@ skipping 13 matching lines @@
   base::Thread* ListenerThread() {
     return overrided_thread_ ? overrided_thread_ : &listener_thread_;
   }
   // Called on the listener thread to create the sync channel.
   void OnStart() {
     // Link ipc_thread_, listener_thread_ and channel_ altogether.
     StartThread(&ipc_thread_, MessageLoop::TYPE_IO);
     channel_.reset(new SyncChannel(
         channel_name_, mode_, this, NULL, ipc_thread_.message_loop(), true,
         TestProcess::GetShutDownEvent()));
-    channel_created_.Set();
+    channel_created_->Signal();
     Run();
   }
 
-  void OnListenerThreadShutdown(HANDLE listener_event, HANDLE ipc_event) {
+  void OnListenerThreadShutdown(WaitableEvent* listener_event,
+                                WaitableEvent* ipc_event) {
     // SyncChannel needs to be destructed on the thread that it was created on.
     channel_.reset();
-    SetEvent(listener_event);
-    
+    listener_event->Signal();
+
     ipc_thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
         this, &Worker::OnIPCThreadShutdown, ipc_event));
   }
 
-  void OnIPCThreadShutdown(HANDLE ipc_event) {
-    SetEvent(ipc_event);
+  void OnIPCThreadShutdown(WaitableEvent* ipc_event) {
+    ipc_event->Signal();
   }
 
   void OnMessageReceived(const Message& message) {
     IPC_BEGIN_MESSAGE_MAP(Worker, message)
      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_Double, OnDoubleDelay)
      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_AnswerToLife,
                                      OnAnswerDelay)
     IPC_END_MESSAGE_MAP()
   }
 
   void StartThread(base::Thread* thread, MessageLoop::Type type) {
     base::Thread::Options options;
     options.message_loop_type = type;
     thread->StartWithOptions(options);
   }
 
-  Event done_;
-  Event channel_created_;
+  scoped_ptr<WaitableEvent> done_;
+  scoped_ptr<WaitableEvent> channel_created_;
   std::wstring channel_name_;
   Channel::Mode mode_;
   scoped_ptr<SyncChannel> channel_;
   base::Thread ipc_thread_;
   base::Thread listener_thread_;
   base::Thread* overrided_thread_;
 
   DISALLOW_EVIL_CONSTRUCTORS(Worker);
 };
 
@@ skipping 12 matching lines @@
     }
   }
 
   // now create the clients
   for (size_t i = 0; i < workers.size(); ++i) {
     if (workers[i]->mode() == Channel::MODE_CLIENT)
       workers[i]->Start();
   }
 
   // wait for all the workers to finish
-  std::vector<HANDLE> done_handles;
   for (size_t i = 0; i < workers.size(); ++i)
-    done_handles.push_back(workers[i]->done_event());
+    workers[i]->done_event()->Wait();
 
-  int count = static_cast<int>(done_handles.size());
-  WaitForMultipleObjects(count, &done_handles.front(), TRUE, INFINITE);
   STLDeleteContainerPointers(workers.begin(), workers.end());
-  
+
   TestProcess::GlobalCleanup();
 }
 
 }  // namespace
 
 //-----------------------------------------------------------------------------
 
 namespace {
 
 class SimpleServer : public Worker {
  public:
   SimpleServer(bool pump_during_send)
       : Worker(Channel::MODE_SERVER, "simpler_server"),
         pump_during_send_(pump_during_send) { }
   void Run() {
-    SendAnswerToLife(pump_during_send_, INFINITE, true);
+    SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
     Done();
   }
 
   bool pump_during_send_;
 };
 
 class SimpleClient : public Worker {
  public:
   SimpleClient() : Worker(Channel::MODE_CLIENT, "simple_client") { }
 
@@ skipping 47 matching lines @@
   DelayReply(false);
   DelayReply(true);
 }
 
 //-----------------------------------------------------------------------------
 
 namespace {
 
 class NoHangServer : public Worker {
  public:
-  explicit NoHangServer(Event* got_first_reply, bool pump_during_send)
+  explicit NoHangServer(WaitableEvent* got_first_reply, bool pump_during_send)
       : Worker(Channel::MODE_SERVER, "no_hang_server"),
         got_first_reply_(got_first_reply),
         pump_during_send_(pump_during_send) { }
   void Run() {
-    SendAnswerToLife(pump_during_send_, INFINITE, true);
-    got_first_reply_->Set();
+    SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
+    got_first_reply_->Signal();
 
-    SendAnswerToLife(pump_during_send_, INFINITE, false);
+    SendAnswerToLife(pump_during_send_, base::kNoTimeout, false);
     Done();
   }
 
-  Event* got_first_reply_;
+  WaitableEvent* got_first_reply_;
   bool pump_during_send_;
 };
 
 class NoHangClient : public Worker {
  public:
-  explicit NoHangClient(Event* got_first_reply)
+  explicit NoHangClient(WaitableEvent* got_first_reply)
     : Worker(Channel::MODE_CLIENT, "no_hang_client"),
       got_first_reply_(got_first_reply) { }
 
   virtual void OnAnswerDelay(Message* reply_msg) {
     // Use the DELAY_REPLY macro so that we can force the reply to be sent
     // before this function returns (when the channel will be reset).
     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
     Send(reply_msg);
     got_first_reply_->Wait();
     CloseChannel();
     Done();
   }
 
-  Event* got_first_reply_;
+  WaitableEvent* got_first_reply_;
 };
 
 void NoHang(bool pump_during_send) {
-  Event got_first_reply;
+  WaitableEvent got_first_reply(false, false);
   std::vector<Worker*> workers;
   workers.push_back(new NoHangServer(&got_first_reply, pump_during_send));
   workers.push_back(new NoHangClient(&got_first_reply));
   RunTest(workers);
 }
 
 }  // namespace
 
 // Tests that caller doesn't hang if receiver dies
 TEST_F(IPCSyncChannelTest, NoHang) {
   NoHang(false);
   NoHang(true);
 }
 
 //-----------------------------------------------------------------------------
 
 namespace {
 
 class UnblockServer : public Worker {
  public:
   UnblockServer(bool pump_during_send)
     : Worker(Channel::MODE_SERVER, "unblock_server"),
       pump_during_send_(pump_during_send) { }
   void Run() {
-    SendAnswerToLife(pump_during_send_, INFINITE, true);
+    SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
     Done();
   }
 
   void OnDouble(int in, int* out) {
     *out = in * 2;
   }
 
   bool pump_during_send_;
 };
 
@@ skipping 40 matching lines @@
     : Worker(Channel::MODE_SERVER, "recursive_server"),
       expected_send_result_(expected_send_result),
       pump_first_(pump_first), pump_second_(pump_second)  { }
   void Run() {
     SendDouble(pump_first_, expected_send_result_);
     Done();
   }
 
   void OnDouble(int in, int* out) {
     *out = in * 2;
-    SendAnswerToLife(pump_second_, INFINITE, expected_send_result_);
+    SendAnswerToLife(pump_second_, base::kNoTimeout, expected_send_result_);
   }
 
   bool expected_send_result_, pump_first_, pump_second_;
 };
 
 class RecursiveClient : public Worker {
  public:
   explicit RecursiveClient(bool pump_during_send, bool close_channel)
     : Worker(Channel::MODE_CLIENT, "recursive_client"),
       pump_during_send_(pump_during_send), close_channel_(close_channel) { }
@@ skipping 86 matching lines @@
   void Run() {
     SendDouble(pump_during_send_, true);
     Done();
   }
 
   bool pump_during_send_;
 };
 
 class MultipleClient1 : public Worker {
  public:
-  MultipleClient1(Event* client1_msg_received, Event* client1_can_reply) :
+  MultipleClient1(WaitableEvent* client1_msg_received,
+                  WaitableEvent* client1_can_reply) :
       Worker(L"test_channel1", Channel::MODE_CLIENT),
       client1_msg_received_(client1_msg_received),
       client1_can_reply_(client1_can_reply) { }
 
   void OnDouble(int in, int* out) {
-    client1_msg_received_->Set();
+    client1_msg_received_->Signal();
     *out = in * 2;
     client1_can_reply_->Wait();
     Done();
   }
 
  private:
-  Event *client1_msg_received_, *client1_can_reply_;
+  WaitableEvent *client1_msg_received_, *client1_can_reply_;
 };
 
 class MultipleServer2 : public Worker {
  public:
   MultipleServer2() : Worker(L"test_channel2", Channel::MODE_SERVER) { }
 
   void OnAnswer(int* result) {
     *result = 42;
     Done();
   }
 };
 
 class MultipleClient2 : public Worker {
  public:
   MultipleClient2(
-    Event* client1_msg_received, Event* client1_can_reply,
+    WaitableEvent* client1_msg_received, WaitableEvent* client1_can_reply,
     bool pump_during_send)
     : Worker(L"test_channel2", Channel::MODE_CLIENT),
       client1_msg_received_(client1_msg_received),
       client1_can_reply_(client1_can_reply),
       pump_during_send_(pump_during_send) { }
 
   void Run() {
     client1_msg_received_->Wait();
-    SendAnswerToLife(pump_during_send_, INFINITE, true);
-    client1_can_reply_->Set();
+    SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
+    client1_can_reply_->Signal();
     Done();
   }
 
  private:
-  Event *client1_msg_received_, *client1_can_reply_;
+  WaitableEvent *client1_msg_received_, *client1_can_reply_;
   bool pump_during_send_;
 };
 
 void Multiple(bool server_pump, bool client_pump) {
   std::vector<Worker*> workers;
 
   // A shared worker thread so that server1 and server2 run on one thread.
   base::Thread worker_thread("Multiple");
   worker_thread.Start();
 
   // Server1 sends a sync msg to client1, which blocks the reply until
   // server2 (which runs on the same worker thread as server1) responds
   // to a sync msg from client2.
-  Event client1_msg_received, client1_can_reply;
+  WaitableEvent client1_msg_received(false, false);
+  WaitableEvent client1_can_reply(false, false);
 
   Worker* worker;
 
   worker = new MultipleServer2();
   worker->OverrideThread(&worker_thread);
   workers.push_back(worker);
 
   worker = new MultipleClient2(
       &client1_msg_received, &client1_can_reply, client_pump);
   workers.push_back(worker);
@@ skipping 32 matching lines @@
   void Run() {
     SendDouble(pump_during_send_, true);
     Done();
   }
 
   bool pump_during_send_;
 };
 
 class QueuedReplyClient1 : public Worker {
  public:
-  QueuedReplyClient1(Event* client1_msg_received, Event* server2_can_reply) :
+  QueuedReplyClient1(WaitableEvent* client1_msg_received,
+                     WaitableEvent* server2_can_reply) :
       Worker(L"test_channel1", Channel::MODE_CLIENT),
       client1_msg_received_(client1_msg_received),
       server2_can_reply_(server2_can_reply) { }
 
   void OnDouble(int in, int* out) {
-    client1_msg_received_->Set();
+    client1_msg_received_->Signal();
     *out = in * 2;
     server2_can_reply_->Wait();
     Done();
   }
 
  private:
-  Event *client1_msg_received_, *server2_can_reply_;
+  WaitableEvent *client1_msg_received_, *server2_can_reply_;
 };
 
 class QueuedReplyServer2 : public Worker {
  public:
-  explicit QueuedReplyServer2(Event* server2_can_reply) :
+  explicit QueuedReplyServer2(WaitableEvent* server2_can_reply) :
       Worker(L"test_channel2", Channel::MODE_SERVER),
       server2_can_reply_(server2_can_reply) { }
 
   void OnAnswer(int* result) {
-    server2_can_reply_->Set();
+    server2_can_reply_->Signal();
 
     // give client1's reply time to reach the server listener thread
-    Sleep(200);
+    PlatformThread::Sleep(200);
 
     *result = 42;
     Done();
   }
 
-  Event *server2_can_reply_;
+  WaitableEvent *server2_can_reply_;
 };
 
 class QueuedReplyClient2 : public Worker {
  public:
   explicit QueuedReplyClient2(
-    Event* client1_msg_received, bool pump_during_send)
+    WaitableEvent* client1_msg_received, bool pump_during_send)
     : Worker(L"test_channel2", Channel::MODE_CLIENT),
       client1_msg_received_(client1_msg_received),
       pump_during_send_(pump_during_send){ }
 
   void Run() {
     client1_msg_received_->Wait();
-    SendAnswerToLife(pump_during_send_, INFINITE, true);
+    SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
     Done();
   }
 
-  Event *client1_msg_received_;
+  WaitableEvent *client1_msg_received_;
   bool pump_during_send_;
 };
 
 void QueuedReply(bool server_pump, bool client_pump) {
   std::vector<Worker*> workers;
 
   // A shared worker thread so that server1 and server2 run on one thread.
   base::Thread worker_thread("QueuedReply");
   worker_thread.Start();
 
-  Event client1_msg_received, server2_can_reply;
+  WaitableEvent client1_msg_received(false, false);
+  WaitableEvent server2_can_reply(false, false);
 
   Worker* worker;
 
   worker = new QueuedReplyServer2(&server2_can_reply);
   worker->OverrideThread(&worker_thread);
   workers.push_back(worker);
 
   worker = new QueuedReplyClient2(&client1_msg_received, client_pump);
   workers.push_back(worker);
 
@@ skipping 225 matching lines @@
 
 //------------------------------------------------------------------------------
 
 namespace {
 
 class NestedTask : public Task {
  public:
   NestedTask(Worker* server) : server_(server) { }
   void Run() {
     // Sleep a bit so that we wake up after the reply has been received.
-    Sleep(250);
-    server_->SendAnswerToLife(true, INFINITE, true);
+    PlatformThread::Sleep(250);
+    server_->SendAnswerToLife(true, base::kNoTimeout, true);
   }
 
   Worker* server_;
 };
 
 static bool timeout_occured = false;
 
 class TimeoutTask : public Task {
  public:
   void Run() {
@@ skipping 23 matching lines @@
 
 // Tests http://b/1474092 - that if after the done_event is set but before
 // OnObjectSignaled is called another message is sent out, then after its
 // reply comes back OnObjectSignaled will be called for the first message.
 TEST_F(IPCSyncChannelTest, DoneEventRace) {
   std::vector<Worker*> workers;
   workers.push_back(new DoneEventRaceServer());
   workers.push_back(new SimpleClient());
   RunTest(workers);
 }