data_pipe_consumer_dispatcher.cc 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617
  1. // Copyright 2013 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 "mojo/core/data_pipe_consumer_dispatcher.h"
  5. #include <stddef.h>
  6. #include <stdint.h>
  7. #include <algorithm>
  8. #include <limits>
  9. #include <utility>
  10. #include "base/bind.h"
  11. #include "base/logging.h"
  12. #include "base/memory/ref_counted.h"
  13. #include "base/numerics/checked_math.h"
  14. #include "base/trace_event/trace_event.h"
  15. #include "mojo/core/core.h"
  16. #include "mojo/core/data_pipe_control_message.h"
  17. #include "mojo/core/node_controller.h"
  18. #include "mojo/core/platform_handle_utils.h"
  19. #include "mojo/core/request_context.h"
  20. #include "mojo/core/user_message_impl.h"
  21. #include "mojo/public/c/system/data_pipe.h"
  22. namespace mojo {
  23. namespace core {
  24. namespace {
  25. const uint8_t kFlagPeerClosed = 0x01;
  26. #pragma pack(push, 1)
  27. struct SerializedState {
  28. MojoCreateDataPipeOptions options;
  29. uint64_t pipe_id;
  30. uint32_t read_offset;
  31. uint32_t bytes_available;
  32. uint8_t flags;
  33. uint64_t buffer_guid_high;
  34. uint64_t buffer_guid_low;
  35. char padding[7];
  36. };
  37. static_assert(sizeof(SerializedState) % 8 == 0,
  38. "Invalid SerializedState size.");
  39. #pragma pack(pop)
  40. } // namespace
  41. // A PortObserver which forwards to a DataPipeConsumerDispatcher. This owns a
  42. // reference to the dispatcher to ensure it lives as long as the observed port.
  43. class DataPipeConsumerDispatcher::PortObserverThunk
  44. : public NodeController::PortObserver {
  45. public:
  46. explicit PortObserverThunk(
  47. scoped_refptr<DataPipeConsumerDispatcher> dispatcher)
  48. : dispatcher_(dispatcher) {}
  49. PortObserverThunk(const PortObserverThunk&) = delete;
  50. PortObserverThunk& operator=(const PortObserverThunk&) = delete;
  51. private:
  52. ~PortObserverThunk() override = default;
  53. // NodeController::PortObserver:
  54. void OnPortStatusChanged() override { dispatcher_->OnPortStatusChanged(); }
  55. scoped_refptr<DataPipeConsumerDispatcher> dispatcher_;
  56. };
  57. // static
  58. scoped_refptr<DataPipeConsumerDispatcher> DataPipeConsumerDispatcher::Create(
  59. NodeController* node_controller,
  60. const ports::PortRef& control_port,
  61. base::UnsafeSharedMemoryRegion shared_ring_buffer,
  62. const MojoCreateDataPipeOptions& options,
  63. uint64_t pipe_id) {
  64. scoped_refptr<DataPipeConsumerDispatcher> consumer =
  65. new DataPipeConsumerDispatcher(node_controller, control_port,
  66. std::move(shared_ring_buffer), options,
  67. pipe_id);
  68. base::AutoLock lock(consumer->lock_);
  69. if (!consumer->InitializeNoLock())
  70. return nullptr;
  71. return consumer;
  72. }
  73. Dispatcher::Type DataPipeConsumerDispatcher::GetType() const {
  74. return Type::DATA_PIPE_CONSUMER;
  75. }
  76. MojoResult DataPipeConsumerDispatcher::Close() {
  77. base::AutoLock lock(lock_);
  78. DVLOG(1) << "Closing data pipe consumer " << pipe_id_;
  79. return CloseNoLock();
  80. }
  81. MojoResult DataPipeConsumerDispatcher::ReadData(
  82. const MojoReadDataOptions& options,
  83. void* elements,
  84. uint32_t* num_bytes) {
  85. base::AutoLock lock(lock_);
  86. if (!shared_ring_buffer_.IsValid() || in_transit_)
  87. return MOJO_RESULT_INVALID_ARGUMENT;
  88. if (in_two_phase_read_)
  89. return MOJO_RESULT_BUSY;
  90. const bool had_new_data = new_data_available_;
  91. new_data_available_ = false;
  92. if ((options.flags & MOJO_READ_DATA_FLAG_QUERY)) {
  93. if ((options.flags & MOJO_READ_DATA_FLAG_PEEK) ||
  94. (options.flags & MOJO_READ_DATA_FLAG_DISCARD))
  95. return MOJO_RESULT_INVALID_ARGUMENT;
  96. DCHECK(!(options.flags & MOJO_READ_DATA_FLAG_DISCARD)); // Handled above.
  97. DVLOG_IF(2, elements) << "Query mode: ignoring non-null |elements|";
  98. *num_bytes = static_cast<uint32_t>(bytes_available_);
  99. if (had_new_data)
  100. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  101. return MOJO_RESULT_OK;
  102. }
  103. bool discard = false;
  104. if ((options.flags & MOJO_READ_DATA_FLAG_DISCARD)) {
  105. // These flags are mutally exclusive.
  106. if (options.flags & MOJO_READ_DATA_FLAG_PEEK)
  107. return MOJO_RESULT_INVALID_ARGUMENT;
  108. DVLOG_IF(2, elements) << "Discard mode: ignoring non-null |elements|";
  109. discard = true;
  110. }
  111. uint32_t max_num_bytes_to_read = *num_bytes;
  112. if (max_num_bytes_to_read % options_.element_num_bytes != 0)
  113. return MOJO_RESULT_INVALID_ARGUMENT;
  114. bool all_or_none = options.flags & MOJO_READ_DATA_FLAG_ALL_OR_NONE;
  115. uint32_t min_num_bytes_to_read = all_or_none ? max_num_bytes_to_read : 0;
  116. if (min_num_bytes_to_read > bytes_available_) {
  117. if (had_new_data)
  118. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  119. return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
  120. : MOJO_RESULT_OUT_OF_RANGE;
  121. }
  122. uint32_t bytes_to_read = std::min(max_num_bytes_to_read, bytes_available_);
  123. if (bytes_to_read == 0) {
  124. if (had_new_data)
  125. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  126. return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
  127. : MOJO_RESULT_SHOULD_WAIT;
  128. }
  129. if (!discard) {
  130. const uint8_t* data =
  131. static_cast<const uint8_t*>(ring_buffer_mapping_.memory());
  132. CHECK(data);
  133. uint8_t* destination = static_cast<uint8_t*>(elements);
  134. CHECK(destination);
  135. DCHECK_LE(read_offset_, options_.capacity_num_bytes);
  136. uint32_t tail_bytes_to_copy =
  137. std::min(options_.capacity_num_bytes - read_offset_, bytes_to_read);
  138. uint32_t head_bytes_to_copy = bytes_to_read - tail_bytes_to_copy;
  139. if (tail_bytes_to_copy > 0)
  140. memcpy(destination, data + read_offset_, tail_bytes_to_copy);
  141. if (head_bytes_to_copy > 0)
  142. memcpy(destination + tail_bytes_to_copy, data, head_bytes_to_copy);
  143. }
  144. *num_bytes = bytes_to_read;
  145. bool peek = !!(options.flags & MOJO_READ_DATA_FLAG_PEEK);
  146. if (discard || !peek) {
  147. read_offset_ = (read_offset_ + bytes_to_read) % options_.capacity_num_bytes;
  148. bytes_available_ -= bytes_to_read;
  149. base::AutoUnlock unlock(lock_);
  150. NotifyRead(bytes_to_read);
  151. }
  152. // We may have just read the last available data and thus changed the signals
  153. // state.
  154. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  155. return MOJO_RESULT_OK;
  156. }
  157. MojoResult DataPipeConsumerDispatcher::BeginReadData(
  158. const void** buffer,
  159. uint32_t* buffer_num_bytes) {
  160. base::AutoLock lock(lock_);
  161. if (!shared_ring_buffer_.IsValid() || in_transit_)
  162. return MOJO_RESULT_INVALID_ARGUMENT;
  163. if (in_two_phase_read_)
  164. return MOJO_RESULT_BUSY;
  165. const bool had_new_data = new_data_available_;
  166. new_data_available_ = false;
  167. if (bytes_available_ == 0) {
  168. if (had_new_data)
  169. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  170. return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
  171. : MOJO_RESULT_SHOULD_WAIT;
  172. }
  173. DCHECK_LT(read_offset_, options_.capacity_num_bytes);
  174. uint32_t bytes_to_read =
  175. std::min(bytes_available_, options_.capacity_num_bytes - read_offset_);
  176. CHECK(ring_buffer_mapping_.IsValid());
  177. uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_.memory());
  178. CHECK(data);
  179. in_two_phase_read_ = true;
  180. *buffer = data + read_offset_;
  181. *buffer_num_bytes = bytes_to_read;
  182. two_phase_max_bytes_read_ = bytes_to_read;
  183. if (had_new_data)
  184. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  185. return MOJO_RESULT_OK;
  186. }
  187. MojoResult DataPipeConsumerDispatcher::EndReadData(uint32_t num_bytes_read) {
  188. base::AutoLock lock(lock_);
  189. if (!in_two_phase_read_)
  190. return MOJO_RESULT_FAILED_PRECONDITION;
  191. if (in_transit_)
  192. return MOJO_RESULT_INVALID_ARGUMENT;
  193. CHECK(shared_ring_buffer_.IsValid());
  194. MojoResult rv;
  195. if (num_bytes_read > two_phase_max_bytes_read_ ||
  196. num_bytes_read % options_.element_num_bytes != 0) {
  197. rv = MOJO_RESULT_INVALID_ARGUMENT;
  198. } else {
  199. rv = MOJO_RESULT_OK;
  200. read_offset_ =
  201. (read_offset_ + num_bytes_read) % options_.capacity_num_bytes;
  202. DCHECK_GE(bytes_available_, num_bytes_read);
  203. bytes_available_ -= num_bytes_read;
  204. base::AutoUnlock unlock(lock_);
  205. NotifyRead(num_bytes_read);
  206. }
  207. in_two_phase_read_ = false;
  208. two_phase_max_bytes_read_ = 0;
  209. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  210. return rv;
  211. }
  212. HandleSignalsState DataPipeConsumerDispatcher::GetHandleSignalsState() const {
  213. base::AutoLock lock(lock_);
  214. return GetHandleSignalsStateNoLock();
  215. }
  216. MojoResult DataPipeConsumerDispatcher::AddWatcherRef(
  217. const scoped_refptr<WatcherDispatcher>& watcher,
  218. uintptr_t context) {
  219. base::AutoLock lock(lock_);
  220. if (is_closed_ || in_transit_)
  221. return MOJO_RESULT_INVALID_ARGUMENT;
  222. return watchers_.Add(watcher, context, GetHandleSignalsStateNoLock());
  223. }
  224. MojoResult DataPipeConsumerDispatcher::RemoveWatcherRef(
  225. WatcherDispatcher* watcher,
  226. uintptr_t context) {
  227. base::AutoLock lock(lock_);
  228. if (is_closed_ || in_transit_)
  229. return MOJO_RESULT_INVALID_ARGUMENT;
  230. return watchers_.Remove(watcher, context);
  231. }
  232. void DataPipeConsumerDispatcher::StartSerialize(uint32_t* num_bytes,
  233. uint32_t* num_ports,
  234. uint32_t* num_handles) {
  235. base::AutoLock lock(lock_);
  236. DCHECK(in_transit_);
  237. *num_bytes = static_cast<uint32_t>(sizeof(SerializedState));
  238. *num_ports = 1;
  239. *num_handles = 1;
  240. }
  241. bool DataPipeConsumerDispatcher::EndSerialize(
  242. void* destination,
  243. ports::PortName* ports,
  244. PlatformHandle* platform_handles) {
  245. SerializedState* state = static_cast<SerializedState*>(destination);
  246. memcpy(&state->options, &options_, sizeof(MojoCreateDataPipeOptions));
  247. memset(state->padding, 0, sizeof(state->padding));
  248. base::AutoLock lock(lock_);
  249. DCHECK(in_transit_);
  250. state->pipe_id = pipe_id_;
  251. state->read_offset = read_offset_;
  252. state->bytes_available = bytes_available_;
  253. state->flags = peer_closed_ ? kFlagPeerClosed : 0;
  254. auto region_handle =
  255. base::UnsafeSharedMemoryRegion::TakeHandleForSerialization(
  256. std::move(shared_ring_buffer_));
  257. const base::UnguessableToken& guid = region_handle.GetGUID();
  258. state->buffer_guid_high = guid.GetHighForSerialization();
  259. state->buffer_guid_low = guid.GetLowForSerialization();
  260. ports[0] = control_port_.name();
  261. PlatformHandle handle;
  262. PlatformHandle ignored_handle;
  263. ExtractPlatformHandlesFromSharedMemoryRegionHandle(
  264. region_handle.PassPlatformHandle(), &handle, &ignored_handle);
  265. if (!handle.is_valid() || ignored_handle.is_valid())
  266. return false;
  267. platform_handles[0] = std::move(handle);
  268. return true;
  269. }
  270. bool DataPipeConsumerDispatcher::BeginTransit() {
  271. base::AutoLock lock(lock_);
  272. if (in_transit_)
  273. return false;
  274. in_transit_ = !in_two_phase_read_;
  275. return in_transit_;
  276. }
  277. void DataPipeConsumerDispatcher::CompleteTransitAndClose() {
  278. node_controller_->SetPortObserver(control_port_, nullptr);
  279. base::AutoLock lock(lock_);
  280. DCHECK(in_transit_);
  281. in_transit_ = false;
  282. transferred_ = true;
  283. CloseNoLock();
  284. }
  285. void DataPipeConsumerDispatcher::CancelTransit() {
  286. base::AutoLock lock(lock_);
  287. DCHECK(in_transit_);
  288. in_transit_ = false;
  289. UpdateSignalsStateNoLock();
  290. }
  291. // static
  292. scoped_refptr<DataPipeConsumerDispatcher>
  293. DataPipeConsumerDispatcher::Deserialize(const void* data,
  294. size_t num_bytes,
  295. const ports::PortName* ports,
  296. size_t num_ports,
  297. PlatformHandle* handles,
  298. size_t num_handles) {
  299. if (num_ports != 1 || num_handles != 1 ||
  300. num_bytes != sizeof(SerializedState)) {
  301. AssertNotExtractingHandlesFromMessage();
  302. return nullptr;
  303. }
  304. const SerializedState* state = static_cast<const SerializedState*>(data);
  305. if (!state->options.capacity_num_bytes || !state->options.element_num_bytes ||
  306. state->options.capacity_num_bytes < state->options.element_num_bytes ||
  307. state->read_offset >= state->options.capacity_num_bytes ||
  308. state->bytes_available > state->options.capacity_num_bytes) {
  309. AssertNotExtractingHandlesFromMessage();
  310. return nullptr;
  311. }
  312. NodeController* node_controller = Core::Get()->GetNodeController();
  313. ports::PortRef port;
  314. if (node_controller->node()->GetPort(ports[0], &port) != ports::OK) {
  315. AssertNotExtractingHandlesFromMessage();
  316. return nullptr;
  317. }
  318. auto region_handle = CreateSharedMemoryRegionHandleFromPlatformHandles(
  319. std::move(handles[0]), PlatformHandle());
  320. auto region = base::subtle::PlatformSharedMemoryRegion::Take(
  321. std::move(region_handle),
  322. base::subtle::PlatformSharedMemoryRegion::Mode::kUnsafe,
  323. state->options.capacity_num_bytes,
  324. base::UnguessableToken::Deserialize(state->buffer_guid_high,
  325. state->buffer_guid_low));
  326. auto ring_buffer =
  327. base::UnsafeSharedMemoryRegion::Deserialize(std::move(region));
  328. if (!ring_buffer.IsValid()) {
  329. DLOG(ERROR) << "Failed to deserialize shared buffer handle.";
  330. AssertNotExtractingHandlesFromMessage();
  331. return nullptr;
  332. }
  333. scoped_refptr<DataPipeConsumerDispatcher> dispatcher =
  334. new DataPipeConsumerDispatcher(node_controller, port,
  335. std::move(ring_buffer), state->options,
  336. state->pipe_id);
  337. {
  338. base::AutoLock lock(dispatcher->lock_);
  339. dispatcher->read_offset_ = state->read_offset;
  340. dispatcher->bytes_available_ = state->bytes_available;
  341. dispatcher->new_data_available_ = state->bytes_available > 0;
  342. dispatcher->peer_closed_ = state->flags & kFlagPeerClosed;
  343. if (!dispatcher->InitializeNoLock()) {
  344. AssertNotExtractingHandlesFromMessage();
  345. return nullptr;
  346. }
  347. if (state->options.capacity_num_bytes >
  348. dispatcher->ring_buffer_mapping_.mapped_size()) {
  349. AssertNotExtractingHandlesFromMessage();
  350. return nullptr;
  351. }
  352. dispatcher->UpdateSignalsStateNoLock();
  353. }
  354. return dispatcher;
  355. }
  356. DataPipeConsumerDispatcher::DataPipeConsumerDispatcher(
  357. NodeController* node_controller,
  358. const ports::PortRef& control_port,
  359. base::UnsafeSharedMemoryRegion shared_ring_buffer,
  360. const MojoCreateDataPipeOptions& options,
  361. uint64_t pipe_id)
  362. : options_(options),
  363. node_controller_(node_controller),
  364. control_port_(control_port),
  365. pipe_id_(pipe_id),
  366. watchers_(this),
  367. shared_ring_buffer_(std::move(shared_ring_buffer)) {}
  368. DataPipeConsumerDispatcher::~DataPipeConsumerDispatcher() {
  369. DCHECK(is_closed_ && !shared_ring_buffer_.IsValid() &&
  370. !ring_buffer_mapping_.IsValid() && !in_transit_);
  371. }
  372. bool DataPipeConsumerDispatcher::InitializeNoLock() {
  373. lock_.AssertAcquired();
  374. if (!shared_ring_buffer_.IsValid())
  375. return false;
  376. DCHECK(!ring_buffer_mapping_.IsValid());
  377. ring_buffer_mapping_ = shared_ring_buffer_.Map();
  378. if (!ring_buffer_mapping_.IsValid()) {
  379. DLOG(ERROR) << "Failed to map shared buffer.";
  380. shared_ring_buffer_ = base::UnsafeSharedMemoryRegion();
  381. is_closed_ = true;
  382. return false;
  383. }
  384. base::AutoUnlock unlock(lock_);
  385. node_controller_->SetPortObserver(
  386. control_port_, base::MakeRefCounted<PortObserverThunk>(this));
  387. return true;
  388. }
  389. MojoResult DataPipeConsumerDispatcher::CloseNoLock() {
  390. lock_.AssertAcquired();
  391. if (is_closed_ || in_transit_)
  392. return MOJO_RESULT_INVALID_ARGUMENT;
  393. is_closed_ = true;
  394. ring_buffer_mapping_ = base::WritableSharedMemoryMapping();
  395. shared_ring_buffer_ = base::UnsafeSharedMemoryRegion();
  396. watchers_.NotifyClosed();
  397. if (!transferred_) {
  398. base::AutoUnlock unlock(lock_);
  399. node_controller_->ClosePort(control_port_);
  400. }
  401. return MOJO_RESULT_OK;
  402. }
  403. HandleSignalsState DataPipeConsumerDispatcher::GetHandleSignalsStateNoLock()
  404. const {
  405. lock_.AssertAcquired();
  406. HandleSignalsState rv;
  407. if (shared_ring_buffer_.IsValid() && bytes_available_) {
  408. if (!in_two_phase_read_) {
  409. rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_READABLE;
  410. if (new_data_available_)
  411. rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE;
  412. }
  413. rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
  414. } else if (!peer_closed_ && shared_ring_buffer_.IsValid()) {
  415. rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_READABLE;
  416. }
  417. if (shared_ring_buffer_.IsValid()) {
  418. if (new_data_available_ || !peer_closed_)
  419. rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_NEW_DATA_READABLE;
  420. }
  421. if (peer_closed_) {
  422. rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
  423. } else {
  424. rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_REMOTE;
  425. if (peer_remote_)
  426. rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_REMOTE;
  427. }
  428. rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
  429. return rv;
  430. }
  431. void DataPipeConsumerDispatcher::NotifyRead(uint32_t num_bytes) {
  432. DVLOG(1) << "Data pipe consumer " << pipe_id_
  433. << " notifying peer: " << num_bytes
  434. << " bytes read. [control_port=" << control_port_.name() << "]";
  435. SendDataPipeControlMessage(node_controller_, control_port_,
  436. DataPipeCommand::DATA_WAS_READ, num_bytes);
  437. }
  438. void DataPipeConsumerDispatcher::OnPortStatusChanged() {
  439. DCHECK(RequestContext::current());
  440. base::AutoLock lock(lock_);
  441. // We stop observing the control port as soon it's transferred, but this can
  442. // race with events which are raised right before that happens. This is fine
  443. // to ignore.
  444. if (transferred_)
  445. return;
  446. DVLOG(1) << "Control port status changed for data pipe producer " << pipe_id_;
  447. UpdateSignalsStateNoLock();
  448. }
  449. void DataPipeConsumerDispatcher::UpdateSignalsStateNoLock() {
  450. lock_.AssertAcquired();
  451. const bool was_peer_closed = peer_closed_;
  452. const bool was_peer_remote = peer_remote_;
  453. size_t previous_bytes_available = bytes_available_;
  454. ports::PortStatus port_status;
  455. int rv = node_controller_->node()->GetStatus(control_port_, &port_status);
  456. peer_remote_ = rv == ports::OK && port_status.peer_remote;
  457. if (rv != ports::OK || !port_status.receiving_messages) {
  458. DVLOG(1) << "Data pipe consumer " << pipe_id_ << " is aware of peer closure"
  459. << " [control_port=" << control_port_.name() << "]";
  460. peer_closed_ = true;
  461. } else if (rv == ports::OK && port_status.has_messages && !in_transit_) {
  462. std::unique_ptr<ports::UserMessageEvent> message_event;
  463. do {
  464. rv = node_controller_->node()->GetMessage(control_port_, &message_event,
  465. nullptr);
  466. if (rv != ports::OK)
  467. peer_closed_ = true;
  468. if (message_event) {
  469. auto* message = message_event->GetMessage<UserMessageImpl>();
  470. if (message->user_payload_size() < sizeof(DataPipeControlMessage)) {
  471. peer_closed_ = true;
  472. break;
  473. }
  474. const DataPipeControlMessage* m =
  475. static_cast<const DataPipeControlMessage*>(message->user_payload());
  476. if (m->command != DataPipeCommand::DATA_WAS_WRITTEN) {
  477. DLOG(ERROR) << "Unexpected control message from producer.";
  478. peer_closed_ = true;
  479. break;
  480. }
  481. TRACE_EVENT0("ipc",
  482. "DataPipeConsumerDispatcher received DATA_WAS_WRITTEN");
  483. uint32_t new_bytes_available;
  484. if (!base::CheckAdd(bytes_available_, m->num_bytes)
  485. .AssignIfValid(&new_bytes_available) ||
  486. new_bytes_available > options_.capacity_num_bytes) {
  487. DLOG(ERROR) << "Producer claims to have written too many bytes.";
  488. peer_closed_ = true;
  489. break;
  490. }
  491. DVLOG(1) << "Data pipe consumer " << pipe_id_ << " is aware that "
  492. << m->num_bytes << " bytes were written. [control_port="
  493. << control_port_.name() << "]";
  494. bytes_available_ = new_bytes_available;
  495. }
  496. } while (message_event);
  497. }
  498. bool has_new_data = bytes_available_ != previous_bytes_available;
  499. if (has_new_data)
  500. new_data_available_ = true;
  501. if (peer_closed_ != was_peer_closed || has_new_data ||
  502. peer_remote_ != was_peer_remote) {
  503. watchers_.NotifyState(GetHandleSignalsStateNoLock());
  504. }
  505. }
  506. } // namespace core
  507. } // namespace mojo