Changeset 12060 for source/ariba/communication/BaseCommunication.h

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/communication/BaseCommunication.h (modified) (16 diffs)

source/ariba/communication/BaseCommunication.h

@@ -50 +50 @@
 #include <boost/foreach.hpp>
 
+#ifdef ECLIPSE_PARSER
+    #define foreach(a, b) for(a : b)
+#else
+    #define foreach(a, b) BOOST_FOREACH(a, b)
+#endif
+
 // utilities
 #include "ariba/utility/types.h"
-#include "ariba/utility/messages.h"
+#include "ariba/utility/messages/MessageReceiver.h"
 #include "ariba/utility/logging/Logging.h"
 #include "ariba/utility/misc/Demultiplexer.hpp"
@@ -58 +64 @@
 
 // new transport and addressing
-#include "ariba/utility/addressing/addressing.hpp"
-#include "ariba/utility/transport/transport.hpp"
-#include "ariba/utility/transport/transport_connection.hpp"
+#include "ariba/utility/transport/transport_peer.hpp"
+#include "ariba/utility/transport/interfaces/transport_connection.hpp"
+#include "ariba/utility/transport/interfaces/transport_listener.hpp"
+#include "ariba/utility/addressing2/endpoint.hpp"
 
 // communication
@@ -72 +79 @@
 #include "ariba/communication/networkinfo/NetworkInformation.h"
 
-// disabled
-//#ifndef UNDERLAY_OMNET
-//  #include "ariba/communication/modules/transport/tcp/TCPTransport.h"
-//  #include "ariba/communication/modules/network/ip/IPv4NetworkProtocol.h"
-//  using ariba::communication::IPv4NetworkProtocol;
-//  using ariba::communication::TCPTransport;
-//#endif
-
 namespace ariba {
-  class SideportListener;
+    class SideportListener;
 }
 
@@ -87 +86 @@
 namespace communication {
 
+
+class communication_message_not_sent: public std::runtime_error
+{
+public:
+    /** Takes a character string describing the error.  */
+    explicit communication_message_not_sent(const string& __arg)  :
+        std::runtime_error(__arg)
+    {
+    }
+    
+    virtual ~communication_message_not_sent() throw() {}
+};
+
+
+
 using namespace std;
-using namespace ariba::addressing;
 using namespace ariba::transport;
 using namespace ariba::utility;
@@ -102 +115 @@
  * protocols and addressing schemes.
  *
- * @author Sebastian Mies, Christoph Mayer
+ * @author Sebastian Mies, Christoph Mayer, Mario Hock
  */
 class BaseCommunication:
         public NetworkChangeInterface,
-        public SystemEventListener,
         public transport_listener {
 
@@ -120 +132 @@
 
         /// Startup the base communication, start modules etc.
-        void start();
+        void start(addressing2::EndpointSetPtr listen_on);
 
         /// Stops the base communication, stop modules etc.
         void stop();
-
-        /// Sets the endpoints
-        void setEndpoints( string& endpoints );
 
         /// Check whether the base communication has been started up
@@ -147 +156 @@
          * @return A sequence number for this message
          */
-        seqnum_t sendMessage(const LinkID lid, const Message* message);
+        seqnum_t sendMessage(const LinkID& lid,
+                reboost::message_t message,
+                uint8_t priority,
+                bool bypass_overlay = false) throw(communication_message_not_sent);
 
         /**
@@ -164 +176 @@
          * @return List of LinkID
          */
-        LinkIDs getLocalLinks(const address_v* addr) const;
+//      LinkIDs getLocalLinks(const address_v* addr) const;  // XXX aktuell
 
         /**
@@ -187 +199 @@
 
         void unregisterEventListener(CommunicationEvents* _events);
-
-        /// called when a system event is emitted by system queue
-        virtual void handleSystemEvent(const SystemEvent& event);
 
         /**
@@ -196 +205 @@
          */ 
         virtual void receive_message(transport_connection::sptr connection,
-                reboost::message_t msg);
-
+                reboost::shared_buffer_t msg);
+
+    /**
+     * called within the ASIO thread
+     * when a connection is terminated (e.g. TCP close)
+     */ 
+    virtual void connection_terminated(transport_connection::sptr connection);
+
+    addressing2::EndpointPtr get_local_endpoint_of_link(const LinkID& linkid);
+    addressing2::EndpointPtr get_remote_endpoint_of_link(const LinkID& linkid);
+
+        
 protected:
 
@@ -205 +224 @@
          */ 
         void receiveMessage(transport_connection::sptr connection,
-                reboost::message_t msg);
+                reboost::shared_buffer_t message);
+        
+    /**
+     * called within the ARIBA thread (System Queue)
+     * when a connection is terminated (e.g. TCP close)
+     */ 
+    void connectionTerminated(transport_connection::sptr connection);
+        
 
         /// called when a network interface change happens
@@ -221 +247 @@
                 /// default constructor
                 LinkDescriptor() :
