Changeset 12060 for source/ariba/utility/transport/transport_peer.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.

Location:

source/ariba/utility/transport

Files:

• . (modified) (1 prop)
• transport_peer.cpp (modified) (4 diffs)

source/ariba/utility/transport/transport_peer.cpp

@@ -1 @@
-
-#include "ariba/config.h"
 #include "transport_peer.hpp"
-#include "transport.hpp"
-#include <boost/asio/ip/tcp.hpp>
+
+// ariba
+#include "StreamTransport/StreamTransport.hpp"
+#include "ariba/utility/addressing2/tcpip_endpoint.hpp"
+
+// boost
 #include <boost/asio/error.hpp>
 #include <boost/foreach.hpp>
-
-#ifdef ECLIPSE_PARSER
-    #define foreach(a, b) for(a : b)
-#else
-    #define foreach(a, b) BOOST_FOREACH(a, b)
-#endif
 
 // namespace ariba::transport
@@ -17 +13 @@
 namespace transport {
 
-using namespace ariba::addressing;
+using namespace addressing2;
 using boost::asio::ip::tcp;
 
@@ -26 +22 @@
 use_logging_cpp(transport_peer);
 
-transport_peer::transport_peer( endpoint_set& local_set ) : local(local_set) {
-    
-    // setup tcp transports
-    foreach(tcp_port_address port, local.tcp) {
+transport_peer::transport_peer()  :
+        local(new addressing2::endpoint_set())
+{
+}
+
+EndpointSetPtr transport_peer::add_listenOn_endpoints(EndpointSetPtr endpoints)
+{
+    // TCP Endpoints
+    BOOST_FOREACH( shared_ptr<tcpip_endpoint> endp, endpoints->get_tcpip_endpoints() )
+    {
+        // automatic port detection
+        bool port_detection = false;
+        uint16_t try_port = 41322;
         
-        if (local.ip.size() > 0) {
-            foreach(ip_address ip_addr, local.ip) {
-                
-                tcp::endpoint endp(ip_addr.asio(), port.asio());
-                create_service(endp);
-            }
-        } else {
-            tcp::endpoint endp_v6(tcp::v6(), port.asio());
-            tcp::endpoint endp_v4(tcp::v4(), port.asio());
-            
-            create_service(endp_v6);
-            create_service(endp_v4);
+        tcp::endpoint asio_endp = endp->to_asio();
+        if ( asio_endp.port() == 0 )
+        {
+            port_detection = true;
         }
         
-    }
-    
+        
+        // create new server socket
+        do
+        {
+            try
+            {
+                // automatic port detection
+                if ( port_detection )
+                {
+                    asio_endp.port(try_port);
+                    endp = tcpip_endpoint::create_TcpIP_Endpoint(asio_endp);
+                }
+                
+                TransportProtocolPtr tmp_ptr(new StreamTransport<tcp>(endp->to_asio()));
+                transport_streams.push_back(tmp_ptr);
+                logging_info("Listening on IP/TCP " << endp->to_string());
+                
+                local->add_endpoint(endp);
+                port_detection = false;
+            }
+            
+            catch (boost::system::system_error& e)
+            {
+                // address in use
+                if (e.code() == boost::asio::error::address_in_use)
+                {
+                    // BRANCH: automatic port detection
+                    if ( port_detection )
+                    {
+                        // give up ?
+                        if ( try_port > 41422 )
+                        {
+                            logging_warn("[WARN] Unable to find free port. Giving up. :-( Last try was: "
+                                << endp->to_string() << ". Endpoint will be ignored!");
+    
+                            port_detection = false;
+                        }
+                        else
+                        {
+                            // try next port
+                            try_port++;
+                        }
+                    }
+                    // BRANCH: explicit given port --> error
+                    else
+                    {
+                        logging_warn("[WARN] Address already in use: "
+                            << endp->to_string() << ". Endpoint will be ignored!");
+                    }
+                }
+    
+                // Rethrow
+                else
+                {
+                    throw;
+                }
+            }
+        } while ( port_detection );
+    }
+    
+    // TODO Bluetooth Endpoints
         #ifdef HAVE_LIBBLUETOOTH
-    foreach(rfcomm_channel_address channel, local.rfcomm) {
-        if (local.bluetooth.size() > 0) {
-                foreach(mac_address mac, local.bluetooth) {
-                        rfcomm::endpoint endp(mac.bluetooth(), channel.value());
-                        create_service(endp);
-                }
-        } else {
-                rfcomm::endpoint endp(channel.value());
-                create_service(endp);
-        }
-    }
+//    foreach(rfcomm_channel_address channel, local.rfcomm) {
+//      if (local.bluetooth.size() > 0) {
+//              foreach(mac_address mac, local.bluetooth) {
+//                      rfcomm::endpoint endp(mac.bluetooth(), channel.value());
+//                      create_service(endp);
+//              }
+//      } else {
+//              rfcomm::endpoint endp(channel.value());
+//              create_service(endp);
+//      }
+//    }
         #endif
-}
-
-void transport_peer::create_service(tcp::endpoint endp) {
-    try {
-        TcpIpPtr tmp_ptr(new tcpip(endp));
-        tcps.push_back(tmp_ptr);
-        logging_info("Listening on IP/TCP " << endp);
-        
-    } catch (boost::system::system_error& e) {
-        if (e.code() == boost::asio::error::address_in_use) {
-            logging_warn("[WARN] Address already in use: "
-                    << endp << ". Endpoint will be ignored!");
-        } else {
-            // Rethrow
-            throw;
-        }
-    }
-}
+    
+    return local;
+}
+
+//void transport_peer::create_service(tcp::endpoint endp) {
+//    try {
+//        TransportProtocolPtr tmp_ptr(new StreamTransport<tcp>(endp));
+//        tcps.push_back(tmp_ptr);
+//        logging_info("Listening on IP/TCP " << endp);
+//        
+//    } catch (boost::system::system_error& e) {
+//        if (e.code() == boost::asio::error::address_in_use) {
+//            logging_warn("[WARN] Address already in use: "
+//                    << endp << ". Endpoint will be ignored!");
+//        } else {
+//            // Rethrow
+//            throw;
+//        }
+//    }
+//}
 
 #ifdef HAVE_LIBBLUETOOTH
-void transport_peer::create_service(rfcomm::endpoint endp) {
-    try {
-        rfcomm_transport::sptr tmp_ptr(new rfcomm_transport(endp));
-        rfcomms.push_back(tmp_ptr);
-        logging_info("Listening on bluetooth/RFCOMM " << endp);
-        
-    } catch (boost::system::system_error& e) {
-        if (e.code() == boost::asio::error::address_in_use) {
-            logging_warn("[WARN] Address already in use: "
-                    << endp << ". Endpoint will be ignored!");
-        } else {
-            // Rethrow
-            throw;
-        }
-    }
-}
+//void transport_peer::create_service(rfcomm::endpoint endp) {
+//    try {
+//        TransportProtocolPtr tmp_ptr(new StreamTransport<rfcomm>(endp));
+//        rfcomms.push_back(tmp_ptr);
+//        logging_info("Listening on bluetooth/RFCOMM " << endp);
+//        
+//    } catch (boost::system::system_error& e) {
+//        if (e.code() == boost::asio::error::address_in_use) {
+//            logging_warn("[WARN] Address already in use: "
+//                    << endp << ". Endpoint will be ignored!");
+//        } else {
+//            // Rethrow
+//            throw;
+//        }
+//    }
+//}
 #endif
 
@@ -101 +159 @@
 }
 
-void transport_peer::start() {
-    foreach(TcpIpPtr tcp, tcps) {
-        tcp->start();
-    }
-    
-#ifdef HAVE_LIBBLUETOOTH
-    foreach(rfcomm_transport::sptr x, rfcomms) {
-        x->start();
-    }
-#endif
-}
-
-void transport_peer::stop() {
-    foreach(TcpIpPtr tcp, tcps) {
-        tcp->stop();
-    }
-    
-#ifdef HAVE_LIBBLUETOOTH
-        foreach(rfcomm_transport::sptr x, rfcomms) {
-                x->stop();
-        }
-#endif
+void transport_peer::start()
+{
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->start();
+    }
+}
+
+void transport_peer::stop()
+{
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->stop();
+    }
 }
 
 
 void transport_peer::send(
-        const endpoint_set& endpoints,
+        const const_EndpointSetPtr endpoints,
         reboost::message_t message,
         uint8_t priority)
 {
-    foreach(TcpIpPtr tcp, tcps) {
-        tcp->send(endpoints, message, priority);
-    }
-    
-#ifdef HAVE_LIBBLUETOOTH
-    foreach(rfcomm_transport::sptr x, rfcomms) {
-                x->send(endpoints, message, priority);
-        }
-#endif
-}
-
-void transport_peer::terminate( const address_v* remote ) {
-        if (remote->instanceof<tcpip_endpoint>())// TODO direkt auf der richtigen verbindung
-        {
-            foreach(TcpIpPtr tcp, tcps) {
-                tcp->terminate(remote);
-            }
-        }
-#ifdef HAVE_LIBBLUETOOTH
-        if (remote->instanceof<rfcomm_endpoint>()) {
-                foreach(rfcomm_transport::sptr x, rfcomms) {
-                        x->terminate(remote);
-                }
-        }
-#endif
-}
-
-void transport_peer::register_listener( transport_listener* listener ) {
-    foreach(TcpIpPtr tcp, tcps) {
-        tcp->register_listener(listener);
-    }
-    
-#ifdef HAVE_LIBBLUETOOTH
-    foreach(rfcomm_transport::sptr x, rfcomms) {
-        x->register_listener(listener);
-    }
-#endif
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->send(endpoints, message, priority);
+    }
+}
+
+// XXX DEPRECATED
+//void transport_peer::terminate( const address_v* remote ) {
+//      if (remote->instanceof<tcpip_endpoint>())// TODO direkt auf der richtigen verbindung
+//      {
+//          foreach(TransportProtocolPtr tcp, tcps) {
+//              tcp->terminate(remote);
+//          }
+//      }
+//#ifdef HAVE_LIBBLUETOOTH
+//      if (remote->instanceof<rfcomm_endpoint>()) {
+//              foreach(TransportProtocolPtr x, rfcomms) {
+//                      x->terminate(remote);
+//              }
+//      }
+//#endif
+//}
+
+void transport_peer::register_listener( transport_listener* listener )
+{
+    BOOST_FOREACH(TransportProtocolPtr stream, transport_streams)
+    {
+        stream->register_listener(listener);
+    }
 }