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 #ifndef MOJO_EDK_SYSTEM_DATA_PIPE_CONSUMER_DISPATCHER_H_ 6 #define MOJO_EDK_SYSTEM_DATA_PIPE_CONSUMER_DISPATCHER_H_ 7 8 #include <stddef.h> 9 #include <stdint.h> 10 11 #include <memory> 12 13 #include "base/macros.h" 14 #include "base/memory/ref_counted.h" 15 #include "base/synchronization/lock.h" 16 #include "mojo/edk/embedder/platform_handle_vector.h" 17 #include "mojo/edk/embedder/platform_shared_buffer.h" 18 #include "mojo/edk/embedder/scoped_platform_handle.h" 19 #include "mojo/edk/system/awakable_list.h" 20 #include "mojo/edk/system/dispatcher.h" 21 #include "mojo/edk/system/ports/port_ref.h" 22 #include "mojo/edk/system/system_impl_export.h" 23 24 namespace mojo { 25 namespace edk { 26 27 struct DataPipeControlMessage; 28 class NodeController; 29 30 // This is the Dispatcher implementation for the consumer handle for data 31 // pipes created by the Mojo primitive MojoCreateDataPipe(). This class is 32 // thread-safe. 33 class MOJO_SYSTEM_IMPL_EXPORT DataPipeConsumerDispatcher final 34 : public Dispatcher { 35 public: 36 DataPipeConsumerDispatcher( 37 NodeController* node_controller, 38 const ports::PortRef& control_port, 39 scoped_refptr<PlatformSharedBuffer> shared_ring_buffer, 40 const MojoCreateDataPipeOptions& options, 41 bool initialized, 42 uint64_t pipe_id); 43 44 // Dispatcher: 45 Type GetType() const override; 46 MojoResult Close() override; 47 MojoResult Watch(MojoHandleSignals signals, 48 const Watcher::WatchCallback& callback, 49 uintptr_t context) override; 50 MojoResult CancelWatch(uintptr_t context) override; 51 MojoResult ReadData(void* elements, 52 uint32_t* num_bytes, 53 MojoReadDataFlags flags) override; 54 MojoResult BeginReadData(const void** buffer, 55 uint32_t* buffer_num_bytes, 56 MojoReadDataFlags flags) override; 57 MojoResult EndReadData(uint32_t num_bytes_read) override; 58 HandleSignalsState GetHandleSignalsState() const override; 59 MojoResult AddAwakable(Awakable* awakable, 60 MojoHandleSignals signals, 61 uintptr_t context, 62 HandleSignalsState* signals_state) override; 63 void RemoveAwakable(Awakable* awakable, 64 HandleSignalsState* signals_state) override; 65 void StartSerialize(uint32_t* num_bytes, 66 uint32_t* num_ports, 67 uint32_t* num_handles) override; 68 bool EndSerialize(void* destination, 69 ports::PortName* ports, 70 PlatformHandle* handles) override; 71 bool BeginTransit() override; 72 void CompleteTransitAndClose() override; 73 void CancelTransit() override; 74 75 static scoped_refptr<DataPipeConsumerDispatcher> 76 Deserialize(const void* data, 77 size_t num_bytes, 78 const ports::PortName* ports, 79 size_t num_ports, 80 PlatformHandle* handles, 81 size_t num_handles); 82 83 private: 84 class PortObserverThunk; 85 friend class PortObserverThunk; 86 87 ~DataPipeConsumerDispatcher() override; 88 89 void InitializeNoLock(); 90 MojoResult CloseNoLock(); 91 HandleSignalsState GetHandleSignalsStateNoLock() const; 92 void NotifyRead(uint32_t num_bytes); 93 void OnPortStatusChanged(); 94 void UpdateSignalsStateNoLock(); 95 96 const MojoCreateDataPipeOptions options_; 97 NodeController* const node_controller_; 98 const ports::PortRef control_port_; 99 const uint64_t pipe_id_; 100 101 // Guards access to the fields below. 102 mutable base::Lock lock_; 103 104 AwakableList awakable_list_; 105 106 scoped_refptr<PlatformSharedBuffer> shared_ring_buffer_; 107 std::unique_ptr<PlatformSharedBufferMapping> ring_buffer_mapping_; 108 ScopedPlatformHandle buffer_handle_for_transit_; 109 110 bool in_two_phase_read_ = false; 111 uint32_t two_phase_max_bytes_read_ = 0; 112 113 bool in_transit_ = false; 114 bool is_closed_ = false; 115 bool peer_closed_ = false; 116 bool transferred_ = false; 117 118 uint32_t read_offset_ = 0; 119 uint32_t bytes_available_ = 0; 120 121 DISALLOW_COPY_AND_ASSIGN(DataPipeConsumerDispatcher); 122 }; 123 124 } // namespace edk 125 } // namespace mojo 126 127 #endif // MOJO_EDK_SYSTEM_DATA_PIPE_CONSUMER_DISPATCHER_H_ 128