overcast.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 Overcast protocol
 *
 *  Adolfo Rodriguez
 */

#include "adolfo_filter.h"

protocol overcast

addressing ip

trace_off

constants {
  int OVERCAST_MAX_CHILDREN = 500;
  int OVERCAST_PAYLOAD = 1000;
  int OVERCAST_SIBLINGS_TO_PROBE = 5;   // how many I can probe in one round
  int OVERCAST_NUM_PROBES = 10;
  int OVERCAST_TRASH_PROBES = 20;
  int OVERCAST_MIN_PROBES = 9;
  int OVERCAST_STOP_PROBE = 14;
  int OVERCAST_INFO_INTERVAL = 1;
  int OVERCAST_REASON_PROBING = 51;
  int OVERCAST_REASON_INVALID = 52;
  double OVERCAST_GOOD_ENOUGH = 0.90;
  int OVERCAST_JOIN_RETRY=2;
  int DEBUG_INTERVAL = 4; //The interval of the printer timer.
}

node_types {
  source;
  receiver;
}

states {
  joining;
  probing;
  probed;
  joined;
}

neighbor_types {
  oparent 1 {
    int probes_to_get;
    double probing_time;
    double start_probing_time;
    int lock_requested;
  }
  ochildren OVERCAST_MAX_CHILDREN {
    int probes_to_get;
    double probing_time;
    double start_probing_time;
    int lock_requested;
  }
}

transports {
  TCP HIGHEST;
  TCP HIGH;
  TCP MED;
  TCP LOW;
  UDP BEST_EFFORT;
}

messages {
  BEST_EFFORT join {
  }
  HIGHEST join_redirect {
    int who;
  }
  HIGHEST join_reply {
    int response;
  }
  HIGHEST add {
  }
  HIGHEST remove {
  }
  HIGHEST probe_request {
    int parent;
    int synch;
    double wait_time;   // how long to wait before probing
  }
  HIGHEST probe_denied {
    int reason;
    int synch;
  }
  LOW probe {
    int sequence;
    int synch;
    int trash;
  }
  HIGHEST probe_done {
  }
  MED parent_info {
    oparent daddy;
    ochildren brothers;
    int height;
    int enable_notify;
    int num_uncles;
  }
  LOW data {
    int receiver;
    int sender;
    int comm_type;
    int transport;
  }
}

state_variables {	 
  oparent papa;
  ochildren kids;
  oparent grandpa;
  ochildren brothers;
  int num_siblings;   // just a current guess
  timer printer DEBUG_INTERVAL; //This is the printer timer, used to debug and dump state periodically.
  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. (""  ""    ""  ""   "")

  // forced tree support
  int forced_parent;
  int forced_children;

  // for debugging
  int height;
  int joined_nodes;

  int notify_on;  // lets people know they should start reporting changes

  // used by probed node
  double probe_skip;

  // used by probing node
  int probed_node;
  int probes_to_send;

  // used by both probed and probing node
  int probe_synch;

  timer sendinfo OVERCAST_INFO_INTERVAL;
  timer keep_probing;
  timer stop_probe;
  timer probe_requester;
  timer join;
}	  

