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