Changeset 12060 for source/ariba/Node.cpp

Timestamp:

Jun 19, 2013, 11:05:49 AM (11 years ago)

Author:

hock@…

Message:

Reintegrate branch: 20130111-hock-message_classes

improvements:

• new message classes (reboost, zero-copy)
• "fast path" for direct links (skip overlay layer)
• link-properties accessible from the application
• SystemQueue can call boost::bind functions
• protlib compatibility removed (32bit overhead saved in every message)
• addressing2
• AddressDiscovery discoveres only addresses on which we're actually listening
• ariba serialization usage reduced (sill used in OverlayMsg)
• Node::connect, easier and cleaner interface to start-up ariba from the application
• ariba configs via JSON, XML, etc (boost::property_tree)
• keep-alive overhead greatly reduced
• (relayed) overlay links can actually be closed now
• lost messages are detected in most cases
• notification to the application when link is transformed into direct-link
• overlay routing: send message to second best hop if it would be dropped otherwise
• SequenceNumbers (only mechanisms, so for: upward compatibility)
• various small fixes

regressions:

• bluetooth is not yet working again
• bootstrap modules deactivated
• liblog4xx is not working (use cout-logging)

This patch brings great performance and stability improvements at cost of backward compatibility.
Also bluetooth and the bootstrap modules have not been ported to the new interfaces, yet.

File:

• source/ariba/Node.cpp (modified) (4 diffs)

source/ariba/Node.cpp

@@ -39 +39 @@
 #include "Node.h"
 
+#include <boost/foreach.hpp>
+
 #include "ariba/overlay/BaseOverlay.h"
+#include "ariba/communication/BaseCommunication.h"
+
 #include "ariba/utility/types/OverlayParameterSet.h"
 #include "ariba/communication/EndpointDescriptor.h"
 
+#include <boost/property_tree/exceptions.hpp>
+
+using namespace std;
 using ariba::communication::EndpointDescriptor;
