coop-agent.cc

来自「模拟器提供了一个简单易用的平台」· CC 代码 · 共 2,021 行 · 第 1/5 页

        if (l->ag_list.Locate(pred_aid) == NULL) { 
	  count_pkts += send_data_pkt_to_layer(l,src_aid,src_nid,src_seqno);
        }
      }
    }
  }

  if (state == JOIN) {
    if ( (curr_join_q_lid < MAX_LAYERS) && (curr_join_q_lid > 0) ) {

      LayerInfo * q_l = layers.arr[curr_join_q_lid];
      assert (q_l != NULL);
      assert (q_l->me_in_layer == false);
      count_pkts += send_data_pkt_to_layer(q_l,src_aid,src_nid,src_seqno);
    }
    if (curr_join_q_lid < 0) { // Starting join query at BSE

      // For join to higher layers, we probably already have the higher
      // target layer from the previous leader which left the cluster or
      // did a root xfer to me. Can use that as a fall back option.
      // This should not be for layer 0.
      LayerInfo * target_layer = layers.arr[target_join_q_lid];
      if (target_layer != NULL) {
	assert (target_layer->me_in_layer == false);
	count_pkts += send_data_pkt_to_layer(target_layer,src_aid,src_nid,src_seqno);
      }
    }
  }

  if (is_source == true) {
       AppPacket *ap = new AppPacket(PACKET_DATA);
       ap->u.data_p.original_src.agent_id = src_aid;
       ap->u.data_p.original_src.node_id = src_nid;
       ap->u.data_p.seq_no = src_seqno;
       ap->u.data_p.lid = -1;
#ifdef LOG_DATA_PACKET_SEND
       printf ("[coop %d %d ] At %8.4f send data-pkt : < [ %d %d ] > seq %d : to < [ %d %d ] > bse\n", id, n->id, Scheduler::Clock(), src_aid, src_nid, src_seqno, bse.agent_id, bse.node_id);
#endif // LOG_DATA_PACKET_SEND
       send_pkt_wrapper(ap,bse.agent_id, bse.node_id);
       count_pkts ++;
  }

#ifdef LOG_DATA_PACKET_SEND_COUNT
  if (count_pkts > 0)
    printf ("[coop %d %d ] At %8.4f fwd-data count %d for < [ %d %d ]> seq-no %d\n",id, n->id, Scheduler::Clock(), count_pkts, src_aid, src_nid, src_seqno);
#endif // LOG_DATA_PACKET_SEND_COUNT
  return;
}

int coopAgent::send_data_pkt_to_layer (LayerInfo * l, int src_aid, int src_nid, int src_seqno) {

  int count_pkts = 0;

  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) == false) {
      if ( ( (bse.agent_id != this_la->ag.agent_id) || (bse.node_id != this_la->ag.node_id) ) &&
         ( ( (this_la->ag.agent_id != src_aid) || (this_la->ag.node_id != src_nid) ) ) ) {
       AppPacket *ap = new AppPacket(PACKET_DATA);
       ap->u.data_p.original_src.agent_id = src_aid;
       ap->u.data_p.original_src.node_id = src_nid;
       ap->u.data_p.seq_no = src_seqno;
       ap->u.data_p.lid = l->lid;
#ifdef LOG_DATA_PACKET_SEND
       printf ("[coop %d %d ] At %8.4f send data-pkt : < [ %d %d ] > seq %d : to < [ %d %d ] > lid %d\n", id, n->id, Scheduler::Clock(), src_aid, src_nid, src_seqno, this_la->ag.agent_id, this_la->ag.node_id, l->lid);
#endif // LOG_DATA_PACKET_SEND
       send_pkt_wrapper(ap,this_la->ag.agent_id, this_la->ag.node_id);
       count_pkts ++;
     }
   }
 }

  return count_pkts;
}

/* Have to be the cluster root to accept this merge */
/* If we want to accept cluster merge at non-root agents, then
 * we have to change the mechanism of the merge, i.e. the merging
 * root should be able to first figure out which is the correct
 * root, and then advise the children appropriately of its
 * change to be the new correct root.
 */
