scribems.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 SCRIBE protocol
 */

#include "adolfo_filter.h" //includes the declaration of the adolfo filter.
#include "scribe-ext.h" //scribe-additional declarations of types and integers.

//These macros are used to turn additional debugging/trace messages on and off.
//#define SCRIBE_TRACE
//#define ANYCAST_TRACE
//#define HEARTBEAT_TRACE

#define UNDEFINED 0 //This constant a temporary use to mark an empty macedon_key.  Requires nodes not to have this value as their id.

protocol scribems uses pastry

addressing hash

trace_off

constants {
  int BITS = 32; //Number of bits in a node identifier.
  int MAX_SESSIONS = 20; //Number of scribe trees to maintain state for.
  int MAX_CHILDREN = 26; //Number of children per tree to maintain state for.
  int MAX_ROOTPATH = 50; //Maximum depth of a path-to-root for a scribe tree.
  int MAX_VNODES = 101;   //Maximum number of visited nodes to maintain during an anycast traversal

  //NOTE: The following constants are used by the heartbeat timer 
  //NOTE: As long as the timeouts are close in granularity it makes sense to only have one.  Otherwise,
  // more may be needed.
  int TIMER_INTERVAL = 5; //Heartbeat timer interval
  int DROP_CHILD_TICKS = 10; //Number of timer transitions before dropping a child who doesn't renew its joined status.
  int RENEW_TICKS = 6; //Number of timer transitions before renewing join status to parents in subscribed trees.
  int HB_MISS_REPAIR = 2; //Number of timer transitions with no heartbeat received before searching for a new parent.
  int HB_TICKS = 0; //Number of timer transitions skipped (-1) before sending hb

  //The interval of the printer timer.
  int DEBUG_INTERVAL = 4;
}

states {
  joined ready;
}

neighbor_types {
  //Note that definitions here are of _types_, not actual variables.  This is similar to 
  //  defining classes.
  children MAX_CHILDREN { //A set of children not to exceed size MAX_CHILDREN.  Defined for use in "session".
    int ticks_since_join; //Used to maintain the number of hbtimer ticks since this child last sent a join message
  }
  uppath MAX_ROOTPATH;    //A set of the nodes to the root of the tree.  Defined for use in "session".
  session MAX_SESSIONS {  //The state type for sessions (scribe trees).  Each tree is a neighbor wiht these contents.
    children kids;        //The children of this node in the tree.
    macedon_key parent;   //The parent of this node in the tree. BUG: Until a general macedon_key class is defined, this approach uses 0 as NULL or undefined.  Thus a node with id 0 will be confused.
    uppath pathtoroot;    //The path to the root of this tree.
    int subscribed;       //If subscribed==1, then this node should deliver data to the upper layer/app.
    int ticks_since_heartbeat;  //Keeps track of how long its been (in hbtimer ticks) since a heartbeat has been received for this tree.
    int tick_count_for_renew;   //Keeps track of how long (in hbtimer ticks) since this node sent a join to its parent for this tree. 
    int tick_count_without_data;//Keeps track of how long (in hbtimer ticks) since this node has sent data to its children (used to determine if a heartbeat needs to be sent to its children)
    bool isRoot;          //Stores whether this node is the root of the tree (to prevent downcalls all the time.
  }
  visited MAX_VNODES {     //Stores a list of nodes visited during an anycast traversal.
    int color;              //0-white, 1-grey, 2-black
  }
}

messages {
  PRIORITY_HIGH create_group {
    int group_id;
  }
  PRIORITY_HIGH join {
    int group_id;
  }
  PRIORITY_HIGH leave {
    int group_id;
  }
  PRIORITY_LOW data {
    int group_id;
    int priority;
  }
  PRIORITY_HIGH pushdown {  //Sent to a child when it is being "kicked out" of this node's children list.
                            //  Contains a list of the pushed down node's siblings.  It will try to join  
                            //  under one of these.
    int group_id;
    children siblings;
  }
  PRIORITY_HIGH heartbeat { //Heartbeats will be sent periodically in the absence of tree data to maintain
                            //  freshness of state.  Each node sends the list of nodes in the path to 
                            //  the root of the tree to its children to prevent loops.
    int group_id;
    uppath pathtoroot;
  }
  PRIORITY_HIGH anycast {   //Message type to be sent for anycast requests.  This is performed via a DFS
                            //  of the tree.  Each node in the tree will pass the message up to the 
                            //  higher layer/app to determine if the anycast should be continued or 
                            //  stopped using upcall_forward.  
    int group_id;
    int from_addr;
    visited dfsState;
  }
}

