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
5 #include "mojo/edk/system/data_pipe_producer_dispatcher.h"
6
7 #include <stddef.h>
8 #include <stdint.h>
9
10 #include <utility>
11
12 #include "base/bind.h"
13 #include "base/logging.h"
14 #include "base/memory/ref_counted.h"
15 #include "base/message_loop/message_loop.h"
16 #include "mojo/edk/embedder/embedder_internal.h"
17 #include "mojo/edk/embedder/platform_shared_buffer.h"
18 #include "mojo/edk/system/configuration.h"
19 #include "mojo/edk/system/core.h"
20 #include "mojo/edk/system/data_pipe_control_message.h"
21 #include "mojo/edk/system/node_controller.h"
22 #include "mojo/edk/system/ports_message.h"
23 #include "mojo/edk/system/request_context.h"
24 #include "mojo/public/c/system/data_pipe.h"
25
26 namespace mojo {
27 namespace edk {
28
29 namespace {
30
31 const uint8_t kFlagPeerClosed = 0x01;
32
33 #pragma pack(push, 1)
34
35 struct SerializedState {
36 MojoCreateDataPipeOptions options;
37 uint64_t pipe_id;
38 uint32_t write_offset;
39 uint32_t available_capacity;
40 uint8_t flags;
41 char padding[7];
42 };
43
44 static_assert(sizeof(SerializedState) % 8 == 0,
45 "Invalid SerializedState size.");
46
47 #pragma pack(pop)
48
49 } // namespace
50
51 // A PortObserver which forwards to a DataPipeProducerDispatcher. This owns a
52 // reference to the dispatcher to ensure it lives as long as the observed port.
53 class DataPipeProducerDispatcher::PortObserverThunk
54 : public NodeController::PortObserver {
55 public:
PortObserverThunk(scoped_refptr<DataPipeProducerDispatcher> dispatcher)56 explicit PortObserverThunk(
57 scoped_refptr<DataPipeProducerDispatcher> dispatcher)
58 : dispatcher_(dispatcher) {}
59
60 private:
~PortObserverThunk()61 ~PortObserverThunk() override {}
62
63 // NodeController::PortObserver:
OnPortStatusChanged()64 void OnPortStatusChanged() override { dispatcher_->OnPortStatusChanged(); }
65
66 scoped_refptr<DataPipeProducerDispatcher> dispatcher_;
67
68 DISALLOW_COPY_AND_ASSIGN(PortObserverThunk);
69 };
70
DataPipeProducerDispatcher(NodeController * node_controller,const ports::PortRef & control_port,scoped_refptr<PlatformSharedBuffer> shared_ring_buffer,const MojoCreateDataPipeOptions & options,bool initialized,uint64_t pipe_id)71 DataPipeProducerDispatcher::DataPipeProducerDispatcher(
72 NodeController* node_controller,
73 const ports::PortRef& control_port,
74 scoped_refptr<PlatformSharedBuffer> shared_ring_buffer,
75 const MojoCreateDataPipeOptions& options,
76 bool initialized,
77 uint64_t pipe_id)
78 : options_(options),
79 node_controller_(node_controller),
80 control_port_(control_port),
81 pipe_id_(pipe_id),
82 shared_ring_buffer_(shared_ring_buffer),
83 available_capacity_(options_.capacity_num_bytes) {
84 if (initialized) {
85 base::AutoLock lock(lock_);
86 InitializeNoLock();
87 }
88 }
89
GetType() const90 Dispatcher::Type DataPipeProducerDispatcher::GetType() const {
91 return Type::DATA_PIPE_PRODUCER;
92 }
93
Close()94 MojoResult DataPipeProducerDispatcher::Close() {
95 base::AutoLock lock(lock_);
96 DVLOG(1) << "Closing data pipe producer " << pipe_id_;
97 return CloseNoLock();
98 }
99
Watch(MojoHandleSignals signals,const Watcher::WatchCallback & callback,uintptr_t context)100 MojoResult DataPipeProducerDispatcher::Watch(
101 MojoHandleSignals signals,
102 const Watcher::WatchCallback& callback,
103 uintptr_t context) {
104 base::AutoLock lock(lock_);
105
106 if (is_closed_ || in_transit_)
107 return MOJO_RESULT_INVALID_ARGUMENT;
108
109 return awakable_list_.AddWatcher(
110 signals, callback, context, GetHandleSignalsStateNoLock());
111 }
112
CancelWatch(uintptr_t context)113 MojoResult DataPipeProducerDispatcher::CancelWatch(uintptr_t context) {
114 base::AutoLock lock(lock_);
115
116 if (is_closed_ || in_transit_)
117 return MOJO_RESULT_INVALID_ARGUMENT;
118
119 return awakable_list_.RemoveWatcher(context);
120 }
121
WriteData(const void * elements,uint32_t * num_bytes,MojoWriteDataFlags flags)122 MojoResult DataPipeProducerDispatcher::WriteData(const void* elements,
123 uint32_t* num_bytes,
124 MojoWriteDataFlags flags) {
125 base::AutoLock lock(lock_);
126 if (!shared_ring_buffer_ || in_transit_)
127 return MOJO_RESULT_INVALID_ARGUMENT;
128
129 if (in_two_phase_write_)
130 return MOJO_RESULT_BUSY;
131
132 if (peer_closed_)
133 return MOJO_RESULT_FAILED_PRECONDITION;
134
135 if (*num_bytes % options_.element_num_bytes != 0)
136 return MOJO_RESULT_INVALID_ARGUMENT;
137 if (*num_bytes == 0)
138 return MOJO_RESULT_OK; // Nothing to do.
139
140 if ((flags & MOJO_WRITE_DATA_FLAG_ALL_OR_NONE) &&
141 (*num_bytes > available_capacity_)) {
142 // Don't return "should wait" since you can't wait for a specified amount of
143 // data.
144 return MOJO_RESULT_OUT_OF_RANGE;
145 }
146
147 DCHECK_LE(available_capacity_, options_.capacity_num_bytes);
148 uint32_t num_bytes_to_write = std::min(*num_bytes, available_capacity_);
149 if (num_bytes_to_write == 0)
150 return MOJO_RESULT_SHOULD_WAIT;
151
152 HandleSignalsState old_state = GetHandleSignalsStateNoLock();
153
154 *num_bytes = num_bytes_to_write;
155
156 CHECK(ring_buffer_mapping_);
157 uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
158 CHECK(data);
159
160 const uint8_t* source = static_cast<const uint8_t*>(elements);
161 CHECK(source);
162
163 DCHECK_LE(write_offset_, options_.capacity_num_bytes);
164 uint32_t tail_bytes_to_write =
165 std::min(options_.capacity_num_bytes - write_offset_,
166 num_bytes_to_write);
167 uint32_t head_bytes_to_write = num_bytes_to_write - tail_bytes_to_write;
168
169 DCHECK_GT(tail_bytes_to_write, 0u);
170 memcpy(data + write_offset_, source, tail_bytes_to_write);
171 if (head_bytes_to_write > 0)
172 memcpy(data, source + tail_bytes_to_write, head_bytes_to_write);
173
174 DCHECK_LE(num_bytes_to_write, available_capacity_);
175 available_capacity_ -= num_bytes_to_write;
176 write_offset_ = (write_offset_ + num_bytes_to_write) %
177 options_.capacity_num_bytes;
178
179 HandleSignalsState new_state = GetHandleSignalsStateNoLock();
180 if (!new_state.equals(old_state))
181 awakable_list_.AwakeForStateChange(new_state);
182
183 base::AutoUnlock unlock(lock_);
184 NotifyWrite(num_bytes_to_write);
185
186 return MOJO_RESULT_OK;
187 }
188
BeginWriteData(void ** buffer,uint32_t * buffer_num_bytes,MojoWriteDataFlags flags)189 MojoResult DataPipeProducerDispatcher::BeginWriteData(
190 void** buffer,
191 uint32_t* buffer_num_bytes,
192 MojoWriteDataFlags flags) {
193 base::AutoLock lock(lock_);
194 if (!shared_ring_buffer_ || in_transit_)
195 return MOJO_RESULT_INVALID_ARGUMENT;
196 if (in_two_phase_write_)
197 return MOJO_RESULT_BUSY;
198 if (peer_closed_)
199 return MOJO_RESULT_FAILED_PRECONDITION;
200
201 if (available_capacity_ == 0) {
202 return peer_closed_ ? MOJO_RESULT_FAILED_PRECONDITION
203 : MOJO_RESULT_SHOULD_WAIT;
204 }
205
206 in_two_phase_write_ = true;
207 *buffer_num_bytes = std::min(options_.capacity_num_bytes - write_offset_,
208 available_capacity_);
209 DCHECK_GT(*buffer_num_bytes, 0u);
210
211 CHECK(ring_buffer_mapping_);
212 uint8_t* data = static_cast<uint8_t*>(ring_buffer_mapping_->GetBase());
213 *buffer = data + write_offset_;
214
215 return MOJO_RESULT_OK;
216 }
217
EndWriteData(uint32_t num_bytes_written)218 MojoResult DataPipeProducerDispatcher::EndWriteData(
219 uint32_t num_bytes_written) {
220 base::AutoLock lock(lock_);
221 if (is_closed_ || in_transit_)
222 return MOJO_RESULT_INVALID_ARGUMENT;
223
224 if (!in_two_phase_write_)
225 return MOJO_RESULT_FAILED_PRECONDITION;
226
227 DCHECK(shared_ring_buffer_);
228 DCHECK(ring_buffer_mapping_);
229
230 // Note: Allow successful completion of the two-phase write even if the other
231 // side has been closed.
232 MojoResult rv = MOJO_RESULT_OK;
233 if (num_bytes_written > available_capacity_ ||
234 num_bytes_written % options_.element_num_bytes != 0 ||
235 write_offset_ + num_bytes_written > options_.capacity_num_bytes) {
236 rv = MOJO_RESULT_INVALID_ARGUMENT;
237 } else {
238 DCHECK_LE(num_bytes_written + write_offset_, options_.capacity_num_bytes);
239 available_capacity_ -= num_bytes_written;
240 write_offset_ = (write_offset_ + num_bytes_written) %
241 options_.capacity_num_bytes;
242
243 base::AutoUnlock unlock(lock_);
244 NotifyWrite(num_bytes_written);
245 }
246
247 in_two_phase_write_ = false;
248
249 // If we're now writable, we *became* writable (since we weren't writable
250 // during the two-phase write), so awake producer awakables.
251 HandleSignalsState new_state = GetHandleSignalsStateNoLock();
252 if (new_state.satisfies(MOJO_HANDLE_SIGNAL_WRITABLE))
253 awakable_list_.AwakeForStateChange(new_state);
254
255 return rv;
256 }
257
GetHandleSignalsState() const258 HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsState() const {
259 base::AutoLock lock(lock_);
260 return GetHandleSignalsStateNoLock();
261 }
262
AddAwakable(Awakable * awakable,MojoHandleSignals signals,uintptr_t context,HandleSignalsState * signals_state)263 MojoResult DataPipeProducerDispatcher::AddAwakable(
264 Awakable* awakable,
265 MojoHandleSignals signals,
266 uintptr_t context,
267 HandleSignalsState* signals_state) {
268 base::AutoLock lock(lock_);
269 if (!shared_ring_buffer_ || in_transit_) {
270 if (signals_state)
271 *signals_state = HandleSignalsState();
272 return MOJO_RESULT_INVALID_ARGUMENT;
273 }
274 UpdateSignalsStateNoLock();
275 HandleSignalsState state = GetHandleSignalsStateNoLock();
276 if (state.satisfies(signals)) {
277 if (signals_state)
278 *signals_state = state;
279 return MOJO_RESULT_ALREADY_EXISTS;
280 }
281 if (!state.can_satisfy(signals)) {
282 if (signals_state)
283 *signals_state = state;
284 return MOJO_RESULT_FAILED_PRECONDITION;
285 }
286
287 awakable_list_.Add(awakable, signals, context);
288 return MOJO_RESULT_OK;
289 }
290
RemoveAwakable(Awakable * awakable,HandleSignalsState * signals_state)291 void DataPipeProducerDispatcher::RemoveAwakable(
292 Awakable* awakable,
293 HandleSignalsState* signals_state) {
294 base::AutoLock lock(lock_);
295 if ((!shared_ring_buffer_ || in_transit_) && signals_state)
296 *signals_state = HandleSignalsState();
297 else if (signals_state)
298 *signals_state = GetHandleSignalsStateNoLock();
299 awakable_list_.Remove(awakable);
300 }
301
StartSerialize(uint32_t * num_bytes,uint32_t * num_ports,uint32_t * num_handles)302 void DataPipeProducerDispatcher::StartSerialize(uint32_t* num_bytes,
303 uint32_t* num_ports,
304 uint32_t* num_handles) {
305 base::AutoLock lock(lock_);
306 DCHECK(in_transit_);
307 *num_bytes = sizeof(SerializedState);
308 *num_ports = 1;
309 *num_handles = 1;
310 }
311
EndSerialize(void * destination,ports::PortName * ports,PlatformHandle * platform_handles)312 bool DataPipeProducerDispatcher::EndSerialize(
313 void* destination,
314 ports::PortName* ports,
315 PlatformHandle* platform_handles) {
316 SerializedState* state = static_cast<SerializedState*>(destination);
317 memcpy(&state->options, &options_, sizeof(MojoCreateDataPipeOptions));
318 memset(state->padding, 0, sizeof(state->padding));
319
320 base::AutoLock lock(lock_);
321 DCHECK(in_transit_);
322 state->pipe_id = pipe_id_;
323 state->write_offset = write_offset_;
324 state->available_capacity = available_capacity_;
325 state->flags = peer_closed_ ? kFlagPeerClosed : 0;
326
327 ports[0] = control_port_.name();
328
329 buffer_handle_for_transit_ = shared_ring_buffer_->DuplicatePlatformHandle();
330 platform_handles[0] = buffer_handle_for_transit_.get();
331
332 return true;
333 }
334
BeginTransit()335 bool DataPipeProducerDispatcher::BeginTransit() {
336 base::AutoLock lock(lock_);
337 if (in_transit_)
338 return false;
339 in_transit_ = !in_two_phase_write_;
340 return in_transit_;
341 }
342
CompleteTransitAndClose()343 void DataPipeProducerDispatcher::CompleteTransitAndClose() {
344 node_controller_->SetPortObserver(control_port_, nullptr);
345
346 base::AutoLock lock(lock_);
347 DCHECK(in_transit_);
348 transferred_ = true;
349 in_transit_ = false;
350 ignore_result(buffer_handle_for_transit_.release());
351 CloseNoLock();
352 }
353
CancelTransit()354 void DataPipeProducerDispatcher::CancelTransit() {
355 base::AutoLock lock(lock_);
356 DCHECK(in_transit_);
357 in_transit_ = false;
358 buffer_handle_for_transit_.reset();
359 awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
360 }
361
362 // static
363 scoped_refptr<DataPipeProducerDispatcher>
Deserialize(const void * data,size_t num_bytes,const ports::PortName * ports,size_t num_ports,PlatformHandle * handles,size_t num_handles)364 DataPipeProducerDispatcher::Deserialize(const void* data,
365 size_t num_bytes,
366 const ports::PortName* ports,
367 size_t num_ports,
368 PlatformHandle* handles,
369 size_t num_handles) {
370 if (num_ports != 1 || num_handles != 1 ||
371 num_bytes != sizeof(SerializedState)) {
372 return nullptr;
373 }
374
375 const SerializedState* state = static_cast<const SerializedState*>(data);
376
377 NodeController* node_controller = internal::g_core->GetNodeController();
378 ports::PortRef port;
379 if (node_controller->node()->GetPort(ports[0], &port) != ports::OK)
380 return nullptr;
381
382 PlatformHandle buffer_handle;
383 std::swap(buffer_handle, handles[0]);
384 scoped_refptr<PlatformSharedBuffer> ring_buffer =
385 PlatformSharedBuffer::CreateFromPlatformHandle(
386 state->options.capacity_num_bytes,
387 false /* read_only */,
388 ScopedPlatformHandle(buffer_handle));
389 if (!ring_buffer) {
390 DLOG(ERROR) << "Failed to deserialize shared buffer handle.";
391 return nullptr;
392 }
393
394 scoped_refptr<DataPipeProducerDispatcher> dispatcher =
395 new DataPipeProducerDispatcher(node_controller, port, ring_buffer,
396 state->options, false /* initialized */,
397 state->pipe_id);
398
399 {
400 base::AutoLock lock(dispatcher->lock_);
401 dispatcher->write_offset_ = state->write_offset;
402 dispatcher->available_capacity_ = state->available_capacity;
403 dispatcher->peer_closed_ = state->flags & kFlagPeerClosed;
404 dispatcher->InitializeNoLock();
405 dispatcher->UpdateSignalsStateNoLock();
406 }
407
408 return dispatcher;
409 }
410
~DataPipeProducerDispatcher()411 DataPipeProducerDispatcher::~DataPipeProducerDispatcher() {
412 DCHECK(is_closed_ && !in_transit_ && !shared_ring_buffer_ &&
413 !ring_buffer_mapping_);
414 }
415
InitializeNoLock()416 void DataPipeProducerDispatcher::InitializeNoLock() {
417 lock_.AssertAcquired();
418
419 if (shared_ring_buffer_) {
420 ring_buffer_mapping_ =
421 shared_ring_buffer_->Map(0, options_.capacity_num_bytes);
422 if (!ring_buffer_mapping_) {
423 DLOG(ERROR) << "Failed to map shared buffer.";
424 shared_ring_buffer_ = nullptr;
425 }
426 }
427
428 base::AutoUnlock unlock(lock_);
429 node_controller_->SetPortObserver(
430 control_port_,
431 make_scoped_refptr(new PortObserverThunk(this)));
432 }
433
CloseNoLock()434 MojoResult DataPipeProducerDispatcher::CloseNoLock() {
435 lock_.AssertAcquired();
436 if (is_closed_ || in_transit_)
437 return MOJO_RESULT_INVALID_ARGUMENT;
438 is_closed_ = true;
439 ring_buffer_mapping_.reset();
440 shared_ring_buffer_ = nullptr;
441
442 awakable_list_.CancelAll();
443 if (!transferred_) {
444 base::AutoUnlock unlock(lock_);
445 node_controller_->ClosePort(control_port_);
446 }
447
448 return MOJO_RESULT_OK;
449 }
450
GetHandleSignalsStateNoLock() const451 HandleSignalsState DataPipeProducerDispatcher::GetHandleSignalsStateNoLock()
452 const {
453 lock_.AssertAcquired();
454 HandleSignalsState rv;
455 if (!peer_closed_) {
456 if (!in_two_phase_write_ && shared_ring_buffer_ &&
457 available_capacity_ > 0)
458 rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
459 rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_WRITABLE;
460 } else {
461 rv.satisfied_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
462 }
463 rv.satisfiable_signals |= MOJO_HANDLE_SIGNAL_PEER_CLOSED;
464 return rv;
465 }
466
NotifyWrite(uint32_t num_bytes)467 void DataPipeProducerDispatcher::NotifyWrite(uint32_t num_bytes) {
468 DVLOG(1) << "Data pipe producer " << pipe_id_ << " notifying peer: "
469 << num_bytes << " bytes written. [control_port="
470 << control_port_.name() << "]";
471
472 SendDataPipeControlMessage(node_controller_, control_port_,
473 DataPipeCommand::DATA_WAS_WRITTEN, num_bytes);
474 }
475
OnPortStatusChanged()476 void DataPipeProducerDispatcher::OnPortStatusChanged() {
477 DCHECK(RequestContext::current());
478
479 base::AutoLock lock(lock_);
480
481 // We stop observing the control port as soon it's transferred, but this can
482 // race with events which are raised right before that happens. This is fine
483 // to ignore.
484 if (transferred_)
485 return;
486
487 DVLOG(1) << "Control port status changed for data pipe producer " << pipe_id_;
488
489 UpdateSignalsStateNoLock();
490 }
491
UpdateSignalsStateNoLock()492 void DataPipeProducerDispatcher::UpdateSignalsStateNoLock() {
493 lock_.AssertAcquired();
494
495 bool was_peer_closed = peer_closed_;
496 size_t previous_capacity = available_capacity_;
497
498 ports::PortStatus port_status;
499 int rv = node_controller_->node()->GetStatus(control_port_, &port_status);
500 if (rv != ports::OK || !port_status.receiving_messages) {
501 DVLOG(1) << "Data pipe producer " << pipe_id_ << " is aware of peer closure"
502 << " [control_port=" << control_port_.name() << "]";
503 peer_closed_ = true;
504 } else if (rv == ports::OK && port_status.has_messages && !in_transit_) {
505 ports::ScopedMessage message;
506 do {
507 int rv = node_controller_->node()->GetMessage(
508 control_port_, &message, nullptr);
509 if (rv != ports::OK)
510 peer_closed_ = true;
511 if (message) {
512 if (message->num_payload_bytes() < sizeof(DataPipeControlMessage)) {
513 peer_closed_ = true;
514 break;
515 }
516
517 const DataPipeControlMessage* m =
518 static_cast<const DataPipeControlMessage*>(
519 message->payload_bytes());
520
521 if (m->command != DataPipeCommand::DATA_WAS_READ) {
522 DLOG(ERROR) << "Unexpected message from consumer.";
523 peer_closed_ = true;
524 break;
525 }
526
527 if (static_cast<size_t>(available_capacity_) + m->num_bytes >
528 options_.capacity_num_bytes) {
529 DLOG(ERROR) << "Consumer claims to have read too many bytes.";
530 break;
531 }
532
533 DVLOG(1) << "Data pipe producer " << pipe_id_ << " is aware that "
534 << m->num_bytes << " bytes were read. [control_port="
535 << control_port_.name() << "]";
536
537 available_capacity_ += m->num_bytes;
538 }
539 } while (message);
540 }
541
542 if (peer_closed_ != was_peer_closed ||
543 available_capacity_ != previous_capacity) {
544 awakable_list_.AwakeForStateChange(GetHandleSignalsStateNoLock());
545 }
546 }
547
548 } // namespace edk
549 } // namespace mojo
550