bool coopAgent::valid_cluster_merge_packet (AppPacket * ap) {

  LayerInfo *l = layers.arr[ap->u.clustermerge_p.layer_id];
  if (l == NULL)
    return false;
  if (l->me_in_layer == false)
    return false;

  if (self_check(l->root) == false)
    return false;
  return true;
}

bool coopAgent::valid_cluster_refresh_packet (AppPacket * ap) {

  LayerInfo * l = layers.arr[ap->u.clusterrefresh_p.layer_id];
  if (l == NULL)
    return false;
  if (l->me_in_layer == false)
    return false;

  void * pos = l->ag_list.Locate(ap->src_agent);
  if (pos == NULL)
    return false;
  return true;
}

bool coopAgent::match_cluster_to_root_info (LayerInfo *l) {

  bool has_change = false;

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

    LayerAgentInfo * la_ag = l->ag_list.GetAt(pos);
    la_ag->valid_tmp = false;
  }

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

    void * pos = l->ag_list.Locate(l->root->agent_dist_arr[i].ag.agent_id);
    LayerAgentInfo * this_la;
    if (pos == NULL) {
      this_la = new LayerAgentInfo (l->root->agent_dist_arr[i].ag.agent_id,l->root->agent_dist_arr[i].ag.node_id);
      assert (self_check(this_la) == false); // Cannot have self unknown
      l->AddClusterMember(this_la);
      has_change = true;
    }
    else {
      this_la = l->ag_list.GetAt(pos);
    }
    this_la->valid_tmp = true;
  }

  // Purge the members that are not in the cluster any more
  for (void * pos = l->ag_list.GetHeadPosition();
       pos != NULL;
       ) {
    
    LayerAgentInfo * la_ag = l->ag_list.GetAt(pos);
    void * old_pos = pos;
    l->ag_list.GetNext(pos);
    if (la_ag->valid_tmp == false) {
      l->ag_list.RemoveAt(old_pos);
      assert (la_ag != NULL);
      delete la_ag;
      has_change = true;
    }
  }

  return has_change;
}

/* Send query to the next lower layer if needed (i.e. to
 * one below this_lid).
 * If this is the target layer, then need to figure out point of
 * attachment
 */
void coopAgent::continue_join_query (int this_lid) {

#ifdef LOG_COOP_JUNK
  printf ("[coop %d %d ] At %8.4f (c-cjq) : continue-join-query : lid %d\n", id, n->id, Scheduler::Clock(), this_lid);
#endif // LOG_COOP_JUNK

  if (this_lid > target_join_q_lid) { /* Probe all the cluster members */
    LayerInfo *l = layers.arr[this_lid];
    void * pos = l->ag_list.Locate(id);
    if (pos != NULL) { // I am considered to be part of this cluster !
      assert (self_check(l->root) == false); // I cannot be the root
      delete_self_from_layer(l);
      }
#ifdef LOG_COOP_JUNK
    printf ("[coop %d %d ] . . (c-cjq) : join-continue\n", id, n->id);
#endif // LOG_COOP_JUNK

    send_join_query_to_cluster(this_lid);
    m_join_resp_received = 0;
    curr_join_q_lid = this_lid;
    jqt.SetTimer(CONST_JOIN_QUERY_TIMEOUT);
    return;
  }
  else { /* Find point of attachment */
    assert (this_lid == target_join_q_lid);
    LayerInfo * lbase = layers.arr[this_lid];
    LayerAgentInfo *self = new LayerAgentInfo(id,n->id);
    self->dist = 0.0;
    if (lbase->AddClusterMember(self) == false) {
      delete self;
      self = lbase->FindClusterMember(id);
      assert(self->dist >= 0.0);
    }
    lbase->me_in_layer = true;
    
    assert (lbase->root != NULL);
    if (lbase->root->dist < 0.0)
      assert (lbase->root->dist >= 0.0);

    AppPacket *pkt = new AppPacket(JOIN_QUERY);
    init_join_query_packet(pkt,this_lid,Scheduler::Clock(),true);
    send_pkt_wrapper(pkt,lbase->root->ag.agent_id, lbase->root->ag.node_id);
    lbase->cr_msgt.SetTimer(CONST_CLUSTER_REFRESH_MSG_TIMEOUT);
    lbase->cr_chkt.SetTimer(CONST_CLUSTER_REFRESH_CHECK_TIMEOUT);
    state = ATTACHED;

    set_higher_layer_ping_timer();

#ifdef LOG_COOP_JUNK
    printf ("[coop %d %d ] . . (c-cjq) : attach to my-root < [ %d %d ] >\n", id, n->id, lbase->root->ag.agent_id,lbase->root->ag.node_id);
#endif // LOG_COOP_JUNK
    log_cluster_change_info(lbase->lid);
#ifdef LOG_COOP_CLUSTER_CHANGE
    display_cluster_info(lbase,lbase->lid);
#endif // LOG_COOP_CLUSTER_CHANGE

  }
  return;
}

