coop-agent.cc

模拟器提供了一个简单易用的平台

#undef LOG_NEW_LEADER_CHOOSE

#ifdef LOG_NEW_LEADER_CHOOSE
  printf ("[coop %d %d ] : At %8.4f (c-fnl) : find-new-ldr : lid %d  %s\n", id, n->id, Scheduler::Clock(), l->lid, (omit_self == true) ? "omit-self" : "include-self");
#endif // LOG_NEW_LEADER_CHOOSE

  double max_dist = -1.0;
  int max_count = -1;
  LayerAgentInfo * new_leader = NULL;
  for (void * pos = l->ag_list.GetHeadPosition();
       pos != NULL;
       l->ag_list.GetNext(pos) ) {

    LayerAgentInfo * this_la = l->ag_list.GetAt(pos);

    if (self_check(this_la) == true) {
      continue;
    }
    // HAVE TO LOOK AT HOW TO ACCOMODATE THIS COUNT THING

    int this_count = count_dist_known_members_for_agent(this_la,omit_self);

#ifdef LOG_NEW_LEADER_CHOOSE
    printf ("[coop %d %d ] : . . (c-fnl) : mbr %d : count %d : ", id, n->id, this_la->ag.agent_id, this_la->agent_dist_arr_count);
    for (int i = 0; i < this_la->agent_dist_arr_count; i++)
      printf ("( %d %f )", this_la->agent_dist_arr[i].ag.agent_id, this_la->agent_dist_arr[i].dist);
    printf ("\n[coop %d %d ] : . . (c-fnl) : this-count %d\n", id, n->id, this_count);
#endif // LOG_NEW_LEADER_CHOOSE

    if (this_count == 0) {
      if (max_count < 0) {
        max_count = 0;
        new_leader = this_la;
      }
      continue;
    }

    assert (this_count > 0);
    if (this_count < max_count)
      continue;
    double this_dist = get_farthest_dist_of_member_to_agent(l,this_la,omit_self,true);
#ifdef LOG_NEW_LEADER_CHOOSE
    printf ("[coop %d %d ] : . . (c-fnl) : farthest-dist %f\n", id, n->id, this_dist);
#endif // LOG_NEW_LEADER_CHOOSE

    assert (this_dist >= 0.0);
    if (this_count > max_count) {
      max_count = this_count;
      max_dist = this_dist;
      new_leader = this_la;
    }
    else { //(this_count == max_count)
      assert (max_dist >= 0.0);
      long this_dist_long = USECS(this_dist);
      long max_dist_long = USECS(max_dist);
      if (this_dist_long < max_dist_long) {
	max_dist = this_dist;
	new_leader = this_la;
      }
    }
  }


  if (omit_self == false) { // Need to compare self too
    void * pos = l->ag_list.Locate(id);
    if (pos != NULL) {

      LayerAgentInfo * self_ag = l->ag_list.GetAt(pos);
      LayerAgentInfo * my_farthest = find_farthest_agent_to_me_in_layer(l);
      int my_count = count_dist_known_members_for_me(l);

#ifdef LOG_NEW_LEADER_CHOOSE
      printf ("[coop %d %d ] : . . (c-fnl) : mbr self : count %d : ", id, n->id, l->ag_list.GetSize() );
      for (void * pos2 = l->ag_list.GetHeadPosition();
	   pos2 != NULL;
	   l->ag_list.GetNext(pos2) ) {
	LayerAgentInfo * la2 = l->ag_list.GetAt(pos2);
	printf ("( %d %f )", la2->ag.agent_id, la2->dist);
      }
      printf ("\n[coop %d %d ] : . . (c-fnl) : my-count %d : farthest ( %d %f )\n", id, n->id, my_count, my_farthest->ag.agent_id, my_farthest->dist);
#endif // LOG_NEW_LEADER_CHOOSE
      
      assert (my_farthest != NULL);
      long max_dist_long = USECS(max_dist);
      long my_farthest_dist_long = USECS(my_farthest->dist);
      if ( (my_count > max_count) ||
	   ( (my_count == max_count) &&
	     ( (max_dist < 0.0) || (my_farthest_dist_long <= max_dist_long) ) ) ) {
	// The my_farthest->dist <= max_dist ensures that the current incumbent
	// (i.e. if I am the ldr) continues as ldr if possible
	new_leader = self_ag;
	max_count = my_count;
	max_dist = my_farthest->dist;
      }
    }
  }

