pathrate_rcv.c

pathrate是pathload的改进型

  
  /* Total per-packet processing time: transfer from NIC to kernel, 
   * processing at the kernel, and transfer at user space. The multiplicative 
   * factor 3 is based on packet processing 
   * times reported in the literature */ 

  min_possible_delta *= 3; 
  sprintf(message,"--> Minimum acceptable packet pair dispersion: %.0f usec\n", 
      min_possible_delta);
  prntmsg(pathrate_fp);
  if (Verbose) prntmsg(stdout);

  /* The default initial train-length in Phase I is 2 packets (packet pairs) */
  train_len_P1 = 2;  
  
  /* 
    Keep a  unique identifier for each round in pathrate's execution. 
       This id is used to ignore between trains of previous rounds.  
  */
  round = 1;
  
  /*
    --------------------------------------------------
    Discover maximum feasible train length in the path
    (stop at a length that causes 3 lossy trains)
    --------------------------------------------------
  */
  sprintf(message,"\n-- Maximum train length discovery -- \n");
  prntmsg(pathrate_fp);
  
  /* Send packet size  to sender -> maximum at this phase */
  no_trains = 1;
  pack_sz=max_pack_sz;
  ctr_code = PCK_LEN | (pack_sz<<8);
  send_ctr_msg(ctr_code);
  
  /* Initial train length */
  train_len=train_len_P1; 
  path_overflow=0;
  bad_trains = 0;
  trains_msrd=0;
  
  /* Repeat for each train length */
  while (train_len <= MAX_TRAIN_LEN && !path_overflow) {
    /* Send train length to sender */
    if (!retransmit) {
      ctr_code = TRAIN_LEN | (train_len<<8);
      send_ctr_msg(ctr_code);
    }
    /* Wait for a complete packet train */
    if ( time_stmps != NULL ) free(time_stmps);
    bad_train = recv_train(train_len, &round, &time_stmps);

    /* Compute dispersion and bandwidth measurement */
    if (bad_train == 2) {
      if( bad_trains++ > 3) path_overflow = 1;
    }
    else if (!bad_train) {
       delta = time_to_us_delta(time_stmps[1], time_stmps[train_len]);
       bw_msr = ((28+pack_sz) << 3) * (train_len-1) / delta;
       sprintf(message,"\tTrain length: %d ->\t", train_len);
       prntmsg(pathrate_fp);
       if (Verbose) prntmsg(stdout);
       print_bw(pathrate_fp, bw_msr); 
       if (Verbose) print_bw(stdout, bw_msr); 
       if (delta > min_possible_delta * (train_len-1)) {
         sprintf(message,"\n");
         prntmsg(pathrate_fp);
         if (Verbose) prntmsg(stdout);
         /* Acceptable measurement */
         measurs_P1[trains_msrd++] = bw_msr;
       }
       else {
         sprintf(message,"(ignored) \n");
         prntmsg(pathrate_fp);
         if (Verbose) prntmsg(stdout);
       }
  
       /* Very slow link; the packet trains take more than their spacing
        * to be transmitted
        */  
       if (delta > train_spacing*1000) {
         path_overflow=1;
         /* Send control message to double train spacing */
         //ctr_code = CONTINUE;
         //send_ctr_msg(ctr_code);
         train_spacing = max_train_spacing;
         ctr_code = TRAIN_SPACING | (train_spacing<<8);
         send_ctr_msg(ctr_code);
       }
      
       /* Increase train length */
       if (!retransmit) {
         prev_train_len = train_len;
         if (train_len<6) train_len ++; 
         else if (train_len<12) train_len += 2; 
         else train_len += 4;
       }
    }
  }
  retransmit = 0;
  
  if (path_overflow || train_len>MAX_TRAIN_LEN) 
    max_train_len=prev_train_len;  /* Undo the previous increase. */
  else 
    max_train_len = train_len;
  sprintf(message,"\t--> Maximum train length: %d packets \n\n", max_train_len);
  prntmsg(pathrate_fp);
  if (Verbose) prntmsg(stdout); 
  /* Tell sender to continue with  next phase */
  //ctr_code = CONTINUE;
  //send_ctr_msg(ctr_code);
  
  /* Ravi: Test IC here and continue if its not IC */ 
  /* get one more train with max_train_len */

  ctr_code = TRAIN_LEN | (max_train_len<<8);
  send_ctr_msg(ctr_code);
  do {
    if ( time_stmps != NULL ) free(time_stmps);
    bad_train = recv_train(max_train_len, &round, &time_stmps);
  }while(bad_train);
  
  tmp = intr_coalescence(time_stmps, max_train_len, min_possible_delta/3);
  if (tmp) {
    sprintf(message,"Detected Interrupt Coalescence... \n");
    prntmsg(pathrate_fp);
    if (Verbose) prntmsg(stdout); 
    gig_path(max_pack_sz, round++,min_possible_delta/3);
  }
  
  /*
    ---------------------------------------------------------
    Check for possible multichannel links and traffic shapers 
    ---------------------------------------------------------
  */
  
  sprintf(message,"\n--Preliminary measurements with increasing packet train lengths--\n");
  
  prntmsg(pathrate_fp);
  if (Verbose) prntmsg(stdout); 
  /* Send number of trains to sender (five of them is good enough..) */
  no_trains = 7;  // Ravi: No need to send this to sender... we ask for each train now.
  
  /* Send packet size to sender -> maximum at this phase */
  pack_sz=max_pack_sz;
  ctr_code=PCK_LEN | (pack_sz<<8);
  send_ctr_msg(ctr_code);
  train_len=train_len_P1;
  do {
    /* Send train length to sender */
    sprintf(message,"  Train length: %d -> ", train_len);
    prntmsg(pathrate_fp);
    if (Verbose) prntmsg(stdout);

    if (!retransmit) {
      ctr_code = TRAIN_LEN | (train_len<<8);
      send_ctr_msg(ctr_code);
    }
    /* Tell sender to start sending packet trains */
    trains_rcvd=0;
    do {  /* for each train-length */ 
      /* Wait for a complete packet train */
      if ( time_stmps != NULL ) free(time_stmps);
      bad_train = recv_train(train_len, &round, &time_stmps);

      /* Compute dispersion and bandwidth measurement */
      if (!bad_train) {
        delta = time_to_us_delta(time_stmps[1], time_stmps[train_len]);
        bw_msr = ((28+pack_sz) << 3) * (train_len-1) / delta;
        print_bw(pathrate_fp, bw_msr);
        if (Verbose) print_bw(stdout, bw_msr);
        if (delta > min_possible_delta * (train_len-1)) { 
          /* Acceptable measurement */
           measurs_P1[trains_msrd++] = bw_msr;
        }
        trains_rcvd++;
      }
    }while(trains_rcvd<no_trains);
    sprintf(message,"\n");
    prntmsg(pathrate_fp);
    if (Verbose) prntmsg(stdout);
  
    if (!retransmit) {
      train_len++;
      /* Tell sender to continue with next phase */
      //ctr_code = CONTINUE;
      //send_ctr_msg(ctr_code);
    }
  } while ( train_len <= MIN_V(max_train_len,10));
  
  /* Actual number of trains measured */
  trains_msrd--;

  /*
    ------------------------------------------------------------------------
    Estimate bandwidth resolution (bin width) and check for "quick estimate" 
    ------------------------------------------------------------------------
  */
  /* Calculate average and standard deviation of last measurements,
     ignoring the five largest and the five smallest values  */
  order(measurs_P1,ord_measurs_P1,trains_msrd);
  enough_data=1;
  if (trains_msrd > 30) {
    avg_bw = get_avg(ord_measurs_P1+5, trains_msrd-10);
    std_dev = get_std(ord_measurs_P1+5, trains_msrd-10);
    outlier_lim=(int) (trains_msrd/10);
    bw_range=ord_measurs_P1[trains_msrd-outlier_lim-1]-ord_measurs_P1[outlier_lim];
  }
  else  {
    avg_bw = get_avg(ord_measurs_P1, trains_msrd);
    std_dev = get_std(ord_measurs_P1, trains_msrd);
    bw_range=ord_measurs_P1[trains_msrd-1]-ord_measurs_P1[0];
  }  
  
  
  /* Estimate bin width based on previous measurements */
  /* Weiling: bin_wd is increased to be twice of earlier value */

  if (bw_range < 1.0)  /* less than 1Mbps */ 
    bin_wd = bw_range*0.25; 
  else
    bin_wd = bw_range*0.125; 

  if (bin_wd == 0) bin_wd = 20.0;
  sprintf(message,"\n\t--> Capacity Resolution: ");
  prntmsg(pathrate_fp);
  if (verbose) prntmsg(stdout);
  print_bw(pathrate_fp, bin_wd); 
  if (verbose) print_bw(stdout, bin_wd);
  sprintf(message,"\n");
  prntmsg(pathrate_fp); 
  if (verbose) prntmsg(stdout);
  /* Check for quick estimate (when measurements are not very spread) */
  if( (std_dev/avg_bw<COEF_VAR_THR) || quick_term) {
    if (quick_term)
      sprintf(message," - Requested Quick Termination");
    else
      sprintf(message,"`Quick Termination' - Sufficiently low measurement noise");
    prntmsg(pathrate_fp);
    if(verbose) prntmsg(stdout);
    sprintf(message,"\n\n--> Coefficient of variation: %.3f \n", std_dev/avg_bw);
    prntmsg(pathrate_fp); 
    if(verbose) prntmsg(stdout);
    happy_end(avg_bw-bin_wd/2., avg_bw+bin_wd/2.);
    termint(0);
  }
  
  /* Tell sender to continue with next phase */
  //ctr_code = CONTINUE;
  //send_ctr_msg(ctr_code);
  
  /*
    ----------------------------------------------------------
    Phase I: Discover local modes with packet pair experiments
    ----------------------------------------------------------
  */
  sprintf(message,"\n\n-- Phase I: Detect possible capacity modes -- \n\n");
  prntmsg(pathrate_fp);
  if (verbose) prntmsg(stdout);
  sleep(1);
  
  /* Phase I uses packet pairs, i.e., 2 packets (the default; it may be larger) */
  train_len = train_len_P1;
  if (train_len > max_train_len) train_len = max_train_len;
  
  /* Compute number of different packet sizes  in Phase I */
  pack_incr_step = (max_pack_sz - min_pack_sz_P1) / no_pack_sizes + 1;
  pack_sz=min_pack_sz_P1;
  
  /* Compute number of trains  per packet size */
  no_trains_per_size = NO_TRAINS_P1 / no_pack_sizes;
  
  /* Number of trains in Phase I */
  no_trains = NO_TRAINS_P1;
  
  /* Send train spacing to sender */
  train_spacing = min_train_spacing;
  ctr_code = TRAIN_SPACING | (train_spacing<<8);
  send_ctr_msg(ctr_code);
  
  trains_msrd=0;
  retransmit = 0;
  /* Compute new packet size (and repeat for several packet sizes) */
  for (i=0; i<no_pack_sizes; i++){
     if (!retransmit) {
       /* Send packet size to sender */
       ctr_code = PCK_LEN | (pack_sz<<8);
       send_ctr_msg(ctr_code);
    
       /* Send train length to sender */
       ctr_code = TRAIN_LEN | (train_len<<8);
       send_ctr_msg(ctr_code);
    
       sprintf(message,"\t-> Train length: %2d - Packet size: %4dB -> %2d%% completed\n",
           train_len, pack_sz+28, i*100/no_pack_sizes);
       prntmsg(pathrate_fp);
       if (verbose) prntmsg(stdout);
       bad_tstamps = 0; 
       trains_per_size = 0; 
     }

     do {
        /* Wait for a complete packet train */
        if ( time_stmps != NULL ) free(time_stmps);
        bad_train = recv_train(train_len, &round, &time_stmps);
  
          /* Compute dispersion and bandwidth measurement */
        if (!bad_train) {
          delta = time_to_us_delta(time_stmps[1], time_stmps[train_len]);
          bw_msr = ((28+pack_sz) << 3) * (train_len-1) / delta;
          sprintf(message,"\tMeasurement-%d:", trains_per_size+1);
          prntmsg(pathrate_fp);
          if (Verbose) prntmsg(stdout);
          print_bw(pathrate_fp, bw_msr);
          if (Verbose) print_bw(stdout, bw_msr);
          sprintf(message," (%.0f usec)",delta);
          prntmsg(pathrate_fp); 
          if (Verbose) prntmsg(stdout);
  
          /* Acceptable measurement */
          if (delta > min_possible_delta*(train_len-1)) {
             measurs_P1[trains_msrd++] = bw_msr;
          }
          else {
            bad_tstamps++;
            sprintf(message," (ignored)");
            prntmsg(pathrate_fp);
            if (Verbose) prntmsg(stdout);
          }  
  
          /* # of trains received in this packet size iteration */
          trains_per_size++;
          sprintf(message,"\n");
          prntmsg(pathrate_fp);
          if (Verbose) prntmsg(stdout);
       }
     } while(trains_per_size < no_trains_per_size);
     if(Verbose) fprintf(stdout,"\n");

     /* Tell sender to continue with next phase */
     pack_sz+=pack_incr_step;
     if (pack_sz > max_pack_sz) pack_sz = max_pack_sz;
     /***** dealing with ignored measurements *****/
     if (bad_tstamps > no_trains_per_size/IGNORE_LIM_FACTOR){
       /* If this is greater than max_train_len? :Ravi */
       train_len+=1;
       if (train_len > MAX_V(max_train_len/4,2)) {
          abort_phase1=1;  
       }
       pack_sz+=275;