kademlia.c

P2P模拟器

Kademlia::common_prefix(Kademlia::NodeID k1, Kademlia::NodeID k2)
{
  unsigned size = 0;
  for(unsigned i=0; i<idsize; i++)
    if(getbit(k1, i) == getbit(k2, i))
      size++;
    else
      break;
  return size;
}

// }}}
// {{{ Kademlia::record_stat
void
Kademlia::record_stat(stat_type type, uint num_ids, uint num_else )
{
  record_bw_stat(type, num_ids, num_else);
  if(collect_stat())
    _rpc_bytes += 20 + num_ids * 4 + num_else; // paper says 40 bytes per node entry
}
// }}}
// {{{ Kademlia::distance
Kademlia::NodeID
Kademlia::distance(Kademlia::NodeID from, Kademlia::NodeID to)
{
  return from ^ to;
}

// }}}
// {{{ Kademlia::printbits
string
Kademlia::printbits(NodeID id)
{
  char buf[128];

  unsigned j=0;
  for(int i=idsize-1; i>=0; i--)
    sprintf(&(buf[j++]), "%llu", (id >> i) & 0x1);
  sprintf(&(buf[j]), ":%llx", id);

  return string(buf);
}

// }}}
// {{{ Kademlia::printID
string
Kademlia::printID(NodeID id)
{
  char buf[128];
  sprintf(buf, "%llx", id);
  return string(buf);
}

// }}}
// {{{ Kademlia::reap
// reaps outstanding RPCs, but snatches out the info that's contained in it.
// I.e., we update k-buckets.
void
Kademlia::reap(void *r)
{
  reap_info *ri = (reap_info *) r;
  // NodeID _id = ri->k->id();

  // cout << "Kademlia::reap" << endl;
  assert((int) ri->rpcset->size() == ri->outstanding_rpcs->size());

  while(ri->outstanding_rpcs->size()) {
    bool ok;
    unsigned donerpc = ri->k->rcvRPC(ri->rpcset, ok);
    callinfo *ci = ri->outstanding_rpcs->find(donerpc);

    // cout << "Kademlia::reap ok = " << ok << ", ki = " << endl;
    if(ok) {
      _ok_by_reaper++;
      if(ri->k->alive() && (!Kademlia::learn_stabilize_only || ci->fa->stattype == STAT_STABILIZE ||
                                                               ci->fa->stattype == STAT_LOOKUP))
        ri->k->update_k_bucket(ci->ki.id, ci->ki.ip);
      ri->k->record_stat(ri->stat, ci->fr->results.size(), 0);

    } else if(ri->k->flyweight[ci->ki.id]) {
      _timeouts_by_reaper++;
      if(ri->k->alive() && (!Kademlia::learn_stabilize_only || ci->fa->stattype == STAT_STABILIZE ||
                                                               ci->fa->stattype == STAT_LOOKUP))
        ri->k->erase(ci->ki.id);
    }

    // Although it slightly affects success rate, it's only fair
    // to only count nodes as dead if you're alive when you receive
    // their response.
    if( !ok && Kademlia::death_notification && ri->k->alive() ) {
      ri->is_dead->insert( ci->ki.id, true );
    }

    ri->outstanding_rpcs->remove(donerpc);
    delete ci;
  }

  // make vectors of dead-nodes sorted by the neighbor that told you.
  if( Kademlia::death_notification ) {

    HashMap<IPAddress, vector<NodeID> *> bad_info;
    HashMap<unsigned, erase_args*> deathmap;
    RPCSet deathrpcset;

    for( HashMap<NodeID, bool>::iterator i=ri->is_dead->begin(); 
	 i != ri->is_dead->end(); ++i ) {
      NodeID dead_node = i.key();
      vector<IPAddress> *v = (*(ri->who_told_me))[dead_node];
      if( v != NULL ) {
	for( unsigned j = 0; j < v->size(); j++ ) {
	  if( bad_info[(*v)[j]] == NULL ) {
	    bad_info.insert( (*v)[j], New vector<NodeID> );
	  }
	  (bad_info[(*v)[j]])->push_back( dead_node );	
	}
      }
    }
    
    // cleanup who_told_me state
    for( HashMap<NodeID, vector<IPAddress> *>::iterator i=
	   ri->who_told_me->begin(); 
	 i != ri->who_told_me->end(); ++i ) {
      
      vector<IPAddress> *v = i.value();
      if( v != NULL ) delete v;
      
    }
    
    // notify neighbors about all the deaths
    for( HashMap<IPAddress,vector<NodeID> *>::iterator i=bad_info.begin(); 
	 i != bad_info.end(); ++i ) {
      
      IPAddress informant = i.key();
      vector<NodeID> *v = i.value();

      if( informant == ri->k->ip() ) {
	delete v;
	continue;
      }
      
      erase_args *ea = New erase_args(v);
      ri->k->record_stat(STAT_ERASE, v->size(), 0);
      
      unsigned rpc = ri->k->asyncRPC(informant, &Kademlia::do_erase, ea, 
				     (erase_result *) NULL, 
				     ri->k->timeout(informant));
      deathrpcset.insert(rpc);
      deathmap.insert(rpc, ea);

    }

    // clean up any death notification state as well
    while(deathmap.size()) {
      bool ok;
      unsigned donerpc = ri->k->rcvRPC(&deathrpcset, ok);
      erase_args *ea = deathmap.find(donerpc);
      // NOTE: there's a conscious decision here to not record_stat 
      // the response.  The reason being that I don't care what the 
      // response is, and in real life I wouldn't wait for it.  I only
      // have to do so here to free the memory.
      delete ea->ids;
      delete ea;
      deathmap.remove(donerpc);
    }
    
  }

  // ri->k->_riset.remove(ri);
  delete ri;

  // cout << "reaper done" << endl;
  taskexit(0);
}
// }}}

