Changeset 187:da63d2c59a8a in finroc_plugins_tcp

Timestamp:

05.05.2020 00:10:53 (3 years ago)

Author:

Max Reichardt <mreichardt@…>

Branch:

17.03

Children:

188:054a5d87d050, 189:a6cc7cd79aef

Parents:

184:675dfcc4d494 (diff), 186:48a9973a8bde (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Phase:

public

Message:

Merge with 14.08

Files:

• internal/tPeerImplementation.h (modified) (1 diff)
• internal/tPeerImplementation.h (modified) (19 diffs)

internal/tPeerImplementation.h

@@ -42 +42 @@
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/deadline_timer.hpp>
+#include <boost/asio/io_service.hpp>
 #include <set>
 

internal/tPeerImplementation.h

@@ -40 +40 @@
 // External includes (system with <>, local with "")
 //----------------------------------------------------------------------
-#include <map>
 #include <boost/asio/ip/tcp.hpp>
 #include <boost/asio/deadline_timer.hpp>
 #include <boost/asio/io_service.hpp>
+#include <set>
+
 #include "rrlib/thread/tThread.h"
-#include "core/tFrameworkElement.h"
 
 //----------------------------------------------------------------------
 // Internal includes with ""
 //----------------------------------------------------------------------
-#include "plugins/network_transport/structure_info/tFrameworkElementInfo.h"
-#include "plugins/tcp/common/tRemoteTypes.h"
-//#include "plugins/tcp/internal/tNetworkPortInfo.h"
 #include "plugins/tcp/internal/tPeerInfo.h"
-#include "plugins/tcp/internal/tPortBufferChangeEvent.h"
-#include "plugins/tcp/internal/tSerializedStructureChange.h"
 
 //----------------------------------------------------------------------
@@ -70 +65 @@
 // Forward declarations / typedefs / enums
 //----------------------------------------------------------------------
-
-class tPlugin;
 class tServer;
+class tConnection;
 
 //----------------------------------------------------------------------
@@ -81 +75 @@
  * Implementation of different variants of TCP peer.
  */
-class tPeerImplementation : public core::tRuntimeListener
+class tPeerImplementation : public tTCPPlugin
 {
 
-  typedef rrlib::buffer_pools::tBufferPool < tPortBufferChangeEvent, rrlib::concurrent_containers::tConcurrency::MULTIPLE_READERS,
-          rrlib::buffer_pools::management::QueueBased, rrlib::buffer_pools::deleting::CollectGarbage,
-          rrlib::buffer_pools::recycling::UseOwnerStorageInBuffer > tPortBufferChangeEventPool;
-
 //----------------------------------------------------------------------
 // Public methods and typedefs
 //----------------------------------------------------------------------
 public:
-
-  typedef typename tPortBufferChangeEventPool::tPointer tChangeEventPointer;
 
   /*!
@@ -99 +87 @@
    * \param Options for peer creation
    */
-  tPeerImplementation(core::tFrameworkElement& framework_element, const tOptions& options);
+  tPeerImplementation();
 
   /*! Shuts peer down */
@@ -111 +99 @@
   /*! Starts actively connecting to the specified network */
   void Connect();
-
-  /*!
-   * Connect local port to port in remote runtime environment using TCP network transport plugin.
-   *
-   * \param local_port Local port to connect
-   * \param remote_runtime_uuid UUID of remote runtime
-   * \param remote_port_handle Handle of remote port
-   * \param remote_port_link Link of port in remote runtime environment
-   * \param disconnect If 'false' the ports are connected - if 'true' the ports are disconnected
-   * \return Returns error message if connecting failed. On success an empty string is returned.
-   */
-  std::string Connect(core::tAbstractPort& local_port, const std::string& remote_runtime_uuid, int remote_port_handle, const std::string remote_port_link, bool disconnect);
 
   /*!
@@ -148 +124 @@
 
   /*!
-   * Gets remote part with specified UUID.
-   * If no such part has been registered yet, creates a new one.
+   * Obtains remote runtime with specified UUID.
+   * If no such runtime has been registered yet, creates a new one.
    *
-   * \param uuid UUID of remote part
+   * \param uuid UUID of remote runtime
    * \param peer_type Peer type
    * \param peer_name Name of peer. Will be displayed in tooling and status messages. Does not need to be unique. Typically the program/process name.
-   * \param address IP address of remote part
+   * \param address IP address of remote runtime
    * \param never_forget Is this a remote peer to never forget?
-   * \return Pointer to remote part
-   */
-  tRemotePart* GetRemotePart(const tUUID& uuid, tPeerType peer_type, const std::string& peer_name, const boost::asio::ip::address& address, bool never_forget);
+   * \return Pointer to remote runtime
+   */
+  network_transport::generic_protocol::tRemoteRuntime* GetOrCreateRemoteRuntime(std::shared_ptr<tConnection>& connection, const tUUID& uuid, tPeerType peer_type, const std::string& peer_name, const boost::asio::ip::address& address, bool never_forget);
+
+  virtual network_transport::generic_protocol::tRemoteRuntime* GetRemoteRuntime(const network_transport::generic_protocol::tUriConnectorData& connector) override;
 
   /*!
@@ -168 +146 @@
   }
 
+  virtual void Init(tXmlNode* config_node) override;
+
   /*!
    * Reference to Boost asio IO service
@@ -174 +154 @@
   {
     return *io_service;
-  }
-
-  void PortChanged(data_ports::tPortDataPointer<const rrlib::rtti::tGenericObject>& value, tNetworkPortInfo* info, data_ports::tChangeContext& change_context)
-  {
-    tChangeEventPointer event_buffer = port_buffer_change_event_buffers.GetUnusedBuffer();
-    if (!event_buffer)
-    {
-      event_buffer = port_buffer_change_event_buffers.AddBuffer(std::unique_ptr<tPortBufferChangeEvent>(new tPortBufferChangeEvent()));
-    }
-    event_buffer->new_value = std::move(value);
-    event_buffer->network_port_info = info;
-    event_buffer->change_type = change_context.ChangeType();
-    incoming_port_buffer_changes.Enqueue(event_buffer);
   }
 
@@ -202 +169 @@
 
   /*!
+   * Processes incoming structure changes from local runtime
+   */
+  void ProcessLocalRuntimeStructureChanges()
+  {
+    tTCPPlugin::ProcessLocalRuntimeStructureChanges();
+  }
+
+  /*!
    * Called in a regular interval to do things like establishing new connections
    */
   void ProcessLowPriorityTasks();
-
-  /*!
-   * Processes all enqueued runtime change events in TCP thread
-   * and distributes them to all connections that are interested.
-   */
-  void ProcessRuntimeChangeEvents();
 
   /*!
@@ -219 +188 @@
 
   /*!
-   * Serializes shared ports and returns that in memory buffer
-   *
-   * \param connection_type_encoder Type encoder object of connection to serialize shared ports for
-   */
-  rrlib::serialization::tMemoryBuffer SerializeSharedPorts(common::tRemoteTypes& connection_type_encoder);
-
-
-  /*!
-   * \return True as soon as peer also serves clients interested in complete application structure
-   */
-  bool ServesStructure() const
-  {
-    return serve_structure.load();
-  }
-
-  /*!
    * Mark the peer list as changed.
    * This will cause the peer list to be sent to all connected peers.
@@ -248 +201 @@
   void StartServer();
 
-  void StartServingStructure()
-  {
-    serve_structure.store(true);
+  /*!
+   * \return Unused buffer for initialization of prototype streams
+   */
+  rrlib::serialization::tStackMemoryBuffer<16>& UnusedInitializationBuffer()
+  {
+    return unused_initialization_buffer;
   }
 
@@ -258 +214 @@
 private:
 
+  friend class tTCPPluginInstance;
   friend class tServer;
   friend class tTCPThread;
@@ -269 +226 @@
   friend struct tConnectorTask;
 
-  /*! Framework element associated with server */
-  core::tFrameworkElement& framework_element;
-
-  /*! Options that peer was created with */
-  tOptions create_options;
-
-  /*! Vector containing all network addresses this peer should try to connect to */
+  /*! Unused buffer for initialization of prototype streams (must not be destroyed before connections are) */
+  rrlib::serialization::tStackMemoryBuffer<16> unused_initialization_buffer;
+
+  /*! Vector containing all network addresses this peer should currently try to connect to */
   std::vector<std::string> connect_to;
+
+  /*! Vector containing all network addresses this peer already connected to (should not be added to connect_to again) */
+  std::set<std::string> connected_to;
 
   /*! Info on this peer */
@@ -286 +243 @@
    * connection to remote part is established.
    */
-  std::vector<std::unique_ptr<tPeerInfo>> other_peers;
+  std::vector<std::shared_ptr<tPeerInfo>> other_peers;
 
   /*! Revision of peer information */
@@ -306 +263 @@
   tServer* server;
 
-  /*! Cached info on ports shared by this peer */
-  std::map<core::tFrameworkElement::tHandle, rrlib::serialization::tFixedBuffer> shared_ports;
-
-  /*! Mutex for 'shared_ports' access */
-  rrlib::thread::tMutex shared_ports_mutex;
-
-  /*!
-   * True as soon as peer also serves clients interested in complete application structure.
-   * This may be enabled later, as this can cause quite a lot of overhead if done during
-   * construction and startup of large (MCA2) applications.
-   */
-  std::atomic<bool> serve_structure;
-
-  /*!
-   * Concurrent queue with incoming structure changes.
-   * Queue is filled when runtime changes occur (structure mutex is acquired by this thread).
-   * Queue is processed when TCP thread calls ProcessIncomingStructureChanges()
-   */
-  rrlib::concurrent_containers::tQueue < std::unique_ptr<tSerializedStructureChange>, rrlib::concurrent_containers::tConcurrency::SINGLE_READER_AND_WRITER,
-        rrlib::concurrent_containers::tDequeueMode::ALL > incoming_structure_changes;
-
   /*!
    * Actively connect to specified network?
@@ -333 +269 @@
   bool actively_connect;
 
-  /*! Ports to check subscriptions for */
-  std::vector<tFrameworkElementHandle> pending_subscription_checks;
-
-  /*! Mutex for 'pending_subscription_checks' access */
-  rrlib::thread::tMutex pending_subscription_checks_mutex;
-
-  /*! Copy for TCP thread */
-  std::vector<tFrameworkElementHandle> pending_subscription_checks_copy;
-
   /*! True when event loop is running (ProcessEvents() is regularly called by TCP thread) */
   bool event_loop_running;
-
-  /*! Concurrent queue with incoming port value changes */
-  rrlib::concurrent_containers::tQueue < tChangeEventPointer, rrlib::concurrent_containers::tConcurrency::MULTIPLE_WRITERS,
-        rrlib::concurrent_containers::tDequeueMode::ALL > incoming_port_buffer_changes;
-
-  /*! Buffer pool with port value change event buffers */
-  tPortBufferChangeEventPool port_buffer_change_event_buffers;
-
-  /*! List with deleted RPC ports - so that buffer pools created in connection for these ports can be deleted */
-  std::vector<core::tFrameworkElement::tHandle> deleted_rpc_ports;
-
-  /*! Mutex for 'deleted_rpc_ports' access */
-  rrlib::thread::tMutex deleted_rpc_ports_mutex;
 
 
@@ -368 +282 @@
   void AddPeerAddresses(tPeerInfo& existing_peer, const std::vector<boost::asio::ip::address>& addresses);
 
+  virtual bool Create(core::tAbstractPort& owner_port, const rrlib::uri::tURI& uri, const rrlib::uri::tURIElements& uri_elements, const core::tUriConnectOptions& connect_options) override;
+
   /*!
    * Scans current peer list and adds missing addresses
@@ -374 +290 @@
   void InferMissingAddresses();
 
-  /*! Is provided element a shared port (to be announced to other peers)? */
-  static bool IsSharedPort(core::tFrameworkElement& framework_element);
-
-  virtual void OnEdgeChange(core::tRuntimeListener::tEvent change_type, core::tAbstractPort& source, core::tAbstractPort& target) override;
-  virtual void OnFrameworkElementChange(core::tRuntimeListener::tEvent change_type, core::tFrameworkElement& element) override;
-
-  /*! Handles runtime changes from callbacks - and forwards info to 'incoming_structure_changes' queue */
-  void ProcessRuntimeChange(core::tRuntimeListener::tEvent change_type, core::tFrameworkElement& element, bool edge_change);
+  virtual void OnStartServingStructure() override;
 
   /*!
@@ -393 +302 @@
   void StartThread();
 
-  /*!
-   * Called whenever a new edge is added and updates network connection info if necessary
-   */
-  void UpdateNetworkConnectionInfo(core::tRuntimeListener::tEvent change_type, core::tAbstractPort& source, core::tAbstractPort& target, bool& target_port_changed);
 };