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