common.cc,v

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

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


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


desc
@@


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

#include <stdio.h>
#include <stdlib.h>
#include <assert.h>

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

//#define CLOCK 0.00

void init_join_query_packet (AppPacket *ap, int lid, double time, bool attach, struct sockaddr_in reply_addr) {

  ap->u.joinq_p.q_lid = lid;
  ap->u.joinq_p.send_time = time;
  ap->u.joinq_p.attach = attach;
  memcpy(& ap->u.joinq_p.reply_addr, & reply_addr, sizeof(struct sockaddr_in));
  return;
}

void put_info_into_join_forward_packet (AgentInfo src_ag, int qlid, AgentInfo orig_dst, double send_time, bool attach, AppPacket *ap) {

  ap->u.joinforward_p.src_ag.agent_id = src_ag.agent_id;
  //  ap->u.joinforward_p.src_ag.node_id = src_ag.node_id;
  memcpy(& ap->u.joinforward_p.src_ag.agent_addr, & src_ag.agent_addr, sizeof(struct sockaddr_in));
  ap->u.joinforward_p.original_dst.agent_id = orig_dst.agent_id;
  //  ap->u.joinforward_p.original_dst.node_id = orig_dst.node_id;
  memcpy(& ap->u.joinforward_p.original_dst.agent_addr, & orig_dst.agent_addr, sizeof(struct sockaddr_in));
  ap->u.joinforward_p.q_lid = qlid;
  ap->u.joinforward_p.attach = attach;
  ap->u.joinforward_p.send_time = send_time;
  return;
}

void put_hl_cluster_info_into_packet (LayerInfo *l, AppPacket *ap, AgentInfo * ignore_ag) {

  assert (ap->st == CLUSTER_REFRESH);
  if (l == NULL)
    return;
  if (l->ag_list.GetSize() == 0)
    return;
  // assert ( (l->me_in_layer == true) && (l->ag_list.GetSize() >= 1) );
  ap->u.clusterrefresh_p.hl_mbr_count = l->ag_list.GetSize();
  ap->u.clusterrefresh_p.hl_agent_arr = (TreeInfo *) safe_malloc(l->ag_list.GetSize() * sizeof(TreeInfo));
  int count = create_cluster_tree_info_in_array(l, ap->u.clusterrefresh_p.hl_agent_arr,ignore_ag,true);
  ap->u.clusterrefresh_p.hl_mbr_count = count;
  return;
}

void put_cluster_refresh_hl_info_into_layer_agent (AppPacket *ap, LayerInfo *l, int self_aid, struct sockaddr_in sa) {

  assert (ap->st == CLUSTER_REFRESH);
  assert (l != NULL);

  assert (l->ag_list.Locate(self_aid) == NULL);

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

  TreeInfo * arr = ap->u.clusterrefresh_p.hl_agent_arr;
  int size = ap->u.clusterrefresh_p.hl_mbr_count;
  for (int i = 0; i < size; i++) {

    if ( (arr[i].ag.agent_id == self_aid) /* && (arr[i].ag.node_id == self_nid) */) {
      printf ("[Err] [ %d ] getting self in higher layer !\n", self_aid);
      continue;
    }
    void * pos = l->ag_list.Locate(arr[i].ag.agent_id);
    LayerAgentInfo * ag;
    if (pos == NULL) {
      ag = new LayerAgentInfo (arr[i].ag.agent_id,arr[i].ag.agent_addr);
      l->ag_list.Add(ag,ag->ag.agent_id);
    }
    else {
      ag = l->ag_list.GetAt(pos);
    }

    ag->valid_tmp = true;

    if ( (ap->src_agent == ag->ag.agent_id) /*&& (ap->src == ag->ag.node_id)*/ ) {
      //ag->dist = /* Scheduler::my_clock() */ my_clock() - ap->send_time;
      if ((ag->valid_dist_clock == false) || (my_clock() - ag->dist_clock >= DIST_CLOCK_THRESHOLD)) {
      //  dist = MAX_RTT;
	struct sockaddr_in their_addr; 
	memcpy(& their_addr, & ag->ag.agent_addr, sizeof(struct sockaddr_in));
	their_addr.sin_port = htons(DIST_EST_PORT);
	ag->dist = estimate_rtt_wrapper(their_addr);
	ag->dist_clock = my_clock();
	ag->valid_dist_clock = true;
      }
    }
    if (0 == i)
      l->root = ag;
  }

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

    LayerAgentInfo * this_la = l->ag_list.GetAt(pos);
    void * old_pos = pos;
    l->ag_list.GetNext(pos);
    if (this_la->valid_tmp == false) {
      if (l->root != NULL) {
        if ( (l->root->ag.agent_id == this_la->ag.agent_id) /*&& (l->root->ag.node_id == this_la->ag.node_id)*/ )
	  l->root = NULL;
      }
      l->ag_list.RemoveAt(old_pos);
      delete this_la;
    }
  }

  return;
}

