bullet.mac

来自「这是一个著名的应用层组播中间件的源码」· MAC 代码 · 共 1,126 行 · 第 1/3 页

    else {
      candidate_set<cand_bullet_summary_ticket>  empty(BULLET_MAX_CANDS);
      route_collect(from, field(sequence), 0, empty, 0, 0, -1);
      printf("RanSub: send empty coll seq %d to %x\n", field(sequence), from);
    }
    delete myticket;
  }

  joined recv collect {
    sprintf(trace_buf_, "RanSub: got coll seq %d(%d desc.) from %x\n", field(sequence), field(descendants), from);
    trace_print();
    if (sequence == field(sequence)) {
      if (neighbor_query(mychildren, from)) {
        neighbor_ransub_children *mykid = neighbor_entry(mychildren, from);
        mykid->gathered = field(mycollect);
        mykid->represents = field(descendants);
        mykid->seq = field(sequence);
      }
      joined_xmit_collect_if_need();
    }
  }     



  API notify {
    // I am being notified of changes in the tree below
    int i;

    if (type == NBR_TYPE_CHILDREN)
    {
      if (size > BULLET_MAX_CHILDREN) {
        printf("Exception: notified of too many children %d!\n", size);
        exit(54);
      }
      for (i=0; i<size; i++) {
        if (!neighbor_query(mychildren, neighbors[i])) {
          sprintf(trace_buf_, "Bullet: child %x added\n", neighbors[i]);
          trace_print();
          neighbor_add (mychildren, neighbors[i]);
          neighbor_ransub_children *just_added = neighbor_entry(mychildren, neighbors[i]);
          just_added->seq = -1;
          just_added->represents = 1;
          just_added->bandwidth_factor = 0.1;
        }
      }
      int remove[BULLET_MAX_CHILDREN];
      int rc=0;
      foreach_neighbor (neighbor_ransub_children *, kid, mychildren) {
        int found = 0;
        for (i=0; i<size; i++) {
          if (neighbors[i] == kid->ipaddr)
            found = 1;
        }
        if (found == 0)
        {
          remove[rc] = kid->ipaddr;
          rc++;
        }
      }
      for (i=0; i<rc; i++) {
        sprintf(trace_buf_, "Bullet: child %x removed\n", remove[i]);
        trace_print();
        neighbor_remove (mychildren, remove[i]);
      }
      joined_xmit_collect_if_need();
    }
    else {
      if (!neighbor_query(myparent, neighbors[0])) {
        sprintf(trace_buf_, "Bullet: parent changed to %x\n", neighbors[0]);
        trace_print();
        neighbor_clear(myparent);
        neighbor_add (myparent, neighbors[0]);
      }
    }
    upcall_notify(mychildren, NBR_TYPE_CHILDREN); // Notify upper layer of change
  }

} //transitions

