koorde.c

P2P模拟器

    if (!_join_scheduled) {
      _join_scheduled++;
      delaycb(0, &Koorde::join, (Args *)0);
    }
    return;
  }

  //mark all the bad nodes
  if (a->dead.ip)
    loctable->add_check(a->dead);

  //printf ("koorde_next (id=%qx, key=%qx, kshift=%qx, i=%qx) succ=(%u,%qx)\n", 
  //me.id, a->k, a->kshift, a->i, succ.ip, succ.id);
  if (ConsistentHash::betweenrightincl (me.id, succ.id, a->k) ||
      (me.id == succ.id)) {
    r->next = succ;
    r->k = a->k;
    r->kshift = a->kshift;
    r->i = a->i;
    r->done = true;
    r->v.clear ();
    r->v = loctable->succs(me.id + 1, a->nsucc);
    assert (r->v.size () > 0);
    CDEBUG(3) << "koorde_next key " << printID(a->k) << "done: succ " << succ.ip 
      << "," << printID(succ.id) << endl;
  } else if (a->i == me.id || ConsistentHash::betweenrightincl (me.id, succ.id, a->i)) {
    r->k = a->k;
    r->kshift = a->kshift;
    r->i = a->i;
    do {
      r->i = nextimagin (r->i, r->kshift);
      r->kshift = r->kshift << logbase;
    }while (ConsistentHash::betweenrightincl(me.id,succ.id,r->i));
    r->next.id = r->i + 1;
    r->next = loctable->pred(r->i, LOC_HEALTHY);
    r->done = false;
    CDEBUG(3) << "koorde_next key " << printID(a->k) << ": follow debruijn " 
      << r->next.ip << "," << printID(r->next.id) << "i=" << printID(r->i) 
      << "kshift=" << printID(r->kshift) << "debruijn " << printID(debruijn) 
      << endl;
  } else {
    r->k = a->k;
    r->i = a->i;
    r->next.id = r->i + 1;
    r->next = loctable->pred(r->i, LOC_HEALTHY);
    r->kshift = a->kshift;
    r->done = false;
 //   assert (ConsistentHash::betweenrightincl (me.id, r->i, r->next.id));
    CDEBUG(3) << "koorde_next key " << printID(a->k) << "follow succ "<<succ.ip 
      << "," << printID(succ.id) << " to " << r->next.ip << "," 
      << printID(r->next.id) << "i=" << printID(r->i) << "kshift=" 
      << printID(r->kshift) << " debruijn " << printID(debruijn) << endl;
  }
}

