common-agent.cc,v

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

head	1.1;
access;
symbols;
locks
	rbraud:1.1; strict;
comment	@// @;


1.1
date	2002.07.02.19.28.35;	author rbraud;	state Exp;
branches;
next	;


desc
@@


1.1
log
@Initial revision
@
text
@/*
 * File: common-agent.cc
 * Author: Suman Banerjee <suman@@cs.umd.edu>
 * Date: 15th February, 2002
 * Terms: GPL
 *
 * NICE Application Layer Multicast
 */

#include <stdio.h>
#include <assert.h>
#include <stdlib.h>
#include <netinet/in.h>

//#include "nicenode.h"
#include "common-agent.h"
#include "app-packet.h"
//#include <scheduler.h>

#define PRECISION_DOUBLE	1000.0

//#define CLOCK 0.00
//#define LOG_DETAILED_PKT_SEND
//#define LOG_DETAILED_PKT_RCV

#define MAX_PAYLOAD 10000
#define HOST_INFO_SIZE	6

void commonAgent::send_pkt_wrapper (AppPacket *ap, int aid, struct sockaddr_in agent_addr) {

  char payload[MAX_PAYLOAD];
#ifdef LOG_DETAILED_PKT_SEND
     printf ("[ %d ] spd %f [ %d ] tp %d", id, /* Scheduler::my_clock() */ my_clock(), aid, ap->st);
     display_detailed_pkt_info(ap);
     fflush(stdout);
#endif
  //  printf("app-packet type is %d\n",ap->st);

  int len = flatten_apppacket(payload, ap, aid);

  //  printf("flattened app-packet type is %d\n",ap->st);
  //  display_detailed_pkt_info(ap);
  
  //  display_detailed_pkt_info(app);
  
  send_pkt(payload,len,aid,agent_addr);
  //  printf("sent payload to agent %d port %d\n",aid, ntohs(((sockaddr_in) agent_addr).sin_port));
  free_app_packet(ap);
  return;
}

void commonAgent::rx_pkt_wrapper (AppPacket *ap) {
#ifdef LOG_DETAILED_PKT_RCV
   printf ("[ %d ] rpd %f [ %d ] tp %d", id, /* Scheduler::my_clock() */ my_clock(), ap->src_agent, ap->st);
   display_detailed_pkt_info(ap);
    fflush(stdout);

#endif
  return;
}

void commonAgent::display_detailed_pkt_info (AppPacket *ap) {
  switch (ap->st) {

  case JOIN_QUERY:
    printf (" sz 5 + 0\n");
    break;

  case JOIN_FORWARD:
    printf (" sz %d + 0\n", 2 * HOST_INFO_SIZE +  9);
    break;

  case PING_QUERY:
    printf (" sz 5 + 0\n");
    break;

  case JOIN_RESPONSE:
    printf (" sz %d + %d\n", HOST_INFO_SIZE + 7, HOST_INFO_SIZE *
ap->u.joinresp_p.mbr_count);
    break;

  case CLUSTER_REFRESH:
    printf (" sz 13 + %d\n", HOST_INFO_SIZE *
(ap->u.clusterrefresh_p.mbr_count+ap->u.clusterrefresh_p.hl_mbr_count));
    break;

  case CLUSTER_MERGE:
    printf (" sz 5 + %d\n", HOST_INFO_SIZE * ap->u.clustermerge_p.mbr_count);
    break;

  case PING_RESPONSE:
    printf (" sz 9 + %d\n", HOST_INFO_SIZE * ap->u.pingresp_p.mbr_count);
    break;

  case PACKET_DATA:
    printf(" sz %d + 0\n", 2 + HOST_INFO_SIZE);
    break;

  case PACKET_DATA_ACK:
    printf(" sz %d + 0\n", 2 + HOST_INFO_SIZE);
    break;

  default:

    assert(0);
  }
  fflush(stdout);

  return;
}