+using boost::property_tree::ptree;
 
 namespace ariba {
 
-Node::Node(AribaModule& ariba_mod, const Name& node_name) :
-        name(node_name), ariba_mod(ariba_mod) {
-        base_overlay = new BaseOverlay();
+//Node::Node(AribaModule& ariba_mod, const Name& node_name) :
+//      name(node_name), ariba_mod(ariba_mod) {
+//      base_overlay = new BaseOverlay();
+//}
+
+Node::Node()  :
+        name(Name::UNSPECIFIED),
+        base_communication(NULL),
+        base_overlay(NULL)
+{
+    base_communication = new BaseCommunication();
+    base_overlay = new BaseOverlay();
 }
 
@@ -55 +72 @@
         delete base_overlay;
         base_overlay = NULL;
-}
-
-void Node::join(const Name& vnetname) {
-        spovnetId = vnetname.toSpoVNetId();
-        nodeId = generateNodeId(name);
-
-        // start base comm if not started
-        if( !ariba_mod.base_comm->isStarted() )
-                ariba_mod.base_comm->start();
-
-        // start base overlay if not started
-        // join against ourselfs
-        if( !base_overlay->isStarted() )
-                base_overlay->start( *ariba_mod.base_comm, nodeId );
-        base_overlay->joinSpoVNet( spovnetId );
-
-        // join against static bootstrap points and
-        // start automatic bootstrapping modules
-        vector<AribaModule::BootstrapMechanism> mechanisms
-                = ariba_mod.getBootstrapMechanisms(vnetname);
-
-        vector<pair<BootstrapManager::BootstrapType,string> > internalmodules;
-
-        BOOST_FOREACH(AribaModule::BootstrapMechanism m, mechanisms){
-                switch(m){
-                        case AribaModule::BootstrapMechanismStatic:
-                        {
-                                const communication::EndpointDescriptor* ep =
-                                                        ariba_mod.getBootstrapNode(vnetname, m);
-                                        if( ep != NULL && ep->isUnspecified() == false )
-                                                base_overlay->joinSpoVNet( spovnetId, *ep);
-                                break;
-                        }
-                        case AribaModule::BootstrapMechanismBroadcast:
-                                internalmodules.push_back(make_pair(
-                                                BootstrapManager::BootstrapTypePeriodicBroadcast,
-                                                ariba_mod.getBootstrapInfo(vnetname, m)));
-                                break;
-                        case AribaModule::BootstrapMechanismMulticastDNS:
-                                internalmodules.push_back(make_pair(
-                                                BootstrapManager::BootstrapTypeMulticastDns,
-                                                ariba_mod.getBootstrapInfo(vnetname, m)));
-                                break;
-                        case AribaModule::BootstrapMechanismSDP:
-                                internalmodules.push_back(make_pair(
-                                                BootstrapManager::BootstrapTypeBluetoothSdp,
-                                                ariba_mod.getBootstrapInfo(vnetname, m)));
-                                break;
-                        default:
-                                break;
-                }
-        }
-
-        // start automatic overlay bootstrapping modules
-        base_overlay->startBootstrapModules(internalmodules);
-
-        // done
-}
-
-void Node::initiate(const Name& vnetname, const SpoVNetProperties& parm) {
-        utility::OverlayParameterSet ovrpset;
-        ovrpset.setOverlayStructure(
-                        (utility::OverlayParameterSet::_OverlayStructure)
-                        parm.getBaseOverlayType()
-                        );
-
-        spovnetId = vnetname.toSpoVNetId();
-        nodeId = generateNodeId(name);
-
-        // start base comm if not started
-        if( !ariba_mod.base_comm->isStarted() )
-                ariba_mod.base_comm->start();
-
-        // start base overlay if not started
-        if( !base_overlay->isStarted() )
-                base_overlay->start( *ariba_mod.base_comm, nodeId );
-
-        base_overlay->createSpoVNet( spovnetId, ovrpset );
-}
+        
+        delete base_communication;
+        base_communication = NULL;
+}
+
+void Node::connect(const ptree& config)
+{
+    using namespace boost::property_tree;
+    using namespace addressing2;
+    
+    assert( ! base_communication->isStarted() );
+    assert( ! base_overlay->isStarted() );
+
+    // XXX needed since »empty_ptree<ptree>()« is not working
+    //   ---> see: http://stackoverflow.com/questions/5003549/where-is-boost-property-treeempty-ptree
+    static const ptree empty_pt;
+    
+    // SpovNet ID
+    Name spovnet_name(config.get<string>("spovnet_name"));
+    spovnetId = spovnet_name.toSpoVNetId();
+    
+    // Node ID
+    try
+    {
+        name = config.get<string>("node_name");
+    }
+    catch ( ptree_bad_path& e )
+    {
+        name = Name::UNSPECIFIED;
+    }
+    nodeId = generateNodeId(name);
+
+    
+    
+    /* Base Communication */
+    EndpointSetPtr listen_on;
+    try
+    {
+        listen_on = endpoint_set::create_EndpointSet(
+            config.get_child("listen_on"));
+    }
+    catch ( ptree_bad_path& e )
+    {
+        /* no endpoints specified, using default: »[::]:41322+« */
+        
+        ptree default_listen_on;
+        default_listen_on.put("endp.category", "TCPIP");
+        default_listen_on.put("endp.addr", "::");
+        default_listen_on.put("endp.port", 0);  // defaults to 41322 (or higher)
+        
+        listen_on = endpoint_set::create_EndpointSet(default_listen_on);
+//        logging_warn("No endpoints specified in config. ---> Using default.");
+        cout << "No endpoints specified in config. ---> Using default." << endl;
+    }
+    base_communication->start(listen_on);
+    
+    // TODO maybe notify the upper layer whether we have any active endpoints
+    
+
+    /* Base Overlay */
+    base_overlay->start( base_communication, nodeId );
+
+    base_overlay->createSpoVNet( spovnetId );
+    base_overlay->joinSpoVNet( spovnetId );
+    
+    
+    
+    /* Bootstrap */
+    const ptree& bootstrap_pt = config.get_child("bootstrap", empty_pt);
+    
+    // Static Bootstrap
+    try
+    {
+        // read endpoint_set from config
+        EndpointSetPtr ep_set = endpoint_set::create_EndpointSet(
+                bootstrap_pt.get_child("direct"));
+        
+        EndpointDescriptor ep = EndpointDescriptor::UNSPECIFIED();
+        ep.replace_endpoint_set(ep_set);
+        
+        // try to connect
+        base_overlay->joinSpoVNet( spovnetId, ep);
+    }
+    catch ( ptree_bad_path& e )
+    {
+//        logging_info("No direct bootstrap info in config.");
+        cout << "No direct bootstrap info in config." << endl;
+    }
+
+    
+    /* Bootstrap modules */
+    vector<pair<BootstrapManager::BootstrapType,string> > internalmodules;
+    
+    // Bootstrap: Broadcast
+    if ( bootstrap_pt.get("broadcast", false) )
+    {
+        internalmodules.push_back(make_pair(
+                BootstrapManager::BootstrapTypePeriodicBroadcast,""));
+    }
+    
+    // Bootstrap: MDNS
+    if ( bootstrap_pt.get("mdns", false) )
+    {
+        internalmodules.push_back(make_pair(
+                BootstrapManager::BootstrapTypeMulticastDns,""));
+    }
+
+    // Bootstrap: SDP
+    if ( bootstrap_pt.get("sdp", false) )
+    {
+        internalmodules.push_back(make_pair(
+                BootstrapManager::BootstrapTypeBluetoothSdp,""));
+    }
+
+    // start automatic overlay bootstrapping modules
+    base_overlay->startBootstrapModules(internalmodules);
+}
+
+//void Node::join(const Name& vnetname) {
+//      spovnetId = vnetname.toSpoVNetId();
+//      nodeId = generateNodeId(name);
+//
+//      // start base comm if not started
+//      if( !ariba_mod.base_comm->isStarted() )
+//              ariba_mod.base_comm->start();
+//
+//      // start base overlay if not started
+//      // join against ourselfs
+//      if( !base_overlay->isStarted() )
+//              base_overlay->start( *ariba_mod.base_comm, nodeId );
+//      base_overlay->joinSpoVNet( spovnetId );
+//
+//      // join against static bootstrap points and
+//      // start automatic bootstrapping modules
+//      vector<AribaModule::BootstrapMechanism> mechanisms
+//              = ariba_mod.getBootstrapMechanisms(vnetname);
+//
+//      vector<pair<BootstrapManager::BootstrapType,string> > internalmodules;
+//
+//      BOOST_FOREACH(AribaModule::BootstrapMechanism m, mechanisms){
+//              switch(m){
+//                      case AribaModule::BootstrapMechanismStatic:
+//                      {
+//                              const communication::EndpointDescriptor* ep =
+//                                                      ariba_mod.getBootstrapNode(vnetname, m);
+//                                      if( ep != NULL && ep->isUnspecified() == false )
+//                                              base_overlay->joinSpoVNet( spovnetId, *ep);
+//                              break;
+//                      }
+//                      case AribaModule::BootstrapMechanismBroadcast:
+//                              internalmodules.push_back(make_pair(
+//                                              BootstrapManager::BootstrapTypePeriodicBroadcast,
+//                                              ariba_mod.getBootstrapInfo(vnetname, m)));
+//                              break;
+//                      case AribaModule::BootstrapMechanismMulticastDNS:
+//                              internalmodules.push_back(make_pair(
+//                                              BootstrapManager::BootstrapTypeMulticastDns,
+//                                              ariba_mod.getBootstrapInfo(vnetname, m)));
+//                              break;
+//                      case AribaModule::BootstrapMechanismSDP:
+//                              internalmodules.push_back(make_pair(
+//                                              BootstrapManager::BootstrapTypeBluetoothSdp,
+//                                              ariba_mod.getBootstrapInfo(vnetname, m)));
+//                              break;
+//                      default:
+//                              break;
+//              }
+//      }
+//
+//      // start automatic overlay bootstrapping modules
+//      base_overlay->startBootstrapModules(internalmodules);
+//
+//      // done
+//}
+//
+//void Node::initiate(const Name& vnetname, const SpoVNetProperties& parm) {
+//      utility::OverlayParameterSet ovrpset;
+//      ovrpset.setOverlayStructure(
+//                      (utility::OverlayParameterSet::_OverlayStructure)
+//                      parm.getBaseOverlayType()
+//                      );
+//
+//      spovnetId = vnetname.toSpoVNetId();
+//      nodeId = generateNodeId(name);
+//
+//      // start base comm if not started
+//      if( !ariba_mod.base_comm->isStarted() )
+//              ariba_mod.base_comm->start();
+//
+//      // start base overlay if not started
+//      if( !base_overlay->isStarted() )
+//              base_overlay->start( *ariba_mod.base_comm, nodeId );
+//
+//      base_overlay->createSpoVNet( spovnetId, ovrpset );
+//}
 
 void Node::leave() {
         base_overlay->stopBootstrapModules();
         base_overlay->leaveSpoVNet();
-        ariba_mod.base_comm->stop();
+        base_communication->stop();  // XXX before »base_overlay->stop()« ??
         base_overlay->stop();
 }
