Changeset 10653 for source/ariba/utility/transport/tcpip/tcpip.cpp

Timestamp:

Jul 25, 2012, 11:41:36 AM (12 years ago)

Author:

Michael Tänzer

Message:

Merge the ASIO branch back into trunk

Location:

source/ariba/utility/transport

Files:

• . (modified) (1 prop)
• tcpip/tcpip.cpp (modified) (1 diff)

source/ariba/utility/transport/tcpip/tcpip.cpp

@@ -1 +1 @@
 #include "tcpip.hpp"
 
-#define _NO_LOGGING
-
-// std includes
-#include <unistd.h>
-#include <iostream>
-#include <string>
-#include <sstream>
-#include <boost/foreach.hpp>
-
-// protlib includes
-#include "protlib/network_message.h"
-#include "protlib/tp_over_tcp.h"
-#include "protlib/tperror.h"
-#include "protlib/logfile.h"
-#include "protlib/queuemanager.h"
-#include "protlib/threadsafe_db.h"
-#include "protlib/setuid.h"
-
-// protlib namespaces
-using namespace protlib;
-using namespace protlib::log;
-
-logfile commonlog;
-protlib::log::logfile& protlib::log::DefaultLog(commonlog);
+#include <boost/array.hpp>
+
+// interface discovery for link-local destinations
+#include <ifaddrs.h>
 
 namespace ariba {
 namespace transport {
 
+use_logging_cpp(tcpip)
+
 using namespace ariba::addressing;
 
-
-tcpip_endpoint convert( const appladdress* addr ) {
-        const char* ip_str = addr->get_ip_str();
-        tcpip_endpoint endpoint( std::string(ip_str), addr->get_port() );
-        return endpoint;
-}
-
-appladdress convert( const tcpip_endpoint& endpoint ) {
-        tcpip_endpoint* e = const_cast<tcpip_endpoint*>(&endpoint);
-        appladdress
-                peer(e->address().to_string().c_str(), "tcp", e->port().asio() );
-//      cout << endpoint.to_string() << " to " << peer.get_ip_str() << ":" << peer.get_port() << endl;
-        return peer;
-}
-
-tcpip::tcpip( uint16_t port ) : 
-        done ( false ),
-        running ( false ),
-        port( port ),
-        tpreceivethread ( NULL ),
-        tpthread ( NULL ),
-        listener ( NULL ) {
-}
-
-tcpip::~tcpip() {
-        if (running) stop();
-}
-
-bool get_message_length( NetMsg& m, uint32& clen_bytes ) {
-        clen_bytes = m.decode32();
-        m.set_pos_r(-4);
-        return true;
-}
-
-void tcpip::start() {
-        done = false;
-        running = false;
-
-        // initalize netdb and setuid
-        protlib::tsdb::init();
-        protlib::setuid::init();
-
-        // set tcp parameters
-        port_t port = this->port; // port
-        TPoverTCPParam tppar(4, get_message_length, port);
-
-        // create receiver thread
-        FastQueue* tpchecker_fq = new FastQueue("TCPTransport", true);
-        QueueManager::instance()->register_queue(tpchecker_fq,
-                        message::qaddr_signaling);
-
-        // start thread
-        pthread_create( &tpreceivethread, NULL, tcpip::receiverThread, this );
-        tpthread = new ThreadStarter<TPoverTCP, TPoverTCPParam> ( 1, tppar );
-        tpthread->start_processing();
-}
-
-void tcpip::stop() {
-        // stop receiver thread
-        done = true;
-
-        // stop TPoverTCP
-        tpthread->stop_processing();
-        tpthread->abort_processing(true);
-        tpthread->wait_until_stopped();
-
-        // unregister TPoverTCP
-        delete QueueManager::instance()->get_queue( message::qaddr_signaling );
-        QueueManager::instance()->unregister_queue( message::qaddr_signaling );
-
-        // destroy QueueManager
-        QueueManager::clear();
-
-        // de-initalize netdb and setuid
-        protlib::setuid::end();
-        protlib::tsdb::end();
-
-        // wait for thread to finish and delete
-        pthread_join(tpreceivethread, NULL);
-}
-
-void tcpip::send( const address_v* remote, const uint8_t* data, size_t size ) {
-
-        // prepare netmsg with length and and port
-        NetMsg* datamsg = new NetMsg(size + 6);
-        datamsg->encode32( size + 2,  true );
-        datamsg->encode16( this->port,true );
-
-        for (size_t i=0; i<size; i++)
-                datamsg->encode8( data[i],true );
-
-        // send message
-        tcpip_endpoint endpoint = *remote;
-        appladdress peer = convert(endpoint);
-
-        // add to output queue
-        tpthread->get_thread_object()->send( datamsg, peer, false );
-}
-
-void tcpip::send( const endpoint_set& endpoints, const uint8_t* data, size_t size ) {
-        // send a message to each combination of ip-address and port
-        BOOST_FOREACH( const ip_address ip, endpoints.ip ) {
-                BOOST_FOREACH( const tcp_port_address port, endpoints.tcp ) {
-                        tcpip_endpoint endpoint(ip,port);
-                        address_vf vf = endpoint;
-                        send(vf,data,size);
-                }
-        }
-}
-
-void tcpip::terminate( const address_v* remote) {
-        tcpip_endpoint endpoint = *remote;
-        appladdress peer = convert(endpoint);
-        peer.convert_to_ipv6();
-        tpthread->get_thread_object()->terminate( peer );
-}
-
-void tcpip::register_listener( transport_listener* listener ) {
-        this->listener = listener;
-}
-
-void* tcpip::receiverThread( void* ptp ) {
-        // get reference to transport object
-        tcpip& tp = *((tcpip*)ptp);
-
-        // get queue
-        FastQueue* fq =
-                QueueManager::instance()->get_queue(message::qaddr_signaling);
-
-        // main processing loop
-        tp.running = true;
-        while (!tp.done) {
-
-                // wait for new message to approach
-                message* msg = fq->dequeue_timedwait(300);
-
-                // message has arrived? no-> continue
-                if (!msg) continue;
-
-                // handle transport message
-                TPMsg* tpmsg = dynamic_cast<TPMsg*> (msg);
-                if (!tpmsg) {
-                        delete msg;
-                        continue;
-                }
-
-                // get address & message
-                const appladdress* remote_peer = static_cast<const appladdress*>( tpmsg->get_peeraddress() );
-                const appladdress* local_peer  = static_cast<const appladdress*>( tpmsg->get_ownaddress() );
-                NetMsg* datamsg = tpmsg->get_message();
-
-                // not a data message? -> continue!
-                if (!datamsg) {
-                        delete tpmsg;
-                        continue;
-                }
-
-                // get length and remote endpoint port
-                datamsg->set_pos(0);
-                uint32_t message_size = datamsg->decode32(true)-2;
-                //uint16_t remote_port = datamsg->decode16(true);
-
-                // inform listener
-                if (tp.listener != NULL) {
-                        tcpip_endpoint remote = convert(remote_peer);
-                        tcpip_endpoint local  = convert(local_peer);
-                        tp.listener->receive_message(
-                                        &tp, local, remote, datamsg->get_buffer()+6, message_size );
-                }
-
-                tpmsg->set_message(NULL);
-                delete datamsg;
-                delete tpmsg;
-        }
-        // clean queue & stop
-        fq->cleanup();
-        tp.running = false;
-        return NULL;
+typedef boost::mutex::scoped_lock unique_lock;
+
+tcpip::tcpip( const tcp::endpoint& endp )  :
+        listener(NULL),
+        acceptor(u_io_service.get_asio_io_service(), endp)
+{
+}
+
+tcpip::~tcpip(){}
+
+void tcpip::start()
+{
+    // open server socket
+    accept();
+    
+    u_io_service.start();
+}
+
+
+void tcpip::stop()
+{
+    acceptor.close();
+    
+    u_io_service.stop();
+}
+
+
+/* see header file for comments */
+void tcpip::send(
+        const tcp::endpoint& dest_addr,
+        reboost::message_t message,
+        uint8_t priority)
+{
+    ConnPtr conn;
+    bool need_to_connect = false;
+    
+    {
+        unique_lock lock(connections_lock);
+        
+        ConnectionMap::iterator it = connections.find(dest_addr);
+        if (it == connections.end())
+        {
+            ConnPtr tmp_ptr(
+                    new tcpip_connection(
+                            u_io_service.get_asio_io_service(),
+                            shared_from_this() )
+                    );
+            conn = tmp_ptr;
+            
+            conn->partner = dest_addr;
+            conn->remote = convert_address(dest_addr);
+            
+            // Note: starting the send is the obligation of the connect_handler
+            // (avoids trying to send while not connected yet)
+            conn->sending =  true;
+            need_to_connect = true;
+            
+            ConnectionMap::value_type item(dest_addr, conn);
+            connections.insert(item);
+            
+        } else {
+            conn = it->second;
+        }
+    }
+    
+    
+    // * the actual send *
+    conn->enqueue_for_sending(message, priority);
+    
+    // if new connection connect to the other party
+    if ( need_to_connect )
+    {
+        conn->sock.async_connect(
+                dest_addr,
+                boost::bind(
+                        &tcpip_connection::async_connect_handler,
+                        conn,
+                        boost::asio::placeholders::error));
+    }
+}
+
+
+/* see header file for comments */
+void tcpip::send(
+        const address_v* remote,
+        reboost::message_t message,
+        uint8_t priority)
+{
+    send(convert_address(remote), message, priority);
+}
+
+
+/* see header file for comments */
+void tcpip::send(
+        const endpoint_set& endpoints,
+        reboost::message_t message,
+        uint8_t priority )
+{
+    // network interfaces scope_ids, for link-local connections (lazy initialization)
+    vector<uint64_t> scope_ids;
+    
+    // send a message to each combination of address-address and port
+    BOOST_FOREACH( const ip_address address, endpoints.ip ) {
+        BOOST_FOREACH( const tcp_port_address port, endpoints.tcp ) {
+            tcp::endpoint endp(address.asio(), port.asio());
+            
+            // special treatment for link local addresses
+            //   ---> send over all (suitable) interfaces
+            if ( endp.address().is_v6() )
+            {
+                boost::asio::ip::address_v6 v6_addr = endp.address().to_v6();
+                
+                if ( v6_addr.is_link_local() )
+                {
+                    // initialize scope_ids
+                    if ( scope_ids.size() == 0 )
+                        scope_ids = get_interface_scope_ids();
+                    
+                    BOOST_FOREACH ( uint64_t id, scope_ids )
+                    {                        
+                        v6_addr.scope_id(id);
+                        endp.address(v6_addr);
+    
+                        logging_debug("------> SEND TO (link-local): " << endp);
+                        // * send *
+                        send(endp, message, priority);
+                    }
+                }
+                
+                continue;
+            }
+            
+            // * send *
+            send(endp, message, priority);
+        }
+    }
+}
+
+
+void tcpip::register_listener( transport_listener* listener )
+{
+    this->listener = listener;
+}
+
+
+void tcpip::terminate( const address_v* remote )
+{
+    terminate(convert_address(remote));
+}
+
+void tcpip::terminate( const tcp::endpoint& remote )
+{
+    ConnPtr conn;
+    
+    // find and forget connection
+    {
+        unique_lock lock(connections_lock);
+        
+        ConnectionMap::iterator it = connections.find(remote);
+        if (it == connections.end())
+        {
+            return;
+        }
+        
+        conn = it->second;
+        
+        connections.erase(it);
+    }
+
+    // close connection
+    boost::system::error_code ec;
+    conn->sock.shutdown(tcp::socket::shutdown_both, ec);
+    conn->sock.close(ec);
+}
+
+
+/* private */
+void tcpip::accept()
+{
+    // create new connection object
+    ConnPtr conn(
+            new tcpip_connection(
+                    u_io_service.get_asio_io_service(),
+                    shared_from_this()
+            )
+    );
+    
+    // wait for incoming connection
+    acceptor.async_accept(
+            conn->sock,
+            boost::bind(&self::async_accept_handler,
+                    this->shared_from_this(),
+                    conn,
+                    boost::asio::placeholders::error)
+    );
+}
+
+void tcpip::async_accept_handler(ConnPtr conn, const error_code& error)
+{
+    if ( ! error )
+    {
+        conn->partner = conn->sock.remote_endpoint();
+        conn->remote = convert_address(conn->partner);
+        conn->local = convert_address(conn->sock.local_endpoint());
+        
+        {
+            unique_lock lock(connections_lock);
+            
+            ConnectionMap::value_type item(conn->sock.remote_endpoint(), conn);
+            connections.insert(item);
+        }
+        
+        // read
+        conn->listen();
+    }
+    
+    // accept further connections
+    accept();
+}
+
+inline tcp::endpoint tcpip::convert_address( const address_v* address )
+{
+    tcpip_endpoint endpoint = *address;
+    
+    return tcp::endpoint(
+        endpoint.address().asio(), endpoint.port().value()
+    );
+}
+
+
+inline tcpip_endpoint tcpip::convert_address(const tcp::endpoint& endpoint)
+{
+    ip_address address;
+    address.asio(endpoint.address());
+    tcp_port_address port;
+    port.value(endpoint.port());
+    return tcpip_endpoint(address, port);
+}
+
+
+vector<uint64_t> tcpip::get_interface_scope_ids()
+{
+    vector<uint64_t> ret;
+    
+    struct ifaddrs* ifaceBuffer = NULL;
+    void*           tmpAddrPtr  = NULL;
+    
+    int ok = getifaddrs( &ifaceBuffer );
+    if( ok != 0 ) return ret;
+
+    for( struct ifaddrs* i=ifaceBuffer; i != NULL; i=i->ifa_next ) {
+
+        // ignore devices that are disabled or have no ip
+        if(i == NULL) continue;
+        struct sockaddr* addr = i->ifa_addr;
+        if (addr==NULL) continue;
+
+        // only use ethX and wlanX devices
+        string device = string(i->ifa_name);
+        if ( (device.find("eth") == string::npos) &&
+              (device.find("wlan")  == string::npos) /* &&
+              (device.find("lo")  == string::npos) XXX */ )
+        {
+            continue;
+        }
+
+        // only use interfaces with ipv6 link-local addresses 
+        if (addr->sa_family == AF_INET6)
+        {
+            // convert address
+            // TODO should be possible without detour over strings
+            char straddr[INET6_ADDRSTRLEN];
+            tmpAddrPtr= &((struct sockaddr_in6*)addr)->sin6_addr;
+            inet_ntop( i->ifa_addr->sa_family, tmpAddrPtr, straddr, sizeof(straddr) );
+
+            address_v6 v6addr = address_v6::from_string(straddr);
+            if ( v6addr.is_link_local() )
+            {
+                // * append the scope_id to the return vector *
+                ret.push_back(if_nametoindex(i->ifa_name));
+            }
+
+        }
+    }
+
+    freeifaddrs(ifaceBuffer);
+    
+    return ret;
+}
+
+
+/*****************
+ ** inner class **
+ *****************/
+
+tcpip::tcpip_connection::tcpip_connection(boost::asio::io_service & io_service, TcpIpPtr parent)  :
+        sock(io_service),
+        valid(true),
+        parent(parent),
+        out_queues(8), //TODO How much priorities shall we have?
+        sending(false)
+{
+        header.length = 0;
+        header.prot = 0;
+}
+
+/*-------------------------------------------
+ | implement transport_connection interface |
+ -------------------------------------------*/
+void tcpip::tcpip_connection::send(
+        reboost::message_t message,
+        uint8_t priority)
+{
+    enqueue_for_sending(message, priority);
+}
+
+
+address_vf tcpip::tcpip_connection::getLocalEndpoint()
+{
+    return local;
+}
+
+
+address_vf tcpip::tcpip_connection::getRemoteEndpoint()
+{
+    return remote;
+}
+
+
+void tcpip::tcpip_connection::terminate()
+{
+    parent->terminate(partner);
+}
+
+
+/*------------------------------
+ | things we defined ourselves |
+ ------------------------------*/
+void tcpip::tcpip_connection::async_connect_handler(const error_code& error)
+{
+    if (error)
+    {
+        parent->terminate(partner);
+
+        return;
+    }
+    
+    // save address in ariba format
+    local = parent->convert_address(sock.local_endpoint());
+    
+    // Note: sending has to be true at this point
+    send_next_package();
+    
+    listen();
+}
+
+
+void tcpip::tcpip_connection::listen()
+{
+    boost::asio::async_read(
+            this->sock,
+            boost::asio::mutable_buffers_1(&this->header, sizeof(header_t)),
+            boost::bind(
+                    &tcpip::tcpip_connection::async_read_header_handler,
+                    this->shared_from_this(),
+                    boost::asio::placeholders::error,
+                    boost::asio::placeholders::bytes_transferred
+            )
+    );
+}
+
+
+void tcpip::tcpip_connection::async_read_header_handler(const error_code& error, size_t bytes_transferred)
+{
+    if (error)
+    {
+        parent->terminate(partner);
+
+        return;
+    }
+
+    // convert byte order
+    header.length = ntohl(header.length);
+    header.length -= 2;  // XXX protlib
+    
+    assert(header.length > 0);
+    
+    // new buffer for the new packet
+    buffy = shared_buffer_t(header.length);
+
+    // * read data *
+    boost::asio::async_read(
+            this->sock,
+            boost::asio::buffer(buffy.mutable_data(), buffy.size()),
+            boost::bind(
+                    &tcpip::tcpip_connection::async_read_data_handler,
+                    this->shared_from_this(),
+                    boost::asio::placeholders::error,
+                    boost::asio::placeholders::bytes_transferred
+            )
+    );
+}
+
+void tcpip::tcpip_connection::async_read_data_handler(
+        const error_code& error, size_t bytes_transferred)
+{
+    if (error)
+    {
+        parent->terminate(partner);
+
+        return;
+    }
+    
+    message_t msg;
+    msg.push_back(buffy);
+    buffy = shared_buffer_t();
+
+    if ( parent->listener )
+        parent->listener->receive_message(shared_from_this(), msg);
+    
+    listen();
+}
+
+/* see header file for comments */
+void tcpip::tcpip_connection::async_write_handler(reboost::shared_buffer_t packet, const error_code& error, size_t bytes_transferred)
+{
+    if ( error )
+    {        
+        // remove this connection
+        parent->terminate(partner); 
+
+        return;
+    }
+    
+    send_next_package();
+}
+
+
+
+void tcpip::tcpip_connection::enqueue_for_sending(Packet packet, uint8_t priority)
+{
+    bool restart_sending = false;
+    
+    // enqueue packet  [locked]
+    {
+        unique_lock(out_queues_lock);
+        
+        assert( priority < out_queues.size() );
+        out_queues[priority].push(packet);
+        
+        if ( ! sending )
+        {
+            restart_sending = true;
+            sending = true;
+        }
+    }
+    
+    // if sending was stopped, we have to restart it here
+    if ( restart_sending )
+    {
+        send_next_package();
+    }
+}
+
+/* see header file for comments */
+void tcpip::tcpip_connection::send_next_package()
+{
+    Packet packet;
+    bool found = false;
+
+    // find packet with highest priority  [locked]
+    {
+        unique_lock(out_queues_lock);
+        
+        for ( vector<OutQueue>::iterator it = out_queues.begin();
+                it != out_queues.end(); it++ )
+        {
+            if ( !it->empty() )
+            {
+                packet = it->front();
+                it->pop();
+                found = true;
+                
+                break;
+            }
+        }
+        
+        // no packets waiting --> stop sending
+        if ( ! found )
+        {
+            sending = false;
+        }
+    }
+    
+    // * send *
+    if ( found )
+    {
+        reboost::shared_buffer_t header_buf(sizeof(header_t));
+        header_t* header = (header_t*)(header_buf.mutable_data());
+        header->length = htonl(packet.size()+2);  // XXX protlib
+        
+        packet.push_front(header_buf);
+        
+        // "convert" message to asio buffer sequence
+        vector<boost::asio::const_buffer> send_sequence(packet.length());
+        for ( int i=0; i < packet.length(); i++ )
+        {
+            shared_buffer_t b = packet.at(i);
+            send_sequence.push_back(boost::asio::buffer(b.data(), b.size()));
+        }
+        
+        // * async write *
+        boost::asio::async_write(
+                this->sock,
+                send_sequence,
+                boost::bind(
+                        &tcpip::tcpip_connection::async_write_handler,
+                        this->shared_from_this(),
+                        packet,  // makes sure our shared pointer lives long enough ;-)
+                        boost::asio::placeholders::error,
+                        boost::asio::placeholders::bytes_transferred)
+        );
+    }
 }