ptrn_Nto1.cpp

/*
 * Copyright (c) 2009 The Trustees of Indiana University and Indiana
 *                    University Research and Technology
 *                    Corporation.  All rights reserved.
 *
 * Author(s): Torsten Hoefler <htor@cs.indiana.edu>
 *
 */

#include "netgauge.h"
#if defined NG_PTRN_NTO1
#include "hrtimer/hrtimer.h"
#include <vector>
#include "statistics.h"
#include <string.h>
#include <algorithm>
#include "ng_tools.hpp"

/* internal function prototypes & extern stuff */
static void Nto1_do_benchmarks(struct ng_module *module);
extern struct ng_options g_options;

/**
 * one-to-many communication pattern registration
 * structure
 */
static struct ng_comm_pattern pattern_Nto1 = {
   pattern_Nto1.name = "Nto1",
   pattern_Nto1.desc = "measures bandwith in a N:1 congestion situation",
   pattern_Nto1.flags = NG_PTRN_NB,
   pattern_Nto1.do_benchmarks = Nto1_do_benchmarks
};

/**
 * registers this comm. pattern with the main prog.,
 * but only if we use MPI
 */
int register_pattern_Nto1() {
  /* there's no use for this mode without MPI */
#if NG_MPI
  ng_register_pattern(&pattern_Nto1);
#endif 
  return 0;
}

