chord.mac

这是一个著名的应用层组播中间件的源码

//Copyright (c) 2004, Charles Killian, Adolfo Rodriguez, Dejan Kostic, Sooraj Bhat, and Amin Vahdat
//All rights reserved.
//
//Redistribution and use in source and binary forms, with or without
//modification, are permitted provided that the following conditions are met:
//
//   * Redistributions of source code must retain the above copyright
//     notice, this list of conditions and the following disclaimer.
//   * Redistributions in binary form must reproduce the above copyright
//     notice, this list of conditions and the following disclaimer in
//     the documentation and/or other materials provided with the
//     distribution.
//   * Neither the names of Duke University nor The University of
//     California, San Diego, nor the names of its contributors
//     may be used to endorse or promote products derived from
//     this software without specific prior written permission.
//
//THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
//AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
//IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
//DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE
//FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
//DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
//SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
//CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
//OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
//USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

/**
   Implementation of the Chord protocol
   Adolfo Rodriguez

   Notes (limitations):
   - 32-bit hash addresses
   - Fixed stabilize and fix-finger timeouts
   - Recursive forwarding
   - Successor list of size 2
   - fix_fingers invokes update_others, which is not consistent with published chord or current implementations
**/

#include "scribe-ext.h"
#include "macedon_cache.h"

protocol chord

addressing ip

trace_off

constants {
  int BITS = 32;
  double STABLE_TIMEOUT = 0.3;  // stabilize timer
  int FIX_TIMEOUT = 1; // fix_fingers timer
  //  int FIX_TIMEOUT = 20;
  int JOIN_TIMEOUT = 10;  // initial join timer
  int PRINT_TIMEOUT = 2; // for debugging
}

states {
  joining;
  joined;
}

neighbor_types {
  finger 1 {
    int start; // the range for this finger
    int hashf; // the hash of this finger
  }
}

transports {
  TCP HIGH;
  TCP LOW;
}

messages {
  HIGH find_pred {  // used to find the predecessor of an ID
    int id;  // the hash in question (the one whose predecessor we want)
    int who; // who originated the query (IP)
  }
  HIGH find_pred_reply {  // response to find_pred
    int pred_id;  // the hash of the predecessor
    int succ; // the successor's IP
    int succ_id;  // the hash of the successor
    finger pf BITS;  // my routing table
  }
  HIGH get_pred { // queries a node for its predecessor and successor
  }	 
  HIGH get_pred_reply { // returns the requested information
    int pred;
    int pred_id;
    int succ;
    int succ_id;
  }
  HIGH update_pred { // instructs a node to potentially update its predecessor
    int id;  // hash of the sending node
  }
  HIGH update_others { // instructs others to potentially correct an entry
    int id; // where i am sending this request
    int node; // sender's hash
    int who; // sender's IP address
  }
  HIGH update_ft { // an "other" sends this msg back to its predecessor to
    // checks its routing table for any related changes
    int node; // sender's hash
    int who; // sender's IP address
  }
  LOW data {
    int receiver_hash;
    int sender_hash;
    int sender_ip;
    int priority;
  }
  LOW data_path_req { // solicits a mapping message
    int receiver_hash;
    int sender_hash;
    int sender_ip;
    int priority;
  }
  HIGH mapping { // Mappings are used to populate the location cache
    int low_hash;
    int high_hash;
    int map_ip;
  }
}


state_variables {	 
  int myhash;
  // disable failure detection for now
  //  fail_detect finger myfinger BITS;
  //  fail_detect finger pred;
  finger myfinger BITS; // routing table
  finger pred; // predecessor entry
  mac_cache map; // location cache
  finger successors_succ; // for error recovery
  timer stabilize STABLE_TIMEOUT;
  timer fix_fingers FIX_TIMEOUT;
  timer printer PRINT_TIMEOUT;
  timer join;
}	  

