chordfingerpns.c

P2P模拟器

	  }
	  k++;
	  if (_samples > 0 && k>= _samples) break;
	}
	if (min_f.ip > 0) {
	  if (x == 0) 
	    loctable->add_node(min_f,false,true,tmpf.id,tmpf.id+lap);
	  else
	    learntable->add_node(min_f,false,true,tmpf.id,tmpf.id+lap);
	  inserted[min_f.ip] = true;
	  min_f.ip = 0;
	  min_l = 1000000;
	} else {
	  break;
	}
      }
    }
  }
  CDEBUG(3) << "initstate sz " << loctable->size() << " pns_samples " 
    << _samples << endl;
  Chord::initstate();
}

bool
ChordFingerPNS::stabilized(vector<CHID> lid)
{
  return true;
}

void
ChordFingerPNS::join(Args *args)
{

  if ((static_sim) || (!alive())) return;

  Chord::join(args);

  if (me.ip!=_wkn.ip) {
    CHID fs,fe;
    which_finger(_base, _wkn.id, me.id, fs, fe);
    learntable->add_node(_wkn,false,true,fs,fe);
  }

  //schedule pns stabilizer, no finger stabilizer
  if (!_stab_pns_running) {
    _stab_pns_running = true;
    if (!_learn)
      reschedule_pns_stabilizer((void *)1);
  }else if (_join_scheduled == 0){ //successfully joined
    if (!_learn)
      ChordFingerPNS::fix_pns_fingers(true);
  }
}

void
ChordFingerPNS::reschedule_pns_stabilizer(void *x)
{
  if (!alive()) {
    _stab_pns_running = false;
    return;
  }
  _stab_pns_running = true;
  if (_stab_pns_outstanding > 0) {
  }else{
    _stab_pns_outstanding++;
    fix_pns_fingers(x > 0?true:false);
    _stab_pns_outstanding--;
    assert(_stab_pns_outstanding == 0);
  }
  delaycb(_stab_pns_timer, &ChordFingerPNS::reschedule_pns_stabilizer, (void *) 0);
}

uint
ChordFingerPNS::num_live_samples(vector<IDMap> ss)
{
  uint num = 0;
  for (uint i = 0; i < ss.size(); i++) {
    if (getpeer(ss[i].ip)->alive()) {
      num++;
    }
  }
  return num;
}

