node.h 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357
  1. // Copyright 2016 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. #ifndef MOJO_CORE_PORTS_NODE_H_
  5. #define MOJO_CORE_PORTS_NODE_H_
  6. #include <stddef.h>
  7. #include <stdint.h>
  8. #include <queue>
  9. #include <unordered_map>
  10. #include "base/component_export.h"
  11. #include "base/containers/flat_map.h"
  12. #include "base/memory/raw_ptr.h"
  13. #include "base/memory/ref_counted.h"
  14. #include "base/synchronization/lock.h"
  15. #include "mojo/core/ports/event.h"
  16. #include "mojo/core/ports/name.h"
  17. #include "mojo/core/ports/port.h"
  18. #include "mojo/core/ports/port_ref.h"
  19. #include "mojo/core/ports/user_data.h"
  20. namespace mojo {
  21. namespace core {
  22. namespace ports {
  23. enum : int {
  24. OK = 0,
  25. ERROR_PORT_UNKNOWN = -10,
  26. ERROR_PORT_EXISTS = -11,
  27. ERROR_PORT_STATE_UNEXPECTED = -12,
  28. ERROR_PORT_CANNOT_SEND_SELF = -13,
  29. ERROR_PORT_PEER_CLOSED = -14,
  30. ERROR_PORT_CANNOT_SEND_PEER = -15,
  31. ERROR_NOT_IMPLEMENTED = -100,
  32. };
  33. struct PortStatus {
  34. bool has_messages;
  35. bool receiving_messages;
  36. bool peer_closed;
  37. bool peer_remote;
  38. size_t queued_message_count;
  39. size_t queued_num_bytes;
  40. size_t unacknowledged_message_count;
  41. };
  42. struct PendingUpdatePreviousPeer {
  43. NodeName receiver;
  44. PortName port;
  45. PortName from_port;
  46. uint64_t sequence_num;
  47. NodeName new_prev_node;
  48. PortName new_prev_port;
  49. };
  50. class MessageFilter;
  51. class NodeDelegate;
  52. // A Node maintains a collection of Ports (see port.h) indexed by unique 128-bit
  53. // addresses (names), performing routing and processing of events among the
  54. // Ports within the Node and to or from other Nodes in the system. Typically
  55. // (and practically, in all uses today) there is a single Node per system
  56. // process. Thus a Node boundary effectively models a process boundary.
  57. //
  58. // New Ports can be created uninitialized using CreateUninitializedPort (and
  59. // later initialized using InitializePort), or created in a fully initialized
  60. // state using CreatePortPair(). Initialized ports have exactly one conjugate
  61. // port which is the ultimate receiver of any user messages sent by that port.
  62. // See SendUserMessage().
  63. //
  64. // In addition to routing user message events, various control events are used
  65. // by Nodes to coordinate Port behavior and lifetime within and across Nodes.
  66. // See Event documentation for description of different types of events used by
  67. // a Node to coordinate behavior.
  68. class COMPONENT_EXPORT(MOJO_CORE_PORTS) Node {
  69. public:
  70. enum class ShutdownPolicy {
  71. DONT_ALLOW_LOCAL_PORTS,
  72. ALLOW_LOCAL_PORTS,
  73. };
  74. // Does not take ownership of the delegate.
  75. Node(const NodeName& name, NodeDelegate* delegate);
  76. Node(const Node&) = delete;
  77. Node& operator=(const Node&) = delete;
  78. ~Node();
  79. // Returns true iff there are no open ports referring to another node or ports
  80. // in the process of being transferred from this node to another. If this
  81. // returns false, then to ensure clean shutdown, it is necessary to keep the
  82. // node alive and continue routing messages to it via AcceptMessage. This
  83. // method may be called again after AcceptMessage to check if the Node is now
  84. // ready to be destroyed.
  85. //
  86. // If |policy| is set to |ShutdownPolicy::ALLOW_LOCAL_PORTS|, this will return
  87. // |true| even if some ports remain alive, as long as none of them are proxies
  88. // to another node.
  89. bool CanShutdownCleanly(
  90. ShutdownPolicy policy = ShutdownPolicy::DONT_ALLOW_LOCAL_PORTS);
  91. // Lookup the named port.
  92. int GetPort(const PortName& port_name, PortRef* port_ref);
  93. // Creates a port on this node. Before the port can be used, it must be
  94. // initialized using InitializePort. This method is useful for bootstrapping
  95. // a connection between two nodes. Generally, ports are created using
  96. // CreatePortPair instead.
  97. int CreateUninitializedPort(PortRef* port_ref);
  98. // Initializes a newly created port.
  99. int InitializePort(const PortRef& port_ref,
  100. const NodeName& peer_node_name,
  101. const PortName& peer_port_name,
  102. const NodeName& prev_node_name,
  103. const PortName& prev_port_name);
  104. // Generates a new connected pair of ports bound to this node. These ports
  105. // are initialized and ready to go.
  106. int CreatePortPair(PortRef* port0_ref, PortRef* port1_ref);
  107. // User data associated with the port.
  108. int SetUserData(const PortRef& port_ref, scoped_refptr<UserData> user_data);
  109. int GetUserData(const PortRef& port_ref, scoped_refptr<UserData>* user_data);
  110. // Prevents further messages from being sent from this port or delivered to
  111. // this port. The port is removed, and the port's peer is notified of the
  112. // closure after it has consumed all pending messages.
  113. int ClosePort(const PortRef& port_ref);
  114. // Returns the current status of the port.
  115. int GetStatus(const PortRef& port_ref, PortStatus* port_status);
  116. // Returns the next available message on the specified port or returns a null
  117. // message if there are none available. Returns ERROR_PORT_PEER_CLOSED to
  118. // indicate that this port's peer has closed. In such cases GetMessage may
  119. // be called until it yields a null message, indicating that no more messages
  120. // may be read from the port.
  121. //
  122. // If |filter| is non-null, the next available message is returned only if it
  123. // is matched by the filter. If the provided filter does not match the next
  124. // available message, GetMessage() behaves as if there is no message
  125. // available. Ownership of |filter| is not taken, and it must outlive the
  126. // extent of this call.
  127. int GetMessage(const PortRef& port_ref,
  128. std::unique_ptr<UserMessageEvent>* message,
  129. MessageFilter* filter);
  130. // Sends a message from the specified port to its peer. Note that the message
  131. // notification may arrive synchronously (via PortStatusChanged() on the
  132. // delegate) if the peer is local to this Node.
  133. int SendUserMessage(const PortRef& port_ref,
  134. std::unique_ptr<UserMessageEvent> message);
  135. // Makes the port send acknowledge requests to its conjugate to acknowledge
  136. // at least every |sequence_number_acknowledge_interval| messages as they're
  137. // read from the conjugate. The number of unacknowledged messages is exposed
  138. // in the |unacknowledged_message_count| field of PortStatus. This allows
  139. // bounding the number of unread and/or in-transit messages from this port
  140. // to its conjugate between zero and |unacknowledged_message_count|.
  141. int SetAcknowledgeRequestInterval(
  142. const PortRef& port_ref,
  143. uint64_t sequence_number_acknowledge_interval);
  144. // Corresponding to NodeDelegate::ForwardEvent.
  145. int AcceptEvent(const NodeName& from_node, ScopedEvent event);
  146. // Called to merge two ports with each other. If you have two independent
  147. // port pairs A <=> B and C <=> D, the net result of merging B and C is a
  148. // single connected port pair A <=> D.
  149. //
  150. // Note that the behavior of this operation is undefined if either port to be
  151. // merged (B or C above) has ever been read from or written to directly, and
  152. // this must ONLY be called on one side of the merge, though it doesn't matter
  153. // which side.
  154. //
  155. // It is safe for the non-merged peers (A and D above) to be transferred,
  156. // closed, and/or written to before, during, or after the merge.
  157. int MergePorts(const PortRef& port_ref,
  158. const NodeName& destination_node_name,
  159. const PortName& destination_port_name);
  160. // Like above but merges two ports local to this node. Because both ports are
  161. // local this can also verify that neither port has been written to before the
  162. // merge. If this fails for any reason, both ports are closed. Otherwise OK
  163. // is returned and the ports' receiving peers are connected to each other.
  164. int MergeLocalPorts(const PortRef& port0_ref, const PortRef& port1_ref);
  165. // Called to inform this node that communication with another node is lost
  166. // indefinitely. This triggers cleanup of ports bound to this node.
  167. int LostConnectionToNode(const NodeName& node_name);
  168. private:
  169. // Helper to ensure that a Node always calls into its delegate safely, i.e.
  170. // without holding any internal locks.
  171. class DelegateHolder {
  172. public:
  173. DelegateHolder(Node* node, NodeDelegate* delegate);
  174. DelegateHolder(const DelegateHolder&) = delete;
  175. DelegateHolder& operator=(const DelegateHolder&) = delete;
  176. ~DelegateHolder();
  177. NodeDelegate* operator->() const {
  178. EnsureSafeDelegateAccess();
  179. return delegate_;
  180. }
  181. private:
  182. #if DCHECK_IS_ON()
  183. void EnsureSafeDelegateAccess() const;
  184. #else
  185. void EnsureSafeDelegateAccess() const {}
  186. #endif
  187. const raw_ptr<Node> node_;
  188. const raw_ptr<NodeDelegate> delegate_;
  189. };
  190. int OnUserMessage(const PortRef& port_ref,
  191. const NodeName& from_node,
  192. std::unique_ptr<UserMessageEvent> message);
  193. int OnPortAccepted(const PortRef& port_ref,
  194. std::unique_ptr<PortAcceptedEvent> event);
  195. int OnObserveProxy(const PortRef& port_ref,
  196. std::unique_ptr<ObserveProxyEvent> event);
  197. int OnObserveProxyAck(const PortRef& port_ref,
  198. std::unique_ptr<ObserveProxyAckEvent> event);
  199. int OnObserveClosure(const PortRef& port_ref,
  200. std::unique_ptr<ObserveClosureEvent> event);
  201. int OnMergePort(const PortRef& port_ref,
  202. std::unique_ptr<MergePortEvent> event);
  203. int OnUserMessageReadAckRequest(
  204. const PortRef& port_ref,
  205. std::unique_ptr<UserMessageReadAckRequestEvent> event);
  206. int OnUserMessageReadAck(const PortRef& port_ref,
  207. std::unique_ptr<UserMessageReadAckEvent> event);
  208. int OnUpdatePreviousPeer(const PortRef& port_ref,
  209. std::unique_ptr<UpdatePreviousPeerEvent> event);
  210. int AddPortWithName(const PortName& port_name, scoped_refptr<Port> port);
  211. void ErasePort(const PortName& port_name);
  212. // Check if the event is sent by the previous peer of the port to decide if
  213. // we can check the sequence number.
  214. // This is not the case for example for PortAccepted or broadcasted events.
  215. bool IsEventFromPreviousPeer(const Event& event);
  216. int AcceptEventInternal(const PortRef& port_ref,
  217. const NodeName& from_node,
  218. ScopedEvent event);
  219. int SendUserMessageInternal(const PortRef& port_ref,
  220. std::unique_ptr<UserMessageEvent>* message);
  221. int MergePortsInternal(const PortRef& port0_ref,
  222. const PortRef& port1_ref,
  223. bool allow_close_on_bad_state);
  224. void ConvertToProxy(Port* port,
  225. const NodeName& to_node_name,
  226. PortName* port_name,
  227. Event::PortDescriptor* port_descriptor,
  228. PendingUpdatePreviousPeer* pending_update);
  229. int AcceptPort(const PortName& port_name,
  230. const Event::PortDescriptor& port_descriptor);
  231. int PrepareToForwardUserMessage(const PortRef& forwarding_port_ref,
  232. Port::State expected_port_state,
  233. bool ignore_closed_peer,
  234. UserMessageEvent* message,
  235. NodeName* forward_to_node);
  236. int BeginProxying(const PortRef& port_ref);
  237. int ForwardUserMessagesFromProxy(const PortRef& port_ref);
  238. void InitiateProxyRemoval(const PortRef& port_ref);
  239. void TryRemoveProxy(const PortRef& port_ref);
  240. void DestroyAllPortsWithPeer(const NodeName& node_name,
  241. const PortName& port_name);
  242. // Changes the peer node and port name referenced by |port|. Note that both
  243. // |ports_lock_| MUST be held through the extent of this method.
  244. // |local_port|'s lock must be held if and only if a reference to |local_port|
  245. // exist in |ports_|.
  246. void UpdatePortPeerAddress(const PortName& local_port_name,
  247. Port* local_port,
  248. const NodeName& new_peer_node,
  249. const PortName& new_peer_port);
  250. // Removes an entry from |peer_port_map_| corresponding to |local_port|'s peer
  251. // address, if valid.
  252. void RemoveFromPeerPortMap(const PortName& local_port_name, Port* local_port);
  253. // Swaps the peer information for two local ports. Used during port merges.
  254. // Note that |ports_lock_| must be held along with each of the two port's own
  255. // locks, through the extent of this method.
  256. void SwapPortPeers(const PortName& port0_name,
  257. Port* port0,
  258. const PortName& port1_name,
  259. Port* port1);
  260. // Sends an acknowledge request to the peer if the port has a non-zero
  261. // |sequence_num_acknowledge_interval|. This needs to be done when the port's
  262. // peer changes, as the previous peer proxy may not have forwarded any prior
  263. // acknowledge request before deleting itself.
  264. void MaybeResendAckRequest(const PortRef& port_ref);
  265. // Forwards a stored acknowledge request to the peer if the proxy has a
  266. // non-zero |sequence_num_acknowledge_interval|.
  267. void MaybeForwardAckRequest(const PortRef& port_ref);
  268. // Sends an acknowledge of the most recently read sequence number to the peer
  269. // if any messages have been read, and the port has a non-zero
  270. // |sequence_num_to_acknowledge|.
  271. void MaybeResendAck(const PortRef& port_ref);
  272. const NodeName name_;
  273. const DelegateHolder delegate_;
  274. // Just to clarify readability of the types below.
  275. using LocalPortName = PortName;
  276. using PeerPortName = PortName;
  277. // Guards access to |ports_| and |peer_port_maps_| below.
  278. //
  279. // This must never be acquired while an individual port's lock is held on the
  280. // same thread. Conversely, individual port locks may be acquired while this
  281. // one is held.
  282. //
  283. // Because UserMessage events may execute arbitrary user code during
  284. // destruction, it is also important to ensure that such events are never
  285. // destroyed while this (or any individual Port) lock is held.
  286. base::Lock ports_lock_;
  287. std::unordered_map<LocalPortName, scoped_refptr<Port>> ports_;
  288. // Maps a peer port name to a list of PortRefs for all local ports which have
  289. // the port name key designated as their peer port. The set of local ports
  290. // which have the same peer port is expected to always be relatively small and
  291. // usually 1. Hence we just use a flat_map of local PortRefs keyed on each
  292. // local port's name.
  293. using PeerPortMap =
  294. std::unordered_map<PeerPortName, base::flat_map<LocalPortName, PortRef>>;
  295. // A reverse mapping which can be used to find all local ports that reference
  296. // a given peer node or a local port that references a specific given peer
  297. // port on a peer node. The key to this map is the corresponding peer node
  298. // name.
  299. std::unordered_map<NodeName, PeerPortMap> peer_port_maps_;
  300. };
  301. } // namespace ports
  302. } // namespace core
  303. } // namespace mojo
  304. #endif // MOJO_CORE_PORTS_NODE_H_