transitions {

  init API init {  // called to initialize protocol
    int i, id, start, adjusted, max = (int)pow(2,BITS);
    neighbor_finger *myent;

    myhash = hashof(me);  
    print_addr(myhash);
    neighbor_add(pred,me);  // just me so far in ring
    myent = neighbor_entry(pred,me);
    myent->hashf = myhash;

    for (i=0; i<BITS; i++) {  // initializes all finger entries to me
      neighbor_add(myfinger[i],me);
      myent = neighbor_entry(myfinger[i],me);
      myent->hashf = myhash;
      start = (int)pow(2,i);  
      adjusted = start + myhash;
      myent->start = adjusted % max;  // the start address of this finger
    }
    timer_resched(printer, PRINT_TIMEOUT);  // for debugging
    if (source_ == me) { // i am the bootstrap
      state_change(joined);  // consider self joined
      replay_experiment();
      timer_resched(stabilize, STABLE_TIMEOUT);  // schedule maintenance
      timer_resched(fix_fingers, FIX_TIMEOUT);
    }
    else { // not the bootstrap
      state_change(joining);  // just joining
      id = myhash + 1; // route find pred to just beyond me
      replay_init();
      route_find_pred(source_, id, me, 0, 0, -1);
      timer_resched(join, JOIN_TIMEOUT);  // my request could get lost
    }
    upcall_notify(pred,NBR_TYPE_PEER);
  }

  joining timer join { // retry if my request was lost
    int id = myhash + 1;
    route_find_pred(source_, id, me, 0, 0, -1);
    timer_resched(join, JOIN_TIMEOUT);
  }


  joined recv find_pred {
    int succ, succ_hash, nextguy, done;
    neighbor_finger *myent;
    done = 0;
    myent = neighbor_random(myfinger[0]);  // gets the successor

    if (isinrange(field(id), myhash, myent->hashf)) { // arrived at destination
      done = 1;
    }

    if (done == 1) {
      succ = myent->ipaddr;
      succ_hash = myent->hashf;
      if (field(who) != me) { // i did not originate request
        route_find_pred_reply( field(who), myhash,
            succ, succ_hash, myfinger, 0, 0, -1);
      }
      // Note that this node's successor field is updated via stabilize timers
      //      else { // update entries if needed
      //	register_knowledge(succ, succ_hash);
      //      }  Adolfo: Probably not needed.
      return;
    }

    nextguy = nexthop(field(id));
    if (nextguy != me) { // route the find_pred
      route_find_pred( nextguy, field(id), field(who), 0, 0, -1);
    }
    else {
      exception(48);
    }
  }

  joining recv find_pred_reply {
    int i;
    int powow;
    neighbor_finger *predent, *myent;

    // my predecessor's old successor may suit me as route entry
    register_knowledge(field(succ), field(succ_id));
    // update p's old successor
    route_update_pred(field(succ), myhash, 0, 0, -1); 
    neighbor_clear(pred);  // set my new predecessor
    neighbor_add( pred, from);
    predent = neighbor_random( pred);
    predent->hashf = field(pred_id);
    state_change( joined); // now joined
    timer_resched( stabilize, STABLE_TIMEOUT); // schedule maintenance
    timer_resched( fix_fingers, FIX_TIMEOUT);
    replay_add( from);
    upcall_notify(pred,NBR_TYPE_PEER);
  }


  joined timer stabilize { // stabilize timer
    // responsible for correcting predecessor and successor values
    neighbor_finger *myent = neighbor_random(myfinger[0]); // successor
    if (myent->ipaddr != me) 
      route_get_pred( myent->ipaddr, 0, 0, -1);  // query successor for pred
  }


  joined recv get_pred {
    neighbor_finger *predent = neighbor_random(pred);
    neighbor_finger *succent = neighbor_random(myfinger[0]);
    route_get_pred_reply( from, predent->ipaddr, predent->hashf, succent->ipaddr, succent->hashf, 0, 0, -1); // let querying node know whom my successor
    // and predecessor are
  }

  joined recv get_pred_reply {
    int old_start, i;
    neighbor_finger *myent = neighbor_random(myfinger[0]);

    if (myent->ipaddr != from) 
      return;  // make sure reply is from successor

    neighbor_clear(successors_succ);
    neighbor_add(successors_succ, field(succ_id));
    if (field(pred_id) == myhash) 
      return;  // if he thinks i am his predecessor, we are done

    register_knowledge(field(pred), field(pred_id)); // update fingers
    myent = neighbor_random(myfinger[0]); // successor has likely changed
    route_update_pred( myent->ipaddr, myhash, 0, 0, -1); // let him know of me
  }