D:/opendownloadmanager/ODM-1.x/Bittorrent/libtorrent-0.12/src/kademlia/rpc_manager.cpp

Go to the documentation of this file.

/*
  Free Download Manager Copyright (c) 2003-2007 FreeDownloadManager.ORG
*/  


 

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

#include <libtorrent/io.hpp>
#include <libtorrent/invariant_check.hpp>
#include <libtorrent/kademlia/rpc_manager.hpp>
#include <libtorrent/kademlia/logging.hpp>
#include <libtorrent/kademlia/routing_table.hpp>
#include <libtorrent/hasher.hpp>

#include <fstream>

using boost::posix_time::ptime;
using boost::posix_time::time_duration;
using boost::posix_time::microsec_clock;
using boost::posix_time::seconds;
using boost::posix_time::milliseconds;
using boost::shared_ptr;
using boost::bind;

namespace libtorrent { namespace dht
{

namespace io = libtorrent::detail;

#ifdef TORRENT_DHT_VERBOSE_LOGGING
TORRENT_DEFINE_LOG(rpc)
#endif

node_id generate_id();

rpc_manager::rpc_manager(fun const& f, node_id const& our_id
        , routing_table& table, send_fun const& sf)
        : m_next_transaction_id(rand() % max_transactions)
        , m_oldest_transaction_id(m_next_transaction_id)
        , m_incoming(f)
        , m_send(sf)
        , m_our_id(our_id)
        , m_table(table)
        , m_timer(boost::posix_time::microsec_clock::universal_time())
        , m_random_number(generate_id())
        , m_destructing(false)
{
        std::srand(time(0));
}

rpc_manager::~rpc_manager()
{
        m_destructing = true;
#ifdef TORRENT_DHT_VERBOSE_LOGGING
        TORRENT_LOG(rpc) << "Destructing";
#endif
        std::for_each(m_aborted_transactions.begin(), m_aborted_transactions.end()
                , bind(&observer::abort, _1));
        
        for (transactions_t::iterator i = m_transactions.begin()
                , end(m_transactions.end()); i != end; ++i)
        {
                if (*i) (*i)->abort();
        }
}

#ifndef NDEBUG
void rpc_manager::check_invariant() const
{
        assert(m_oldest_transaction_id >= 0);
        assert(m_oldest_transaction_id < max_transactions);
        assert(m_next_transaction_id >= 0);
        assert(m_next_transaction_id < max_transactions);
        assert(!m_transactions[m_next_transaction_id]);

        for (int i = (m_next_transaction_id + 1) % max_transactions;
                i != m_oldest_transaction_id; i = (i + 1) % max_transactions)
        {
                assert(!m_transactions[i]);
        }
}
#endif

bool rpc_manager::incoming(msg const& m)
{
        INVARIANT_CHECK;

        if (m_destructing) return false;

        if (m.reply)
        {
                
                

                if (m.transaction_id.size() != 2)
                {
#ifdef TORRENT_DHT_VERBOSE_LOGGING
                        TORRENT_LOG(rpc) << "Reply with invalid transaction id size: " 
                                << m.transaction_id.size() << " from " << m.addr;
#endif
                        return false;
                }
        
                std::string::const_iterator i = m.transaction_id.begin();       
                int tid = io::read_uint16(i);

                if (tid >= (int)m_transactions.size()
                        || tid < 0)
                {
#ifdef TORRENT_DHT_VERBOSE_LOGGING
                        TORRENT_LOG(rpc) << "Reply with unknown transaction id: " 
                                << tid << " from " << m.addr;
#endif
                        return false;
                }
                
                boost::shared_ptr<observer> o = m_transactions[tid];

                if (!o)
                {
#ifdef TORRENT_DHT_VERBOSE_LOGGING
                        TORRENT_LOG(rpc) << "Reply with unknown transaction id: " 
                                << tid << " from " << m.addr;
#endif
                        return false;
                }
                
                if (m.addr != o->target_addr)
                {
#ifdef TORRENT_DHT_VERBOSE_LOGGING
                        TORRENT_LOG(rpc) << "Reply with incorrect address and valid transaction id: " 
                                << tid << " from " << m.addr;
#endif
                        return false;
                }

#ifdef TORRENT_DHT_VERBOSE_LOGGING
                std::ofstream reply_stats("libtorrent_logs/round_trip_ms.log", std::ios::app);
                reply_stats << m.addr << "\t" << (microsec_clock::universal_time()
                        - o->sent).total_milliseconds() << std::endl;
#endif
                o->reply(m);
                m_transactions[tid].reset();
                
                if (m.piggy_backed_ping)
                {
                        
                        
                        msg ph;
                        ph.message_id = messages::ping;
                        ph.transaction_id = m.ping_transaction_id;
                        ph.id = m_our_id;
                        ph.addr = m.addr;

                        msg empty;
                        
                        reply(empty, ph);
                }
                return m_table.node_seen(m.id, m.addr);
        }
        else
        {
                
                m_incoming(m);
        }
        return false;
}

time_duration rpc_manager::tick()
{
        INVARIANT_CHECK;

        using boost::posix_time::microsec_clock;

        const int timeout_ms = 10 * 1000;

        

        if (m_next_transaction_id == m_oldest_transaction_id) return milliseconds(timeout_ms);

        std::vector<shared_ptr<observer> > timeouts;

        for (;m_next_transaction_id != m_oldest_transaction_id;
                m_oldest_transaction_id = (m_oldest_transaction_id + 1) % max_transactions)
        {
                assert(m_oldest_transaction_id >= 0);
                assert(m_oldest_transaction_id < max_transactions);

                boost::shared_ptr<observer> o = m_transactions[m_oldest_transaction_id];
                if (!o) continue;

                time_duration diff = o->sent + milliseconds(timeout_ms)
                        - microsec_clock::universal_time();
                if (diff > seconds(0))
                {
                        if (diff < seconds(1)) return seconds(1);
                        return diff;
                }
                
                try
                {
                        m_transactions[m_oldest_transaction_id].reset();
                        timeouts.push_back(o);
                } catch (std::exception) {}
        }
        
        std::for_each(timeouts.begin(), timeouts.end(), bind(&observer::timeout, _1));
        timeouts.clear();
        
        
        
        
        std::vector<shared_ptr<observer> >().swap(m_aborted_transactions);
        return milliseconds(timeout_ms);
}

unsigned int rpc_manager::new_transaction_id(shared_ptr<observer> o)
{
        INVARIANT_CHECK;

        unsigned int tid = m_next_transaction_id;
        m_next_transaction_id = (m_next_transaction_id + 1) % max_transactions;
        if (m_transactions[m_next_transaction_id])
        {
                
                
                
                m_aborted_transactions.push_back(m_transactions[m_next_transaction_id]);
                m_transactions[m_next_transaction_id].reset();
                assert(m_oldest_transaction_id == m_next_transaction_id);
        }
        assert(!m_transactions[tid]);
        m_transactions[tid] = o;
        if (m_oldest_transaction_id == m_next_transaction_id)
        {
                m_oldest_transaction_id = (m_oldest_transaction_id + 1) % max_transactions;
#ifdef TORRENT_DHT_VERBOSE_LOGGING
                TORRENT_LOG(rpc) << "WARNING: transaction limit reached! Too many concurrent"
                        " messages! limit: " << (int)max_transactions;
#endif
                update_oldest_transaction_id();
        }

        return tid;
}

void rpc_manager::update_oldest_transaction_id()
{
        INVARIANT_CHECK;

        assert(m_oldest_transaction_id != m_next_transaction_id);
        while (!m_transactions[m_oldest_transaction_id])
        {
                m_oldest_transaction_id = (m_oldest_transaction_id + 1)
                        % max_transactions;
                if (m_oldest_transaction_id == m_next_transaction_id)
                        break;
        }
}

void rpc_manager::invoke(int message_id, udp::endpoint target_addr
        , shared_ptr<observer> o)
{
        INVARIANT_CHECK;

        if (m_destructing)
        {
                o->abort();
                return;
        }

        msg m;
        m.message_id = message_id;
        m.reply = false;
        m.id = m_our_id;
        m.addr = target_addr;
        assert(!m_transactions[m_next_transaction_id]);
#ifndef NDEBUG
        int potential_new_id = m_next_transaction_id;
#endif
        try
        {
                m.transaction_id.clear();
                std::back_insert_iterator<std::string> out(m.transaction_id);
                io::write_uint16(m_next_transaction_id, out);
                
                o->send(m);

                o->sent = boost::posix_time::microsec_clock::universal_time();
                o->target_addr = target_addr;

        #ifdef TORRENT_DHT_VERBOSE_LOGGING
                TORRENT_LOG(rpc) << "Invoking " << messages::ids[message_id] 
                        << " -> " << target_addr;
        #endif  
                m_send(m);
                new_transaction_id(o);
        }
        catch (std::exception& e)
        {
                
                assert(potential_new_id == m_next_transaction_id);
                o->abort();
        }
}

void rpc_manager::reply(msg& m, msg const& reply_to)
{
        INVARIANT_CHECK;

        if (m_destructing) return;

        if (m.message_id != messages::error)
                m.message_id = reply_to.message_id;
        m.addr = reply_to.addr;
        m.reply = true;
        m.piggy_backed_ping = false;
        m.id = m_our_id;
        m.transaction_id = reply_to.transaction_id;
        
        m_send(m);
}

namespace
{
        struct dummy_observer : observer
        {
                virtual void reply(msg const&) {}
                virtual void timeout() {}
                virtual void send(msg&) {}
                void abort() {}
        };
}

void rpc_manager::reply_with_ping(msg& m, msg const& reply_to)
{
        INVARIANT_CHECK;

        if (m_destructing) return;

        if (m.message_id != messages::error)
                m.message_id = reply_to.message_id;
        m.addr = reply_to.addr;
        m.reply = true;
        m.piggy_backed_ping = true;
        m.id = m_our_id;
        m.transaction_id = reply_to.transaction_id;

        try
        {
                m.ping_transaction_id.clear();
                std::back_insert_iterator<std::string> out(m.ping_transaction_id);
                io::write_uint16(m_next_transaction_id, out);

                boost::shared_ptr<observer> o(new dummy_observer);
                assert(!m_transactions[m_next_transaction_id]);
                o->sent = boost::posix_time::microsec_clock::universal_time();
                o->target_addr = m.addr;
                
                m_send(m);
                new_transaction_id(o);
        }
        catch (std::exception& e)
        {
                
        }
}  

} } 

---