kademlia.c

P2P模拟器

// {{{ headers
/*
 * Copyright (c) 2003-2005 Thomer M. Gil
 *                    Massachusetts Institute of Technology
 *
 * Permission is hereby granted, free of charge, to any person obtaining
 * a copy of this software and associated documentation files (the
 * "Software"), to deal in the Software without restriction, including
 * without limitation the rights to use, copy, modify, merge, publish,
 * distribute, sublicense, and/or sell copies of the Software, and to
 * permit persons to whom the Software is furnished to do so, subject to
 * the following conditions:
 *
 * The above copyright notice and this permission notice shall be
 * included in all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
 * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
 * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 */

#include "kademlia.h"
#include "p2psim/network.h"
#include "p2psim/threadmanager.h"
#include <deque>
#include <iostream>
using namespace std;

Kademlia::use_replacement_cache_t Kademlia::use_replacement_cache = FULL;
unsigned Kademlia::debugcounter = 1;
unsigned Kademlia::_nkademlias = 0;
Kademlia::k_nodeinfo_pool *Kademlia::pool = 0;

bool Kademlia::docheckrep = false;

unsigned Kademlia::k = 0;
unsigned Kademlia::k_tell = 0;
unsigned Kademlia::alpha = 0;
unsigned Kademlia::stabilize_timer = 0;
unsigned Kademlia::refresh_rate = 0;
unsigned Kademlia::erase_count = 0;
Time Kademlia::max_lookup_time = 0;
bool Kademlia::learn_stabilize_only = false;
bool Kademlia::force_stabilization = false;
bool Kademlia::death_notification = false;

long long unsigned Kademlia::_rpc_bytes = 0;
long unsigned Kademlia::_good_rpcs = 0;
long unsigned Kademlia::_bad_rpcs = 0;
long unsigned Kademlia::_ok_by_reaper = 0;
long unsigned Kademlia::_timeouts_by_reaper = 0;

long unsigned Kademlia::_good_lookups = 0;
long unsigned Kademlia::_good_attempts = 0;
long unsigned Kademlia::_bad_attempts = 0;
long unsigned Kademlia::_lookup_dead_node = 0;
long unsigned Kademlia::_ok_failures = 0;
long unsigned Kademlia::_bad_failures = 0;

Time Kademlia::_good_total_latency = 0;
Time Kademlia::_good_lookup_latency = 0;
Time Kademlia::_good_ping_latency = 0;
long unsigned Kademlia::_good_timeouts = 0;

long unsigned Kademlia::_good_hops = 0;
Time Kademlia::_good_hop_latency = 0;

Time Kademlia::_bad_lookup_latency = 0;
long unsigned Kademlia::_bad_timeouts = 0;
long unsigned Kademlia::_bad_hops = 0;
Time Kademlia::_bad_hop_latency = 0;
Time Kademlia::_default_timeout = 0;

unsigned Kademlia::_to_multiplier = 0;
unsigned Kademlia::_to_cheat = 0;


Kademlia::NodeID *Kademlia::_all_kademlias = 0;
HashMap<Kademlia::NodeID, Kademlia*> *Kademlia::_nodeid2kademlia = 0;

#define NODES_ITER(x) for(set<k_nodeinfo, Kademlia::closer>::const_iterator i = (x)->begin(); i != (x)->end(); ++i)


// }}}
// {{{ Kademlia::Kademlia
Kademlia::Kademlia(IPAddress i, Args a)
  : P2Protocol(i), _id(ConsistentHash::ip2chid(ip()))
{
  // KDEBUG(1) << "id: " << printID(_id) << ", ip: " << ip() << endl;
  if(getenv("P2PSIM_CHECKREP"))
    docheckrep = strcmp(getenv("P2PSIM_CHECKREP"), "0") ? true : false;

  use_replacement_cache = a.nget<use_replacement_cache_t>("rcache", ENABLED, 10);
  max_lookup_time = a.nget<Time>("maxlookuptime", 4000, 10);
  learn_stabilize_only = a.nget<unsigned>("stablearn_only", 0, 10) == 1 ? true : false;
  force_stabilization = a.nget<unsigned>("force_stab", 0, 10) == 1 ? true : false;
  death_notification = 
    a.nget<unsigned>("death_notify", 1, 10) == 1 ? true : false;

  if(!k) {
    k = a.nget<unsigned>("k", 20, 10);
  }

  if(!k_tell) {
    k_tell = a.nget<unsigned>("k_tell", k, 10);
  }

  if(!alpha) {
    alpha = a.nget<unsigned>("alpha", 3, 10);
  }

  if(!stabilize_timer) {
    refresh_rate = stabilize_timer = a.nget<unsigned>("stabilize_timer", 10000, 10);
  }

  if(!erase_count) {
    erase_count = a.nget<unsigned>("erase_count", 5, 10);
  }

  if(!_default_timeout) {
    _default_timeout = a.nget<Time>("default_timeout", 1000, 10);
  }

  if (!_to_multiplier) {
    _to_multiplier = a.nget<unsigned>("timeout_multiplier",3,10);
  }
  _to_cheat = a.nget<unsigned>("timeout_cheat",1,10);

  if(!Kademlia::pool)
    Kademlia::pool = New k_nodeinfo_pool();
  _me = k_nodeinfo(_id, ip());
  _joined = false;

  init_k_bucket_tree();
  _nkademlias++;
}

