common-agent.cc

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

    //Time added
    memcpy(&ap->u.dataack_p.orig_time, & data->send_time, sizeof(double));
    
  }
    break; 

//sunny
	case (QUERY_RANDOM) : {
    		struct query_random * data = (struct query_random *) ((char *)recv_buffer + newlen);
    		ap->u.random_p.src_id.agent_id = ntohl(data->src_id.agent_id);
    		memcpy(&ap->u.random_p.src_id.agent_addr, &data->src_id.agent_addr, sizeof(struct sockaddr_in));
    		ap->u.random_p.lid = (short) ntohl(data->lid); 
    		ap->u.random_p.direction = (short) ntohl(data->direction); 
    		ap->u.random_p.hop_count = (short) ntohl(data->hop_count); 
    		ap->u.random_p.level = (short) ntohl(data->level); 

    		ap->u.random_p.rand_id.agent_id = ntohl(data->rand_id.agent_id);
    		memcpy(&ap->u.random_p.rand_id.agent_addr, &data->rand_id.agent_addr, sizeof(struct sockaddr_in));
    		ap->u.random_p.seq_no = (int) ntohl(data->seq_no); 
    		memcpy(ap->u.random_p.data, data->data, NUM_BUDDY_BITMAP*sizeof(char));

    		newlen += sizeof(struct query_random);

		//Time added
		memcpy(& ap->u.random_p.send_time, & data->send_time, sizeof(double));
	}
		break;

	case (RESPONSE_RANDOM) : {
    		struct query_random * data = (struct query_random *) ((char *)recv_buffer + newlen);
    		ap->u.random_p.src_id.agent_id = ntohl(data->src_id.agent_id);
    		memcpy(&ap->u.random_p.src_id.agent_addr, &data->src_id.agent_addr, sizeof(struct sockaddr_in));
    		ap->u.random_p.lid = (short) ntohl(data->lid); 
    		ap->u.random_p.direction = (short) ntohl(data->direction); 
    		ap->u.random_p.hop_count = (short) ntohl(data->hop_count); 
    		ap->u.random_p.level = (short) ntohl(data->level); 

    		ap->u.random_p.rand_id.agent_id = ntohl(data->rand_id.agent_id);
    		memcpy(&ap->u.random_p.rand_id.agent_addr, &data->rand_id.agent_addr, sizeof(struct sockaddr_in));
    		ap->u.random_p.seq_no = (int) ntohl(data->seq_no); 
    		memcpy(ap->u.random_p.data, data->data, NUM_BUDDY_BITMAP*sizeof(char));

    		newlen += sizeof(struct query_random);

		 //Time added
		memcpy(& ap->u.random_p.send_time, & data->send_time, sizeof(double));
	}
		break;

	case (RETRANSMIT_REQUEST) : {
    		struct retransmit_pkt * data = (struct retransmit_pkt *) ((char *)recv_buffer + newlen);
    		ap->u.retransmit_p.src.agent_id = ntohl(data->src.agent_id);
    		memcpy(&ap->u.retransmit_p.src.agent_addr, &data->src.agent_addr, sizeof(struct sockaddr_in));
    		ap->u.retransmit_p.seq_no = (int) ntohl(data->seq_no); 
    		memcpy(ap->u.retransmit_p.bitmap, data->bitmap, BITMAP_SIZE*sizeof(char));
    		newlen += sizeof(struct retransmit_pkt);

		//Time added
		memcpy(& ap->u.retransmit_p.send_time, & data->send_time, sizeof(double));
	}
		break;
//sunny
      
  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);
    data->src_ag.agent_id = htonl(ap->u.joinforward_p.src_ag.agent_id);
    memcpy(&data->src_ag.agent_addr, &ap->u.joinforward_p.src_ag.agent_addr, sizeof(struct sockaddr_in));
    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);
    data->original_dst.agent_id = htonl(ap->u.joinforward_p.original_dst.agent_id);
    memcpy(&data->original_dst.agent_addr, &ap->u.joinforward_p.original_dst.agent_addr, sizeof(struct sockaddr_in));
    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));
      
    // made a change here

  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;

    data->exp_src.agent_id = htonl(ap->u.joinresp_p.exp_src.agent_id);
    memcpy(&data->exp_src.agent_addr, &ap->u.joinresp_p.exp_src.agent_addr, sizeof(struct sockaddr_in));
    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);
    data->original_src.agent_id = htonl(ap->u.data_p.original_src.agent_id);
    memcpy(&data->original_src.agent_addr, &ap->u.data_p.original_src.agent_addr, sizeof(struct sockaddr_in));
    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);

     data->type = htonl(ap->u.data_p.type);
    memcpy(data->bitmap,ap->u.data_p.bitmap, BITMAP_SIZE*sizeof(char));
    data->hop_count = htonl(ap->u.data_p.hop_count);
    data->overlay_hop = htonl(ap->u.data_p.overlay_hop);
     //Time added
    memcpy(& data->orig_time, & ap->u.data_p.orig_time, sizeof(double));

    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);
    data->original_src.agent_id = htonl(ap->u.dataack_p.original_src.agent_id);
    memcpy(&data->original_src.agent_addr, &ap->u.dataack_p.original_src.agent_addr, sizeof(struct sockaddr_in));
    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) */

    //Time added
    memcpy(& data->send_time, & ap->u.dataack_p.orig_time, sizeof(double));
  
  }
    break; 

//sunny
	case (QUERY_RANDOM) : {
    		struct query_random * data = (struct query_random *) ((char *)payload + len);
    		data->src_id.agent_id = htonl(ap->u.random_p.src_id.agent_id);
    		memcpy(&data->src_id.agent_addr, &ap->u.random_p.src_id.agent_addr, sizeof(struct sockaddr_in));
    		data->lid = htonl(ap->u.random_p.lid);
    		data->direction = htonl(ap->u.random_p.direction);
    		data->hop_count = htonl(ap->u.random_p.hop_count);
    		data->level = htonl(ap->u.random_p.level);

    		data->rand_id.agent_id = htonl(ap->u.random_p.rand_id.agent_id);
    		memcpy(&data->rand_id.agent_addr, &ap->u.random_p.rand_id.agent_addr, sizeof(struct sockaddr_in));
		
		//Time added
		memcpy(& data->send_time, & ap->u.random_p.send_time, sizeof(double));
  

    		assert((unsigned short) (5000 + ap->u.random_p.src_id.agent_id) == ntohs(ap->u.random_p.src_id.agent_addr.sin_port));
    		data->seq_no = htonl(ap->u.random_p.seq_no);
    		memcpy(data->data, ap->u.random_p.data, NUM_BUDDY_BITMAP*sizeof(char));
    		len += sizeof(struct query_random);


	}
	break;
	case (RESPONSE_RANDOM) : {
    		struct query_random * data = (struct query_random *) ((char *)payload + len);
    		data->src_id.agent_id = htonl(ap->u.random_p.src_id.agent_id);
    		memcpy(&data->src_id.agent_addr, &ap->u.random_p.src_id.agent_addr, sizeof(struct sockaddr_in));