ipc/ipc_sync_channel_unittest.cc

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