// {{{ k_nodeinfo_cmp
inline int
k_nodeinfo_cmp(const void *k1, const void *k2)
{
  k_nodeinfo *kx1, *kx2;
  kx1 = *((k_nodeinfo**) k1);
  kx2 = *((k_nodeinfo**) k2);
  return kx1->lastts - kx2->lastts;
}
// }}}
// {{{ k_nodes::k_nodes
k_nodes::k_nodes(k_bucket *parent) : _parent(parent)
{
  _map.clear();
  _redo = REBUILD;
  _nodes = 0;
}
// }}}
// {{{ k_nodes::~k_nodes
k_nodes::~k_nodes()
{
  if(_nodes)
    delete _nodes;
  _map.clear();
}
// }}}
// {{{ k_nodes::insert
/*
 * pre: n is in the flyweight.
 * post: if n was contained in the set, its lastts is updated and
 *              its position in the set is updated.
 *       if it wasn't in the set, it is inserted.
 */
void
k_nodes::insert(Kademlia::NodeID n, bool touch = false)
{
//  Kademlia::NodeID _id = _parent->kademlia()->id();
  // KDEBUG(1) << "k_nodes::insert " << Kademlia::printID(n) << endl;
  checkrep();

  // assert(n);
  // assert(_parent->kademlia()->flyweight.find(n, 0));
  k_nodeinfo *ninfo = _parent->kademlia()->flyweight[n];
  // assert(ninfo);
  ninfo->checkrep();

  // if already in set, and we're not going to touch the timestamp, we're done.
  bool contained = contains(n);
  if(contained && !touch) {
    checkrep();
    return;
  }

  if(contained && touch && (ninfo->lastts != now())) {
    ninfo->lastts = now();
    ninfo->timeouts = 0;
    _redo = _redo ? _redo : RESORT;
    checkrep();
    return;
  }

  // insert into set
  _map.insert(ninfo, true);
  // assert(_parent->kademlia()->flyweight.find_pair(ninfo->id)->value == ninfo);
  _redo = REBUILD;

  checkrep();
}
// }}}
// {{{ k_nodes::clear
void
k_nodes::clear()
{
  checkrep();

  _map.clear();
  if(_nodes)
    delete _nodes;
  _nodes = 0;
  _redo = REBUILD;

  checkrep();
}
// }}}
// {{{ k_nodes::erase
/*
 * pre: n is in the flyweight. n is contained in this k_nodes.
 * post: n is erased from this k_nodes
 */
