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