moved time duration measurement of ping from scheduler to pingers

[pingcheck] / src / tcp / tcppinger.cpp

/*
The software in this package is distributed under the GNU General
Public License version 2 (with a special exception described below).

A copy of GNU General Public License (GPL) is included in this distribution,
in the file COPYING.GPL.

As a special exception, if other files instantiate templates or use macros
or inline functions from this file, or you compile this file and link it
with other works to produce a work based on this file, this file
does not by itself cause the resulting work to be covered
by the GNU General Public License.

However the source code for this file must still be made available
in accordance with section (3) of the GNU General Public License.

This exception does not invalidate any other reasons why a work based
on this file might be covered by the GNU General Public License.
*/
#include "tcp/tcppinger.h"

#include <errno.h>
#include <net/if.h>
#include <sys/ioctl.h>
#include <sys/socket.h>

#include <istream>
#include <ostream>
#include <limits>

#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/date_time/posix_time/posix_time_types.hpp>
#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_generators.hpp>

#include <logfunc.hpp>

#include "boost_assert_handler.h"
#include "tcp/tcpsegmentfactory.h"

using namespace std;
using boost::asio::const_buffers_1;
using boost::asio::io_service;
using boost::asio::ip::address;
using boost::asio::ip::tcp_raw_protocol;
using boost::date_time::time_resolution_traits_adapted64_impl;
using boost::function;
using boost::posix_time::microsec_clock;
using boost::posix_time::ptime;
using boost::posix_time::seconds;
using boost::shared_ptr;
using I2n::Logger::GlobalLogger;

//-----------------------------------------------------------------------------
// TcpPinger
//-----------------------------------------------------------------------------

/**
 * @brief Parameterized constructor.
 *
 * @param io_serv The one @c io_service object that controls async processing
 * @param protocol The network layer protocol to use.
 * @param source_network_interface The network interface name from where to
 * send the segments.
 * @param echo_reply_timeout_in_sec The amount of time to wait for a reply.
 */
TcpPinger::TcpPinger(
        const IoServiceItem io_serv,
        const tcp_raw_protocol::socket::protocol_type &protocol,
        const string &source_network_interface_name,
        const int rst_reply_timeout_in_sec
) :
    DestinationEndpoint(),
    Protocol( protocol ),
    Socket( *io_serv, Protocol ),
    NetInterface( source_network_interface_name, Socket ),
    TcpSegmentReceiveTimer( *io_serv ),
    Identifier( 0 ),
    SequenceNumber( 0 ),
    TimeSent( microsec_clock::universal_time() ),
    ReplyBuffer(),
    ReceivedReply( false ),
    RstReplyTimeoutInSec( rst_reply_timeout_in_sec ),
    PingerStatus( PingStatus_NotSent ),
    PingDoneCallback()
{
    if ( !NetInterface.bind() )
    {
        GlobalLogger.error() << "Could not bind the socket with the local interface. "
                << ::strerror( errno )  << endl;
    }

    // Create "unique" identifier
    boost::uuids::random_generator random_gen;
    boost::uuids::uuid random_tag = random_gen();

    BOOST_ASSERT( sizeof(Identifier) <= random_tag.size() );
    memcpy( &Identifier, random_tag.data, sizeof(Identifier) );
}

TcpPinger::~TcpPinger()
{
}

/**
 * @brief Ping a destination address from an available local source.
 *
 * @param destination_ip The address of the host to ping.
 * @param destination_port The port at the destination host to ping.
 * @param done_handler Done handler will be called on successful ping or timeout.
 *
 * @return void.
 */
void TcpPinger::ping(
        const address &destination_ip,
        const uint16_t destination_port,
        function<void(PingStatus,long)> ping_done_callback
)
{
    BOOST_ASSERT( ( 0 < destination_port ) && ( destination_port < numeric_limits<uint16_t>::max() ) );

    PingDoneCallback = ping_done_callback;

    // Prepare ping
    set_ping_status( PingStatus_NotSent );

    set_destination_endpoint( destination_ip, destination_port );

    start_send();
    start_receive();
}

void TcpPinger::stop_pinging()
{
}

address TcpPinger::get_source_address() const
{
    return NetInterface.get_address( Protocol );
}

address TcpPinger::get_destination_address() const
{
    return DestinationEndpoint.address();
}

uint16_t TcpPinger::get_source_port() const
{
    uint16_t source_port = static_cast<uint16_t>( ( random() % 16383u ) + 49152u ); // same as random() % 65536;

    return source_port;
}

uint16_t TcpPinger::get_destination_port() const
{
    return DestinationEndpoint.port();
}

void TcpPinger::set_destination_endpoint(
        const address &destination_ip,
        const uint16_t destination_port
)
{
    BOOST_ASSERT( ( 0 < destination_port ) && ( destination_port < numeric_limits<uint16_t>::max() ) );
    BOOST_ASSERT( sizeof(uint16_t) <= sizeof(destination_port) );

    uint16_t port = static_cast<uint16_t>( destination_port );
    DestinationEndpoint = tcp_raw_protocol::endpoint( destination_ip, port );
}

void TcpPinger::start_send()
{
    ++SequenceNumber;

    // Create a TCP header for a SYN request.
    address source_address = get_source_address();
    address destination_address = get_destination_address();
    uint16_t source_port = get_source_port();
    uint16_t destination_port = get_destination_port();
    TcpSegmentItem tcp_segment = TcpSegmentFactory::create_tcp_segment_syn_request(
            Protocol,
            source_address, destination_address,
            source_port, destination_port, SequenceNumber );

    send_ack_request( tcp_segment );
}

