common-agent.cc,v

这是P2P流媒体方案－NICE的实现源码

    memcpy(& ap->u.dataack_p.original_src, & data->original_src, sizeof(struct PacketAgentInfo));
    ap->u.dataack_p.original_src.agent_id = (int) ntohl(ap->u.dataack_p.original_src.agent_id);
    ap->u.dataack_p.seq_no = (int) ntohl(data->seq_no); 
    newlen += sizeof(struct dataack_pkt);
    /* copy variable part */
    /* len += sizeof(variable part) */
  }
    break; 
      
  case (PACKET_UNDEFINED) :
       
    printf("[Err] undefined packet\n");
    free(ap);
    ap = NULL;
    break; 
      
  default : 

    /* printf("undefined packet\n"); */
    free(ap);
    ap = NULL;
    break;

  }
  fflush(stdout);
  return ap;
}

int commonAgent::flatten_apppacket(char * payload, AppPacket * ap, int dst_ag) {

  int len = sizeof(ap->st);
  long int tempint;
  tempint = htonl(ap->st);
  memcpy(payload, & tempint, len);

  tempint = htonl(id);
  memcpy(payload + len, & tempint, sizeof(int));
  len += sizeof(int);
  
  memcpy(payload + len, & udp_recv_agent_addr, sizeof(struct sockaddr_in));
  len += sizeof(struct sockaddr_in);
  
  assert(ntohs(udp_recv_agent_addr.sin_port) == (unsigned short) (id + 5000));

  tempint = htonl(dst_ag);
  memcpy(payload + len, & tempint, sizeof(int));
  len += sizeof(int);

  switch (ap->st) {
  case (JOIN_QUERY) : {
	
    struct const_join_query_pkt * data = (struct const_join_query_pkt *) ((char *)payload + len);
    int temp = htonl(ap->u.joinq_p.q_lid);
    data->q_lid = temp;
    data->attach = htonl((int) ap->u.joinq_p.attach);
    data->send_time = (long) htonl((long) (ap->u.joinq_p.send_time * PRECISION_DOUBLE));
    //    memcpy(& data->send_time, & ap->u.joinq_p.send_time, sizeof(data->send_time));
    memcpy(& data->reply_addr, & ap->u.joinq_p.reply_addr, sizeof(struct sockaddr_in));
    len += sizeof(struct const_join_query_pkt);
  }
    break;

  case (JOIN_FORWARD) : {
	
    struct const_join_forward_pkt * data = (struct const_join_forward_pkt *) ((char *)payload + len);
    memcpy(& data->src_ag, & ap->u.joinforward_p.src_ag, sizeof(struct PacketAgentInfo));
    data->src_ag.agent_id = (int) htonl(data->src_ag.agent_id);
    assert((unsigned short) (ap->u.joinforward_p.src_ag.agent_id + 5000) == ntohs(ap->u.joinforward_p.src_ag.agent_addr.sin_port));
    int temp = htonl(ap->u.joinforward_p.q_lid);
    data->q_lid = temp;
    /* data->q_lid = (int) htonl(data->q_lid); BIG ERROR */

    data->attach = htonl((int) ap->u.joinforward_p.attach);
    memcpy(& data->original_dst, & ap->u.joinforward_p.original_dst, sizeof(struct PacketAgentInfo));
    data->original_dst.agent_id = (int) htonl(data->original_dst.agent_id);
    assert((unsigned short) (5000 + ap->u.joinforward_p.original_dst.agent_id) == ntohs(ap->u.joinforward_p.original_dst.agent_addr.sin_port));
    data->send_time = (long) htonl((long) (ap->u.joinforward_p.send_time * PRECISION_DOUBLE));
    //    memcpy(& data->send_time, & ap->u.joinforward_p.send_time, sizeof(double));
    len += sizeof(struct const_join_forward_pkt);
  }
    break;
	
  case (JOIN_RESPONSE) : {
	
    struct const_join_response_pkt * data = (struct const_join_response_pkt *) ((char *)payload + len);
    data->accept = htonl((int) ap->u.joinresp_p.accept);
    int temp = htonl(ap->u.joinresp_p.layer_id);
    data->layer_id = temp;
    temp = htonl(ap->u.joinresp_p.mbr_count);
    data->mbr_count = temp;
    temp = htonl(ap->u.joinresp_p.your_id);
    data->your_id = temp;

    memcpy(& data->exp_src, & ap->u.joinresp_p.exp_src, sizeof(struct PacketAgentInfo));
    data->exp_src.agent_id = (int) htonl(data->exp_src.agent_id);
    assert((unsigned short) (ap->u.joinresp_p.exp_src.agent_id + 5000) == ntohs(ap->u.joinresp_p.exp_src.agent_addr.sin_port));
    len += sizeof(struct const_join_response_pkt);

    //    memcpy(payload + len, ap->u.joinresp_p.agent_arr, (ap->u.joinresp_p.mbr_count)*sizeof(struct TreeInfo));
    for (int i = 0; i < ap->u.joinresp_p.mbr_count; i++) {
      struct sriTreeInfo * tempinfo = (struct sriTreeInfo *)(payload + len);
      tempinfo[i].ag.agent_id = (int) htonl(ap->u.joinresp_p.agent_arr[i].ag.agent_id);
      assert((unsigned short) (5000 + ap->u.joinresp_p.agent_arr[i].ag.agent_id) == ntohs(ap->u.joinresp_p.agent_arr[i].ag.agent_addr.sin_port));
      memcpy( & tempinfo[i].ag.agent_addr, & ap->u.joinresp_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
      tempinfo[i].num_children = (int) htonl(ap->u.joinresp_p.agent_arr[i].num_children);
      tempinfo[i].dist = (long) htonl((long) (ap->u.joinresp_p.agent_arr[i].dist * PRECISION_DOUBLE));

//        double temp_double;
//        memcpy(& temp_double, & tempinfo[i].dist, sizeof(double));
//        tempinfo[i].long_dist = htonl((long) temp_double);
    }
    len += (ap->u.joinresp_p.mbr_count)*sizeof(struct sriTreeInfo);

    /*
    printf("flattening debug:\n");
    printf("mbr_count %d mbrs : ",ap->u.joinresp_p.mbr_count);
    for (int i = 0; i < ap->u.joinresp_p.mbr_count; i++) {
      printf("%d ",ap->u.joinresp_p.agent_arr[i].ag.agent_id);
    }
    printf("\n");
    */

  }
    break;

  case (CLUSTER_REFRESH) : {

    struct const_cluster_refresh_pkt * data = (struct const_cluster_refresh_pkt *) ((char *)payload + len);
    data->layer_id = (int) htonl(ap->u.clusterrefresh_p.layer_id);
    data->mbr_count = (int) htonl(ap->u.clusterrefresh_p.mbr_count);
    data->hl_mbr_count = (int) htonl(ap->u.clusterrefresh_p.hl_mbr_count);
    data->is_root = htonl((int)ap->u.clusterrefresh_p.is_root);
    data->root_xfer = htonl((int)ap->u.clusterrefresh_p.root_xfer);
    data->cluster_remove = htonl((int)ap->u.clusterrefresh_p.cluster_remove);
    len += sizeof(struct const_cluster_refresh_pkt);

    //    memcpy(payload + len, ap->u.clusterrefresh_p.agent_arr, (ap->u.clusterrefresh_p.mbr_count)*sizeof(struct TreeInfo));
    for (int i = 0; i < ap->u.clusterrefresh_p.mbr_count; i++) { 
      struct sriTreeInfo * tempinfo = (struct sriTreeInfo *)(payload + len);
      tempinfo[i].ag.agent_id = (int) htonl(ap->u.clusterrefresh_p.agent_arr[i].ag.agent_id);
      assert((unsigned short) (5000 + ap->u.clusterrefresh_p.agent_arr[i].ag.agent_id) == ntohs(ap->u.clusterrefresh_p.agent_arr[i].ag.agent_addr.sin_port));
      memcpy( & tempinfo[i].ag.agent_addr, & ap->u.clusterrefresh_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
      tempinfo[i].num_children = (int) htonl(ap->u.clusterrefresh_p.agent_arr[i].num_children);
      tempinfo[i].dist = (long) htonl((long) ap->u.clusterrefresh_p.agent_arr[i].dist);
    }
    len += (ap->u.clusterrefresh_p.mbr_count)*sizeof(struct sriTreeInfo);

    //    memcpy(payload + len, ap->u.clusterrefresh_p.hl_agent_arr, (ap->u.clusterrefresh_p.hl_mbr_count)*sizeof(struct TreeInfo));
    for (int i = 0; i < ap->u.clusterrefresh_p.hl_mbr_count; i++) { 
      struct sriTreeInfo * tempinfo = (struct sriTreeInfo *)(payload + len);
      tempinfo[i].ag.agent_id = (int) htonl(ap->u.clusterrefresh_p.hl_agent_arr[i].ag.agent_id);
      assert((unsigned short) (5000 + ap->u.clusterrefresh_p.hl_agent_arr[i].ag.agent_id) == ntohs(ap->u.clusterrefresh_p.hl_agent_arr[i].ag.agent_addr.sin_port));
      memcpy( & tempinfo[i].ag.agent_addr, & ap->u.clusterrefresh_p.hl_agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
      tempinfo[i].num_children = (int) htonl(ap->u.clusterrefresh_p.hl_agent_arr[i].num_children);
      tempinfo[i].dist = (long) htonl((long) ap->u.clusterrefresh_p.hl_agent_arr[i].dist);
    }
    len += (ap->u.clusterrefresh_p.hl_mbr_count)*sizeof(struct sriTreeInfo);
  }
    break; 

  case (CLUSTER_MERGE) : {
    
    struct const_cluster_merge_pkt * data = (struct const_cluster_merge_pkt *) ((char *)payload + len);
    data->layer_id = (int) htonl(ap->u.clustermerge_p.layer_id);
    data->mbr_count = (int) htonl(ap->u.clustermerge_p.mbr_count);
    len += sizeof(struct const_cluster_merge_pkt);

    //    memcpy(payload + len, ap->u.clustermerge_p.agent_arr, (ap->u.clustermerge_p.mbr_count)*sizeof(struct TreeInfo));
    for (int i = 0; i < ap->u.clustermerge_p.mbr_count; i++) { 
      struct sriTreeInfo * tempinfo = (struct sriTreeInfo *)(payload + len);
      tempinfo[i].ag.agent_id = (int) htonl(ap->u.clustermerge_p.agent_arr[i].ag.agent_id);
      assert((unsigned short) (5000 + ap->u.clustermerge_p.agent_arr[i].ag.agent_id) == ntohs(ap->u.clustermerge_p.agent_arr[i].ag.agent_addr.sin_port));
      memcpy( & tempinfo[i].ag.agent_addr, & ap->u.clustermerge_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
      tempinfo[i].num_children = (int) htonl(ap->u.clustermerge_p.agent_arr[i].num_children);
      tempinfo[i].dist = (long) htonl((long) ap->u.clustermerge_p.agent_arr[i].dist);
    }
    len += (ap->u.clustermerge_p.mbr_count)*sizeof(struct sriTreeInfo);

  }
    break; 
    
  case (PING_QUERY) : {

    struct const_ping_query_pkt * data = (struct const_ping_query_pkt *) ((char *)payload + len);
    data->src_time = (long) htonl((long) (ap->u.pingq_p.src_time * PRECISION_DOUBLE));
    //    memcpy(& data->src_time, & ap->u.pingq_p.src_time, sizeof(double));
    data->lid = (int) htonl(ap->u.pingq_p.lid);
    len += sizeof(struct ping_query_pkt);
  }
    break; 

  case (PING_RESPONSE) : {

    struct const_ping_response_pkt * data = (struct const_ping_response_pkt *) ((char *)payload + len);
    data->accept = htonl((int)ap->u.pingresp_p.accept);
    data->lid = (int) htonl(ap->u.pingresp_p.lid);
    data->src_time = (long) htonl((long) (ap->u.pingresp_p.src_time * PRECISION_DOUBLE) );
    //    memcpy( & data->src_time, & ap->u.pingresp_p.src_time, sizeof(double));
    data->mbr_count = (int) htonl(ap->u.pingresp_p.mbr_count);
    data->dist = (long) htonl((long) (ap->u.pingresp_p.dist * PRECISION_DOUBLE) );
    //    memcpy(& data->dist, & ap->u.pingresp_p.dist, sizeof(double));
    len += sizeof(struct join_query_pkt);

    //    memcpy(payload + len, ap->u.pingresp_p.agent_arr, (ap->u.pingresp_p.mbr_count)*sizeof(struct TreeInfo));
    for (int i = 0; i < ap->u.pingresp_p.mbr_count; i++) { 
      struct sriTreeInfo * tempinfo = (struct sriTreeInfo *)(payload + len);
      tempinfo[i].ag.agent_id = (int) htonl(ap->u.pingresp_p.agent_arr[i].ag.agent_id);
      assert((unsigned short) (5000 + ap->u.pingresp_p.agent_arr[i].ag.agent_id) == ntohs(ap->u.pingresp_p.agent_arr[i].ag.agent_addr.sin_port));
      memcpy( & tempinfo[i].ag.agent_addr, & ap->u.pingresp_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
      tempinfo[i].num_children = (int) htonl(ap->u.pingresp_p.agent_arr[i].num_children);
      tempinfo[i].dist = (long) htonl((long) ap->u.pingresp_p.agent_arr[i].dist);
    }
    len += (ap->u.pingresp_p.mbr_count)*sizeof(struct sriTreeInfo);

  }
    break;

  case (PACKET_DATA) : {

    struct const_data_pkt * data = (struct const_data_pkt *) (payload + len);
    memcpy(& data->original_src, & ap->u.data_p.original_src, sizeof(struct PacketAgentInfo));
    data->original_src.agent_id = (int) htonl(data->original_src.agent_id);
    assert((unsigned short) (5000 + ap->u.data_p.original_src.agent_id) == ntohs(ap->u.data_p.original_src.agent_addr.sin_port));
    data->seq_no = htonl(ap->u.data_p.seq_no);
    data->lid = htonl(ap->u.data_p.lid);
    data->base_lid = htonl(ap->u.data_p.base_lid);
    data->data_len = htonl(ap->u.data_p.data_len);
    len += sizeof(struct const_data_pkt);
    /* copy variable part */
    memcpy(payload + len, ap->u.data_p.payload, ap->u.data_p.data_len);
    len += ap->u.data_p.data_len;
  }
    break; 
      
  case (PACKET_DATA_ACK) : {

    struct dataack_pkt * data = (struct dataack_pkt *) ((char *)payload + len);
    memcpy(& data->original_src, & ap->u.dataack_p.original_src, sizeof(struct PacketAgentInfo));
    data->original_src.agent_id = (int) htonl(data->original_src.agent_id);
    assert((unsigned short) (5000 + ap->u.dataack_p.original_src.agent_id) == ntohs(ap->u.dataack_p.original_src.agent_addr.sin_port));
    data->seq_no = htonl(ap->u.dataack_p.seq_no);
    len += sizeof(struct dataack_pkt);
    /* copy variable part */
    /* len += sizeof(variable part) */
  }
    break; 
      
  case (PACKET_UNDEFINED) :
       
    printf("[Err] undefined packet\n");
    break; 
      
  default : 

    /* printf("[Err] undefined packet\n"); */
    break;

  }
  fflush(stdout);
  return len;
}

void commonAgent::free_app_packet(AppPacket *ap) {

  switch (ap->st) {

  case (CLUSTER_REFRESH) : {
    
    if (ap->u.clusterrefresh_p.hl_mbr_count > 0) {
      assert(ap->u.clusterrefresh_p.hl_agent_arr);
      free(ap->u.clusterrefresh_p.hl_agent_arr);
    }
    break; 
  }

  case (PING_RESPONSE) : {

    if (ap->u.pingresp_p.mbr_count > 0) {
      assert(ap->u.pingresp_p.agent_arr);
      free(ap->u.pingresp_p.agent_arr);
    }
  }
    break;

  case (PACKET_UNDEFINED) :
       
    printf("[Err] undefined packet\n");
    break; 

  case (PACKET_DATA) : {
    assert(ap->u.data_p.payload);
    free(ap->u.data_p.payload);
  }
    break;

  default : 

    /* printf("undefined packet\n"); */
    break;
  }
  assert(ap);
  free(ap);
  fflush(stdout);
  return;
}

/*
void commonAgent::handle_expired_timers() {
  
  int call_unmask_timer;

  call_unmask_timer = mask_timer();
  glob_handling_expired_timers = true;
  for (void *pos = expired_timer_list.GetHeadPosition();
       pos != NULL;
       expired_timer_list.GetNext(pos) ) {
    Timer *t = expired_timer_list.GetAt(pos);
    if (verify_and_remove_timer_instance(t->get_instance()) == true)
      t->Handler(true);
  }

  expired_timer_list.RemoveAll();
  glob_handling_expired_timers = false;
  if(call_unmask_timer == CALL_UNMASK_TIMER)
    unmask_timer();

  fflush(stdout);
  return;
}

void commonAgent::add_expired_timer(Timer * t) {

  expired_timer_list.Add(t, t->get_id());
  return;
}
*/
@