seanhalle@0: /* 
seanhalle@0:  * 
seanhalle@0:  */
seanhalle@0: 
seanhalle@0: #include "main.h"
seanhalle@0: 
seanhalle@0: //==========  Global Vars  ===========
seanhalle@0: 
seanhalle@0: const char *usage = 
seanhalle@0:  {
seanhalle@0:    "Usage: k_tuple_async [options]\n"
seanhalle@0:    "  Creates a number of workers, and one consumer that packages productions "
seanhalle@0:    "   into a tuple.\n\n"
seanhalle@0:    "Options:\n"
seanhalle@0:    "  -p <num>   The number of producer threads to create.\n"
seanhalle@0:    "  -t <num>   the number of tuples to create\n"
seanhalle@0:    "  -h         this help screen\n\n"
seanhalle@0:  };
seanhalle@0: 
seanhalle@0: char __ProgrammName[] = "K-tuple_async";
seanhalle@0: char __DataSet[255];
seanhalle@0: 
seanhalle@0: #ifdef MEASURE_PERF
seanhalle@0: int cycles_counter_fd[NUM_CORES];
seanhalle@0: int instrs_counter_fd[NUM_CORES];
seanhalle@0: int cycles_counter_main_fd;
seanhalle@0: #endif
seanhalle@0: 
seanhalle@0: pthread_mutex_t  waitForAllDoneLock;
seanhalle@0: pthread_cond_t   waitForAllDoneCond;   
seanhalle@0: 
seanhalle@0: 
seanhalle@0: //===================================
seanhalle@0: /* provide a millisecond-resolution timer for each system */
seanhalle@0: #if defined(unix) || defined(__unix__)
seanhalle@0: #include <time.h>
seanhalle@0: #include <sys/time.h>
seanhalle@0: unsigned long get_msec(void) {
seanhalle@0: 	static struct timeval timeval, first_timeval;
seanhalle@0: 	
seanhalle@0: 	gettimeofday(&timeval, 0);
seanhalle@0: 	if(first_timeval.tv_sec == 0) {
seanhalle@0: 		first_timeval = timeval;
seanhalle@0: 		return 0;
seanhalle@0: 	}
seanhalle@0: 	return (timeval.tv_sec - first_timeval.tv_sec) * 1000 + (timeval.tv_usec - first_timeval.tv_usec) / 1000;
seanhalle@0: }
seanhalle@0: #elif defined(__WIN32__) || defined(WIN32)
seanhalle@0: #include <windows.h>
seanhalle@0: unsigned long get_msec(void) {
seanhalle@0: 	return GetTickCount();
seanhalle@0: }
seanhalle@0: #else
seanhalle@0: #error "I don't know how to measure time on your platform"
seanhalle@0: #endif
seanhalle@0: 
seanhalle@0: /*Initializes the performance counters, and opens the file descriptors used
seanhalle@0:  * to read from the performance counters
seanhalle@0:  */
seanhalle@0: void
seanhalle@0: set_up_performance_counters()
seanhalle@0:  { int i;
seanhalle@0:  
seanhalle@0:  #ifdef MEASURE_PERF
seanhalle@0:     //setup performance counters
seanhalle@0:     struct perf_event_attr hw_event;
seanhalle@0:     memset(&hw_event,0,sizeof(hw_event));
seanhalle@0:         hw_event.type = PERF_TYPE_HARDWARE;
seanhalle@0:         hw_event.size = sizeof(hw_event);
seanhalle@0:         hw_event.disabled = 0;
seanhalle@0:         hw_event.freq = 0;
seanhalle@0:         hw_event.inherit = 1; /* children inherit it   */
seanhalle@0:         hw_event.pinned = 1; /* must always be on PMU */
seanhalle@0:         hw_event.exclusive = 0; /* only group on PMU     */
seanhalle@0:         hw_event.exclude_user = 0; /* don't count user      */
seanhalle@0:         hw_event.exclude_kernel = 1; /* ditto kernel          */
seanhalle@0:         hw_event.exclude_hv = 1; /* ditto hypervisor      */
seanhalle@0:         hw_event.exclude_idle = 1; /* don't count when idle */
seanhalle@0:         hw_event.mmap = 0; /* include mmap data     */
seanhalle@0:         hw_event.comm = 0; /* include comm data     */
seanhalle@0: 
seanhalle@0: 
seanhalle@0:     for( i = 0; i < NUM_CORES; i++ )
seanhalle@0:     {
seanhalle@0:         hw_event.config = PERF_COUNT_HW_CPU_CYCLES; //cycles
seanhalle@0:         cycles_counter_fd[i] = syscall(__NR_perf_event_open, &hw_event,
seanhalle@0:                 0,//pid_t pid, 
seanhalle@0:                 i,//int cpu, 
seanhalle@0:                 -1,//int group_fd,
seanhalle@0:                 0//unsigned long flags
seanhalle@0:         );
seanhalle@0:         if (cycles_counter_fd[i]<0){
seanhalle@0:             fprintf(stderr,"On core %d: ",i);
seanhalle@0:             perror("Failed to open cycles counter");
seanhalle@0:         }
seanhalle@0:     }        
seanhalle@0:        
seanhalle@0:     int cycles_counter_main_fd;
seanhalle@0:     hw_event.config = PERF_COUNT_HW_CPU_CYCLES; //cycles
seanhalle@0:     hw_event.exclude_kernel=0;
seanhalle@0:     cycles_counter_main_fd = syscall(__NR_perf_event_open, &hw_event,
seanhalle@0:             0,//pid_t pid, 
seanhalle@0:             -1,//int cpu, 
seanhalle@0:             -1,//int group_fd,
seanhalle@0:             0//unsigned long flags
seanhalle@0:     );
seanhalle@0:     if (cycles_counter_main_fd<0){
seanhalle@0:         perror("Failed to open main cycles counter");
seanhalle@0:     }
seanhalle@0:     
seanhalle@0:  #endif
seanhalle@0:  }
seanhalle@0: 
seanhalle@0: 
seanhalle@0: void
seanhalle@0: init_stuff()
seanhalle@0:  { 
seanhalle@0:    pthread_mutex_init(&tupleIterLock, NULL);
seanhalle@0:    pthread_cond_init( &tupleIterCond, NULL );
seanhalle@0:    tupleIter = 0;
seanhalle@0: 
seanhalle@0:    pthread_mutex_init(&producerAccessMutex, NULL);
seanhalle@0:    pthread_mutex_init(&productionReadyLock, NULL);
seanhalle@0:    pthread_cond_init( &productionReadyCond, NULL );
seanhalle@0:    currProductionNum = 0;
seanhalle@0: 
seanhalle@0:    pthread_mutex_init(&consumerReceivedAckLock, NULL);
seanhalle@0:    pthread_cond_init( &consumerReceivedAckCond, NULL );
seanhalle@0:    currConsumerReceivedACKNum = 0;
seanhalle@0:  }
seanhalle@0: 
seanhalle@0: 
seanhalle@0: typedef struct
seanhalle@0:  {
seanhalle@0:    int numProducers;
seanhalle@0:    int numTuplesToCreate;   
seanhalle@0:  }
seanhalle@0: ParsedArgs;
seanhalle@0: 
seanhalle@0: /*The benchmark Fn creates the producers and the consumer, then gives the
seanhalle@0:  * "go" signal.  It measures time from go until the consumer produces the
seanhalle@0:  * last tuple as output.
seanhalle@0:  */
seanhalle@0: void 
seanhalle@0: benchmark( ParsedArgs *args )
seanhalle@0:  {
seanhalle@0:    int i;
seanhalle@0:    ProducerParams producerParams[args->numProducers];
seanhalle@0:    pthread_t producerThds[args->numProducers];
seanhalle@0:    pthread_t consumerThd;
seanhalle@0:    
seanhalle@0:    ConsumerParams consumerParams;
seanhalle@0:    
seanhalle@0:    //Set up the param structs for producers.. gives them the mutex and cond var
seanhalle@0:    // to communicate with consumer
seanhalle@0:    //Also the core the producer should pin its thread to
seanhalle@0:    for(i=0; i < args->numProducers; i++)
seanhalle@0:     {
seanhalle@0:       producerParams[i].producerID = i + 1; //no ID of 0, a fact used in handshake
seanhalle@0:       producerParams[i].numTuplesToCreate = args->numTuplesToCreate;
seanhalle@0:       producerParams[i].coreID = i % NUM_CORES;
seanhalle@0:     }
seanhalle@0: 
seanhalle@0:    consumerParams.numProducers = args->numProducers;
seanhalle@0:    consumerParams.numTuplesToCreate = args->numTuplesToCreate;
seanhalle@0:            
seanhalle@0:    //take measurement before creation of threads, to get total exetime
seanhalle@0:    MeasStruct benchStartMeas, benchEndMeas;
seanhalle@0:        
seanhalle@0:    takeAMeas(0, benchStartMeas);
seanhalle@0:    
seanhalle@0:    for(i=0; i < args->numProducers; i++) 
seanhalle@0:     { pthread_create( &producerThds[i], NULL, &producer_birthFn, (void*)&producerParams[i]);  
seanhalle@0:     }
seanhalle@0:    
seanhalle@0:    pthread_create( &consumerThd, NULL, &consumer_birthFn, (void*)&consumerParams );
seanhalle@0:    
seanhalle@0:    for(i=0; i<args->numProducers; i++)
seanhalle@0:     { pthread_join( producerThds[i], NULL );
seanhalle@0:     }
seanhalle@0:    pthread_join( consumerThd, NULL );
seanhalle@0:    
seanhalle@0:    //work is all done, so take a measurement snapshot at end
seanhalle@0:    takeAMeas(0, benchEndMeas);
seanhalle@0:    
seanhalle@0:    
seanhalle@0: #ifdef MEASURE_PERF
seanhalle@0:    uint64_t totalExeCycles = ( benchEndMeas.cycles - benchStartMeas.cycles);
seanhalle@0:    printf("Total Execution: %lu\n", totalExeCycles);
seanhalle@0: #else
seanhalle@0:    uint64_t totalExeCycles = ( benchEndMeas.total - benchStartMeas.total);
seanhalle@0:    printf("Total Cycles of Execution: %lu\n", totalExeCycles);   
seanhalle@0: #endif
seanhalle@0: 
seanhalle@0:     //======================================================
seanhalle@0:  }
seanhalle@0: 
seanhalle@0: 
seanhalle@0: /*This parsed the command line arguments and returns the values in a struct
seanhalle@0:  * Command line args should be a '-' followed by a single letter, then a value
seanhalle@0:  */
seanhalle@0: ParsedArgs *
seanhalle@0: parse_arguments( int argc, char **argv )
seanhalle@0:  { ParsedArgs *parsedArgs;
seanhalle@0:    int i;
seanhalle@0:    
seanhalle@0:    parsedArgs = malloc(sizeof(ParsedArgs));
seanhalle@0:    if(argc < 2)
seanhalle@0:     { fprintf(stdout, "must give arguments");
seanhalle@0:       fputs(usage, stdout);
seanhalle@0:       return EXIT_FAILURE;
seanhalle@0:     }
seanhalle@0:    for( i=1; i < argc; i++ ) 
seanhalle@0:     { if(argv[i][0] == '-' && argv[i][2] == 0) 
seanhalle@0:        { switch(argv[i][1]) 
seanhalle@0:           { case 'p':
seanhalle@0:              { if(!isdigit(argv[++i][0])) 
seanhalle@0:                 { fprintf(stderr, "-p must be followed by the number of producer threads to spawn\n");
seanhalle@0:                   return EXIT_FAILURE;
seanhalle@0:                 }
seanhalle@0:                parsedArgs->numProducers = atoi(argv[i]);
seanhalle@0:                if( parsedArgs->numProducers == 0 ) 
seanhalle@0:                 { fprintf(stderr, "invalid number of producers specified: %d\n", parsedArgs->numProducers);
seanhalle@0:                   return EXIT_FAILURE;
seanhalle@0:                 }
seanhalle@0:                else
seanhalle@0:                 { DEBUG__printf1("num producers: %d\n", parsedArgs->numProducers );
seanhalle@0:                 }
seanhalle@0:              }
seanhalle@0:             break;
seanhalle@0:             case 't':
seanhalle@0:              { if( !isdigit( argv[++i][0] ) ) 
seanhalle@0:                 { fputs("-t must be followed by a number\n", stderr);
seanhalle@0:                   return EXIT_FAILURE;
seanhalle@0:                 }
seanhalle@0:                parsedArgs->numTuplesToCreate = atoi(argv[i]);
seanhalle@0:                DEBUG__printf1("num tuples to produce: %d\n", parsedArgs->numTuplesToCreate );
seanhalle@0:              }
seanhalle@0:             break;
seanhalle@0:             case 'h':
seanhalle@0:              { fputs(usage, stdout);
seanhalle@0:                return 0;
seanhalle@0:              }
seanhalle@0:             default:
seanhalle@0:              { fprintf(stderr, "unrecognized argument: %s\n", argv[i]);
seanhalle@0:                fputs(usage, stderr);
seanhalle@0:                return EXIT_FAILURE;
seanhalle@0:              }
seanhalle@0:           }
seanhalle@0:        }
seanhalle@0:       else 
seanhalle@0:        { fprintf(stdout, "unrecognized argument: %s\n", argv[i]);
seanhalle@0:          fputs(usage, stdout);
seanhalle@0:          return EXIT_FAILURE;
seanhalle@0:        }
seanhalle@0:     }//for
seanhalle@0:    return parsedArgs;
seanhalle@0:  }
seanhalle@0: 
seanhalle@0: int main(int argc, char **argv)
seanhalle@0:  { ParsedArgs *args;
seanhalle@0:    int i;
seanhalle@0:         
seanhalle@0:     
seanhalle@0:    set_up_performance_counters();
seanhalle@0:  
seanhalle@0:    init_stuff();
seanhalle@0:    
seanhalle@0:    args = parse_arguments( argc, argv);
seanhalle@0:    
seanhalle@0:    if( args < 10 ) return args +1; //non-zero exit when parsing went wrong
seanhalle@0:    
seanhalle@0:    benchmark( args );
seanhalle@0: 
seanhalle@0:    return 0;
seanhalle@0:  }
seanhalle@0: