/*
cc  -o forkthread-rel4.5 forkthread-rel4.5.c -xc99 -xarch=v9 -lpthread -lm -D_GNU_SOURCE
gcc -o forkthread-rel4.5 forkthread-rel4.5.c -Wall -pedantic -O3 -std=gnu99 -lpthread -lm -D_GNU_SOURCE
icc -o forkthread-rel4.5 forkthread-rel4.5.c -c99 -O3 -ip -restrict -w1 -lpthread -lm -D_GNU_SOURCE 

*/

#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>
#include <time.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <sys/resource.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <math.h>

#define VERSION         "4.5"
#define ALIGN_SIZE    	512uL
#define PAGE_SIZE     	16384uL
#define MIN_PROCESSES 	1uL
#define LG_MAX_PROCS    12uL
#define MAX_PROCESSES   (1uL<<LG_MAX_PROCS)
#define SEED_SHIFT      (32uL-LG_MAX_PROCS)
#define PAGE_READ       0uL
#define PAGE_NOP        1uL
#define IO_SYS_ERR     -126L
//#define TiBYTES         ((double)(1uL<<40))  
#define TiBYTES         1099511627776.0

#define UI64(c)    (c##uL)
#define _ZERO64    UI64(0x0)
#define _maskr(x)  (((x) == 0) ? _ZERO64   : ((~_ZERO64) >> (64-(x))))
#define _maskl(x)  (((x) == 0) ? _ZERO64   : ((~_ZERO64) << (64-(x))))
#define MAX(X,Y)   ((X)>(Y)?(X):(Y))

typedef unsigned long uint64; 
typedef signed long    int64;

typedef struct
{
  off_t  offset;        // offset into file to begin read       
  void  *addr;          // addr of write buffer                 
  int    opcode;        // flag allows to ignore request        
  int    retval;        // return value 0 or errno              
} read_desc_t;

read_desc_t    *read_desc; // stack of reads to do 
int             gFD;       // global file descriptor
int             gVerify;   // global flag to verify data from reads
int             gThreads;  // number of threads per process
int             gReads;    // number o reads per worker

// we treat the reads as a stack
// threads pop off requests, using the mutex for safety
static pthread_mutex_t   mutex = PTHREAD_MUTEX_INITIALIZER;
static int64             stack_idx;

double wall              (void)

{
  struct timeval tp;

  gettimeofday (&tp, NULL);
  return tp.tv_sec + tp.tv_usec/(double)1.0e6;
}

void   usage             ( char *p )

{

  printf ("Usage:\t%s -p <num processes> -t <threads/process> -f <file>\n"
	  "\t-r <reads/thread or process> -s <seed> -o <O_DIRECT - on(1)/off(0)> [-v]\n",
	  p);
  printf ("Forks <n> processes, creates <t> threads per process,\n"
	  "and each worker does <r> random, 16KB reads from <file>, \n"
	  "where <seed> seeds srandom() to create random 16KB-aligned offsets.\n");
  printf( "The -v option turns on verification to ensure the data read is the\n"
	  "data expected.  Verification requires that <file> be created with\n"
	  "the mkfile utility.\n");
  printf("Version: %s built on %s\n", VERSION, __DATE__  ); 
  printf("\n");
  exit(1);
}

void   seed_rng(uint64 seed) 
{ 
  
  srand48( seed );
  //printf("[%5ld]  Seed %016lx\n", (long)getpid(), seed );
}
static inline long rng(void) 
{ 
  uint64 myrand;

  myrand  = lrand48() ^ (lrand48() << 31); 
  myrand *= 0x33c83d5da4629d47uL;
  myrand >>= 16;
  //printf("[%5ld]  Rand %016lx\n", (long)getpid(), myrand );
  return myrand;
}

void * read_entry        (void *arg)