Kademlia::NodeID Kademlia::closer::n = 0;
Kademlia::NodeID Kademlia::closerRTT::n = 0;
// }}}
// {{{ Kademlia::~Kademlia
Kademlia::~Kademlia()
{
  // KDEBUG(1) << "Kademlia::~Kademlia" << endl;
  for(HashMap<NodeID, k_nodeinfo*>::iterator i = flyweight.begin(); i != flyweight.end(); ++i)
    Kademlia::pool->push(i.value());
  flyweight.clear();

  delete _root;

  if(ip() == 1) {
    printf("\
#  1: k\n\
#  2: k_tell\n\
#  3: alpha\n\
#  4: stabilize_timer\n\
#  5: refresh_timer\n\
#  6: learn\n\
#  7: rcache\n\
#\n\
#  8: total number of bytes for RPCs\n\
#  9: number of RPCs sent in lookup (good)\n\
# 10: number of RPCs sent in lookup (bad)\n\
# 11: number of good      RPCs reaped by reaper\n\
# 12: number of timed out RPCs reaped by reaper\n\
#\n\
# 13: number of good lookups\n\
# 14: number of attempts for good lookups\n\
# 15: number of attempts for bad lookups\n\
# 16: number of good lookups, but node was dead\n\
# 17: number of bad lookups, node was dead\n\
# 18: number of bad lookups, node was alive\n\
#\n\
# 19: avg total lookup latency (good)\n\
# 20: avg pure lookup latency (good)\n\
# 21: avg pure ping latency (good)\n\
# 22: number of timeouts suffered during lookup (good)\n\
#\n\
# 23: avg number of hops (good)\n\
# 24: avg latency per hop (good)\n\
#\n\
# 25: avg pure lookup latency (bad)\n\
# 26: number of timeouts suffered during lookup (bad)\n\
# 27: avg number of hops (bad)\n\
# 28: avg latency per hop (bad)\n\
#\n\
%u %u %u %u %u %u %u      %llu %lu %lu %lu %lu     %lu %lu %lu %lu %lu %lu    %.2f %.2f %.2f %lu    %.2f %.2f   %.2f %lu %.2f %.2f\n",
        Kademlia::k,
        Kademlia::k_tell,
        Kademlia::alpha,
        Kademlia::stabilize_timer,
        Kademlia::refresh_rate,
        Kademlia::learn_stabilize_only ? 1 : 0,
        Kademlia::use_replacement_cache,

        _rpc_bytes,
        _good_rpcs,
        _bad_rpcs,
        _ok_by_reaper,
        _timeouts_by_reaper,

        _good_lookups,
        _good_attempts,
        _bad_attempts,
        _lookup_dead_node,
        _ok_failures,
        _bad_failures,

        (double) _good_total_latency / _good_lookups,
        (double) _good_lookup_latency / _good_lookups,
        (double) _good_ping_latency / _good_lookups,
        _good_timeouts,

        (double) _good_hops / _good_lookups,
        (double) _good_hop_latency / _good_hops,

        (double) _bad_lookup_latency / _bad_attempts,
        _bad_timeouts,
        (double) _bad_hops / _bad_attempts,
        (double) _bad_hop_latency / _bad_attempts);


    print_stats();
  }

  if(_all_kademlias) {
    delete _all_kademlias;
    _all_kademlias = 0;
  }

  if(_nodeid2kademlia) {
    delete _nodeid2kademlia;
    _nodeid2kademlia = 0;
  }


  if(--_nkademlias == 0) {
    delete pool;
    pool = 0;
  }
}

void
Kademlia::init_k_bucket_tree()
{
  _root = New k_bucket(0, this);
  _root->leaf = false;

// build the entire k-bucket tree
  k_bucket *b = _root;
  unsigned bit;
  for(unsigned i=0; ; i++) {
    bit = Kademlia::getbit(_id, i);

    // leafside
    b->child[bit^1] = New k_bucket(b, this);
    b->child[bit^1]->nodes = New k_nodes(b->child[bit^1]);
    b->child[bit^1]->replacement_cache = New k_nodes(b->child[bit^1]);
    b->child[bit^1]->leaf = true;

    // nodeside
    b->child[bit] = New k_bucket(b, this);
    b = b->child[bit];
    b->leaf = false;

    if(i == Kademlia::idsize-2)
      break;
  }
  b->nodes = New k_nodes(b);
  b->replacement_cache = New k_nodes(b);
  b->leaf = true;
}

// jy
Time
Kademlia::timeout(IPAddress dst)
{
  if (_to_cheat)
    return _to_multiplier * 2 * Network::Instance()->gettopology()->latency(_me.ip,dst);
  else
    return Kademlia::_default_timeout;
}