void
Koorde::fix_debruijn () 
{

  get_predsucc_args gsa;
  get_predsucc_ret gsr;
  bool ok = false;
  IDMap last;

  //try a cheap way to fix debruijn predecessor first
  assert(resilience <= _nsucc);
  IDMap dpred = loctable->pred(debruijnpred);

  if ((dpred.ip == 0) && (ConsistentHash::between(debruijnpred,debruijn, me.id))) 
    goto NEXT;
    
  gsa.m = _nsucc;
  gsa.pred = true;
  //record_stat();
  if (dpred.ip)
    ok = failure_detect(dpred, &Chord::get_predsucc_handler, &gsa,&gsr, TYPE_FINGER_UP,1,0);
  CDEBUG(3) << "fix_debruijn debruijnpred " << printID(debruijnpred) << dpred.ip << "," 
    << printID(dpred.id) << "ok? " << (ok?1:0) << endl;
  if (!alive()) return;
  if (ok) { //&& gsr.v.size() > 0 && ConsistentHash::between(dpred.id, gsr.v[0].id, debruijnpred)) {
    record_stat(dpred.ip,me.ip,TYPE_FINGER_UP,1+gsr.v.size(),0);
    loctable->add_node(dpred);
    loctable->add_node(gsr.n);
    for (uint i = 0; i < gsr.v.size(); i++) 
      loctable->add_node(gsr.v[i]);
  }else {
    if ((!ok) && dpred.ip)
      loctable->del_node(dpred);
    vector<IDMap> scs = find_successors(debruijnpred, resilience-1, TYPE_FINGER_LOOKUP, &last, NULL);
    if (scs.size() > 0) {
      loctable->add_node(last);
      for (uint i = 0; i < scs.size(); i++) {
	loctable->add_node(scs[i]);
      }
    }
  }
NEXT:
  if (!alive()) return;
  //try a cheap way to test for the validity of debruijn fingers first
  dpred = loctable->pred(debruijn);
  if (!dpred.ip) return;
  gsa.m = fingers;
  gsa.pred = true;
  assert(fingers <= _nsucc);

  ok = failure_detect(dpred, &Chord::get_predsucc_handler, &gsa, &gsr, TYPE_FINGER_LOOKUP,1,0); 
  CDEBUG(3) << "fix_debruijn debruijn " << printID(debruijn) << dpred.ip << "," 
    << printID(dpred.id) << "ok? " << (ok?1:0) << endl;
  if (!alive()) return;
  if (ok && gsr.v.size() > 0 && 
      ConsistentHash::betweenrightincl(dpred.id, gsr.v[gsr.v.size()-1].id, debruijn)) {
    record_stat(dpred.ip,me.ip,TYPE_FINGER_UP,1+gsr.v.size(),0);
    loctable->add_node(dpred);
    loctable->add_node(gsr.n);
    for (uint i = 0; i < gsr.v.size(); i++) {
      loctable->add_node(gsr.v[i]);
    }
    //printf("%s stabilize fix_debruijn cheap finished %qx, its succ %d,%qx its last(pred) %d,%qx\n", 
//	ts(), debruijn, gsr.v[0].ip, gsr.v[0].id, dpred.ip, dpred.id);
  } else {
    if (!ok) loctable->del_node(dpred);
    vector<IDMap> scs = find_successors (debruijn, fingers - 1, TYPE_FINGER_LOOKUP, &last, NULL);
    if (scs.size() > 0) {
 //     printf("%s stabilize fix_debruijn finished debruijn %qx, its succ %d,%qx its last(pred) %d,%qx\n", 
//	  ts(), debruijn, scs[0].ip, scs[0].id, last.ip, last.id);
      loctable->add_node (last);
      for (uint i = 0; i < scs.size (); i++) 
	loctable->add_node (scs[i]);
    }
  }
  
#if 0
  printf ("fix_debruijn (%u,%qx): debruijn %qx succ %qx d %u,%qx at %lu\n",
	  me.ip, me.id, debruijn, scs[0].id, last.ip, last.id, now ());
  for (uint i = 0; i < scs.size(); i++) {
    printf ("  Koorde %d debruijn fingers %u,%qx\n", i+1, scs[i].ip,
	    scs[i].id);
  }
#endif
  
  if (vis) {
    bool change = false;
    IDMap d = loctable->pred (debruijn);
    vector<IDMap> sc = loctable->succs (d.id + 1, fingers - 1);
    vector<IDMap> dfingers;

    dfingers.push_back (d);
    for (uint i = 0; i < sc.size (); i++) {
      dfingers.push_back (sc[i]);
    }

    for (uint i = 0; i < dfingers.size (); i++) {
      if ((i >= lastdfingers.size ()) || lastdfingers[i].id != dfingers[i].id) {
	change = true;
      }
    }

    if (change) {
      printf ( "vis %llu dfingers %16qx %16qx", now (), me.id, debruijn);
      for (uint i = 0; i < dfingers.size (); i++) {
	printf ( " %16qx", dfingers[i].id);
      }
      printf ( "\n");
    }
    lastdfingers = dfingers;
  }

}

void
Koorde::reschedule_debruijn_stabilizer(void *x)
{
  assert(!static_sim);
  if (!alive()) {
    _stab_debruijn_running = false;
    return;
  }
  _stab_debruijn_running = true;
  if (_stab_debruijn_outstanding > 0) {
  }else{
    _stab_debruijn_outstanding++;
    if (_inited)
      fix_debruijn();
    _stab_debruijn_outstanding--;
    assert(_stab_debruijn_outstanding == 0);
  }
  delaycb(_stab_debruijn_timer, &Koorde::reschedule_debruijn_stabilizer, (void *) 0);
}