void
k_nodes::erase(Kademlia::NodeID n)
{
  checkrep();

  // assert(n);
  // assert(_parent->kademlia()->flyweight.find(n, 0));
  k_nodeinfo* ninfo = _parent->kademlia()->flyweight[n];
  ninfo->checkrep();
  // assert(_map.find(ninfo, false) || _parent->replacement_cache->contains(ninfo->id));

  _map.remove(ninfo);
  _redo = REBUILD;

  checkrep();
}
// }}}
// {{{ k_nodes::contains
bool
k_nodes::contains(Kademlia::NodeID n)
{
  checkrep();
  k_nodeinfo *ki = _parent->kademlia()->flyweight[n];
  // assert(ki);
  return _map.find(ki, false);
}
// }}}
// {{{ k_nodes::rebuild
void
k_nodes::rebuild()
{
  static unsigned _rebuild = 0;
  // Kademlia::NodeID _id = _parent->kademlia()->id();
  // KDEBUG(2) << "k_nodes::rebuild" << endl;

  if(_nodes)
    delete _nodes;
  _nodes = New k_nodeinfo*[_map.size()];

  // fill with entries from _map.
  unsigned j = 0;
  for(HashMap<k_nodeinfo*, bool>::iterator i = _map.begin(); i; i++)
    _nodes[j++] = i.key();
  _redo = RESORT;

  // once in a while throw entries beyond index k out of the set into the
  // replacement cache.  (don't do this if this *is* the replacement cache;
  // k_nodes also implements replacement cache).
  if(_rebuild++ < 100)
    return;

  bool i_am_replacement_cache = (this == _parent->replacement_cache);
  unsigned oldsize = _map.size();
  for(unsigned i=Kademlia::k; i<oldsize; i++) {
    k_nodeinfo *ki = _nodes[i];
    _map.remove(ki);
    if(i_am_replacement_cache || Kademlia::use_replacement_cache == Kademlia::DISABLED) {
      _parent->kademlia()->flyweight.remove(ki->id);
      Kademlia::pool->push(ki);
    } else
      _parent->replacement_cache->insert(ki->id);
  }
  _rebuild = NOTHING;
  rebuild(); // after cleanup, rebuild _nodes again.
}
// }}}
// {{{ k_nodes::get
inline k_nodeinfo*
k_nodes::get(unsigned i)
{
  // assert(i >= 0 && (int) i < _map.size());
  // Kademlia::NodeID _id = _parent->kademlia()->id();

  if(_redo == NOTHING)
    return _nodes[i];

  if(_redo == REBUILD)
    rebuild();

  if(_redo == RESORT) {
    if(_map.size() >= 2)
      qsort(_nodes, _map.size(), sizeof(k_nodeinfo*), &k_nodeinfo_cmp);
    _redo = NOTHING;
  }

  return _nodes[i];
}
// }}}
// {{{ k_nodes::checkrep
inline void
k_nodes::checkrep()
{
  if(!Kademlia::docheckrep)
    return;

  // Kademlia::NodeID _id = _parent->kademlia()->id();

  assert(_parent);
  assert(_parent->nodes == this || _parent->replacement_cache == this);
  assert(_parent->kademlia());
  assert(_parent->kademlia()->id());

  // own ID should never be in flyweight
  assert(!_parent->kademlia()->flyweight.find(_parent->kademlia()->id(), 0));

  if(!size())
    return;

  // all nodes are in flyweight
  // all nodes are unique
  // if we're replacement_cache, entry doesn't exist in nodes and vice versa
  k_nodes *other = _parent->nodes == this ? _parent->replacement_cache : this;
  HashMap<Kademlia::NodeID, bool> haveseen;
  for(unsigned i=0; i<size(); i++) {
    assert(_parent->kademlia()->flyweight.find(get(i)->id, 0));
    assert(_parent->kademlia()->flyweight.find_pair(get(i)->id)->value == get(i));
    assert(!haveseen.find(get(i)->id, false));
    assert(!other->contains(get(i)->id));
    haveseen.insert(get(i)->id, true);
  }

  if(size() == 1) {
    assert(get(0) == last());
    assert(get(0)->id == last()->id);
    return;
  }

  Time prev = get(0)->lastts;
  for(unsigned i=1; i<size(); i++) {
    assert(prev <= get(i)->lastts);
    prev = get(i)->lastts;
  }
}
// }}}

// {{{ k_nodeinfo::k_nodeinfo
k_nodeinfo::k_nodeinfo(NodeID id, IPAddress ip, Time RTT) : id(id), ip(ip), RTT(RTT)
{
  lastts = 0;
  checkrep();
}
// }}}
// {{{ k_nodeinfo::k_nodeinfo
k_nodeinfo::k_nodeinfo(k_nodeinfo *k) :
  id(k->id), ip(k->ip), lastts(k->lastts), timeouts(timeouts), RTT(RTT)
{
  checkrep();
}
// }}}
// {{{ k_nodeinfo::checkrep
void
k_nodeinfo::checkrep() const
{
  if(!Kademlia::docheckrep)