-                        localLink(LinkID::UNSPECIFIED), localLocator(NULL),
-                        remoteLink(LinkID::UNSPECIFIED), remoteLocator(NULL),
+                        localLink(LinkID::UNSPECIFIED),
+                        remoteLink(LinkID::UNSPECIFIED),
                         up(false) {
                 }
 
-                ~LinkDescriptor() {
-                        if (localLocator!=NULL)  delete localLocator;
-                        if (remoteLocator!=NULL) delete remoteLocator;
+                ~LinkDescriptor()
+                {
+                        if ( connection )
+                        {
+                            connection->unregister_communication_link(&localLink);
+                        }
                 }
 
@@ -240 +269 @@
                         return *unspec;
                 }
+                
+                
+                transport_connection::sptr get_connection() const
+                {
+                    return connection;
+                }
+                
+                void set_connection(const transport_connection::sptr& conn)
+                {
+                    // unregister from old connection,
+                    // if any (but normally there shouldn't..)
+                    if ( connection )
+                    {
+                        connection->unregister_communication_link(&localLink);
+                    }
+
+                    // * set_connection *
+                    connection = conn;
+                    
+                    // register this link with the connection
+                    conn->register_communication_link(&localLink);
+                }
 
                 bool unspecified;
@@ -245 +296 @@
                 /// link identifiers
                 LinkID localLink;
-                const address_v* localLocator;
+                addressing2::EndpointPtr localLocator;
 
                 /// used underlay addresses for the link
                 LinkID remoteLink;
-                const address_v* remoteLocator;
+                addressing2::EndpointPtr remoteLocator;
 
                 /// the remote end-point descriptor
-                EndpointDescriptor remoteEndpoint;
+                EndpointDescriptor remoteDescriptor;
 
                 /// flag, whether this link is up
                 bool up;
                 
+                
+        private:
                 /// connection if link is up
                 transport_connection::sptr connection;
@@ -281 +334 @@
         /// The local end-point descriptor
         EndpointDescriptor localDescriptor;
-
-#ifndef UNDERLAY_OMNET
+        
+        /**
+         * endpoint_set holding the addresses of the "server"-sockets,
+         * ---> that should be opened
+         * 
+         * (e.g. 0.0.0.0:41322)
+         */
+        addressing2::EndpointSetPtr listenOn_endpoints;
+        
+    /**
+     * endpoint_set holding the addresses of the "server"-sockets,
+     * ---> that are actually open
+     * 
+     * (e.g. 0.0.0.0:41322)
+     * 
+     * XXX should only be in transport_peer
+     */
+        addressing2::EndpointSetPtr active_listenOn_endpoints;
+        
+    /**
+     * endpoint_set holding the addresses of the "server"-sockets,
+     * ---> here the discovered "addressable" addresses are stored
+     * 
+     * (e.g. 192.168.0.5:41322)
+     * 
+     * XXX should only be in localDescriptor
+     */
+        addressing2::EndpointSetPtr local_endpoints;
+
         /// network change detector
         NetworkChangeDetection networkMonitor;
-#endif
 
         /// list of all remote addresses of links to end-points
-        class endpoint_reference {
-        public:
-                int count; ///< the number of open links to this end-point
-                const address_v* endpoint; ///< the end-point itself
-        };
-        vector<endpoint_reference> remote_endpoints;
-
-        /// adds an end-point to the list
-        void add_endpoint( const address_v* endpoint );
-
-        /// removes an end-point from the list
-        void remove_endpoint( const address_v* endpoint );
+        // XXX DEPRECATED
+//      class endpoint_reference {
+//      public:
+//              int count; ///< the number of open links to this end-point
+//              const address_v* endpoint; ///< the end-point itself
+//      };
+//      vector<endpoint_reference> remote_endpoints;
+
+        // XXX DEPRECATED
+//      /// adds an end-point to the list
+//      void add_endpoint( const address_v* endpoint );
+//
+//      /// removes an end-point from the list
+//      void remove_endpoint( const address_v* endpoint );
 
         /// event listener
@@ -314 +395 @@
         MessageReceiver* messageReceiver;
 
-        /// convenience: send message to peer
-        void send( Message* message, const EndpointDescriptor& endpoint );
-        void send( Message* message, const LinkDescriptor& descriptor );
+        
+        /* 
+         * Sends a message over an existing link.
+         *   ---> Adds »Link Message« Header
+         */
+        bool send_over_link(
+                const uint8_t type,
+                reboost::message_t message,
+                const LinkDescriptor& desc,
+                const uint8_t priority);
+
+    /* 
+     * Sends a message to a known peer. (To all known endpoints.)
+     *   ---> Adds »Peer Message« Header
+     */
+        void send_to_peer(
+                const uint8_t type,
+                const PeerID& peer_id,
+                reboost::message_t message,
+                const EndpointDescriptor& endpoint,
+                const uint8_t priority );
+        
 
         /// state of the base communication