#ifdef LOG_NEW_LEADER_CHOOSE
  if (new_leader != NULL)
      printf ("\n[coop %d %d ] : . . (c-fnl) : new-leader %d\n", id, n->id, new_leader->ag.agent_id);
  else
    printf ("\n[coop %d %d ] : . . (c-fnl) : new-leader NULL\n", id, n->id);
#endif // LOG_NEW_LEADER_CHOOSE
  return new_leader;
}

/* Check all the cluster members, if any of them should be the new
 * cluster leader
 */
LayerAgentInfo * coopAgent::check_all_change_cluster_leader (LayerInfo *l) {

  assert (l != NULL);
  // Should be called for a cluster root
  assert ( (l->root->ag.agent_id == id) && (l->root->ag.node_id == n->id) );

  LayerAgentInfo * new_leader = NULL;
  double max_dist = -1.0;
  for (void * pos = l->ag_list.GetHeadPosition();
       pos != NULL;
       l->ag_list.GetNext(pos) ) {

    LayerAgentInfo * la_ag = l->ag_list.GetAt(pos);
    if ( (la_ag->ag.agent_id == id) && (la_ag->ag.node_id == n->id) )
      continue;
    double temp_max;
    if (check_change_cluster_leader(l,la_ag,temp_max) == true) {
      assert (temp_max >= 0.0);
      long max_dist_long = USECS(max_dist);
      long temp_max_long = USECS(temp_max);
      if ( (max_dist < 0.0) || (temp_max_long < max_dist_long) ) {
	new_leader = la_ag;
	max_dist = temp_max;
      }
    }
  }

  return new_leader;
}

bool coopAgent::check_change_cluster_leader (LayerInfo *l, LayerAgentInfo *la_ag, double & la_dist) {

  assert (l != NULL);
  // First should check if this member knows all the agents that I know.
  if (are_agent_members_contained_in_mine(l,la_ag) == false)
    return false;

  LayerAgentInfo *farthest = find_farthest_agent_to_me_in_layer(l);
  assert (farthest != NULL);

  la_dist = get_farthest_dist_of_member_to_agent(l,la_ag,false,true);

  long la_dist_long = USECS(la_dist);
  long farthest_dist_long = USECS(farthest->dist);
  if (la_dist_long < farthest_dist_long)
    return true;
  return false;
}

/* ignore_self: true, if we do not consider the distance to self from
 * this agent in the calculation
 * use_only_members_known_to_self: true, if we only consider in this
 * calculation those members that are known to self, and no other
 * members
 */
double coopAgent::get_farthest_dist_of_member_to_agent (LayerInfo *l, LayerAgentInfo * la, bool ignore_self, bool use_only_members_known_to_self) {

#undef LOG_DIST_INFO

#ifdef LOG_DIST_INFO
  printf ("[coop %d %d ] mbr-dist-info : %d : ", id, n->id, la->ag.agent_id);
#endif // LOG_DIST_INFO

  double max_dist = -1.0;
  for (int i = 0; i < la->agent_dist_arr_count; i++) {

    if (la->agent_dist_arr[i].dist < 0.0)
      continue;
    if ( (ignore_self == true) &&
	 ( (la->agent_dist_arr[i].ag.agent_id == id) && (la->agent_dist_arr[i].ag.node_id == n->id) ) )
      continue;

    if (use_only_members_known_to_self == true) {
      void * pos = l->ag_list.Locate(la->agent_dist_arr[i].ag.agent_id);
      if (pos == NULL)
	continue;
      LayerAgentInfo * this_la = l->ag_list.GetAt(pos);
      if (this_la->dist < 0.0)
	continue;
    }
#ifdef LOG_DIST_INFO
    printf ("( %d %5.3f )", la->agent_dist_arr[i].ag.agent_id,la->agent_dist_arr[i].dist);
#endif // LOG_DIST_INFO

    long la_agent_dist_arr_i_dist_long = USECS(la->agent_dist_arr[i].dist);
    long max_dist_long = USECS(max_dist);
    if ( (max_dist < 0.0) || (la_agent_dist_arr_i_dist_long > max_dist_long) ) {
      max_dist = la->agent_dist_arr[i].dist;
    }
  }
#ifdef LOG_DIST_INFO
  printf ("\n");
#endif // LOG_DIST_INFO

  return max_dist;
}

