layerinfo.cc,v

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

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


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


desc
@@


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

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

//#define CLOCK 0.00 

NamedConstant const_join_query_timeout("cjqt",2.0);
NamedConstant const_cluster_refresh_msg_timeout("ccrmt",10.0);
NamedConstant const_cluster_refresh_check_timeout("ccrct",20.0);
NamedConstant const_higher_layer_ping_response_timeout("chlprt",15.0);
NamedConstant const_higher_layer_ping_interval_timeout("chlpit",30.0);

LayerAgentInfo::~LayerAgentInfo (void) {
 if (lower_layer != NULL) 
   delete lower_layer;
 if (agent_dist_arr_count > 0)
   delete [] agent_dist_arr;
}

void LayerAgentInfo::refresh_agent (AppPacket *ap) {

  assert (ap->st == CLUSTER_REFRESH);
  refresh = true;
  if (agent_dist_arr != NULL) {
    delete [] agent_dist_arr;
    agent_dist_arr = NULL;
  }
  put_cluster_refresh_packet_dist_info_into_layer_agent (ap,this);
  //  dist = /* Scheduler::my_clock() */ my_clock() - ap->send_time;
  //  printf("calculating wrong distance\n");
  
  if ((valid_dist_clock == false) || (my_clock() - dist_clock >= DIST_CLOCK_THRESHOLD)) {
    //  dist = MAX_RTT;
    struct sockaddr_in their_addr; 
    memcpy(& their_addr, & ag.agent_addr, sizeof(struct sockaddr_in));
    their_addr.sin_port = htons(DIST_EST_PORT);
    dist = estimate_rtt_wrapper(their_addr);
    dist_clock = my_clock();
    valid_dist_clock = true;
  }

  return;
}

LayerInfo::LayerInfo (void) {

  lid = -1; 
  me_in_layer = false; 
  root = NULL;
  delay_root_xfer = false;
}

LayerInfo::LayerInfo (int Lid, Agent * A) {

  lid = Lid; 
  me_in_layer = false; 
  root = NULL;
  delay_root_xfer = false;
  cr_msgt.a = A;
  cr_msgt.l = this;
  cr_chkt.a = A;
  cr_chkt.l = this;
}

/* Adds all members in the refresh/merge packet that are not known to me
 * Called only when I am the root and the pkt src thinks it is root too
 * or when merge is happening
 * Returns the number of new members added.
 */
int LayerInfo::AddExtraMembersFromAltRootPacket (AppPacket * ap) {

  int ret_count = 0;
  int count;
  if (ap->st == CLUSTER_REFRESH)
    count = ap->u.clusterrefresh_p.mbr_count;
  else {
    assert (ap->st == CLUSTER_MERGE);
    count = ap->u.clustermerge_p.mbr_count;
  }

  for (int i = 0; i < count; i++) {

    AgentInfo this_ag;
    if (ap->st == CLUSTER_REFRESH) {
      this_ag.agent_id = ap->u.clusterrefresh_p.agent_arr[i].ag.agent_id;
      memcpy(& this_ag.agent_addr, & ap->u.clusterrefresh_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));
      //this_ag.node_id = ap->u.clusterrefresh_p.agent_arr[i].ag.node_id;
    }
    else {
      this_ag.agent_id = ap->u.clustermerge_p.agent_arr[i].ag.agent_id;
      memcpy(& this_ag.agent_addr, & ap->u.clustermerge_p.agent_arr[i].ag.agent_addr, sizeof(struct sockaddr_in));      
      //this_ag.node_id = ap->u.clustermerge_p.agent_arr[i].ag.node_id;
    }

    if (ag_list.Locate(this_ag.agent_id) == NULL) {
      LayerAgentInfo * new_la = new LayerAgentInfo(this_ag.agent_id,this_ag.agent_addr);
      bool success = AddClusterMember(new_la);
      assert (success == true);
      ret_count ++;
    }
  }
  return ret_count;
}

LayerInfo::~LayerInfo (void) {

  MakeEmpty();
}

void LayerInfo::MakeEmpty (void) {

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

    delete (ag_list.GetAt(pos));
  }
  RemoveAgents();
  cr_chkt.CancelTimer();
  cr_msgt.CancelTimer();
  return;
}

void LayerInfo::RemoveAgents (void) {

  ag_list.RemoveAll();
  root = NULL;
  me_in_layer = false;
  // cluster_tree.RemoveAll();
  // me_pos = NULL;
  return;
}

bool LayerInfo::AddClusterMember (LayerAgentInfo *a) {

  return AddToLayerCluster(a);

  // void * root_pos = cluster_tree.GetRootPosition();
  // assert (root_pos != NULL);
  // return cluster_tree.AddChild(a,root_pos);
}


