ipc/ipc_sync_channel_unittest.cc

   1 // Copyright (c) 2011 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4 //
   5 // Unit test for SyncChannel.
   6
   7 #include "ipc/ipc_sync_channel.h"
   8
   9 #include <string>
  10 #include <vector>
  11
  12 #include "base/basictypes.h"
  13 #include "base/logging.h"
  14 #include "base/memory/scoped_ptr.h"
  15 #include "base/message_loop.h"
  16 #include "base/stl_util.h"
  17 #include "base/string_util.h"
  18 #include "base/third_party/dynamic_annotations/dynamic_annotations.h"
  19 #include "base/threading/platform_thread.h"
  20 #include "base/threading/thread.h"
  21 #include "base/synchronization/waitable_event.h"
  22 #include "ipc/ipc_message.h"
  23 #include "ipc/ipc_sync_message_filter.h"
  24 #include "ipc/ipc_sync_message_unittest.h"
  25 #include "testing/gtest/include/gtest/gtest.h"
  26
  27 using base::WaitableEvent;
  28
  29 namespace IPC {
  30
  31 namespace {
  32
  33 // Base class for a "process" with listener and IPC threads.
  34 class Worker : public Channel::Listener, public Message::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     // The data race on vfptr is real but is very hard
  46     // to suppress using standard Valgrind mechanism (suppressions).
  47     // We have to use ANNOTATE_BENIGN_RACE to hide the reports and
  48     // make ThreadSanitizer bots green.
  49     ANNOTATE_BENIGN_RACE(this, "Race on vfptr, http://crbug.com/25841");
  50   }
  51
  52   // Will create a named channel and use this name for the threads' name.
  53   Worker(const std::string& channel_name, Channel::Mode mode)
  54       : done_(new WaitableEvent(false, false)),
  55         channel_created_(new WaitableEvent(false, false)),
  56         channel_name_(channel_name),
  57         mode_(mode),
  58         ipc_thread_((channel_name + "_ipc").c_str()),
  59         listener_thread_((channel_name + "_listener").c_str()),
  60         overrided_thread_(NULL),
  61         shutdown_event_(true, false) {
  62     // The data race on vfptr is real but is very hard
  63     // to suppress using standard Valgrind mechanism (suppressions).
  64     // We have to use ANNOTATE_BENIGN_RACE to hide the reports and
  65     // make ThreadSanitizer bots green.
  66     ANNOTATE_BENIGN_RACE(this, "Race on vfptr, http://crbug.com/25841");
  67   }
  68
  69   // The IPC thread needs to outlive SyncChannel, so force the correct order of
  70   // destruction.
  71   virtual ~Worker() {
  72     WaitableEvent listener_done(false, false), ipc_done(false, false);
  73     ListenerThread()->message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
  74         this, &Worker::OnListenerThreadShutdown1, &listener_done,
  75         &ipc_done));
  76     listener_done.Wait();
  77     ipc_done.Wait();
  78     ipc_thread_.Stop();
  79     listener_thread_.Stop();
  80   }
  81   void AddRef() { }
  82   void Release() { }
  83   static bool ImplementsThreadSafeReferenceCounting() { return true; }
  84   bool Send(Message* msg) { return channel_->Send(msg); }
  85   bool SendWithTimeout(Message* msg, int timeout_ms) {
  86     return channel_->SendWithTimeout(msg, timeout_ms);
  87   }
  88   void WaitForChannelCreation() { channel_created_->Wait(); }
  89   void CloseChannel() {
  90     DCHECK(MessageLoop::current() == ListenerThread()->message_loop());
  91     channel_->Close();
  92   }
  93   void Start() {
  94     StartThread(&listener_thread_, MessageLoop::TYPE_DEFAULT);
  95     ListenerThread()->message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
  96         this, &Worker::OnStart));
  97   }
  98   void OverrideThread(base::Thread* overrided_thread) {
  99     DCHECK(overrided_thread_ == NULL);
 100     overrided_thread_ = overrided_thread;
 101   }
 102   bool SendAnswerToLife(bool pump, int timeout, bool succeed) {
 103     int answer = 0;
 104     SyncMessage* msg = new SyncChannelTestMsg_AnswerToLife(&answer);
 105     if (pump)
 106       msg->EnableMessagePumping();
 107     bool result = SendWithTimeout(msg, timeout);
 108     DCHECK_EQ(result, succeed);
 109     DCHECK_EQ(answer, (succeed ? 42 : 0));
 110     return result;
 111   }
 112   bool SendDouble(bool pump, bool succeed) {
 113     int answer = 0;
 114     SyncMessage* msg = new SyncChannelTestMsg_Double(5, &answer);
 115     if (pump)
 116       msg->EnableMessagePumping();
 117     bool result = Send(msg);
 118     DCHECK_EQ(result, succeed);
 119     DCHECK_EQ(answer, (succeed ? 10 : 0));
 120     return result;
 121   }
 122   Channel::Mode mode() { return mode_; }
 123   WaitableEvent* done_event() { return done_.get(); }
 124   WaitableEvent* shutdown_event() { return &shutdown_event_; }
 125   void ResetChannel() { channel_.reset(); }
 126   // Derived classes need to call this when they've completed their part of
 127   // the test.
 128   void Done() { done_->Signal(); }
 129
 130  protected:
 131   SyncChannel* channel() { return channel_.get(); }
 132   // Functions for dervied classes to implement if they wish.
 133   virtual void Run() { }
 134   virtual void OnAnswer(int* answer) { NOTREACHED(); }
 135   virtual void OnAnswerDelay(Message* reply_msg) {
 136     // The message handler map below can only take one entry for
 137     // SyncChannelTestMsg_AnswerToLife, so since some classes want
 138     // the normal version while other want the delayed reply, we
 139     // call the normal version if the derived class didn't override
 140     // this function.
 141     int answer;
 142     OnAnswer(&answer);
 143     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, answer);
 144     Send(reply_msg);
 145   }
 146   virtual void OnDouble(int in, int* out) { NOTREACHED(); }
 147   virtual void OnDoubleDelay(int in, Message* reply_msg) {
 148     int result;
 149     OnDouble(in, &result);
 150     SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, result);
 151     Send(reply_msg);
 152   }
 153
 154   virtual void OnNestedTestMsg(Message* reply_msg) {
 155     NOTREACHED();
 156   }
 157
 158   base::Thread* ListenerThread() {
 159     return overrided_thread_ ? overrided_thread_ : &listener_thread_;
 160   }
 161
 162   const base::Thread& ipc_thread() const { return ipc_thread_; }
 163
 164  private:
 165   // Called on the listener thread to create the sync channel.
 166   void OnStart() {
 167     // Link ipc_thread_, listener_thread_ and channel_ altogether.
 168     StartThread(&ipc_thread_, MessageLoop::TYPE_IO);
 169     channel_.reset(new SyncChannel(
 170         channel_name_, mode_, this, ipc_thread_.message_loop_proxy(), true,
 171         &shutdown_event_));
 172     channel_created_->Signal();
 173     Run();
 174   }
 175
 176   void OnListenerThreadShutdown1(WaitableEvent* listener_event,
 177                                  WaitableEvent* ipc_event) {
 178     // SyncChannel needs to be destructed on the thread that it was created on.
 179     channel_.reset();
 180
 181     MessageLoop::current()->RunAllPending();
 182
 183     ipc_thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 184         this, &Worker::OnIPCThreadShutdown, listener_event, ipc_event));
 185   }
 186
 187   void OnIPCThreadShutdown(WaitableEvent* listener_event,
 188                            WaitableEvent* ipc_event) {
 189     MessageLoop::current()->RunAllPending();
 190     ipc_event->Signal();
 191
 192     listener_thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
 193         this, &Worker::OnListenerThreadShutdown2, listener_event));
 194   }
 195
 196   void OnListenerThreadShutdown2(WaitableEvent* listener_event) {
 197     MessageLoop::current()->RunAllPending();
 198     listener_event->Signal();
 199   }
 200
 201   bool OnMessageReceived(const Message& message) {
 202     IPC_BEGIN_MESSAGE_MAP(Worker, message)
 203      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_Double, OnDoubleDelay)
 204      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_AnswerToLife,
 205                                      OnAnswerDelay)
 206      IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelNestedTestMsg_String,
 207                                      OnNestedTestMsg)
 208     IPC_END_MESSAGE_MAP()
 209     return true;
 210   }
 211
 212   void StartThread(base::Thread* thread, MessageLoop::Type type) {
 213     base::Thread::Options options;
 214     options.message_loop_type = type;
 215     thread->StartWithOptions(options);
 216   }
 217
 218   scoped_ptr<WaitableEvent> done_;
 219   scoped_ptr<WaitableEvent> channel_created_;
 220   std::string channel_name_;
 221   Channel::Mode mode_;
 222   scoped_ptr<SyncChannel> channel_;
 223   base::Thread ipc_thread_;
 224   base::Thread listener_thread_;
 225   base::Thread* overrided_thread_;
 226
 227   base::WaitableEvent shutdown_event_;
 228
 229   DISALLOW_COPY_AND_ASSIGN(Worker);
 230 };
 231
 232
 233 // Starts the test with the given workers.  This function deletes the workers
 234 // when it's done.
 235 void RunTest(std::vector<Worker*> workers) {
 236   // First we create the workers that are channel servers, or else the other
 237   // workers' channel initialization might fail because the pipe isn't created..
 238   for (size_t i = 0; i < workers.size(); ++i) {
 239     if (workers[i]->mode() & Channel::MODE_SERVER_FLAG) {
 240       workers[i]->Start();
 241       workers[i]->WaitForChannelCreation();
 242     }
 243   }
 244
 245   // now create the clients
 246   for (size_t i = 0; i < workers.size(); ++i) {
 247     if (workers[i]->mode() & Channel::MODE_CLIENT_FLAG)
 248       workers[i]->Start();
 249   }
 250
 251   // wait for all the workers to finish
 252   for (size_t i = 0; i < workers.size(); ++i)
 253     workers[i]->done_event()->Wait();
 254
 255   STLDeleteContainerPointers(workers.begin(), workers.end());
 256 }
 257
 258 }  // namespace
 259
 260 class IPCSyncChannelTest : public testing::Test {
 261  private:
 262   MessageLoop message_loop_;
 263 };
 264
 265 //-----------------------------------------------------------------------------
 266
 267 namespace {
 268
 269 class SimpleServer : public Worker {
 270  public:
 271   explicit SimpleServer(bool pump_during_send)
 272       : Worker(Channel::MODE_SERVER, "simpler_server"),
 273         pump_during_send_(pump_during_send) { }
 274   void Run() {
 275     SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 276     Done();
 277   }
 278
 279   bool pump_during_send_;
 280 };
 281
 282 class SimpleClient : public Worker {
 283  public:
 284   SimpleClient() : Worker(Channel::MODE_CLIENT, "simple_client") { }
 285
 286   void OnAnswer(int* answer) {
 287     *answer = 42;
 288     Done();
 289   }
 290 };
 291
 292 void Simple(bool pump_during_send) {
 293   std::vector<Worker*> workers;
 294   workers.push_back(new SimpleServer(pump_during_send));
 295   workers.push_back(new SimpleClient());
 296   RunTest(workers);
 297 }
 298
 299 }  // namespace
 300
 301 // Tests basic synchronous call
 302 TEST_F(IPCSyncChannelTest, Simple) {
 303   Simple(false);
 304   Simple(true);
 305 }
 306
 307 //-----------------------------------------------------------------------------
 308
 309 namespace {
 310
 311 class DelayClient : public Worker {
 312  public:
 313   DelayClient() : Worker(Channel::MODE_CLIENT, "delay_client") { }
 314
 315   void OnAnswerDelay(Message* reply_msg) {
 316     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 317     Send(reply_msg);
 318     Done();
 319   }
 320 };
 321
 322 void DelayReply(bool pump_during_send) {
 323   std::vector<Worker*> workers;
 324   workers.push_back(new SimpleServer(pump_during_send));
 325   workers.push_back(new DelayClient());
 326   RunTest(workers);
 327 }
 328
 329 }  // namespace
 330
 331 // Tests that asynchronous replies work
 332 TEST_F(IPCSyncChannelTest, DelayReply) {
 333   DelayReply(false);
 334   DelayReply(true);
 335 }
 336
 337 //-----------------------------------------------------------------------------
 338
 339 namespace {
 340
 341 class NoHangServer : public Worker {
 342  public:
 343   explicit NoHangServer(WaitableEvent* got_first_reply, bool pump_during_send)
 344       : Worker(Channel::MODE_SERVER, "no_hang_server"),
 345         got_first_reply_(got_first_reply),
 346         pump_during_send_(pump_during_send) { }
 347   void Run() {
 348     SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 349     got_first_reply_->Signal();
 350
 351     SendAnswerToLife(pump_during_send_, base::kNoTimeout, false);
 352     Done();
 353   }
 354
 355   WaitableEvent* got_first_reply_;
 356   bool pump_during_send_;
 357 };
 358
 359 class NoHangClient : public Worker {
 360  public:
 361   explicit NoHangClient(WaitableEvent* got_first_reply)
 362     : Worker(Channel::MODE_CLIENT, "no_hang_client"),
 363       got_first_reply_(got_first_reply) { }
 364
 365   virtual void OnAnswerDelay(Message* reply_msg) {
 366     // Use the DELAY_REPLY macro so that we can force the reply to be sent
 367     // before this function returns (when the channel will be reset).
 368     SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 369     Send(reply_msg);
 370     got_first_reply_->Wait();
 371     CloseChannel();
 372     Done();
 373   }
 374
 375   WaitableEvent* got_first_reply_;
 376 };
 377
 378 void NoHang(bool pump_during_send) {
 379   WaitableEvent got_first_reply(false, false);
 380   std::vector<Worker*> workers;
 381   workers.push_back(new NoHangServer(&got_first_reply, pump_during_send));
 382   workers.push_back(new NoHangClient(&got_first_reply));
 383   RunTest(workers);
 384 }
 385
 386 }  // namespace
 387
 388 // Tests that caller doesn't hang if receiver dies
 389 TEST_F(IPCSyncChannelTest, NoHang) {
 390   NoHang(false);
 391   NoHang(true);
 392 }
 393
 394 //-----------------------------------------------------------------------------
 395
 396 namespace {
 397
 398 class UnblockServer : public Worker {
 399  public:
 400   UnblockServer(bool pump_during_send, bool delete_during_send)
 401     : Worker(Channel::MODE_SERVER, "unblock_server"),
 402       pump_during_send_(pump_during_send),
 403       delete_during_send_(delete_during_send) { }
 404   void Run() {
 405     if (delete_during_send_) {
 406       // Use custom code since race conditions mean the answer may or may not be
 407       // available.
 408       int answer = 0;
 409       SyncMessage* msg = new SyncChannelTestMsg_AnswerToLife(&answer);
 410       if (pump_during_send_)
 411         msg->EnableMessagePumping();
 412       Send(msg);
 413     } else {
 414       SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 415     }
 416     Done();
 417   }
 418
 419   void OnDoubleDelay(int in, Message* reply_msg) {
 420     SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, in * 2);
 421     Send(reply_msg);
 422     if (delete_during_send_)
 423       ResetChannel();
 424   }
 425
 426   bool pump_during_send_;
 427   bool delete_during_send_;
 428 };
 429
 430 class UnblockClient : public Worker {
 431  public:
 432   explicit UnblockClient(bool pump_during_send)
 433     : Worker(Channel::MODE_CLIENT, "unblock_client"),
 434       pump_during_send_(pump_during_send) { }
 435
 436   void OnAnswer(int* answer) {
 437     SendDouble(pump_during_send_, true);
 438     *answer = 42;
 439     Done();
 440   }
 441
 442   bool pump_during_send_;
 443 };
 444
 445 void Unblock(bool server_pump, bool client_pump, bool delete_during_send) {
 446   std::vector<Worker*> workers;
 447   workers.push_back(new UnblockServer(server_pump, delete_during_send));
 448   workers.push_back(new UnblockClient(client_pump));
 449   RunTest(workers);
 450 }
 451
 452 }  // namespace
 453
 454 // Tests that the caller unblocks to answer a sync message from the receiver.
 455 TEST_F(IPCSyncChannelTest, Unblock) {
 456   Unblock(false, false, false);
 457   Unblock(false, true, false);
 458   Unblock(true, false, false);
 459   Unblock(true, true, false);
 460 }
 461
 462 //-----------------------------------------------------------------------------
 463
 464 // Tests that the the SyncChannel object can be deleted during a Send.
 465 TEST_F(IPCSyncChannelTest, ChannelDeleteDuringSend) {
 466   Unblock(false, false, true);
 467   Unblock(false, true, true);
 468   Unblock(true, false, true);
 469   Unblock(true, true, true);
 470 }
 471
 472 //-----------------------------------------------------------------------------
 473
 474 namespace {
 475
 476 class RecursiveServer : public Worker {
 477  public:
 478   explicit RecursiveServer(
 479     bool expected_send_result, bool pump_first, bool pump_second)
 480     : Worker(Channel::MODE_SERVER, "recursive_server"),
 481       expected_send_result_(expected_send_result),
 482       pump_first_(pump_first), pump_second_(pump_second)  { }
 483   void Run() {
 484     SendDouble(pump_first_, expected_send_result_);
 485     Done();
 486   }
 487
 488   void OnDouble(int in, int* out) {
 489     *out = in * 2;
 490     SendAnswerToLife(pump_second_, base::kNoTimeout, expected_send_result_);
 491   }
 492
 493   bool expected_send_result_, pump_first_, pump_second_;
 494 };
 495
 496 class RecursiveClient : public Worker {
 497  public:
 498   explicit RecursiveClient(bool pump_during_send, bool close_channel)
 499     : Worker(Channel::MODE_CLIENT, "recursive_client"),
 500       pump_during_send_(pump_during_send), close_channel_(close_channel) { }
 501
 502   void OnDoubleDelay(int in, Message* reply_msg) {
 503     SendDouble(pump_during_send_, !close_channel_);
 504     if (close_channel_) {
 505       delete reply_msg;
 506     } else {
 507       SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, in * 2);
 508       Send(reply_msg);
 509     }
 510     Done();
 511   }
 512
 513   void OnAnswerDelay(Message* reply_msg) {
 514     if (close_channel_) {
 515       delete reply_msg;
 516       CloseChannel();
 517     } else {
 518       SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 519       Send(reply_msg);
 520     }
 521   }
 522
 523   bool pump_during_send_, close_channel_;
 524 };
 525
 526 void Recursive(
 527     bool server_pump_first, bool server_pump_second, bool client_pump) {
 528   std::vector<Worker*> workers;
 529   workers.push_back(
 530       new RecursiveServer(true, server_pump_first, server_pump_second));
 531   workers.push_back(new RecursiveClient(client_pump, false));
 532   RunTest(workers);
 533 }
 534
 535 }  // namespace
 536
 537 // Tests a server calling Send while another Send is pending.
 538 TEST_F(IPCSyncChannelTest, Recursive) {
 539   Recursive(false, false, false);
 540   Recursive(false, false, true);
 541   Recursive(false, true, false);
 542   Recursive(false, true, true);
 543   Recursive(true, false, false);
 544   Recursive(true, false, true);
 545   Recursive(true, true, false);
 546   Recursive(true, true, true);
 547 }
 548
 549 //-----------------------------------------------------------------------------
 550
 551 namespace {
 552
 553 void RecursiveNoHang(
 554     bool server_pump_first, bool server_pump_second, bool client_pump) {
 555   std::vector<Worker*> workers;
 556   workers.push_back(
 557       new RecursiveServer(false, server_pump_first, server_pump_second));
 558   workers.push_back(new RecursiveClient(client_pump, true));
 559   RunTest(workers);
 560 }
 561
 562 }  // namespace
 563
 564 // Tests that if a caller makes a sync call during an existing sync call and
 565 // the receiver dies, neither of the Send() calls hang.
 566 TEST_F(IPCSyncChannelTest, RecursiveNoHang) {
 567   RecursiveNoHang(false, false, false);
 568   RecursiveNoHang(false, false, true);
 569   RecursiveNoHang(false, true, false);
 570   RecursiveNoHang(false, true, true);
 571   RecursiveNoHang(true, false, false);
 572   RecursiveNoHang(true, false, true);
 573   RecursiveNoHang(true, true, false);
 574   RecursiveNoHang(true, true, true);
 575 }
 576
 577 //-----------------------------------------------------------------------------
 578
 579 namespace {
 580
 581 class MultipleServer1 : public Worker {
 582  public:
 583   explicit MultipleServer1(bool pump_during_send)
 584     : Worker("test_channel1", Channel::MODE_SERVER),
 585       pump_during_send_(pump_during_send) { }
 586
 587   void Run() {
 588     SendDouble(pump_during_send_, true);
 589     Done();
 590   }
 591
 592   bool pump_during_send_;
 593 };
 594
 595 class MultipleClient1 : public Worker {
 596  public:
 597   MultipleClient1(WaitableEvent* client1_msg_received,
 598                   WaitableEvent* client1_can_reply) :
 599       Worker("test_channel1", Channel::MODE_CLIENT),
 600       client1_msg_received_(client1_msg_received),
 601       client1_can_reply_(client1_can_reply) { }
 602
 603   void OnDouble(int in, int* out) {
 604     client1_msg_received_->Signal();
 605     *out = in * 2;
 606     client1_can_reply_->Wait();
 607     Done();
 608   }
 609
 610  private:
 611   WaitableEvent *client1_msg_received_, *client1_can_reply_;
 612 };
 613
 614 class MultipleServer2 : public Worker {
 615  public:
 616   MultipleServer2() : Worker("test_channel2", Channel::MODE_SERVER) { }
 617
 618   void OnAnswer(int* result) {
 619     *result = 42;
 620     Done();
 621   }
 622 };
 623
 624 class MultipleClient2 : public Worker {
 625  public:
 626   MultipleClient2(
 627     WaitableEvent* client1_msg_received, WaitableEvent* client1_can_reply,
 628     bool pump_during_send)
 629     : Worker("test_channel2", Channel::MODE_CLIENT),
 630       client1_msg_received_(client1_msg_received),
 631       client1_can_reply_(client1_can_reply),
 632       pump_during_send_(pump_during_send) { }
 633
 634   void Run() {
 635     client1_msg_received_->Wait();
 636     SendAnswerToLife(pump_during_send_, base::kNoTimeout, true);
 637     client1_can_reply_->Signal();
 638     Done();
 639   }
 640
 641  private:
 642   WaitableEvent *client1_msg_received_, *client1_can_reply_;
 643   bool pump_during_send_;
 644 };
 645
 646 void Multiple(bool server_pump, bool client_pump) {
 647   std::vector<Worker*> workers;
 648
 649   // A shared worker thread so that server1 and server2 run on one thread.
 650   base::Thread worker_thread("Multiple");
 651   ASSERT_TRUE(worker_thread.Start());
 652
 653   // Server1 sends a sync msg to client1, which blocks the reply until
 654   // server2 (which runs on the same worker thread as server1) responds
 655   // to a sync msg from client2.
 656   WaitableEvent client1_msg_received(false, false);
 657   WaitableEvent client1_can_reply(false, false);
 658
 659   Worker* worker;
 660
 661   worker = new MultipleServer2();
 662   worker->OverrideThread(&worker_thread);
 663   workers.push_back(worker);
 664
 665   worker = new MultipleClient2(
 666       &client1_msg_received, &client1_can_reply, client_pump);
 667   workers.push_back(worker);
 668
 669   worker = new MultipleServer1(server_pump);
 670   worker->OverrideThread(&worker_thread);
 671   workers.push_back(worker);
 672
 673   worker = new MultipleClient1(
 674       &client1_msg_received, &client1_can_reply);
 675   workers.push_back(worker);
 676
 677   RunTest(workers);
 678 }
 679
 680 }  // namespace
 681
 682 // Tests that multiple SyncObjects on the same listener thread can unblock each
 683 // other.
 684 TEST_F(IPCSyncChannelTest, Multiple) {
 685   Multiple(false, false);
 686   Multiple(false, true);
 687   Multiple(true, false);
 688   Multiple(true, true);
 689 }
 690
 691 //-----------------------------------------------------------------------------
 692
 693 namespace {
 694
 695 // This class provides server side functionality to test the case where
 696 // multiple sync channels are in use on the same thread on the client and
 697 // nested calls are issued.
 698 class QueuedReplyServer : public Worker {
 699  public:
 700   QueuedReplyServer(base::Thread* listener_thread,
 701                     const std::string& channel_name,
 702                     const std::string& reply_text)
 703       : Worker(channel_name, Channel::MODE_SERVER),
 704         reply_text_(reply_text) {
 705     Worker::OverrideThread(listener_thread);
 706   }
 707
 708   virtual void OnNestedTestMsg(Message* reply_msg) {
 709     VLOG(1) << __FUNCTION__ << " Sending reply: " << reply_text_;
 710     SyncChannelNestedTestMsg_String::WriteReplyParams(reply_msg, reply_text_);
 711     Send(reply_msg);
 712     Done();
 713   }
 714
 715  private:
 716   std::string reply_text_;
 717 };
 718
 719 // The QueuedReplyClient class provides functionality to test the case where
 720 // multiple sync channels are in use on the same thread and they make nested
 721 // sync calls, i.e. while the first channel waits for a response it makes a
 722 // sync call on another channel.
 723 // The callstack should unwind correctly, i.e. the outermost call should
 724 // complete first, and so on.
 725 class QueuedReplyClient : public Worker {
 726  public:
 727   QueuedReplyClient(base::Thread* listener_thread,
 728                     const std::string& channel_name,
 729                     const std::string& expected_text,
 730                     bool pump_during_send)
 731       : Worker(channel_name, Channel::MODE_CLIENT),
 732         pump_during_send_(pump_during_send),
 733         expected_text_(expected_text) {
 734     Worker::OverrideThread(listener_thread);
 735   }
 736
 737   virtual void Run() {
 738     std::string response;
 739     SyncMessage* msg = new SyncChannelNestedTestMsg_String(&response);
 740     if (pump_during_send_)
 741       msg->EnableMessagePumping();
 742     bool result = Send(msg);
 743     DCHECK(result);
 744     DCHECK_EQ(response, expected_text_);
 745
 746     VLOG(1) << __FUNCTION__ << " Received reply: " << response;
 747     Done();
 748   }
 749
 750  private:
 751   bool pump_during_send_;
 752   std::string expected_text_;
 753 };
 754
 755 void QueuedReply(bool client_pump) {
 756   std::vector<Worker*> workers;
 757
 758   // A shared worker thread for servers
 759   base::Thread server_worker_thread("QueuedReply_ServerListener");
 760   ASSERT_TRUE(server_worker_thread.Start());
 761
 762   base::Thread client_worker_thread("QueuedReply_ClientListener");
 763   ASSERT_TRUE(client_worker_thread.Start());
 764
 765   Worker* worker;
 766
 767   worker = new QueuedReplyServer(&server_worker_thread,
 768                                  "QueuedReply_Server1",
 769                                  "Got first message");
 770   workers.push_back(worker);
 771
 772   worker = new QueuedReplyServer(&server_worker_thread,
 773                                  "QueuedReply_Server2",
 774                                  "Got second message");
 775   workers.push_back(worker);
 776
 777   worker = new QueuedReplyClient(&client_worker_thread,
 778                                  "QueuedReply_Server1",
 779                                  "Got first message",
 780                                  client_pump);
 781   workers.push_back(worker);
 782
 783   worker = new QueuedReplyClient(&client_worker_thread,
 784                                  "QueuedReply_Server2",
 785                                  "Got second message",
 786                                  client_pump);
 787   workers.push_back(worker);
 788
 789   RunTest(workers);
 790 }
 791
 792 }  // namespace
 793
 794 // While a blocking send is in progress, the listener thread might answer other
 795 // synchronous messages.  This tests that if during the response to another
 796 // message the reply to the original messages comes, it is queued up correctly
 797 // and the original Send is unblocked later.
 798 // We also test that the send call stacks unwind correctly when the channel
 799 // pumps messages while waiting for a response.
 800 TEST_F(IPCSyncChannelTest, QueuedReply) {
 801   QueuedReply(false);
 802   QueuedReply(true);
 803 }
 804
 805 //-----------------------------------------------------------------------------
 806
 807 namespace {
 808
 809 class ChattyClient : public Worker {
 810  public:
 811   ChattyClient() :
 812       Worker(Channel::MODE_CLIENT, "chatty_client") { }
 813
 814   void OnAnswer(int* answer) {
 815     // The PostMessage limit is 10k.  Send 20% more than that.
 816     const int kMessageLimit = 10000;
 817     const int kMessagesToSend = kMessageLimit * 120 / 100;
 818     for (int i = 0; i < kMessagesToSend; ++i) {
 819       if (!SendDouble(false, true))
 820         break;
 821     }
 822     *answer = 42;
 823     Done();
 824   }
 825 };
 826
 827 void ChattyServer(bool pump_during_send) {
 828   std::vector<Worker*> workers;
 829   workers.push_back(new UnblockServer(pump_during_send, false));
 830   workers.push_back(new ChattyClient());
 831   RunTest(workers);
 832 }
 833
 834 }  // namespace
 835
 836 // Tests http://b/1093251 - that sending lots of sync messages while
 837 // the receiver is waiting for a sync reply does not overflow the PostMessage
 838 // queue.
 839 TEST_F(IPCSyncChannelTest, ChattyServer) {
 840   ChattyServer(false);
 841   ChattyServer(true);
 842 }
 843
 844 //------------------------------------------------------------------------------
 845
 846 namespace {
 847
 848 class TimeoutServer : public Worker {
 849  public:
 850   TimeoutServer(int timeout_ms,
 851                 std::vector<bool> timeout_seq,
 852                 bool pump_during_send)
 853       : Worker(Channel::MODE_SERVER, "timeout_server"),
 854         timeout_ms_(timeout_ms),
 855         timeout_seq_(timeout_seq),
 856         pump_during_send_(pump_during_send) {
 857   }
 858
 859   void Run() {
 860     for (std::vector<bool>::const_iterator iter = timeout_seq_.begin();
 861          iter != timeout_seq_.end(); ++iter) {
 862       SendAnswerToLife(pump_during_send_, timeout_ms_, !*iter);
 863     }
 864     Done();
 865   }
 866
 867  private:
 868   int timeout_ms_;
 869   std::vector<bool> timeout_seq_;
 870   bool pump_during_send_;
 871 };
 872
 873 class UnresponsiveClient : public Worker {
 874  public:
 875   explicit UnresponsiveClient(std::vector<bool> timeout_seq)
 876       : Worker(Channel::MODE_CLIENT, "unresponsive_client"),
 877         timeout_seq_(timeout_seq) {
 878   }
 879
 880   void OnAnswerDelay(Message* reply_msg) {
 881     DCHECK(!timeout_seq_.empty());
 882     if (!timeout_seq_[0]) {
 883       SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
 884       Send(reply_msg);
 885     } else {
 886       // Don't reply.
 887       delete reply_msg;
 888     }
 889     timeout_seq_.erase(timeout_seq_.begin());
 890     if (timeout_seq_.empty())
 891       Done();
 892   }
 893
 894  private:
 895   // Whether we should time-out or respond to the various messages we receive.
 896   std::vector<bool> timeout_seq_;
 897 };
 898
 899 void SendWithTimeoutOK(bool pump_during_send) {
 900   std::vector<Worker*> workers;
 901   std::vector<bool> timeout_seq;
 902   timeout_seq.push_back(false);
 903   timeout_seq.push_back(false);
 904   timeout_seq.push_back(false);
 905   workers.push_back(new TimeoutServer(5000, timeout_seq, pump_during_send));
 906   workers.push_back(new SimpleClient());
 907   RunTest(workers);
 908 }
 909
 910 void SendWithTimeoutTimeout(bool pump_during_send) {
 911   std::vector<Worker*> workers;
 912   std::vector<bool> timeout_seq;
 913   timeout_seq.push_back(true);
 914   timeout_seq.push_back(false);
 915   timeout_seq.push_back(false);
 916   workers.push_back(new TimeoutServer(100, timeout_seq, pump_during_send));
 917   workers.push_back(new UnresponsiveClient(timeout_seq));
 918   RunTest(workers);
 919 }
 920
 921 void SendWithTimeoutMixedOKAndTimeout(bool pump_during_send) {
 922   std::vector<Worker*> workers;
 923   std::vector<bool> timeout_seq;
 924   timeout_seq.push_back(true);
 925   timeout_seq.push_back(false);
 926   timeout_seq.push_back(false);
 927   timeout_seq.push_back(true);
 928   timeout_seq.push_back(false);
 929   workers.push_back(new TimeoutServer(100, timeout_seq, pump_during_send));
 930   workers.push_back(new UnresponsiveClient(timeout_seq));
 931   RunTest(workers);
 932 }
 933
 934 }  // namespace
 935
 936 // Tests that SendWithTimeout does not time-out if the response comes back fast
 937 // enough.
 938 TEST_F(IPCSyncChannelTest, SendWithTimeoutOK) {
 939   SendWithTimeoutOK(false);
 940   SendWithTimeoutOK(true);
 941 }
 942
 943 // Tests that SendWithTimeout does time-out.
 944 TEST_F(IPCSyncChannelTest, SendWithTimeoutTimeout) {
 945   SendWithTimeoutTimeout(false);
 946   SendWithTimeoutTimeout(true);
 947 }
 948
 949 // Sends some message that time-out and some that succeed.
 950 // Crashes flakily, http://crbug.com/70075.
 951 TEST_F(IPCSyncChannelTest, DISABLED_SendWithTimeoutMixedOKAndTimeout) {
 952   SendWithTimeoutMixedOKAndTimeout(false);
 953   SendWithTimeoutMixedOKAndTimeout(true);
 954 }
 955
 956 //------------------------------------------------------------------------------
 957
 958 namespace {
 959
 960 class NestedTask : public Task {
 961  public:
 962   explicit NestedTask(Worker* server) : server_(server) { }
 963   void Run() {
 964     // Sleep a bit so that we wake up after the reply has been received.
 965     base::PlatformThread::Sleep(250);
 966     server_->SendAnswerToLife(true, base::kNoTimeout, true);
 967   }
 968
 969   Worker* server_;
 970 };
 971
 972 static bool timeout_occured = false;
 973
 974 class TimeoutTask : public Task {
 975  public:
 976   void Run() {
 977     timeout_occured = true;
 978   }
 979 };
 980
 981 class DoneEventRaceServer : public Worker {
 982  public:
 983   DoneEventRaceServer()
 984       : Worker(Channel::MODE_SERVER, "done_event_race_server") { }
 985
 986   void Run() {
 987     MessageLoop::current()->PostTask(FROM_HERE, new NestedTask(this));
 988     MessageLoop::current()->PostDelayedTask(FROM_HERE, new TimeoutTask(), 9000);
 989     // Even though we have a timeout on the Send, it will succeed since for this
 990     // bug, the reply message comes back and is deserialized, however the done
 991     // event wasn't set.  So we indirectly use the timeout task to notice if a
 992     // timeout occurred.
 993     SendAnswerToLife(true, 10000, true);
 994     DCHECK(!timeout_occured);
 995     Done();
 996   }
 997 };
 998
 999 }  // namespace
