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