bool LayerInfo::AddClusterRoot (LayerAgentInfo *a) {

  bool ret_val = AddToLayerCluster(a);
  if (ret_val == false)
    root = FindClusterMember(a->ag.agent_id);
  else
    root = a;
  return ret_val;

  // void * root_pos = cluster_tree.GetRootPosition();
  // assert (root_pos == NULL);
  // return cluster_tree.SetRoot(a);
}

bool LayerInfo::AddToLayerCluster (LayerAgentInfo *a) {

  void *pos = ag_list.Locate(a->ag.agent_id);
  if (pos != NULL)
    return false;

  ag_list.Add(a,a->ag.agent_id);
  return true;
}

bool LayerInfo::DeleteClusterMember (LayerAgentInfo *a) {

  void * pos = ag_list.Locate(a->ag.agent_id);
  if (pos == NULL) {
    return false;
  }
  ag_list.RemoveAt(pos);
  if (root != NULL) {
    if (root->ag.agent_id == a->ag.agent_id) //&& (root->ag.node_id == a->ag.node_id) )
      root = NULL;
  }
  return true;
}

LayerAgentInfo * LayerInfo::FindClusterMember (int aid) {

  void * pos = ag_list.Locate(aid);
  if (pos == NULL)
    return NULL;
  return ag_list.GetAt(pos);
}
  
LayerAgentInfo ** LayerInfo::CreateLayerAgentInfoArray (void) {

  LayerAgentInfo ** la_arr = (LayerAgentInfo **) safe_malloc (sizeof(LayerAgentInfo *) * ag_list.GetSize());
  int i = 0;
  for (void * pos = ag_list.GetHeadPosition();
       pos != NULL;
       ag_list.GetNext(pos) ) {

    la_arr[i++] = ag_list.GetAt(pos);
  }
  return la_arr;
}

double * create_cost_matrix (int self_aid, /* int self_nid, */ LayerAgentInfo ** ag_arr, int ag_arr_count) {

  int my_agent_count = ag_arr_count;
  int num_square = my_agent_count * my_agent_count;

  double * cost_matrix = (double *) safe_malloc (sizeof(double) * num_square );
  for (int i = 0; i < num_square; i++)
    cost_matrix[i] = -1.0;

  int self_index = -1;
  // Fill in the other rows
  for (int i = 0; i < my_agent_count; i++) {
    if (ag_arr[i]->ag.agent_id != self_aid) /* || (ag_arr[i]->ag.node_id  != self_nid) ) */ {
      for (int j = 0; j < ag_arr[i]->agent_dist_arr_count; j++) {
	for (int k = 0; k < my_agent_count; k++) {
	  if ( (ag_arr[i]->agent_dist_arr[j].ag.agent_id == ag_arr[k]->ag.agent_id)/* && (ag_arr[i]->agent_dist_arr[j].ag.node_id == ag_arr[k]->ag.node_id)*/ ) {
	    cost_matrix[i*my_agent_count + k] = ag_arr[i]->agent_dist_arr[j].dist;
	  }
	}
      }
    }
    else
      self_index = i;
  }

  // Fill in the my row
  if (self_index >= 0) {
    for (int i = 0; i < my_agent_count; i++)
      cost_matrix[self_index*my_agent_count + i] = ag_arr[self_index]->dist;
  }

  // Make matrix symmetrical
  int unfilled_count = 0;
  for (int i = 0; i < my_agent_count; i++) {
    for (int j = i; j < my_agent_count; j++) {
      if (i == j)
	cost_matrix[i*my_agent_count+j]= 0.0;
      else {
	if ( (cost_matrix[i*my_agent_count+j] < 0.0) && (cost_matrix[j*my_agent_count+i] < 0.0) ) {
	  unfilled_count ++;
	  continue;
	}
	if (cost_matrix[i*my_agent_count+j] < 0.0)
	  cost_matrix[i*my_agent_count+j] = cost_matrix[j*my_agent_count+i];
	else if (cost_matrix[j*my_agent_count+i] < 0.0)
	  cost_matrix[j*my_agent_count+i] = cost_matrix[i*my_agent_count+j];
	else { // both are >= 0.0
	  double avg = (cost_matrix[i*my_agent_count+j] + cost_matrix[j*my_agent_count+i]) / 2.0;
	  cost_matrix[i*my_agent_count+j] = avg;
	  cost_matrix[j*my_agent_count+i] = avg;
	}
      }
    }
  }

  if (unfilled_count > 0) {

    int * unfilled_x = (int *) safe_malloc (sizeof(int) * unfilled_count);
    int * unfilled_y = (int *) safe_malloc (sizeof(int) * unfilled_count);

    // Now try triangle inequality to fill in gaps
    int index = 0;
    for (int i = 0; i < my_agent_count; i++) {
      for (int j = i+1; j < my_agent_count; j++) {
	
	if (cost_matrix[i*my_agent_count+j] < 0.0) {
	  unfilled_x[index] = i;
	  unfilled_y[index] = j;
	  index ++;
	}
      }
    }
    
    int unfilled_left = unfilled_count;
    bool cost_matrix_improved;
    do {
      cost_matrix_improved = false;
      for (int i = 0; i < unfilled_count; i++) {
	  for (int j = 0; j < my_agent_count; j++) {
	    if ( (j == unfilled_x[i]) || (j == unfilled_y[i]) )
	      continue;
	    if ( ( cost_matrix[j*my_agent_count+unfilled_x[i]] < 0.0) ||
		 ( cost_matrix[j*my_agent_count+unfilled_y[i]] < 0.0) )
	      continue;
	    int this_index = unfilled_x[i]*my_agent_count + unfilled_y[i];
	    double sum = cost_matrix[j*my_agent_count+unfilled_x[i]] + cost_matrix[j*my_agent_count+unfilled_y[i]];
	    if ( (cost_matrix[this_index] < 0.0) || (cost_matrix[this_index] > sum) ) {
	      if (cost_matrix[this_index] < 0.0)
		unfilled_left --;
	      cost_matrix_improved = true;
	      cost_matrix[this_index] = sum;
	      cost_matrix[unfilled_y[i]*my_agent_count + unfilled_x[i]] = sum;
	    }
	  }
      }
    } while ( (unfilled_left > 0) && (cost_matrix_improved == true) );
    free (unfilled_x);
    free (unfilled_y);
    
    if (unfilled_left > 0) {
      printf ("[Cost] cost-matrix-begin\n[Cost]    : ");
      for (int i = 0; i < my_agent_count; i++)
	printf ("%d  ", ag_arr[i]->ag.agent_id);
      printf("\n[Cost]\n");
      
      for (int i = 0; i < my_agent_count; i++) {
	printf ("[Cost] %d : ", ag_arr[i]->ag.agent_id);
	for (int j = 0; j < my_agent_count; j++) {
	  printf ("%5.3f ", cost_matrix[i*my_agent_count+j]);
	}
	printf ("\n");
      }
      printf ("\n[Cost] cost-matrix-end\n\n");
    }

    assert (unfilled_left == 0);
  }

  return cost_matrix;
}