{
  read_desc_t *rd;
  int64        ret, read_idx;
  char         buf[PAGE_SIZE+ALIGN_SIZE]; 

  /* aligned memory; all reads for each thread goto same place        */
  /* ... trying to save memory by using global fd and stnd PAGE_SIZE  */
  char    *pbuf = (char *)(((uint64)((char *)buf) + 511uL) & _maskl(64-9));

  while (1) {

    pthread_mutex_lock   (&mutex);
    read_idx = stack_idx;
    if (stack_idx >= 0)  stack_idx--;
    pthread_mutex_unlock (&mutex);

    if (read_idx < 0)  break;  // all reads in progress so break

    rd = &read_desc[read_idx];
    // ignore requests with opcodes set to PAGE_NOP 
    if   (rd->opcode == PAGE_NOP) {
      rd->retval = 0;  // nothing to do
    }
    else {

      // do the seek and read w/o updating file pointer
      ret = pread ( gFD, pbuf, PAGE_SIZE, rd->offset);

      if  (ret == PAGE_SIZE)  { 
	rd->retval = 0;

	// if -v (verify) on, then word read will be the value of the word's position in file 0-based
	// if file is created with mkfile utility
	if( gVerify ) { 
	  if( (rd->offset / 8 ) !=  *((uint64 *)pbuf) ) {
	    
	    printf("Warning: Seek to offset %016lX did not read expected data.\n", 
		   rd->offset);
	    printf("Warning: Expected 0x%016lx\tGot 0x%016lx\n", 
		   (rd->offset / 8), *((uint64 *)pbuf ));
	    exit(-1);
	  }
	}
      }
      else {
	rd->retval = IO_SYS_ERR;
	fprintf( stderr, "Error. PID %ld - pread() returned %ld.  Offset: 0x%016lx\n", 
		 (long)getpid(), ret, rd->offset );
	exit(-1);
      }
    }

  }  // END: while reads remaining

  return NULL;
}

void   work              ( int nthreads, int nreads, uint64 file_pages, 
			   uint64 seed )

{
  int             i, spawn_threads;
  uint64          offset_cksum = 0;
  pthread_attr_t  attr;
  pthread_t      *pthreads;
  
  /* seed per process */ 
  seed_rng( seed );

  pthread_attr_init( &attr ); 
  
  // used by pthread_create()   
  pthreads   = calloc(nthreads, sizeof(pthread_t));
  if( !pthreads ){
    fprintf( stderr, "calloc() for pthreads failed.\n");
    exit(1);
  }
  
  // make memory for stack of reads
  read_desc = calloc(MAX(nthreads,1)*nreads, sizeof(read_desc_t));
  if( !read_desc ){
    fprintf(stderr, "calloc() for pthreads failed.\n");
    exit(1);
  }
  
  for( i=0; i<MAX(nthreads,1)*nreads; i++ ) {
    read_desc[i].offset  = (rng() % file_pages ) * PAGE_SIZE; 
    read_desc[i].opcode  = PAGE_READ; 
    offset_cksum        ^= read_desc[i].offset;
  }

  stack_idx = ((MAX(nthreads,1)) * nreads) - 1;
  
  spawn_threads = nthreads - 1;
  
  for( i=0; i<spawn_threads; i++ )  {
    int retval;
    
    retval = pthread_create( &pthreads[i], &attr, &read_entry, NULL );
    if (retval) {
      fprintf (stderr, "ERROR-- pthread_create failed: %s\n", strerror(retval));
      exit(1);
    }
  } 
  
  // parent process does the last set of requests
  read_entry( NULL ); 

  // wait for our threads to complete
  for( i=0; i<spawn_threads; i++ ) {
    pthread_join( pthreads[i], NULL ); 
  }

  free( read_desc ); 

  offset_cksum /= PAGE_SIZE;
  printf("[%5ld] Process Checksum: 0x%08lx\n", (long)getpid(), offset_cksum );
  return;
}

int    main              ( int argc, char *argv[] )

{

  char    *file = NULL;
  int     status, workers, ttl_rds, retval;
  long    i;
  long    processes   = 1L;
  long    reads       = 64L;
  long    threads     = 1L;
  uint64  o_direct    = 1uL; 
  uint64  file_pages, seed;
  pid_t   child = 0, children[MAX_PROCESSES]; 
  double  elapsed, iops, score; 
  double  ttl_utime, ttl_stime;
  double  user_usage, sys_usage;
  struct  stat file_info; 
  struct  rusage parent_usage;
  struct  rusage child_usage;

  if( argc == 1 )  {
    usage( argv[0] ); 
  }

  // default seed
  seed     = getpid();
  seed   <<= 16uL;
  seed    ^= time(NULL) & _maskr(SEED_SHIFT);
  gVerify  = 0uL;
  gFD      = 0uL;
  ttl_utime = ttl_stime =  0.0;
  user_usage = sys_usage = 0.0;

  for(i=1; i<argc; i++ ){
    if( strncmp( argv[i], "-p", 2 ) == 0 ){
      processes = atoi( argv[++i] ); 
    }
    else if ( strncmp( argv[i], "-t", 2 ) == 0 ){
      threads = atoi( argv[++i] ); 
    } 
    else if ( strncmp( argv[i], "-r", 2 ) == 0 ){
      reads = atoi( argv[++i] ); 
    }
    else if( strncmp( argv[i], "-s", 2 ) == 0 ){
      seed = atol( argv[++i] );
    }
    else if ( strncmp( argv[i], "-f", 2 ) == 0 ){
      file = argv[++i]; 
    } 
    else if ( strncmp( argv[i], "-v", 2 ) == 0 ){
      gVerify++;
    } 
    else if( strncmp( argv[i], "-o", 2 ) == 0 ) {
      if( argv[++i][0] == '0' ){
	o_direct = 0;
      }
    } 
    else {
      usage( argv[0] ); 
    }
  }

  gReads = reads;
  gThreads = threads;

  printf("Version: %s built on %s\n", VERSION, __DATE__  ); 
      
  if( (processes<MIN_PROCESSES) || (processes>MAX_PROCESSES) ){
    fprintf(stderr, "Must use [%ld,%ld] processes.\n", MIN_PROCESSES, MAX_PROCESSES);
    exit( -1 ); 
  }

  if( !file ) {
    fprintf(stderr, "Please supply a filename.\n");
    exit(1); 
  }

  seed &= _maskr(32uL);
  printf("Using seed: %ld (0x%08lx).\n", seed, seed );

  // open file 
#ifdef __sun
  gFD = open( file, O_RDONLY ); 
#else
  if( o_direct ) {
    gFD = open( file, O_RDONLY | O_DIRECT ); 
    printf("Using O_DIRECT.\n");
  } 
  else {
    gFD = open( file, O_RDONLY  ); 
    printf("Not using O_DIRECT.\n");
  }
#endif
  if( gFD <= 0 ) {
    fprintf( stderr, "Failed to open %s.\n", file );
    exit(1); 
  }

  //if Solaris and O_DIRECT, then call directio() 
#ifdef __sun
  if( o_direct ) {
    retval = directio( gFD, DIRECTIO_ON );
    if( retval == -1 ) {
      fprintf(stderr, "Failed to set DIRECTIO_ON.\n");
      exit( 1 );
    } 
    printf("Setting DIRECTIO_ON on Solaris.\n");
  } 
  else {
    printf("Using DIRECTIO_OFF on Solaris.\n");
  }
#endif

  if( gVerify ) {
    printf("Verify mode on.\n");
  }
    
  if( fstat( gFD, &file_info ) == -1 ) {
    fprintf( stderr, "Could not stat %s. %s\n", file, strerror( errno ) );
    exit(1); 
  }
  file_pages = file_info.st_size / PAGE_SIZE;
  
  printf( "Filesize:       %ld bytes\n", file_info.st_size );
  printf( "Filesize:   	%0.4f x 2^40 bytes\n",  file_info.st_size / TiBYTES );
  printf( "Pages:      	%ld\n", file_pages );

  printf( "Creating %ld processes each w/ %ld threads doing %ld "
	   "reads/worker from %s.\n", 
	   processes, threads, reads, file ); 
  
  elapsed = wall(); 
  
  for(i=0; i<processes-1; i++ ) {
    fflush( stdout );
    child = fork();

    // child 
    if( child == 0 ) {
      work( threads, reads, file_pages, seed ^ (i<<SEED_SHIFT) ); 
      close( gFD ); 
      exit(0); 
    } // end if child

    // parent 
    if( child > 0 ) {
      children[i] = child; 
    }

    // error 
    if( child == -1 ){
      fprintf( stderr, "Fork() failed. %s.\n", strerror(errno) );
      exit( -1 ); 
    } 

  } // end for processes

  // parent process does one iteration
  work( threads, reads, file_pages, seed ^ (i<<SEED_SHIFT) ); 

  close( gFD ); 

  // wait for child processes
  for(i=0; i<processes-1; i++ ){

    if( waitpid( children[i], &status, 0 ) == -1 ) {
      fprintf( stderr, "Waitpid() error on PID %ld. %s\n", (long)children[i], strerror(errno));
      exit( -1 );
    }
    
    if( WIFEXITED( status )){
      if( WEXITSTATUS( status )){
	fprintf( stderr, "Error. PID %ld had non-zero exit status of %d.\n", 
		 (long)children[i], WEXITSTATUS( status ));       
	exit( -1 );
      }
    }
    else {
      fprintf( stderr, "Error. PID %ld exited abnormally.\n", (long)children[i] );
      exit( -1 );
    } // END: if WIFEXITED
  } //END: for i<processes-1

  elapsed = wall() - elapsed; 

  retval = getrusage( RUSAGE_SELF, &parent_usage );
  if( retval == -1 ) {
    fprintf( stderr, "getrusage() for parent failed. %s\n", strerror( errno ));
    exit( 1 );
  }

  if( processes > 1 ) {
    retval = getrusage( RUSAGE_CHILDREN, &child_usage );
    if( retval == -1 ) {
      fprintf(stderr, "getrusage() for children failed. %s\n", strerror( errno ));
      exit( 1 );
    }

    ttl_utime = parent_usage.ru_utime.tv_sec + (parent_usage.ru_utime.tv_usec / (double)1.0e6) +
      child_usage.ru_utime.tv_sec  + (child_usage.ru_utime.tv_usec / (double) 1.0e6 );

    ttl_stime = parent_usage.ru_stime.tv_sec + (parent_usage.ru_stime.tv_usec / (double)1.0e6) +
      child_usage.ru_stime.tv_sec  + (child_usage.ru_stime.tv_usec / (double)1.0e6 );
  
  } 
  else {
    ttl_utime =   parent_usage.ru_utime.tv_sec + (parent_usage.ru_utime.tv_usec / (double)1.0e6);
    ttl_stime =   parent_usage.ru_stime.tv_sec + (parent_usage.ru_stime.tv_usec / (double)1.0e6);
  } // END: if process > 1   

  if (elapsed > 0.) {
    user_usage = ttl_utime / elapsed;
    sys_usage  = ttl_stime / elapsed;
  } 

  if( threads ) {
    workers = processes * threads;
  } 
  else {
    workers = processes; 
  }

  ttl_rds = reads * workers; 

  if( elapsed ) {
    iops    = ttl_rds / elapsed; 
  } 
  else {
    iops    = 0;
  }

  if( iops ) {
    score   = sqrt(( file_info.st_size / TiBYTES ) * iops );
  } 
  else {
    score = 0.0;
  }

  printf( "Elapsed:    	%.3f\n", elapsed ); 
  printf( "TtlReads:   	%d\n", ttl_rds ); 
  printf( "Reads/Sec:  	%.2f\n", iops );
  printf( "Score:      	%.3f\n", score );
  printf( "User Time:  	%-9.3f (%7.3f%% of one core)\n", ttl_utime, user_usage * 100 );
  printf( "Sys Time:   	%-9.3f (%7.3f%% of one core)\n", ttl_stime, sys_usage  * 100 );
  return 0; 
}