pathrate_rcv_func.c

pathrate是pathload的改进型

/*
 This file is part of pathrate.

 pathrate is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation; either version 2 of the License, or
 (at your option) any later version.

 pathrate is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with pathrate; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

/*-------------------------------------------------
   pathrate : an end-to-end capcity estimation tool
   Author   : Constantinos Dovrolis ([e-mail] )
              Ravi S Prasad            ( [e-mail] )
              
   Release  : Ver 2.4.1
   Support  : This work was supported by the SciDAC
              program of the US department 
--------------------------------------------------*/


#include "pathrate.h"
#include "pathrate_rcv.h"

/*-----------------------------------------------------------------------------*
 *  * Weiling: function get_kurtosis
 *   * requires at least 3 samples,
 *    * returns -99999 if error
 *     *-----------------------------------------------------------------------------*/
double get_kurtosis(double* bell_array, int size){
  double kurtosis, var, mean, temp, diff;
  int i;

  if(size < 3)
    return -99999;

  temp = 0;
  for(i = 0; i < size; i++)
    temp += bell_array[i];
  mean = temp / size;

  temp = 0;
  for(i = 0; i < size; i++) {
    diff = bell_array[i] - mean;
    temp += diff * diff;
  }
  var = temp / size;

  if(var == 0)
    return -99999;

  temp = 0;
  for(i = 0; i < size; i++){
    diff = bell_array[i] - mean;
    temp += diff * diff * diff * diff;
  }

  kurtosis = temp / (var * var);
  return kurtosis;
}

void prntmsg(FILE *fp) {
  fprintf(fp, "%s", message);
  fflush(fp);
}

/*
  Send a message through the control stream
*/
void send_ctr_msg(long ctr_code) 
{
  char ctr_buff[24];
  long ctr_code_n = htonl(ctr_code);
  memcpy((void*)ctr_buff, &ctr_code_n, sizeof(long));
  if (write(sock_tcp, ctr_buff, sizeof(long)) != sizeof(long)) {
    fprintf(stderr, "send control message failed:\n");
    exit(-1);
  }
  //printf ("Done sending %x\n", (ctr_code & 0x000000FF));
}


/*
        Receive an empty message from the control stream
*/
long recv_ctr_msg(int ctr_strm, char *ctr_buff)
{
  long ctr_code;
  if (read(ctr_strm, ctr_buff, sizeof(long)) != sizeof(long)){
    fprintf(stderr, "recv control message failed:\n");
    return(-1);
  }
  memcpy(&ctr_code, ctr_buff, sizeof(long));
  return(ntohl(ctr_code));
}


/*
  Print a bandwidth measurement (given in Mbps) in the appropriate units
*/
void print_bw(FILE *fp, double bw_v) 
{
  if (bw_v < 1.0) {
    sprintf(message," %.0f kbps ", bw_v*1000);prntmsg(fp);
  }
  else if (bw_v < 15.0) {
    sprintf(message," %.1f Mbps ", bw_v);prntmsg(fp);
  }
  else {   
    sprintf(message," %.0f Mbps ", bw_v);prntmsg(fp);
  }
}


/* 
   Terminate measurements  
*/
void termint(int exit_code) {
  int ctr_code;

  ctr_code = CONTINUE;
  send_ctr_msg(ctr_code);
  ctr_code = GAME_OVER;
  send_ctr_msg(ctr_code);
  fclose(pathrate_fp); close(sock_tcp); close(sock_udp);
  exit(exit_code);
}