void Nto1_do_benchmarks(struct ng_module *module) {
#if NG_MPI
  /** current data size for transmission tests */
  long data_size;

  /** # times to run with current data_size */
  long test_count = g_options.testcount;
  
  /** send/receive buffer */
  void *buffer;

  /** enumerating things is such a fun! */
  int i;

  /* server role - receive from clients */
  NG_Request *reqs = NULL;

  /* measurement data ... we need to send it from rank 1 to rank 0 since
   * rank 0 doe all the I/O */
  double results[2];
    
  /* allocate send / recv buffer */
  ng_info(NG_VLEV2, "Allocating %d bytes data buffer", g_options.max_datasize + module->headerlen);
  NG_MALLOC(module, void*, g_options.max_datasize + module->headerlen, buffer);

  int rank = g_options.mpi_opts->worldrank;

  /* initialize the statistics stuff - rank 1 does the whole measurement */
  if(rank == 0) {
    /* allocate Request structure at server */
    reqs = (NG_Request*)malloc(sizeof(NG_Request) * (g_options.mpi_opts->worldsize-1) );
  }

  /* buffer for header ... */
  char* txtbuf = (char *)malloc(2048 * sizeof(char));
  if (txtbuf == NULL) {
    ng_error("Could not (re)allocate 2048 byte for output buffer");
    ng_exit(10);
  }
  memset(txtbuf, '\0', 2048);

  /* header printing */
  if(rank == 0) {
    // if very verbose - long output
    if (NG_VLEV2 & g_options.verbose) {
      snprintf(txtbuf, 2047,
	      "## Netgauge v%s - mode %s - 2 processes\n"
	      "##\n"
        "## A...packet size [byte]\n"
        "##\n"
        "## B...minimum send blocking time\n"
        "## C...average send blocking time\n"
        "## D...median send blocking time\n"
        "## E...maximum send blocking time\n"
        "## F...standard deviation for send blocking time\n"
        "## G...number of send blocking time values, that were bigger than 2*avg.\n"
        "##\n"
        "## H...minimum RTT/2\n"
        "## I...average RTT/2\n"
        "## J...median RTT/2\n"
        "## K...maximum RTT/2\n"
        "## L...standard deviation for RTT/2\n"
        "## M...number of RTT/2 values, that were bigger than 2*avg.\n"
        "##\n"
        "## N...minimum throughput [Mbit/sec]\n"
        "## O...average throughput [Mbit/sec]\n"
        "## P...median throughput [Mbit/sec]\n"
        "## Q...maximum throughput [Mbit/sec]\n"
        "##\n"
        "## A  -  B  C  S  D  (E F) - G  H  T  I  (J K)  -  L  M  U  N\n",
        NG_VERSION,
        g_options.mode);

      printf("%s", txtbuf);
    } else
    // if verbose - short output
    if (NG_VLEV1 & g_options.verbose) {
      snprintf(txtbuf, 2047,
	      "## Netgauge v%s - mode %s - 2 processes\n"
	      "##\n"
        "## A...packet size [byte]\n"
        "##\n"
        "## B...minimum send blocking time\n"
        "## C...average send blocking time\n"
        "## D...median send blocking time\n"
        "## E...maximum send blocking time\n"
        "##\n"
        "## H...minimum RTT/2\n"
        "## I...average RTT/2\n"
        "## J...median RTT/2\n"
        "## K...maximum RTT/2\n"
        "##\n"
        "## N...minimum throughput [Mbit/sec]\n"
        "## O...average throughput [Mbit/sec]\n"
        "## P...median throughput [Mbit/sec]\n"
        "## Q...maximum throughput [Mbit/sec]\n"
        "##\n"
        "## A  -  B  C  S  D - G  H  T  I - L  M  U  N\n",
        NG_VERSION,
        g_options.mode);

      printf("%s", txtbuf);
      
    } else
    // if not verbose - short output
    {
      // no header ...
    }
  }

  /* outer test loop, sets data_size and test_count */
  for (data_size = NG_START_PACKET_SIZE; !g_stop_tests && data_size > 0;
       get_next_testparams(&data_size, &test_count, &g_options, module)) {
    if(data_size == -1) goto shutdown;

    /* TODO: introduce internal dissemination barrier */
    ng_info(NG_VLEV1, "Barrier and testing %d times with %d bytes:", test_count, data_size);
    // if we print dots ...
    if ( (rank==0) && (NG_VLEV1 & g_options.verbose) ) {
      printf("# ");
    }
    
    // the benchmark results
    std::vector<double> tblock, trtt;
    
    for (long test=-1 /* 1 warmup */; (test < test_count) && !g_stop_tests; test++) {
      /* TODO: introduce internal dissemination barrier */
      MPI_Barrier(MPI_COMM_WORLD);

	    if ( rank == 0 && (NG_VLEV1 & g_options.verbose) && ( test_count < NG_DOT_COUNT || !(test % (int)(test_count / NG_DOT_COUNT)) )) {
	      printf(".");
	      fflush(stdout);
	    }

      if (g_options.mpi_opts->worldrank == 0) {
        for(i = 1; i<g_options.mpi_opts->worldsize; i++) {
          /* TODO: all receive in one buffer ... is this a problem? */
          module->irecvfrom(i, buffer, data_size, &reqs[i-1]);
        }
        /* loop until all requests done */
        while(1) {
          int ctr;
          ctr = 0;
          for(i = 1; i<g_options.mpi_opts->worldsize; i++) {
            ctr += module->test(&reqs[i-1]);
          }
          /* all returned 0 :) */
          if(ctr == 0) break;
        }
        /* send ack to rank 1 */
        module->sendto(1, buffer, 1);

        /* receive measurement data from rank 1 */
        MPI_Recv(&results, 2, MPI_DOUBLE, 1, 1, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
        /* calculate results */
        if(test >= 0) {
          trtt.push_back(results[1]/2);
          tblock.push_back(results[0]);
        }
          
      } else {
        /* client role - send to server */
        unsigned long long tibl, tirtt;
        HRT_TIMESTAMP_T t[3];
        
        if(g_options.mpi_opts->worldrank == 1) {
          /* get start time - timer is *ONLY* meaningful on rank 1 !! */
          HRT_GET_TIMESTAMP(t[0]);
        }
            
        module->sendto(0, buffer, data_size);

        /* get after-sending time */
        if(g_options.mpi_opts->worldrank == 1) HRT_GET_TIMESTAMP(t[1]);
        
        /* rank 1 receives ack from server */
        if(g_options.mpi_opts->worldrank == 1) {
          module->recvfrom(0,buffer,1);
          /* get after-receiving time */
          HRT_GET_TIMESTAMP(t[2]);

          HRT_GET_ELAPSED_TICKS(t[0],t[1],&tibl);
          HRT_GET_ELAPSED_TICKS(t[0],t[2],&tirtt);

          /* send results to rank 0 */
          results[0] = HRT_GET_USEC(tibl);
          results[1] = HRT_GET_USEC(tirtt);
          MPI_Send(&results, 2, MPI_DOUBLE, 0, 1, MPI_COMM_WORLD);
        }
      }
    }

    /* add linebreak if we made dots ... */
    if ( (NG_VLEV1 & g_options.verbose) ) {
      ng_info(NG_VLEV1, "\n");
    }

    if(rank==0) {
      /* output statistics - blocking time */
      double tblock_avg = std::accumulate(tblock.begin(), tblock.end(), (double)0)/(double)tblock.size(); 
      double tblock_min = *std::min_element(tblock.begin(), tblock.end()); 
      double tblock_max = *std::max_element(tblock.begin(), tblock.end()); 
      std::vector<double>::iterator nthblock = tblock.begin()+tblock.size()/2;
      std::nth_element(tblock.begin(), nthblock, tblock.end());
      double tblock_med = *nthblock;
      double tblock_var = standard_deviation(tblock.begin(), tblock.end(), tblock_avg);
      int tblock_fail = count_range(tblock.begin(), tblock.end(), tblock_avg-tblock_var*2, tblock_avg+tblock_var*2);

      /* output statistics - rtt time */
      double trtt_avg = std::accumulate(trtt.begin(), trtt.end(), (double)0)/(double)trtt.size(); 
      double trtt_min = *std::min_element(trtt.begin(), trtt.end()); 
      double trtt_max = *std::max_element(trtt.begin(), trtt.end()); 
      std::vector<double>::iterator nthrtt = trtt.begin()+trtt.size()/2;
      std::nth_element(trtt.begin(), nthrtt, trtt.end());
      double trtt_med = *nthrtt;
      double trtt_var = standard_deviation(trtt.begin(), trtt.end(), trtt_avg);
      int trtt_fail = count_range(trtt.begin(), trtt.end(), trtt_avg-trtt_var*2, trtt_avg+trtt_var*2);

      // if very verbose - long output
      if (NG_VLEV2 & g_options.verbose) {
        memset(txtbuf, '\0', 2048);
        snprintf(txtbuf, 2047,
          "%li - %.2lf %.2lf %.2lf %.2lf (%.2lf %i) - %.2lf %.2lf %.2lf %.2lf (%.2lf %i) - %.2lf %.2lf %.2lf %.2lf\n",
          data_size, /* packet size */
          tblock_min, /* minimum send blocking time */
          tblock_avg, /* average send blocking time */
          tblock_med, /* median send blocking time */
          tblock_max, /* maximum send blocking time */ 
          tblock_var, /* standard deviation */
          tblock_fail, /* how many are bigger than twice the standard deviation? */
          trtt_min, /* minimum RTT time */
          trtt_avg, /* average RTT time */
          trtt_med, /* median RTT time */
          trtt_max, /* maximum RTT time */
          trtt_var, /* standard deviation */
          trtt_fail, /* how many are bigger than twice the standard deviation? */
          data_size/trtt_max*8, /* minimum bandwidth */
          data_size/trtt_avg*8, /* average bandwidth */
          data_size/trtt_med*8, /* median bandwidth */
          data_size/trtt_min*8 /* maximum bandwidth */
          );
        printf("%s", txtbuf);
        
      } else
      // if verbose - short output
      if (NG_VLEV1 & g_options.verbose) {
        memset(txtbuf, '\0', 2048);
        snprintf(txtbuf, 2047,
          "%li - %.2lf %.2lf %.2lf %.2lf - %.2lf %.2lf %.2lf %.2lf - %.2lf %.2lf %.2lf %.2lf\n",
          data_size, /* packet size */
          tblock_min, /* minimum send blocking time */
          tblock_avg, /* average send blocking time */
          tblock_med, /* median send blocking time */
          tblock_max, /* maximum send blocking time */ 
          trtt_min, /* minimum RTT time */
          trtt_avg, /* average RTT time */
          trtt_med, /* median RTT time */
          trtt_max, /* maximum RTT time */
          data_size/trtt_max*8, /* minimum bandwidth */
          data_size/trtt_avg*8, /* average bandwidth */
          data_size/trtt_med*8, /* median bandwidth */
          data_size/trtt_min*8 /* maximum bandwidth */
          );
        printf("%s", txtbuf);
        
      } else
      // if not verbose - short output
      {
        memset(txtbuf, '\0', 2048);
        snprintf(txtbuf, 2047,
          "%li bytes \t -> %.2lf us \t == %.2lf Mbit/s\n",
          data_size, /* packet size */
          trtt_med, /* median RTT time */
          data_size/trtt_med*8 /* median bandwidth */
          );
        printf("%s", txtbuf);
      }
    }
    
  } /* outer test loop */

 shutdown:
  /* clean up */
  if (buffer) free(buffer);
#endif
}

#else 
extern "C" {
int register_pattern_Nto1(void) {return 0;}
}
#endif