routines {

  void joined_got_multi_data() {
    neighbor_ransub_parent *papa;
    neighbor_senders_to_me *givuh;
    double application_spacing = (double)(parameters.getint("data_packet_size")) *8/(1000.0*(double) parameters.getint("streaming_rate"));

    //#define TRACE_KEY_MOD 100
#ifdef TRACE_KEY_MOD
    if (got_key%TRACE_KEY_MOD == 0 && source_ != me) {
      sprintf(trace_buf_, "REPLAY_PACKET %s %s %d\n", get_hostname(got_from),  get_hostname(), got_key);
      trace_print();
    }
#endif

//      int useful = working_file.insert( got_key, 0 );
    int useful = working_file.insert( got_key, 0, (unsigned char *)got_msg, got_size );
    if (got_type == BULLET_DATA_PARENT 
        && neighbor_query(myparent, got_from)) {
      papa = neighbor_entry(myparent, got_from);
      parent_file.insert( got_key, 1);
      papa->arrivals.update();
      if (useful)
        papa->useful.update();
    }
    else if (got_type == BULLET_DATA_PARALLEL
        && neighbor_query(givers, got_from)) {
      givuh = neighbor_entry(givers, got_from);
      parallel_file.insert( got_key, 1);
      givuh->arrivals.update();
      if (useful)
        givuh->useful.update();
    }
    master.update();
    if (useful)
      {
	total_received++;
	master_useful.update();
	upcall_deliver( got_msg, got_size, COMM_TYPE_MULTICAST);
	sprintf(trace_buf_, "got useful data %d, type %d, from %.8x\n", got_key, got_type, got_from);
	cut_trace();
      }
    else {
      sprintf(trace_buf_, "got useless data %d, type %d, from %.8x\n", got_key, got_type, got_from);
      cut_trace();
      got_msg = 0;
      return;
    }

    if (!neighbor_size(mychildren)) {
      got_msg = 0;
      multicast_success_code = 0; // technically, we're OK
      return; // no kids, noone to send to
    }
    int current = 0;
    double myrandom = 0.0;
    int sent_this_interval=0;

    //    int strategy = BULLET_STRAT_SPLITTER;
    //    int strategy = BULLET_STRAT_DISJOINT;
    //    int strategy = BULLET_STRAT_STREAMTOALL;
    //    int strategy = BULLET_STRAT_TRY;
    int bullet_strategy = parameters.getint("bullet_strategy");
    foreach_neighbor (neighbor_ransub_children*, kid, mychildren ) {
      sent_this_interval += kid->sent;
      kid->density = (double)kid->represents/(double) (descendants);
      if (kid->density > 1.0)
        kid->density = 1.0;
    }

    unsigned int splitter = got_key % neighbor_size(mychildren);  // for splitter

    int sent_packet = 0;
    int assigned = neighbor_random(mychildren)->ipaddr;
    if (bullet_strategy == BULLET_STRAT_TRY) {
      // find out who this packet belongs and try to send it to him
      foreach_neighbor (neighbor_ransub_children*, kid, mychildren ) {
        if (sent_this_interval == 0 ||
            (((double)kid->sent / (double)sent_this_interval) < kid->density)) {
          assigned = kid->ipaddr;
        }
      }
      neighbor_ransub_children *assigned_ent = neighbor_entry (mychildren, assigned);
      route_papa_data( assigned,
			      COMM_TYPE_MULTICAST, 
			      got_key, got_msg, got_size, -1);
      if (!macedon_sendret) {
        assigned_ent->sent++;
        assigned_ent->kid_digest.insert(got_key);
        sent_packet = 1;
      }
      else {
        assigned_ent->not_sent++;
      }
    }

    foreach_neighbor (neighbor_ransub_children*, kid, mychildren ) {
      int should_send = 0;
      if (bullet_strategy == BULLET_STRAT_STREAMTOALL) {
        should_send = 1;
      }
      else if (bullet_strategy == BULLET_STRAT_SPLITTER) {
        if (splitter == 0) {
          should_send = 1;
        }
        splitter--;
      }
      else if (bullet_strategy == BULLET_STRAT_TRY) {
	if (kid->ipaddr != assigned ) {
	  if (!sent_packet) { // this packet has not been sent yet, I will try
	    should_send = 1;
	  }
	  else {  // have gas if I havent missed out sending more than a little amount of what I was supposed to send
	    int gas_to_send = (int)(1.0/kid->bandwidth_factor);
	    // the commented kid->sent allows us to send gas to a child even 
	    // if we didn't send it any packets it's supposed to own
	    // in this interval. Important for performance when using
	    // unencoded content
	    if ( // kid->sent &&
		 got_key % gas_to_send == 0 ) {
	      should_send = 1;
	    }
	  }
	}
      }

      should_send &= (!kid->kid_digest.contains(got_key));      

      if (should_send) {
	//	sprintf(trace_buf_, "Sending %d to %x strat %d\n", got_key, kid->ipaddr, bullet_strategy);
	//	trace_print();
	route_papa_data( kid->ipaddr, 
				COMM_TYPE_MULTICAST, 
				got_key, got_msg, got_size, -1);
	if (!macedon_sendret) {
	  if (!sent_packet) {
	    kid->sent++; 
	  }
	  else {  // was a dupe
	    kid->gas_sent++;
	    kid->bandwidth_factor = min(kid->bandwidth_factor+application_spacing/(double)BULLET_REFRESH_INTERVAL, 1.0);
	  }
	  kid->kid_digest.insert(got_key);
	  sent_packet = 1;
	}
	else  // the send didn't work
	  if (!sent_packet) {  
	    kid->not_sent++;   
	  }
	  else {    // was a dupe anyway
	    kid->bandwidth_factor = max(kid->bandwidth_factor-application_spacing/(double)BULLET_REFRESH_INTERVAL, 0.00001);
	  }
      }
      else {
        //	sprintf(trace_buf_, "Not sending %d to %x strat %d\n", got_key, kid->ipaddr, bullet_strategy);
        //	trace_print();
      }
    }
    got_msg = 0;
    multicast_success_code = !sent_packet;
  }

// ---------------------------------------------- 
// close_peers
// ---------------------------------------------- 

  void any_close_peers() {
    // close peers if need be
    int drop = 0;
    double drop_bw;
    double least_from_me=1.0;

    if (sequence%3 != 0)
      return;

    // first nuke the receiver who gets the least from me
    if ( !neighbor_space(getters) ) {   // only take one out if no space      
      foreach_neighbor (neighbor_receivers_from_me*, gettuh, getters) {
        if ( curtime - gettuh->start_time > 
            ((double)3 * BULLET_NORMAL_RANSUB) ) {
          // Allow at least 3 RanSub epochs for each guy to get data
          if (gettuh->reported_peer_bandwidth/gettuh->reported_total_bandwidth < least_from_me ) {
            drop = gettuh->ipaddr;
            least_from_me = gettuh->reported_peer_bandwidth/gettuh->reported_total_bandwidth;
          }
        }
      }
      if (drop) {
	neighbor_receivers_from_me *nuke = neighbor_entry(getters, drop);
	route_remove(nuke->ipaddr, 0, 0, -1);
	sprintf(trace_buf_, "Sender: Closing useless receiver %x with values %f %f %f\n", nuke->ipaddr, nuke->reported_peer_bandwidth, nuke->reported_total_bandwidth, nuke->start_time);
	trace_print();
	neighbor_remove (getters, nuke->ipaddr);
      }
    }
//      foreach_neighbor (neighbor_senders_to_me*, givuh, givers) {
//        sprintf(trace_buf_, "rcvdst %9s %8x bw %6d use %6d lr %4.2f%% \n", 
//  	      get_hostname(givuh->ipaddr), 
//  	      givuh->ipaddr,
//  	      (int) givuh->arrivals.get_value(),
//  	      (int) givuh->useful.get_value(),
//  	      0
//  	      //  	       (application->parallel_transport ? 100*application->parallel_transport->loss_rate(givuh->ipaddr): 0)
//  	      );
//        trace_print();
//      }    
    drop = 0;
    int closed_irrat=0;
    // then nuke the sender who gives me nothing or the one who gives me the least
    if ( !neighbor_space(givers) ) {   // only take one out if there is no space
      foreach_neighbor (neighbor_senders_to_me*, givuh, givers) {
        if ( curtime - givuh->start_time > ((double)3 * BULLET_NORMAL_RANSUB) ) {
          // Allow at least 3 RanSub epochs for each guy to give us data
          if (givuh->useful.get_value()/givuh->arrivals.get_value() < BULLET_CLOSING_RATIO && 
              !neighbor_query(myparent, givuh->ipaddr)) {
            sprintf(trace_buf_, "Receiver: Closing irrational sender %x with values %f %f\n", givuh->ipaddr, givuh->useful.get_value(), givuh->arrivals.get_value());
            trace_print();
            closed_irrat = givuh->ipaddr;
          }
          else if ( !drop ||
              givuh->useful.get_value() < drop_bw) {
            drop = givuh->ipaddr;
            drop_bw = givuh->useful.get_value();
          }
        }
      }
      if (closed_irrat) {
	neighbor_remove (givers, closed_irrat);
	route_remove(closed_irrat, 0, 0, -1);
      }
      if (!closed_irrat &&
	  drop) {
	neighbor_senders_to_me *nuke = neighbor_entry(givers, drop);
	route_remove(nuke->ipaddr, 0, 0, -1);
	sprintf(trace_buf_, "Receiver: Closing useless sender %x with values %f %f %f\n", nuke->ipaddr, nuke->useful.get_value(), nuke->arrivals.get_value(), nuke->start_time);
	trace_print();
	neighbor_remove (givers, nuke->ipaddr);
      }
    }

  }

  void joined_xmit_collect_if_need() {
    bullet_summary_ticket* myticket = new bullet_summary_ticket(working_set::UNIVERSE_SIZE);
    * myticket = working_file.get_sketch();
    myticket->st.address = me;
    int propagate=1;
    foreach_neighbor(neighbor_ransub_children *, kid, mychildren) {
      if (kid->seq != sequence && kid->seq !=-1)
        propagate = 0;
    }
    if (propagate) {      
      sprintf(trace_buf_,"joined_xmit_collect_if_need: kids are okay\n");
      trace_print ();
    }
    if (collect_expired)
      propagate = 1;
    if (propagate && collect_missing) {      
      collect_missing = 0;      
      candidate_set<cand_bullet_summary_ticket>  tosend(BULLET_MAX_CANDS);
      if ( neighbor_space(getters)  && source_ != me )
	tosend.addj(myticket->st);
      int sofar = 1;
      descendants = 0;
      foreach_neighbor(neighbor_ransub_children *, kid, mychildren) 
      {
        descendants += kid->represents;
        tosend.compact(kid->gathered, kid->represents, sofar);
        sofar+=kid->represents;
        if (collect_expired &&
            kid->seq != sequence && kid->seq !=-1)
	  {
	    // this kid isn't ready, we timed out the collect
	    sprintf(trace_buf_,"kid->address: %x  timed out seq %d sequence %d\n", kid->ipaddr, kid->seq, sequence);
	    trace_print ();
	    kid->represents=0;  
	  }
      }
      if (source_ == me) {
        curset = tosend;
        //	curset.printem(me);
      }
      else {
	neighbor_ransub_parent *papa = neighbor_random(myparent);

	sprintf(trace_buf_, "RanSub: sending collect sequence %d (%d desc.)to %x.\n", sequence, descendants+1, papa->ipaddr);
	trace_print();
	route_collect(papa->ipaddr, sequence, descendants+1, tosend, 0, 0, -1);
      }
    }
    delete myticket;
  }

}