chordfingerpns.c

P2P模拟器

/*
 * Copyright (c) 2003-2005 Jinyang Li
 *                    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  "chordfingerpns.h"
#include "observers/chordobserver.h"
#include "p2psim/network.h"
#include <stdio.h>
#include <iostream>

using namespace std;

extern bool static_sim;

ChordFingerPNS::ChordFingerPNS(IPAddress i, Args& a, LocTable *l, const char *name) 
  : Chord(i, a, New LocTable(), name) 
{ 
  _base = a.nget<uint>("base",2,10);
  _samples = a.nget<int>("samples",_nsucc,10);
  _fingerlets = a.nget<uint>("fingerlets",1,10);

  _stab_pns_running = false;
  _stab_pns_outstanding = 0;

  _stab_pns_timer = a.nget<uint>("pnstimer",_stab_succlist_timer,10);

  learntable = New LocTable();
  learntable->set_timeout(10*_stab_pns_timer);
  learntable->set_evict(false);
  learntable->init(me);
}

void
ChordFingerPNS::oracle_node_died(IDMap n)
{
  Chord::oracle_node_died(n);

  IDMap tmp = loctable->succ(n.id,LOC_ONCHECK);
  if (tmp.ip != n.ip) return;

  //lost one my of finger?
  vector<IDMap> ids = ChordObserver::Instance(NULL)->get_sorted_nodes();
  uint sz = ids.size();
  CHID lap = (CHID) -1;
  while (lap > 1) {
    lap = lap / _base;
    for (uint j = 1; j <= (_base-1); j++) {
      if (ConsistentHash::between(me.id+lap*j, me.id+lap*(j+1), n.id)) {
	loctable->del_node(n);
	//find a replacement for this finger
	IDMap tmpf;
	tmpf.id = lap * j + me.id;
	uint s_pos = upper_bound(ids.begin(), ids.end(), tmpf, Chord::IDMap::cmp) - ids.begin();
	s_pos = s_pos % sz;
	tmpf.id = lap * (j+1) + me.id;
	uint e_pos = upper_bound(ids.begin(), ids.end(), tmpf, Chord::IDMap::cmp) - ids.begin();
	e_pos = e_pos % sz;
	if (s_pos == e_pos) return;

	IDMap min_f = ids[s_pos];
	Topology *t = Network::Instance()->gettopology();
	uint min_l = t->latency(me.ip,min_f.ip);
	int k = 0;
	for (uint i = s_pos; i!= e_pos; i = (i+1)%sz) {
	  if (t->latency(me.ip,ids[i].ip) < min_l) {
	    min_f = ids[i];
	    min_l = t->latency(me.ip, ids[i].ip);
	    k++;
	    if (_samples > 0 && k>= _samples) break;
	  }
	}
	loctable->add_node(min_f);//add replacement pns finger
	CDEBUG(3) << "oracle_node_died del " << n.ip << "," << printID(n.id)
	  << "add " << min_f.ip << "," << printID(min_f.id) << endl;
	return;
      }
    }
  }
  assert(0);
}

void
ChordFingerPNS::oracle_node_joined(IDMap n)
{
  Chord::oracle_node_joined(n);

  IDMap tmp = loctable->succ(n.id,LOC_ONCHECK);
  if (tmp.ip == n.ip) return;

  //is the new node one of my finger?
  vector<IDMap> ids = ChordObserver::Instance(NULL)->get_sorted_nodes();
  uint sz = ids.size();
  CHID lap = (CHID) -1;
  while (lap > 1) {
    lap = lap / _base;
    for (uint j = 1; j <= (_base-1); j++) {
      if (ConsistentHash::between(me.id+lap*j, me.id+lap*(j+1), n.id)) {
	IDMap s = loctable->succ(me.id + lap * j,LOC_ONCHECK);
	if (!ConsistentHash::between(me.id+lap*j, me.id+lap*(j+1), s.id)) {
	  loctable->add_node(n);
	  CDEBUG(3) << "oracle_node_joined finger add " << n.ip << ","
	    << printID(n.id) << endl;
	  return;
	}
	IDMap tmpf;
	tmpf.id = lap * j + me.id;
	uint s_pos = upper_bound(ids.begin(), ids.end(), tmpf, Chord::IDMap::cmp) - ids.begin();
	uint n_pos = find(ids.begin(),ids.end(),n) - ids.begin();
	n_pos = n_pos + sz;
	if (_samples < 0 || (int)(n_pos - s_pos) <= _samples) {
	  //choose the closer one
	  Topology *t = Network::Instance()->gettopology();
	  if (t->latency(me.ip, n.ip) < t->latency(me.ip, s.ip)) {
	    loctable->del_node(s,true);
	    loctable->add_node(n);
	    CDEBUG(3) << "oracle_node_joined finger del " << s.ip << ","
	      << printID(s.id) << "add " << n.ip << "," << printID(n.id)
	      << endl;
	  }
	} 
	return;
      }
    }
  }
  assert(0);
}

inline static 
void which_finger(uint base, ConsistentHash::CHID id, ConsistentHash::CHID meid, 
    ConsistentHash::CHID &fs, ConsistentHash::CHID &fe)
{
  ConsistentHash::CHID lap = (ConsistentHash::CHID) -1;
  ConsistentHash::CHID finger;
  while (1) {
    lap = lap/base;
    for (uint j = (base-1); j >= 1; j--) {
      finger = lap * j + meid;
      if (ConsistentHash::between(finger,finger+lap,id)) {
	fs = finger;
	fe = finger + lap;
	return;
      }
    }
  }
}

void
ChordFingerPNS::learn_info(IDMap n)
{
  assert(learntable);

  if(loctable->update_ifexists(n)) 
    return;

  if (learntable->update_ifexists(n))
    return;

  Topology *t = Network::Instance()->gettopology();
  IDMap ns = loctable->succ(n.id+1,LOC_ONCHECK);
  LocTable::idmapwrap *naked_ns = NULL;

  if ((ns.ip==0) || (ns.ip == me.ip))
    goto NEXT_CHECK;

  naked_ns = loctable->get_naked_node(ns.id);
  assert(naked_ns);

  if (naked_ns->is_succ) {
    loctable->add_node(n,true); //accidentally discovered a new successor
    return;
  }else if (!naked_ns->fs) 
    goto NEXT_CHECK;

  //assert(naked_ns->status == LOC_HEALTHY);
  if (!naked_ns->fs) return;

  if (ConsistentHash::between(naked_ns->fs,naked_ns->fe,n.id)) {
    if ((t->latency(me.ip,ns.ip) > t->latency(me.ip,n.ip)) || (naked_ns->status==LOC_ONCHECK)) {
      loctable->add_node(n,false,true,naked_ns->fs,naked_ns->fe); //a better finger in latency
      loctable->del_node(ns,true);
      return;
    }else{
      learntable->add_node(n);
      return;
    }
  } 

NEXT_CHECK:
  IDMap ps = loctable->pred(n.id-1,LOC_ONCHECK);
  LocTable::idmapwrap *naked_ps = loctable->get_naked_node(ps.id);
  if (naked_ps && naked_ns && (!naked_ps->is_succ)) {
    if ((naked_ps->fe == naked_ns->fs) || ConsistentHash::between(naked_ps->fs,naked_ps->fe,n.id)) {
      //good good
      if (t->latency(me.ip,ns.ip) > t->latency(me.ip,n.ip)) {
	loctable->add_node(n,false,true,naked_ps->fs,naked_ps->fe);
	loctable->del_node(ns,true);
	return;
      }
    }else{
      learntable->add_node(n);
      return;
    }
  }
  ConsistentHash::CHID fs,fe;

  which_finger(_base,n.id,me.id,fs,fe);
  loctable->add_node(n,false,true,fs,fe);
}

bool
ChordFingerPNS::replace_node(IDMap n, IDMap &replacement)
{
  assert(learntable);
  LocTable::idmapwrap *naked = loctable->get_naked_node(n.id);
  if (!naked)
    return false;
  if (naked->is_succ) {
    //printf("%s no replacement for successor list %u,%qx\n",ts(), n.ip,n.id);
    //no replacement
    return false;
  }else{
    vector<IDMap> v = learntable->succs(naked->fs,_samples);
    IDMap min_f = me;
    Time min_lat = 10000000;
    Topology *t = Network::Instance()->gettopology();
    for (uint i = 0; i < v.size(); i++) {
      if (ConsistentHash::between(naked->fs,naked->fe,v[i].id)) { 
	if (min_lat > t->latency(me.ip,v[i].ip) && v[i].ip!=n.ip) {
	  min_lat = t->latency(me.ip,v[i].ip);
	  min_f = v[i];
	}
      }else{
	break;
      }
    }
    if (min_f.ip!=me.ip) {
      if (t->latency(me.ip,min_f.ip) < t->latency(me.ip,n.ip)) 
	loctable->add_node(min_f,false,true,naked->fs,naked->fe);
      else
	loctable->add_node(min_f,false,true,naked->fs,naked->fe,true);
      loctable->del_node(n);
      learntable->del_node(min_f,true);
      replacement = min_f;
      assert(min_f.ip != 0);
      //printf("%s replaced finger %u,%qx range (%qx,%qx) latency %u with %u,%qx latency %u\n", ts(), n.ip, n.id, naked->fs, naked->fe, (uint) t->latency(me.ip, n.ip), replacement.ip, replacement.id, (uint)t->latency(me.ip,replacement.ip));
      return true;
    }else{
      //printf("%s not replaced finger %u,%qx range (%qx,%qx) latency %u\n",ts(), n.ip,n.id, naked->fs,naked->fe, (uint)t->latency(me.ip, n.ip));
    }

  }
  return false;
}


void
ChordFingerPNS::initstate()
{
  vector<IDMap> ids = ChordObserver::Instance(NULL)->get_sorted_nodes();
  uint sz = ids.size();
  uint my_pos = find(ids.begin(), ids.end(), me) - ids.begin();
  assert(my_pos < sz && ids[my_pos].id == me.id);
  CHID min_lap = ids[(my_pos+_nsucc) % sz].id - me.id;

  CHID lap = (CHID) -1;

  Topology *t = Network::Instance()->gettopology();
  while (lap > min_lap) {
    lap = lap / _base;
    for (uint j = 1; j <= (_base -1); j++) {
      if (lap * j < min_lap) continue;
      IDMap tmpf;
      tmpf.id = lap * j + me.id;
      uint s_pos = upper_bound(ids.begin(), ids.end(), tmpf, Chord::IDMap::cmp) - ids.begin();

      s_pos = s_pos % sz;
      tmpf.id = lap * (j+1) + me.id + 1;
      uint e_pos = upper_bound(ids.begin(), ids.end(), tmpf, Chord::IDMap::cmp) - ids.begin();
      e_pos = e_pos % sz;
      if (s_pos == e_pos) continue;

      IDMap min_f;
      min_f.ip = 0;
      uint min_l = 1000000;
      hash_map<IPAddress, bool> inserted;

      for (uint x = 0; x < _fingerlets; x++) {
	int k = 0;
	for (uint i = s_pos; i!= e_pos; i = (i+1)%sz) {
	  if (inserted.find(ids[i].ip)==inserted.end() && t->latency(me.ip,ids[i].ip) < min_l) { 
	    min_f = ids[i];
	    min_l = t->latency(me.ip, ids[i].ip);