randtree.mac

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

  joined recv join_reply {
    debug_macro ("joined! join_reply from: %.8x\n", from);
    // Already joined, this is a duplicate join
    neighbor_rparent* existing = neighbor_random(papa);
    if (existing->ipaddr != from)
      route_remove(from, 0, 0, -1);  // Just remove yourself from this parent
  }

  joined recv join_redirect { 
    neighbor_rchildren * redir;
    if ( !neighbor_query(kids, field(who)) && neighbor_space(kids) ) {
      debug_macro ("taking redirected child: %.8x redirected by %.8x\n", field(who), from);
      // He is not my child and I have space for him
      neighbor_add(kids, field(who));
      upcall_notify(kids, NBR_TYPE_CHILDREN); // Notify upper layer of change
      route_join_reply(field(who), 0, 0, -1);
    }
    else if (neighbor_query (kids, field(who))) {
      // already our child, do nothing
    }
    else {  // must find someone to redirect to
      redir = neighbor_random(kids);
      route_join_redirect(redir->ipaddr, field(who), 0, 0, -1);
    }
  }

  joined recv remove { 
    // Child is requesting to be removed
    if (neighbor_query(kids, from)) {  // Only remove it if he is my child
      debug_macro ("asking to be removed from: %.8x\n", from);

      neighbor_remove(kids, from);
      upcall_notify(kids, NBR_TYPE_CHILDREN); // Notify upper layer of change
    }
  }

  recv data {
    // check to see if we should process this data
    master.update(); // Record that we got something
    int should_forward = upcall_forward(0, msg, size, field(comm_type));
    if (should_forward)
      return; // Leave if upper layer says not to take this msg
    master_useful.update(); // The data was useful
    upcall_deliver( msg, size, field(comm_type));

    if (field(comm_type) == COMM_TYPE_MULTICAST) {
      // forward to all children if it is multicast
      foreach_neighbor (neighbor_rchildren*, afriend, kids ) {
	// This function routes a "data" message to each child, 
	// using the child addr, the local addr, "send_comm", and 
	// "send_transport" as the headers, and attaches the message 
	route_data (afriend->ipaddr, field(comm_type), field(priority), msg, size, field(priority));
      }
    }
    else if (field(comm_type) == COMM_TYPE_COLLECT) {
      // forward to parent if it is collect
      foreach_neighbor (neighbor_rparent*, afriend, papa ) {
	route_data (afriend->ipaddr, field(comm_type), field(priority), msg, size, field(priority));
      }
    }

  }      
  API transport_error {
    //CHIP TEMP
    if(transport_error == TCP_PEER_ALREADY_EXISTS || transport_error == TCP_PEER_ALREADY_EXISTS_OTHER) {
    } else {
      debug_macro("Neighbor %.8x transport error!\n", dest_addr);
      
      if (neighbor_query(papa, dest_addr)) {  // Is it my parent?
        debug_macro("Parent %.8x transport error!\n", dest_addr);
        neighbor_remove(papa, dest_addr);
        parent_suggestion_give_up = 0.0;
        state_change(joining);  // Retry joining at the root
        {
          route_join(source_, 0, 0, -1);  // Source is the bootstrap
        }
        forced_parent = source_;
        timer_resched (join, RANDTREE_JOIN_TIMEOUT);
      }
      else if(neighbor_query(kids, dest_addr)) {  // Is it my child
        debug_macro("Child %.8x transport_error!\n", dest_addr);
        neighbor_remove(kids, dest_addr);  
        upcall_notify(kids, NBR_TYPE_CHILDREN); // Notify upper layer of change
      }
    // automatic params: transport_error, dest_addr, port, known_lost
    transport_error_struct tr;
    tr.transport_error = transport_error;
    tr.dest_addr = dest_addr;
    tr.port = port;
    tr.known_lost = known_lost;

    upcall_ext(AUTOEXT_TRANSPORT_ERROR, &tr);
    }
  }
  API create_group {
    // For now, do nothing. In a real protocol, we would register the session.
  }

  API join {
    // For now, do nothing
  }

  API leave {
    // For now, ignore it. 
  }


  joined recv wean { 
    if (neighbor_query(papa, from)) {  // Is it my parent?
      sprintf(trace_buf_, "Parent %.8x wean!\n", from);
      trace_print();
      neighbor_remove(papa, from);
      state_change(joining);  // Retry joining at the root
      route_join(source_, 0, 0, -1);  // Source is the bootstrap
      timer_resched (join, RANDTREE_JOIN_TIMEOUT);
      forced_parent = source_;
    }
  }

  API error {
    // This function is called when the peer "neighbor" for which fail_detect was
    // request has failed.
    sprintf(trace_buf_, "Neighbor %.8x died!\n", neighbor);
    trace_print();
//      return;
    
    if (neighbor_query(papa, neighbor)) {  // Is it my parent?
      sprintf(trace_buf_, "Parent %.8x died!\n", neighbor);
      trace_print();
      neighbor_remove(papa, neighbor);
      route_remove(neighbor, 0, 0, -1);
      parent_suggestion_give_up = 0.0;
      state_change(joining);  // Retry joining at the root
      route_join(source_, 0, 0, -1);  // Source is the bootstrap
      timer_resched (join, RANDTREE_JOIN_TIMEOUT);
      forced_parent = source_;
    }
    else if(neighbor_query(kids, neighbor)) {  // Is it my child
      sprintf(trace_buf_, "Child %.8x died!\n", neighbor);
      trace_print();
      neighbor_remove(kids, neighbor);  
      upcall_notify(kids, NBR_TYPE_CHILDREN); // Notify upper layer of change
      route_wean(neighbor, 0, 0, -1);
    }
  }

  API route { // just route over unicast IP
    route_data (dest, COMM_TYPE_UNICAST, transport, msg, size, transport);
    return_code = macedon_sendret;
  }

  joined API multicast { // send the data to each child
    foreach_neighbor (neighbor_rchildren*, afriend, kids ) {
      route_data (afriend->ipaddr, COMM_TYPE_MULTICAST, transport, msg, size, transport);
    }
    return_code = 0;
  }

  joined API collect { // send the data to our parent in the tree
    neighbor_rparent *mypa = neighbor_random (papa);
    if (mypa) {
      // Only forward it I have a parent
      route_data (mypa->ipaddr, COMM_TYPE_COLLECT, transport, msg, size, transport);
    }
    return_code = 0;
  }

  API downcall_ext {
    //Takes int operation, void* arg
    clean_suggestions();
    switch(operation) {
      case SUGGESTION_NBR_SIZE: 
        {
          suggestion_nbr_size* sugg = (suggestion_nbr_size*)arg;
          debug_macro("Changing MAX_CHILDREN based on hint. curr %d prev %d\n",CURRENT_MAX_CHILDREN,sugg->size);
          CURRENT_MAX_CHILDREN = sugg->size;
          break;
        } 
      case SUGGESTION_NBR:
        {
          suggestion_nbr* sugg = (suggestion_nbr*)arg;
          if(!neighbor_query(papa, sugg->neighbor_id) && sugg->neighbor_type == NBR_TYPE_PARENT && sugg->action == ADD_NEW_NBR) {
            state_change(joining);
            route_join(sugg->neighbor_id, 0, 0, -1);
            timer_resched (join, RANDTREE_JOIN_TIMEOUT);
            forced_parent = sugg->neighbor_id;
            parent_suggestion_give_up = curtime + sugg->suggestion_period;
          } else if(sugg->neighbor_type == NBR_TYPE_CHILDREN && sugg->action == ADD_NEW_NBR) {
            if(!neighbor_query(kids, sugg->neighbor_id) && !neighbor_query(suggestions, sugg->neighbor_id)) {
              neighbor_add(suggestions, sugg->neighbor_id);
              neighbor_info(suggestions, sugg->neighbor_id, end_time) = curtime+sugg->suggestion_period;
            }
          } else if(sugg->neighbor_type == NBR_TYPE_CHILDREN && sugg->action == DROP_NBR) {
            if(neighbor_size(kids) > 0 && sugg->neighbor_id == RANDOM_NEIGHBOR) {
              neighbor_rchildren* child = neighbor_random(kids);
              route_remove(child->ipaddr, 0, 0, -1);
              neighbor_remove(kids, child->ipaddr);
            }
            else if(neighbor_query(kids, sugg->neighbor_id)) {
              route_remove(sugg->neighbor_id, 0, 0, -1);
              neighbor_remove(kids, sugg->neighbor_id);
            }
          }
          break;
        }
      default: 
        {
          debug_macro( "Unrecognized Extensible Downcall Made (type=%d).\n",operation);
          return -1; //CALL UNRECOGNIZED
        }
    }
    return 0;
  }

  timer printer {
    extern MACEDON_Agent *globalmacedon;
    if (globalmacedon != this)  // am i the highest layer
      return; 
    if ( ( parameters.getint("streaming_time") == -1.0 ||
	   curtime > time_booted + parameters.getint("streaming_time")) )
      printf("%s %f %d REPLAY_RANDTREE_BANDWIDTH %d %d %d %d %d %d\n", 
	     get_hostname(), Scheduler::instance().clock(), pthread_self(),
	     (int) master.get_value(),
	     0,   
	     (int) master_useful.get_value(),
	     (int) master.get_value(),
	     (int) master_useful.get_value(),
	     0
	     );
  }

}