void put_cluster_remove_info_into_packet (LayerInfo *layer, AppPacket *ap, bool root_xfer, LayerInfo * higher, int self_aid, struct sockaddr_in self_agent_addr) {

  ap->u.clusterrefresh_p.layer_id = layer->lid;
  ap->u.clusterrefresh_p.mbr_count = layer->ag_list.GetSize();
  ap->u.clusterrefresh_p.is_root = root_xfer;
  ap->u.clusterrefresh_p.root_xfer = root_xfer;
  ap->u.clusterrefresh_p.cluster_remove = true;
  create_cluster_tree_info_in_array(layer, ap->u.clusterrefresh_p.agent_arr,NULL,false);
  if ( (root_xfer == true) && (higher != NULL) ) {
    AgentInfo * self = new AgentInfo (self_aid,self_agent_addr);
    put_hl_cluster_info_into_packet(higher,ap,self);
    delete self;
  }
  return;
}

/* Copies cluster information from a layer structure to a packet */
void put_layer_cluster_info_into_packet (LayerInfo * layer, AppPacket *ap) {

  assert (layer->ag_list.GetSize() > 0);
  assert (layer->ag_list.GetSize() <= MAX_RET_COUNT);
  switch (ap->st) {

    case JOIN_RESPONSE :
      ap->u.joinresp_p.layer_id = layer->lid;
      ap->u.joinresp_p.mbr_count = layer->ag_list.GetSize();
      create_cluster_tree_info_in_array(layer, ap->u.joinresp_p.agent_arr,NULL,false);
      break;

    case CLUSTER_REFRESH :
      ap->u.clusterrefresh_p.layer_id = layer->lid;
      ap->u.clusterrefresh_p.mbr_count = layer->ag_list.GetSize();
      create_cluster_tree_info_in_array(layer, ap->u.clusterrefresh_p.agent_arr,NULL,false);
      break;

    case CLUSTER_MERGE :
      ap->u.clustermerge_p.layer_id = layer->lid;
      ap->u.clustermerge_p.mbr_count = layer->ag_list.GetSize();
      create_cluster_tree_info_in_array(layer, ap->u.clustermerge_p.agent_arr,NULL,false);
      break;

    case PING_RESPONSE :
      ap->u.pingresp_p.mbr_count = layer->ag_list.GetSize();
      ap->u.pingresp_p.agent_arr = (TreeInfo *) safe_malloc (sizeof(TreeInfo) * layer->ag_list.GetSize());
      create_cluster_tree_info_in_array(layer, ap->u.pingresp_p.agent_arr,NULL,false);
      break;

    default :
      printf ("[Err] Unknown packet subtype to copy packet\n");
      exit(-1);
  }

  return;
}

/* Copies cluster refresh dist information from a packet 
 * to a layer agent structure */
void put_cluster_refresh_packet_dist_info_into_layer_agent (AppPacket *ap, LayerAgentInfo *la_ag) {

  la_ag->agent_dist_arr_count = ap->u.clusterrefresh_p.mbr_count;
  if (la_ag->agent_dist_arr_count > 0) {
    la_ag->agent_dist_arr = new AgentDistInfo [la_ag->agent_dist_arr_count];
  }

  for (int i = 0; i < la_ag->agent_dist_arr_count; i++) {
    la_ag->agent_dist_arr[i].ag.agent_id = ap->u.clusterrefresh_p.agent_arr[i].ag.agent_id;
    memcpy(& la_ag->agent_dist_arr[i].ag.agent_addr, & ap->u.clusterrefresh_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
    //la_ag->agent_dist_arr[i].ag.node_id = ap->u.clusterrefresh_p.agent_arr[i].ag.node_id;
    la_ag->agent_dist_arr[i].dist = ap->u.clusterrefresh_p.agent_arr[i].dist;
  }
  return;
}

/* Copies cluster information from a packet to a layer structure */
void put_packet_cluster_info_into_layer (AppPacket *ap, LayerInfo *layer) {

  TreeInfo *arr;
  int arr_size;

  switch (ap->st) {

    case JOIN_RESPONSE :
      arr_size = ap->u.joinresp_p.mbr_count;
      arr = ap->u.joinresp_p.agent_arr;
      break;

    case PING_RESPONSE :
      arr_size = ap->u.pingresp_p.mbr_count;
      arr = ap->u.pingresp_p.agent_arr;
      break;

    default :
      printf ("[Err] Unknown packet subtype to copy packet\n");
      exit(-1);
  }

  put_array_info_in_cluster_tree(arr,arr_size,layer);
  // create_array_info_in_tree(arr,layer->cluster_tree);
  // add_tree_to_list(layer->cluster_tree,layer->ag_list);
  return;
}

void put_array_info_in_cluster_tree  (TreeInfo * arr, int size, LayerInfo *l) {

  assert (l->ag_list.GetSize() == 0);
  for (int i = 0; i < size; i++) {

    LayerAgentInfo *ag = new LayerAgentInfo (arr[i].ag.agent_id,arr[i].ag.agent_addr);
    l->ag_list.Add(ag,ag->ag.agent_id);
    if (0 == i)
      l->root = ag;
  } 
  return;
}

int create_cluster_tree_info_in_array (LayerInfo *l, TreeInfo *arr, AgentInfo * ignore_ag, bool root_null_possible) {

  int index = 0;
  if (l->root == NULL)
  if (root_null_possible == false)
    assert (l->root != NULL);
  bool add_root = true;
  if (ignore_ag != NULL) {
    if (l->root != NULL) {
      if (ignore_ag->agent_id == l->root->ag.agent_id) // && (ignore_ag->node_id == l->root->ag.node_id) )
        add_root = false;
    }
  }

  if ( (l->root != NULL) && (add_root == true) ) {
    put_info_into_tree_array_struct(arr[index], l->root->ag.agent_id, l->root->ag.agent_addr, l->ag_list.GetSize() - 1, l->root->dist);
    index ++;
  }

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

    if (ignore_ag != NULL) 
      if (ignore_ag->agent_id == la_ag->ag.agent_id) //&& (ignore_ag->node_id == la_ag->ag.node_id) )
        continue;

    if (l->root != NULL)
      if (la_ag->ag.agent_id == l->root->ag.agent_id) //&& (la_ag->ag.node_id == l->root->ag.node_id) )
        continue;
    put_info_into_tree_array_struct(arr[index], la_ag->ag.agent_id, la_ag->ag.agent_addr, 0, la_ag->dist);
    index ++;
  }

  return index;
}