AppPacket * commonAgent::reclaim_apppacket(char * recv_buffer, int len) {

  enum AppPacketSubType st; 
  memcpy(& st, recv_buffer, sizeof(enum AppPacketSubType));
  st = (enum AppPacketSubType) ntohl(st);
  AppPacket *ap = new AppPacket(st); 
  int newlen = sizeof(enum AppPacketSubType);

  //  printf("within reclaim: type %d\n",ap->st);
  memcpy(& ap->src_agent, recv_buffer + newlen, sizeof(int));
  ap->src_agent = (int) ntohl(ap->src_agent);
  newlen += sizeof(int);

  memcpy(&ap->src_agent_addr, recv_buffer + newlen, sizeof(struct sockaddr_in));
  newlen += sizeof(struct sockaddr_in);

  memcpy(& ap->dst_agent, recv_buffer + newlen, sizeof(int));
  ap->dst_agent = (int) ntohl(ap->dst_agent);
  newlen += sizeof(int);
    
  switch (ap->st) {
  case (JOIN_QUERY) : {
    
    struct const_join_query_pkt * data = (struct const_join_query_pkt *) ((char *)recv_buffer + newlen);
    
    
    //    memcpy(& ap->u.joinq_p, data, sizeof(struct join_query_pkt));
    ap->u.joinq_p.q_lid = (int) ntohl(data->q_lid);
    //    ap->u.joinq_p.q_lid = (int) ntohl(ap->u.joinq_p.q_lid);
    ap->u.joinq_p.attach = (bool) (ntohl(data->attach));
    //    printf("send time is %f\n",data->send_time);

    ap->u.joinq_p.send_time = ((double) (long) ntohl(data->send_time) )/ PRECISION_DOUBLE;
    //    memcpy ((char *)&ap->u.joinq_p.send_time, (char *)&data->send_time, sizeof(double));


    //    printf("send time is %f\n",ap->u.joinq_p.send_time);
    memcpy(& ap->u.joinq_p.reply_addr, & data->reply_addr, sizeof(struct sockaddr_in));
    newlen += sizeof(struct const_join_query_pkt);
  }
    break;

  case (JOIN_FORWARD) : {
   
    struct const_join_forward_pkt * data = (struct const_join_forward_pkt *) ((char *)recv_buffer + newlen);
    memcpy(& ap->u.joinforward_p.src_ag, & data->src_ag, sizeof(struct PacketAgentInfo));
    ap->u.joinforward_p.src_ag.agent_id = (int) ntohl(ap->u.joinforward_p.src_ag.agent_id);
    ap->u.joinforward_p.q_lid = (int) ntohl(data->q_lid);
    ap->u.joinforward_p.attach = (bool) (ntohl(data->attach));
    memcpy(& ap->u.joinforward_p.original_dst, & data->original_dst, sizeof(struct PacketAgentInfo));
    ap->u.joinforward_p.original_dst.agent_id = (int) ntohl(ap->u.joinforward_p.original_dst.agent_id);
    ap->u.joinforward_p.send_time = ((double) (long) ntohl(data->send_time)) / PRECISION_DOUBLE;
    //    memcpy(& ap->u.joinforward_p.send_time, & data->send_time, sizeof(double));
    newlen += sizeof(struct const_join_forward_pkt);
  }
    break;

  case (JOIN_RESPONSE) : {
    
    struct const_join_response_pkt * data = (struct const_join_response_pkt *) ((char *) recv_buffer + newlen);
    //    memcpy(& ap->u.joinresp_p, data, sizeof(struct const_join_response_pkt));
    ap->u.joinresp_p.accept = (bool)(ntohl(data->accept));
    ap->u.joinresp_p.layer_id = (int) ntohl(data->layer_id);
    ap->u.joinresp_p.mbr_count = (int) ntohl(data->mbr_count);
    ap->u.joinresp_p.your_id = (int) ntohl(data->your_id);
    /*    ap->u.joinresp_p.layer_id = (int) ntohl(ap->u.joinresp_p.layer_id);
    ap->u.joinresp_p.mbr_count = (int) ntohl(ap->u.joinresp_p.mbr_count);
    ap->u.joinresp_p.your_id = (int) ntohl(ap->u.joinresp_p.your_id);
    */
    memcpy(& ap->u.joinresp_p.exp_src, & data->exp_src, sizeof(struct PacketAgentInfo));
    ap->u.joinresp_p.exp_src.agent_id = (int) ntohl(ap->u.joinresp_p.exp_src.agent_id);
    newlen += sizeof(struct const_join_response_pkt);

    //    memcpy(ap->u.joinresp_p.agent_arr, (recv_buffer + newlen), (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 *) ((char *) recv_buffer + newlen);
      ap->u.joinresp_p.agent_arr[i].num_children = (int) ntohl(tempinfo[i].num_children);
      ap->u.joinresp_p.agent_arr[i].ag.agent_id = (int) ntohl(tempinfo[i].ag.agent_id);
      memcpy(& ap->u.joinresp_p.agent_arr[i].ag.agent_addr, & tempinfo[i].ag.agent_addr, sizeof(struct sockaddr_in));
      ap->u.joinresp_p.agent_arr[i].dist = ((double) (long) ntohl(tempinfo[i].dist)) / PRECISION_DOUBLE;
    }
    newlen += (ap->u.joinresp_p.mbr_count)*sizeof(struct sriTreeInfo);

    /*
    printf("reclaiming 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 *)recv_buffer + newlen);
    ap->u.clusterrefresh_p.layer_id = (int) ntohl(data->layer_id);
    ap->u.clusterrefresh_p.mbr_count = (int) ntohl(data->mbr_count);
    ap->u.clusterrefresh_p.hl_mbr_count = (int) ntohl(data->hl_mbr_count); 
    ap->u.clusterrefresh_p.is_root = (bool) ntohl(data->is_root);
    ap->u.clusterrefresh_p.root_xfer = (bool) ntohl(data->root_xfer);
    ap->u.clusterrefresh_p.cluster_remove = (bool) ntohl(data->cluster_remove);
    newlen += sizeof(struct const_cluster_refresh_pkt);

    //    memcpy(ap->u.clusterrefresh_p.agent_arr, recv_buffer + newlen, (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 *) ((char *) recv_buffer + newlen);
      ap->u.clusterrefresh_p.agent_arr[i].num_children = (int) ntohl(tempinfo[i].num_children);
      ap->u.clusterrefresh_p.agent_arr[i].ag.agent_id = (int) ntohl(tempinfo[i].ag.agent_id);
      memcpy(& ap->u.clusterrefresh_p.agent_arr[i].ag.agent_addr, & tempinfo[i].ag.agent_addr, sizeof(struct sockaddr_in));
      ap->u.clusterrefresh_p.agent_arr[i].dist = (double) (long) ntohl(tempinfo[i].dist);
    }
    newlen += (ap->u.clusterrefresh_p.mbr_count)*sizeof(struct sriTreeInfo);

    ap->u.clusterrefresh_p.hl_agent_arr = (struct TreeInfo *) malloc((ap->u.clusterrefresh_p.hl_mbr_count)*sizeof(struct TreeInfo));
    //    memcpy(ap->u.clusterrefresh_p.hl_agent_arr, recv_buffer + newlen, (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 *) ((char *) recv_buffer + newlen);
      ap->u.clusterrefresh_p.hl_agent_arr[i].num_children = (int) ntohl(tempinfo[i].num_children);
      ap->u.clusterrefresh_p.hl_agent_arr[i].ag.agent_id = (int) ntohl(tempinfo[i].ag.agent_id);
      memcpy(& ap->u.clusterrefresh_p.hl_agent_arr[i].ag.agent_addr, & tempinfo[i].ag.agent_addr, sizeof(struct sockaddr_in));
      ap->u.clusterrefresh_p.hl_agent_arr[i].dist = (double) (long) ntohl(tempinfo[i].dist);
    }
    newlen += (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 *)recv_buffer + newlen);
    ap->u.clustermerge_p.layer_id = (int) ntohl(data->layer_id); 
    ap->u.clustermerge_p.mbr_count = (int) ntohl(data->mbr_count); 
    newlen += sizeof(struct const_cluster_merge_pkt);

    memcpy(ap->u.clustermerge_p.agent_arr, recv_buffer + newlen, (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 *) ((char *) recv_buffer + newlen);
      ap->u.clustermerge_p.agent_arr[i].num_children = (int) ntohl(tempinfo[i].num_children);
      ap->u.clustermerge_p.agent_arr[i].ag.agent_id = (int) ntohl(tempinfo[i].ag.agent_id);
      memcpy(& ap->u.clustermerge_p.agent_arr[i].ag.agent_addr, & tempinfo[i].ag.agent_addr, sizeof(struct sockaddr_in));
      ap->u.clustermerge_p.agent_arr[i].dist = (double) (long) ntohl(tempinfo[i].dist);
    }
    newlen += (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 *)recv_buffer + newlen);
    ap->u.pingq_p.src_time = ((double) (long) ntohl(data->src_time)) / PRECISION_DOUBLE;
    //    memcpy(& ap->u.pingq_p.src_time, & data->src_time, sizeof(double)); 
    ap->u.pingq_p.lid = (int) ntohl(data->lid);
    newlen += sizeof(struct ping_query_pkt);
  }
    break; 

  case (PING_RESPONSE) : {

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

    ap->u.pingresp_p.agent_arr = (struct TreeInfo *) malloc((ap->u.pingresp_p.mbr_count)*sizeof(struct TreeInfo));
    memcpy(ap->u.pingresp_p.agent_arr, recv_buffer + newlen, (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 *) ((char *) recv_buffer + newlen);
      ap->u.pingresp_p.agent_arr[i].num_children = (int) ntohl(tempinfo[i].num_children);
      ap->u.pingresp_p.agent_arr[i].ag.agent_id = (int) ntohl(tempinfo[i].ag.agent_id);
      memcpy(& ap->u.pingresp_p.agent_arr[i].ag.agent_addr, & tempinfo[i].ag.agent_addr, sizeof(struct sockaddr_in));
      ap->u.pingresp_p.agent_arr[i].dist = (double) (long) ntohl(tempinfo[i].dist);
    }
    newlen += (ap->u.pingresp_p.mbr_count)*sizeof(struct sriTreeInfo);

  }
    break;

  case (PACKET_DATA) : {

    struct const_data_pkt * data = (struct const_data_pkt *) ((char *)recv_buffer + newlen);
    memcpy(& ap->u.data_p.original_src, & data->original_src, sizeof(struct PacketAgentInfo));
    ap->u.data_p.original_src.agent_id = (int) ntohl(ap->u.data_p.original_src.agent_id);
    ap->u.data_p.seq_no = (int) ntohl(data->seq_no); 
    ap->u.data_p.lid = (int) ntohl(data->lid); 
    ap->u.data_p.base_lid = (int) ntohl(data->base_lid);
    ap->u.data_p.data_len = (int) ntohl(data->data_len);
    newlen += sizeof(struct const_data_pkt);
    /* copy variable part */
    ap->u.data_p.payload = (char *)malloc(ap->u.data_p.data_len);
    memcpy(ap->u.data_p.payload, recv_buffer + newlen, ap->u.data_p.data_len);
    newlen += ap->u.data_p.data_len;
  }
    break; 

  case (PACKET_DATA_ACK) : {

    struct dataack_pkt * data = (struct dataack_pkt *) ((char *)recv_buffer + newlen);