transitions {

  init API init {
    timer_resched(sendinfo, OVERCAST_INFO_INTERVAL);
    timer_resched(printer, DEBUG_INTERVAL);
    joined_read_parent_file();
    if ( source_ == me ) {
      if (forced_children)
        notify_on = 1;
      state_change(joined);
      replay_experiment();
      height = 0;
    }
    else {
      state_change(joining);
      replay_init();
      timer_resched (join, OVERCAST_JOIN_RETRY);
      if (!forced_parent) {
        route_join(source_, 0, 0, -1);
      }
      else {
        notify_on = 1;
        route_join (forced_parent, 0, 0, -1);
      }
    }
  }

  !(joining|init) recv join {
    neighbor_ochildren * redir;
    if ( !neighbor_query(kids, from) && neighbor_space(kids) ) {
      debug_macro("ADD child join %.8x\n", from);
      neighbor_add(kids, from);
      if (notify_on) {
        upcall_notify(kids, NBR_TYPE_CHILDREN); 
      }
      route_join_reply(from, 1, 0, 0, -1);
      forced_children--;
      if (source_ == me)
        joined_nodes++;
      if (joined_nodes == num_nodes_-1) {
        debug_macro("REPLAY all %d nodes joined.\n", num_nodes_);
      }
      else {
        debug_macro("REPLAY seen %d nodes so far (%.8x).\n", joined_nodes, from);
      }
    }
    else if (neighbor_query (kids, from)) {
      route_join_reply(from, 1, 0, 0, -1);
    }
    else {
      double hold_time = parameters.getdouble("hold_time");
      if ( ( hold_time != -1.0 &&   // we passed the hold time without all joined
            curtime > time_booted + hold_time) ) {
        redir = neighbor_random(kids);
        route_join_redirect(redir->ipaddr, from, 0, 0, -1);
        if (source_ == me)
          joined_nodes++;
        if (joined_nodes == num_nodes_) {
          debug_macro("REPLAY all %d nodes joined with haste enabled.\n", num_nodes_);
        }
        else {
          debug_macro("REPLAY seen %d nodes so far, haste enabled.\n", joined_nodes);
        }
      }
      else {
        route_join_reply(from, 0, 0, 0, -1);
      }
    }
  }

  joining recv join { 
    route_join_reply(from, 0, 0, 0, -1);
  }

  !(joining|init) recv add {
    if (forced_parent && forced_children) {
      route_join_reply(from, 0, 0, 0, -1);
    }
    else if ( !neighbor_query(kids, from) && neighbor_space(kids) ) {
      neighbor_add(kids, from);
      if (notify_on) {
        upcall_notify(kids, NBR_TYPE_CHILDREN); 
      }
      route_join_reply(from, 1, 0, 0, -1);
    }
    else if (neighbor_query (kids, from)) {
      route_join_reply(from, 1, 0, 0, -1);
    }
    else {
      route_join_reply(from, 0, 0, 0, -1);
    }
  }

  joining recv join_reply {
    //    debug_macro("join rep resp %d\n", field(response));
    if (field(response) == 1) {
      if (neighbor_size(papa)) {
        neighbor_oparent *pops = neighbor_random(papa);
        route_remove(pops->ipaddr, 0, 0, -1);
        replay_remove(pops->ipaddr);
        any_unlock_nodes();
        debug_macro("Move from %.8x %f\n", pops->ipaddr, pops->delay);
        neighbor_clear(papa);
      }
      replay_add(from);
      neighbor_add(papa, from);
      state_change (joined);
      neighbor_oparent *pops = neighbor_random (papa);
      if (neighbor_query (brothers, from)) {
        neighbor_ochildren *newp = neighbor_entry (brothers, from);
        pops->delay = newp->delay;
      }
      debug_macro("Move to %.8x %f\n", from, pops->delay);
      if (notify_on) {
        upcall_notify(papa, NBR_TYPE_PARENT); 
      }
    }
    else {
      if (neighbor_size(papa)) { 
        debug_macro("New parent rejected.  Returning to joined state.\n");
        state_change(joined);
        any_unlock_nodes();
      }
      else {
        debug_macro("New parent rejected.  Setting retry timer.\n");
        timer_cancel (join);
        timer_resched (join, OVERCAST_JOIN_RETRY);
      }
    }
  }

  joining timer join {
    //    debug_macro("Trying to join.\n");
    if (!forced_parent) {
      route_join(source_, 0, 0, -1);
    }
    else {
      route_join (forced_parent, 0, 0, -1);
    }
    timer_resched (join, OVERCAST_JOIN_RETRY);
  }

  joined recv join_redirect {
    neighbor_ochildren * redir;
    if ( neighbor_space(kids) ) {
      neighbor_add(kids, field(who));
      if (notify_on) {
        upcall_notify(kids, NBR_TYPE_CHILDREN); 
      }
      route_join_reply(field(who), 1, 0, 0, -1);
      forced_children--;
    }
    else {  // must find someone to redirect to
      redir = neighbor_random(kids);
      route_join_redirect(redir->ipaddr, field(who), 0, 0, -1);
    }
  }

  !joined recv join_redirect {
    route_join_redirect(from, field(who), 0, 0, -1);    // send it back
  }

  !init recv remove {
    if (neighbor_query(kids, from) ) {
      neighbor_remove(kids, from);
      if (notify_on) {
        upcall_notify(kids, NBR_TYPE_CHILDREN); 
      }
    }
  }

  probed recv probe_denied {
    if (probe_synch != field(synch))
      return;
    if (!neighbor_query(papa, from) && 
        !neighbor_query(grandpa, from) &&
        !neighbor_query(brothers, from) )
      return;
    debug_macro("Probe: denied from %.8x\n", from);
    timer_cancel (probe_requester);
    neighbor_oparent *pops = neighbor_random (papa);
    if (pops->lock_requested && neighbor_query(papa, from)) {
      // the parent request fail, this is fatal
      pops->lock_requested = 0;
      any_unlock_nodes();
      state_change(joined);
      return;
    }
    if (neighbor_size(grandpa)) {
      neighbor_oparent *gramps = neighbor_random(grandpa);
      if (gramps->lock_requested && neighbor_query(grandpa, from)) {
        gramps->lock_requested = 0;
        neighbor_clear(grandpa);
      }
    }
    int nuke_bro=0;
    foreach_neighbor (neighbor_ochildren*, hermano, brothers) {
      if (hermano->lock_requested && hermano->ipaddr == from) {
        hermano->lock_requested = 0;
        nuke_bro = hermano->ipaddr;
      }
    }
    if (nuke_bro) {
      neighbor_remove (brothers, nuke_bro);
    }
    probed_request_more_probes();
  }

  !joined recv probe_request {
    route_probe_denied(from, OVERCAST_REASON_PROBING, field(synch), 0, 0, -1);
  }

  joined recv probe_request {
    if ( (me == field(parent) &&     // he thinks I am his parent
          neighbor_query(kids, from)) ||   // and I am 
        neighbor_query(kids, field(parent)) || // I am his grandpa
        neighbor_query(papa, field(parent)) ) { // we are siblings
      state_change (probing);
      probes_to_send = OVERCAST_NUM_PROBES+OVERCAST_TRASH_PROBES;
      probed_node = from;
      debug_macro("Probe: locked by %.8x\n", probed_node);
      probe_synch = field(synch);
      if (field(wait_time > 0.0)) {
        debug_macro("Probe: sched initial keep for %.8x, time %f\n", from, field(wait_time));
        timer_resched(keep_probing, field(wait_time));
      }
      else
        probing_send_some_probes();
    }
    else {  // I don't know this guy
      route_probe_denied(from, OVERCAST_REASON_INVALID, field(synch), 0, 0, -1);
    }
  }

  probing timer keep_probing {
    //    debug_macro("Probe: keep probing timer hit with %d to go\n", probes_to_send);
    probing_send_some_probes();
  }

  probed recv probe {
    if (field(synch) != probe_synch)
      return;

    if (field(trash))
      return;

    if (neighbor_query(brothers, from)) {
      neighbor_ochildren *twin = neighbor_entry(brothers, from);
      twin->probes_to_get--;
      twin->probing_time = curtime;
      if (twin->start_probing_time == 0.0)
        twin->start_probing_time = curtime;
      //debug_macro("Probe: got seq %d need %d more from %.8x\n", field(sequence), twin->probes_to_get, from);
    }
    else if (neighbor_query(papa, from)) {
      neighbor_oparent *pops = neighbor_entry(papa, from);
      pops->probes_to_get --;
      pops->probing_time = curtime;