ipc_sync_channel_unittest.cc 55 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700
  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. #include "ipc/ipc_sync_channel.h"
  5. #include <stddef.h>
  6. #include <memory>
  7. #include <string>
  8. #include <utility>
  9. #include <vector>
  10. #include "base/bind.h"
  11. #include "base/location.h"
  12. #include "base/logging.h"
  13. #include "base/memory/raw_ptr.h"
  14. #include "base/message_loop/message_pump_type.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/task/single_thread_task_runner.h"
  20. #include "base/test/task_environment.h"
  21. #include "base/threading/platform_thread.h"
  22. #include "base/threading/thread.h"
  23. #include "base/threading/thread_task_runner_handle.h"
  24. #include "build/build_config.h"
  25. #include "ipc/ipc_listener.h"
  26. #include "ipc/ipc_message.h"
  27. #include "ipc/ipc_sender.h"
  28. #include "ipc/ipc_sync_message_filter.h"
  29. #include "ipc/ipc_sync_message_unittest.h"
  30. #include "mojo/public/cpp/system/message_pipe.h"
  31. #include "testing/gtest/include/gtest/gtest.h"
  32. using base::WaitableEvent;
  33. namespace IPC {
  34. namespace {
  35. // Base class for a "process" with listener and IPC threads.
  36. class Worker : public Listener, public Sender {
  37. public:
  38. // Will create a channel without a name.
  39. Worker(Channel::Mode mode,
  40. const std::string& thread_name,
  41. mojo::ScopedMessagePipeHandle channel_handle)
  42. : done_(
  43. new WaitableEvent(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  44. base::WaitableEvent::InitialState::NOT_SIGNALED)),
  45. channel_created_(
  46. new WaitableEvent(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  47. base::WaitableEvent::InitialState::NOT_SIGNALED)),
  48. channel_handle_(std::move(channel_handle)),
  49. mode_(mode),
  50. ipc_thread_((thread_name + "_ipc").c_str()),
  51. listener_thread_((thread_name + "_listener").c_str()),
  52. overrided_thread_(nullptr),
  53. shutdown_event_(base::WaitableEvent::ResetPolicy::MANUAL,
  54. base::WaitableEvent::InitialState::NOT_SIGNALED),
  55. is_shutdown_(false) {}
  56. // Will create a named channel and use this name for the threads' name.
  57. Worker(mojo::ScopedMessagePipeHandle channel_handle, Channel::Mode mode)
  58. : done_(
  59. new WaitableEvent(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  60. base::WaitableEvent::InitialState::NOT_SIGNALED)),
  61. channel_created_(
  62. new WaitableEvent(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  63. base::WaitableEvent::InitialState::NOT_SIGNALED)),
  64. channel_handle_(std::move(channel_handle)),
  65. mode_(mode),
  66. ipc_thread_("ipc thread"),
  67. listener_thread_("listener thread"),
  68. overrided_thread_(nullptr),
  69. shutdown_event_(base::WaitableEvent::ResetPolicy::MANUAL,
  70. base::WaitableEvent::InitialState::NOT_SIGNALED),
  71. is_shutdown_(false) {}
  72. Worker(const Worker&) = delete;
  73. Worker& operator=(const Worker&) = delete;
  74. ~Worker() override {
  75. // Shutdown() must be called before destruction.
  76. CHECK(is_shutdown_);
  77. }
  78. bool Send(Message* msg) override { return channel_->Send(msg); }
  79. void WaitForChannelCreation() { channel_created_->Wait(); }
  80. void CloseChannel() {
  81. DCHECK(ListenerThread()->task_runner()->BelongsToCurrentThread());
  82. channel_->Close();
  83. }
  84. void Start() {
  85. StartThread(&listener_thread_, base::MessagePumpType::DEFAULT);
  86. ListenerThread()->task_runner()->PostTask(
  87. FROM_HERE, base::BindOnce(&Worker::OnStart, base::Unretained(this)));
  88. }
  89. void Shutdown() {
  90. // The IPC thread needs to outlive SyncChannel. We can't do this in
  91. // ~Worker(), since that'll reset the vtable pointer (to Worker's), which
  92. // may result in a race conditions. See http://crbug.com/25841.
  93. WaitableEvent listener_done(
  94. base::WaitableEvent::ResetPolicy::AUTOMATIC,
  95. base::WaitableEvent::InitialState::NOT_SIGNALED),
  96. ipc_done(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  97. base::WaitableEvent::InitialState::NOT_SIGNALED);
  98. ListenerThread()->task_runner()->PostTask(
  99. FROM_HERE,
  100. base::BindOnce(&Worker::OnListenerThreadShutdown1,
  101. base::Unretained(this), &listener_done, &ipc_done));
  102. listener_done.Wait();
  103. ipc_done.Wait();
  104. ipc_thread_.Stop();
  105. listener_thread_.Stop();
  106. is_shutdown_ = true;
  107. }
  108. void OverrideThread(base::Thread* overrided_thread) {
  109. DCHECK(!overrided_thread_);
  110. overrided_thread_ = overrided_thread;
  111. }
  112. bool SendAnswerToLife(bool succeed) {
  113. int answer = 0;
  114. SyncMessage* msg = new SyncChannelTestMsg_AnswerToLife(&answer);
  115. bool result = Send(msg);
  116. DCHECK_EQ(result, succeed);
  117. DCHECK_EQ(answer, (succeed ? 42 : 0));
  118. return result;
  119. }
  120. bool SendDouble(bool succeed) {
  121. int answer = 0;
  122. SyncMessage* msg = new SyncChannelTestMsg_Double(5, &answer);
  123. bool result = Send(msg);
  124. DCHECK_EQ(result, succeed);
  125. DCHECK_EQ(answer, (succeed ? 10 : 0));
  126. return result;
  127. }
  128. mojo::MessagePipeHandle TakeChannelHandle() {
  129. DCHECK(channel_handle_.is_valid());
  130. return channel_handle_.release();
  131. }
  132. Channel::Mode mode() { return mode_; }
  133. WaitableEvent* done_event() { return done_.get(); }
  134. WaitableEvent* shutdown_event() { return &shutdown_event_; }
  135. void ResetChannel() { channel_.reset(); }
  136. // Derived classes need to call this when they've completed their part of
  137. // the test.
  138. void Done() { done_->Signal(); }
  139. protected:
  140. SyncChannel* channel() { return channel_.get(); }
  141. // Functions for derived classes to implement if they wish.
  142. virtual void Run() { }
  143. virtual void OnAnswer(int* answer) { NOTREACHED(); }
  144. virtual void OnAnswerDelay(Message* reply_msg) {
  145. // The message handler map below can only take one entry for
  146. // SyncChannelTestMsg_AnswerToLife, so since some classes want
  147. // the normal version while other want the delayed reply, we
  148. // call the normal version if the derived class didn't override
  149. // this function.
  150. int answer;
  151. OnAnswer(&answer);
  152. SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, answer);
  153. Send(reply_msg);
  154. }
  155. virtual void OnDouble(int in, int* out) { NOTREACHED(); }
  156. virtual void OnDoubleDelay(int in, Message* reply_msg) {
  157. int result;
  158. OnDouble(in, &result);
  159. SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, result);
  160. Send(reply_msg);
  161. }
  162. virtual void OnNestedTestMsg(Message* reply_msg) {
  163. NOTREACHED();
  164. }
  165. virtual SyncChannel* CreateChannel() {
  166. std::unique_ptr<SyncChannel> channel = SyncChannel::Create(
  167. TakeChannelHandle(), mode_, this, ipc_thread_.task_runner(),
  168. base::ThreadTaskRunnerHandle::Get(), true, &shutdown_event_);
  169. return channel.release();
  170. }
  171. base::Thread* ListenerThread() {
  172. return overrided_thread_ ? overrided_thread_.get() : &listener_thread_;
  173. }
  174. const base::Thread& ipc_thread() const { return ipc_thread_; }
  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_, base::MessagePumpType::IO);
  180. channel_.reset(CreateChannel());
  181. channel_created_->Signal();
  182. Run();
  183. }
  184. void OnListenerThreadShutdown1(WaitableEvent* listener_event,
  185. WaitableEvent* ipc_event) {
  186. // SyncChannel needs to be destructed on the thread that it was created on.
  187. channel_.reset();
  188. base::RunLoop().RunUntilIdle();
  189. ipc_thread_.task_runner()->PostTask(
  190. FROM_HERE,
  191. base::BindOnce(&Worker::OnIPCThreadShutdown, base::Unretained(this),
  192. listener_event, ipc_event));
  193. }
  194. void OnIPCThreadShutdown(WaitableEvent* listener_event,
  195. WaitableEvent* ipc_event) {
  196. base::RunLoop().RunUntilIdle();
  197. ipc_event->Signal();
  198. listener_thread_.task_runner()->PostTask(
  199. FROM_HERE, base::BindOnce(&Worker::OnListenerThreadShutdown2,
  200. base::Unretained(this), listener_event));
  201. }
  202. void OnListenerThreadShutdown2(WaitableEvent* listener_event) {
  203. base::RunLoop().RunUntilIdle();
  204. listener_event->Signal();
  205. }
  206. bool OnMessageReceived(const Message& message) override {
  207. IPC_BEGIN_MESSAGE_MAP(Worker, message)
  208. IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_Double, OnDoubleDelay)
  209. IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_AnswerToLife,
  210. OnAnswerDelay)
  211. IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelNestedTestMsg_String,
  212. OnNestedTestMsg)
  213. IPC_END_MESSAGE_MAP()
  214. return true;
  215. }
  216. void StartThread(base::Thread* thread, base::MessagePumpType type) {
  217. base::Thread::Options options;
  218. options.message_pump_type = type;
  219. thread->StartWithOptions(std::move(options));
  220. }
  221. std::unique_ptr<WaitableEvent> done_;
  222. std::unique_ptr<WaitableEvent> channel_created_;
  223. mojo::ScopedMessagePipeHandle channel_handle_;
  224. Channel::Mode mode_;
  225. std::unique_ptr<SyncChannel> channel_;
  226. base::Thread ipc_thread_;
  227. base::Thread listener_thread_;
  228. raw_ptr<base::Thread> overrided_thread_;
  229. base::WaitableEvent shutdown_event_;
  230. bool is_shutdown_;
  231. };
  232. // Starts the test with the given workers. This function deletes the workers
  233. // when it's done.
  234. void RunTest(std::vector<Worker*> workers) {
  235. // First we create the workers that are channel servers, or else the other
  236. // workers' channel initialization might fail because the pipe isn't created..
  237. for (size_t i = 0; i < workers.size(); ++i) {
  238. if (workers[i]->mode() & Channel::MODE_SERVER_FLAG) {
  239. workers[i]->Start();
  240. workers[i]->WaitForChannelCreation();
  241. }
  242. }
  243. // now create the clients
  244. for (size_t i = 0; i < workers.size(); ++i) {
  245. if (workers[i]->mode() & Channel::MODE_CLIENT_FLAG)
  246. workers[i]->Start();
  247. }
  248. // wait for all the workers to finish
  249. for (size_t i = 0; i < workers.size(); ++i)
  250. workers[i]->done_event()->Wait();
  251. for (size_t i = 0; i < workers.size(); ++i) {
  252. workers[i]->Shutdown();
  253. delete workers[i];
  254. }
  255. }
  256. class IPCSyncChannelTest : public testing::Test {
  257. private:
  258. base::test::SingleThreadTaskEnvironment task_environment_;
  259. };
  260. //------------------------------------------------------------------------------
  261. class SimpleServer : public Worker {
  262. public:
  263. explicit SimpleServer(mojo::ScopedMessagePipeHandle channel_handle)
  264. : Worker(Channel::MODE_SERVER,
  265. "simpler_server",
  266. std::move(channel_handle)) {}
  267. void Run() override {
  268. SendAnswerToLife(true);
  269. Done();
  270. }
  271. };
  272. class SimpleClient : public Worker {
  273. public:
  274. explicit SimpleClient(mojo::ScopedMessagePipeHandle channel_handle)
  275. : Worker(Channel::MODE_CLIENT,
  276. "simple_client",
  277. std::move(channel_handle)) {}
  278. void OnAnswer(int* answer) override {
  279. *answer = 42;
  280. Done();
  281. }
  282. };
  283. void Simple() {
  284. std::vector<Worker*> workers;
  285. mojo::MessagePipe pipe;
  286. workers.push_back(new SimpleServer(std::move(pipe.handle0)));
  287. workers.push_back(new SimpleClient(std::move(pipe.handle1)));
  288. RunTest(workers);
  289. }
  290. #if BUILDFLAG(IS_ANDROID)
  291. #define MAYBE_Simple DISABLED_Simple
  292. #else
  293. #define MAYBE_Simple Simple
  294. #endif
  295. // Tests basic synchronous call
  296. TEST_F(IPCSyncChannelTest, MAYBE_Simple) {
  297. Simple();
  298. }
  299. //------------------------------------------------------------------------------
  300. // Worker classes which override how the sync channel is created to use the
  301. // two-step initialization (calling the lightweight constructor and then
  302. // ChannelProxy::Init separately) process.
  303. class TwoStepServer : public Worker {
  304. public:
  305. TwoStepServer(bool create_pipe_now,
  306. mojo::ScopedMessagePipeHandle channel_handle)
  307. : Worker(Channel::MODE_SERVER,
  308. "simpler_server",
  309. std::move(channel_handle)),
  310. create_pipe_now_(create_pipe_now) {}
  311. void Run() override {
  312. SendAnswerToLife(true);
  313. Done();
  314. }
  315. SyncChannel* CreateChannel() override {
  316. SyncChannel* channel =
  317. SyncChannel::Create(TakeChannelHandle(), mode(), this,
  318. ipc_thread().task_runner(),
  319. base::ThreadTaskRunnerHandle::Get(),
  320. create_pipe_now_, shutdown_event())
  321. .release();
  322. return channel;
  323. }
  324. bool create_pipe_now_;
  325. };
  326. class TwoStepClient : public Worker {
  327. public:
  328. TwoStepClient(bool create_pipe_now,
  329. mojo::ScopedMessagePipeHandle channel_handle)
  330. : Worker(Channel::MODE_CLIENT,
  331. "simple_client",
  332. std::move(channel_handle)),
  333. create_pipe_now_(create_pipe_now) {}
  334. void OnAnswer(int* answer) override {
  335. *answer = 42;
  336. Done();
  337. }
  338. SyncChannel* CreateChannel() override {
  339. SyncChannel* channel =
  340. SyncChannel::Create(TakeChannelHandle(), mode(), this,
  341. ipc_thread().task_runner(),
  342. base::ThreadTaskRunnerHandle::Get(),
  343. create_pipe_now_, shutdown_event())
  344. .release();
  345. return channel;
  346. }
  347. bool create_pipe_now_;
  348. };
  349. void TwoStep(bool create_server_pipe_now, bool create_client_pipe_now) {
  350. std::vector<Worker*> workers;
  351. mojo::MessagePipe pipe;
  352. workers.push_back(
  353. new TwoStepServer(create_server_pipe_now, std::move(pipe.handle0)));
  354. workers.push_back(
  355. new TwoStepClient(create_client_pipe_now, std::move(pipe.handle1)));
  356. RunTest(workers);
  357. }
  358. // Tests basic two-step initialization, where you call the lightweight
  359. // constructor then Init.
  360. TEST_F(IPCSyncChannelTest, TwoStepInitialization) {
  361. TwoStep(false, false);
  362. TwoStep(false, true);
  363. TwoStep(true, false);
  364. TwoStep(true, true);
  365. }
  366. //------------------------------------------------------------------------------
  367. class DelayClient : public Worker {
  368. public:
  369. explicit DelayClient(mojo::ScopedMessagePipeHandle channel_handle)
  370. : Worker(Channel::MODE_CLIENT,
  371. "delay_client",
  372. std::move(channel_handle)) {}
  373. void OnAnswerDelay(Message* reply_msg) override {
  374. SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
  375. Send(reply_msg);
  376. Done();
  377. }
  378. };
  379. void DelayReply() {
  380. std::vector<Worker*> workers;
  381. mojo::MessagePipe pipe;
  382. workers.push_back(new SimpleServer(std::move(pipe.handle0)));
  383. workers.push_back(new DelayClient(std::move(pipe.handle1)));
  384. RunTest(workers);
  385. }
  386. // Tests that asynchronous replies work
  387. TEST_F(IPCSyncChannelTest, DelayReply) {
  388. DelayReply();
  389. }
  390. //------------------------------------------------------------------------------
  391. class NoHangServer : public Worker {
  392. public:
  393. NoHangServer(WaitableEvent* got_first_reply,
  394. mojo::ScopedMessagePipeHandle channel_handle)
  395. : Worker(Channel::MODE_SERVER,
  396. "no_hang_server",
  397. std::move(channel_handle)),
  398. got_first_reply_(got_first_reply) {}
  399. void Run() override {
  400. SendAnswerToLife(true);
  401. got_first_reply_->Signal();
  402. SendAnswerToLife(false);
  403. Done();
  404. }
  405. raw_ptr<WaitableEvent> got_first_reply_;
  406. };
  407. class NoHangClient : public Worker {
  408. public:
  409. NoHangClient(WaitableEvent* got_first_reply,
  410. mojo::ScopedMessagePipeHandle channel_handle)
  411. : Worker(Channel::MODE_CLIENT,
  412. "no_hang_client",
  413. std::move(channel_handle)),
  414. got_first_reply_(got_first_reply) {}
  415. void OnAnswerDelay(Message* reply_msg) override {
  416. // Use the DELAY_REPLY macro so that we can force the reply to be sent
  417. // before this function returns (when the channel will be reset).
  418. SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
  419. Send(reply_msg);
  420. got_first_reply_->Wait();
  421. CloseChannel();
  422. Done();
  423. }
  424. raw_ptr<WaitableEvent> got_first_reply_;
  425. };
  426. void NoHang() {
  427. WaitableEvent got_first_reply(
  428. base::WaitableEvent::ResetPolicy::AUTOMATIC,
  429. base::WaitableEvent::InitialState::NOT_SIGNALED);
  430. std::vector<Worker*> workers;
  431. mojo::MessagePipe pipe;
  432. workers.push_back(
  433. new NoHangServer(&got_first_reply, std::move(pipe.handle0)));
  434. workers.push_back(
  435. new NoHangClient(&got_first_reply, std::move(pipe.handle1)));
  436. RunTest(workers);
  437. }
  438. // Tests that caller doesn't hang if receiver dies
  439. TEST_F(IPCSyncChannelTest, NoHang) {
  440. NoHang();
  441. }
  442. //------------------------------------------------------------------------------
  443. class UnblockServer : public Worker {
  444. public:
  445. UnblockServer(bool delete_during_send,
  446. mojo::ScopedMessagePipeHandle channel_handle)
  447. : Worker(Channel::MODE_SERVER,
  448. "unblock_server",
  449. std::move(channel_handle)),
  450. delete_during_send_(delete_during_send) {}
  451. void Run() override {
  452. if (delete_during_send_) {
  453. // Use custom code since race conditions mean the answer may or may not be
  454. // available.
  455. int answer = 0;
  456. SyncMessage* msg = new SyncChannelTestMsg_AnswerToLife(&answer);
  457. Send(msg);
  458. } else {
  459. SendAnswerToLife(true);
  460. }
  461. Done();
  462. }
  463. void OnDoubleDelay(int in, Message* reply_msg) override {
  464. SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, in * 2);
  465. Send(reply_msg);
  466. if (delete_during_send_)
  467. ResetChannel();
  468. }
  469. bool delete_during_send_;
  470. };
  471. class UnblockClient : public Worker {
  472. public:
  473. explicit UnblockClient(mojo::ScopedMessagePipeHandle channel_handle)
  474. : Worker(Channel::MODE_CLIENT,
  475. "unblock_client",
  476. std::move(channel_handle)) {}
  477. void OnAnswer(int* answer) override {
  478. SendDouble(true);
  479. *answer = 42;
  480. Done();
  481. }
  482. };
  483. void Unblock(bool delete_during_send) {
  484. std::vector<Worker*> workers;
  485. mojo::MessagePipe pipe;
  486. workers.push_back(
  487. new UnblockServer(delete_during_send, std::move(pipe.handle0)));
  488. workers.push_back(new UnblockClient(std::move(pipe.handle1)));
  489. RunTest(workers);
  490. }
  491. // Tests that the caller unblocks to answer a sync message from the receiver.
  492. TEST_F(IPCSyncChannelTest, Unblock) {
  493. Unblock(false);
  494. }
  495. //------------------------------------------------------------------------------
  496. #if BUILDFLAG(IS_ANDROID)
  497. #define MAYBE_ChannelDeleteDuringSend DISABLED_ChannelDeleteDuringSend
  498. #else
  499. #define MAYBE_ChannelDeleteDuringSend ChannelDeleteDuringSend
  500. #endif
  501. // Tests that the the SyncChannel object can be deleted during a Send.
  502. TEST_F(IPCSyncChannelTest, MAYBE_ChannelDeleteDuringSend) {
  503. Unblock(true);
  504. }
  505. //------------------------------------------------------------------------------
  506. class RecursiveServer : public Worker {
  507. public:
  508. RecursiveServer(bool expected_send_result,
  509. mojo::ScopedMessagePipeHandle channel_handle)
  510. : Worker(Channel::MODE_SERVER,
  511. "recursive_server",
  512. std::move(channel_handle)),
  513. expected_send_result_(expected_send_result) {}
  514. void Run() override {
  515. SendDouble(expected_send_result_);
  516. Done();
  517. }
  518. void OnDouble(int in, int* out) override {
  519. *out = in * 2;
  520. SendAnswerToLife(expected_send_result_);
  521. }
  522. bool expected_send_result_;
  523. };
  524. class RecursiveClient : public Worker {
  525. public:
  526. RecursiveClient(bool close_channel,
  527. mojo::ScopedMessagePipeHandle channel_handle)
  528. : Worker(Channel::MODE_CLIENT,
  529. "recursive_client",
  530. std::move(channel_handle)),
  531. close_channel_(close_channel) {}
  532. void OnDoubleDelay(int in, Message* reply_msg) override {
  533. SendDouble(!close_channel_);
  534. if (close_channel_) {
  535. delete reply_msg;
  536. } else {
  537. SyncChannelTestMsg_Double::WriteReplyParams(reply_msg, in * 2);
  538. Send(reply_msg);
  539. }
  540. Done();
  541. }
  542. void OnAnswerDelay(Message* reply_msg) override {
  543. if (close_channel_) {
  544. delete reply_msg;
  545. CloseChannel();
  546. } else {
  547. SyncChannelTestMsg_AnswerToLife::WriteReplyParams(reply_msg, 42);
  548. Send(reply_msg);
  549. }
  550. }
  551. bool close_channel_;
  552. };
  553. void Recursive() {
  554. std::vector<Worker*> workers;
  555. mojo::MessagePipe pipe;
  556. workers.push_back(new RecursiveServer(true, std::move(pipe.handle0)));
  557. workers.push_back(new RecursiveClient(false, std::move(pipe.handle1)));
  558. RunTest(workers);
  559. }
  560. // Tests a server calling Send while another Send is pending.
  561. TEST_F(IPCSyncChannelTest, Recursive) {
  562. Recursive();
  563. }
  564. //------------------------------------------------------------------------------
  565. void RecursiveNoHang() {
  566. std::vector<Worker*> workers;
  567. mojo::MessagePipe pipe;
  568. workers.push_back(new RecursiveServer(false, std::move(pipe.handle0)));
  569. workers.push_back(new RecursiveClient(true, std::move(pipe.handle1)));
  570. RunTest(workers);
  571. }
  572. // Tests that if a caller makes a sync call during an existing sync call and
  573. // the receiver dies, neither of the Send() calls hang.
  574. TEST_F(IPCSyncChannelTest, RecursiveNoHang) {
  575. RecursiveNoHang();
  576. }
  577. //------------------------------------------------------------------------------
  578. class MultipleServer1 : public Worker {
  579. public:
  580. explicit MultipleServer1(mojo::ScopedMessagePipeHandle channel_handle)
  581. : Worker(std::move(channel_handle), Channel::MODE_SERVER) {}
  582. void Run() override {
  583. SendDouble(true);
  584. Done();
  585. }
  586. };
  587. class MultipleClient1 : public Worker {
  588. public:
  589. MultipleClient1(WaitableEvent* client1_msg_received,
  590. WaitableEvent* client1_can_reply,
  591. mojo::ScopedMessagePipeHandle channel_handle)
  592. : Worker(std::move(channel_handle), Channel::MODE_CLIENT),
  593. client1_msg_received_(client1_msg_received),
  594. client1_can_reply_(client1_can_reply) {}
  595. void OnDouble(int in, int* out) override {
  596. client1_msg_received_->Signal();
  597. *out = in * 2;
  598. client1_can_reply_->Wait();
  599. Done();
  600. }
  601. private:
  602. WaitableEvent *client1_msg_received_, *client1_can_reply_;
  603. };
  604. class MultipleServer2 : public Worker {
  605. public:
  606. explicit MultipleServer2(mojo::ScopedMessagePipeHandle channel_handle)
  607. : Worker(std::move(channel_handle), Channel::MODE_SERVER) {}
  608. void OnAnswer(int* result) override {
  609. *result = 42;
  610. Done();
  611. }
  612. };
  613. class MultipleClient2 : public Worker {
  614. public:
  615. MultipleClient2(WaitableEvent* client1_msg_received,
  616. WaitableEvent* client1_can_reply,
  617. mojo::ScopedMessagePipeHandle channel_handle)
  618. : Worker(std::move(channel_handle), Channel::MODE_CLIENT),
  619. client1_msg_received_(client1_msg_received),
  620. client1_can_reply_(client1_can_reply) {}
  621. void Run() override {
  622. client1_msg_received_->Wait();
  623. SendAnswerToLife(true);
  624. client1_can_reply_->Signal();
  625. Done();
  626. }
  627. private:
  628. WaitableEvent *client1_msg_received_, *client1_can_reply_;
  629. };
  630. void Multiple() {
  631. std::vector<Worker*> workers;
  632. // A shared worker thread so that server1 and server2 run on one thread.
  633. base::Thread worker_thread("Multiple");
  634. ASSERT_TRUE(worker_thread.Start());
  635. // Server1 sends a sync msg to client1, which blocks the reply until
  636. // server2 (which runs on the same worker thread as server1) responds
  637. // to a sync msg from client2.
  638. WaitableEvent client1_msg_received(
  639. base::WaitableEvent::ResetPolicy::AUTOMATIC,
  640. base::WaitableEvent::InitialState::NOT_SIGNALED);
  641. WaitableEvent client1_can_reply(
  642. base::WaitableEvent::ResetPolicy::AUTOMATIC,
  643. base::WaitableEvent::InitialState::NOT_SIGNALED);
  644. Worker* worker;
  645. mojo::MessagePipe pipe1, pipe2;
  646. worker = new MultipleServer2(std::move(pipe2.handle0));
  647. worker->OverrideThread(&worker_thread);
  648. workers.push_back(worker);
  649. worker = new MultipleClient2(&client1_msg_received, &client1_can_reply,
  650. std::move(pipe2.handle1));
  651. workers.push_back(worker);
  652. worker = new MultipleServer1(std::move(pipe1.handle0));
  653. worker->OverrideThread(&worker_thread);
  654. workers.push_back(worker);
  655. worker = new MultipleClient1(&client1_msg_received, &client1_can_reply,
  656. std::move(pipe1.handle1));
  657. workers.push_back(worker);
  658. RunTest(workers);
  659. }
  660. // Tests that multiple SyncObjects on the same listener thread can unblock each
  661. // other.
  662. TEST_F(IPCSyncChannelTest, Multiple) {
  663. Multiple();
  664. }
  665. //------------------------------------------------------------------------------
  666. // This class provides server side functionality to test the case where
  667. // multiple sync channels are in use on the same thread on the client.
  668. class QueuedReplyServer : public Worker {
  669. public:
  670. QueuedReplyServer(base::Thread* listener_thread,
  671. mojo::ScopedMessagePipeHandle channel_handle,
  672. const std::string& reply_text)
  673. : Worker(std::move(channel_handle), Channel::MODE_SERVER),
  674. reply_text_(reply_text) {
  675. Worker::OverrideThread(listener_thread);
  676. }
  677. void OnNestedTestMsg(Message* reply_msg) override {
  678. VLOG(1) << __FUNCTION__ << " Sending reply: " << reply_text_;
  679. SyncChannelNestedTestMsg_String::WriteReplyParams(reply_msg, reply_text_);
  680. Send(reply_msg);
  681. Done();
  682. }
  683. private:
  684. std::string reply_text_;
  685. };
  686. // The QueuedReplyClient class provides functionality to test the case where
  687. // multiple sync channels are in use on the same thread.
  688. class QueuedReplyClient : public Worker {
  689. public:
  690. QueuedReplyClient(base::Thread* listener_thread,
  691. mojo::ScopedMessagePipeHandle channel_handle,
  692. const std::string& expected_text)
  693. : Worker(std::move(channel_handle), Channel::MODE_CLIENT),
  694. expected_text_(expected_text) {
  695. Worker::OverrideThread(listener_thread);
  696. }
  697. void Run() override {
  698. std::string response;
  699. SyncMessage* msg = new SyncChannelNestedTestMsg_String(&response);
  700. bool result = Send(msg);
  701. DCHECK(result);
  702. DCHECK_EQ(response, expected_text_);
  703. VLOG(1) << __FUNCTION__ << " Received reply: " << response;
  704. Done();
  705. }
  706. private:
  707. std::string expected_text_;
  708. };
  709. void QueuedReply() {
  710. std::vector<Worker*> workers;
  711. // A shared worker thread for servers
  712. base::Thread server_worker_thread("QueuedReply_ServerListener");
  713. ASSERT_TRUE(server_worker_thread.Start());
  714. base::Thread client_worker_thread("QueuedReply_ClientListener");
  715. ASSERT_TRUE(client_worker_thread.Start());
  716. Worker* worker;
  717. mojo::MessagePipe pipe1, pipe2;
  718. worker = new QueuedReplyServer(&server_worker_thread,
  719. std::move(pipe1.handle0), "Got first message");
  720. workers.push_back(worker);
  721. worker = new QueuedReplyServer(
  722. &server_worker_thread, std::move(pipe2.handle0), "Got second message");
  723. workers.push_back(worker);
  724. worker = new QueuedReplyClient(&client_worker_thread,
  725. std::move(pipe1.handle1), "Got first message");
  726. workers.push_back(worker);
  727. worker = new QueuedReplyClient(
  728. &client_worker_thread, std::move(pipe2.handle1), "Got second message");
  729. workers.push_back(worker);
  730. RunTest(workers);
  731. }
  732. // While a blocking send is in progress, the listener thread might answer other
  733. // synchronous messages. This tests that if during the response to another
  734. // message the reply to the original messages comes, it is queued up correctly
  735. // and the original Send is unblocked later.
  736. TEST_F(IPCSyncChannelTest, QueuedReply) {
  737. QueuedReply();
  738. }
  739. //------------------------------------------------------------------------------
  740. class TestSyncMessageFilter : public SyncMessageFilter {
  741. public:
  742. TestSyncMessageFilter(
  743. base::WaitableEvent* shutdown_event,
  744. Worker* worker,
  745. scoped_refptr<base::SingleThreadTaskRunner> task_runner)
  746. : SyncMessageFilter(shutdown_event),
  747. worker_(worker),
  748. task_runner_(task_runner) {}
  749. void OnFilterAdded(Channel* channel) override {
  750. SyncMessageFilter::OnFilterAdded(channel);
  751. task_runner_->PostTask(
  752. FROM_HERE,
  753. base::BindOnce(&TestSyncMessageFilter::SendMessageOnHelperThread,
  754. this));
  755. }
  756. void SendMessageOnHelperThread() {
  757. int answer = 0;
  758. bool result = Send(new SyncChannelTestMsg_AnswerToLife(&answer));
  759. DCHECK(result);
  760. DCHECK_EQ(answer, 42);
  761. worker_->Done();
  762. }
  763. private:
  764. ~TestSyncMessageFilter() override = default;
  765. raw_ptr<Worker> worker_;
  766. scoped_refptr<base::SingleThreadTaskRunner> task_runner_;
  767. };
  768. class SyncMessageFilterServer : public Worker {
  769. public:
  770. explicit SyncMessageFilterServer(mojo::ScopedMessagePipeHandle channel_handle)
  771. : Worker(Channel::MODE_SERVER,
  772. "sync_message_filter_server",
  773. std::move(channel_handle)),
  774. thread_("helper_thread") {
  775. base::Thread::Options options;
  776. options.message_pump_type = base::MessagePumpType::DEFAULT;
  777. thread_.StartWithOptions(std::move(options));
  778. filter_ = new TestSyncMessageFilter(shutdown_event(), this,
  779. thread_.task_runner());
  780. }
  781. void Run() override {
  782. channel()->AddFilter(filter_.get());
  783. }
  784. base::Thread thread_;
  785. scoped_refptr<TestSyncMessageFilter> filter_;
  786. };
  787. // This class provides functionality to test the case that a Send on the sync
  788. // channel does not crash after the channel has been closed.
  789. class ServerSendAfterClose : public Worker {
  790. public:
  791. explicit ServerSendAfterClose(mojo::ScopedMessagePipeHandle channel_handle)
  792. : Worker(Channel::MODE_SERVER,
  793. "simpler_server",
  794. std::move(channel_handle)),
  795. send_result_(true) {}
  796. bool SendDummy() {
  797. ListenerThread()->task_runner()->PostTask(
  798. FROM_HERE,
  799. base::BindOnce(base::IgnoreResult(&ServerSendAfterClose::Send),
  800. base::Unretained(this), new SyncChannelTestMsg_NoArgs));
  801. return true;
  802. }
  803. bool send_result() const {
  804. return send_result_;
  805. }
  806. private:
  807. void Run() override {
  808. CloseChannel();
  809. Done();
  810. }
  811. bool Send(Message* msg) override {
  812. send_result_ = Worker::Send(msg);
  813. Done();
  814. return send_result_;
  815. }
  816. bool send_result_;
  817. };
  818. // Tests basic synchronous call
  819. TEST_F(IPCSyncChannelTest, SyncMessageFilter) {
  820. std::vector<Worker*> workers;
  821. mojo::MessagePipe pipe;
  822. workers.push_back(new SyncMessageFilterServer(std::move(pipe.handle0)));
  823. workers.push_back(new SimpleClient(std::move(pipe.handle1)));
  824. RunTest(workers);
  825. }
  826. // Test the case when the channel is closed and a Send is attempted after that.
  827. TEST_F(IPCSyncChannelTest, SendAfterClose) {
  828. mojo::MessagePipe pipe;
  829. ServerSendAfterClose server(std::move(pipe.handle0));
  830. server.Start();
  831. server.done_event()->Wait();
  832. server.done_event()->Reset();
  833. server.SendDummy();
  834. server.done_event()->Wait();
  835. EXPECT_FALSE(server.send_result());
  836. server.Shutdown();
  837. }
  838. //------------------------------------------------------------------------------
  839. class RestrictedDispatchServer : public Worker {
  840. public:
  841. RestrictedDispatchServer(WaitableEvent* sent_ping_event,
  842. WaitableEvent* wait_event,
  843. mojo::ScopedMessagePipeHandle channel_handle)
  844. : Worker(std::move(channel_handle), Channel::MODE_SERVER),
  845. sent_ping_event_(sent_ping_event),
  846. wait_event_(wait_event) {}
  847. void OnDoPing(int ping) {
  848. // Send an asynchronous message that unblocks the caller.
  849. Message* msg = new SyncChannelTestMsg_Ping(ping);
  850. msg->set_unblock(true);
  851. Send(msg);
  852. // Signal the event after the message has been sent on the channel, on the
  853. // IPC thread.
  854. ipc_thread().task_runner()->PostTask(
  855. FROM_HERE, base::BindOnce(&RestrictedDispatchServer::OnPingSent,
  856. base::Unretained(this)));
  857. }
  858. void OnPingTTL(int ping, int* out) {
  859. *out = ping;
  860. wait_event_->Wait();
  861. }
  862. base::Thread* ListenerThread() { return Worker::ListenerThread(); }
  863. private:
  864. bool OnMessageReceived(const Message& message) override {
  865. IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchServer, message)
  866. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
  867. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_PingTTL, OnPingTTL)
  868. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
  869. IPC_END_MESSAGE_MAP()
  870. return true;
  871. }
  872. void OnPingSent() {
  873. sent_ping_event_->Signal();
  874. }
  875. void OnNoArgs() { }
  876. raw_ptr<WaitableEvent> sent_ping_event_;
  877. raw_ptr<WaitableEvent> wait_event_;
  878. };
  879. class NonRestrictedDispatchServer : public Worker {
  880. public:
  881. NonRestrictedDispatchServer(WaitableEvent* signal_event,
  882. mojo::ScopedMessagePipeHandle channel_handle)
  883. : Worker(std::move(channel_handle), Channel::MODE_SERVER),
  884. signal_event_(signal_event) {}
  885. base::Thread* ListenerThread() { return Worker::ListenerThread(); }
  886. void OnDoPingTTL(int ping) {
  887. int value = 0;
  888. Send(new SyncChannelTestMsg_PingTTL(ping, &value));
  889. signal_event_->Signal();
  890. }
  891. private:
  892. bool OnMessageReceived(const Message& message) override {
  893. IPC_BEGIN_MESSAGE_MAP(NonRestrictedDispatchServer, message)
  894. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
  895. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
  896. IPC_END_MESSAGE_MAP()
  897. return true;
  898. }
  899. void OnNoArgs() { }
  900. raw_ptr<WaitableEvent> signal_event_;
  901. };
  902. class RestrictedDispatchClient : public Worker {
  903. public:
  904. RestrictedDispatchClient(
  905. WaitableEvent* sent_ping_event,
  906. RestrictedDispatchServer* server,
  907. NonRestrictedDispatchServer* server2,
  908. int* success,
  909. mojo::ScopedMessagePipeHandle restricted_channel_handle,
  910. mojo::ScopedMessagePipeHandle non_restricted_channel_handle)
  911. : Worker(std::move(restricted_channel_handle), Channel::MODE_CLIENT),
  912. ping_(0),
  913. server_(server),
  914. server2_(server2),
  915. success_(success),
  916. sent_ping_event_(sent_ping_event),
  917. non_restricted_channel_handle_(
  918. std::move(non_restricted_channel_handle)) {}
  919. void Run() override {
  920. // Incoming messages from our channel should only be dispatched when we
  921. // send a message on that same channel.
  922. channel()->SetRestrictDispatchChannelGroup(1);
  923. server_->ListenerThread()->task_runner()->PostTask(
  924. FROM_HERE, base::BindOnce(&RestrictedDispatchServer::OnDoPing,
  925. base::Unretained(server_), 1));
  926. sent_ping_event_->Wait();
  927. Send(new SyncChannelTestMsg_NoArgs);
  928. if (ping_ == 1)
  929. ++*success_;
  930. else
  931. LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
  932. non_restricted_channel_ = SyncChannel::Create(
  933. non_restricted_channel_handle_.release(), IPC::Channel::MODE_CLIENT,
  934. this, ipc_thread().task_runner(), base::ThreadTaskRunnerHandle::Get(),
  935. true, shutdown_event());
  936. server_->ListenerThread()->task_runner()->PostTask(
  937. FROM_HERE, base::BindOnce(&RestrictedDispatchServer::OnDoPing,
  938. base::Unretained(server_), 2));
  939. sent_ping_event_->Wait();
  940. // Check that the incoming message is *not* dispatched when sending on the
  941. // non restricted channel.
  942. // TODO(piman): there is a possibility of a false positive race condition
  943. // here, if the message that was posted on the server-side end of the pipe
  944. // is not visible yet on the client side, but I don't know how to solve this
  945. // without hooking into the internals of SyncChannel. I haven't seen it in
  946. // practice (i.e. not setting SetRestrictDispatchToSameChannel does cause
  947. // the following to fail).
  948. non_restricted_channel_->Send(new SyncChannelTestMsg_NoArgs);
  949. if (ping_ == 1)
  950. ++*success_;
  951. else
  952. LOG(ERROR) << "Send dispatched message from restricted channel";
  953. Send(new SyncChannelTestMsg_NoArgs);
  954. if (ping_ == 2)
  955. ++*success_;
  956. else
  957. LOG(ERROR) << "Send failed to dispatch incoming message on same channel";
  958. // Check that the incoming message on the non-restricted channel is
  959. // dispatched when sending on the restricted channel.
  960. server2_->ListenerThread()->task_runner()->PostTask(
  961. FROM_HERE, base::BindOnce(&NonRestrictedDispatchServer::OnDoPingTTL,
  962. base::Unretained(server2_), 3));
  963. int value = 0;
  964. Send(new SyncChannelTestMsg_PingTTL(4, &value));
  965. if (ping_ == 3 && value == 4)
  966. ++*success_;
  967. else
  968. LOG(ERROR) << "Send failed to dispatch message from unrestricted channel";
  969. non_restricted_channel_->Send(new SyncChannelTestMsg_Done);
  970. non_restricted_channel_.reset();
  971. Send(new SyncChannelTestMsg_Done);
  972. Done();
  973. }
  974. private:
  975. bool OnMessageReceived(const Message& message) override {
  976. IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchClient, message)
  977. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Ping, OnPing)
  978. IPC_MESSAGE_HANDLER_DELAY_REPLY(SyncChannelTestMsg_PingTTL, OnPingTTL)
  979. IPC_END_MESSAGE_MAP()
  980. return true;
  981. }
  982. void OnPing(int ping) {
  983. ping_ = ping;
  984. }
  985. void OnPingTTL(int ping, IPC::Message* reply) {
  986. ping_ = ping;
  987. // This message comes from the NonRestrictedDispatchServer, we have to send
  988. // the reply back manually.
  989. SyncChannelTestMsg_PingTTL::WriteReplyParams(reply, ping);
  990. non_restricted_channel_->Send(reply);
  991. }
  992. int ping_;
  993. raw_ptr<RestrictedDispatchServer> server_;
  994. raw_ptr<NonRestrictedDispatchServer> server2_;
  995. raw_ptr<int> success_;
  996. raw_ptr<WaitableEvent> sent_ping_event_;
  997. std::unique_ptr<SyncChannel> non_restricted_channel_;
  998. mojo::ScopedMessagePipeHandle non_restricted_channel_handle_;
  999. };
  1000. TEST_F(IPCSyncChannelTest, RestrictedDispatch) {
  1001. WaitableEvent sent_ping_event(
  1002. base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1003. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1004. WaitableEvent wait_event(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1005. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1006. mojo::MessagePipe restricted_pipe, non_restricted_pipe;
  1007. RestrictedDispatchServer* server = new RestrictedDispatchServer(
  1008. &sent_ping_event, &wait_event, std::move(restricted_pipe.handle0));
  1009. NonRestrictedDispatchServer* server2 = new NonRestrictedDispatchServer(
  1010. &wait_event, std::move(non_restricted_pipe.handle0));
  1011. int success = 0;
  1012. std::vector<Worker*> workers;
  1013. workers.push_back(server);
  1014. workers.push_back(server2);
  1015. workers.push_back(
  1016. new RestrictedDispatchClient(&sent_ping_event, server, server2, &success,
  1017. std::move(restricted_pipe.handle1),
  1018. std::move(non_restricted_pipe.handle1)));
  1019. RunTest(workers);
  1020. EXPECT_EQ(4, success);
  1021. }
  1022. //------------------------------------------------------------------------------
  1023. // This test case inspired by crbug.com/108491
  1024. // We create two servers that use the same ListenerThread but have
  1025. // SetRestrictDispatchToSameChannel set to true.
  1026. // We create clients, then use some specific WaitableEvent wait/signalling to
  1027. // ensure that messages get dispatched in a way that causes a deadlock due to
  1028. // a nested dispatch and an eligible message in a higher-level dispatch's
  1029. // delayed_queue. Specifically, we start with client1 about so send an
  1030. // unblocking message to server1, while the shared listener thread for the
  1031. // servers server1 and server2 is about to send a non-unblocking message to
  1032. // client1. At the same time, client2 will be about to send an unblocking
  1033. // message to server2. Server1 will handle the client1->server1 message by
  1034. // telling server2 to send a non-unblocking message to client2.
  1035. // What should happen is that the send to server2 should find the pending,
  1036. // same-context client2->server2 message to dispatch, causing client2 to
  1037. // unblock then handle the server2->client2 message, so that the shared
  1038. // servers' listener thread can then respond to the client1->server1 message.
  1039. // Then client1 can handle the non-unblocking server1->client1 message.
  1040. // The old code would end up in a state where the server2->client2 message is
  1041. // sent, but the client2->server2 message (which is eligible for dispatch, and
  1042. // which is what client2 is waiting for) is stashed in a local delayed_queue
  1043. // that has server1's channel context, causing a deadlock.
  1044. // WaitableEvents in the events array are used to:
  1045. // event 0: indicate to client1 that server listener is in OnDoServerTask
  1046. // event 1: indicate to client1 that client2 listener is in OnDoClient2Task
  1047. // event 2: indicate to server1 that client2 listener is in OnDoClient2Task
  1048. // event 3: indicate to client2 that server listener is in OnDoServerTask
  1049. class RestrictedDispatchDeadlockServer : public Worker {
  1050. public:
  1051. RestrictedDispatchDeadlockServer(int server_num,
  1052. WaitableEvent* server_ready_event,
  1053. WaitableEvent** events,
  1054. RestrictedDispatchDeadlockServer* peer,
  1055. mojo::ScopedMessagePipeHandle channel_handle)
  1056. : Worker(std::move(channel_handle), Channel::MODE_SERVER),
  1057. server_num_(server_num),
  1058. server_ready_event_(server_ready_event),
  1059. events_(events),
  1060. peer_(peer) {}
  1061. void OnDoServerTask() {
  1062. events_[3]->Signal();
  1063. events_[2]->Wait();
  1064. events_[0]->Signal();
  1065. SendMessageToClient();
  1066. }
  1067. void Run() override {
  1068. channel()->SetRestrictDispatchChannelGroup(1);
  1069. server_ready_event_->Signal();
  1070. }
  1071. base::Thread* ListenerThread() { return Worker::ListenerThread(); }
  1072. private:
  1073. bool OnMessageReceived(const Message& message) override {
  1074. IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockServer, message)
  1075. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
  1076. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, Done)
  1077. IPC_END_MESSAGE_MAP()
  1078. return true;
  1079. }
  1080. void OnNoArgs() {
  1081. if (server_num_ == 1) {
  1082. DCHECK(peer_);
  1083. peer_->SendMessageToClient();
  1084. }
  1085. }
  1086. void SendMessageToClient() {
  1087. Message* msg = new SyncChannelTestMsg_NoArgs;
  1088. msg->set_unblock(false);
  1089. DCHECK(!msg->should_unblock());
  1090. Send(msg);
  1091. }
  1092. int server_num_;
  1093. raw_ptr<WaitableEvent> server_ready_event_;
  1094. raw_ptr<WaitableEvent*> events_;
  1095. raw_ptr<RestrictedDispatchDeadlockServer> peer_;
  1096. };
  1097. class RestrictedDispatchDeadlockClient2 : public Worker {
  1098. public:
  1099. RestrictedDispatchDeadlockClient2(
  1100. RestrictedDispatchDeadlockServer* server,
  1101. WaitableEvent* server_ready_event,
  1102. WaitableEvent** events,
  1103. mojo::ScopedMessagePipeHandle channel_handle)
  1104. : Worker(std::move(channel_handle), Channel::MODE_CLIENT),
  1105. server_ready_event_(server_ready_event),
  1106. events_(events),
  1107. received_msg_(false),
  1108. received_noarg_reply_(false),
  1109. done_issued_(false) {}
  1110. void Run() override {
  1111. server_ready_event_->Wait();
  1112. }
  1113. void OnDoClient2Task() {
  1114. events_[3]->Wait();
  1115. events_[1]->Signal();
  1116. events_[2]->Signal();
  1117. DCHECK(received_msg_ == false);
  1118. Message* message = new SyncChannelTestMsg_NoArgs;
  1119. message->set_unblock(true);
  1120. Send(message);
  1121. received_noarg_reply_ = true;
  1122. }
  1123. base::Thread* ListenerThread() { return Worker::ListenerThread(); }
  1124. private:
  1125. bool OnMessageReceived(const Message& message) override {
  1126. IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockClient2, message)
  1127. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
  1128. IPC_END_MESSAGE_MAP()
  1129. return true;
  1130. }
  1131. void OnNoArgs() {
  1132. received_msg_ = true;
  1133. PossiblyDone();
  1134. }
  1135. void PossiblyDone() {
  1136. if (received_noarg_reply_ && received_msg_) {
  1137. DCHECK(done_issued_ == false);
  1138. done_issued_ = true;
  1139. Send(new SyncChannelTestMsg_Done);
  1140. Done();
  1141. }
  1142. }
  1143. raw_ptr<WaitableEvent> server_ready_event_;
  1144. raw_ptr<WaitableEvent*> events_;
  1145. bool received_msg_;
  1146. bool received_noarg_reply_;
  1147. bool done_issued_;
  1148. };
  1149. class RestrictedDispatchDeadlockClient1 : public Worker {
  1150. public:
  1151. RestrictedDispatchDeadlockClient1(
  1152. RestrictedDispatchDeadlockServer* server,
  1153. RestrictedDispatchDeadlockClient2* peer,
  1154. WaitableEvent* server_ready_event,
  1155. WaitableEvent** events,
  1156. mojo::ScopedMessagePipeHandle channel_handle)
  1157. : Worker(std::move(channel_handle), Channel::MODE_CLIENT),
  1158. server_(server),
  1159. peer_(peer),
  1160. server_ready_event_(server_ready_event),
  1161. events_(events),
  1162. received_msg_(false),
  1163. received_noarg_reply_(false),
  1164. done_issued_(false) {}
  1165. void Run() override {
  1166. server_ready_event_->Wait();
  1167. server_->ListenerThread()->task_runner()->PostTask(
  1168. FROM_HERE,
  1169. base::BindOnce(&RestrictedDispatchDeadlockServer::OnDoServerTask,
  1170. base::Unretained(server_)));
  1171. peer_->ListenerThread()->task_runner()->PostTask(
  1172. FROM_HERE,
  1173. base::BindOnce(&RestrictedDispatchDeadlockClient2::OnDoClient2Task,
  1174. base::Unretained(peer_)));
  1175. events_[0]->Wait();
  1176. events_[1]->Wait();
  1177. DCHECK(received_msg_ == false);
  1178. Message* message = new SyncChannelTestMsg_NoArgs;
  1179. message->set_unblock(true);
  1180. Send(message);
  1181. received_noarg_reply_ = true;
  1182. PossiblyDone();
  1183. }
  1184. private:
  1185. bool OnMessageReceived(const Message& message) override {
  1186. IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchDeadlockClient1, message)
  1187. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_NoArgs, OnNoArgs)
  1188. IPC_END_MESSAGE_MAP()
  1189. return true;
  1190. }
  1191. void OnNoArgs() {
  1192. received_msg_ = true;
  1193. PossiblyDone();
  1194. }
  1195. void PossiblyDone() {
  1196. if (received_noarg_reply_ && received_msg_) {
  1197. DCHECK(done_issued_ == false);
  1198. done_issued_ = true;
  1199. Send(new SyncChannelTestMsg_Done);
  1200. Done();
  1201. }
  1202. }
  1203. raw_ptr<RestrictedDispatchDeadlockServer> server_;
  1204. raw_ptr<RestrictedDispatchDeadlockClient2> peer_;
  1205. raw_ptr<WaitableEvent> server_ready_event_;
  1206. raw_ptr<WaitableEvent*> events_;
  1207. bool received_msg_;
  1208. bool received_noarg_reply_;
  1209. bool done_issued_;
  1210. };
  1211. TEST_F(IPCSyncChannelTest, RestrictedDispatchDeadlock) {
  1212. std::vector<Worker*> workers;
  1213. // A shared worker thread so that server1 and server2 run on one thread.
  1214. base::Thread worker_thread("RestrictedDispatchDeadlock");
  1215. ASSERT_TRUE(worker_thread.Start());
  1216. WaitableEvent server1_ready(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1217. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1218. WaitableEvent server2_ready(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1219. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1220. WaitableEvent event0(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1221. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1222. WaitableEvent event1(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1223. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1224. WaitableEvent event2(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1225. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1226. WaitableEvent event3(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1227. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1228. WaitableEvent* events[4] = {&event0, &event1, &event2, &event3};
  1229. RestrictedDispatchDeadlockServer* server1;
  1230. RestrictedDispatchDeadlockServer* server2;
  1231. RestrictedDispatchDeadlockClient1* client1;
  1232. RestrictedDispatchDeadlockClient2* client2;
  1233. mojo::MessagePipe pipe1, pipe2;
  1234. server2 = new RestrictedDispatchDeadlockServer(
  1235. 2, &server2_ready, events, nullptr, std::move(pipe2.handle0));
  1236. server2->OverrideThread(&worker_thread);
  1237. workers.push_back(server2);
  1238. client2 = new RestrictedDispatchDeadlockClient2(
  1239. server2, &server2_ready, events, std::move(pipe2.handle1));
  1240. workers.push_back(client2);
  1241. server1 = new RestrictedDispatchDeadlockServer(
  1242. 1, &server1_ready, events, server2, std::move(pipe1.handle0));
  1243. server1->OverrideThread(&worker_thread);
  1244. workers.push_back(server1);
  1245. client1 = new RestrictedDispatchDeadlockClient1(
  1246. server1, client2, &server1_ready, events, std::move(pipe1.handle1));
  1247. workers.push_back(client1);
  1248. RunTest(workers);
  1249. }
  1250. //------------------------------------------------------------------------------
  1251. // This test case inspired by crbug.com/120530
  1252. // We create 4 workers that pipe to each other W1->W2->W3->W4->W1 then we send a
  1253. // message that recurses through 3, 4 or 5 steps to make sure, say, W1 can
  1254. // re-enter when called from W4 while it's sending a message to W2.
  1255. // The first worker drives the whole test so it must be treated specially.
  1256. class RestrictedDispatchPipeWorker : public Worker {
  1257. public:
  1258. RestrictedDispatchPipeWorker(mojo::ScopedMessagePipeHandle channel_handle1,
  1259. WaitableEvent* event1,
  1260. mojo::ScopedMessagePipeHandle channel_handle2,
  1261. WaitableEvent* event2,
  1262. int group,
  1263. int* success)
  1264. : Worker(std::move(channel_handle1), Channel::MODE_SERVER),
  1265. event1_(event1),
  1266. event2_(event2),
  1267. other_channel_handle_(std::move(channel_handle2)),
  1268. group_(group),
  1269. success_(success) {}
  1270. void OnPingTTL(int ping, int* ret) {
  1271. *ret = 0;
  1272. if (!ping)
  1273. return;
  1274. other_channel_->Send(new SyncChannelTestMsg_PingTTL(ping - 1, ret));
  1275. ++*ret;
  1276. }
  1277. void OnDone() {
  1278. if (is_first())
  1279. return;
  1280. other_channel_->Send(new SyncChannelTestMsg_Done);
  1281. other_channel_.reset();
  1282. Done();
  1283. }
  1284. void Run() override {
  1285. channel()->SetRestrictDispatchChannelGroup(group_);
  1286. if (is_first())
  1287. event1_->Signal();
  1288. event2_->Wait();
  1289. other_channel_ = SyncChannel::Create(
  1290. other_channel_handle_.release(), IPC::Channel::MODE_CLIENT, this,
  1291. ipc_thread().task_runner(), base::ThreadTaskRunnerHandle::Get(), true,
  1292. shutdown_event());
  1293. other_channel_->SetRestrictDispatchChannelGroup(group_);
  1294. if (!is_first()) {
  1295. event1_->Signal();
  1296. return;
  1297. }
  1298. *success_ = 0;
  1299. int value = 0;
  1300. OnPingTTL(3, &value);
  1301. *success_ += (value == 3);
  1302. OnPingTTL(4, &value);
  1303. *success_ += (value == 4);
  1304. OnPingTTL(5, &value);
  1305. *success_ += (value == 5);
  1306. other_channel_->Send(new SyncChannelTestMsg_Done);
  1307. other_channel_.reset();
  1308. Done();
  1309. }
  1310. bool is_first() { return !!success_; }
  1311. private:
  1312. bool OnMessageReceived(const Message& message) override {
  1313. IPC_BEGIN_MESSAGE_MAP(RestrictedDispatchPipeWorker, message)
  1314. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_PingTTL, OnPingTTL)
  1315. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Done, OnDone)
  1316. IPC_END_MESSAGE_MAP()
  1317. return true;
  1318. }
  1319. std::unique_ptr<SyncChannel> other_channel_;
  1320. raw_ptr<WaitableEvent> event1_;
  1321. raw_ptr<WaitableEvent> event2_;
  1322. mojo::ScopedMessagePipeHandle other_channel_handle_;
  1323. int group_;
  1324. raw_ptr<int> success_;
  1325. };
  1326. #if BUILDFLAG(IS_ANDROID)
  1327. #define MAYBE_RestrictedDispatch4WayDeadlock \
  1328. DISABLED_RestrictedDispatch4WayDeadlock
  1329. #else
  1330. #define MAYBE_RestrictedDispatch4WayDeadlock RestrictedDispatch4WayDeadlock
  1331. #endif
  1332. TEST_F(IPCSyncChannelTest, MAYBE_RestrictedDispatch4WayDeadlock) {
  1333. int success = 0;
  1334. std::vector<Worker*> workers;
  1335. WaitableEvent event0(base::WaitableEvent::ResetPolicy::MANUAL,
  1336. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1337. WaitableEvent event1(base::WaitableEvent::ResetPolicy::MANUAL,
  1338. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1339. WaitableEvent event2(base::WaitableEvent::ResetPolicy::MANUAL,
  1340. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1341. WaitableEvent event3(base::WaitableEvent::ResetPolicy::MANUAL,
  1342. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1343. mojo::MessagePipe pipe0, pipe1, pipe2, pipe3;
  1344. workers.push_back(new RestrictedDispatchPipeWorker(
  1345. std::move(pipe0.handle0), &event0, std::move(pipe1.handle1), &event1, 1,
  1346. &success));
  1347. workers.push_back(new RestrictedDispatchPipeWorker(
  1348. std::move(pipe1.handle0), &event1, std::move(pipe2.handle1), &event2, 2,
  1349. nullptr));
  1350. workers.push_back(new RestrictedDispatchPipeWorker(
  1351. std::move(pipe2.handle0), &event2, std::move(pipe3.handle1), &event3, 3,
  1352. nullptr));
  1353. workers.push_back(new RestrictedDispatchPipeWorker(
  1354. std::move(pipe3.handle0), &event3, std::move(pipe0.handle1), &event0, 4,
  1355. nullptr));
  1356. RunTest(workers);
  1357. EXPECT_EQ(3, success);
  1358. }
  1359. //------------------------------------------------------------------------------
  1360. // This test case inspired by crbug.com/122443
  1361. // We want to make sure a reply message with the unblock flag set correctly
  1362. // behaves as a reply, not a regular message.
  1363. // We have 3 workers. Server1 will send a message to Server2 (which will block),
  1364. // during which it will dispatch a message comming from Client, at which point
  1365. // it will send another message to Server2. While sending that second message it
  1366. // will receive a reply from Server1 with the unblock flag.
  1367. class ReentrantReplyServer1 : public Worker {
  1368. public:
  1369. ReentrantReplyServer1(WaitableEvent* server_ready,
  1370. mojo::ScopedMessagePipeHandle channel_handle1,
  1371. mojo::ScopedMessagePipeHandle channel_handle2)
  1372. : Worker(std::move(channel_handle1), Channel::MODE_SERVER),
  1373. server_ready_(server_ready),
  1374. other_channel_handle_(std::move(channel_handle2)) {}
  1375. void Run() override {
  1376. server2_channel_ = SyncChannel::Create(
  1377. other_channel_handle_.release(), IPC::Channel::MODE_CLIENT, this,
  1378. ipc_thread().task_runner(), base::ThreadTaskRunnerHandle::Get(), true,
  1379. shutdown_event());
  1380. server_ready_->Signal();
  1381. Message* msg = new SyncChannelTestMsg_Reentrant1();
  1382. server2_channel_->Send(msg);
  1383. server2_channel_.reset();
  1384. Done();
  1385. }
  1386. private:
  1387. bool OnMessageReceived(const Message& message) override {
  1388. IPC_BEGIN_MESSAGE_MAP(ReentrantReplyServer1, message)
  1389. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Reentrant2, OnReentrant2)
  1390. IPC_REPLY_HANDLER(OnReply)
  1391. IPC_END_MESSAGE_MAP()
  1392. return true;
  1393. }
  1394. void OnReentrant2() {
  1395. Message* msg = new SyncChannelTestMsg_Reentrant3();
  1396. server2_channel_->Send(msg);
  1397. }
  1398. void OnReply(const Message& message) {
  1399. // If we get here, the Send() will never receive the reply (thus would
  1400. // hang), so abort instead.
  1401. LOG(FATAL) << "Reply message was dispatched";
  1402. }
  1403. raw_ptr<WaitableEvent> server_ready_;
  1404. std::unique_ptr<SyncChannel> server2_channel_;
  1405. mojo::ScopedMessagePipeHandle other_channel_handle_;
  1406. };
  1407. class ReentrantReplyServer2 : public Worker {
  1408. public:
  1409. ReentrantReplyServer2(mojo::ScopedMessagePipeHandle channel_handle)
  1410. : Worker(std::move(channel_handle), Channel::MODE_SERVER),
  1411. reply_(nullptr) {}
  1412. private:
  1413. bool OnMessageReceived(const Message& message) override {
  1414. IPC_BEGIN_MESSAGE_MAP(ReentrantReplyServer2, message)
  1415. IPC_MESSAGE_HANDLER_DELAY_REPLY(
  1416. SyncChannelTestMsg_Reentrant1, OnReentrant1)
  1417. IPC_MESSAGE_HANDLER(SyncChannelTestMsg_Reentrant3, OnReentrant3)
  1418. IPC_END_MESSAGE_MAP()
  1419. return true;
  1420. }
  1421. void OnReentrant1(Message* reply) {
  1422. DCHECK(!reply_);
  1423. reply_ = reply;
  1424. }
  1425. void OnReentrant3() {
  1426. DCHECK(reply_);
  1427. Message* reply = reply_;
  1428. reply_ = nullptr;
  1429. reply->set_unblock(true);
  1430. Send(reply);
  1431. Done();
  1432. }
  1433. raw_ptr<Message> reply_;
  1434. };
  1435. class ReentrantReplyClient : public Worker {
  1436. public:
  1437. ReentrantReplyClient(WaitableEvent* server_ready,
  1438. mojo::ScopedMessagePipeHandle channel_handle)
  1439. : Worker(std::move(channel_handle), Channel::MODE_CLIENT),
  1440. server_ready_(server_ready) {}
  1441. void Run() override {
  1442. server_ready_->Wait();
  1443. Send(new SyncChannelTestMsg_Reentrant2());
  1444. Done();
  1445. }
  1446. private:
  1447. raw_ptr<WaitableEvent> server_ready_;
  1448. };
  1449. TEST_F(IPCSyncChannelTest, ReentrantReply) {
  1450. std::vector<Worker*> workers;
  1451. WaitableEvent server_ready(base::WaitableEvent::ResetPolicy::AUTOMATIC,
  1452. base::WaitableEvent::InitialState::NOT_SIGNALED);
  1453. mojo::MessagePipe pipe1, pipe2;
  1454. workers.push_back(new ReentrantReplyServer2(std::move(pipe2.handle0)));
  1455. workers.push_back(new ReentrantReplyServer1(
  1456. &server_ready, std::move(pipe1.handle0), std::move(pipe2.handle1)));
  1457. workers.push_back(
  1458. new ReentrantReplyClient(&server_ready, std::move(pipe1.handle1)));
  1459. RunTest(workers);
  1460. }
  1461. } // namespace
  1462. } // namespace IPC