state_variables {	          //These are not type definitions but actual variable declarations.
  int myhash;               //used to store the local node's node identifier
  session my_sessions;      //The set of scribe trees we are members of.
  adolfo_filter master;     //This filter collects stats on all data received. (for bookkeeping & evaluation)
  adolfo_filter master_useful;  //This filter collects stats on useful data received. (""  ""    ""  ""   "")
  timer hbTimer TIMER_INTERVAL; //This is the heartbeat timer, used for periodic table maintenance
  timer printer DEBUG_INTERVAL; //This is the printer timer, used to debug and dump state periodically.
}	  

transitions {

  //The automatic transition which occurs at the beginning.
  init API init {
    myhash = hashof(me);
    timer_resched(hbTimer, TIMER_INTERVAL); //All timers have to be scheduled the first time.
    timer_resched(printer, DEBUG_INTERVAL);
  }

  //Message: create_group
  //Action: recv
  //
  //If state already exists for this group, simply set isRoot to true, upstream 
  //  maintenance state to 0, and clear out data about parents.
  //Otherwise, first add state for the group.
  //
  //Known Problems: Does not verify there is room for the new session.
  recv create_group {
    int mygroup = field(group_id);
    neighbor_session* temp;
    if(!neighbor_query(my_sessions, mygroup)) {
      neighbor_add( my_sessions, mygroup );
      upcall_notify(my_sessions, NBR_TYPE_TREE);
      temp = neighbor_entry(my_sessions, mygroup);
      temp->subscribed=0;
    } else {
      temp = neighbor_entry(my_sessions, mygroup);
      neighbor_clear(temp->pathtoroot);
      temp->parent = UNDEFINED;
    }
    temp->isRoot=true;
    temp->ticks_since_heartbeat = 0;
    temp->tick_count_for_renew = 0;
    temp->tick_count_without_data = 0;
  }

  //Message: join
  //Action: forward
  //
  //Comments inline.
  //CASE 1: Forward join because this node sent it
  //CASE 2: Reject joining node based on loop test.
  //CASE 3: Accept new child
  //CASE 4: Reject child because there are too many children.
  //
  forward join {
    int mygroup;
    neighbor_session *temp;

    //CASE 1: Forward join because this node sent it
    //upcall_forward will be called on all nodes, including origin node.  Thus, if this node
    //is the origin node, automatically forward it.
    if (from == myhash) 
      return 0; 

    mygroup = field(group_id);
    if (neighbor_query( my_sessions, mygroup )) {
      //retrieve state for use in this transition
      temp = neighbor_entry( my_sessions, mygroup );

      if(neighbor_query(temp->pathtoroot, from)) {
        //CASE 2: Reject joining node based on loop test.
        //If the joining node is in our path to the root, forward it on, because we cannot allow loops.
        debug_macro("WARNING: Forwarding a join (fr:%.8x to:%.8x) because it is in our path to root.\n", from, dest);
        return 0; //Continue forwarding the join, because we cannot take the node.
      }
    }
    else {
      //add state for the new session, joining the group
      neighbor_add( my_sessions, mygroup );
      upcall_notify(my_sessions, NBR_TYPE_TREE);  //notify the upper layer of the state change.
      //FIXME: If upcall_notify causes the group to be removed -- what happens here?
      temp = neighbor_entry( my_sessions, mygroup );
      temp->subscribed = 0;
      downcall_ext(temp->isRoot, TEST_OWNER, (void*)&mygroup);
      temp->ticks_since_heartbeat = 0;
      temp->tick_count_for_renew = 0;
      temp->tick_count_without_data = 0;
      if(temp->isRoot <= 0) {
        //Need to join the new group
        route_join(mygroup, mygroup, 0, 0, -1);
      }
    }

    if (!neighbor_query( temp->kids, from )) {
      if(neighbor_space(temp->kids)) {
        //CASE 3: Accept new child
        neighbor_add( temp->kids, from );
        routeIP_heartbeat(from,mygroup,temp->pathtoroot, 0, 0, -1);
        debug_macro("Scribe: added kid %.8x for group %.8x, now have %d\n", from, mygroup, neighbor_size(temp->kids));
        //group_change defined in scribe_ext.h
        group_change *gchg = new group_change;
        gchg->groupId = mygroup;
        gchg->childId = from;
        gchg->newJoin = true;

        //Notify upper layer that the child has been added to the given group.
        upcall_ext(GROUP_CHANGE, (void*)gchg);

        delete gchg;
      } else {
        //CASE 4: Reject child because there are too many children.
        route_pushdown(from,temp->ipaddr,temp->kids, 0, 0, -1);
      }
    }
    if(neighbor_query(temp->kids,from)) {
      //Set the child's counter to 0 since we recieved a join message.
      neighbor_info(temp->kids,from,ticks_since_join) = 0;
    }
    return 1;
  }	 

  //Message: join
  //Action: recv
  //
  //If a join message is received at a node, this means it has already been forwarded by all 
  //  intermediate nodes, including this one.  If this happens frequently, it may be an 
  //  indicator of a more serious problem.  Likely indicates stale state.
  recv join {
    debug_macro("Scribe: WARNING: join failed! Join msg not accepted by any node. (fr: %.8x gr: %.8x)\n", from, field(group_id));
  }

  //Message: leave
  //Action: forward
  //
  //Process:
  //1 - Remove child if found (otherwise stale state somewhere)
  //2 - Conditionally notify higher layer
  //3 - If we have no remaining children and are not subscribed, leave the group. (notify upper layer)
  forward leave {
    // returns 0 if it should be forwarded
    int mygroup;
    neighbor_session *temp;

    if (from == myhash) return 0; //forward message if we sent it.

    mygroup = field(group_id);

    if (neighbor_query(my_sessions, mygroup)) {
      temp = neighbor_entry(my_sessions, mygroup);

      //remove "from" from the list of neighbor kids
      if (neighbor_query(temp->kids, from)) {
        if(true) { //will be replaced by if(upper layer is subscribed to this event type)
          group_change *gchg = new group_change;
          gchg->groupId = mygroup;
          gchg->childId = from;
          gchg->newJoin = false;

          upcall_ext(GROUP_CHANGE, (void*)gchg);

          delete gchg;
        }	
        neighbor_remove(temp->kids, from);
      }	

      if (neighbor_size(temp->kids) <= 0 && temp->subscribed == 0)
      { // now we check both for children and our own subscription.
        if( temp->parent != UNDEFINED ) {
          routeIP_leave (temp->parent, mygroup, 0, 0, -1);
        }
        neighbor_clear(temp->kids);
        neighbor_remove(my_sessions, mygroup);
        upcall_notify(my_sessions, NBR_TYPE_TREE);
      }
    } //else no processing done here (probably stale data)
    return 1;
  }

  //Message: leave
  //Action: recv 
  //
  //ASSERTion error.  Should not happen.
  recv leave {
    ASSERT(false);
  }

  //Message: data
  //Action: recv
  //
  //Process:
  //1 - Update bookkeeping filter
  //2 - If we do not have state for this tree, assume stale data, and route a leave to the source.
  //3 - Ask upper layer if we should forward this message (else done)
  //4 - Reset counters
  //5 - Deliver message to upper layer
  //6 - forward message to children.
  recv data {
    neighbor_session *tempsess;
    
    master.update();
    if (dest != field(group_id) && dest != myhash) {
      debug_macro("Scribe: WARNING: Dropped packet cause dest %.8x, group %.8x, me %.8x\n", dest, field(group_id), myhash);
    } else {
      if (neighbor_query(my_sessions, field(group_id))) {
        if(upcall_forward(field(group_id), msg, size, COMM_TYPE_MULTICAST) == 0) {
          tempsess = neighbor_entry(my_sessions, field(group_id));
          tempsess->tick_count_without_data = 0;
          tempsess->ticks_since_heartbeat = 0;
          if(tempsess->subscribed) {
            master_useful.update();
            upcall_deliver(msg, size, COMM_TYPE_MULTICAST);
          }
          //Update the parent neighbor type, if needed (need lock inside).