bool coopAgent::valid_join_query_forward_packet (int qlid, bool attach) {
  
  if ( (qlid < 0) || ( (qlid == 0) && (attach == false) ) ) { 
    /* Wrong packet */
    printf ("[Err] Found packet with qlid %d\n",qlid);
    assert (0);
    return false;
  }  
  int lower_lid;
  if (attach == true) /* This is an attach request */
    lower_lid = qlid;
  else
    lower_lid = qlid - 1;
  if (layers.arr[lower_lid] == NULL)
    return false;
  /* I must be part of this cluster. */
  if (layers.arr[lower_lid]->me_in_layer == false)
    return false;

  return true;
}

/* Returns true if this is a valid join response packet that
 * we are expecting
 */
bool coopAgent::valid_join_response_packet (AppPacket *ap) {

  if (state != JOIN)
    return false;

  int this_lid = ap->u.joinresp_p.layer_id;
  if ( (ap->src_agent == bse.agent_id) && (ap->src == bse.node_id) ) {
    if (curr_join_q_lid != -1) // BSE-JF && (curr_join_q_lid != (this_lid+1) )
      return false;
    if (target_join_q_lid > this_lid) {
      assert (0);
      return false;
    }
  }
  else { /* Check for other layer queries */
    if (curr_join_q_lid != (this_lid+1))
      return false;
    LayerInfo * last_layer = layers.arr[curr_join_q_lid];
    if (last_layer == NULL)
      return false;
    if (last_layer->ag_list.Locate(ap->u.joinresp_p.exp_src.agent_id) == NULL)
      return false;
  }

  return true;
}

/* Sends a join query to all members specified in the given
 * layer. The expected response is all members in the immediate
 * lower layer.
 */
void coopAgent::send_join_query_to_cluster (int layer_lid) {

  LayerInfo * this_layer = layers.arr[layer_lid];
#ifdef LOG_COOP_SEND
    printf ("[coop %d %d ] At %8.4f (c-jqtc) : join-query-to-cluster : lid %d\n", id, n->id, Scheduler::Clock(), layer_lid);
#endif // LOG_COOP_SEND

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

    LayerAgentInfo * coop_ag = this_layer->ag_list.GetAt(pos);
    if ( (coop_ag->ag.agent_id == bse.agent_id) && (coop_ag->ag.node_id == bse.node_id) ) // Dont send this join query to the BSE
      continue;
    AppPacket *p = new AppPacket(JOIN_QUERY);
    init_join_query_packet (p,layer_lid,Scheduler::Clock(),false);
    send_pkt_wrapper(p,coop_ag->ag.agent_id,coop_ag->ag.node_id);

#ifdef LOG_COOP_SEND
    printf ("[coop %d %d ] . . (c-jqtc) < [ %d %d ] >\n", id, n->id, coop_ag->ag.agent_id, coop_ag->ag.node_id);
#endif // LOG_COOP_SEND
  }

#ifdef LOG_COOP_SEND
  printf ("\n");
#endif // LOG_COOP_SEND
  return;
}