/* Successful termination. Print result.  */
void happy_end(double bw_lo, double bw_hi)
{
  sprintf(message,"\n\n-------------------------------------------------\n");
  prntmsg(pathrate_fp);
  sprintf(message,"Final capacity estimate : ");prntmsg(pathrate_fp);
  print_bw(pathrate_fp, bw_lo); 
  sprintf(message," to ");prntmsg(pathrate_fp); 
  print_bw(pathrate_fp, bw_hi); 
  sprintf(message,"\n");prntmsg(pathrate_fp);
  sprintf(message,"-------------------------------------------------\n\n");
  prntmsg(pathrate_fp);
  if(verbose){
    fprintf(stdout,"-------------------------------------------------\nFinal capacity estimate : ");
    print_bw(stdout,bw_lo);
    fprintf(stdout," to ");
    print_bw(stdout,bw_hi);
    fprintf(stdout," \n-------------------------------------------------\n");
  }
  if (netlog) { /* print result for netlooger */
    struct tm *tm;
    struct hostent *rcv_host, *snd_host;
    char rcv_name[256];
    struct timeval curr_time;

    gettimeofday(&curr_time,NULL);
    tm = gmtime(&curr_time.tv_sec);
    fprintf(netlog_fp,"DATE=%4d",tm->tm_year+1900);
    print_time(netlog_fp,tm->tm_mon+1);
    print_time(netlog_fp,tm->tm_mday);
    print_time(netlog_fp,tm->tm_hour);
    print_time(netlog_fp,tm->tm_min);
    if (tm->tm_sec <10) {
      fprintf(netlog_fp,"0");
      fprintf(netlog_fp,"%1.6f",tm->tm_sec+curr_time.tv_usec/1000000.0);
    }
    else{
      fprintf(netlog_fp,"%1.6f",tm->tm_sec+curr_time.tv_usec/1000000.0);
    }
    gethostname(rcv_name, 255);
    rcv_host = gethostbyname(rcv_name);
    if(strcmp(rcv_name, "\0")!=0) fprintf(netlog_fp," HOST=%s",rcv_host->h_name);
    else fprintf(netlog_fp," HOST=NO_NAME");
    fprintf(netlog_fp," PROG=pathrate");
    fprintf(netlog_fp," LVL=Usage");
    if ((snd_host = gethostbyname(hostname)) == 0) {
        snd_host = gethostbyaddr(hostname,256,AF_INET);
    }
    fprintf(netlog_fp," PATHRATE.SNDR=%s",snd_host->h_name);
    fprintf(netlog_fp," PATHRATE.CAPL=%.1fMbps",bw_lo);
    fprintf(netlog_fp," PATHRATE.CAPH=%.1fMbps\n",bw_hi);
    fclose(netlog_fp);
  }
}

/* print time */
void print_time(FILE *fp, int time)
{
  if( time<10){
    fprintf(fp,"0");
    fprintf(fp,"%1d",time);
  }
  else{
    fprintf(fp,"%2d",time);
  }
}

/*
  Compute the time difference in microseconds
  between two timeval measurements
*/
double time_to_us_delta(struct timeval tv1, struct timeval tv2) 
{
   double time_us;
   time_us = (double)
     ((tv2.tv_sec-tv1.tv_sec)*1000000+(tv2.tv_usec-tv1.tv_usec));
   return time_us;
}

void time_copy(struct timeval time_val_old, struct timeval *time_val_new) 
{
   (*time_val_new).tv_sec =time_val_old.tv_sec;
   (*time_val_new).tv_usec=time_val_old.tv_usec;
}


/* 
  Order an array of doubles using bubblesort 
*/
void order(double unord_arr[], double ord_arr[], long no_elems)
{
  long i,j;
  double temp;
  for (i=0; i<no_elems; i++) ord_arr[i]=unord_arr[i];
  for (i=1; i<no_elems; i++) 
    for (j=i-1; j>=0; j--) {
      if (ord_arr[j+1]<ord_arr[j]) {
         temp=ord_arr[j]; ord_arr[j]=ord_arr[j+1]; ord_arr[j+1]=temp;
      }
      else break;
    }
}


/*
  Compute the average of the set of measurements <data>.
*/
double get_avg(double data[], long no_values)
{
   long i;
   double sum_;
   sum_ = 0; 
   for (i=0; i<no_values; i++) sum_ += data[i];
   return (sum_ / (double)no_values);
}



/*
  Compute the standard deviation of the set of measurements <data>.
*/
double get_std(double data[], long no_values)
{
   long i;
   double sum_, mean;
   mean = get_avg(data, no_values);
   sum_ = 0;
   for (i=0; i<no_values; i++) sum_ += pow(data[i]-mean, 2.);
   return sqrt(sum_ / (double)(no_values-1));
}


