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_);
 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 // Tests basic synchronous call
 305 TEST_F(IPCSyncChannelTest, Simple) {
 306   Simple(false);
 307   Simple(true);
 308 }
 309
 310 //------------------------------------------------------------------------------
 311
 312 // Worker classes which override how the sync channel is created to use the
 313 // two-step initialization (calling the lightweight constructor and then
 314 // ChannelProxy::Init separately) process.
 315 class TwoStepServer : public Worker {
 316  public:
 317   explicit TwoStepServer(bool create_pipe_now)
 318       : Worker(Channel::MODE_SERVER, "simpler_server"),
 319         create_pipe_now_(create_pipe_now) { }
 320
 321   void Run() override {
 322     SendAnswerToLife(false, true);
 323     Done();
 324   }
 325
 326   SyncChannel* CreateChannel() override {
 327     SyncChannel* channel =
 328         SyncChannel::Create(channel_name(), mode(), this,
 329                             ipc_thread().task_runner().get(), create_pipe_now_,
 330                             shutdown_event()).release();
 331     return channel;
 332   }
 333
 334   bool create_pipe_now_;
 335 };
 336
 337 class TwoStepClient : public Worker {
 338  public:
 339   TwoStepClient(bool create_pipe_now)
 340       : Worker(Channel::MODE_CLIENT, "simple_client"),
 341         create_pipe_now_(create_pipe_now) { }
 342
 343   void OnAnswer(int* answer) override {
 344     *answer = 42;
 345     Done();
 346   }
 347
 348   SyncChannel* CreateChannel() override {
 349     SyncChannel* channel =
 350         SyncChannel::Create(channel_name(), mode(), this,
 351                             ipc_thread().task_runner().get(), 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   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   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   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   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   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   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   void Run() override {
 537     SendDouble(pump_first_, expected_send_result_);
 538     Done();
 539   }
 540
 541   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   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   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   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   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   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   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   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   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   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   void Run() override {
 899     base::ThreadTaskRunnerHandle::Get()->PostTask(
 900         FROM_HERE, base::Bind(&NestedCallback, this));
 901     base::ThreadTaskRunnerHandle::Get()->PostDelayedTask(
 902         FROM_HERE, base::Bind(&TimeoutCallback),
 903         base::TimeDelta::FromSeconds(9));
 904     // Even though we have a timeout on the Send, it will succeed since for this
 905     // bug, the reply message comes back and is deserialized, however the done
 906     // event wasn't set.  So we indirectly use the timeout task to notice if a
 907     // timeout occurred.
 908     SendAnswerToLife(true, true);
 909     DCHECK(!timeout_occurred);
 910     Done();
 911   }
 912 };
 913
 914 // Tests http://b/1474092 - that if after the done_event is set but before
 915 // OnObjectSignaled is called another message is sent out, then after its
 916 // reply comes back OnObjectSignaled will be called for the first message.
 917 TEST_F(IPCSyncChannelTest, DoneEventRace) {
 918   std::vector<Worker*> workers;
 919   workers.push_back(new DoneEventRaceServer());
 920   workers.push_back(new SimpleClient());
 921   RunTest(workers);
 922 }
 923
 924 //------------------------------------------------------------------------------
 925
 926 class TestSyncMessageFilter : public SyncMessageFilter {
 927  public:
 928   TestSyncMessageFilter(
 929       base::WaitableEvent* shutdown_event,
 930       Worker* worker,
 931       scoped_refptr<base::SingleThreadTaskRunner> task_runner)
 932       : SyncMessageFilter(shutdown_event),
 933         worker_(worker),
 934         task_runner_(task_runner) {}
 935
 936   void OnFilterAdded(Sender* sender) override {
 937     SyncMessageFilter::OnFilterAdded(sender);
 938     task_runner_->PostTask(
 939         FROM_HERE,
 940         base::Bind(&TestSyncMessageFilter::SendMessageOnHelperThread, this));
 941   }
 942
 943   void SendMessageOnHelperThread() {
 944     int answer = 0;
 945     bool result = Send(new SyncChannelTestMsg_AnswerToLife(&answer));
 946     DCHECK(result);
 947     DCHECK_EQ(answer, 42);
 948
 949     worker_->Done();
 950   }
 951
 952  private:
 953   ~TestSyncMessageFilter() override {}
 954
 955   Worker* worker_;
 956   scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
 957 };
 958
 959 class SyncMessageFilterServer : public Worker {
 960  public:
 961   SyncMessageFilterServer()
 962       : Worker(Channel::MODE_SERVER, "sync_message_filter_server"),
 963         thread_("helper_thread") {
 964     base::Thread::Options options;
 965     options.message_loop_type = base::MessageLoop::TYPE_DEFAULT;
 966     thread_.StartWithOptions(options);
 967     filter_ = new TestSyncMessageFilter(shutdown_event(), this,
 968                                         thread_.task_runner());
 969   }
 970
 971   void Run() override {
 972     channel()->AddFilter(filter_.get());
 973   }
 974
 975   base::Thread thread_;
 976   scoped_refptr<TestSyncMessageFilter> filter_;
 977 };
 978
 979 // This class provides functionality to test the case that a Send on the sync
 980 // channel does not crash after the channel has been closed.
 981 class ServerSendAfterClose : public Worker {
 982  public:
 983   ServerSendAfterClose()
 984      : Worker(Channel::MODE_SERVER, "simpler_server"),
 985        send_result_(true) {
 986   }
 987
 988   bool SendDummy() {
 989     ListenerThread()->task_runner()->PostTask(
 990         FROM_HERE, base::Bind(base::IgnoreResult(&ServerSendAfterClose::Send),
 991                               this, new SyncChannelTestMsg_NoArgs));
 992     return true;
 993   }
 994
 995   bool send_result() const {
 996     return send_result_;
 997   }
 998
 999  private:
1000   void Run() override {
1001     CloseChannel();
1002     Done();
1003   }
1004
1005   bool Send(Message* msg) override {
1006     send_result_ = Worker::Send(msg);
1007     Done();
1008     return send_result_;
1009   }
1010
1011   bool send_result_;
1012 };
1013
1014 // Tests basic synchronous call
1015 TEST_F(IPCSyncChannelTest, SyncMessageFilter) {
1016   std::vector<Worker*> workers;
1017   workers.push_back(new SyncMessageFilterServer());
1018   workers.push_back(new SimpleClient());
1019   RunTest(workers);
1020 }
1021
1022 // Test the case when the channel is closed and a Send is attempted after that.
1023 TEST_F(IPCSyncChannelTest, SendAfterClose) {
1024   ServerSendAfterClose server;
1025   server.Start();
1026
1027   server.done_event()->Wait();
1028   server.done_event()->Reset();
1029
1030   server.SendDummy();
1031   server.done_event()->Wait();
1032
1033   EXPECT_FALSE(server.send_result());
1034
1035   server.Shutdown();
1036 }
1037
1038 //------------------------------------------------------------------------------
1039
1040 class RestrictedDispatchServer : public Worker {
1041  public:
1042   RestrictedDispatchServer(WaitableEvent* sent_ping_event,
1043                            WaitableEvent* wait_event)
1044       : Worker("restricted_channel", Channel::MODE_SERVER),
1045         sent_ping_event_(sent_ping_event),
1046         wait_event_(wait_event) { }
1047
1048   void OnDoPing(int ping) {
1049     // Send an asynchronous message that unblocks the caller.
1050     Message* msg = new SyncChannelTestMsg_Ping(ping);
1051     msg->set_unblock(true);
1052     Send(msg);
1053     // Signal the event after the message has been sent on the channel, on the
1054     // IPC thread.
1055     ipc_thread().task_runner()->PostTask(
1056         FROM_HERE, base::Bind(&RestrictedDispatchServer::OnPingSent, this));
1057   }
1058
1059   void OnPingTTL(int ping, int* out) {
1060     *out = ping;
1061     wait_event_->Wait();
1062   }
1063
1064   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1065
1066  private:
1067   bool OnMessageReceived(const Message& message) override {
1068     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchServer, message)
1069      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1070      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_PingTTL, OnPingTTL)
1071      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1072     IPC_END_MESSAGE_MAP()
1073     return true;
1074   }
1075
1076   void OnPingSent() {
1077     sent_ping_event_->Signal();
1078   }
1079
1080   void OnNoArgs() { }
1081   WaitableEvent* sent_ping_event_;
1082   WaitableEvent* wait_event_;
1083 };
1084
1085 class NonRestrictedDispatchServer : public Worker {
1086  public:
1087   NonRestrictedDispatchServer(WaitableEvent* signal_event)
1088       : Worker("non_restricted_channel", Channel::MODE_SERVER),
1089         signal_event_(signal_event) {}
1090
1091   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1092
1093   void OnDoPingTTL(int ping) {
1094     int value = 0;
1095     Send(new SyncChannelTestMsg_PingTTL(ping, &value));
1096     signal_event_->Signal();
1097   }
1098
1099  private:
1100   bool OnMessageReceived(const Message& message) override {
1101     IPC_BEGIN_MESSAGE_MAP(NonRestrictedDispatchServer, message)
1102      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1103      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1104     IPC_END_MESSAGE_MAP()
1105     return true;
1106   }
1107
1108   void OnNoArgs() { }
1109   WaitableEvent* signal_event_;
1110 };
1111
1112 class RestrictedDispatchClient : public Worker {
1113  public:
1114   RestrictedDispatchClient(WaitableEvent* sent_ping_event,
1115                            RestrictedDispatchServer* server,
1116                            NonRestrictedDispatchServer* server2,
1117                            int* success)
1118       : Worker("restricted_channel", Channel::MODE_CLIENT),
1119         ping_(0),
1120         server_(server),
1121         server2_(server2),
1122         success_(success),
1123         sent_ping_event_(sent_ping_event) {}
1124
1125   void Run() override {
1126     // Incoming messages from our channel should only be dispatched when we
1127     // send a message on that same channel.
1128     channel()->SetRestrictDispatchChannelGroup(1);
1129
1130     server_->ListenerThread()->task_runner()->PostTask(
1131         FROM_HERE, base::Bind(&RestrictedDispatchServer::OnDoPing, server_, 1));
1132     sent_ping_event_->Wait();
1133     Send(new SyncChannelTestMsg_NoArgs);
1134     if (ping_ == 1)
1135       ++*success_;
1136     else
1137       LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
1138
1139     non_restricted_channel_ = SyncChannel::Create(
1140         "non_restricted_channel", IPC::Channel::MODE_CLIENT, this,
1141         ipc_thread().task_runner().get(), true, shutdown_event());
1142
1143     server_->ListenerThread()->task_runner()->PostTask(
1144         FROM_HERE, base::Bind(&RestrictedDispatchServer::OnDoPing, server_, 2));
1145     sent_ping_event_->Wait();
1146     // Check that the incoming message is *not* dispatched when sending on the
1147     // non restricted channel.
1148     // TODO(piman): there is a possibility of a false positive race condition
1149     // here, if the message that was posted on the server-side end of the pipe
1150     // is not visible yet on the client side, but I don't know how to solve this
1151     // without hooking into the internals of SyncChannel. I haven't seen it in
1152     // practice (i.e. not setting SetRestrictDispatchToSameChannel does cause
1153     // the following to fail).
1154     non_restricted_channel_->Send(new SyncChannelTestMsg_NoArgs);
1155     if (ping_ == 1)
1156       ++*success_;
1157     else
1158       LOG(ERROR) << "Send dispatched message from restricted channel";
1159
1160     Send(new SyncChannelTestMsg_NoArgs);
1161     if (ping_ == 2)
1162       ++*success_;
1163     else
1164       LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
1165
1166     // Check that the incoming message on the non-restricted channel is
1167     // dispatched when sending on the restricted channel.
1168     server2_->ListenerThread()->task_runner()->PostTask(
1169         FROM_HERE,
1170         base::Bind(&NonRestrictedDispatchServer::OnDoPingTTL, server2_, 3));
1171     int value = 0;
1172     Send(new SyncChannelTestMsg_PingTTL(4, &value));
1173     if (ping_ == 3 && value == 4)
1174       ++*success_;
1175     else
1176       LOG(ERROR) << "Send failed to dispatch message from unrestricted channel";
1177
1178     non_restricted_channel_->Send(new SyncChannelTestMsg_Done);
1179     non_restricted_channel_.reset();
1180     Send(new SyncChannelTestMsg_Done);
1181     Done();
1182   }
1183
1184  private:
1185   bool OnMessageReceived(const Message& message) override {
1186     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchClient, message)
1187      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Ping, OnPing)
1188      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_PingTTL, OnPingTTL)
1189     IPC_END_MESSAGE_MAP()
1190     return true;
1191   }
1192
1193   void OnPing(int ping) {
1194     ping_ = ping;
1195   }
1196
1197   void OnPingTTL(int ping, IPC::Message* reply) {
1198     ping_ = ping;
1199     // This message comes from the NonRestrictedDispatchServer, we have to send
1200     // the reply back manually.
1201     SyncChannelTestMsg_PingTTL::WriteReplyParams(reply, ping);
1202     non_restricted_channel_->Send(reply);
1203   }
1204
1205   int ping_;
1206   RestrictedDispatchServer* server_;
1207   NonRestrictedDispatchServer* server2_;
1208   int* success_;
1209   WaitableEvent* sent_ping_event_;
1210   scoped_ptr<SyncChannel> non_restricted_channel_;
1211 };
1212
1213 TEST_F(IPCSyncChannelTest, RestrictedDispatch) {
1214   WaitableEvent sent_ping_event(false, false);
1215   WaitableEvent wait_event(false, false);
1216   RestrictedDispatchServer* server =
1217       new RestrictedDispatchServer(&sent_ping_event, &wait_event);
1218   NonRestrictedDispatchServer* server2 =
1219       new NonRestrictedDispatchServer(&wait_event);
1220
1221   int success = 0;
1222   std::vector<Worker*> workers;
1223   workers.push_back(server);
1224   workers.push_back(server2);
1225   workers.push_back(new RestrictedDispatchClient(
1226       &sent_ping_event, server, server2, &success));
1227   RunTest(workers);
1228   EXPECT_EQ(4, success);
1229 }
1230
1231 //------------------------------------------------------------------------------
1232
1233 // This test case inspired by crbug.com/108491
1234 // We create two servers that use the same ListenerThread but have
1235 // SetRestrictDispatchToSameChannel set to true.
1236 // We create clients, then use some specific WaitableEvent wait/signalling to
1237 // ensure that messages get dispatched in a way that causes a deadlock due to
1238 // a nested dispatch and an eligible message in a higher-level dispatch's
1239 // delayed_queue. Specifically, we start with client1 about so send an
1240 // unblocking message to server1, while the shared listener thread for the
1241 // servers server1 and server2 is about to send a non-unblocking message to
1242 // client1. At the same time, client2 will be about to send an unblocking
1243 // message to server2. Server1 will handle the client1->server1 message by
1244 // telling server2 to send a non-unblocking message to client2.
1245 // What should happen is that the send to server2 should find the pending,
1246 // same-context client2->server2 message to dispatch, causing client2 to
1247 // unblock then handle the server2->client2 message, so that the shared
1248 // servers' listener thread can then respond to the client1->server1 message.
1249 // Then client1 can handle the non-unblocking server1->client1 message.
1250 // The old code would end up in a state where the server2->client2 message is
1251 // sent, but the client2->server2 message (which is eligible for dispatch, and
1252 // which is what client2 is waiting for) is stashed in a local delayed_queue
1253 // that has server1's channel context, causing a deadlock.
1254 // WaitableEvents in the events array are used to:
1255 //   event 0: indicate to client1 that server listener is in OnDoServerTask
1256 //   event 1: indicate to client1 that client2 listener is in OnDoClient2Task
1257 //   event 2: indicate to server1 that client2 listener is in OnDoClient2Task
1258 //   event 3: indicate to client2 that server listener is in OnDoServerTask
1259
1260 class RestrictedDispatchDeadlockServer : public Worker {
1261  public:
1262   RestrictedDispatchDeadlockServer(int server_num,
1263                                    WaitableEvent* server_ready_event,
1264                                    WaitableEvent** events,
1265                                    RestrictedDispatchDeadlockServer* peer)
1266       : Worker(server_num == 1 ? "channel1" : "channel2", Channel::MODE_SERVER),
1267         server_num_(server_num),
1268         server_ready_event_(server_ready_event),
1269         events_(events),
1270         peer_(peer) { }
1271
1272   void OnDoServerTask() {
1273     events_[3]->Signal();
1274     events_[2]->Wait();
1275     events_[0]->Signal();
1276     SendMessageToClient();
1277   }
1278
1279   void Run() override {
1280     channel()->SetRestrictDispatchChannelGroup(1);
1281     server_ready_event_->Signal();
1282   }
1283
1284   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1285
1286  private:
1287   bool OnMessageReceived(const Message& message) override {
1288     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockServer, message)
1289      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1290      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1291     IPC_END_MESSAGE_MAP()
1292     return true;
1293   }
1294
1295   void OnNoArgs() {
1296     if (server_num_ == 1) {
1297       DCHECK(peer_ != NULL);
1298       peer_->SendMessageToClient();
1299     }
1300   }
1301
1302   void SendMessageToClient() {
1303     Message* msg = new SyncChannelTestMsg_NoArgs;
1304     msg->set_unblock(false);
1305     DCHECK(!msg->should_unblock());
1306     Send(msg);
1307   }
1308
1309   int server_num_;
1310   WaitableEvent* server_ready_event_;
1311   WaitableEvent** events_;
1312   RestrictedDispatchDeadlockServer* peer_;
1313 };
1314
1315 class RestrictedDispatchDeadlockClient2 : public Worker {
1316  public:
1317   RestrictedDispatchDeadlockClient2(RestrictedDispatchDeadlockServer* server,
1318                                     WaitableEvent* server_ready_event,
1319                                     WaitableEvent** events)
1320       : Worker("channel2", Channel::MODE_CLIENT),
1321         server_ready_event_(server_ready_event),
1322         events_(events),
1323         received_msg_(false),
1324         received_noarg_reply_(false),
1325         done_issued_(false) {}
1326
1327   void Run() override {
1328     server_ready_event_->Wait();
1329   }
1330
1331   void OnDoClient2Task() {
1332     events_[3]->Wait();
1333     events_[1]->Signal();
1334     events_[2]->Signal();
1335     DCHECK(received_msg_ == false);
1336
1337     Message* message = new SyncChannelTestMsg_NoArgs;
1338     message->set_unblock(true);
1339     Send(message);
1340     received_noarg_reply_ = true;
1341   }
1342
1343   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1344  private:
1345   bool OnMessageReceived(const Message& message) override {
1346     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockClient2, message)
1347      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1348     IPC_END_MESSAGE_MAP()
1349     return true;
1350   }
1351
1352   void OnNoArgs() {
1353     received_msg_ = true;
1354     PossiblyDone();
1355   }
1356
1357   void PossiblyDone() {
1358     if (received_noarg_reply_ && received_msg_) {
1359       DCHECK(done_issued_ == false);
1360       done_issued_ = true;
1361       Send(new SyncChannelTestMsg_Done);
1362       Done();
1363     }
1364   }
1365
1366   WaitableEvent* server_ready_event_;
1367   WaitableEvent** events_;
1368   bool received_msg_;
1369   bool received_noarg_reply_;
1370   bool done_issued_;
1371 };
1372
1373 class RestrictedDispatchDeadlockClient1 : public Worker {
1374  public:
1375   RestrictedDispatchDeadlockClient1(RestrictedDispatchDeadlockServer* server,
1376                                     RestrictedDispatchDeadlockClient2* peer,
1377                                     WaitableEvent* server_ready_event,
1378                                     WaitableEvent** events)
1379       : Worker("channel1", Channel::MODE_CLIENT),
1380         server_(server),
1381         peer_(peer),
1382         server_ready_event_(server_ready_event),
1383         events_(events),
1384         received_msg_(false),
1385         received_noarg_reply_(false),
1386         done_issued_(false) {}
1387
1388   void Run() override {
1389     server_ready_event_->Wait();
1390     server_->ListenerThread()->task_runner()->PostTask(
1391         FROM_HERE,
1392         base::Bind(&RestrictedDispatchDeadlockServer::OnDoServerTask, server_));
1393     peer_->ListenerThread()->task_runner()->PostTask(
1394         FROM_HERE,
1395         base::Bind(&RestrictedDispatchDeadlockClient2::OnDoClient2Task, peer_));
1396     events_[0]->Wait();
1397     events_[1]->Wait();
1398     DCHECK(received_msg_ == false);
1399
1400     Message* message = new SyncChannelTestMsg_NoArgs;
1401     message->set_unblock(true);
1402     Send(message);
1403     received_noarg_reply_ = true;
1404     PossiblyDone();
1405   }
1406
1407  private:
1408   bool OnMessageReceived(const Message& message) override {
1409     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockClient1, message)
1410      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1411     IPC_END_MESSAGE_MAP()
1412     return true;
1413   }
1414
1415   void OnNoArgs() {
1416     received_msg_ = true;
1417     PossiblyDone();
1418   }
1419
1420   void PossiblyDone() {
1421     if (received_noarg_reply_ && received_msg_) {
1422       DCHECK(done_issued_ == false);
1423       done_issued_ = true;
1424       Send(new SyncChannelTestMsg_Done);
1425       Done();
1426     }
1427   }
1428
1429   RestrictedDispatchDeadlockServer* server_;
1430   RestrictedDispatchDeadlockClient2* peer_;
1431   WaitableEvent* server_ready_event_;
1432   WaitableEvent** events_;
1433   bool received_msg_;
1434   bool received_noarg_reply_;
1435   bool done_issued_;
1436 };
1437
1438 TEST_F(IPCSyncChannelTest, RestrictedDispatchDeadlock) {
1439   std::vector<Worker*> workers;
1440
1441   // A shared worker thread so that server1 and server2 run on one thread.
1442   base::Thread worker_thread("RestrictedDispatchDeadlock");
1443   ASSERT_TRUE(worker_thread.Start());
1444
1445   WaitableEvent server1_ready(false, false);
1446   WaitableEvent server2_ready(false, false);
1447
1448   WaitableEvent event0(false, false);
1449   WaitableEvent event1(false, false);
1450   WaitableEvent event2(false, false);
1451   WaitableEvent event3(false, false);
1452   WaitableEvent* events[4] = {&event0, &event1, &event2, &event3};
1453
1454   RestrictedDispatchDeadlockServer* server1;
1455   RestrictedDispatchDeadlockServer* server2;
1456   RestrictedDispatchDeadlockClient1* client1;
1457   RestrictedDispatchDeadlockClient2* client2;
1458
1459   server2 = new RestrictedDispatchDeadlockServer(2, &server2_ready, events,
1460                                                  NULL);
1461   server2->OverrideThread(&worker_thread);
1462   workers.push_back(server2);
1463
1464   client2 = new RestrictedDispatchDeadlockClient2(server2, &server2_ready,
1465                                                   events);
1466   workers.push_back(client2);
1467
1468   server1 = new RestrictedDispatchDeadlockServer(1, &server1_ready, events,
1469                                                  server2);
1470   server1->OverrideThread(&worker_thread);
1471   workers.push_back(server1);
1472
1473   client1 = new RestrictedDispatchDeadlockClient1(server1, client2,
1474                                                   &server1_ready, events);
1475   workers.push_back(client1);
1476
1477   RunTest(workers);
1478 }
1479
1480 //------------------------------------------------------------------------------
1481
1482 // This test case inspired by crbug.com/120530
1483 // We create 4 workers that pipe to each other W1->W2->W3->W4->W1 then we send a
1484 // message that recurses through 3, 4 or 5 steps to make sure, say, W1 can
1485 // re-enter when called from W4 while it's sending a message to W2.
1486 // The first worker drives the whole test so it must be treated specially.
1487
1488 class RestrictedDispatchPipeWorker : public Worker {
1489  public:
1490   RestrictedDispatchPipeWorker(
1491       const std::string &channel1,
1492       WaitableEvent* event1,
1493       const std::string &channel2,
1494       WaitableEvent* event2,
1495       int group,
1496       int* success)
1497       : Worker(channel1, Channel::MODE_SERVER),
1498         event1_(event1),
1499         event2_(event2),
1500         other_channel_name_(channel2),
1501         group_(group),
1502         success_(success) {
1503   }
1504
1505   void OnPingTTL(int ping, int* ret) {
1506     *ret = 0;
1507     if (!ping)
1508       return;
1509     other_channel_->Send(new SyncChannelTestMsg_PingTTL(ping - 1, ret));
1510     ++*ret;
1511   }
1512
1513   void OnDone() {
1514     if (is_first())
1515       return;
1516     other_channel_->Send(new SyncChannelTestMsg_Done);
1517     other_channel_.reset();
1518     Done();
1519   }
1520
1521   void Run() override {
1522     channel()->SetRestrictDispatchChannelGroup(group_);
1523     if (is_first())
1524       event1_->Signal();
1525     event2_->Wait();
1526     other_channel_ = SyncChannel::Create(
1527         other_channel_name_, IPC::Channel::MODE_CLIENT, this,
1528         ipc_thread().task_runner().get(), true, shutdown_event());
1529     other_channel_->SetRestrictDispatchChannelGroup(group_);
1530     if (!is_first()) {
1531       event1_->Signal();
1532       return;
1533     }
1534     *success_ = 0;
1535     int value = 0;
1536     OnPingTTL(3, &value);
1537     *success_ += (value == 3);
1538     OnPingTTL(4, &value);
1539     *success_ += (value == 4);
1540     OnPingTTL(5, &value);
1541     *success_ += (value == 5);
1542     other_channel_->Send(new SyncChannelTestMsg_Done);
1543     other_channel_.reset();
1544     Done();
1545   }
1546
1547   bool is_first() { return !!success_; }
1548
1549  private:
1550   bool OnMessageReceived(const Message& message) override {
1551     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchPipeWorker, message)
1552      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_PingTTL, OnPingTTL)
1553      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, OnDone)
1554     IPC_END_MESSAGE_MAP()
1555     return true;
1556   }
1557
1558   scoped_ptr<SyncChannel> other_channel_;
1559   WaitableEvent* event1_;
1560   WaitableEvent* event2_;
1561   std::string other_channel_name_;
1562   int group_;
1563   int* success_;
1564 };
1565
1566 TEST_F(IPCSyncChannelTest, RestrictedDispatch4WayDeadlock) {
1567   int success = 0;
1568   std::vector<Worker*> workers;
1569   WaitableEvent event0(true, false);
1570   WaitableEvent event1(true, false);
1571   WaitableEvent event2(true, false);
1572   WaitableEvent event3(true, false);
1573   workers.push_back(new RestrictedDispatchPipeWorker(
1574         "channel0", &event0, "channel1", &event1, 1, &success));
1575   workers.push_back(new RestrictedDispatchPipeWorker(
1576         "channel1", &event1, "channel2", &event2, 2, NULL));
1577   workers.push_back(new RestrictedDispatchPipeWorker(
1578         "channel2", &event2, "channel3", &event3, 3, NULL));
1579   workers.push_back(new RestrictedDispatchPipeWorker(
1580         "channel3", &event3, "channel0", &event0, 4, NULL));
1581   RunTest(workers);
1582   EXPECT_EQ(3, success);
1583 }
1584
1585 //------------------------------------------------------------------------------
1586
1587 // This test case inspired by crbug.com/122443
1588 // We want to make sure a reply message with the unblock flag set correctly
1589 // behaves as a reply, not a regular message.
1590 // We have 3 workers. Server1 will send a message to Server2 (which will block),
1591 // during which it will dispatch a message comming from Client, at which point
1592 // it will send another message to Server2. While sending that second message it
1593 // will receive a reply from Server1 with the unblock flag.
1594
1595 class ReentrantReplyServer1 : public Worker {
1596  public:
1597   ReentrantReplyServer1(WaitableEvent* server_ready)
1598       : Worker("reentrant_reply1", Channel::MODE_SERVER),
1599         server_ready_(server_ready) { }
1600
1601   void Run() override {
1602     server2_channel_ = SyncChannel::Create(
1603         "reentrant_reply2", IPC::Channel::MODE_CLIENT, this,
1604         ipc_thread().task_runner().get(), true, shutdown_event());
1605     server_ready_->Signal();
1606     Message* msg = new SyncChannelTestMsg_Reentrant1();
1607     server2_channel_->Send(msg);
1608     server2_channel_.reset();
1609     Done();
1610   }
1611
1612  private:
1613   bool OnMessageReceived(const Message& message) override {
1614     IPC_BEGIN_MESSAGE_MAP(ReentrantReplyServer1, message)
1615      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Reentrant2, OnReentrant2)
1616      IPC_REPLY_HANDLER(OnReply)
1617     IPC_END_MESSAGE_MAP()
1618     return true;
1619   }
1620
1621   void OnReentrant2() {
1622     Message* msg = new SyncChannelTestMsg_Reentrant3();
1623     server2_channel_->Send(msg);
1624   }
1625
1626   void OnReply(const Message& message) {
1627     // If we get here, the Send() will never receive the reply (thus would
1628     // hang), so abort instead.
1629     LOG(FATAL) << "Reply message was dispatched";
1630   }
1631
1632   WaitableEvent* server_ready_;
1633   scoped_ptr<SyncChannel> server2_channel_;
1634 };
1635
1636 class ReentrantReplyServer2 : public Worker {
1637  public:
1638   ReentrantReplyServer2()
1639       : Worker("reentrant_reply2", Channel::MODE_SERVER),
1640         reply_(NULL) { }
1641
1642  private:
1643   bool OnMessageReceived(const Message& message) override {
1644     IPC_BEGIN_MESSAGE_MAP(ReentrantReplyServer2, message)
1645      IPC_MESSAGE_HANDLER_DELAY_REPLY(
1646          SyncChannelTestMsg_Reentrant1, OnReentrant1)
1647      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Reentrant3, OnReentrant3)
1648     IPC_END_MESSAGE_MAP()
1649     return true;
1650   }
1651
1652   void OnReentrant1(Message* reply) {
1653     DCHECK(!reply_);
1654     reply_ = reply;
1655   }
1656
1657   void OnReentrant3() {
1658     DCHECK(reply_);
1659     Message* reply = reply_;
1660     reply_ = NULL;
1661     reply->set_unblock(true);
1662     Send(reply);
1663     Done();
1664   }
1665
1666   Message* reply_;
1667 };
1668
1669 class ReentrantReplyClient : public Worker {
1670  public:
1671   ReentrantReplyClient(WaitableEvent* server_ready)
1672       : Worker("reentrant_reply1", Channel::MODE_CLIENT),
1673         server_ready_(server_ready) { }
1674
1675   void Run() override {
1676     server_ready_->Wait();
1677     Send(new SyncChannelTestMsg_Reentrant2());
1678     Done();
1679   }
1680
1681  private:
1682   WaitableEvent* server_ready_;
1683 };
1684
1685 TEST_F(IPCSyncChannelTest, ReentrantReply) {
1686   std::vector<Worker*> workers;
1687   WaitableEvent server_ready(false, false);
1688   workers.push_back(new ReentrantReplyServer2());
1689   workers.push_back(new ReentrantReplyServer1(&server_ready));
1690   workers.push_back(new ReentrantReplyClient(&server_ready));
1691   RunTest(workers);
1692 }
1693
1694 //------------------------------------------------------------------------------
1695
1696 // Generate a validated channel ID using Channel::GenerateVerifiedChannelID().
1697
1698 class VerifiedServer : public Worker {
1699  public:
1700   VerifiedServer(base::Thread* listener_thread,
1701                  const std::string& channel_name,
1702                  const std::string& reply_text)
1703       : Worker(channel_name, Channel::MODE_SERVER),
1704         reply_text_(reply_text) {
1705     Worker::OverrideThread(listener_thread);
1706   }
1707
1708   void OnNestedTestMsg(Message* reply_msg) override {
1709     VLOG(1) << __FUNCTION__ << " Sending reply: " << reply_text_;
1710     SyncChannelNestedTestMsg_String::WriteReplyParams(reply_msg, reply_text_);
1711     Send(reply_msg);
1712     ASSERT_EQ(channel()->GetPeerPID(), base::GetCurrentProcId());
1713     Done();
1714   }
1715
1716  private:
1717   std::string reply_text_;
1718 };
1719
1720 class VerifiedClient : public Worker {
1721  public:
1722   VerifiedClient(base::Thread* listener_thread,
1723                  const std::string& channel_name,
1724                  const std::string& expected_text)
1725       : Worker(channel_name, Channel::MODE_CLIENT),
1726         expected_text_(expected_text) {
1727     Worker::OverrideThread(listener_thread);
1728   }
1729
1730   void Run() override {
1731     std::string response;
1732     SyncMessage* msg = new SyncChannelNestedTestMsg_String(&response);
1733     bool result = Send(msg);
1734     DCHECK(result);
1735     DCHECK_EQ(response, expected_text_);
1736     // expected_text_ is only used in the above DCHECK. This line suppresses the
1737     // "unused private field" warning in release builds.
1738     (void)expected_text_;
1739
1740     VLOG(1) << __FUNCTION__ << " Received reply: " << response;
1741     ASSERT_EQ(channel()->GetPeerPID(), base::GetCurrentProcId());
1742     Done();
1743   }
1744
1745  private:
1746   std::string expected_text_;
1747 };
1748
1749 void Verified() {
1750   std::vector<Worker*> workers;
1751
1752   // A shared worker thread for servers
1753   base::Thread server_worker_thread("Verified_ServerListener");
1754   ASSERT_TRUE(server_worker_thread.Start());
1755
1756   base::Thread client_worker_thread("Verified_ClientListener");
1757   ASSERT_TRUE(client_worker_thread.Start());
1758
1759   std::string channel_id = Channel::GenerateVerifiedChannelID("Verified");
1760   Worker* worker;
1761
1762   worker = new VerifiedServer(&server_worker_thread,
1763                               channel_id,
1764                               "Got first message");
1765   workers.push_back(worker);
1766
1767   worker = new VerifiedClient(&client_worker_thread,
1768                               channel_id,
1769                               "Got first message");
1770   workers.push_back(worker);
1771
1772   RunTest(workers);
1773 }
1774
1775 // Windows needs to send an out-of-band secret to verify the client end of the
1776 // channel. Test that we still connect correctly in that case.
1777 TEST_F(IPCSyncChannelTest, Verified) {
1778   Verified();
1779 }
1780
1781 }  // namespace
1782 }  // namespace IPC