data_pipe_unittest.cc 80 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108
  1. // Copyright 2015 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 <stddef.h>
  5. #include <stdint.h>
  6. #include <memory>
  7. #include "base/bind.h"
  8. #include "base/check_op.h"
  9. #include "base/location.h"
  10. #include "base/run_loop.h"
  11. #include "base/test/task_environment.h"
  12. #include "build/build_config.h"
  13. #include "mojo/core/embedder/embedder.h"
  14. #include "mojo/core/test/mojo_test_base.h"
  15. #include "mojo/public/c/system/data_pipe.h"
  16. #include "mojo/public/c/system/functions.h"
  17. #include "mojo/public/c/system/message_pipe.h"
  18. #include "mojo/public/cpp/system/message_pipe.h"
  19. #include "mojo/public/cpp/system/simple_watcher.h"
  20. #include "testing/gtest/include/gtest/gtest.h"
  21. namespace mojo {
  22. namespace core {
  23. namespace {
  24. const uint32_t kSizeOfOptions =
  25. static_cast<uint32_t>(sizeof(MojoCreateDataPipeOptions));
  26. // In various places, we have to poll (since, e.g., we can't yet wait for a
  27. // certain amount of data to be available). This is the maximum number of
  28. // iterations (separated by a short sleep).
  29. // TODO(vtl): Get rid of this.
  30. const size_t kMaxPoll = 100;
  31. // Used in Multiprocess test.
  32. const size_t kMultiprocessCapacity = 37;
  33. const char kMultiprocessTestData[] = "hello i'm a string that is 36 bytes";
  34. const int kMultiprocessMaxIter = 5;
  35. // Capacity that will cause data pipe creation to fail.
  36. constexpr size_t kOversizedCapacity = std::numeric_limits<uint32_t>::max();
  37. // A timeout smaller than |TestTimeouts::tiny_timeout()|, as a |MojoDeadline|.
  38. // Warning: This may lead to flakiness, but this is unavoidable if, e.g., you're
  39. // trying to ensure that functions with timeouts are reasonably accurate. We
  40. // want this to be as small as possible without causing too much flakiness.
  41. base::TimeDelta EpsilonDeadline() {
  42. const int64_t tiny_timeout = TestTimeouts::tiny_timeout().InMicroseconds();
  43. // Originally, our epsilon timeout was 10 ms, which was mostly fine but flaky on
  44. // some Windows bots. I don't recall ever seeing flakes on other bots. At 30 ms
  45. // tests seem reliable on Windows bots, but not at 25 ms. We'd like this timeout
  46. // to be as small as possible (see the description in the .h file).
  47. //
  48. // Currently, |tiny_timeout()| is usually 100 ms (possibly scaled under ASAN,
  49. // etc.). Based on this, set it to (usually be) 30 ms on Windows and 20 ms
  50. // elsewhere.
  51. #if BUILDFLAG(IS_WIN) || BUILDFLAG(IS_ANDROID)
  52. const int64_t deadline = (tiny_timeout * 3) / 10;
  53. #else
  54. const int64_t deadline = (tiny_timeout * 2) / 10;
  55. #endif
  56. return base::Microseconds(deadline);
  57. }
  58. // TODO(rockot): There are many uses of ASSERT where EXPECT would be more
  59. // appropriate. Fix this.
  60. class DataPipeTest : public test::MojoTestBase {
  61. public:
  62. DataPipeTest()
  63. : producer_(MOJO_HANDLE_INVALID), consumer_(MOJO_HANDLE_INVALID) {}
  64. DataPipeTest(const DataPipeTest&) = delete;
  65. DataPipeTest& operator=(const DataPipeTest&) = delete;
  66. ~DataPipeTest() override {
  67. if (producer_ != MOJO_HANDLE_INVALID)
  68. CHECK_EQ(MOJO_RESULT_OK, MojoClose(producer_));
  69. if (consumer_ != MOJO_HANDLE_INVALID)
  70. CHECK_EQ(MOJO_RESULT_OK, MojoClose(consumer_));
  71. }
  72. MojoResult ReadEmptyMessageWithHandles(MojoHandle pipe,
  73. MojoHandle* out_handles,
  74. uint32_t num_handles) {
  75. std::vector<uint8_t> bytes;
  76. std::vector<ScopedHandle> handles;
  77. MojoResult rv = ReadMessageRaw(MessagePipeHandle(pipe), &bytes, &handles,
  78. MOJO_READ_MESSAGE_FLAG_NONE);
  79. if (rv == MOJO_RESULT_OK) {
  80. CHECK_EQ(0u, bytes.size());
  81. CHECK_EQ(num_handles, handles.size());
  82. for (size_t i = 0; i < num_handles; ++i)
  83. out_handles[i] = handles[i].release().value();
  84. }
  85. return rv;
  86. }
  87. MojoResult Create(const MojoCreateDataPipeOptions* options) {
  88. return MojoCreateDataPipe(options, &producer_, &consumer_);
  89. }
  90. MojoResult WriteData(const void* elements,
  91. uint32_t* num_bytes,
  92. bool all_or_none = false) {
  93. MojoWriteDataOptions options;
  94. options.struct_size = sizeof(options);
  95. options.flags = all_or_none ? MOJO_WRITE_DATA_FLAG_ALL_OR_NONE
  96. : MOJO_WRITE_DATA_FLAG_NONE;
  97. return MojoWriteData(producer_, elements, num_bytes, &options);
  98. }
  99. MojoResult ReadData(void* elements,
  100. uint32_t* num_bytes,
  101. bool all_or_none = false,
  102. bool peek = false) {
  103. MojoReadDataFlags flags = MOJO_READ_DATA_FLAG_NONE;
  104. if (all_or_none)
  105. flags |= MOJO_READ_DATA_FLAG_ALL_OR_NONE;
  106. if (peek)
  107. flags |= MOJO_READ_DATA_FLAG_PEEK;
  108. MojoReadDataOptions options;
  109. options.struct_size = sizeof(options);
  110. options.flags = flags;
  111. return MojoReadData(consumer_, &options, elements, num_bytes);
  112. }
  113. MojoResult QueryData(uint32_t* num_bytes) {
  114. MojoReadDataOptions options;
  115. options.struct_size = sizeof(options);
  116. options.flags = MOJO_READ_DATA_FLAG_QUERY;
  117. return MojoReadData(consumer_, &options, nullptr, num_bytes);
  118. }
  119. MojoResult DiscardData(uint32_t* num_bytes, bool all_or_none = false) {
  120. MojoReadDataFlags flags = MOJO_READ_DATA_FLAG_DISCARD;
  121. if (all_or_none)
  122. flags |= MOJO_READ_DATA_FLAG_ALL_OR_NONE;
  123. MojoReadDataOptions options;
  124. options.struct_size = sizeof(options);
  125. options.flags = flags;
  126. return MojoReadData(consumer_, &options, nullptr, num_bytes);
  127. }
  128. MojoResult BeginReadData(const void** elements, uint32_t* num_bytes) {
  129. return MojoBeginReadData(consumer_, nullptr, elements, num_bytes);
  130. }
  131. MojoResult EndReadData(uint32_t num_bytes_read) {
  132. return MojoEndReadData(consumer_, num_bytes_read, nullptr);
  133. }
  134. MojoResult BeginWriteData(void** elements, uint32_t* num_bytes) {
  135. return MojoBeginWriteData(producer_, nullptr, elements, num_bytes);
  136. }
  137. MojoResult EndWriteData(uint32_t num_bytes_written) {
  138. return MojoEndWriteData(producer_, num_bytes_written, nullptr);
  139. }
  140. MojoResult CloseProducer() {
  141. MojoResult rv = MojoClose(producer_);
  142. producer_ = MOJO_HANDLE_INVALID;
  143. return rv;
  144. }
  145. MojoResult CloseConsumer() {
  146. MojoResult rv = MojoClose(consumer_);
  147. consumer_ = MOJO_HANDLE_INVALID;
  148. return rv;
  149. }
  150. MojoHandle producer_, consumer_;
  151. };
  152. TEST_F(DataPipeTest, Basic) {
  153. const MojoCreateDataPipeOptions options = {
  154. kSizeOfOptions, // |struct_size|.
  155. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  156. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  157. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  158. };
  159. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  160. // We can write to a data pipe handle immediately.
  161. int32_t elements[10] = {};
  162. uint32_t num_bytes = 0;
  163. num_bytes = static_cast<uint32_t>(std::size(elements) * sizeof(elements[0]));
  164. elements[0] = 123;
  165. elements[1] = 456;
  166. num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  167. ASSERT_EQ(MOJO_RESULT_OK, WriteData(&elements[0], &num_bytes));
  168. // Now wait for the other side to become readable.
  169. MojoHandleSignalsState state;
  170. ASSERT_EQ(MOJO_RESULT_OK,
  171. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &state));
  172. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  173. state.satisfied_signals);
  174. elements[0] = -1;
  175. elements[1] = -1;
  176. ASSERT_EQ(MOJO_RESULT_OK, ReadData(&elements[0], &num_bytes));
  177. ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
  178. ASSERT_EQ(elements[0], 123);
  179. ASSERT_EQ(elements[1], 456);
  180. }
  181. // Tests creation of data pipes with various (valid) options.
  182. TEST_F(DataPipeTest, CreateAndMaybeTransfer) {
  183. MojoCreateDataPipeOptions test_options[] = {
  184. // Default options.
  185. {},
  186. // Trivial element size, non-default capacity.
  187. {kSizeOfOptions, // |struct_size|.
  188. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  189. 1, // |element_num_bytes|.
  190. 1000}, // |capacity_num_bytes|.
  191. // Nontrivial element size, non-default capacity.
  192. {kSizeOfOptions, // |struct_size|.
  193. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  194. 4, // |element_num_bytes|.
  195. 4000}, // |capacity_num_bytes|.
  196. // Nontrivial element size, default capacity.
  197. {kSizeOfOptions, // |struct_size|.
  198. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  199. 100, // |element_num_bytes|.
  200. 0} // |capacity_num_bytes|.
  201. };
  202. for (size_t i = 0; i < std::size(test_options); i++) {
  203. MojoHandle producer_handle, consumer_handle;
  204. MojoCreateDataPipeOptions* options = i ? &test_options[i] : nullptr;
  205. ASSERT_EQ(MOJO_RESULT_OK,
  206. MojoCreateDataPipe(options, &producer_handle, &consumer_handle));
  207. ASSERT_EQ(MOJO_RESULT_OK, MojoClose(producer_handle));
  208. ASSERT_EQ(MOJO_RESULT_OK, MojoClose(consumer_handle));
  209. }
  210. }
  211. TEST_F(DataPipeTest, SimpleReadWrite) {
  212. const MojoCreateDataPipeOptions options = {
  213. kSizeOfOptions, // |struct_size|.
  214. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  215. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  216. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  217. };
  218. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  219. MojoHandleSignalsState hss;
  220. int32_t elements[10] = {};
  221. uint32_t num_bytes = 0;
  222. // Try reading; nothing there yet.
  223. num_bytes = static_cast<uint32_t>(std::size(elements) * sizeof(elements[0]));
  224. ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, ReadData(elements, &num_bytes));
  225. // Query; nothing there yet.
  226. num_bytes = 0;
  227. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  228. ASSERT_EQ(0u, num_bytes);
  229. // Discard; nothing there yet.
  230. num_bytes = static_cast<uint32_t>(5u * sizeof(elements[0]));
  231. ASSERT_EQ(MOJO_RESULT_SHOULD_WAIT, DiscardData(&num_bytes));
  232. // Read with invalid |num_bytes|.
  233. num_bytes = sizeof(elements[0]) + 1;
  234. ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, ReadData(elements, &num_bytes));
  235. // Write two elements.
  236. elements[0] = 123;
  237. elements[1] = 456;
  238. num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  239. ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes));
  240. // It should have written everything (even without "all or none").
  241. ASSERT_EQ(2u * sizeof(elements[0]), num_bytes);
  242. // Wait.
  243. ASSERT_EQ(MOJO_RESULT_OK,
  244. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  245. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  246. hss.satisfied_signals);
  247. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  248. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  249. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  250. hss.satisfiable_signals);
  251. // Query.
  252. // TODO(vtl): It's theoretically possible (though not with the current
  253. // implementation/configured limits) that not all the data has arrived yet.
  254. // (The theoretically-correct assertion here is that |num_bytes| is |1 * ...|
  255. // or |2 * ...|.)
  256. num_bytes = 0;
  257. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  258. ASSERT_EQ(2 * sizeof(elements[0]), num_bytes);
  259. // Read one element.
  260. elements[0] = -1;
  261. elements[1] = -1;
  262. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  263. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes));
  264. ASSERT_EQ(1u * sizeof(elements[0]), num_bytes);
  265. ASSERT_EQ(123, elements[0]);
  266. ASSERT_EQ(-1, elements[1]);
  267. // Query.
  268. // TODO(vtl): See previous TODO. (If we got 2 elements there, however, we
  269. // should get 1 here.)
  270. num_bytes = 0;
  271. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  272. ASSERT_EQ(1 * sizeof(elements[0]), num_bytes);
  273. // Peek one element.
  274. elements[0] = -1;
  275. elements[1] = -1;
  276. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  277. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, false, true));
  278. ASSERT_EQ(1u * sizeof(elements[0]), num_bytes);
  279. ASSERT_EQ(456, elements[0]);
  280. ASSERT_EQ(-1, elements[1]);
  281. // Query. Still has 1 element remaining.
  282. num_bytes = 0;
  283. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  284. ASSERT_EQ(1 * sizeof(elements[0]), num_bytes);
  285. // Try to read two elements, with "all or none".
  286. elements[0] = -1;
  287. elements[1] = -1;
  288. num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  289. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE,
  290. ReadData(elements, &num_bytes, true, false));
  291. ASSERT_EQ(-1, elements[0]);
  292. ASSERT_EQ(-1, elements[1]);
  293. // Try to read two elements, without "all or none".
  294. elements[0] = -1;
  295. elements[1] = -1;
  296. num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  297. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, false, false));
  298. ASSERT_EQ(1u * sizeof(elements[0]), num_bytes);
  299. ASSERT_EQ(456, elements[0]);
  300. ASSERT_EQ(-1, elements[1]);
  301. // Query.
  302. num_bytes = 0;
  303. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  304. ASSERT_EQ(0u, num_bytes);
  305. }
  306. // Note: The "basic" waiting tests test that the "wait states" are correct in
  307. // various situations; they don't test that waiters are properly awoken on state
  308. // changes. (For that, we need to use multiple threads.)
  309. TEST_F(DataPipeTest, BasicProducerWaiting) {
  310. // Note: We take advantage of the fact that current for current
  311. // implementations capacities are strict maximums. This is not guaranteed by
  312. // the API.
  313. const MojoCreateDataPipeOptions options = {
  314. kSizeOfOptions, // |struct_size|.
  315. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  316. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  317. 2 * sizeof(int32_t) // |capacity_num_bytes|.
  318. };
  319. Create(&options);
  320. MojoHandleSignalsState hss;
  321. // Never readable. Already writable.
  322. hss = GetSignalsState(producer_);
  323. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
  324. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  325. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  326. hss.satisfiable_signals);
  327. // Write two elements.
  328. int32_t elements[2] = {123, 456};
  329. uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  330. ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
  331. ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
  332. // Wait for data to become available to the consumer.
  333. ASSERT_EQ(MOJO_RESULT_OK,
  334. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  335. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  336. hss.satisfied_signals);
  337. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  338. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  339. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  340. hss.satisfiable_signals);
  341. // Peek one element.
  342. elements[0] = -1;
  343. elements[1] = -1;
  344. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  345. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, true));
  346. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  347. ASSERT_EQ(123, elements[0]);
  348. ASSERT_EQ(-1, elements[1]);
  349. // Read one element.
  350. elements[0] = -1;
  351. elements[1] = -1;
  352. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  353. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, false));
  354. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  355. ASSERT_EQ(123, elements[0]);
  356. ASSERT_EQ(-1, elements[1]);
  357. // Try writing, using a two-phase write.
  358. void* buffer = nullptr;
  359. num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
  360. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&buffer, &num_bytes));
  361. EXPECT_TRUE(buffer);
  362. ASSERT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(elements[0])));
  363. static_cast<int32_t*>(buffer)[0] = 789;
  364. ASSERT_EQ(MOJO_RESULT_OK,
  365. EndWriteData(static_cast<uint32_t>(1u * sizeof(elements[0]))));
  366. // Read one element, using a two-phase read.
  367. const void* read_buffer = nullptr;
  368. num_bytes = 0u;
  369. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes));
  370. EXPECT_TRUE(read_buffer);
  371. // The two-phase read should be able to read at least one element.
  372. ASSERT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(elements[0])));
  373. ASSERT_EQ(456, static_cast<const int32_t*>(read_buffer)[0]);
  374. ASSERT_EQ(MOJO_RESULT_OK,
  375. EndReadData(static_cast<uint32_t>(1u * sizeof(elements[0]))));
  376. // Write one element.
  377. elements[0] = 123;
  378. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  379. ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes));
  380. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  381. // Close the consumer.
  382. CloseConsumer();
  383. // It should now be never-writable.
  384. hss = MojoHandleSignalsState();
  385. ASSERT_EQ(MOJO_RESULT_OK,
  386. WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  387. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  388. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
  389. }
  390. TEST_F(DataPipeTest, PeerClosedProducerWaiting) {
  391. const MojoCreateDataPipeOptions options = {
  392. kSizeOfOptions, // |struct_size|.
  393. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  394. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  395. 2 * sizeof(int32_t) // |capacity_num_bytes|.
  396. };
  397. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  398. MojoHandleSignalsState hss;
  399. // Close the consumer.
  400. CloseConsumer();
  401. // It should be signaled.
  402. hss = MojoHandleSignalsState();
  403. ASSERT_EQ(MOJO_RESULT_OK,
  404. WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  405. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  406. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
  407. }
  408. TEST_F(DataPipeTest, PeerClosedConsumerWaiting) {
  409. const MojoCreateDataPipeOptions options = {
  410. kSizeOfOptions, // |struct_size|.
  411. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  412. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  413. 2 * sizeof(int32_t) // |capacity_num_bytes|.
  414. };
  415. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  416. MojoHandleSignalsState hss;
  417. // Close the producer.
  418. CloseProducer();
  419. // It should be signaled.
  420. hss = MojoHandleSignalsState();
  421. ASSERT_EQ(MOJO_RESULT_OK,
  422. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  423. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  424. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
  425. }
  426. TEST_F(DataPipeTest, BasicConsumerWaiting) {
  427. const MojoCreateDataPipeOptions options = {
  428. kSizeOfOptions, // |struct_size|.
  429. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  430. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  431. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  432. };
  433. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  434. MojoHandleSignalsState hss;
  435. // Never writable.
  436. hss = MojoHandleSignalsState();
  437. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  438. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_WRITABLE, &hss));
  439. EXPECT_EQ(0u, hss.satisfied_signals);
  440. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  441. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  442. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  443. hss.satisfiable_signals);
  444. // Write two elements.
  445. int32_t elements[2] = {123, 456};
  446. uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  447. ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
  448. // Wait for readability.
  449. hss = MojoHandleSignalsState();
  450. ASSERT_EQ(MOJO_RESULT_OK,
  451. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  452. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  453. hss.satisfied_signals);
  454. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  455. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  456. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  457. hss.satisfiable_signals);
  458. // Discard one element.
  459. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  460. ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true));
  461. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  462. // Should still be readable.
  463. hss = MojoHandleSignalsState();
  464. ASSERT_EQ(MOJO_RESULT_OK,
  465. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  466. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
  467. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  468. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  469. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  470. hss.satisfiable_signals);
  471. // Peek one element.
  472. elements[0] = -1;
  473. elements[1] = -1;
  474. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  475. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true, true));
  476. ASSERT_EQ(456, elements[0]);
  477. ASSERT_EQ(-1, elements[1]);
  478. // Should still be readable.
  479. hss = MojoHandleSignalsState();
  480. ASSERT_EQ(MOJO_RESULT_OK,
  481. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  482. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
  483. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  484. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  485. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  486. hss.satisfiable_signals);
  487. // Read one element.
  488. elements[0] = -1;
  489. elements[1] = -1;
  490. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  491. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true));
  492. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  493. ASSERT_EQ(456, elements[0]);
  494. ASSERT_EQ(-1, elements[1]);
  495. // Write one element.
  496. elements[0] = 789;
  497. elements[1] = -1;
  498. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  499. ASSERT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
  500. // Waiting should now succeed.
  501. hss = MojoHandleSignalsState();
  502. ASSERT_EQ(MOJO_RESULT_OK,
  503. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  504. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  505. hss.satisfied_signals);
  506. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  507. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  508. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  509. hss.satisfiable_signals);
  510. // Close the producer.
  511. CloseProducer();
  512. // Should still be readable.
  513. hss = MojoHandleSignalsState();
  514. ASSERT_EQ(MOJO_RESULT_OK,
  515. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  516. EXPECT_TRUE(hss.satisfied_signals & (MOJO_HANDLE_SIGNAL_READABLE |
  517. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE));
  518. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  519. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  520. hss.satisfiable_signals);
  521. // Wait for the peer closed signal.
  522. hss = MojoHandleSignalsState();
  523. ASSERT_EQ(MOJO_RESULT_OK,
  524. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  525. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  526. MOJO_HANDLE_SIGNAL_PEER_CLOSED,
  527. hss.satisfied_signals);
  528. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  529. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  530. hss.satisfiable_signals);
  531. // Read one element.
  532. elements[0] = -1;
  533. elements[1] = -1;
  534. num_bytes = static_cast<uint32_t>(1u * sizeof(elements[0]));
  535. ASSERT_EQ(MOJO_RESULT_OK, ReadData(elements, &num_bytes, true));
  536. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  537. ASSERT_EQ(789, elements[0]);
  538. ASSERT_EQ(-1, elements[1]);
  539. // Should be never-readable.
  540. hss = MojoHandleSignalsState();
  541. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  542. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  543. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  544. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
  545. }
  546. TEST_F(DataPipeTest, ConsumerNewDataReadable) {
  547. const MojoCreateDataPipeOptions create_options = {
  548. kSizeOfOptions, // |struct_size|.
  549. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  550. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  551. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  552. };
  553. EXPECT_EQ(MOJO_RESULT_OK, Create(&create_options));
  554. int32_t elements[2] = {123, 456};
  555. uint32_t num_bytes = static_cast<uint32_t>(2u * sizeof(elements[0]));
  556. EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
  557. // The consumer handle should appear to be readable and have new data.
  558. EXPECT_EQ(MOJO_RESULT_OK,
  559. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE));
  560. EXPECT_TRUE(GetSignalsState(consumer_).satisfied_signals &
  561. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE);
  562. // Now try to read a minimum of 6 elements.
  563. int32_t read_elements[6];
  564. uint32_t num_read_bytes = sizeof(read_elements);
  565. MojoReadDataOptions read_options;
  566. read_options.struct_size = sizeof(read_options);
  567. read_options.flags = MOJO_READ_DATA_FLAG_ALL_OR_NONE;
  568. EXPECT_EQ(
  569. MOJO_RESULT_OUT_OF_RANGE,
  570. MojoReadData(consumer_, &read_options, read_elements, &num_read_bytes));
  571. // The consumer should still appear to be readable but not with new data.
  572. EXPECT_TRUE(GetSignalsState(consumer_).satisfied_signals &
  573. MOJO_HANDLE_SIGNAL_READABLE);
  574. EXPECT_FALSE(GetSignalsState(consumer_).satisfied_signals &
  575. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE);
  576. // Write four more elements.
  577. EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
  578. EXPECT_EQ(MOJO_RESULT_OK, WriteData(elements, &num_bytes, true));
  579. // The consumer handle should once again appear to be readable.
  580. EXPECT_EQ(MOJO_RESULT_OK,
  581. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE));
  582. // Try again to read a minimum of 6 elements. Should succeed this time.
  583. EXPECT_EQ(MOJO_RESULT_OK, MojoReadData(consumer_, &read_options,
  584. read_elements, &num_read_bytes));
  585. // And now the consumer is unreadable.
  586. EXPECT_FALSE(GetSignalsState(consumer_).satisfied_signals &
  587. MOJO_HANDLE_SIGNAL_READABLE);
  588. EXPECT_FALSE(GetSignalsState(consumer_).satisfied_signals &
  589. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE);
  590. }
  591. // Test with two-phase APIs and also closing the producer with an active
  592. // consumer waiter.
  593. TEST_F(DataPipeTest, ConsumerWaitingTwoPhase) {
  594. const MojoCreateDataPipeOptions options = {
  595. kSizeOfOptions, // |struct_size|.
  596. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  597. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  598. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  599. };
  600. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  601. MojoHandleSignalsState hss;
  602. // Write two elements.
  603. int32_t* elements = nullptr;
  604. void* buffer = nullptr;
  605. // Request room for three (but we'll only write two).
  606. uint32_t num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
  607. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&buffer, &num_bytes));
  608. EXPECT_TRUE(buffer);
  609. EXPECT_GE(num_bytes, static_cast<uint32_t>(3u * sizeof(elements[0])));
  610. elements = static_cast<int32_t*>(buffer);
  611. elements[0] = 123;
  612. elements[1] = 456;
  613. ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(2u * sizeof(elements[0])));
  614. // Wait for readability.
  615. hss = MojoHandleSignalsState();
  616. ASSERT_EQ(MOJO_RESULT_OK,
  617. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  618. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  619. hss.satisfied_signals);
  620. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  621. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  622. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  623. hss.satisfiable_signals);
  624. // Read one element.
  625. // Two should be available, but only read one.
  626. const void* read_buffer = nullptr;
  627. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes));
  628. EXPECT_TRUE(read_buffer);
  629. ASSERT_EQ(static_cast<uint32_t>(2u * sizeof(elements[0])), num_bytes);
  630. const int32_t* read_elements = static_cast<const int32_t*>(read_buffer);
  631. ASSERT_EQ(123, read_elements[0]);
  632. ASSERT_EQ(MOJO_RESULT_OK, EndReadData(1u * sizeof(elements[0])));
  633. // Should still be readable.
  634. hss = MojoHandleSignalsState();
  635. ASSERT_EQ(MOJO_RESULT_OK,
  636. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  637. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
  638. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  639. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  640. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  641. hss.satisfiable_signals);
  642. // Read one element.
  643. // Request three, but not in all-or-none mode.
  644. read_buffer = nullptr;
  645. num_bytes = static_cast<uint32_t>(3u * sizeof(elements[0]));
  646. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes));
  647. EXPECT_TRUE(read_buffer);
  648. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(elements[0])), num_bytes);
  649. read_elements = static_cast<const int32_t*>(read_buffer);
  650. ASSERT_EQ(456, read_elements[0]);
  651. ASSERT_EQ(MOJO_RESULT_OK, EndReadData(1u * sizeof(elements[0])));
  652. // Close the producer.
  653. CloseProducer();
  654. // Should be never-readable.
  655. hss = MojoHandleSignalsState();
  656. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  657. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  658. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  659. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
  660. }
  661. // Tests that data pipes aren't writable/readable during two-phase writes/reads.
  662. TEST_F(DataPipeTest, BasicTwoPhaseWaiting) {
  663. const MojoCreateDataPipeOptions options = {
  664. kSizeOfOptions, // |struct_size|.
  665. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  666. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  667. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  668. };
  669. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  670. MojoHandleSignalsState hss;
  671. // It should be writable.
  672. hss = GetSignalsState(producer_);
  673. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
  674. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  675. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  676. hss.satisfiable_signals);
  677. uint32_t num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
  678. void* write_ptr = nullptr;
  679. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes));
  680. EXPECT_TRUE(write_ptr);
  681. EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
  682. // At this point, it shouldn't be writable.
  683. hss = GetSignalsState(producer_);
  684. ASSERT_EQ(0u, hss.satisfied_signals);
  685. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  686. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  687. hss.satisfiable_signals);
  688. // It shouldn't be readable yet either (we'll wait later).
  689. hss = GetSignalsState(consumer_);
  690. ASSERT_EQ(0u, hss.satisfied_signals);
  691. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  692. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  693. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  694. hss.satisfiable_signals);
  695. static_cast<int32_t*>(write_ptr)[0] = 123;
  696. ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(1u * sizeof(int32_t)));
  697. // It should immediately be writable again.
  698. hss = GetSignalsState(producer_);
  699. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
  700. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  701. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  702. hss.satisfiable_signals);
  703. // It should become readable.
  704. hss = MojoHandleSignalsState();
  705. ASSERT_EQ(MOJO_RESULT_OK,
  706. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  707. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  708. hss.satisfied_signals);
  709. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  710. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  711. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  712. hss.satisfiable_signals);
  713. // Start another two-phase write and check that it's readable even in the
  714. // middle of it.
  715. num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
  716. write_ptr = nullptr;
  717. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes));
  718. EXPECT_TRUE(write_ptr);
  719. EXPECT_GE(num_bytes, static_cast<uint32_t>(1u * sizeof(int32_t)));
  720. // It should be readable.
  721. hss = MojoHandleSignalsState();
  722. ASSERT_EQ(MOJO_RESULT_OK,
  723. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  724. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  725. hss.satisfied_signals);
  726. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  727. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  728. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  729. hss.satisfiable_signals);
  730. // End the two-phase write without writing anything.
  731. ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(0u));
  732. // Start a two-phase read.
  733. num_bytes = static_cast<uint32_t>(1u * sizeof(int32_t));
  734. const void* read_ptr = nullptr;
  735. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes));
  736. EXPECT_TRUE(read_ptr);
  737. ASSERT_EQ(static_cast<uint32_t>(1u * sizeof(int32_t)), num_bytes);
  738. // At this point, it should still be writable.
  739. hss = GetSignalsState(producer_);
  740. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE, hss.satisfied_signals);
  741. ASSERT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  742. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  743. hss.satisfiable_signals);
  744. // But not readable.
  745. hss = GetSignalsState(consumer_);
  746. ASSERT_EQ(0u, hss.satisfied_signals);
  747. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  748. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  749. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  750. hss.satisfiable_signals);
  751. // End the two-phase read without reading anything.
  752. ASSERT_EQ(MOJO_RESULT_OK, EndReadData(0u));
  753. // It should be readable again.
  754. hss = GetSignalsState(consumer_);
  755. ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE, hss.satisfied_signals);
  756. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  757. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  758. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  759. hss.satisfiable_signals);
  760. }
  761. void Seq(int32_t start, size_t count, int32_t* out) {
  762. for (size_t i = 0; i < count; i++)
  763. out[i] = start + static_cast<int32_t>(i);
  764. }
  765. TEST_F(DataPipeTest, AllOrNone) {
  766. const MojoCreateDataPipeOptions options = {
  767. kSizeOfOptions, // |struct_size|.
  768. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  769. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  770. 10 * sizeof(int32_t) // |capacity_num_bytes|.
  771. };
  772. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  773. MojoHandleSignalsState hss;
  774. // Try writing more than the total capacity of the pipe.
  775. uint32_t num_bytes = 20u * sizeof(int32_t);
  776. int32_t buffer[100];
  777. Seq(0, std::size(buffer), buffer);
  778. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, WriteData(buffer, &num_bytes, true));
  779. // Should still be empty.
  780. num_bytes = ~0u;
  781. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  782. ASSERT_EQ(0u, num_bytes);
  783. // Write some data.
  784. num_bytes = 5u * sizeof(int32_t);
  785. Seq(100, std::size(buffer), buffer);
  786. ASSERT_EQ(MOJO_RESULT_OK, WriteData(buffer, &num_bytes, true));
  787. ASSERT_EQ(5u * sizeof(int32_t), num_bytes);
  788. // Wait for data.
  789. // TODO(vtl): There's no real guarantee that all the data will become
  790. // available at once (except that in current implementations, with reasonable
  791. // limits, it will). Eventually, we'll be able to wait for a specified amount
  792. // of data to become available.
  793. hss = MojoHandleSignalsState();
  794. ASSERT_EQ(MOJO_RESULT_OK,
  795. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  796. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  797. hss.satisfied_signals);
  798. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  799. MOJO_HANDLE_SIGNAL_PEER_CLOSED | MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  800. hss.satisfiable_signals);
  801. // Half full.
  802. num_bytes = 0u;
  803. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  804. ASSERT_EQ(5u * sizeof(int32_t), num_bytes);
  805. // Try writing more than the available capacity of the pipe, but less than the
  806. // total capacity.
  807. num_bytes = 6u * sizeof(int32_t);
  808. Seq(200, std::size(buffer), buffer);
  809. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, WriteData(buffer, &num_bytes, true));
  810. // Try reading too much.
  811. num_bytes = 11u * sizeof(int32_t);
  812. memset(buffer, 0xab, sizeof(buffer));
  813. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, ReadData(buffer, &num_bytes, true));
  814. int32_t expected_buffer[100];
  815. memset(expected_buffer, 0xab, sizeof(expected_buffer));
  816. ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
  817. // Try discarding too much.
  818. num_bytes = 11u * sizeof(int32_t);
  819. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, DiscardData(&num_bytes, true));
  820. // Just a little.
  821. num_bytes = 2u * sizeof(int32_t);
  822. Seq(300, std::size(buffer), buffer);
  823. ASSERT_EQ(MOJO_RESULT_OK, WriteData(buffer, &num_bytes, true));
  824. ASSERT_EQ(2u * sizeof(int32_t), num_bytes);
  825. // Just right.
  826. num_bytes = 3u * sizeof(int32_t);
  827. Seq(400, std::size(buffer), buffer);
  828. ASSERT_EQ(MOJO_RESULT_OK, WriteData(buffer, &num_bytes, true));
  829. ASSERT_EQ(3u * sizeof(int32_t), num_bytes);
  830. // TODO(vtl): Hack (see also the TODO above): We can't currently wait for a
  831. // specified amount of data to be available, so poll.
  832. for (size_t i = 0; i < kMaxPoll; i++) {
  833. num_bytes = 0u;
  834. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  835. if (num_bytes >= 10u * sizeof(int32_t))
  836. break;
  837. base::PlatformThread::Sleep(EpsilonDeadline());
  838. }
  839. ASSERT_EQ(10u * sizeof(int32_t), num_bytes);
  840. // Read half.
  841. num_bytes = 5u * sizeof(int32_t);
  842. memset(buffer, 0xab, sizeof(buffer));
  843. ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, true));
  844. ASSERT_EQ(5u * sizeof(int32_t), num_bytes);
  845. memset(expected_buffer, 0xab, sizeof(expected_buffer));
  846. Seq(100, 5, expected_buffer);
  847. ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
  848. // Try reading too much again.
  849. num_bytes = 6u * sizeof(int32_t);
  850. memset(buffer, 0xab, sizeof(buffer));
  851. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, ReadData(buffer, &num_bytes, true));
  852. memset(expected_buffer, 0xab, sizeof(expected_buffer));
  853. ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
  854. // Try discarding too much again.
  855. num_bytes = 6u * sizeof(int32_t);
  856. ASSERT_EQ(MOJO_RESULT_OUT_OF_RANGE, DiscardData(&num_bytes, true));
  857. // Discard a little.
  858. num_bytes = 2u * sizeof(int32_t);
  859. ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true));
  860. ASSERT_EQ(2u * sizeof(int32_t), num_bytes);
  861. // Three left.
  862. num_bytes = 0u;
  863. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  864. ASSERT_EQ(3u * sizeof(int32_t), num_bytes);
  865. // Close the producer, then test producer-closed cases.
  866. CloseProducer();
  867. // Wait.
  868. hss = MojoHandleSignalsState();
  869. ASSERT_EQ(MOJO_RESULT_OK,
  870. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  871. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
  872. hss.satisfied_signals);
  873. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED,
  874. hss.satisfiable_signals);
  875. // Try reading too much; "failed precondition" since the producer is closed.
  876. num_bytes = 4u * sizeof(int32_t);
  877. memset(buffer, 0xab, sizeof(buffer));
  878. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  879. ReadData(buffer, &num_bytes, true));
  880. memset(expected_buffer, 0xab, sizeof(expected_buffer));
  881. ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
  882. // Try discarding too much; "failed precondition" again.
  883. num_bytes = 4u * sizeof(int32_t);
  884. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, DiscardData(&num_bytes, true));
  885. // Read a little.
  886. num_bytes = 2u * sizeof(int32_t);
  887. memset(buffer, 0xab, sizeof(buffer));
  888. ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, true));
  889. ASSERT_EQ(2u * sizeof(int32_t), num_bytes);
  890. memset(expected_buffer, 0xab, sizeof(expected_buffer));
  891. Seq(400, 2, expected_buffer);
  892. ASSERT_EQ(0, memcmp(buffer, expected_buffer, sizeof(buffer)));
  893. // Discard the remaining element.
  894. num_bytes = 1u * sizeof(int32_t);
  895. ASSERT_EQ(MOJO_RESULT_OK, DiscardData(&num_bytes, true));
  896. ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
  897. // Empty again.
  898. num_bytes = ~0u;
  899. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  900. ASSERT_EQ(0u, num_bytes);
  901. }
  902. // Tests that |ProducerWriteData()| and |ConsumerReadData()| writes and reads,
  903. // respectively, as much as possible, even if it may have to "wrap around" the
  904. // internal circular buffer. (Note that the two-phase write and read need not do
  905. // this.)
  906. TEST_F(DataPipeTest, WrapAround) {
  907. unsigned char test_data[1000];
  908. for (size_t i = 0; i < std::size(test_data); i++)
  909. test_data[i] = static_cast<unsigned char>(i);
  910. const MojoCreateDataPipeOptions options = {
  911. kSizeOfOptions, // |struct_size|.
  912. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  913. 1u, // |element_num_bytes|.
  914. 100u // |capacity_num_bytes|.
  915. };
  916. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  917. MojoHandleSignalsState hss;
  918. // Write 20 bytes.
  919. uint32_t num_bytes = 20u;
  920. ASSERT_EQ(MOJO_RESULT_OK, WriteData(&test_data[0], &num_bytes, true));
  921. ASSERT_EQ(20u, num_bytes);
  922. // Wait for data.
  923. ASSERT_EQ(MOJO_RESULT_OK,
  924. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  925. EXPECT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_READABLE);
  926. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  927. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  928. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  929. hss.satisfiable_signals);
  930. // Read 10 bytes.
  931. unsigned char read_buffer[1000] = {0};
  932. num_bytes = 10u;
  933. ASSERT_EQ(MOJO_RESULT_OK, ReadData(read_buffer, &num_bytes, true));
  934. ASSERT_EQ(10u, num_bytes);
  935. ASSERT_EQ(0, memcmp(read_buffer, &test_data[0], 10u));
  936. // Check that a two-phase write can now only write (at most) 80 bytes. (This
  937. // checks an implementation detail; this behavior is not guaranteed.)
  938. void* write_buffer_ptr = nullptr;
  939. num_bytes = 0u;
  940. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes));
  941. EXPECT_TRUE(write_buffer_ptr);
  942. ASSERT_EQ(80u, num_bytes);
  943. ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(0));
  944. size_t total_num_bytes = 0;
  945. while (total_num_bytes < 90) {
  946. // Wait to write.
  947. ASSERT_EQ(MOJO_RESULT_OK,
  948. WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_WRITABLE, &hss));
  949. ASSERT_EQ(hss.satisfied_signals, MOJO_HANDLE_SIGNAL_WRITABLE);
  950. ASSERT_EQ(hss.satisfiable_signals, MOJO_HANDLE_SIGNAL_WRITABLE |
  951. MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  952. MOJO_HANDLE_SIGNAL_PEER_REMOTE);
  953. // Write as much as we can.
  954. num_bytes = 100;
  955. ASSERT_EQ(MOJO_RESULT_OK,
  956. WriteData(&test_data[20 + total_num_bytes], &num_bytes, false));
  957. total_num_bytes += num_bytes;
  958. }
  959. ASSERT_EQ(90u, total_num_bytes);
  960. num_bytes = 0;
  961. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  962. ASSERT_EQ(100u, num_bytes);
  963. // Check that a two-phase read can now only read (at most) 90 bytes. (This
  964. // checks an implementation detail; this behavior is not guaranteed.)
  965. const void* read_buffer_ptr = nullptr;
  966. num_bytes = 0;
  967. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes));
  968. EXPECT_TRUE(read_buffer_ptr);
  969. ASSERT_EQ(90u, num_bytes);
  970. ASSERT_EQ(MOJO_RESULT_OK, EndReadData(0));
  971. // Read as much as possible. We should read 100 bytes.
  972. num_bytes =
  973. static_cast<uint32_t>(std::size(read_buffer) * sizeof(read_buffer[0]));
  974. memset(read_buffer, 0, num_bytes);
  975. ASSERT_EQ(MOJO_RESULT_OK, ReadData(read_buffer, &num_bytes));
  976. ASSERT_EQ(100u, num_bytes);
  977. ASSERT_EQ(0, memcmp(read_buffer, &test_data[10], 100u));
  978. }
  979. // Tests the behavior of writing (simple and two-phase), closing the producer,
  980. // then reading (simple and two-phase).
  981. TEST_F(DataPipeTest, WriteCloseProducerRead) {
  982. const char kTestData[] = "hello world";
  983. const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
  984. const MojoCreateDataPipeOptions options = {
  985. kSizeOfOptions, // |struct_size|.
  986. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  987. 1u, // |element_num_bytes|.
  988. 1000u // |capacity_num_bytes|.
  989. };
  990. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  991. // Write some data, so we'll have something to read.
  992. uint32_t num_bytes = kTestDataSize;
  993. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes, false));
  994. ASSERT_EQ(kTestDataSize, num_bytes);
  995. // Write it again, so we'll have something left over.
  996. num_bytes = kTestDataSize;
  997. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes, false));
  998. ASSERT_EQ(kTestDataSize, num_bytes);
  999. // Start two-phase write.
  1000. void* write_buffer_ptr = nullptr;
  1001. num_bytes = 0u;
  1002. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes));
  1003. EXPECT_TRUE(write_buffer_ptr);
  1004. EXPECT_GT(num_bytes, 0u);
  1005. // TODO(vtl): (See corresponding TODO in TwoPhaseAllOrNone.)
  1006. for (size_t i = 0; i < kMaxPoll; i++) {
  1007. num_bytes = 0u;
  1008. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1009. if (num_bytes >= 2u * kTestDataSize)
  1010. break;
  1011. base::PlatformThread::Sleep(EpsilonDeadline());
  1012. }
  1013. ASSERT_EQ(2u * kTestDataSize, num_bytes);
  1014. // Start two-phase read.
  1015. const void* read_buffer_ptr = nullptr;
  1016. num_bytes = 0u;
  1017. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes));
  1018. EXPECT_TRUE(read_buffer_ptr);
  1019. ASSERT_EQ(2u * kTestDataSize, num_bytes);
  1020. // Close the producer.
  1021. CloseProducer();
  1022. // The consumer can finish its two-phase read.
  1023. ASSERT_EQ(0, memcmp(read_buffer_ptr, kTestData, kTestDataSize));
  1024. ASSERT_EQ(MOJO_RESULT_OK, EndReadData(kTestDataSize));
  1025. // And start another.
  1026. read_buffer_ptr = nullptr;
  1027. num_bytes = 0u;
  1028. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes));
  1029. EXPECT_TRUE(read_buffer_ptr);
  1030. ASSERT_EQ(kTestDataSize, num_bytes);
  1031. }
  1032. // Tests the behavior of interrupting a two-phase read and write by closing the
  1033. // consumer.
  1034. TEST_F(DataPipeTest, TwoPhaseWriteReadCloseConsumer) {
  1035. const char kTestData[] = "hello world";
  1036. const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
  1037. const MojoCreateDataPipeOptions options = {
  1038. kSizeOfOptions, // |struct_size|.
  1039. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1040. 1u, // |element_num_bytes|.
  1041. 1000u // |capacity_num_bytes|.
  1042. };
  1043. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1044. MojoHandleSignalsState hss;
  1045. // Write some data, so we'll have something to read.
  1046. uint32_t num_bytes = kTestDataSize;
  1047. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
  1048. ASSERT_EQ(kTestDataSize, num_bytes);
  1049. // Start two-phase write.
  1050. void* write_buffer_ptr = nullptr;
  1051. num_bytes = 0u;
  1052. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes));
  1053. EXPECT_TRUE(write_buffer_ptr);
  1054. ASSERT_GT(num_bytes, kTestDataSize);
  1055. // Wait for data.
  1056. // TODO(vtl): (See corresponding TODO in AllOrNone.)
  1057. hss = MojoHandleSignalsState();
  1058. ASSERT_EQ(MOJO_RESULT_OK,
  1059. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1060. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1061. hss.satisfied_signals);
  1062. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1063. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  1064. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  1065. hss.satisfiable_signals);
  1066. // Start two-phase read.
  1067. const void* read_buffer_ptr = nullptr;
  1068. num_bytes = 0u;
  1069. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &num_bytes));
  1070. EXPECT_TRUE(read_buffer_ptr);
  1071. ASSERT_EQ(kTestDataSize, num_bytes);
  1072. // Close the consumer.
  1073. CloseConsumer();
  1074. // Wait for producer to know that the consumer is closed.
  1075. hss = MojoHandleSignalsState();
  1076. ASSERT_EQ(MOJO_RESULT_OK,
  1077. WaitForSignals(producer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  1078. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  1079. ASSERT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfiable_signals);
  1080. // Actually write some data. (Note: Premature freeing of the buffer would
  1081. // probably only be detected under ASAN or similar.)
  1082. memcpy(write_buffer_ptr, kTestData, kTestDataSize);
  1083. // Note: Even though the consumer has been closed, ending the two-phase
  1084. // write will report success.
  1085. ASSERT_EQ(MOJO_RESULT_OK, EndWriteData(kTestDataSize));
  1086. // But trying to write should result in failure.
  1087. num_bytes = kTestDataSize;
  1088. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, WriteData(kTestData, &num_bytes));
  1089. // As will trying to start another two-phase write.
  1090. write_buffer_ptr = nullptr;
  1091. num_bytes = 0u;
  1092. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  1093. BeginWriteData(&write_buffer_ptr, &num_bytes));
  1094. }
  1095. // Tests the behavior of "interrupting" a two-phase write by closing both the
  1096. // producer and the consumer.
  1097. TEST_F(DataPipeTest, TwoPhaseWriteCloseBoth) {
  1098. const uint32_t kTestDataSize = 15u;
  1099. const MojoCreateDataPipeOptions options = {
  1100. kSizeOfOptions, // |struct_size|.
  1101. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1102. 1u, // |element_num_bytes|.
  1103. 1000u // |capacity_num_bytes|.
  1104. };
  1105. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1106. // Start two-phase write.
  1107. void* write_buffer_ptr = nullptr;
  1108. uint32_t num_bytes = 0u;
  1109. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_buffer_ptr, &num_bytes));
  1110. EXPECT_TRUE(write_buffer_ptr);
  1111. ASSERT_GT(num_bytes, kTestDataSize);
  1112. }
  1113. // Tests the behavior of writing, closing the producer, and then reading (with
  1114. // and without data remaining).
  1115. TEST_F(DataPipeTest, WriteCloseProducerReadNoData) {
  1116. const char kTestData[] = "hello world";
  1117. const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
  1118. const MojoCreateDataPipeOptions options = {
  1119. kSizeOfOptions, // |struct_size|.
  1120. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1121. 1u, // |element_num_bytes|.
  1122. 1000u // |capacity_num_bytes|.
  1123. };
  1124. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1125. MojoHandleSignalsState hss;
  1126. // Write some data, so we'll have something to read.
  1127. uint32_t num_bytes = kTestDataSize;
  1128. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
  1129. ASSERT_EQ(kTestDataSize, num_bytes);
  1130. // Close the producer.
  1131. CloseProducer();
  1132. // Wait. (Note that once the consumer knows that the producer is closed, it
  1133. // must also know about all the data that was sent.)
  1134. hss = MojoHandleSignalsState();
  1135. ASSERT_EQ(MOJO_RESULT_OK,
  1136. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  1137. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1138. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1139. hss.satisfied_signals);
  1140. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1141. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1142. hss.satisfiable_signals);
  1143. // Peek that data.
  1144. char buffer[1000];
  1145. num_bytes = static_cast<uint32_t>(sizeof(buffer));
  1146. ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes, false, true));
  1147. ASSERT_EQ(kTestDataSize, num_bytes);
  1148. ASSERT_EQ(0, memcmp(buffer, kTestData, kTestDataSize));
  1149. // Read that data.
  1150. memset(buffer, 0, 1000);
  1151. num_bytes = static_cast<uint32_t>(sizeof(buffer));
  1152. ASSERT_EQ(MOJO_RESULT_OK, ReadData(buffer, &num_bytes));
  1153. ASSERT_EQ(kTestDataSize, num_bytes);
  1154. ASSERT_EQ(0, memcmp(buffer, kTestData, kTestDataSize));
  1155. // A second read should fail.
  1156. num_bytes = static_cast<uint32_t>(sizeof(buffer));
  1157. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, ReadData(buffer, &num_bytes));
  1158. // A two-phase read should also fail.
  1159. const void* read_buffer_ptr = nullptr;
  1160. num_bytes = 0u;
  1161. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  1162. BeginReadData(&read_buffer_ptr, &num_bytes));
  1163. // Ditto for discard.
  1164. num_bytes = 10u;
  1165. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, DiscardData(&num_bytes));
  1166. }
  1167. // Test that during a two phase read the memory stays valid even if more data
  1168. // comes in.
  1169. TEST_F(DataPipeTest, TwoPhaseReadMemoryStable) {
  1170. const char kTestData[] = "hello world";
  1171. const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
  1172. const MojoCreateDataPipeOptions options = {
  1173. kSizeOfOptions, // |struct_size|.
  1174. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1175. 1u, // |element_num_bytes|.
  1176. 1000u // |capacity_num_bytes|.
  1177. };
  1178. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1179. MojoHandleSignalsState hss;
  1180. // Write some data.
  1181. uint32_t num_bytes = kTestDataSize;
  1182. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
  1183. ASSERT_EQ(kTestDataSize, num_bytes);
  1184. // Wait for the data.
  1185. hss = MojoHandleSignalsState();
  1186. ASSERT_EQ(MOJO_RESULT_OK,
  1187. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1188. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1189. hss.satisfied_signals);
  1190. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1191. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  1192. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  1193. hss.satisfiable_signals);
  1194. // Begin a two-phase read.
  1195. const void* read_buffer_ptr = nullptr;
  1196. uint32_t read_buffer_size = 0u;
  1197. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer_ptr, &read_buffer_size));
  1198. // Write more data.
  1199. const char kExtraData[] = "bye world";
  1200. const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData));
  1201. num_bytes = kExtraDataSize;
  1202. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes));
  1203. ASSERT_EQ(kExtraDataSize, num_bytes);
  1204. // Close the producer.
  1205. CloseProducer();
  1206. // Wait. (Note that once the consumer knows that the producer is closed, it
  1207. // must also have received the extra data).
  1208. hss = MojoHandleSignalsState();
  1209. ASSERT_EQ(MOJO_RESULT_OK,
  1210. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &hss));
  1211. EXPECT_EQ(MOJO_HANDLE_SIGNAL_PEER_CLOSED, hss.satisfied_signals);
  1212. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1213. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1214. hss.satisfiable_signals);
  1215. // Read the two phase memory to check it's still valid.
  1216. ASSERT_EQ(0, memcmp(read_buffer_ptr, kTestData, kTestDataSize));
  1217. EndReadData(read_buffer_size);
  1218. }
  1219. // Test that two-phase reads/writes behave correctly when given invalid
  1220. // arguments.
  1221. TEST_F(DataPipeTest, TwoPhaseMoreInvalidArguments) {
  1222. const MojoCreateDataPipeOptions options = {
  1223. kSizeOfOptions, // |struct_size|.
  1224. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1225. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  1226. 10 * sizeof(int32_t) // |capacity_num_bytes|.
  1227. };
  1228. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1229. MojoHandleSignalsState hss;
  1230. // No data.
  1231. uint32_t num_bytes = 1000u;
  1232. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1233. ASSERT_EQ(0u, num_bytes);
  1234. // Try "ending" a two-phase write when one isn't active.
  1235. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION,
  1236. EndWriteData(1u * sizeof(int32_t)));
  1237. // Wait a bit, to make sure that if a signal were (incorrectly) sent, it'd
  1238. // have time to propagate.
  1239. base::PlatformThread::Sleep(EpsilonDeadline());
  1240. // Still no data.
  1241. num_bytes = 1000u;
  1242. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1243. ASSERT_EQ(0u, num_bytes);
  1244. // Try ending a two-phase write with an invalid amount (too much).
  1245. num_bytes = 0u;
  1246. void* write_ptr = nullptr;
  1247. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes));
  1248. ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  1249. EndWriteData(num_bytes + static_cast<uint32_t>(sizeof(int32_t))));
  1250. // But the two-phase write still ended.
  1251. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndWriteData(0u));
  1252. // Wait a bit (as above).
  1253. base::PlatformThread::Sleep(EpsilonDeadline());
  1254. // Still no data.
  1255. num_bytes = 1000u;
  1256. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1257. ASSERT_EQ(0u, num_bytes);
  1258. // Try ending a two-phase write with an invalid amount (not a multiple of the
  1259. // element size).
  1260. num_bytes = 0u;
  1261. write_ptr = nullptr;
  1262. ASSERT_EQ(MOJO_RESULT_OK, BeginWriteData(&write_ptr, &num_bytes));
  1263. EXPECT_GE(num_bytes, 1u);
  1264. ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, EndWriteData(1u));
  1265. // But the two-phase write still ended.
  1266. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndWriteData(0u));
  1267. // Wait a bit (as above).
  1268. base::PlatformThread::Sleep(EpsilonDeadline());
  1269. // Still no data.
  1270. num_bytes = 1000u;
  1271. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1272. ASSERT_EQ(0u, num_bytes);
  1273. // Now write some data, so we'll be able to try reading.
  1274. int32_t element = 123;
  1275. num_bytes = 1u * sizeof(int32_t);
  1276. ASSERT_EQ(MOJO_RESULT_OK, WriteData(&element, &num_bytes));
  1277. // Wait for data.
  1278. // TODO(vtl): (See corresponding TODO in AllOrNone.)
  1279. hss = MojoHandleSignalsState();
  1280. ASSERT_EQ(MOJO_RESULT_OK,
  1281. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1282. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1283. hss.satisfied_signals);
  1284. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1285. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  1286. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  1287. hss.satisfiable_signals);
  1288. // One element available.
  1289. num_bytes = 0u;
  1290. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1291. ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
  1292. // Try "ending" a two-phase read when one isn't active.
  1293. ASSERT_EQ(MOJO_RESULT_FAILED_PRECONDITION, EndReadData(1u * sizeof(int32_t)));
  1294. // Still one element available.
  1295. num_bytes = 0u;
  1296. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1297. ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
  1298. // Try ending a two-phase read with an invalid amount (too much).
  1299. num_bytes = 0u;
  1300. const void* read_ptr = nullptr;
  1301. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes));
  1302. ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT,
  1303. EndReadData(num_bytes + static_cast<uint32_t>(sizeof(int32_t))));
  1304. // Still one element available.
  1305. num_bytes = 0u;
  1306. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1307. ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
  1308. // Try ending a two-phase read with an invalid amount (not a multiple of the
  1309. // element size).
  1310. num_bytes = 0u;
  1311. read_ptr = nullptr;
  1312. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_ptr, &num_bytes));
  1313. ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
  1314. ASSERT_EQ(123, static_cast<const int32_t*>(read_ptr)[0]);
  1315. ASSERT_EQ(MOJO_RESULT_INVALID_ARGUMENT, EndReadData(1u));
  1316. // Still one element available.
  1317. num_bytes = 0u;
  1318. ASSERT_EQ(MOJO_RESULT_OK, QueryData(&num_bytes));
  1319. ASSERT_EQ(1u * sizeof(int32_t), num_bytes);
  1320. }
  1321. // Test that a producer can be sent over a MP.
  1322. TEST_F(DataPipeTest, SendProducer) {
  1323. const char kTestData[] = "hello world";
  1324. const uint32_t kTestDataSize = static_cast<uint32_t>(sizeof(kTestData));
  1325. const MojoCreateDataPipeOptions options = {
  1326. kSizeOfOptions, // |struct_size|.
  1327. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1328. 1u, // |element_num_bytes|.
  1329. 1000u // |capacity_num_bytes|.
  1330. };
  1331. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1332. MojoHandleSignalsState hss;
  1333. // Write some data.
  1334. uint32_t num_bytes = kTestDataSize;
  1335. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kTestData, &num_bytes));
  1336. ASSERT_EQ(kTestDataSize, num_bytes);
  1337. // Wait for the data.
  1338. hss = MojoHandleSignalsState();
  1339. ASSERT_EQ(MOJO_RESULT_OK,
  1340. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1341. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1342. hss.satisfied_signals);
  1343. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1344. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  1345. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  1346. hss.satisfiable_signals);
  1347. // Check the data.
  1348. const void* read_buffer = nullptr;
  1349. num_bytes = 0u;
  1350. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes));
  1351. ASSERT_EQ(0, memcmp(read_buffer, kTestData, kTestDataSize));
  1352. EndReadData(num_bytes);
  1353. // Now send the producer over a MP so that it's serialized.
  1354. MojoHandle pipe0, pipe1;
  1355. ASSERT_EQ(MOJO_RESULT_OK, MojoCreateMessagePipe(nullptr, &pipe0, &pipe1));
  1356. ASSERT_EQ(MOJO_RESULT_OK,
  1357. WriteMessageRaw(MessagePipeHandle(pipe0), nullptr, 0, &producer_, 1,
  1358. MOJO_WRITE_MESSAGE_FLAG_NONE));
  1359. producer_ = MOJO_HANDLE_INVALID;
  1360. ASSERT_EQ(MOJO_RESULT_OK,
  1361. WaitForSignals(pipe1, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1362. ASSERT_EQ(MOJO_RESULT_OK, ReadEmptyMessageWithHandles(pipe1, &producer_, 1));
  1363. // Write more data.
  1364. const char kExtraData[] = "bye world";
  1365. const uint32_t kExtraDataSize = static_cast<uint32_t>(sizeof(kExtraData));
  1366. num_bytes = kExtraDataSize;
  1367. ASSERT_EQ(MOJO_RESULT_OK, WriteData(kExtraData, &num_bytes));
  1368. ASSERT_EQ(kExtraDataSize, num_bytes);
  1369. // Wait for it.
  1370. hss = MojoHandleSignalsState();
  1371. ASSERT_EQ(MOJO_RESULT_OK,
  1372. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1373. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1374. hss.satisfied_signals);
  1375. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1376. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE |
  1377. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  1378. hss.satisfiable_signals);
  1379. // Check the second write.
  1380. num_bytes = 0u;
  1381. ASSERT_EQ(MOJO_RESULT_OK, BeginReadData(&read_buffer, &num_bytes));
  1382. ASSERT_EQ(0, memcmp(read_buffer, kExtraData, kExtraDataSize));
  1383. EndReadData(num_bytes);
  1384. ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0));
  1385. ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe1));
  1386. }
  1387. // Ensures that if a data pipe consumer whose producer has closed is passed over
  1388. // a message pipe, the deserialized dispatcher is also marked as having a closed
  1389. // peer.
  1390. TEST_F(DataPipeTest, ConsumerWithClosedProducerSent) {
  1391. const MojoCreateDataPipeOptions options = {
  1392. kSizeOfOptions, // |struct_size|.
  1393. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1394. static_cast<uint32_t>(sizeof(int32_t)), // |element_num_bytes|.
  1395. 1000 * sizeof(int32_t) // |capacity_num_bytes|.
  1396. };
  1397. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1398. // We can write to a data pipe handle immediately.
  1399. int32_t data = 123;
  1400. uint32_t num_bytes = sizeof(data);
  1401. ASSERT_EQ(MOJO_RESULT_OK, WriteData(&data, &num_bytes));
  1402. ASSERT_EQ(MOJO_RESULT_OK, CloseProducer());
  1403. // Now wait for the other side to become readable and to see the peer closed.
  1404. MojoHandleSignalsState state;
  1405. ASSERT_EQ(MOJO_RESULT_OK,
  1406. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &state));
  1407. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1408. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1409. state.satisfied_signals);
  1410. ASSERT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1411. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1412. state.satisfiable_signals);
  1413. // Now send the consumer over a MP so that it's serialized.
  1414. MojoHandle pipe0, pipe1;
  1415. ASSERT_EQ(MOJO_RESULT_OK, MojoCreateMessagePipe(nullptr, &pipe0, &pipe1));
  1416. ASSERT_EQ(MOJO_RESULT_OK,
  1417. WriteMessageRaw(MessagePipeHandle(pipe0), nullptr, 0, &consumer_, 1,
  1418. MOJO_WRITE_MESSAGE_FLAG_NONE));
  1419. consumer_ = MOJO_HANDLE_INVALID;
  1420. ASSERT_EQ(MOJO_RESULT_OK,
  1421. WaitForSignals(pipe1, MOJO_HANDLE_SIGNAL_READABLE, &state));
  1422. ASSERT_EQ(MOJO_RESULT_OK, ReadEmptyMessageWithHandles(pipe1, &consumer_, 1));
  1423. ASSERT_EQ(MOJO_RESULT_OK,
  1424. WaitForSignals(consumer_, MOJO_HANDLE_SIGNAL_PEER_CLOSED, &state));
  1425. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1426. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1427. state.satisfied_signals);
  1428. EXPECT_EQ(MOJO_HANDLE_SIGNAL_READABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1429. MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE,
  1430. state.satisfiable_signals);
  1431. int32_t read_data;
  1432. ASSERT_EQ(MOJO_RESULT_OK, ReadData(&read_data, &num_bytes));
  1433. ASSERT_EQ(sizeof(read_data), num_bytes);
  1434. ASSERT_EQ(data, read_data);
  1435. ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe0));
  1436. ASSERT_EQ(MOJO_RESULT_OK, MojoClose(pipe1));
  1437. }
  1438. bool WriteAllData(MojoHandle producer,
  1439. const void* elements,
  1440. uint32_t num_bytes) {
  1441. for (size_t i = 0; i < kMaxPoll; i++) {
  1442. // Write as much data as we can.
  1443. uint32_t write_bytes = num_bytes;
  1444. MojoResult result =
  1445. MojoWriteData(producer, elements, &write_bytes, nullptr);
  1446. if (result == MOJO_RESULT_OK) {
  1447. num_bytes -= write_bytes;
  1448. elements = static_cast<const uint8_t*>(elements) + write_bytes;
  1449. if (num_bytes == 0)
  1450. return true;
  1451. } else {
  1452. EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT, result);
  1453. }
  1454. MojoHandleSignalsState hss = MojoHandleSignalsState();
  1455. EXPECT_EQ(MOJO_RESULT_OK, test::MojoTestBase::WaitForSignals(
  1456. producer, MOJO_HANDLE_SIGNAL_WRITABLE, &hss));
  1457. EXPECT_TRUE(hss.satisfied_signals & MOJO_HANDLE_SIGNAL_WRITABLE);
  1458. EXPECT_EQ(MOJO_HANDLE_SIGNAL_WRITABLE | MOJO_HANDLE_SIGNAL_PEER_CLOSED |
  1459. MOJO_HANDLE_SIGNAL_PEER_REMOTE,
  1460. hss.satisfiable_signals);
  1461. }
  1462. return false;
  1463. }
  1464. // If |expect_empty| is true, expect |consumer| to be empty after reading.
  1465. bool ReadAllData(MojoHandle consumer,
  1466. void* elements,
  1467. uint32_t num_bytes,
  1468. bool expect_empty) {
  1469. for (size_t i = 0; i < kMaxPoll; i++) {
  1470. // Read as much data as we can.
  1471. uint32_t read_bytes = num_bytes;
  1472. MojoResult result = MojoReadData(consumer, nullptr, elements, &read_bytes);
  1473. if (result == MOJO_RESULT_OK) {
  1474. num_bytes -= read_bytes;
  1475. elements = static_cast<uint8_t*>(elements) + read_bytes;
  1476. if (num_bytes == 0) {
  1477. if (expect_empty) {
  1478. // Expect no more data.
  1479. base::PlatformThread::Sleep(TestTimeouts::tiny_timeout());
  1480. MojoReadDataOptions options;
  1481. options.struct_size = sizeof(options);
  1482. options.flags = MOJO_READ_DATA_FLAG_QUERY;
  1483. MojoReadData(consumer, &options, nullptr, &num_bytes);
  1484. EXPECT_EQ(0u, num_bytes);
  1485. }
  1486. return true;
  1487. }
  1488. } else {
  1489. EXPECT_EQ(MOJO_RESULT_SHOULD_WAIT, result);
  1490. }
  1491. MojoHandleSignalsState hss = MojoHandleSignalsState();
  1492. EXPECT_EQ(MOJO_RESULT_OK, test::MojoTestBase::WaitForSignals(
  1493. consumer, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1494. // Peer could have become closed while we're still waiting for data.
  1495. EXPECT_TRUE(MOJO_HANDLE_SIGNAL_READABLE & hss.satisfied_signals);
  1496. EXPECT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_READABLE);
  1497. EXPECT_TRUE(hss.satisfiable_signals & MOJO_HANDLE_SIGNAL_PEER_CLOSED);
  1498. }
  1499. return num_bytes == 0;
  1500. }
  1501. TEST_F(DataPipeTest, CreateOversized) {
  1502. const MojoCreateDataPipeOptions options = {
  1503. kSizeOfOptions, // |struct_size|.
  1504. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1505. 1, // |element_num_bytes|.
  1506. kOversizedCapacity, // |capacity_num_bytes|.
  1507. };
  1508. ASSERT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED, Create(&options));
  1509. }
  1510. #if !BUILDFLAG(IS_IOS)
  1511. TEST_F(DataPipeTest, Multiprocess) {
  1512. const uint32_t kTestDataSize =
  1513. static_cast<uint32_t>(sizeof(kMultiprocessTestData));
  1514. const MojoCreateDataPipeOptions options = {
  1515. kSizeOfOptions, // |struct_size|.
  1516. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1517. 1, // |element_num_bytes|.
  1518. kMultiprocessCapacity // |capacity_num_bytes|.
  1519. };
  1520. ASSERT_EQ(MOJO_RESULT_OK, Create(&options));
  1521. RunTestClient("MultiprocessClient", [&](MojoHandle server_mp) {
  1522. // Send some data before serialising and sending the data pipe over.
  1523. // This is the first write so we don't need to use WriteAllData.
  1524. uint32_t num_bytes = kTestDataSize;
  1525. ASSERT_EQ(MOJO_RESULT_OK,
  1526. WriteData(kMultiprocessTestData, &num_bytes, true));
  1527. ASSERT_EQ(kTestDataSize, num_bytes);
  1528. // Send child process the data pipe.
  1529. ASSERT_EQ(MOJO_RESULT_OK,
  1530. WriteMessageRaw(MessagePipeHandle(server_mp), nullptr, 0,
  1531. &consumer_, 1, MOJO_WRITE_MESSAGE_FLAG_NONE));
  1532. // Send a bunch of data of varying sizes.
  1533. uint8_t buffer[100];
  1534. int seq = 0;
  1535. for (int i = 0; i < kMultiprocessMaxIter; ++i) {
  1536. for (uint32_t size = 1; size <= kMultiprocessCapacity; size++) {
  1537. for (unsigned int j = 0; j < size; ++j)
  1538. buffer[j] = seq + j;
  1539. EXPECT_TRUE(WriteAllData(producer_, buffer, size));
  1540. seq += size;
  1541. }
  1542. }
  1543. // Write the test string in again.
  1544. ASSERT_TRUE(WriteAllData(producer_, kMultiprocessTestData, kTestDataSize));
  1545. // Swap ends.
  1546. ASSERT_EQ(MOJO_RESULT_OK,
  1547. WriteMessageRaw(MessagePipeHandle(server_mp), nullptr, 0,
  1548. &producer_, 1, MOJO_WRITE_MESSAGE_FLAG_NONE));
  1549. // Receive the consumer from the other side.
  1550. producer_ = MOJO_HANDLE_INVALID;
  1551. MojoHandleSignalsState hss = MojoHandleSignalsState();
  1552. ASSERT_EQ(MOJO_RESULT_OK,
  1553. WaitForSignals(server_mp, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1554. ASSERT_EQ(MOJO_RESULT_OK,
  1555. ReadEmptyMessageWithHandles(server_mp, &consumer_, 1));
  1556. // Read the test string twice. Once for when we sent it, and once for the
  1557. // other end sending it.
  1558. for (int i = 0; i < 2; ++i) {
  1559. EXPECT_TRUE(ReadAllData(consumer_, buffer, kTestDataSize, i == 1));
  1560. EXPECT_EQ(0, memcmp(buffer, kMultiprocessTestData, kTestDataSize));
  1561. }
  1562. WriteMessage(server_mp, "quit");
  1563. // Don't have to close the consumer here because it will be done for us.
  1564. });
  1565. }
  1566. DEFINE_TEST_CLIENT_TEST_WITH_PIPE(MultiprocessClient, DataPipeTest, client_mp) {
  1567. const uint32_t kTestDataSize =
  1568. static_cast<uint32_t>(sizeof(kMultiprocessTestData));
  1569. // Receive the data pipe from the other side.
  1570. MojoHandle consumer = MOJO_HANDLE_INVALID;
  1571. MojoHandleSignalsState hss = MojoHandleSignalsState();
  1572. ASSERT_EQ(MOJO_RESULT_OK,
  1573. WaitForSignals(client_mp, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1574. ASSERT_EQ(MOJO_RESULT_OK,
  1575. ReadEmptyMessageWithHandles(client_mp, &consumer, 1));
  1576. // Read the initial string that was sent.
  1577. int32_t buffer[100];
  1578. EXPECT_TRUE(ReadAllData(consumer, buffer, kTestDataSize, false));
  1579. EXPECT_EQ(0, memcmp(buffer, kMultiprocessTestData, kTestDataSize));
  1580. // Receive the main data and check it is correct.
  1581. int seq = 0;
  1582. uint8_t expected_buffer[100];
  1583. for (int i = 0; i < kMultiprocessMaxIter; ++i) {
  1584. for (uint32_t size = 1; size <= kMultiprocessCapacity; ++size) {
  1585. for (unsigned int j = 0; j < size; ++j)
  1586. expected_buffer[j] = seq + j;
  1587. EXPECT_TRUE(ReadAllData(consumer, buffer, size, false));
  1588. EXPECT_EQ(0, memcmp(buffer, expected_buffer, size));
  1589. seq += size;
  1590. }
  1591. }
  1592. // Swap ends.
  1593. ASSERT_EQ(MOJO_RESULT_OK,
  1594. WriteMessageRaw(MessagePipeHandle(client_mp), nullptr, 0, &consumer,
  1595. 1, MOJO_WRITE_MESSAGE_FLAG_NONE));
  1596. // Receive the producer from the other side.
  1597. MojoHandle producer = MOJO_HANDLE_INVALID;
  1598. hss = MojoHandleSignalsState();
  1599. ASSERT_EQ(MOJO_RESULT_OK,
  1600. WaitForSignals(client_mp, MOJO_HANDLE_SIGNAL_READABLE, &hss));
  1601. ASSERT_EQ(MOJO_RESULT_OK,
  1602. ReadEmptyMessageWithHandles(client_mp, &producer, 1));
  1603. // Write the test string one more time.
  1604. EXPECT_TRUE(WriteAllData(producer, kMultiprocessTestData, kTestDataSize));
  1605. // We swapped ends, so close the producer.
  1606. EXPECT_EQ(MOJO_RESULT_OK, MojoClose(producer));
  1607. // Wait to receive a "quit" message before exiting.
  1608. EXPECT_EQ("quit", ReadMessage(client_mp));
  1609. }
  1610. DEFINE_TEST_CLIENT_TEST_WITH_PIPE(WriteAndCloseProducer, DataPipeTest, h) {
  1611. MojoHandle p;
  1612. std::string message = ReadMessageWithHandles(h, &p, 1);
  1613. // Write some data to the producer and close it.
  1614. uint32_t num_bytes = static_cast<uint32_t>(message.size());
  1615. EXPECT_EQ(MOJO_RESULT_OK,
  1616. MojoWriteData(p, message.data(), &num_bytes, nullptr));
  1617. EXPECT_EQ(num_bytes, static_cast<uint32_t>(message.size()));
  1618. // Close the producer before quitting.
  1619. EXPECT_EQ(MOJO_RESULT_OK, MojoClose(p));
  1620. // Wait for a quit message.
  1621. EXPECT_EQ("quit", ReadMessage(h));
  1622. }
  1623. DEFINE_TEST_CLIENT_TEST_WITH_PIPE(ReadAndCloseConsumer, DataPipeTest, h) {
  1624. MojoHandle c;
  1625. std::string expected_message = ReadMessageWithHandles(h, &c, 1);
  1626. // Wait for the consumer to become readable.
  1627. EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(c, MOJO_HANDLE_SIGNAL_READABLE));
  1628. // Drain the consumer and expect to find the given message.
  1629. uint32_t num_bytes = static_cast<uint32_t>(expected_message.size());
  1630. std::vector<char> bytes(expected_message.size());
  1631. EXPECT_EQ(MOJO_RESULT_OK, MojoReadData(c, nullptr, bytes.data(), &num_bytes));
  1632. EXPECT_EQ(num_bytes, static_cast<uint32_t>(bytes.size()));
  1633. std::string message(bytes.data(), bytes.size());
  1634. EXPECT_EQ(expected_message, message);
  1635. EXPECT_EQ(MOJO_RESULT_OK, MojoClose(c));
  1636. // Wait for a quit message.
  1637. EXPECT_EQ("quit", ReadMessage(h));
  1638. }
  1639. TEST_F(DataPipeTest, SendConsumerAndCloseProducer) {
  1640. // Create a new data pipe.
  1641. MojoHandle p, c;
  1642. EXPECT_EQ(MOJO_RESULT_OK, MojoCreateDataPipe(nullptr, &p, &c));
  1643. RunTestClient("WriteAndCloseProducer", [&](MojoHandle producer_client) {
  1644. RunTestClient("ReadAndCloseConsumer", [&](MojoHandle consumer_client) {
  1645. const std::string kMessage = "Hello, world!";
  1646. WriteMessageWithHandles(producer_client, kMessage, &p, 1);
  1647. WriteMessageWithHandles(consumer_client, kMessage, &c, 1);
  1648. WriteMessage(consumer_client, "quit");
  1649. });
  1650. WriteMessage(producer_client, "quit");
  1651. });
  1652. }
  1653. DEFINE_TEST_CLIENT_TEST_WITH_PIPE(CreateAndWrite, DataPipeTest, h) {
  1654. const MojoCreateDataPipeOptions options = {
  1655. kSizeOfOptions, // |struct_size|.
  1656. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1657. 1, // |element_num_bytes|.
  1658. kMultiprocessCapacity // |capacity_num_bytes|.
  1659. };
  1660. MojoHandle p, c;
  1661. ASSERT_EQ(MOJO_RESULT_OK, MojoCreateDataPipe(&options, &p, &c));
  1662. const std::string kMessage = "Hello, world!";
  1663. WriteMessageWithHandles(h, kMessage, &c, 1);
  1664. // Write some data to the producer and close it.
  1665. uint32_t num_bytes = static_cast<uint32_t>(kMessage.size());
  1666. EXPECT_EQ(MOJO_RESULT_OK,
  1667. MojoWriteData(p, kMessage.data(), &num_bytes, nullptr));
  1668. EXPECT_EQ(num_bytes, static_cast<uint32_t>(kMessage.size()));
  1669. EXPECT_EQ(MOJO_RESULT_OK, MojoClose(p));
  1670. // Wait for a quit message.
  1671. EXPECT_EQ("quit", ReadMessage(h));
  1672. }
  1673. TEST_F(DataPipeTest, CreateInChild) {
  1674. RunTestClient("CreateAndWrite", [&](MojoHandle child) {
  1675. MojoHandle c;
  1676. std::string expected_message = ReadMessageWithHandles(child, &c, 1);
  1677. // Wait for the consumer to become readable.
  1678. EXPECT_EQ(MOJO_RESULT_OK, WaitForSignals(c, MOJO_HANDLE_SIGNAL_READABLE));
  1679. // Drain the consumer and expect to find the given message.
  1680. uint32_t num_bytes = static_cast<uint32_t>(expected_message.size());
  1681. std::vector<char> bytes(expected_message.size());
  1682. EXPECT_EQ(MOJO_RESULT_OK,
  1683. MojoReadData(c, nullptr, bytes.data(), &num_bytes));
  1684. EXPECT_EQ(num_bytes, static_cast<uint32_t>(bytes.size()));
  1685. std::string message(bytes.data(), bytes.size());
  1686. EXPECT_EQ(expected_message, message);
  1687. EXPECT_EQ(MOJO_RESULT_OK, MojoClose(c));
  1688. WriteMessage(child, "quit");
  1689. });
  1690. }
  1691. DEFINE_TEST_CLIENT_TEST_WITH_PIPE(DataPipeStatusChangeInTransitClient,
  1692. DataPipeTest,
  1693. parent) {
  1694. // This test verifies that peer closure is detectable through various
  1695. // mechanisms when it races with handle transfer.
  1696. MojoHandle handles[6];
  1697. EXPECT_EQ("o_O", ReadMessageWithHandles(parent, handles, 6));
  1698. MojoHandle* producers = &handles[0];
  1699. MojoHandle* consumers = &handles[3];
  1700. // Wait on producer 0
  1701. EXPECT_EQ(MOJO_RESULT_OK,
  1702. WaitForSignals(producers[0], MOJO_HANDLE_SIGNAL_PEER_CLOSED));
  1703. // Wait on consumer 0
  1704. EXPECT_EQ(MOJO_RESULT_OK,
  1705. WaitForSignals(consumers[0], MOJO_HANDLE_SIGNAL_PEER_CLOSED));
  1706. base::test::SingleThreadTaskEnvironment task_environment;
  1707. // Wait on producer 1 and consumer 1 using SimpleWatchers.
  1708. {
  1709. base::RunLoop run_loop;
  1710. int count = 0;
  1711. auto callback = base::BindRepeating(
  1712. [](base::RunLoop* loop, int* count, MojoResult result) {
  1713. EXPECT_EQ(MOJO_RESULT_OK, result);
  1714. if (++*count == 2)
  1715. loop->Quit();
  1716. },
  1717. &run_loop, &count);
  1718. SimpleWatcher producer_watcher(FROM_HERE,
  1719. SimpleWatcher::ArmingPolicy::AUTOMATIC,
  1720. base::SequencedTaskRunnerHandle::Get());
  1721. SimpleWatcher consumer_watcher(FROM_HERE,
  1722. SimpleWatcher::ArmingPolicy::AUTOMATIC,
  1723. base::SequencedTaskRunnerHandle::Get());
  1724. producer_watcher.Watch(Handle(producers[1]), MOJO_HANDLE_SIGNAL_PEER_CLOSED,
  1725. callback);
  1726. consumer_watcher.Watch(Handle(consumers[1]), MOJO_HANDLE_SIGNAL_PEER_CLOSED,
  1727. callback);
  1728. run_loop.Run();
  1729. EXPECT_EQ(2, count);
  1730. }
  1731. // Wait on producer 2 by polling with MojoWriteData.
  1732. MojoResult result;
  1733. do {
  1734. uint32_t num_bytes = 0;
  1735. result = MojoWriteData(producers[2], nullptr, &num_bytes, nullptr);
  1736. } while (result == MOJO_RESULT_OK);
  1737. EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result);
  1738. // Wait on consumer 2 by polling with MojoReadData.
  1739. do {
  1740. char byte;
  1741. uint32_t num_bytes = 1;
  1742. result = MojoReadData(consumers[2], nullptr, &byte, &num_bytes);
  1743. } while (result == MOJO_RESULT_SHOULD_WAIT);
  1744. EXPECT_EQ(MOJO_RESULT_FAILED_PRECONDITION, result);
  1745. for (size_t i = 0; i < 6; ++i)
  1746. CloseHandle(handles[i]);
  1747. }
  1748. TEST_F(DataPipeTest, StatusChangeInTransit) {
  1749. MojoHandle producers[6];
  1750. MojoHandle consumers[6];
  1751. for (size_t i = 0; i < 6; ++i)
  1752. CreateDataPipe(&producers[i], &consumers[i], 1);
  1753. RunTestClient("DataPipeStatusChangeInTransitClient", [&](MojoHandle child) {
  1754. MojoHandle handles[] = {producers[0], producers[1], producers[2],
  1755. consumers[3], consumers[4], consumers[5]};
  1756. // Send 3 producers and 3 consumers, and let their transfer race with their
  1757. // peers' closure.
  1758. WriteMessageWithHandles(child, "o_O", handles, 6);
  1759. for (size_t i = 0; i < 3; ++i)
  1760. CloseHandle(consumers[i]);
  1761. for (size_t i = 3; i < 6; ++i)
  1762. CloseHandle(producers[i]);
  1763. });
  1764. }
  1765. DEFINE_TEST_CLIENT_TEST_WITH_PIPE(CreateOversizedChild, DataPipeTest, h) {
  1766. const MojoCreateDataPipeOptions options = {
  1767. kSizeOfOptions, // |struct_size|.
  1768. MOJO_CREATE_DATA_PIPE_FLAG_NONE, // |flags|.
  1769. 1, // |element_num_bytes|.
  1770. kOversizedCapacity // |capacity_num_bytes|.
  1771. };
  1772. MojoHandle p, c;
  1773. ASSERT_EQ(MOJO_RESULT_RESOURCE_EXHAUSTED,
  1774. MojoCreateDataPipe(&options, &p, &c));
  1775. WriteMessage(h, "success");
  1776. // Wait for a quit message.
  1777. EXPECT_EQ("quit", ReadMessage(h));
  1778. }
  1779. TEST_F(DataPipeTest, CreateOversizedInChild) {
  1780. RunTestClient("CreateOversizedChild", [&](MojoHandle child) {
  1781. // Wait for the child to finish the test.
  1782. std::string expected_message = ReadMessage(child);
  1783. EXPECT_EQ("success", expected_message);
  1784. WriteMessage(child, "quit");
  1785. });
  1786. }
  1787. #endif // !BUILDFLAG(IS_IOS)
  1788. } // namespace
  1789. } // namespace core
  1790. } // namespace mojo