/* Finds the farthest agent in this cluster from this agent */
LayerAgentInfo * coopAgent::find_farthest_agent_to_me_in_layer (LayerInfo * l) {

  assert (l != NULL);

  LayerAgentInfo * farthest = NULL;
  double max_dist = -1.0;
  for (void * pos = l->ag_list.GetHeadPosition();
       pos != NULL;
       l->ag_list.GetNext(pos) ) {

    LayerAgentInfo * la_ag = l->ag_list.GetAt(pos);
    if (la_ag->dist < 0.0)
      continue;
    long max_dist_long = USECS(max_dist);
    long la_ag_dist_long = USECS(la_ag->dist);
    if ( (max_dist < 0.0) || (la_ag_dist_long > max_dist_long) ) {
      max_dist = la_ag->dist;
      farthest = la_ag;
    }
  }
  return farthest;
}

/* Find the closest agent in the specified layer among all
 * the agents which have responded
 */
LayerAgentInfo * coopAgent::find_closest_agent_in_layer (LayerInfo * this_layer, AgentInfo * ignore_agent, bool match_join_resp) {

  assert (this_layer != NULL);
  int count_resps = 0;
  double min_dist = -1.0;
  LayerAgentInfo * min_dist_la_ag = NULL;
  for (void * pos = this_layer->ag_list.GetHeadPosition();
       pos != NULL;
       this_layer->ag_list.GetNext(pos) ) {

    LayerAgentInfo * la_ag = this_layer->ag_list.GetAt(pos);
    if (self_check(la_ag) == true)
      continue;

    if (ignore_agent != NULL) {
      if ( (ignore_agent->agent_id == la_ag->ag.agent_id) && (ignore_agent->node_id == la_ag->ag.node_id)) {
	continue;
      }
    }

    if ( (match_join_resp == false) || (la_ag->lower_layer != NULL) ) { 
      /* For join responses, lower_layer != NULL => responded */
      count_resps ++;
      long min_dist_long = USECS(min_dist);
      long la_ag_dist_long = USECS(la_ag->dist);
      if ( (min_dist < 0.0) || (min_dist_long > la_ag_dist_long) ) {
	min_dist = la_ag->dist;
	min_dist_la_ag = la_ag;
      }
    }
  }
  return min_dist_la_ag;
}

/* Checks the member set known to the specified agent. Returns true
 * if all the members known to this agent is also known to me.
 * Known => distance to the member is also known.
 */
bool coopAgent::are_agent_members_contained_in_mine (LayerInfo *l, LayerAgentInfo * la) {

  for (int i = 0; i < la->agent_dist_arr_count; i++) {

    if (la->agent_dist_arr[i].dist >= 0.0) {
      void * pos = l->ag_list.Locate(la->agent_dist_arr[i].ag.agent_id);
      if (pos == NULL)
	return false;
      LayerAgentInfo * this_la = l->ag_list.GetAt(pos);
      if (this_la->dist < 0.0)
	return false;
    }
  }
  return true;
}