@@ -172 +305 @@
 }
 
+bool Node::isLinkDirect(const ariba::LinkID& lnk) const
+{
+    return base_overlay->isLinkDirect(lnk);
+}
+
+int Node::getHopCount(const ariba::LinkID& lnk) const
+{
+    return base_overlay->getHopCount(lnk);
+}
+
+
+
+
+/* +++++ Message sending +++++ */
+void Node::check_send_priority(uint8_t priority)
+{
+    if ( priority < send_priority::HIGHEST || priority > send_priority::LOWEST )
+        throw std::invalid_argument("Illegal priority");     
+}
+
+
+// +++ new interface +++
+const SequenceNumber& Node::sendMessage(reboost::message_t msg, const LinkID& lnk, uint8_t priority)
+{
+    // check priority
+    check_send_priority(priority);
+        
+    // * call base overlay *
+    return base_overlay->sendMessage(msg, lnk, priority);
+}
+
+// +++ legacy interface +++
+seqnum_t Node::sendMessage(const DataMessage& msg, const LinkID& lnk)
+{
+    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+    
+    try
+    {
+        base_overlay->sendMessage(message, lnk, send_priority::NORMAL);
+        return 0;
+    }
+    catch ( ariba::overlay::message_not_sent& e )
+    {
+        logging_warn("Message could not be sent. Dropped.");
+        return -1;
+    }
+}
+
+
+// +++ new interface +++
+const SequenceNumber& Node::sendMessage(reboost::message_t msg, const NodeID& nid,
+        const ServiceID& sid, uint8_t priority, const LinkProperties& req) {
+
+    // check priority
+    check_send_priority(priority);
+        
+    // * call base overlay *
+    return base_overlay->sendMessage(msg, nid, priority, sid);
+}
+
+// +++ legacy interface +++
 seqnum_t Node::sendMessage(const DataMessage& msg, const NodeID& nid,
                 const ServiceID& sid, const LinkProperties& req) {
-        return base_overlay->sendMessage((Message*) msg, nid, sid);
-}
-
+    
+//    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+    reboost::message_t message;
+    message.push_back( ((Message*) msg)->serialize_into_shared_buffer() );
+    
+    try
+    {
+        sendMessage(message, nid, sid, send_priority::NORMAL, req);
+        return 0;
+    }
+    catch ( ariba::overlay::message_not_sent& e )
+    {
+        logging_warn("Message could not be sent. Dropped.");
+        return -1;
+    }
+}
+
+
+// +++ new interface +++
+NodeID Node::sendMessageCloserToNodeID(reboost::message_t msg, const NodeID& nid, const ServiceID& sid,
+        uint8_t priority, const LinkProperties& req) {
+
+    // check priority
+    check_send_priority(priority);
+        
+    // * call base overlay *
+    return base_overlay->sendMessageCloserToNodeID(msg, nid, priority, sid);
+}
+
+// +++ legacy interface +++
 NodeID Node::sendMessageCloserToNodeID(const DataMessage& msg, const NodeID& nid, const ServiceID& sid,
         const LinkProperties& req) {
     
-    return base_overlay->sendMessageCloserToNodeID((Message*) msg, nid, sid);
-}
-
-
-seqnum_t Node::sendMessage(const DataMessage& msg, const LinkID& lnk) {
-        return base_overlay->sendMessage((Message*) msg, lnk);
-}
-
+    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+    
+    return sendMessageCloserToNodeID(message, nid, sid, send_priority::NORMAL, req);
+}
+
+
+// +++ new interface +++
+void Node::sendBroadcastMessage(reboost::message_t msg, const ServiceID& sid, uint8_t priority) {
+
+    // check priority
+    check_send_priority(priority);
+        
+    // * call base overlay *
+    return base_overlay->broadcastMessage(msg, sid, priority);
+}
+
+// +++ legacy interface +++
 void Node::sendBroadcastMessage(const DataMessage& msg, const ServiceID& sid) {
-        return base_overlay->broadcastMessage((Message*)msg, sid);
-}
+    reboost::message_t message = ((Message*) msg)->wrap_up_for_sending();
+
+        return sendBroadcastMessage(message, sid);
+}
+
+
+/* +++++ [Message sending] +++++ */
+
+
+
 
 bool Node::bind(NodeListener* listener) {
@@ -204 +442 @@
         bool ret = base_overlay->bind(listener, sid);
 
-        // now that we have a listener, we can ask if sniffing is ok
-        if( ariba_mod.sideport_sniffer != NULL ){
-                base_overlay->registerSidePort(ariba_mod.sideport_sniffer);
-        }
+//      // now that we have a listener, we can ask if sniffing is ok
+//      if( ariba_mod.sideport_sniffer != NULL ){
+//              base_overlay->registerSidePort(ariba_mod.sideport_sniffer);
+//      }
 
         return ret;