void
ChordFingerPNS::fix_pns_fingers(bool restart)
{
  if (!alive()) return;
  vector<IDMap> scs = loctable->succs(me.id + 1, _nsucc);

  CDEBUG(3) << "fix_pns_fingers start sz " << loctable->size() << " scs " << scs.size() << endl;

  uint dead_finger = 0;
  uint missing_finger = 0;
  uint real_missing_finger = 0;
  uint new_finger = 0;
  uint new_added_finger = 0;
  uint new_deleted_finger = 0;
  uint total_finger= 0;
  uint valid_finger = 0;
  uint skipped_finger = 0;
  uint finger_lookup = 0;
  hash_map<IPAddress,bool> inserted;

  if (scs.size() == 0) return;

  vector<IDMap> v;
  vector<IDMap> newsucc;
  CHID finger;
  IDMap currf, prevf, prevfpred, deadf, min_f, tmp;
  bool ok;
  uint min_l;

  Topology *t = Network::Instance()->gettopology();

  CHID lap = (CHID) -1;

  prevf.ip = 0;
  prevfpred.ip = 0;

  IDMap pred = loctable->pred(me.id-1);
  IDMap haha = scs[scs.size()-1];

  while (1) {
    lap = lap/_base;
    for (uint j = (_base-1); j >= 1; j--) {
      finger = lap * j + me.id;
      if ((ConsistentHash::betweenrightincl(me.id,scs[scs.size()-1].id,finger+lap)) 
	  || (!lap) || (!alive()))
	goto PNS_DONE;
      currf = loctable->succ(finger);
      if (currf.ip == me.ip) 
	currf.ip = 0;

      if (currf.ip) {
	LocTable::idmapwrap *naked = loctable->get_naked_node(currf.id);
	if (naked->status == LOC_REPLACEMENT) {
	  //printf("%s re-install replaced crap %u,%qx\n",ts(),currf.ip,currf.id);
	  loctable->del_node(currf);
	  currf.ip = 0;
	}
      }
      total_finger++;//testing
      //
      if (currf.ip == pred.ip) currf.ip = 0;
      //if (currf.ip == me.ip) continue;

      if ((!restart) && (currf.ip)) {

	if (ConsistentHash::between(finger, finger + lap, currf.id)) {
	  LocTable::idmapwrap *naked = loctable->get_naked_node(currf.id);
	  if (naked->is_succ) goto PNS_DONE;
	  assert(naked);
	  if (!naked->fs) {
	    naked->fs = finger;
	    naked->fe = finger+lap;
	  }
	  //if lookup has updated the timestamp of this finger, ignore it
	  if ((now() - naked->n.timestamp) < _stab_pns_timer) {
	    skipped_finger++; //testing
	    continue;
	  }
	  assert(currf.ip != prevf.ip);
	  prevf = currf;
	  prevfpred.ip = 0;
	  //just ping this finger to see if it is alive
	  //record_stat(TYPE_PNS_UP,0);
	  ok = failure_detect(currf, &Chord::null_handler, (void *)NULL,&currf,TYPE_PNS_UP,0,0);
	  if (!alive()) goto PNS_DONE;
	  if(ok) {
	    record_stat(currf.ip,me.ip,TYPE_PNS_UP,0);
	    loctable->update_ifexists(currf);//update timestamp
	    valid_finger++;//testing
	    continue;
	  }else{
	    loctable->del_node(currf);
	    naked = NULL;
	    dead_finger++;//testing
	    deadf = currf;
	    inserted[deadf.ip] = true;
	  }
	} else if (!restart) {
	  missing_finger++;//testing
	  if ((currf.ip == prevf.ip) && (prevfpred.ip) && (ConsistentHash::between(prevfpred.id,currf.id,finger))) {
	    skipped_finger++;
	    continue;
	  } else if ((currf.ip != prevf.ip) || (!prevfpred.ip)) {
	    prevf = currf;
	    prevfpred.ip = 0;
	    //get predecessor, coz new finger within the candidate range might show up
	    get_predsucc_args gpa;
	    gpa.pred = true;
	    gpa.m=0;
	    get_predsucc_ret gpr;
	    record_stat(me.ip,currf.ip,TYPE_PNS_UP,0);
	    ok = doRPC(currf.ip, &Chord::get_predsucc_handler, &gpa, &gpr, TIMEOUT(me.ip,currf.ip));
	    if (!alive()) goto PNS_DONE;
	    if(ok) {
	      record_stat(currf.ip,me.ip,TYPE_PNS_UP,1);
	      loctable->add_node(currf);//update timestamp
	      prevfpred = gpr.n;
	      if (ConsistentHash::between(gpr.n.id,currf.id, finger)) { //pred is beyond finger, there is no real missing finger
		continue;
	      } else {
		real_missing_finger++; //testing
	      }
	    }else{
	      loctable->del_node(currf);
	      deadf = currf;
	      inserted[deadf.ip] = true;
	      dead_finger++; //testing
	    }
	  }else{
	    real_missing_finger++; //testing
	  }
	}
      }

      finger_lookup++;
      if (_recurs)
	v = find_successors_recurs(finger, _samples, TYPE_FINGER_LOOKUP);
      else
	v = find_successors(finger, _samples, TYPE_FINGER_LOOKUP);

      if (!alive()) goto PNS_DONE;
 
      if (v.size() > 0) {
	new_finger++; //testing

	uint x = 0;
	while (x < _fingerlets) {
	  min_l = 1000000;
	  min_f.ip = 0;
	  for (uint k = 0; k < v.size(); k++) {
	    if (ConsistentHash::between(finger,finger+lap,v[k].id)
	      && t->latency(me.ip, v[k].ip) < min_l && inserted.find(v[k].ip)==inserted.end()) {
	      min_f = v[k];
	      min_l = t->latency(me.ip, v[k].ip);
	    }
	  }

	  if (min_f.ip == 0) break;
	  newsucc = loctable->succs(me.id+1, _nsucc, LOC_HEALTHY);
	  if ((min_f.ip) && newsucc.size() && (!ConsistentHash::betweenrightincl(finger,finger+lap,newsucc[newsucc.size()-1].id))) {
	    new_added_finger++; //testing
	    assert(ConsistentHash::between(finger,finger+lap,min_f.id));
	    //ping this node, coz it might have been dead
	    record_stat(me.ip,min_f.ip,TYPE_PNS_UP,0); 
	    ok = doRPC(min_f.ip, &Chord::null_handler, (void *)NULL, &min_f, TIMEOUT(me.ip,min_f.ip));
	    if (!alive()) goto PNS_DONE;
	    if (ok) {
	      record_stat(min_f.ip,me.ip,TYPE_PNS_UP,0); 
	      if (x == 0) {
		loctable->add_node(min_f,false,true,finger,finger+lap);
		tmp = loctable->succ(finger,LOC_HEALTHY);
		while (tmp.ip != min_f.ip) {
		  if (ConsistentHash::between(finger,finger+lap,tmp.id)) {
		    loctable->del_node(tmp,true);
		    new_deleted_finger++; //testing
		  }else{
		    break;
		  }
		  tmp = loctable->succ(finger,LOC_HEALTHY);
		}
	      }else{
		learntable->add_node(min_f,false,true,finger,finger+lap);
	      }
	      x++;
	    }
	    inserted[min_f.ip] = true;
	  }else
	    break;
	}
      } else{
      }

    }
  }
PNS_DONE:
  CDEBUG(3) << "fix_pns_fingers finished sz " << loctable->size()
    << " lookup " << finger_lookup << " total " << total_finger 
    << " valid " << valid_finger << " skipped " << skipped_finger 
    << " dead " << dead_finger << " missing " << missing_finger 
    << " real_missing " << real_missing_finger << " new " << new_finger
    << " new_added " << new_added_finger << " new_deleted " 
    << new_deleted_finger << endl;
  return;
}

void
ChordFingerPNS::dump()
{
  Chord::dump();
  IDMap succ = loctable->succ(me.id + 1,LOC_ONCHECK);
  CHID min_lap = succ.id - me.id;
  CHID lap = (CHID) -1;
  CHID finger;

  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;
      finger = lap * j + me.id;
      succ = loctable->succ(finger,LOC_HEALTHY);
      if (succ.ip > 0) 
        printf("%qx: finger: %qx,%d : %qx : succ %qx lat %d\n", me.id, lap, j, finger, succ.id, (unsigned) t->latency(me.ip, succ.ip));
    }
  }
}