bool coopAgent::are_my_members_contained_in_agent (LayerInfo *l, LayerAgentInfo * la) {

  for (void * pos = l->ag_list.GetHeadPosition();
       pos != NULL;
       l->ag_list.GetNext(pos) ) {

    LayerAgentInfo * this_la = l->ag_list.GetAt(pos);
    if (this_la->dist >= 0.0) {
      for (int i = 0; i < la->agent_dist_arr_count; i++) {

	if ( (la->agent_dist_arr[i].ag.agent_id == this_la->ag.agent_id) &&
	     (la->agent_dist_arr[i].ag.node_id == this_la->ag.node_id) )
	  continue;
      }
      return false;
    }
  }
  return true;
}

/* ignore self true means that do not include "this" agent in
 * the count
 */
int coopAgent::count_dist_known_members_for_agent (LayerAgentInfo *la, bool ignore_self) {

  int count = 0;
  for (int i = 0; i < la->agent_dist_arr_count; i++) {

    if ( (ignore_self == true) &&
	 ( (la->agent_dist_arr[i].ag.agent_id == id) && (la->agent_dist_arr[i].ag.node_id == n->id) ) )
      continue;
    if (la->agent_dist_arr[i].dist >= 0.0)
      count ++;
  }
  return count;
}

int coopAgent::count_dist_known_members_for_me (LayerInfo *l) {

  assert (l != NULL);
  int count = 0;
  for (void * pos = l->ag_list.GetHeadPosition();
       pos != NULL;
       l->ag_list.GetNext(pos) ) {

    LayerAgentInfo * la = l->ag_list.GetAt(pos);
    if (la->dist >= 0.0)
      count ++;
  }
  return count;
}

void coopAgent::flush_layer (int layer_id) {

  LayerInfo * this_layer = layers.arr[layer_id];
  if (this_layer != NULL) {
    delete this_layer;
    layers.arr[layer_id] = NULL;
  }
  return;
/*
  for (void * pos = this_layer->ag_list.GetHeadPosition();
       pos != NULL;
       this_layer->ag_list.GetNext(pos) ) {

    LayerAgentInfo * la_ag = this_layer->ag_list.GetAt(pos);
    if (la_ag->lower_layer != NULL) {
      delete la_ag->lower_layer;
      la_ag->lower_layer = NULL;
    }
  }
  return;
*/
}

void coopAgent::send_root_challenge (LayerInfo *l, LayerAgentInfo *la) {

  send_specific_cluster_refresh_packet(l,la,false,true);
  return;
}

/* This is a root transfer to a member which is not part of this
 * cluster.
 */
void coopAgent::do_cluster_merge_root_transfer (LayerInfo *l, AgentInfo *new_root) {

  assert (l != NULL);
  assert ( (l->root != NULL) && (self_check(l->root) == true) );
  assert ( (new_root->agent_id != id) || (new_root->node_id != n->id) );
  AppPacket *ap = new AppPacket(CLUSTER_MERGE);
  put_layer_cluster_info_into_packet(l,ap);
  send_pkt_wrapper(ap,new_root->agent_id,new_root->node_id);

  LayerAgentInfo * root_info = new LayerAgentInfo(new_root->agent_id,new_root->node_id);
  if (l->AddClusterRoot(root_info) == false) {
    delete root_info;
    root_info = l->FindClusterMember(new_root->agent_id);
  }

  l->cr_msgt.CancelTimer();
  send_all_cluster_refresh_packet(l,true,true,root_info);
  l->cr_msgt.SetTimer(CONST_CLUSTER_REFRESH_MSG_TIMEOUT);

  return;
}

/* Root transfer to a member already part of the cluster */
void coopAgent::do_internal_root_transfer (LayerInfo *l, LayerAgentInfo *la) {

  assert ( (l->root->ag.agent_id == id) && (l->root->ag.node_id == n->id) );
  l->root = la; // This is the root change
  
  l->cr_msgt.CancelTimer();
  send_all_cluster_refresh_packet