double * delete_agent_from_cost_matrix (double * in_cost, LayerAgentInfo ** ag_arr, int in_count, int aid /*, int nid*/) {

  int ag_index = -1;
  for (int i = 0; i < in_count; i++) {
    if ( (ag_arr[i]->ag.agent_id == aid) /*&& (ag_arr[i]->ag.node_id == nid)*/ ) {
      ag_index = i;
      break;
    }
  }
  assert (ag_index >= 0);

  int out_count = in_count - 1;
  double * out_cost = (double *) safe_malloc (sizeof(double) * out_count * out_count);

  int out_i = 0;
  for (int i = 0; i < in_count; i++) {
    int out_j = 0;
    if (i != ag_index) {
      for (int j = 0; j < in_count; j++) {
	
	if (j != ag_index) {
	  out_cost[out_i*out_count + out_j] = in_cost[i*in_count + j];
	  out_j ++;
	}
      }
      out_i ++;
    }
  }

  for (int i = ag_index+1; i < in_count; i++)
    ag_arr[i-1] = ag_arr[i];

  return out_cost;
}

void ClusterRefreshMsgTimer::EventHandler (bool from_handle_expired_timers) {

  //if (((commonAgent *)a)->handle_timer_now || from_handle_expired_timers) {
    if (a->t == AGENT_APPLICATION_BSE) 
      ((bseAgent*)a)->handle_cluster_refresh_msg_timeout();
    else
      ((coopAgent*)a)->handle_cluster_refresh_msg_timeout(l->lid);
  //}
  //else {
    //((commonAgent *)a)->add_expired_timer(this);
  //  add_expired_timer(this);
  //}
  return;
}

void ClusterRefreshCheckTimer::EventHandler (bool from_handle_expired_timers) {

  //if (((commonAgent *)a)->handle_timer_now || from_handle_expired_timers) {
    if (a->t == AGENT_APPLICATION_BSE)
      ((bseAgent*)a)->handle_cluster_refresh_check_timeout();
    else
      ((coopAgent*)a)->handle_cluster_refresh_check_timeout(l->lid);
  //}
  //else {
    //((commonAgent *)a)->add_expired_timer(this);
  //  add_expired_timer(this);
  //}
  return;
}
@