ipc/ipc_sync_channel_unittest.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4 //
   5 // Unit test for SyncChannel.
   6
   7 #include "ipc/ipc_sync_channel.h"
   8
   9 #include <string>
  10 #include <vector>
  11
  12 #include "base/basictypes.h"
  13 #include "base/bind.h"
  14 #include "base/logging.h"
  15 #include "base/memory/scoped_ptr.h"
  16 #include "base/message_loop.h"
  17 #include "base/process_util.h"
  18 #include "base/string_util.h"
  19 #include "base/threading/platform_thread.h"
  20 #include "base/threading/thread.h"
  21 #include "base/synchronization/waitable_event.h"
  22 #include "ipc/ipc_listener.h"
  23 #include "ipc/ipc_message.h"
  24 #include "ipc/ipc_sender.h"
  25 #include "ipc/ipc_sync_message_filter.h"
  26 #include "ipc/ipc_sync_message_unittest.h"
  27 #include "testing/gtest/include/gtest/gtest.h"
  28
  29 using base::WaitableEvent;
  30 using namespace IPC;
  31
  32 namespace {
  33
  34 // Base class for a "process" with listener and IPC threads.
  35 class Worker : public Listener, public Sender {
  36  public:
  37   // Will create a channel without a name.
  38   Worker(Channel::Mode mode, const std::string& thread_name)
  39       : done_(new WaitableEvent(false, false)),
  40         channel_created_(new WaitableEvent(false, false)),
  41         mode_(mode),
  42         ipc_thread_((thread_name + "_ipc").c_str()),
  43         listener_thread_((thread_name + "_listener").c_str()),
  44         overrided_thread_(NULL),
  45         shutdown_event_(true, false),
  46         is_shutdown_(false) {
  47   }
  48
  49   // Will create a named channel and use this name for the threads' name.
  50   Worker(const std::string& channel_name, Channel::Mode mode)
  51       : done_(new WaitableEvent(false, false)),
  52         channel_created_(new WaitableEvent(false, false)),
  53         channel_name_(channel_name),
  54         mode_(mode),
  55         ipc_thread_((channel_name + "_ipc").c_str()),
  56         listener_thread_((channel_name + "_listener").c_str()),
  57         overrided_thread_(NULL),
  58         shutdown_event_(true, false),
  59         is_shutdown_(false) {
  60   }
  61
  62   virtual ~Worker() {
  63     // Shutdown() must be called before destruction.
  64     CHECK(is_shutdown_);
  65   }
  66   void AddRef() { }
  67   void Release() { }
  68   virtual bool Send(Message* msg) OVERRIDE { return channel_->Send(msg); }
  69   bool SendWithTimeout(Message* msg, int timeout_ms) {
  70     return channel_->SendWithTimeout(msg, timeout_ms);
  71   }
  72   void WaitForChannelCreation() { channel_created_->Wait(); }
  73   void CloseChannel() {
  74     DCHECK(MessageLoop::current() == ListenerThread()->message_loop());
  75     channel_->Close();
  76   }
  77   void Start() {
  78     StartThread(&listener_thread_, MessageLoop::TYPE_DEFAULT);
  79     ListenerThread()->message_loop()->PostTask(
  80         FROM_HERE, base::Bind(&Worker::OnStart, this));
  81   }
  82   void Shutdown() {
  83     // The IPC thread needs to outlive SyncChannel. We can't do this in
  84     // ~Worker(), since that'll reset the vtable pointer (to Worker's), which
  85     // may result in a race conditions. See http://crbug.com/25841.
  86     WaitableEvent listener_done(false, false), ipc_done(false, false);
  87     ListenerThread()->message_loop()->PostTask(
  88         FROM_HERE, base::Bind(&Worker::OnListenerThreadShutdown1, this,
  89                               &listener_done, &ipc_done));
  90     listener_done.Wait();
  91     ipc_done.Wait();
  92     ipc_thread_.Stop();
  93     listener_thread_.Stop();
  94     is_shutdown_ = true;
  95   }
  96   void OverrideThread(base::Thread* overrided_thread) {
  97     DCHECK(overrided_thread_ == NULL);
  98     overrided_thread_ = overrided_thread;
  99   }
 100   bool SendAnswerToLife(bool pump, int timeout, bool succeed) {
 101     int answer = 0;
 102     SyncMessage* msg = new SyncChannelTestMsg_AnswerToLife(&answer);
 103     if (pump)
 104       msg->EnableMessagePumping();
 105     bool result = SendWithTimeout(msg, timeout);
 106     DCHECK_EQ(result, succeed);
 107     DCHECK_EQ(answer, (succeed ? 42 : 0));
 108     return result;
 109   }
 110   bool SendDouble(bool pump, bool succeed) {
 111     int answer = 0;
 112     SyncMessage* msg = new SyncChannelTestMsg_Double(5, &answer);
 113     if (pump)
 114       msg->EnableMessagePumping();
 115     bool result = Send(msg);
 116     DCHECK_EQ(result, succeed);
 117     DCHECK_EQ(answer, (succeed ? 10 : 0));
 118     return result;
 119   }
 120   const std::string& channel_name() { return channel_name_; }
 121   Channel::Mode mode() { return mode_; }
 122   WaitableEvent* done_event() { return done_.get(); }
 123   WaitableEvent* shutdown_event() { return &shutdown_event_; }
 124   void ResetChannel() { channel_.reset(); }
 125   // Derived classes need to call this when they've completed their part of
 126   // the test.
 127   void Done() { done_->Signal(); }
 128
 129  protected:
 130   SyncChannel* channel() { return channel_.get(); }
 131   // Functions for dervied classes to implement if they wish.
 132   virtual void Run() { }
 133   virtual void OnAnswer(int* answer) { NOTREACHED(); }
 134   virtual void OnAnswerDelay(Message* reply_msg) {
 135     // The message handler map below can only take one entry for
 136     // SyncChannelTestMsg_AnswerToLife, so since some classes want
 137     // the normal version while other want the delayed reply, we
 138     // call the normal version if the derived class didn't override
 139     // this function.
 140     int answer;
 141     OnAnswer(&answer);
 142     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, answer);
 143     Send(reply_msg);
 144   }
 145   virtual void OnDouble(int in, int* out) { NOTREACHED(); }
 146   virtual void OnDoubleDelay(int in, Message* reply_msg) {
 147     int result;
 148     OnDouble(in, &result);
 149     SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, result);
 150     Send(reply_msg);
 151   }
 152
 153   virtual void OnNestedTestMsg(Message* reply_msg) {
 154     NOTREACHED();
 155   }
 156
 157   virtual SyncChannel* CreateChannel() {
 158     return new SyncChannel(
 159         channel_name_, mode_, this, ipc_thread_.message_loop_proxy(), true,
 160         &shutdown_event_);
 161   }
 162
 163   base::Thread* ListenerThread() {
 164     return overrided_thread_ ? overrided_thread_ : &listener_thread_;
 165   }
 166
 167   const base::Thread& ipc_thread() const { return ipc_thread_; }
 168
 169  private:
 170   // Called on the listener thread to create the sync channel.
 171   void OnStart() {
 172     // Link ipc_thread_, listener_thread_ and channel_ altogether.
 173     StartThread(&ipc_thread_, MessageLoop::TYPE_IO);
 174     channel_.reset(CreateChannel());
 175     channel_created_->Signal();
 176     Run();
 177   }
 178
 179   void OnListenerThreadShutdown1(WaitableEvent* listener_event,
 180                                  WaitableEvent* ipc_event) {
 181     // SyncChannel needs to be destructed on the thread that it was created on.
 182     channel_.reset();
 183
 184     MessageLoop::current()->RunUntilIdle();
 185
 186     ipc_thread_.message_loop()->PostTask(
 187         FROM_HERE, base::Bind(&Worker::OnIPCThreadShutdown, this,
 188                               listener_event, ipc_event));
 189   }
 190
 191   void OnIPCThreadShutdown(WaitableEvent* listener_event,
 192                            WaitableEvent* ipc_event) {
 193     MessageLoop::current()->RunUntilIdle();
 194     ipc_event->Signal();
 195
 196     listener_thread_.message_loop()->PostTask(
 197         FROM_HERE, base::Bind(&Worker::OnListenerThreadShutdown2, this,
 198                               listener_event));
 199   }
 200
 201   void OnListenerThreadShutdown2(WaitableEvent* listener_event) {
 202     MessageLoop::current()->RunUntilIdle();
 203     listener_event->Signal();
 204   }
 205
 206   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
 207     IPC_BEGIN_MESSAGE_MAP(Worker, message)
 208      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_Double, OnDoubleDelay)
 209      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_AnswerToLife,
 210                                      OnAnswerDelay)
 211      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelNestedTestMsg_String,
 212                                      OnNestedTestMsg)
 213     IPC_END_MESSAGE_MAP()
 214     return true;
 215   }
 216
 217   void StartThread(base::Thread* thread, MessageLoop::Type type) {
 218     base::Thread::Options options;
 219     options.message_loop_type = type;
 220     thread->StartWithOptions(options);
 221   }
 222
 223   scoped_ptr<WaitableEvent> done_;
 224   scoped_ptr<WaitableEvent> channel_created_;
 225   std::string channel_name_;
 226   Channel::Mode mode_;
 227   scoped_ptr<SyncChannel> channel_;
 228   base::Thread ipc_thread_;
 229   base::Thread listener_thread_;
 230   base::Thread* overrided_thread_;
 231
 232   base::WaitableEvent shutdown_event_;
 233
 234   bool is_shutdown_;
 235
 236   DISALLOW_COPY_AND_ASSIGN(Worker);
 237 };
 238
 239
 240 // Starts the test with the given workers.  This function deletes the workers
 241 // when it's done.
 242 void RunTest(std::vector<Worker*> workers) {
 243   // First we create the workers that are channel servers, or else the other
 244   // workers' channel initialization might fail because the pipe isn't created..
 245   for (size_t i = 0; i < workers.size(); ++i) {
 246     if (workers[i]->mode() & Channel::MODE_SERVER_FLAG) {
 247       workers[i]->Start();
 248       workers[i]->WaitForChannelCreation();
 249     }
 250   }
 251
 252   // now create the clients
 253   for (size_t i = 0; i < workers.size(); ++i) {
 254     if (workers[i]->mode() & Channel::MODE_CLIENT_FLAG)
 255       workers[i]->Start();
 256   }
 257
 258   // wait for all the workers to finish
 259   for (size_t i = 0; i < workers.size(); ++i)
 260     workers[i]->done_event()->Wait();
 261
 262   for (size_t i = 0; i < workers.size(); ++i) {
 263     workers[i]->Shutdown();
 264     delete workers[i];
 265   }
 266 }
 267
 268 class IPCSyncChannelTest : public testing::Test {
 269  private:
 270   MessageLoop message_loop_;
 271 };
 272
 273 //------------------------------------------------------------------------------
 274
 275 class SimpleServer : public Worker {
 276  public:
 277   explicit SimpleServer(bool pump_during_send)
 278       : Worker(Channel::MODE_SERVER, "simpler_server"),
 279         pump_during_send_(pump_during_send) { }
 280   virtual void Run() OVERRIDE {
 281     SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 282     Done();
 283   }
 284
 285   bool pump_during_send_;
 286 };
 287
 288 class SimpleClient : public Worker {
 289  public:
 290   SimpleClient() : Worker(Channel::MODE_CLIENT, "simple_client") { }
 291
 292   virtual void OnAnswer(int* answer) OVERRIDE {
 293     *answer = 42;
 294     Done();
 295   }
 296 };
 297
 298 void Simple(bool pump_during_send) {
 299   std::vector<Worker*> workers;
 300   workers.push_back(new SimpleServer(pump_during_send));
 301   workers.push_back(new SimpleClient());
 302   RunTest(workers);
 303 }
 304
 305 // Tests basic synchronous call
 306 TEST_F(IPCSyncChannelTest, Simple) {
 307   Simple(false);
 308   Simple(true);
 309 }
 310
 311 //------------------------------------------------------------------------------
 312
 313 // Worker classes which override how the sync channel is created to use the
 314 // two-step initialization (calling the lightweight constructor and then
 315 // ChannelProxy::Init separately) process.
 316 class TwoStepServer : public Worker {
 317  public:
 318   explicit TwoStepServer(bool create_pipe_now)
 319       : Worker(Channel::MODE_SERVER, "simpler_server"),
 320         create_pipe_now_(create_pipe_now) { }
 321
 322   virtual void Run() OVERRIDE {
 323     SendAnswerToLife(false, base::kNoTimeout, true);
 324     Done();
 325   }
 326
 327   virtual SyncChannel* CreateChannel() OVERRIDE {
 328     SyncChannel* channel = new SyncChannel(
 329         this, ipc_thread().message_loop_proxy(), shutdown_event());
 330     channel->Init(channel_name(), mode(), create_pipe_now_);
 331     return channel;
 332   }
 333
 334   bool create_pipe_now_;
 335 };
 336
 337 class TwoStepClient : public Worker {
 338  public:
 339   TwoStepClient(bool create_pipe_now)
 340       : Worker(Channel::MODE_CLIENT, "simple_client"),
 341         create_pipe_now_(create_pipe_now) { }
 342
 343   virtual void OnAnswer(int* answer) OVERRIDE {
 344     *answer = 42;
 345     Done();
 346   }
 347
 348   virtual SyncChannel* CreateChannel() OVERRIDE {
 349     SyncChannel* channel = new SyncChannel(
 350         this, ipc_thread().message_loop_proxy(), shutdown_event());
 351     channel->Init(channel_name(), mode(), create_pipe_now_);
 352     return channel;
 353   }
 354
 355   bool create_pipe_now_;
 356 };
 357
 358 void TwoStep(bool create_server_pipe_now, bool create_client_pipe_now) {
 359   std::vector<Worker*> workers;
 360   workers.push_back(new TwoStepServer(create_server_pipe_now));
 361   workers.push_back(new TwoStepClient(create_client_pipe_now));
 362   RunTest(workers);
 363 }
 364
 365 // Tests basic two-step initialization, where you call the lightweight
 366 // constructor then Init.
 367 TEST_F(IPCSyncChannelTest, TwoStepInitialization) {
 368   TwoStep(false, false);
 369   TwoStep(false, true);
 370   TwoStep(true, false);
 371   TwoStep(true, true);
 372 }
 373
 374 //------------------------------------------------------------------------------
 375
 376 class DelayClient : public Worker {
 377  public:
 378   DelayClient() : Worker(Channel::MODE_CLIENT, "delay_client") { }
 379
 380   virtual void OnAnswerDelay(Message* reply_msg) OVERRIDE {
 381     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 382     Send(reply_msg);
 383     Done();
 384   }
 385 };
 386
 387 void DelayReply(bool pump_during_send) {
 388   std::vector<Worker*> workers;
 389   workers.push_back(new SimpleServer(pump_during_send));
 390   workers.push_back(new DelayClient());
 391   RunTest(workers);
 392 }
 393
 394 // Tests that asynchronous replies work
 395 TEST_F(IPCSyncChannelTest, DelayReply) {
 396   DelayReply(false);
 397   DelayReply(true);
 398 }
 399
 400 //------------------------------------------------------------------------------
 401
 402 class NoHangServer : public Worker {
 403  public:
 404   NoHangServer(WaitableEvent* got_first_reply, bool pump_during_send)
 405       : Worker(Channel::MODE_SERVER, "no_hang_server"),
 406         got_first_reply_(got_first_reply),
 407         pump_during_send_(pump_during_send) { }
 408   virtual void Run() OVERRIDE {
 409     SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 410     got_first_reply_->Signal();
 411
 412     SendAnswerToLife(pump_during_send_, base::kNoTimeout, false);
 413     Done();
 414   }
 415
 416   WaitableEvent* got_first_reply_;
 417   bool pump_during_send_;
 418 };
 419
 420 class NoHangClient : public Worker {
 421  public:
 422   explicit NoHangClient(WaitableEvent* got_first_reply)
 423     : Worker(Channel::MODE_CLIENT, "no_hang_client"),
 424       got_first_reply_(got_first_reply) { }
 425
 426   virtual void OnAnswerDelay(Message* reply_msg) OVERRIDE {
 427     // Use the DELAY_REPLY macro so that we can force the reply to be sent
 428     // before this function returns (when the channel will be reset).
 429     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 430     Send(reply_msg);
 431     got_first_reply_->Wait();
 432     CloseChannel();
 433     Done();
 434   }
 435
 436   WaitableEvent* got_first_reply_;
 437 };
 438
 439 void NoHang(bool pump_during_send) {
 440   WaitableEvent got_first_reply(false, false);
 441   std::vector<Worker*> workers;
 442   workers.push_back(new NoHangServer(&got_first_reply, pump_during_send));
 443   workers.push_back(new NoHangClient(&got_first_reply));
 444   RunTest(workers);
 445 }
 446
 447 // Tests that caller doesn't hang if receiver dies
 448 TEST_F(IPCSyncChannelTest, NoHang) {
 449   NoHang(false);
 450   NoHang(true);
 451 }
 452
 453 //------------------------------------------------------------------------------
 454
 455 class UnblockServer : public Worker {
 456  public:
 457   UnblockServer(bool pump_during_send, bool delete_during_send)
 458     : Worker(Channel::MODE_SERVER, "unblock_server"),
 459       pump_during_send_(pump_during_send),
 460       delete_during_send_(delete_during_send) { }
 461   virtual void Run() OVERRIDE {
 462     if (delete_during_send_) {
 463       // Use custom code since race conditions mean the answer may or may not be
 464       // available.
 465       int answer = 0;
 466       SyncMessage* msg = new SyncChannelTestMsg_AnswerToLife(&answer);
 467       if (pump_during_send_)
 468         msg->EnableMessagePumping();
 469       Send(msg);
 470     } else {
 471       SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 472     }
 473     Done();
 474   }
 475
 476   virtual void OnDoubleDelay(int in, Message* reply_msg) OVERRIDE {
 477     SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, in * 2);
 478     Send(reply_msg);
 479     if (delete_during_send_)
 480       ResetChannel();
 481   }
 482
 483   bool pump_during_send_;
 484   bool delete_during_send_;
 485 };
 486
 487 class UnblockClient : public Worker {
 488  public:
 489   explicit UnblockClient(bool pump_during_send)
 490     : Worker(Channel::MODE_CLIENT, "unblock_client"),
 491       pump_during_send_(pump_during_send) { }
 492
 493   virtual void OnAnswer(int* answer) OVERRIDE {
 494     SendDouble(pump_during_send_, true);
 495     *answer = 42;
 496     Done();
 497   }
 498
 499   bool pump_during_send_;
 500 };
 501
 502 void Unblock(bool server_pump, bool client_pump, bool delete_during_send) {
 503   std::vector<Worker*> workers;
 504   workers.push_back(new UnblockServer(server_pump, delete_during_send));
 505   workers.push_back(new UnblockClient(client_pump));
 506   RunTest(workers);
 507 }
 508
 509 // Tests that the caller unblocks to answer a sync message from the receiver.
 510 TEST_F(IPCSyncChannelTest, Unblock) {
 511   Unblock(false, false, false);
 512   Unblock(false, true, false);
 513   Unblock(true, false, false);
 514   Unblock(true, true, false);
 515 }
 516
 517 //------------------------------------------------------------------------------
 518
 519 // Tests that the the SyncChannel object can be deleted during a Send.
 520 TEST_F(IPCSyncChannelTest, ChannelDeleteDuringSend) {
 521   Unblock(false, false, true);
 522   Unblock(false, true, true);
 523   Unblock(true, false, true);
 524   Unblock(true, true, true);
 525 }
 526
 527 //------------------------------------------------------------------------------
 528
 529 class RecursiveServer : public Worker {
 530  public:
 531   RecursiveServer(bool expected_send_result, bool pump_first, bool pump_second)
 532       : Worker(Channel::MODE_SERVER, "recursive_server"),
 533         expected_send_result_(expected_send_result),
 534         pump_first_(pump_first), pump_second_(pump_second) {}
 535   virtual void Run() OVERRIDE {
 536     SendDouble(pump_first_, expected_send_result_);
 537     Done();
 538   }
 539
 540   virtual void OnDouble(int in, int* out) OVERRIDE {
 541     *out = in * 2;
 542     SendAnswerToLife(pump_second_, base::kNoTimeout, expected_send_result_);
 543   }
 544
 545   bool expected_send_result_, pump_first_, pump_second_;
 546 };
 547
 548 class RecursiveClient : public Worker {
 549  public:
 550   RecursiveClient(bool pump_during_send, bool close_channel)
 551       : Worker(Channel::MODE_CLIENT, "recursive_client"),
 552         pump_during_send_(pump_during_send), close_channel_(close_channel) {}
 553
 554   virtual void OnDoubleDelay(int in, Message* reply_msg) OVERRIDE {
 555     SendDouble(pump_during_send_, !close_channel_);
 556     if (close_channel_) {
 557       delete reply_msg;
 558     } else {
 559       SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, in * 2);
 560       Send(reply_msg);
 561     }
 562     Done();
 563   }
 564
 565   virtual void OnAnswerDelay(Message* reply_msg) OVERRIDE {
 566     if (close_channel_) {
 567       delete reply_msg;
 568       CloseChannel();
 569     } else {
 570       SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 571       Send(reply_msg);
 572     }
 573   }
 574
 575   bool pump_during_send_, close_channel_;
 576 };
 577
 578 void Recursive(
 579     bool server_pump_first, bool server_pump_second, bool client_pump) {
 580   std::vector<Worker*> workers;
 581   workers.push_back(
 582       new RecursiveServer(true, server_pump_first, server_pump_second));
 583   workers.push_back(new RecursiveClient(client_pump, false));
 584   RunTest(workers);
 585 }
 586
 587 // Tests a server calling Send while another Send is pending.
 588 TEST_F(IPCSyncChannelTest, Recursive) {
 589   Recursive(false, false, false);
 590   Recursive(false, false, true);
 591   Recursive(false, true, false);
 592   Recursive(false, true, true);
 593   Recursive(true, false, false);
 594   Recursive(true, false, true);
 595   Recursive(true, true, false);
 596   Recursive(true, true, true);
 597 }
 598
 599 //------------------------------------------------------------------------------
 600
 601 void RecursiveNoHang(
 602     bool server_pump_first, bool server_pump_second, bool client_pump) {
 603   std::vector<Worker*> workers;
 604   workers.push_back(
 605       new RecursiveServer(false, server_pump_first, server_pump_second));
 606   workers.push_back(new RecursiveClient(client_pump, true));
 607   RunTest(workers);
 608 }
 609
 610 // Tests that if a caller makes a sync call during an existing sync call and
 611 // the receiver dies, neither of the Send() calls hang.
 612 TEST_F(IPCSyncChannelTest, RecursiveNoHang) {
 613   RecursiveNoHang(false, false, false);
 614   RecursiveNoHang(false, false, true);
 615   RecursiveNoHang(false, true, false);
 616   RecursiveNoHang(false, true, true);
 617   RecursiveNoHang(true, false, false);
 618   RecursiveNoHang(true, false, true);
 619   RecursiveNoHang(true, true, false);
 620   RecursiveNoHang(true, true, true);
 621 }
 622
 623 //------------------------------------------------------------------------------
 624
 625 class MultipleServer1 : public Worker {
 626  public:
 627   explicit MultipleServer1(bool pump_during_send)
 628     : Worker("test_channel1", Channel::MODE_SERVER),
 629       pump_during_send_(pump_during_send) { }
 630
 631   virtual void Run() OVERRIDE {
 632     SendDouble(pump_during_send_, true);
 633     Done();
 634   }
 635
 636   bool pump_during_send_;
 637 };
 638
 639 class MultipleClient1 : public Worker {
 640  public:
 641   MultipleClient1(WaitableEvent* client1_msg_received,
 642                   WaitableEvent* client1_can_reply) :
 643       Worker("test_channel1", Channel::MODE_CLIENT),
 644       client1_msg_received_(client1_msg_received),
 645       client1_can_reply_(client1_can_reply) { }
 646
 647   virtual void OnDouble(int in, int* out) OVERRIDE {
 648     client1_msg_received_->Signal();
 649     *out = in * 2;
 650     client1_can_reply_->Wait();
 651     Done();
 652   }
 653
 654  private:
 655   WaitableEvent *client1_msg_received_, *client1_can_reply_;
 656 };
 657
 658 class MultipleServer2 : public Worker {
 659  public:
 660   MultipleServer2() : Worker("test_channel2", Channel::MODE_SERVER) { }
 661
 662   virtual void OnAnswer(int* result) OVERRIDE {
 663     *result = 42;
 664     Done();
 665   }
 666 };
 667
 668 class MultipleClient2 : public Worker {
 669  public:
 670   MultipleClient2(
 671     WaitableEvent* client1_msg_received, WaitableEvent* client1_can_reply,
 672     bool pump_during_send)
 673     : Worker("test_channel2", Channel::MODE_CLIENT),
 674       client1_msg_received_(client1_msg_received),
 675       client1_can_reply_(client1_can_reply),
 676       pump_during_send_(pump_during_send) { }
 677
 678   virtual void Run() OVERRIDE {
 679     client1_msg_received_->Wait();
 680     SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 681     client1_can_reply_->Signal();
 682     Done();
 683   }
 684
 685  private:
 686   WaitableEvent *client1_msg_received_, *client1_can_reply_;
 687   bool pump_during_send_;
 688 };
 689
 690 void Multiple(bool server_pump, bool client_pump) {
 691   std::vector<Worker*> workers;
 692
 693   // A shared worker thread so that server1 and server2 run on one thread.
 694   base::Thread worker_thread("Multiple");
 695   ASSERT_TRUE(worker_thread.Start());
 696
 697   // Server1 sends a sync msg to client1, which blocks the reply until
 698   // server2 (which runs on the same worker thread as server1) responds
 699   // to a sync msg from client2.
 700   WaitableEvent client1_msg_received(false, false);
 701   WaitableEvent client1_can_reply(false, false);
 702
 703   Worker* worker;
 704
 705   worker = new MultipleServer2();
 706   worker->OverrideThread(&worker_thread);
 707   workers.push_back(worker);
 708
 709   worker = new MultipleClient2(
 710       &client1_msg_received, &client1_can_reply, client_pump);
 711   workers.push_back(worker);
 712
 713   worker = new MultipleServer1(server_pump);
 714   worker->OverrideThread(&worker_thread);
 715   workers.push_back(worker);
 716
 717   worker = new MultipleClient1(
 718       &client1_msg_received, &client1_can_reply);
 719   workers.push_back(worker);
 720
 721   RunTest(workers);
 722 }
 723
 724 // Tests that multiple SyncObjects on the same listener thread can unblock each
 725 // other.
 726 TEST_F(IPCSyncChannelTest, Multiple) {
 727   Multiple(false, false);
 728   Multiple(false, true);
 729   Multiple(true, false);
 730   Multiple(true, true);
 731 }
 732
 733 //------------------------------------------------------------------------------
 734
 735 // This class provides server side functionality to test the case where
 736 // multiple sync channels are in use on the same thread on the client and
 737 // nested calls are issued.
 738 class QueuedReplyServer : public Worker {
 739  public:
 740   QueuedReplyServer(base::Thread* listener_thread,
 741                     const std::string& channel_name,
 742                     const std::string& reply_text)
 743       : Worker(channel_name, Channel::MODE_SERVER),
 744         reply_text_(reply_text) {
 745     Worker::OverrideThread(listener_thread);
 746   }
 747
 748   virtual void OnNestedTestMsg(Message* reply_msg) OVERRIDE {
 749     VLOG(1) << __FUNCTION__ << " Sending reply: " << reply_text_;
 750     SyncChannelNestedTestMsg_String::WriteReplyParams(reply_msg, reply_text_);
 751     Send(reply_msg);
 752     Done();
 753   }
 754
 755  private:
 756   std::string reply_text_;
 757 };
 758
 759 // The QueuedReplyClient class provides functionality to test the case where
 760 // multiple sync channels are in use on the same thread and they make nested
 761 // sync calls, i.e. while the first channel waits for a response it makes a
 762 // sync call on another channel.
 763 // The callstack should unwind correctly, i.e. the outermost call should
 764 // complete first, and so on.
 765 class QueuedReplyClient : public Worker {
 766  public:
 767   QueuedReplyClient(base::Thread* listener_thread,
 768                     const std::string& channel_name,
 769                     const std::string& expected_text,
 770                     bool pump_during_send)
 771       : Worker(channel_name, Channel::MODE_CLIENT),
 772         pump_during_send_(pump_during_send),
 773         expected_text_(expected_text) {
 774     Worker::OverrideThread(listener_thread);
 775   }
 776
 777   virtual void Run() OVERRIDE {
 778     std::string response;
 779     SyncMessage* msg = new SyncChannelNestedTestMsg_String(&response);
 780     if (pump_during_send_)
 781       msg->EnableMessagePumping();
 782     bool result = Send(msg);
 783     DCHECK(result);
 784     DCHECK_EQ(response, expected_text_);
 785
 786     VLOG(1) << __FUNCTION__ << " Received reply: " << response;
 787     Done();
 788   }
 789
 790  private:
 791   bool pump_during_send_;
 792   std::string expected_text_;
 793 };
 794
 795 void QueuedReply(bool client_pump) {
 796   std::vector<Worker*> workers;
 797
 798   // A shared worker thread for servers
 799   base::Thread server_worker_thread("QueuedReply_ServerListener");
 800   ASSERT_TRUE(server_worker_thread.Start());
 801
 802   base::Thread client_worker_thread("QueuedReply_ClientListener");
 803   ASSERT_TRUE(client_worker_thread.Start());
 804
 805   Worker* worker;
 806
 807   worker = new QueuedReplyServer(&server_worker_thread,
 808                                  "QueuedReply_Server1",
 809                                  "Got first message");
 810   workers.push_back(worker);
 811
 812   worker = new QueuedReplyServer(&server_worker_thread,
 813                                  "QueuedReply_Server2",
 814                                  "Got second message");
 815   workers.push_back(worker);
 816
 817   worker = new QueuedReplyClient(&client_worker_thread,
 818                                  "QueuedReply_Server1",
 819                                  "Got first message",
 820                                  client_pump);
 821   workers.push_back(worker);
 822
 823   worker = new QueuedReplyClient(&client_worker_thread,
 824                                  "QueuedReply_Server2",
 825                                  "Got second message",
 826                                  client_pump);
 827   workers.push_back(worker);
 828
 829   RunTest(workers);
 830 }
 831
 832 // While a blocking send is in progress, the listener thread might answer other
 833 // synchronous messages.  This tests that if during the response to another
 834 // message the reply to the original messages comes, it is queued up correctly
 835 // and the original Send is unblocked later.
 836 // We also test that the send call stacks unwind correctly when the channel
 837 // pumps messages while waiting for a response.
 838 TEST_F(IPCSyncChannelTest, QueuedReply) {
 839   QueuedReply(false);
 840   QueuedReply(true);
 841 }
 842
 843 //------------------------------------------------------------------------------
 844
 845 class ChattyClient : public Worker {
 846  public:
 847   ChattyClient() :
 848       Worker(Channel::MODE_CLIENT, "chatty_client") { }
 849
 850   virtual void OnAnswer(int* answer) OVERRIDE {
 851     // The PostMessage limit is 10k.  Send 20% more than that.
 852     const int kMessageLimit = 10000;
 853     const int kMessagesToSend = kMessageLimit * 120 / 100;
 854     for (int i = 0; i < kMessagesToSend; ++i) {
 855       if (!SendDouble(false, true))
 856         break;
 857     }
 858     *answer = 42;
 859     Done();
 860   }
 861 };
 862
 863 void ChattyServer(bool pump_during_send) {
 864   std::vector<Worker*> workers;
 865   workers.push_back(new UnblockServer(pump_during_send, false));
 866   workers.push_back(new ChattyClient());
 867   RunTest(workers);
 868 }
 869
 870 // Tests http://b/1093251 - that sending lots of sync messages while
 871 // the receiver is waiting for a sync reply does not overflow the PostMessage
 872 // queue.
 873 TEST_F(IPCSyncChannelTest, ChattyServer) {
 874   ChattyServer(false);
 875   ChattyServer(true);
 876 }
 877
 878 //------------------------------------------------------------------------------
 879
 880 class TimeoutServer : public Worker {
 881  public:
 882   TimeoutServer(int timeout_ms,
 883                 std::vector<bool> timeout_seq,
 884                 bool pump_during_send)
 885       : Worker(Channel::MODE_SERVER, "timeout_server"),
 886         timeout_ms_(timeout_ms),
 887         timeout_seq_(timeout_seq),
 888         pump_during_send_(pump_during_send) {
 889   }
 890
 891   virtual void Run() OVERRIDE {
 892     for (std::vector<bool>::const_iterator iter = timeout_seq_.begin();
 893          iter != timeout_seq_.end(); ++iter) {
 894       SendAnswerToLife(pump_during_send_, timeout_ms_, !*iter);
 895     }
 896     Done();
 897   }
 898
 899  private:
 900   int timeout_ms_;
 901   std::vector<bool> timeout_seq_;
 902   bool pump_during_send_;
 903 };
 904
 905 class UnresponsiveClient : public Worker {
 906  public:
 907   explicit UnresponsiveClient(std::vector<bool> timeout_seq)
 908       : Worker(Channel::MODE_CLIENT, "unresponsive_client"),
 909         timeout_seq_(timeout_seq) {
 910   }
 911
 912   virtual void OnAnswerDelay(Message* reply_msg) OVERRIDE {
 913     DCHECK(!timeout_seq_.empty());
 914     if (!timeout_seq_[0]) {
 915       SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 916       Send(reply_msg);
 917     } else {
 918       // Don't reply.
 919       delete reply_msg;
 920     }
 921     timeout_seq_.erase(timeout_seq_.begin());
 922     if (timeout_seq_.empty())
 923       Done();
 924   }
 925
 926  private:
 927   // Whether we should time-out or respond to the various messages we receive.
 928   std::vector<bool> timeout_seq_;
 929 };
 930
 931 void SendWithTimeoutOK(bool pump_during_send) {
 932   std::vector<Worker*> workers;
 933   std::vector<bool> timeout_seq;
 934   timeout_seq.push_back(false);
 935   timeout_seq.push_back(false);
 936   timeout_seq.push_back(false);
 937   workers.push_back(new TimeoutServer(5000, timeout_seq, pump_during_send));
 938   workers.push_back(new SimpleClient());
 939   RunTest(workers);
 940 }
 941
 942 void SendWithTimeoutTimeout(bool pump_during_send) {
 943   std::vector<Worker*> workers;
 944   std::vector<bool> timeout_seq;
 945   timeout_seq.push_back(true);
 946   timeout_seq.push_back(false);
 947   timeout_seq.push_back(false);
 948   workers.push_back(new TimeoutServer(100, timeout_seq, pump_during_send));
 949   workers.push_back(new UnresponsiveClient(timeout_seq));
 950   RunTest(workers);
 951 }
 952
 953 void SendWithTimeoutMixedOKAndTimeout(bool pump_during_send) {
 954   std::vector<Worker*> workers;
 955   std::vector<bool> timeout_seq;
 956   timeout_seq.push_back(true);
 957   timeout_seq.push_back(false);
 958   timeout_seq.push_back(false);
 959   timeout_seq.push_back(true);
 960   timeout_seq.push_back(false);
 961   workers.push_back(new TimeoutServer(100, timeout_seq, pump_during_send));
 962   workers.push_back(new UnresponsiveClient(timeout_seq));
 963   RunTest(workers);
 964 }
 965
 966 // Tests that SendWithTimeout does not time-out if the response comes back fast
 967 // enough.
 968 TEST_F(IPCSyncChannelTest, SendWithTimeoutOK) {
 969   SendWithTimeoutOK(false);
 970   SendWithTimeoutOK(true);
 971 }
 972
 973 // Tests that SendWithTimeout does time-out.
 974 TEST_F(IPCSyncChannelTest, SendWithTimeoutTimeout) {
 975   SendWithTimeoutTimeout(false);
 976   SendWithTimeoutTimeout(true);
 977 }
 978
 979 // Sends some message that time-out and some that succeed.
 980 TEST_F(IPCSyncChannelTest, SendWithTimeoutMixedOKAndTimeout) {
 981   SendWithTimeoutMixedOKAndTimeout(false);
 982   SendWithTimeoutMixedOKAndTimeout(true);
 983 }
 984
 985 //------------------------------------------------------------------------------
 986
 987 void NestedCallback(Worker* server) {
 988   // Sleep a bit so that we wake up after the reply has been received.
 989   base::PlatformThread::Sleep(base::TimeDelta::FromMilliseconds(250));
 990   server->SendAnswerToLife(true, base::kNoTimeout, true);
 991 }
 992
 993 bool timeout_occurred = false;
 994
 995 void TimeoutCallback() {
 996   timeout_occurred = true;
 997 }
 998
 999 class DoneEventRaceServer : public Worker {
1000  public:
1001   DoneEventRaceServer()
1002       : Worker(Channel::MODE_SERVER, "done_event_race_server") { }
1003
1004   virtual void Run() OVERRIDE {
1005     MessageLoop::current()->PostTask(FROM_HERE,
1006                                      base::Bind(&NestedCallback, this));
1007     MessageLoop::current()->PostDelayedTask(
1008         FROM_HERE,
1009         base::Bind(&TimeoutCallback),
1010         base::TimeDelta::FromSeconds(9));
1011     // Even though we have a timeout on the Send, it will succeed since for this
1012     // bug, the reply message comes back and is deserialized, however the done
1013     // event wasn't set.  So we indirectly use the timeout task to notice if a
1014     // timeout occurred.
1015     SendAnswerToLife(true, 10000, true);
1016     DCHECK(!timeout_occurred);
1017     Done();
1018   }
1019 };
1020
1021 // Tests http://b/1474092 - that if after the done_event is set but before
1022 // OnObjectSignaled is called another message is sent out, then after its
1023 // reply comes back OnObjectSignaled will be called for the first message.
1024 TEST_F(IPCSyncChannelTest, DoneEventRace) {
1025   std::vector<Worker*> workers;
1026   workers.push_back(new DoneEventRaceServer());
1027   workers.push_back(new SimpleClient());
1028   RunTest(workers);
1029 }
1030
1031 //------------------------------------------------------------------------------
1032
1033 class TestSyncMessageFilter : public SyncMessageFilter {
1034  public:
1035   TestSyncMessageFilter(base::WaitableEvent* shutdown_event,
1036                         Worker* worker,
1037                         scoped_refptr<base::MessageLoopProxy> message_loop)
1038       : SyncMessageFilter(shutdown_event),
1039         worker_(worker),
1040         message_loop_(message_loop) {
1041   }
1042
1043   virtual void OnFilterAdded(Channel* channel) OVERRIDE {
1044     SyncMessageFilter::OnFilterAdded(channel);
1045     message_loop_->PostTask(
1046         FROM_HERE,
1047         base::Bind(&TestSyncMessageFilter::SendMessageOnHelperThread, this));
1048   }
1049
1050   void SendMessageOnHelperThread() {
1051     int answer = 0;
1052     bool result = Send(new SyncChannelTestMsg_AnswerToLife(&answer));
1053     DCHECK(result);
1054     DCHECK_EQ(answer, 42);
1055
1056     worker_->Done();
1057   }
1058
1059  private:
1060   virtual ~TestSyncMessageFilter() {}
1061
1062   Worker* worker_;
1063   scoped_refptr<base::MessageLoopProxy> message_loop_;
1064 };
1065
1066 class SyncMessageFilterServer : public Worker {
1067  public:
1068   SyncMessageFilterServer()
1069       : Worker(Channel::MODE_SERVER, "sync_message_filter_server"),
1070         thread_("helper_thread") {
1071     base::Thread::Options options;
1072     options.message_loop_type = MessageLoop::TYPE_DEFAULT;
1073     thread_.StartWithOptions(options);
1074     filter_ = new TestSyncMessageFilter(shutdown_event(), this,
1075                                         thread_.message_loop_proxy());
1076   }
1077
1078   virtual void Run() OVERRIDE {
1079     channel()->AddFilter(filter_.get());
1080   }
1081
1082   base::Thread thread_;
1083   scoped_refptr<TestSyncMessageFilter> filter_;
1084 };
1085
1086 // This class provides functionality to test the case that a Send on the sync
1087 // channel does not crash after the channel has been closed.
1088 class ServerSendAfterClose : public Worker {
1089  public:
1090   ServerSendAfterClose()
1091      : Worker(Channel::MODE_SERVER, "simpler_server"),
1092        send_result_(true) {
1093   }
1094
1095   bool SendDummy() {
1096     ListenerThread()->message_loop()->PostTask(
1097         FROM_HERE, base::Bind(base::IgnoreResult(&ServerSendAfterClose::Send),
1098                               this, new SyncChannelTestMsg_NoArgs));
1099     return true;
1100   }
1101
1102   bool send_result() const {
1103     return send_result_;
1104   }
1105
1106  private:
1107   virtual void Run() OVERRIDE {
1108     CloseChannel();
1109     Done();
1110   }
1111
1112   virtual bool Send(Message* msg) OVERRIDE {
1113     send_result_ = Worker::Send(msg);
1114     Done();
1115     return send_result_;
1116   }
1117
1118   bool send_result_;
1119 };
1120
1121 // Tests basic synchronous call
1122 TEST_F(IPCSyncChannelTest, SyncMessageFilter) {
1123   std::vector<Worker*> workers;
1124   workers.push_back(new SyncMessageFilterServer());
1125   workers.push_back(new SimpleClient());
1126   RunTest(workers);
1127 }
1128
1129 // Test the case when the channel is closed and a Send is attempted after that.
1130 TEST_F(IPCSyncChannelTest, SendAfterClose) {
1131   ServerSendAfterClose server;
1132   server.Start();
1133
1134   server.done_event()->Wait();
1135   server.done_event()->Reset();
1136
1137   server.SendDummy();
1138   server.done_event()->Wait();
1139
1140   EXPECT_FALSE(server.send_result());
1141
1142   server.Shutdown();
1143 }
1144
1145 //------------------------------------------------------------------------------
1146
1147 class RestrictedDispatchServer : public Worker {
1148  public:
1149   RestrictedDispatchServer(WaitableEvent* sent_ping_event,
1150                            WaitableEvent* wait_event)
1151       : Worker("restricted_channel", Channel::MODE_SERVER),
1152         sent_ping_event_(sent_ping_event),
1153         wait_event_(wait_event) { }
1154
1155   void OnDoPing(int ping) {
1156     // Send an asynchronous message that unblocks the caller.
1157     Message* msg = new SyncChannelTestMsg_Ping(ping);
1158     msg->set_unblock(true);
1159     Send(msg);
1160     // Signal the event after the message has been sent on the channel, on the
1161     // IPC thread.
1162     ipc_thread().message_loop()->PostTask(
1163         FROM_HERE, base::Bind(&RestrictedDispatchServer::OnPingSent, this));
1164   }
1165
1166   void OnPingTTL(int ping, int* out) {
1167     *out = ping;
1168     wait_event_->Wait();
1169   }
1170
1171   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1172
1173  private:
1174   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1175     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchServer, message)
1176      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1177      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_PingTTL, OnPingTTL)
1178      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1179     IPC_END_MESSAGE_MAP()
1180     return true;
1181   }
1182
1183   void OnPingSent() {
1184     sent_ping_event_->Signal();
1185   }
1186
1187   void OnNoArgs() { }
1188   WaitableEvent* sent_ping_event_;
1189   WaitableEvent* wait_event_;
1190 };
1191
1192 class NonRestrictedDispatchServer : public Worker {
1193  public:
1194   NonRestrictedDispatchServer(WaitableEvent* signal_event)
1195       : Worker("non_restricted_channel", Channel::MODE_SERVER),
1196         signal_event_(signal_event) {}
1197
1198   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1199
1200   void OnDoPingTTL(int ping) {
1201     int value = 0;
1202     Send(new SyncChannelTestMsg_PingTTL(ping, &value));
1203     signal_event_->Signal();
1204   }
1205
1206  private:
1207   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1208     IPC_BEGIN_MESSAGE_MAP(NonRestrictedDispatchServer, message)
1209      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1210      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1211     IPC_END_MESSAGE_MAP()
1212     return true;
1213   }
1214
1215   void OnNoArgs() { }
1216   WaitableEvent* signal_event_;
1217 };
1218
1219 class RestrictedDispatchClient : public Worker {
1220  public:
1221   RestrictedDispatchClient(WaitableEvent* sent_ping_event,
1222                            RestrictedDispatchServer* server,
1223                            NonRestrictedDispatchServer* server2,
1224                            int* success)
1225       : Worker("restricted_channel", Channel::MODE_CLIENT),
1226         ping_(0),
1227         server_(server),
1228         server2_(server2),
1229         success_(success),
1230         sent_ping_event_(sent_ping_event) {}
1231
1232   virtual void Run() OVERRIDE {
1233     // Incoming messages from our channel should only be dispatched when we
1234     // send a message on that same channel.
1235     channel()->SetRestrictDispatchChannelGroup(1);
1236
1237     server_->ListenerThread()->message_loop()->PostTask(
1238         FROM_HERE, base::Bind(&RestrictedDispatchServer::OnDoPing, server_, 1));
1239     sent_ping_event_->Wait();
1240     Send(new SyncChannelTestMsg_NoArgs);
1241     if (ping_ == 1)
1242       ++*success_;
1243     else
1244       LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
1245
1246     non_restricted_channel_.reset(new SyncChannel(
1247         "non_restricted_channel", Channel::MODE_CLIENT, this,
1248         ipc_thread().message_loop_proxy(), true, shutdown_event()));
1249
1250     server_->ListenerThread()->message_loop()->PostTask(
1251         FROM_HERE, base::Bind(&RestrictedDispatchServer::OnDoPing, server_, 2));
1252     sent_ping_event_->Wait();
1253     // Check that the incoming message is *not* dispatched when sending on the
1254     // non restricted channel.
1255     // TODO(piman): there is a possibility of a false positive race condition
1256     // here, if the message that was posted on the server-side end of the pipe
1257     // is not visible yet on the client side, but I don't know how to solve this
1258     // without hooking into the internals of SyncChannel. I haven't seen it in
1259     // practice (i.e. not setting SetRestrictDispatchToSameChannel does cause
1260     // the following to fail).
1261     non_restricted_channel_->Send(new SyncChannelTestMsg_NoArgs);
1262     if (ping_ == 1)
1263       ++*success_;
1264     else
1265       LOG(ERROR) << "Send dispatched message from restricted channel";
1266
1267     Send(new SyncChannelTestMsg_NoArgs);
1268     if (ping_ == 2)
1269       ++*success_;
1270     else
1271       LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
1272
1273     // Check that the incoming message on the non-restricted channel is
1274     // dispatched when sending on the restricted channel.
1275     server2_->ListenerThread()->message_loop()->PostTask(
1276         FROM_HERE,
1277         base::Bind(&NonRestrictedDispatchServer::OnDoPingTTL, server2_, 3));
1278     int value = 0;
1279     Send(new SyncChannelTestMsg_PingTTL(4, &value));
1280     if (ping_ == 3 && value == 4)
1281       ++*success_;
1282     else
1283       LOG(ERROR) << "Send failed to dispatch message from unrestricted channel";
1284
1285     non_restricted_channel_->Send(new SyncChannelTestMsg_Done);
1286     non_restricted_channel_.reset();
1287     Send(new SyncChannelTestMsg_Done);
1288     Done();
1289   }
1290
1291  private:
1292   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1293     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchClient, message)
1294      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Ping, OnPing)
1295      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_PingTTL, OnPingTTL)
1296     IPC_END_MESSAGE_MAP()
1297     return true;
1298   }
1299
1300   void OnPing(int ping) {
1301     ping_ = ping;
1302   }
1303
1304   void OnPingTTL(int ping, IPC::Message* reply) {
1305     ping_ = ping;
1306     // This message comes from the NonRestrictedDispatchServer, we have to send
1307     // the reply back manually.
1308     SyncChannelTestMsg_PingTTL::WriteReplyParams(reply, ping);
1309     non_restricted_channel_->Send(reply);
1310   }
1311
1312   int ping_;
1313   RestrictedDispatchServer* server_;
1314   NonRestrictedDispatchServer* server2_;
1315   int* success_;
1316   WaitableEvent* sent_ping_event_;
1317   scoped_ptr<SyncChannel> non_restricted_channel_;
1318 };
1319
1320 TEST_F(IPCSyncChannelTest, RestrictedDispatch) {
1321   WaitableEvent sent_ping_event(false, false);
1322   WaitableEvent wait_event(false, false);
1323   RestrictedDispatchServer* server =
1324       new RestrictedDispatchServer(&sent_ping_event, &wait_event);
1325   NonRestrictedDispatchServer* server2 =
1326       new NonRestrictedDispatchServer(&wait_event);
1327
1328   int success = 0;
1329   std::vector<Worker*> workers;
1330   workers.push_back(server);
1331   workers.push_back(server2);
1332   workers.push_back(new RestrictedDispatchClient(
1333       &sent_ping_event, server, server2, &success));
1334   RunTest(workers);
1335   EXPECT_EQ(4, success);
1336 }
1337
1338 //------------------------------------------------------------------------------
1339
1340 // This test case inspired by crbug.com/108491
1341 // We create two servers that use the same ListenerThread but have
1342 // SetRestrictDispatchToSameChannel set to true.
1343 // We create clients, then use some specific WaitableEvent wait/signalling to
1344 // ensure that messages get dispatched in a way that causes a deadlock due to
1345 // a nested dispatch and an eligible message in a higher-level dispatch's
1346 // delayed_queue. Specifically, we start with client1 about so send an
1347 // unblocking message to server1, while the shared listener thread for the
1348 // servers server1 and server2 is about to send a non-unblocking message to
1349 // client1. At the same time, client2 will be about to send an unblocking
1350 // message to server2. Server1 will handle the client1->server1 message by
1351 // telling server2 to send a non-unblocking message to client2.
1352 // What should happen is that the send to server2 should find the pending,
1353 // same-context client2->server2 message to dispatch, causing client2 to
1354 // unblock then handle the server2->client2 message, so that the shared
1355 // servers' listener thread can then respond to the client1->server1 message.
1356 // Then client1 can handle the non-unblocking server1->client1 message.
1357 // The old code would end up in a state where the server2->client2 message is
1358 // sent, but the client2->server2 message (which is eligible for dispatch, and
1359 // which is what client2 is waiting for) is stashed in a local delayed_queue
1360 // that has server1's channel context, causing a deadlock.
1361 // WaitableEvents in the events array are used to:
1362 //   event 0: indicate to client1 that server listener is in OnDoServerTask
1363 //   event 1: indicate to client1 that client2 listener is in OnDoClient2Task
1364 //   event 2: indicate to server1 that client2 listener is in OnDoClient2Task
1365 //   event 3: indicate to client2 that server listener is in OnDoServerTask
1366
1367 class RestrictedDispatchDeadlockServer : public Worker {
1368  public:
1369   RestrictedDispatchDeadlockServer(int server_num,
1370                                    WaitableEvent* server_ready_event,
1371                                    WaitableEvent** events,
1372                                    RestrictedDispatchDeadlockServer* peer)
1373       : Worker(server_num == 1 ? "channel1" : "channel2", Channel::MODE_SERVER),
1374         server_num_(server_num),
1375         server_ready_event_(server_ready_event),
1376         events_(events),
1377         peer_(peer) { }
1378
1379   void OnDoServerTask() {
1380     events_[3]->Signal();
1381     events_[2]->Wait();
1382     events_[0]->Signal();
1383     SendMessageToClient();
1384   }
1385
1386   virtual void Run() OVERRIDE {
1387     channel()->SetRestrictDispatchChannelGroup(1);
1388     server_ready_event_->Signal();
1389   }
1390
1391   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1392
1393  private:
1394   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1395     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockServer, message)
1396      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1397      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1398     IPC_END_MESSAGE_MAP()
1399     return true;
1400   }
1401
1402   void OnNoArgs() {
1403     if (server_num_ == 1) {
1404       DCHECK(peer_ != NULL);
1405       peer_->SendMessageToClient();
1406     }
1407   }
1408
1409   void SendMessageToClient() {
1410     Message* msg = new SyncChannelTestMsg_NoArgs;
1411     msg->set_unblock(false);
1412     DCHECK(!msg->should_unblock());
1413     Send(msg);
1414   }
1415
1416   int server_num_;
1417   WaitableEvent* server_ready_event_;
1418   WaitableEvent** events_;
1419   RestrictedDispatchDeadlockServer* peer_;
1420 };
1421
1422 class RestrictedDispatchDeadlockClient2 : public Worker {
1423  public:
1424   RestrictedDispatchDeadlockClient2(RestrictedDispatchDeadlockServer* server,
1425                                     WaitableEvent* server_ready_event,
1426                                     WaitableEvent** events)
1427       : Worker("channel2", Channel::MODE_CLIENT),
1428         server_ready_event_(server_ready_event),
1429         events_(events),
1430         received_msg_(false),
1431         received_noarg_reply_(false),
1432         done_issued_(false) {}
1433
1434   virtual void Run() OVERRIDE {
1435     server_ready_event_->Wait();
1436   }
1437
1438   void OnDoClient2Task() {
1439     events_[3]->Wait();
1440     events_[1]->Signal();
1441     events_[2]->Signal();
1442     DCHECK(received_msg_ == false);
1443
1444     Message* message = new SyncChannelTestMsg_NoArgs;
1445     message->set_unblock(true);
1446     Send(message);
1447     received_noarg_reply_ = true;
1448   }
1449
1450   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1451  private:
1452   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1453     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockClient2, message)
1454      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1455     IPC_END_MESSAGE_MAP()
1456     return true;
1457   }
1458
1459   void OnNoArgs() {
1460     received_msg_ = true;
1461     PossiblyDone();
1462   }
1463
1464   void PossiblyDone() {
1465     if (received_noarg_reply_ && received_msg_) {
1466       DCHECK(done_issued_ == false);
1467       done_issued_ = true;
1468       Send(new SyncChannelTestMsg_Done);
1469       Done();
1470     }
1471   }
1472
1473   WaitableEvent* server_ready_event_;
1474   WaitableEvent** events_;
1475   bool received_msg_;
1476   bool received_noarg_reply_;
1477   bool done_issued_;
1478 };
1479
1480 class RestrictedDispatchDeadlockClient1 : public Worker {
1481  public:
1482   RestrictedDispatchDeadlockClient1(RestrictedDispatchDeadlockServer* server,
1483                                     RestrictedDispatchDeadlockClient2* peer,
1484                                     WaitableEvent* server_ready_event,
1485                                     WaitableEvent** events)
1486       : Worker("channel1", Channel::MODE_CLIENT),
1487         server_(server),
1488         peer_(peer),
1489         server_ready_event_(server_ready_event),
1490         events_(events),
1491         received_msg_(false),
1492         received_noarg_reply_(false),
1493         done_issued_(false) {}
1494
1495   virtual void Run() OVERRIDE {
1496     server_ready_event_->Wait();
1497     server_->ListenerThread()->message_loop()->PostTask(
1498         FROM_HERE,
1499         base::Bind(&RestrictedDispatchDeadlockServer::OnDoServerTask, server_));
1500     peer_->ListenerThread()->message_loop()->PostTask(
1501         FROM_HERE,
1502         base::Bind(&RestrictedDispatchDeadlockClient2::OnDoClient2Task, peer_));
1503     events_[0]->Wait();
1504     events_[1]->Wait();
1505     DCHECK(received_msg_ == false);
1506
1507     Message* message = new SyncChannelTestMsg_NoArgs;
1508     message->set_unblock(true);
1509     Send(message);
1510     received_noarg_reply_ = true;
1511     PossiblyDone();
1512   }
1513
1514   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1515  private:
1516   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1517     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockClient1, message)
1518      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1519     IPC_END_MESSAGE_MAP()
1520     return true;
1521   }
1522
1523   void OnNoArgs() {
1524     received_msg_ = true;
1525     PossiblyDone();
1526   }
1527
1528   void PossiblyDone() {
1529     if (received_noarg_reply_ && received_msg_) {
1530       DCHECK(done_issued_ == false);
1531       done_issued_ = true;
1532       Send(new SyncChannelTestMsg_Done);
1533       Done();
1534     }
1535   }
1536
1537   RestrictedDispatchDeadlockServer* server_;
1538   RestrictedDispatchDeadlockClient2* peer_;
1539   WaitableEvent* server_ready_event_;
1540   WaitableEvent** events_;
1541   bool received_msg_;
1542   bool received_noarg_reply_;
1543   bool done_issued_;
1544 };
1545
1546 TEST_F(IPCSyncChannelTest, RestrictedDispatchDeadlock) {
1547   std::vector<Worker*> workers;
1548
1549   // A shared worker thread so that server1 and server2 run on one thread.
1550   base::Thread worker_thread("RestrictedDispatchDeadlock");
1551   ASSERT_TRUE(worker_thread.Start());
1552
1553   WaitableEvent server1_ready(false, false);
1554   WaitableEvent server2_ready(false, false);
1555
1556   WaitableEvent event0(false, false);
1557   WaitableEvent event1(false, false);
1558   WaitableEvent event2(false, false);
1559   WaitableEvent event3(false, false);
1560   WaitableEvent* events[4] = {&event0, &event1, &event2, &event3};
1561
1562   RestrictedDispatchDeadlockServer* server1;
1563   RestrictedDispatchDeadlockServer* server2;
1564   RestrictedDispatchDeadlockClient1* client1;
1565   RestrictedDispatchDeadlockClient2* client2;
1566
1567   server2 = new RestrictedDispatchDeadlockServer(2, &server2_ready, events,
1568                                                  NULL);
1569   server2->OverrideThread(&worker_thread);
1570   workers.push_back(server2);
1571
1572   client2 = new RestrictedDispatchDeadlockClient2(server2, &server2_ready,
1573                                                   events);
1574   workers.push_back(client2);
1575
1576   server1 = new RestrictedDispatchDeadlockServer(1, &server1_ready, events,
1577                                                  server2);
1578   server1->OverrideThread(&worker_thread);
1579   workers.push_back(server1);
1580
1581   client1 = new RestrictedDispatchDeadlockClient1(server1, client2,
1582                                                   &server1_ready, events);
1583   workers.push_back(client1);
1584
1585   RunTest(workers);
1586 }
1587
1588 //------------------------------------------------------------------------------
1589
1590 // This test case inspired by crbug.com/120530
1591 // We create 4 workers that pipe to each other W1->W2->W3->W4->W1 then we send a
1592 // message that recurses through 3, 4 or 5 steps to make sure, say, W1 can
1593 // re-enter when called from W4 while it's sending a message to W2.
1594 // The first worker drives the whole test so it must be treated specially.
1595
1596 class RestrictedDispatchPipeWorker : public Worker {
1597  public:
1598   RestrictedDispatchPipeWorker(
1599       const std::string &channel1,
1600       WaitableEvent* event1,
1601       const std::string &channel2,
1602       WaitableEvent* event2,
1603       int group,
1604       int* success)
1605       : Worker(channel1, Channel::MODE_SERVER),
1606         event1_(event1),
1607         event2_(event2),
1608         other_channel_name_(channel2),
1609         group_(group),
1610         success_(success) {
1611   }
1612
1613   void OnPingTTL(int ping, int* ret) {
1614     *ret = 0;
1615     if (!ping)
1616       return;
1617     other_channel_->Send(new SyncChannelTestMsg_PingTTL(ping - 1, ret));
1618     ++*ret;
1619   }
1620
1621   void OnDone() {
1622     if (is_first())
1623       return;
1624     other_channel_->Send(new SyncChannelTestMsg_Done);
1625     other_channel_.reset();
1626     Done();
1627   }
1628
1629   virtual void Run() OVERRIDE {
1630     channel()->SetRestrictDispatchChannelGroup(group_);
1631     if (is_first())
1632       event1_->Signal();
1633     event2_->Wait();
1634     other_channel_.reset(new SyncChannel(
1635         other_channel_name_, Channel::MODE_CLIENT, this,
1636         ipc_thread().message_loop_proxy(), true, shutdown_event()));
1637     other_channel_->SetRestrictDispatchChannelGroup(group_);
1638     if (!is_first()) {
1639       event1_->Signal();
1640       return;
1641     }
1642     *success_ = 0;
1643     int value = 0;
1644     OnPingTTL(3, &value);
1645     *success_ += (value == 3);
1646     OnPingTTL(4, &value);
1647     *success_ += (value == 4);
1648     OnPingTTL(5, &value);
1649     *success_ += (value == 5);
1650     other_channel_->Send(new SyncChannelTestMsg_Done);
1651     other_channel_.reset();
1652     Done();
1653   }
1654
1655   bool is_first() { return !!success_; }
1656
1657  private:
1658   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1659     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchPipeWorker, message)
1660      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_PingTTL, OnPingTTL)
1661      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, OnDone)
1662     IPC_END_MESSAGE_MAP()
1663     return true;
1664   }
1665
1666   scoped_ptr<SyncChannel> other_channel_;
1667   WaitableEvent* event1_;
1668   WaitableEvent* event2_;
1669   std::string other_channel_name_;
1670   int group_;
1671   int* success_;
1672 };
1673
1674 TEST_F(IPCSyncChannelTest, RestrictedDispatch4WayDeadlock) {
1675   int success = 0;
1676   std::vector<Worker*> workers;
1677   WaitableEvent event0(true, false);
1678   WaitableEvent event1(true, false);
1679   WaitableEvent event2(true, false);
1680   WaitableEvent event3(true, false);
1681   workers.push_back(new RestrictedDispatchPipeWorker(
1682         "channel0", &event0, "channel1", &event1, 1, &success));
1683   workers.push_back(new RestrictedDispatchPipeWorker(
1684         "channel1", &event1, "channel2", &event2, 2, NULL));
1685   workers.push_back(new RestrictedDispatchPipeWorker(
1686         "channel2", &event2, "channel3", &event3, 3, NULL));
1687   workers.push_back(new RestrictedDispatchPipeWorker(
1688         "channel3", &event3, "channel0", &event0, 4, NULL));
1689   RunTest(workers);
1690   EXPECT_EQ(3, success);
1691 }
1692
1693 //------------------------------------------------------------------------------
1694
1695 // This test case inspired by crbug.com/122443
1696 // We want to make sure a reply message with the unblock flag set correctly
1697 // behaves as a reply, not a regular message.
1698 // We have 3 workers. Server1 will send a message to Server2 (which will block),
1699 // during which it will dispatch a message comming from Client, at which point
1700 // it will send another message to Server2. While sending that second message it
1701 // will receive a reply from Server1 with the unblock flag.
1702
1703 class ReentrantReplyServer1 : public Worker {
1704  public:
1705   ReentrantReplyServer1(WaitableEvent* server_ready)
1706       : Worker("reentrant_reply1", Channel::MODE_SERVER),
1707         server_ready_(server_ready) { }
1708
1709   virtual void Run() OVERRIDE {
1710     server2_channel_.reset(new SyncChannel(
1711         "reentrant_reply2", Channel::MODE_CLIENT, this,
1712         ipc_thread().message_loop_proxy(), true, shutdown_event()));
1713     server_ready_->Signal();
1714     Message* msg = new SyncChannelTestMsg_Reentrant1();
1715     server2_channel_->Send(msg);
1716     server2_channel_.reset();
1717     Done();
1718   }
1719
1720  private:
1721   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1722     IPC_BEGIN_MESSAGE_MAP(ReentrantReplyServer1, message)
1723      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Reentrant2, OnReentrant2)
1724      IPC_REPLY_HANDLER(OnReply)
1725     IPC_END_MESSAGE_MAP()
1726     return true;
1727   }
1728
1729   void OnReentrant2() {
1730     Message* msg = new SyncChannelTestMsg_Reentrant3();
1731     server2_channel_->Send(msg);
1732   }
1733
1734   void OnReply(const Message& message) {
1735     // If we get here, the Send() will never receive the reply (thus would
1736     // hang), so abort instead.
1737     LOG(FATAL) << "Reply message was dispatched";
1738   }
1739
1740   WaitableEvent* server_ready_;
1741   scoped_ptr<SyncChannel> server2_channel_;
1742 };
1743
1744 class ReentrantReplyServer2 : public Worker {
1745  public:
1746   ReentrantReplyServer2()
1747       : Worker("reentrant_reply2", Channel::MODE_SERVER),
1748         reply_(NULL) { }
1749
1750  private:
1751   virtual bool OnMessageReceived(const Message& message) OVERRIDE {
1752     IPC_BEGIN_MESSAGE_MAP(ReentrantReplyServer2, message)
1753      IPC_MESSAGE_HANDLER_DELAY_REPLY(
1754          SyncChannelTestMsg_Reentrant1, OnReentrant1)
1755      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Reentrant3, OnReentrant3)
1756     IPC_END_MESSAGE_MAP()
1757     return true;
1758   }
1759
1760   void OnReentrant1(Message* reply) {
1761     DCHECK(!reply_);
1762     reply_ = reply;
1763   }
1764
1765   void OnReentrant3() {
1766     DCHECK(reply_);
1767     Message* reply = reply_;
1768     reply_ = NULL;
1769     reply->set_unblock(true);
1770     Send(reply);
1771     Done();
1772   }
1773
1774   Message* reply_;
1775 };
1776
1777 class ReentrantReplyClient : public Worker {
1778  public:
1779   ReentrantReplyClient(WaitableEvent* server_ready)
1780       : Worker("reentrant_reply1", Channel::MODE_CLIENT),
1781         server_ready_(server_ready) { }
1782
1783   virtual void Run() OVERRIDE {
1784     server_ready_->Wait();
1785     Send(new SyncChannelTestMsg_Reentrant2());
1786     Done();
1787   }
1788
1789  private:
1790   WaitableEvent* server_ready_;
1791 };
1792
1793 TEST_F(IPCSyncChannelTest, ReentrantReply) {
1794   std::vector<Worker*> workers;
1795   WaitableEvent server_ready(false, false);
1796   workers.push_back(new ReentrantReplyServer2());
1797   workers.push_back(new ReentrantReplyServer1(&server_ready));
1798   workers.push_back(new ReentrantReplyClient(&server_ready));
1799   RunTest(workers);
1800 }
1801
1802 //------------------------------------------------------------------------------
1803
1804 // Generate a validated channel ID using Channel::GenerateVerifiedChannelID().
1805
1806 class VerifiedServer : public Worker {
1807  public:
1808   VerifiedServer(base::Thread* listener_thread,
1809                  const std::string& channel_name,
1810                  const std::string& reply_text)
1811       : Worker(channel_name, Channel::MODE_SERVER),
1812         reply_text_(reply_text) {
1813     Worker::OverrideThread(listener_thread);
1814   }
1815
1816   virtual void OnNestedTestMsg(Message* reply_msg) OVERRIDE {
1817     VLOG(1) << __FUNCTION__ << " Sending reply: " << reply_text_;
1818     SyncChannelNestedTestMsg_String::WriteReplyParams(reply_msg, reply_text_);
1819     Send(reply_msg);
1820     ASSERT_EQ(channel()->peer_pid(), base::GetCurrentProcId());
1821     Done();
1822   }
1823
1824  private:
1825   std::string reply_text_;
1826 };
1827
1828 class VerifiedClient : public Worker {
1829  public:
1830   VerifiedClient(base::Thread* listener_thread,
1831                  const std::string& channel_name,
1832                  const std::string& expected_text)
1833       : Worker(channel_name, Channel::MODE_CLIENT),
1834         expected_text_(expected_text) {
1835     Worker::OverrideThread(listener_thread);
1836   }
1837
1838   virtual void Run() OVERRIDE {
1839     std::string response;
1840     SyncMessage* msg = new SyncChannelNestedTestMsg_String(&response);
1841     bool result = Send(msg);
1842     DCHECK(result);
1843     DCHECK_EQ(response, expected_text_);
1844     // expected_text_ is only used in the above DCHECK. This line suppresses the
1845     // "unused private field" warning in release builds.
1846     (void)expected_text_;
1847
1848     VLOG(1) << __FUNCTION__ << " Received reply: " << response;
1849     ASSERT_EQ(channel()->peer_pid(), base::GetCurrentProcId());
1850     Done();
1851   }
1852
1853  private:
1854   std::string expected_text_;
1855 };
1856
1857 void Verified() {
1858   std::vector<Worker*> workers;
1859
1860   // A shared worker thread for servers
1861   base::Thread server_worker_thread("Verified_ServerListener");
1862   ASSERT_TRUE(server_worker_thread.Start());
1863
1864   base::Thread client_worker_thread("Verified_ClientListener");
1865   ASSERT_TRUE(client_worker_thread.Start());
1866
1867   std::string channel_id = Channel::GenerateVerifiedChannelID("Verified");
1868   Worker* worker;
1869
1870   worker = new VerifiedServer(&server_worker_thread,
1871                               channel_id,
1872                               "Got first message");
1873   workers.push_back(worker);
1874
1875   worker = new VerifiedClient(&client_worker_thread,
1876                               channel_id,
1877                               "Got first message");
1878   workers.push_back(worker);
1879
1880   RunTest(workers);
1881 }
1882
1883 // Windows needs to send an out-of-band secret to verify the client end of the
1884 // channel. Test that we still connect correctly in that case.
1885 TEST_F(IPCSyncChannelTest, Verified) {
1886   Verified();
1887 }
1888
1889 }  // namespace