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