1000
1001 // Tests http://b/1474092 - that if after the done_event is set but before
1002 // OnObjectSignaled is called another message is sent out, then after its
1003 // reply comes back OnObjectSignaled will be called for the first message.
1004 TEST_F(IPCSyncChannelTest, DoneEventRace) {
1005   std::vector<Worker*> workers;
1006   workers.push_back(new DoneEventRaceServer());
1007   workers.push_back(new SimpleClient());
1008   RunTest(workers);
1009 }
1010
1011 //-----------------------------------------------------------------------------
1012
1013 namespace {
1014
1015 class TestSyncMessageFilter : public SyncMessageFilter {
1016  public:
1017   TestSyncMessageFilter(base::WaitableEvent* shutdown_event, Worker* worker)
1018       : SyncMessageFilter(shutdown_event),
1019         worker_(worker),
1020         thread_("helper_thread") {
1021     base::Thread::Options options;
1022     options.message_loop_type = MessageLoop::TYPE_DEFAULT;
1023     thread_.StartWithOptions(options);
1024   }
1025
1026   virtual void OnFilterAdded(Channel* channel) {
1027     SyncMessageFilter::OnFilterAdded(channel);
1028     thread_.message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
1029         this, &TestSyncMessageFilter::SendMessageOnHelperThread));
1030   }
1031
1032   void SendMessageOnHelperThread() {
1033     int answer = 0;
1034     bool result = Send(new SyncChannelTestMsg_AnswerToLife(&answer));
1035     DCHECK(result);
1036     DCHECK_EQ(answer, 42);
1037
1038     worker_->Done();
1039   }
1040
1041   Worker* worker_;
1042   base::Thread thread_;
1043 };
1044
1045 class SyncMessageFilterServer : public Worker {
1046  public:
1047   SyncMessageFilterServer()
1048       : Worker(Channel::MODE_SERVER, "sync_message_filter_server") {
1049     filter_ = new TestSyncMessageFilter(shutdown_event(), this);
1050   }
1051
1052   void Run() {
1053     channel()->AddFilter(filter_.get());
1054   }
1055
1056   scoped_refptr<TestSyncMessageFilter> filter_;
1057 };
1058
1059 // This class provides functionality to test the case that a Send on the sync
1060 // channel does not crash after the channel has been closed.
1061 class ServerSendAfterClose : public Worker {
1062  public:
1063   ServerSendAfterClose()
1064      : Worker(Channel::MODE_SERVER, "simpler_server"),
1065        send_result_(true) {
1066   }
1067
1068   bool SendDummy() {
1069     ListenerThread()->message_loop()->PostTask(FROM_HERE, NewRunnableMethod(
1070         this, &ServerSendAfterClose::Send, new SyncChannelTestMsg_NoArgs));
1071     return true;
1072   }
1073
1074   bool send_result() const {
1075     return send_result_;
1076   }
1077
1078  private:
1079   virtual void Run() {
1080     CloseChannel();
1081     Done();
1082   }
1083
1084   bool Send(Message* msg) {
1085     send_result_ = Worker::Send(msg);
1086     Done();
1087     return send_result_;
1088   }
1089
1090   bool send_result_;
1091 };
1092
1093 }  // namespace
1094
1095 // Tests basic synchronous call
1096 TEST_F(IPCSyncChannelTest, SyncMessageFilter) {
1097   std::vector<Worker*> workers;
1098   workers.push_back(new SyncMessageFilterServer());
1099   workers.push_back(new SimpleClient());
1100   RunTest(workers);
1101 }
1102
1103 // Test the case when the channel is closed and a Send is attempted after that.
1104 TEST_F(IPCSyncChannelTest, SendAfterClose) {
1105   ServerSendAfterClose server;
1106   server.Start();
1107
1108   server.done_event()->Wait();
1109   server.done_event()->Reset();
1110
1111   server.SendDummy();
1112   server.done_event()->Wait();
1113
1114   EXPECT_FALSE(server.send_result());
1115 }
1116
1117 //-----------------------------------------------------------------------------
1118
1119 namespace {
1120
1121 class RestrictedDispatchServer : public Worker {
1122  public:
1123   RestrictedDispatchServer(WaitableEvent* sent_ping_event)
1124       : Worker("restricted_channel", Channel::MODE_SERVER),
1125         sent_ping_event_(sent_ping_event) { }
1126
1127   void OnDoPing(int ping) {
1128     // Send an asynchronous message that unblocks the caller.
1129     Message* msg = new SyncChannelTestMsg_Ping(ping);
1130     msg->set_unblock(true);
1131     Send(msg);
1132     // Signal the event after the message has been sent on the channel, on the
1133     // IPC thread.
1134     ipc_thread().message_loop()->PostTask(FROM_HERE,
1135         NewRunnableMethod(this, &RestrictedDispatchServer::OnPingSent));
1136   }
1137
1138   base::Thread* ListenerThread() { return Worker::ListenerThread(); }
1139
1140  private:
1141   bool OnMessageReceived(const Message& message) {
1142     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchServer, message)
1143      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1144      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1145     IPC_END_MESSAGE_MAP()
1146     return true;
1147   }
1148
1149   void OnPingSent() {
1150     sent_ping_event_->Signal();
1151   }
1152
1153   void OnNoArgs() { }
1154   WaitableEvent* sent_ping_event_;
1155 };
1156
1157 class NonRestrictedDispatchServer : public Worker {
1158  public:
1159   NonRestrictedDispatchServer()
1160       : Worker("non_restricted_channel", Channel::MODE_SERVER) {}
1161
1162  private:
1163   bool OnMessageReceived(const Message& message) {
1164     IPC_BEGIN_MESSAGE_MAP(NonRestrictedDispatchServer, message)
1165      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
1166      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
1167     IPC_END_MESSAGE_MAP()
1168     return true;
1169   }
1170
1171   void OnNoArgs() { }
1172 };
1173
1174 class RestrictedDispatchClient : public Worker {
1175  public:
1176   RestrictedDispatchClient(WaitableEvent* sent_ping_event,
1177                            RestrictedDispatchServer* server,
1178                            int* success)
1179       : Worker("restricted_channel", Channel::MODE_CLIENT),
1180         ping_(0),
1181         server_(server),
1182         success_(success),
1183         sent_ping_event_(sent_ping_event) {}
1184
1185   void Run() {
1186     // Incoming messages from our channel should only be dispatched when we
1187     // send a message on that same channel.
1188     channel()->SetRestrictDispatchToSameChannel(true);
1189
1190     server_->ListenerThread()->message_loop()->PostTask(FROM_HERE,
1191         NewRunnableMethod(server_, &RestrictedDispatchServer::OnDoPing, 1));
1192     sent_ping_event_->Wait();
1193     Send(new SyncChannelTestMsg_NoArgs);
1194     if (ping_ == 1)
1195       ++*success_;
1196     else
1197       LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
1198
1199     scoped_ptr<SyncChannel> non_restricted_channel(new SyncChannel(
1200         "non_restricted_channel", Channel::MODE_CLIENT, this,
1201         ipc_thread().message_loop_proxy(), true, shutdown_event()));
1202
1203     server_->ListenerThread()->message_loop()->PostTask(FROM_HERE,
1204         NewRunnableMethod(server_, &RestrictedDispatchServer::OnDoPing, 2));
1205     sent_ping_event_->Wait();
1206     // Check that the incoming message is *not* dispatched when sending on the
1207     // non restricted channel.
1208     // TODO(piman): there is a possibility of a false positive race condition
1209     // here, if the message that was posted on the server-side end of the pipe
1210     // is not visible yet on the client side, but I don't know how to solve this
1211     // without hooking into the internals of SyncChannel. I haven't seen it in
1212     // practice (i.e. not setting SetRestrictDispatchToSameChannel does cause
1213     // the following to fail).
1214     non_restricted_channel->Send(new SyncChannelTestMsg_NoArgs);
1215     if (ping_ == 1)
1216       ++*success_;
1217     else
1218       LOG(ERROR) << "Send dispatched message from restricted channel";
1219
1220     Send(new SyncChannelTestMsg_NoArgs);
1221     if (ping_ == 2)
1222       ++*success_;
1223     else
1224       LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
1225
1226     non_restricted_channel->Send(new SyncChannelTestMsg_Done);
1227     non_restricted_channel.reset();
1228     Send(new SyncChannelTestMsg_Done);
1229     Done();
1230   }
1231
1232  private:
1233   bool OnMessageReceived(const Message& message) {
1234     IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchClient, message)
1235      IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Ping, OnPing)
1236     IPC_END_MESSAGE_MAP()
1237     return true;
1238   }
1239
1240   void OnPing(int ping) {
1241     ping_ = ping;
1242   }
1243
1244   int ping_;
1245   RestrictedDispatchServer* server_;
1246   int* success_;
1247   WaitableEvent* sent_ping_event_;
1248 };
1249
1250 }  // namespace
1251
1252 TEST_F(IPCSyncChannelTest, RestrictedDispatch) {
1253   WaitableEvent sent_ping_event(false, false);
1254
1255   RestrictedDispatchServer* server =
1256       new RestrictedDispatchServer(&sent_ping_event);
1257   int success = 0;
1258   std::vector<Worker*> workers;
1259   workers.push_back(new NonRestrictedDispatchServer);
1260   workers.push_back(server);
1261   workers.push_back(
1262       new RestrictedDispatchClient(&sent_ping_event, server, &success));
1263   RunTest(workers);
1264   EXPECT_EQ(3, success);
1265 }
1266
1267 }  // namespace IPC