// }}}
// {{{ Kademlia::initstate
void
Kademlia::initstate()
{
  const set<Node*> *l = Network::Instance()->getallnodes();
  // KDEBUG(1) << "Kademlia::initstate l->size = " << l->size() << endl;

  if(!_all_kademlias) {
    // KDEBUG(1) << "allocating all" << endl;
    _all_kademlias = New NodeID[l->size()];
    _nodeid2kademlia = New HashMap<NodeID, Kademlia*>;
    unsigned j = 0;
    for(set<Node*>::const_iterator i = l->begin(); i != l->end(); ++i) {
      Kademlia *k = (Kademlia *) *i;
      _all_kademlias[j++] = k->id();
      _nodeid2kademlia->insert(k->id(), k);
    }
  }

  Kademlia::NodeID upper = 0, lower = 0;
  for(unsigned depth=0; depth<Kademlia::idsize; depth++) {
    upper = lower = 0;
    for(unsigned i=0; i<Kademlia::idsize; i++) {
      unsigned bit = Kademlia::getbit(_id, i);
      if(i < depth) {
        lower |= (((Kademlia::NodeID) bit) << Kademlia::idsize-i-1);
        upper |= (((Kademlia::NodeID) bit) << Kademlia::idsize-i-1);
      } else if(i == depth) {
        lower |= (((Kademlia::NodeID) bit^((Kademlia::NodeID)1)) << Kademlia::idsize-i-1);
        upper |= (((Kademlia::NodeID) bit^((Kademlia::NodeID)1)) << Kademlia::idsize-i-1);
      } else if(i > depth)
        upper |= (((Kademlia::NodeID) 1) << Kademlia::idsize-i-1);
    }

    // // KDEBUG(2) << " initstate lower = " << Kademlia::printbits(lower) << endl;
    // // KDEBUG(2) << " initstate upper = " << Kademlia::printbits(upper) << endl;

    // create a set of canditates
    vector<NodeID> candidates;
    for(unsigned i = 0; i < l->size(); i++) {
      if(_all_kademlias[i] >= lower && _all_kademlias[i] <= upper)
        candidates.push_back(_all_kademlias[i]);
    }

    // if candidate set size is less than k, then insert them all.
    if(candidates.size() <= Kademlia::k) {
      for(vector<NodeID>::const_iterator i = candidates.begin(); i != candidates.end(); ++i) {
        Kademlia *k = (*_nodeid2kademlia)[*i];
        // KDEBUG(2) << "initstate: inserting candidate = " << Kademlia::printID(k->id()) << endl;
        // assert(k);
        insert(k->id(), k->ip(), 0, 0, true);
      }
      continue;
    }

    // if candidate set is too large, pick k at random
    unsigned count = 0;
    while(count <= Kademlia::k && candidates.size()) {
      unsigned index = random() % candidates.size();
      Kademlia *k = (*_nodeid2kademlia)[candidates[index]];
      // assert(k);

      if(flyweight[k->id()])
        continue;

      insert(k->id(), k->ip(), 0, 0, true);
      count++;

      // make a copy of candidates and remove the selected index
      vector<NodeID> tmp;
      for(unsigned i=0; i<candidates.size(); i++) {
        if(i == index)
          continue;
        tmp.push_back(candidates[i]);
      }
      candidates.clear();
      candidates = tmp;
    }
  }
  _joined = true;
}

// }}}
// {{{ Kademlia::join
void
Kademlia::join(Args *args)
{
  if(_joined)
    return;

  IPAddress wkn = args->nget<IPAddress>("wellknown");

  // pick a new ID
  //KDEBUG(1) << "Kademlia::join ip " << ip() << " id " << printID(_id) << " now " << now() << endl;
  assert(!_root);
  _nodeid2kademlia->remove(_id);
  _id = ConsistentHash::ip2chid(ip());
  _me = k_nodeinfo(_id, ip());
  assert(!_nodeid2kademlia->find_pair(_id));
  _nodeid2kademlia->insert(_id, this);
  assert(!_root);
  init_k_bucket_tree();

  // I am the well-known node
  if(wkn == ip()) {
    if(stabilize_timer)
      delaycb(stabilize_timer, &Kademlia::reschedule_stabilizer, (void *) 0);
    _joined = true;
    return;
  }

join_restart:
  // lookup my own key with well known node.
  lookup_args la(_id, ip(), _id);
  la.stattype = Kademlia::STAT_JOIN;
  lookup_result lr;
  bool b = false;
  Time before = 0;
  do {
    record_stat(STAT_JOIN, 1, 0);
    before = now();
    b = doRPC(wkn, &Kademlia::do_lookup, &la, &lr, timeout(wkn));
    record_stat(STAT_JOIN, lr.results.size(), 0);
  } while(!b);

  if(!alive())
    return;

  // put well known node in k-buckets
  // KDEBUG(2) << "join: lr.rid = " << printID(lr.rid) << endl;