void put_info_into_tree_array_struct (TreeInfo& t, int aid, struct sockaddr_in sa, int num_children, double dist) {

  t.ag.agent_id = aid;
  memcpy(& t.ag.agent_addr, & sa, sizeof(struct sockaddr_in));
  //  t.ag.node_id = nid;
  t.num_children = num_children;
  t.dist = dist;
  return;
}

/*
void add_tree_to_list (SimpleTree<LayerAgentInfo *> & tree, LinkedList<LayerAgentInfo *, int> & list) {

  void * pos = tree.GetRootPosition();
  if (pos != NULL)
    recursive_add_tree_to_list(tree,pos,list);
  return;
}

void recursive_add_tree_to_list (SimpleTree<LayerAgentInfo *> & tree, void * pos, LinkedList<LayerAgentInfo *, int> & list) {

  LayerAgentInfo * la_ag = tree.GetAt(pos);
  list.Add(la_ag,la_ag->ag.agent_id);
  int cc = tree.GetChildCount(pos);
  void ** child_pos_arr = tree.GetChildList(pos);
  for (int i = 0; i < cc; i++)
    recursive_add_tree_to_list(tree,child_pos_arr[i],list);
  delete [] child_pos_arr;
  return;
}
*/

/*
void create_array_info_in_tree (TreeInfo * arr, SimpleTree<LayerAgentInfo *> & tree) {

  assert (tree.GetSize() == 0);

  int count = add_array_info_in_tree(arr,0,tree,NULL);
  assert (count == tree.GetSize());
  return;
}
*/

/* Given the tree array, it creates the simple tree structure back.
 * Should be called as (arr,0,tree,NULL)
 */
/*
int add_array_info_in_tree (TreeInfo * arr, int index, SimpleTree<LayerAgentInfo *> & tree, void * curr_pos) {

  LayerAgentInfo *ag = new LayerAgentInfo (arr[index].ag.agent_id,arr[index].ag.node_id);

  void * new_curr_pos = tree.AddChild(curr_pos);
  int child_count = arr[index].num_children;
  index++;
  for (int i = 0; i < child_count; i++)
    index = add_array_info_in_tree(arr,index,tree,new_curr_pos);
  return index;
}
*/

/* Create an array for the tree defined in the tree structure.
 * It should be called with (tree,tree->GetRootPos(),arr,0)
 */
/*
int add_tree_info_to_array (SimpleTree<LayerAgentInfo *> & tree, void * pos, TreeInfo * arr, int index) {

  if (pos != NULL)
    LayerAgentInfo * node = tree.GetAt(pos);
  else {
    assert (index == 0);
    return index;
  }
  arr[index].ag.agent_id = node->ag.agent_id;
  arr[index].ag.node_id = node->ag.node_id;
  arr[index].num_children = tree.GetChildCount(pos);
  arr[index].dist = node->dist;
  index ++;

  void ** child_pos_arr = tree.GetChildList(pos);
  for (int i = 0; i < tree.GetChildCount(pos); i++) {
    index = add_tree_info_to_array(tree,child_pos_arr[i],arr,index);
  }

  if (tree.GetChildCount(pos) > 0)
    free(child_pos_arr);

  return index;
}
*/

void * safe_malloc (int size) {

  void *p = malloc (size);
  if (p == NULL) {
     perror("malloc:");
     exit(-2);
     return NULL;
  }
  return p;
}

@