Changeset 26:9f93ea63d74e in finroc_plugins_network_transport

Timestamp:

08.06.2017 02:40:59 (4 years ago)

Author:

Max Reichardt <mreichardt@…>

Branch:

17.03

Phase:

public

Message:

Adapts to removal of reverse pushing - by adding a parameter to CONNECT_PORTS opcode as to whether connection is used for publishing or subscribing

Location:

generic_protocol

Files:

• definitions.h (modified) (1 diff)
• tDynamicConnectionData.h (modified) (1 diff)
• tNetworkPortInfo.cpp (modified) (2 diffs)
• tNetworkPortInfo.h (modified) (2 diffs)
• tNetworkPortInfoClient.cpp (modified) (1 diff)
• tNetworkTransportPlugin.cpp (modified) (1 diff)
• tRemoteRuntime.cpp (modified) (11 diffs)
• tStaticConnectorParameters.h (modified) (5 diffs)

generic_protocol/definitions.h

@@ -211 +211 @@
 
 
-// Parameters: [connection handle == client port handle] after message: [static connector parameters][dynamic connection data]
-typedef tMessage <tOpCode, tOpCode::CONNECT_PORTS, tMessageSize::VARIABLE_UP_TO_4GB, tFrameworkElementHandle> tConnectPortsMessage;
+// Parameters: [connection handle == client port handle][connect direction: connection for publishing data to server?] after message: [static connector parameters][dynamic connection data]
+typedef tMessage <tOpCode, tOpCode::CONNECT_PORTS, tMessageSize::VARIABLE_UP_TO_4GB, tFrameworkElementHandle, bool> tConnectPortsMessage;
 
 // Parameters: [connection handle == client port handle] after message: [string: error message]

generic_protocol/tDynamicConnectionData.h

@@ -72 +72 @@
   uint16_t strategy = -1;
 
-  /*! bool  */
 };
 

generic_protocol/tNetworkPortInfo.cpp

@@ -66 +66 @@
 // Const values
 //----------------------------------------------------------------------
+__thread tNetworkPortInfo* tNetworkPortInfo::current_publishing_network_port = nullptr;
 
 //----------------------------------------------------------------------