void TcpPinger::send_ack_request( const TcpSegmentItem tcp_segment )
{
    // Encode the request packet.
    boost::asio::streambuf request_buffer;
    ostream os( &request_buffer );
    if ( !tcp_segment->write( os ) )
    {
        GlobalLogger.error() << "Fail writing ping data." << endl;
    }

    TimeSent = microsec_clock::universal_time();

    address dest_address = DestinationEndpoint.address();
    string dest_address_string = dest_address.to_string();
    BOOST_ASSERT( !dest_address_string.empty() );

    // Send the request
    try
    {
        const_buffers_1 data = request_buffer.data();
        // Block until send the data
        size_t bytes_sent = Socket.send_to( data, DestinationEndpoint );
        if ( bytes_sent != buffer_size( data ) )
        {
            GlobalLogger.error() << "Fail sending ping data."
                    << "Amount of bytes sent differs from the buffer." << endl;
        }
    }
    catch ( const exception &ex )
    {
        GlobalLogger.error() << "Fail sending ping data. " << ex.what() << endl;
    }

    // Tell how long to wait for the reply
    schedule_timeout_rst_reply();
}

void TcpPinger::schedule_timeout_rst_reply()
{
    // Wait up to N seconds for a reply.
    ReceivedReply = false;
    (void) TcpSegmentReceiveTimer.expires_at(
            TimeSent + seconds( RstReplyTimeoutInSec )
    );
    TcpSegmentReceiveTimer.async_wait(
            boost::bind( &TcpPinger::handle_ping_done, this )
    );
}

/**
 * @brief Gets called when the ping is finished: Either on timeout or on ping reply
 *
 * @return void
 **/
void TcpPinger::handle_ping_done()
{
    // Check ReceivedReply as the timer handler
    // is also called by Timer.cancel();
    if ( !ReceivedReply )
    {
        GlobalLogger.info() << "Request timed out" << endl;

        set_ping_status( PingStatus_FailureTimeout );
    }

    // Call ping-done handler
    PingDoneCallback( PingerStatus,
           (microsec_clock::universal_time() - TimeSent).total_microseconds() );
}

void TcpPinger::start_receive()
{
    // Discard any data already in the buffer.
    ReplyBuffer.consume( ReplyBuffer.size() );

    // Waiting for a reply. We prepare the buffer to receive up to 64KB.
    Socket.async_receive(
            ReplyBuffer.prepare( 65536 ),
            boost::bind( &TcpPinger::handle_receive_tcp_segment, this, _2 )
    );
}

void TcpPinger::handle_receive_tcp_segment( const size_t &bytes_transferred )
{
    // The actual number of bytes received is committed to the buffer so that we
    // can extract it using a std::istream object.
    ReplyBuffer.commit( bytes_transferred );

    try
    {
        istream is( &ReplyBuffer );
        if ( !is )
        {
            GlobalLogger.error() << "Can't handle ReplyBuffer" << endl;
            return;
        }

        // Decode the reply segment.
        TcpSegmentItem tcp_segment = TcpSegmentFactory::create_tcp_segment( Protocol, is );

        // The TCP SYN ping will receive a RST if the port is closed.
        // Filter out the TCP reset (RST) replies. Note that the sequence
        // number from RST does not match the sent SYN's sequence number.
        if ( tcp_segment->match_rst_reply( DestinationEndpoint.address() ) )
        {
            ReceivedReply = true;

            tcp_segment->print_rst_reply( TimeSent );

            set_ping_status( PingStatus_SuccessReply );

            TcpSegmentReceiveTimer.cancel();                                                                                                                //lint !e534
        }
        // The TCP SYN ping will receive a SYN/ACK if the port is open.
        // Filter out the TCP synchronize (SYN/ACK) replies. Note that the sequence
        // number from SYN/ACK does not match the sent SYN's sequence number.
        else if ( tcp_segment->match_syn_ack_reply( DestinationEndpoint.address() ) )
        {
            ReceivedReply = true;

            tcp_segment->print_syn_ack_reply( TimeSent );

            set_ping_status( PingStatus_SuccessReply );

            TcpSegmentReceiveTimer.cancel();                                                                                                                //lint !e534
        }
        // Unknown TCP reply, start another receive till timeout
        else
        {
            start_receive();
        }
    }
    catch ( ... )
    {
        GlobalLogger.notice() << "Exception during ICMP parse. "
            << "Starting another receive till timeout." << endl;
        start_receive();
    }
}

void TcpPinger::set_ping_status( PingStatus ping_status )
{
    PingerStatus = ping_status;
}

PingerItem TcpPinger::create(
        const IoServiceItem io_serv,
        const tcp_raw_protocol::socket::protocol_type &protocol,
        const string &source_network_interface_name,
        const int rst_reply_timeout_in_sec )
{
    TcpPinger *ptr = new TcpPinger(io_serv, protocol, source_network_interface_name,
                                           rst_reply_timeout_in_sec);
    PingerItem shared_ptr(ptr);
    Pinger::WeakPtr weak_ptr( shared_ptr );

    // keep weak pointer to self
    ptr->set_myself( weak_ptr );
    // shared_ptr->set_myself( weak_ptr ); Error: Pinger::set_myself is protected

    // done, return shared ptr
    return shared_ptr;
}