/*
  Detect a local mode in the set of measurements <ord_data>.
  Take into account only the valid (unmarked) measurements (vld_data[i]=1). 
  The bin width of the local mode detection process is <bin_wd>. 
  The set has <no_values> elements, and it is ordered in increasing sequence.
  The function returns the center value of the modal bin.
  Also, the following call-by-ref arguments return:
    mode_cnt:    # of measurements in modal bin
    bell_cnt:    # of measurements in entire mode   
             (modal bin+surrounding bins of the same `bell')
    bell_lo:     low bandwidth threshold of modal `bell'
    bell_hi:     high bandwidth threshold of modal `bell'
*/ 
double get_mode(double ord_data[], short vld_data[], 
    double bin_wd, int no_values, mode_struct *curr_mode)
{
  int 
    i, j, done, tmp_cnt, 
    mode_lo_ind, mode_hi_ind, bell_lo_ind, bell_hi_ind,
    bin_cnt,  bin_lo_ind,  bin_hi_ind,  
    lbin_cnt, lbin_lo_ind=0, lbin_hi_ind=0,  /* left bin */
    rbin_cnt, rbin_lo_ind=0, rbin_hi_ind=0;  /* right bin */
  double
    mode_lo, mode_hi, bin_lo,  bin_hi;
  /*Weiling: bin_toler, used as */
  double bin_cnt_toler;
  

  /* Check if all measurements have been already marked */
  j=0;
  for (i=0; i<no_values; i++) j+=vld_data[i];
  if (j==0) return LAST_MODE; /* no more modes */
  #ifdef MODES 
    printf("\n%d valid measurements\n", j);
  #endif
  /* Initialize mode struct */
  curr_mode->mode_cnt=0; 
  curr_mode->bell_cnt=0; 
  curr_mode->bell_lo=0; 
  curr_mode->bell_hi=0; 

  /* 
    Find the bin of the primary mode from non-marked values 
  */

  /* Find window of length bin_wd with maximum number of consecutive values */
  mode_hi=0; mode_hi_ind=0; mode_lo=0; mode_lo_ind=0;
  tmp_cnt=0;
  for (i=0;i<no_values;i++) {
    if (vld_data[i]) {
      j=i;
      while (j<no_values && vld_data[j] && ord_data[j]<=ord_data[i]+bin_wd) j++;
      if (tmp_cnt<j-i) {
        tmp_cnt = j-i;
        mode_lo_ind = i;
        mode_hi_ind = j-1;
      }
    }
  }

  curr_mode->mode_cnt  = tmp_cnt;
  mode_lo   = ord_data[mode_lo_ind];
  mode_hi   = ord_data[mode_hi_ind];
  #ifdef MODES 
    printf("Central mode bin from %.3f to %.3f (%d msrms)", mode_lo, mode_hi, *mode_cnt);
    printf(" - (%d-%d)\n", mode_lo_ind, mode_hi_ind);
  #endif

  curr_mode->bell_cnt  = tmp_cnt;
  curr_mode->bell_lo  = mode_lo;
  curr_mode->bell_hi  = mode_hi;
  bell_lo_ind     = mode_lo_ind;
  bell_hi_ind     = mode_hi_ind;
  
  /*
        Find all the bins at the *left* of the central bin that are part of the
        same mode's bell. Stop when another local mode is detected.
  */
  bin_cnt    = curr_mode->mode_cnt;
  bin_lo_ind =  mode_lo_ind;
  bin_hi_ind =  mode_hi_ind;
  bin_lo     =  mode_lo;
  bin_hi     =  mode_hi;
  
    /* Weiling:  noise tolerance is determined by bin_cnt_toler, and it's 
     * proportional to previous bin_cnt instead of constant BIN_NOISE_TOLER . */

  done=0;
  bin_cnt_toler = BIN_CNT_TOLER_kernel_percent * (bin_cnt);
  do {
      /* find window of measurements left of the leftmost modal bin with 
       (at most) bin_wd width and with the maximum number of measurements */
    lbin_cnt=0;
    if (bin_lo_ind>0) {
      for (i=bin_hi_ind-1; i>=bin_lo_ind-1; i--) {
        tmp_cnt=0;
        for (j=i; j>=0; j--) {
          if (ord_data[j]>=ord_data[i]-bin_wd) tmp_cnt++;
          else break;
        }
        if (tmp_cnt >= lbin_cnt) {
          lbin_cnt    = tmp_cnt;
          lbin_lo_ind = j+1;
          lbin_hi_ind = i;
        }
      }
    }
    #ifdef MODES 
         printf("Left bin: %.3f to %.3f", ord_data[lbin_lo_ind], ord_data[lbin_hi_ind]);
         printf(" (%d msrms)", lbin_cnt);
         printf(" - (%d-%d)", lbin_lo_ind, lbin_hi_ind);
    #endif
  
    if (lbin_cnt>0) {
      if (lbin_cnt < bin_cnt+bin_cnt_toler) {
         /* the bin is inside the modal bell */
        #ifdef MODES 
          printf(" - Inside mode");
        #endif

        /* update bell counters */
        /* Weiling:
         In prevoius version (2.3.0), the count and lo threshold counters are updated
         only if bin has a significant number of measurements
         In this version (2.3.1), the count and lo threshold counters are updated anyway
        */

        curr_mode->bell_cnt += bin_lo_ind-lbin_lo_ind;
        curr_mode->bell_lo   = ord_data[lbin_lo_ind];
        bell_lo_ind = lbin_lo_ind;
  
        /* reset bin counters for next iteration */
        bin_cnt     = lbin_cnt;
        bin_lo_ind  = lbin_lo_ind;
        bin_hi_ind  = lbin_hi_ind;
        bin_lo      = ord_data[lbin_lo_ind];
        bin_hi      = ord_data[lbin_hi_ind];
        bin_cnt_toler = BIN_CNT_TOLER_kernel_percent * (bin_cnt);
      } 
      else {