@@ -92 +93 @@
   if (data_ports::IsDataFlowType(port->GetDataType()))
   {
-    if (port->IsOutputPort())
-    {
-      values_to_send.SetMaxLength(port->GetFlag(core::tFrameworkElement::tFlag::PUSH_STRATEGY_REVERSE) ? 1 : 0);
-    }
-    else
+    if (!port->IsOutputPort())
     {
       values_to_send.SetMaxLength(strategy < 0 ? 0 : strategy);

generic_protocol/tNetworkPortInfo.h

@@ -164 +164 @@
   {
     //FINROC_LOG_PRINT(DEBUG, "Port Changed ", port.GetWrapped()->GetQualifiedName(), " ", rrlib::serialization::Serialize(*value));
-    if (strategy > 0)
+    if (strategy > 0 && (!(current_publishing_network_port && current_publishing_network_port->connection_handle == -connection_handle && (&current_publishing_network_port->remote_runtime == &remote_runtime))))
     {
       remote_runtime.GetLocalRuntimeInfo().OnPortChange(value, this, change_context);
@@ -293 +293 @@
   std::unique_ptr<tNetworkPortInfoClient> client_info;
 
+  /*! Network port that currently publishes data using the current thread (null if current thread does not publish data from the net) */
+  static __thread tNetworkPortInfo* current_publishing_network_port;
+
 
   virtual void OnManagedDelete() override;

generic_protocol/tNetworkPortInfoClient.cpp

@@ -122 +122 @@
   {
     tStaticNetworkConnectorParameters static_parameters(used_by_connectors[0]->StaticParameters(), NetworkPortInfo().remote_port_handle);
-    tConnectPortsMessage::Serialize(false, true, stream, port->GetHandle());
+    tConnectPortsMessage::Serialize(false, true, stream, port->GetHandle(), port->IsInputPort());
     stream << static_parameters << new_connection_data;
     tConnectPortsMessage::FinishMessage(stream);

generic_protocol/tNetworkTransportPlugin.cpp

@@ -201 +201 @@
     {
       connector.temporary_conversion_port.reset(new data_ports::tGenericPort(remote_runtime.client_ports, rrlib::uri::tURI(connector.StaticParameters().server_port_id.path).ToString(), connector.OwnerPort().GetDataType(), tFlag::ACCEPTS_DATA | tFlag::PUSH_STRATEGY | tFlag::EMITS_DATA | (created_port.GetWrapped()->IsOutputPort() ? tFlag::OUTPUT_PORT : tFlag::PORT)), core::tPortWrapperBase::tDeleter());
+      connector.temporary_conversion_port->ConnectTo(connector.OwnerPort(), core::tConnectionFlag::NON_PRIMARY_CONNECTOR); // Do this first so that initial pushing works (does not send default first) for remote input ports
       connector.temporary_conversion_port->ConnectTo(connector.temporary_connector_port->GetPort(), core::tConnectOptions(connector.LocalConversionOperations(), core::tConnectionFlag::NON_PRIMARY_CONNECTOR));
-      connector.temporary_conversion_port->ConnectTo(connector.OwnerPort(), core::tConnectionFlag::NON_PRIMARY_CONNECTOR);
     }
     else

generic_protocol/tRemoteRuntime.cpp

@@ -302 +302 @@
     if (port.first && port.first->IsReady() && data_ports::IsDataFlowType(port.first->GetDataType()))
     {
+      tNetworkPortInfo::current_publishing_network_port = port.second;
       data_ports::tGenericPort generic_port = data_ports::tGenericPort::Wrap(*port.first);
 
@@ -346 +347 @@
       while (another_value);
 
+      tNetworkPortInfo::current_publishing_network_port = nullptr;
       message.FinishDeserialize(stream);
       connection.received_data_after_last_connect = true;
@@ -542 +544 @@
         if (lock.TryLock())
         {
+          bool publish_connection = message.Get<1>();
+          bool tool_connection = GetDesiredStructureInfo() != runtime_info::tStructureExchange::SHARED_PORTS;
+
           // Read subscription data
           tStaticNetworkConnectorParameters static_subscription_parameters;
@@ -556 +561 @@
 
           tFlags flags = tFlag::NETWORK_ELEMENT | tFlag::VOLATILE;
-          if (port->IsOutputPort())
+          if ((!tool_connection) && port->IsOutputPort() && publish_connection)
+          {
+            throw std::runtime_error("Cannot publish to output ports with basic (SHARED_PORTS) connection");
+          }
+          if (!publish_connection)
           {
             flags |= tFlag::ACCEPTS_DATA; // create input port
@@ -562 +571 @@
           else
           {
-            flags |= tFlag::OUTPUT_PORT | tFlag::EMITS_DATA; // create output io port
-          }
-          if (GetDesiredStructureInfo() != runtime_info::tStructureExchange::SHARED_PORTS)
+            flags |= tFlag::OUTPUT_PORT | tFlag::EMITS_DATA | tFlag::NO_INITIAL_PUSHING; // create output io port
+          }
+          if (tool_connection)
           {
             flags |= tFlag::TOOL_PORT;
@@ -572 +581 @@
             flags |= tFlag::PUSH_STRATEGY;
           }
-          if (static_subscription_parameters.reverse_push)
-          {
-            flags |= tFlag::PUSH_STRATEGY_REVERSE;
-          }
-
+
+          std::string port_name = rrlib::uri::tURI(port->GetPath()).ToString() + (publish_connection ? " (Publish)" : " (Subscribe)");
           core::tAbstractPort* port_to_connect_to = port;
           if (!static_subscription_parameters.server_side_conversion.NoConversion())
@@ -583 +589 @@
 
             // Check whether port already exists
-            if (port->IsOutputPort())
+            if (!publish_connection)
             {
               for (auto it = port->OutgoingConnectionsBegin(); it != port->OutgoingConnectionsEnd(); ++it)
@@ -647 +653 @@
 
               // Create and connect new conversion port
-              data_ports::tGenericPort created_port(rrlib::uri::tURI(port->GetPath()).ToString(), &GetServerPortsElement(), destination_type, tFlag::NETWORK_ELEMENT | tFlag::VOLATILE | tFlag::EMITS_DATA | tFlag::ACCEPTS_DATA | (port->IsOutputPort() ? tFlag::OUTPUT_PORT : tFlag::PORT));
-              if (created_port.ConnectTo(port, core::tConnectOptions(conversion, core::tConnectionFlag::NON_PRIMARY_CONNECTOR)))
+              data_ports::tGenericPort created_port(port_name + " to " + destination_type.GetName(), &GetServerPortsElement(), destination_type, tFlag::NETWORK_ELEMENT | tFlag::VOLATILE | tFlag::EMITS_DATA | tFlag::ACCEPTS_DATA | (publish_connection ? (tFlag::OUTPUT_PORT | tFlag::NO_INITIAL_PUSHING) : tFlag::PORT) | (tool_connection ? tFlag::TOOL_PORT : tFlag::PORT));
+              if (created_port.ConnectTo(port, core::tConnectOptions(conversion, core::tConnectionFlag::NON_PRIMARY_CONNECTOR | (publish_connection ? core::tConnectionFlag::DIRECTION_TO_DESTINATION : core::tConnectionFlag::DIRECTION_TO_SOURCE))))
               {
                 created_port.GetWrapped()->EmplaceAnnotation<tServerSideConversionAnnotation>(static_subscription_parameters.server_side_conversion);
@@ -662 +668 @@
           }
 
-          data_ports::tGenericPort created_port(rrlib::uri::tURI(port->GetPath()).ToString(), &GetServerPortsElement(), &GetServerPortsElement(), port_to_connect_to->GetDataType(), flags);
+          data_ports::tGenericPort created_port(port_name, &GetServerPortsElement(), &GetServerPortsElement(), port_to_connect_to->GetDataType(), flags);
           tNetworkPortInfo* network_port_info = new tNetworkPortInfo(*this, message.Get<0>(), message.Get<0>(), dynamic_connection_data.strategy, *created_port.GetWrapped(), port->GetHandle());
           network_port_info->SetDesiredEncoding(static_subscription_parameters.server_side_conversion.encoding);
@@ -669 +675 @@
           created_port.SetPullRequestHandler(this);
           created_port.Init();
-          created_port.ConnectTo(*port_to_connect_to, core::tConnectionFlag::NON_PRIMARY_CONNECTOR);
+          created_port.ConnectTo(*port_to_connect_to, core::tConnectionFlag::NON_PRIMARY_CONNECTOR | (publish_connection ? core::tConnectionFlag::DIRECTION_TO_DESTINATION : core::tConnectionFlag::DIRECTION_TO_SOURCE));
           server_port_map.emplace(message.Get<0>(), network_port_info);
           FINROC_LOG_PRINT(DEBUG_VERBOSE_1, "Created server port ", created_port.GetWrapped());
@@ -784 +790 @@
           data_ports::common::tAbstractDataPort& data_port = static_cast<data_ports::common::tAbstractDataPort&>(*client_port->GetPort());
           data_port.SetPushStrategy(dynamic_info.strategy > 0);
-          //data_port.SetReversePushStrategy(dynamic_info.flags.Get(tFlag::PUSH_STRATEGY_REVERSE));
           //data_port.SetMinNetUpdateIntervalRaw(dynamic_info.min_net_update_time);
           //tNetworkPortInfo* network_port_info = data_port.GetAnnotation<tNetworkPortInfo>();

generic_protocol/tStaticConnectorParameters.h

@@ -77 +77 @@
   TServerPortId server_port_id;
 
-  /*! Whether to subscribe at input port for reverse pushing */
-  bool reverse_push;
 
   tStaticConnectorParameters() :
-    server_port_id(TServerPortId()),
-    reverse_push(false)
+    server_port_id(TServerPortId())
   {}
 
@@ -89 +86 @@
   tStaticConnectorParameters(const tStaticConnectorParameters<TOtherId>& other, const TServerPortId& this_id) :
     server_side_conversion(other.server_side_conversion),
-    server_port_id(this_id),
-    reverse_push(other.reverse_push)
+    server_port_id(this_id)
   {}
 
@@ -99 +95 @@
 inline bool operator==(const tStaticConnectorParameters<TServerPortId>& lhs, const tStaticConnectorParameters<TServerPortId>& rhs)
 {
-  return lhs.server_port_id == rhs.server_port_id && lhs.reverse_push == rhs.reverse_push && lhs.server_side_conversion == rhs.server_side_conversion;
+  return lhs.server_port_id == rhs.server_port_id && lhs.server_side_conversion == rhs.server_side_conversion;
 }
 
@@ -111 +107 @@
 inline rrlib::serialization::tOutputStream& operator << (rrlib::serialization::tOutputStream& stream, const tStaticConnectorParameters<TServerPortId>& data)
 {
-  stream << data.server_port_id << data.reverse_push << data.server_side_conversion;
+  stream << data.server_port_id << data.server_side_conversion;
   return stream;
 }
@@ -118 +114 @@
 inline rrlib::serialization::tInputStream& operator >> (rrlib::serialization::tInputStream& stream, tStaticConnectorParameters<TServerPortId>& data)
 {
-  stream >> data.server_port_id >> data.reverse_push >> data.server_side_conversion;
+  stream >> data.server_port_id >> data.server_side_conversion;
   return stream;
 }