bool
Koorde::debruijn_stabilized (ConsistentHash::CHID finger, uint n,
			     vector<ConsistentHash::CHID> lid)
{
  IDMap d = loctable->pred (finger);
  vector<IDMap> sc = loctable->succs (d.id + 1, n - 1);
  vector<IDMap> dfingers;

  dfingers.clear ();
  dfingers.push_back (d);
  for (uint i = 0; i < sc.size (); i++) {
    dfingers.push_back (sc[i]);
    //   printf ("succ finger %d is %qx\n", i, sc[i].id);
  }
  assert (dfingers.size () <= n);

  bool r = false;
  // printf ("koorde::stabilized? %u,%qx debruijn %qx d %qx at %lu %d\n", me.ip,
  //  me.id, finger, dfingers[0].id, now (), isstable);

  assert (lid.size () > 0);

  if (lid.size () == 1) {
    assert (d.id == lid.front ());
    r = true;
  } else {
    uint i, j;

    for (i = 0; i < lid.size (); i++) {
      // printf ("stable? %qx %qx %qx, %qx\n", lid[i], lid[(i+1)%lid.size ()], 
      //      dfingers[0].id, finger);
      if (ConsistentHash::betweenrightincl (lid[i], lid[(i+1) % lid.size ()], 
					    finger)) {
	break;
      }
    }

    for (j = 0; j < n; j++) {
      if (lid[i] != dfingers[j].id) {
	break;
      }
      i = (i + 1) % lid.size ();
    }

    if (j == n) {
      r = true;
    } else {
      printf ("loop %u,%qx not stable: finger %d should be %qx but is %qx\n", 
	      me.ip, me.id, j, lid[i], dfingers[j].id);
      for (uint l = 0; l < dfingers.size (); l++) {
	printf ("succ debruijn finger %d is %qx\n", l, dfingers[l].id);
      }

    }
  }
  return r;
}

bool
Koorde::stabilized (vector<ConsistentHash::CHID> lid)
{
  bool r = Chord::stabilized (lid);
  if (!r) return false;
  r = debruijn_stabilized (debruijn, fingers, lid);
  if (r && (resilience > 0)) 
    r = debruijn_stabilized (debruijnpred, resilience, lid);
  return r;
}

void 
Koorde::initstate()
{
  vector<IDMap> ids = ChordObserver::Instance(NULL)->get_sorted_nodes();
  uint nnodes = ids.size ();
  IDMap tmp;
  tmp.id = debruijn;
  uint pos = upper_bound(ids.begin(), ids.end(), tmp, Chord::IDMap::cmp) - ids.begin();
  for (uint i = 0; i < resilience; i++) 
    loctable->add_node(ids[(pos-i)%nnodes], false, true);
  debruijnpred = debruijn - ((Chord::CHID)-1/nnodes)*resilience;
  tmp.id = debruijn;
  pos = upper_bound(ids.begin(), ids.end(), tmp, Chord::IDMap::cmp) - ids.begin();

  CDEBUG(3) << "koorde_init_state sz " << loctable->size() << " debruijn "
    << printID(debruijn) << ids[pos%nnodes].ip << ","
    << printID(ids[pos%nnodes].id) << endl;

  for (uint i = 0; i < fingers; i++) 
    loctable->add_node(ids[(i+pos)%nnodes],false,true);

  Chord::initstate();
}

void
Koorde::debruijn_dump (Chord::CHID finger, uint n)
{
  IDMap d = loctable->pred (finger);
  vector<IDMap> sc = loctable->succs (d.id + 1, n - 1);
  vector<IDMap> dfingers;

  dfingers.clear ();
  dfingers.push_back (d);
  for (uint i = 0; i < sc.size (); i++) {
    dfingers.push_back (sc[i]);
    // printf ("succ finger %d is %qx\n", i, sc[i].id);
  }
  assert (dfingers.size () <= n);

  printf ("Koorde (%5u,%16qx) finger %16qx\n", me.ip, me.id, finger);
  for (uint m = 0; m < dfingers.size (); m++) {
    printf ("finger %3u is %16qx\n", m, dfingers[m].id);
  }
}

void
Koorde::dump ()
{
  isstable = true;

  Chord::dump ();
  debruijn_dump (debruijn, fingers);
  if (resilience > 0) debruijn_dump (debruijnpred, resilience);
}