void coopAgent::cancel_higher_layer_ping_timers (void) {

  if (m_ping_in_progress == true) {
    m_ping_in_progress = false;
    m_hlprt.CancelTimer();
    flush_lower_layers(layers.arr[m_ping_lid]);
  }
  m_hlpit.CancelTimer();
  return;
}

void coopAgent::set_higher_layer_ping_timer (void) {
  m_hlpit.SetTimer(CONST_HIGHER_LAYER_PING_INTERVAL_TIMEOUT);
  return;
}

void coopAgent::handle_join_query_timeout (void) {

  process_received_join_responses();
  return;
}

void coopAgent::process_received_join_responses (void) {

  assert (state == JOIN);

  if (curr_join_q_lid  == -1) { /* BSE defaulted ! */
    printf ("[coop %d %d ] At %8.4f BSE did not respond to join-query\n", id, n->id, Scheduler::Clock());
    init_join_query(-1);
#ifdef LOG_COOP_JUNK
    printf ("[coop %d %d ] At %8.4f process-join-resps : bse defaults target lid %d\n", id, n->id, Scheduler::Clock(), target_join_q_lid);
#endif // LOG_COOP_JUNK
    return;
  }

  /* Find the closest agent to this member and re-start
   * the query business for the lower layer */
  LayerInfo * this_layer = layers.arr[curr_join_q_lid];

  void * pos = this_layer->ag_list.Locate(id);
  assert (pos == NULL);

  LayerAgentInfo * min_dist_la_ag = find_closest_agent_in_layer(this_layer,NULL,true);

  if (min_dist_la_ag != NULL) {
    /* Send next set of queries if needed. */
    // assert (layers.arr[curr_join_q_lid] == NULL); THIS ONE WAS WRONG
    if (min_dist_la_ag->dist < 0.0)
      printf ("[Err] At %f [ %d %d ] found closest agent [ %d %d ] with neg dist\n", Scheduler::Clock(), id, n->id, min_dist_la_ag->ag.agent_id, min_dist_la_ag->ag.node_id);

    int lower_lid = curr_join_q_lid - 1;

    flush_layer(lower_lid);

    layers.arr[lower_lid] = min_dist_la_ag->lower_layer;
    min_dist_la_ag->lower_layer = NULL;
    
    LayerAgentInfo * mdla_in_lower = layers.arr[lower_lid]->FindClusterMember(min_dist_la_ag->ag.agent_id);
    assert (mdla_in_lower != NULL);
    mdla_in_lower->dist = min_dist_la_ag->dist;

#ifdef LOG_COOP_JUNK
    printf ("[coop %d %d ] At %8.4f process-join-resps : lid %d closest < [ %d %d ] > dist %f\n", id, n->id, Scheduler::Clock(), curr_join_q_lid, min_dist_la_ag->ag.agent_id, min_dist_la_ag->ag.node_id, min_dist_la_ag->dist);
#endif // LOG_COOP_JUNK

    continue_join_query(lower_lid);
  }
  else { /* Have to re-query the higher layer if no one in
	  * curr_join_q_lid responded. */
#ifdef LOG_COOP_JUNK
    printf ("[coop %d %d ] At %8.4f process-join-resps  layer defaults lid %d\n", id, n->id, Scheduler::Clock(), this_layer->lid);
#endif // LOG_COOP_JUNK

    delete this_layer;
    layers.arr[curr_join_q_lid] = NULL;
    int higher_lid = curr_join_q_lid + 1;
    if (higher_lid < MAX_LAYERS) {
      if (layers.arr[higher_lid] != NULL) { /* Retry from the higher layer */
	continue_join_query(higher_lid);
      }
      else { /* Have to re-start with the BSE */
	init_join_query(-1);
      }
    }
  }
  return;
}

/* Find a new leader in the cluster. I am leaving this cluster,
 * so I am not included in this.
 */
LayerAgentInfo * coopAgent::find_new_cluster_leader (LayerInfo *l, bool omit_self) {

  if (l->root != NULL)
    assert (self_check(l->root) == true);