/*
 * Define routines for reusability. These can only be called by
 * transitions from this protocol. (Or other routines of this protocol).
 * These are defined just like C++ functions.
 */
routines {  

  void clean_suggestions() {
    foreach_neighbor(neighbor_suggestion_list*, suggestion, suggestions) {
      if(suggestion->end_time < curtime) {
        neighbor_remove(suggestions, suggestion->ipaddr);
      }
    }
  }

  /*
   * This function shows how by leveraging a general purpose language,
   * MACEDON does not limit the possible actions a protocol can take.
   * In this case, it is reading in a parent file. This same code
   * could be written as part of a transition, but is here to simplify
   * the transition code.
   */
  void read_parent_file () 
  {
    struct hostent *phe;
    struct hostent *mhe;
    struct in_addr paddr;
    struct in_addr myaddr;
    FILE *myfile;

    int i=0;
    char name[200];
    char parent[200];
    char rest[1024];
    char whole_line[1024];
    char *parent_filename = parameters.getstr("parent_file");
    if (!parameters.getint("quiet")) {
      if(parent_filename != NULL) {
	printf("forced parent_filename: %s\n", parent_filename);
      }
      else {
	printf("parent_filename was NULL\n");
	return;
      }
    }

    forced_children=0;
    myfile = (FILE *)fopen(parent_filename, "r");
    if (!myfile) {
      if (!parameters.getint("quiet")) {
	printf("nonexistent forced parent_file  %s\n", parent_filename);
      }
      return;
    }
    while (fgets(whole_line, 512, myfile)!=NULL) {
      int successful = 0;
      if ((successful =sscanf(whole_line, "%s %s %[^\n]s", 
              &name, &parent, &rest))!= 2) {
        continue;
      }   
      if ((mhe = gethostbyname(name)) == 0) {
        printf("Bad host lookup.\n");
        exit(24);
      }
      memcpy(&myaddr, mhe->h_addr_list[0], sizeof(struct in_addr)); 

      if ((phe = gethostbyname(parent)) == 0) {
        printf("Bad host lookup.\n");
        exit(24);
      }

      memcpy(&paddr, phe->h_addr_list[0], sizeof(struct in_addr));

//        printf("forced parent line: %s %.8x %s %.8x\n", name, myaddr, parent, paddr);

      if (paddr.s_addr == me) {   // i am parent
        forced_children++;
      }
      else if (myaddr.s_addr == me) {   // i am kid
        forced_parent = paddr.s_addr;
        sprintf(trace_buf_, "forced_parent: %.8x\n", forced_parent);
        trace_print();
      }
    }
    fclose(myfile);
//      if ( source_ != forced_parent )
//  	exit(0);

  }

}