PR/Applications/Vthread/Vthread__Best_Effort_Msg_

comparison main.c @ 20:29b273cf3b1f

Benchmark that tests msg loss

author	Merten Sach <msach@mailbox.tu-berlin.de>
date	Tue, 13 Mar 2012 12:25:48 +0100
parents	fdc2f264f3d6
children	08b37152b48d

comparison

equal deleted inserted replaced

-:cf8ba4d052e6
+:c24d544a2a0a
 #include <errno.h>
 #include <pthread.h>
 #include <unistd.h>
 #include "VMS_Implementations/Vthread_impl/VPThread.h"
 #include "C_Libraries/Queue_impl/PrivateQueue.h"
+#include "C_Libraries/DynArray/DynArray.h"
+#include "C_Libraries/BestEffortMessaging/LossyCom.h"
 #include <linux/perf_event.h>
 #include <linux/prctl.h>
 #include <sys/syscall.h>
 #undef DEBUG
 //#define DEBUG
-#define MEASURE_PERF
+//#define MEASURE_PERF
 #if !defined(unix) && !defined(__unix__)
 #ifdef __MACH__
 #define unix		1
 #define __unix__	1
 uint64 cycles;
 uint64 instructions;
 };
 const char *usage = {
-	"Usage: malloc_test [options]\n"
+	"Usage: msg_passing_test [options]\n"
-	"  Spwans a number of threads and allocates memory.\n\n"
+	"  Starts threads equal to the number of cores and sends\n"
+"  messages to random receivers\n\n"
 	"Options:\n"
-	"  -t <num>   how many threads to use (default: 1). This is internaly multiplied by the number of cores.\n"
+	"  -n <num>   This specifies the number of sends done by each thread.\n"
-	"  -o <num>   repeat workload and sync operation <m> times\n"
-"  -i <num>   size of workload, repeat <n> times\n"
 	"  -h         this help screen\n\n"
 };
+/***************************
+* Barrier Implementation
+***************************/
 struct barrier_t
 {
 int counter;
 int nthreads;
 }
 VPThread__mutex_unlock(barr->mutex, animatingPr);
 }
+/**************************
+* Worker Parameters
+**************************/
 typedef struct
 { struct barrier_t* barrier;
 uint64_t  totalWorkCycles;
 uint64_t  totalBadCycles;
 uint64_t  totalSyncCycles;
 uint64_t  totalBadSyncCycles;
 uint64     numGoodSyncs;
 uint64     numGoodTasks;
+uint64_t   coreID;
+lossyCom__endpoint_t* localEndpoint;
+lossyCom__exchange_t* centralMsgExchange;
+unsigned int receivedACKs;
+unsigned int broadcasterStatus;
+unsigned int terminate;
 }
 WorkerParams;
 typedef struct
 { measurement_t *startExeCycles;
 measurement_t *endExeCycles;
 }
 BenchParams;
+typedef struct
+{
+lossyCom__endpointID_t receiverID;
+lossyCom__msgBody_t msg;
+} savedMsg_t;
 //======================== Globals =========================
 char __ProgrammName[] = "overhead_test";
 char __DataSet[255];
-int outer_iters, inner_iters, num_threads;
+int num_msg_to_send;
 size_t chunk_size = 0;
 int cycles_counter_fd[NUM_CORES];
 struct perf_event_attr* hw_event;
 WorkerParams *workerParamsArray;
+// init random number
+uint32_t seed1;
+uint32_t seed2;
 //======================== App Code =========================
 /*
 * Workload
 */
 perror("Error reading cycles counter");          \
 cycles = 0;                                      \
 }                                                    \
 } while (0) //macro magic for scoping
+extern inline uint32_t
+randomNumber(uint32_t* seed1, uint32_t* seed2);
+#define BROADCAST BROADCAST_ID
+#define BROADCAST_ACK BROADCAST_ID-1
+#define TERMINATE BROADCAST_ID-2
+#define RECEIVING_BROADCAST 0
+#define BROADCASTING 1
+#define RECEIVING_ACK 2
+/*
+* Message Handler Function
+*/
+void msgHandler(lossyCom__endpointID_t senderID, lossyCom__msgBody_t msg, void* data)
+{
+WorkerParams* threadData = (WorkerParams*)data;
+lossyCom__endpoint_t* comEndpoint = threadData->localEndpoint;
+lossyCom__endpointID_t receiverID;
+if(msg == BROADCAST_ID) //answer broadcast message
+{
+lossyCom__sendMsg(comEndpoint, senderID, BROADCAST_ACK);
+return;
+}
+if(msg == (BROADCAST_ACK) && threadData->broadcasterStatus == RECEIVING_ACK)
+{
+threadData->receivedACKs++;
+if(threadData->receivedACKs == NUM_CORES/2)//chose next broadcaster
+{
+do{
+receiverID = randomNumber(&seed1, &seed2) % NUM_CORES;
+}while(receiverID == comEndpoint->endpointID);
+//send the receiverID to the receiver to notify him that he is next
+lossyCom__sendMsg(comEndpoint, receiverID, receiverID);
+threadData->broadcasterStatus = RECEIVING_BROADCAST;
+}
+return;
+}
+if(msg == TERMINATE) //termination message
+{
+printf("endpoint %d received termination request\n", comEndpoint->endpointID);
+threadData->terminate = TRUE;
+return;
+}
+//
+threadData->broadcasterStatus = BROADCASTING;
+}
+unsigned int global_broadcast_counter;
 double
 worker_TLF(void* _params, VirtProcr* animatingPr)
 {
-int i,o;
+unsigned int msgCounter;
+unsigned int broadcaster;
+uint32_t wait_iterations;
 WorkerParams* params = (WorkerParams*)_params;
 unsigned int totalWorkCycles = 0, totalBadCycles = 0;
 unsigned int totalSyncCycles = 0, totalBadSyncCycles = 0;
 unsigned int workspace1=0, numGoodSyncs = 0, numGoodTasks = 0;
 double workspace2=0.0;
+//core 0 always starts
+params->broadcasterStatus = params->coreID==0?BROADCASTING:RECEIVING_BROADCAST;
+/*
 int32 privateMutex = VPThread__make_mutex(animatingPr);
 int cpuid = sched_getcpu();
 measurement_t startWorkload, endWorkload, startWorkload2, endWorkload2;
 uint64 numCycles;
-for(o=0; o < outer_iters; o++)
+*/
-{
 #ifdef MEASURE_PERF
 saveCyclesAndInstrs(cpuid,startWorkload.cycles);
 #endif
+//initialize endpoint for communication
+lossyCom__endpoint_t comEndpoint;
+params->localEndpoint = &comEndpoint;
+lossyCom__initialize_endpoint(&comEndpoint,
+params->centralMsgExchange,
+params->coreID,
+msgHandler,
+params);
+lossyCom__endpointID_t receiverID;
+msgCounter = 0;
+while(msgCounter <= num_msg_to_send)
+{
+int i;
-//workltask
+if(params->broadcasterStatus == BROADCASTING)
-for(i=0; i < inner_iters; i++)
 {
-workspace1 += (workspace1 + 32)/2;
+if(msgCounter == num_msg_to_send)//send termination msg
-workspace2 += (workspace2 + 23.2)/1.4;
+{
+lossyCom__sendMsg(&comEndpoint,BROADCAST_ID, TERMINATE);
+break;
+}else{ //generate and send random message
+params->receivedACKs = 0;
+lossyCom__sendMsg(&comEndpoint, BROADCAST_ID, BROADCAST);
+global_broadcast_counter++;
+if(global_broadcast_counter % 1000 == 0){
+printf("broadcast count: %d\n", global_broadcast_counter);
+}
+params->broadcasterStatus = RECEIVING_ACK; //mark msg as send
+msgCounter++;
+}
 }
+//check if the benchmark should terminate
+if(params->terminate)
+break;
+//receive msg
+lossyCom__receiveMsg(&comEndpoint);
+}
 #ifdef MEASURE_PERF
 saveCyclesAndInstrs(cpuid,endWorkload.cycles);
 numCycles = endWorkload.cycles - startWorkload.cycles;
 //sanity check (400K is about 20K iters)
 if( numCycles < 400000 ) {totalWorkCycles += numCycles; numGoodTasks++;}
 else                     {totalBadCycles  += numCycles; }
 #endif
-//mutex access often causes switch to different Slave VP
+barrier_wait(params->barrier, animatingPr);
-VPThread__mutex_lock(privateMutex, animatingPr);
-/*
-saveCyclesAndInstrs(cpuid,startWorkload2.cycles);
-//Task
-for(i=0; i < inner_iters; i++)
-{
-workspace1 += (workspace1 + 32)/2;
-workspace2 += (workspace2 + 23.2)/1.4;
-}
-saveCyclesAndInstrs(cpuid,endWorkload2.cycles);
-numCycles = endWorkload2.cycles - startWorkload2.cycles;
-//sanity check (400K is about 20K iters)
-if( numCycles < 400000 ) {totalWorkCycles += numCycles; numGoodTasks++;}
-else                     {totalBadCycles  += numCycles; }
-*/
-VPThread__mutex_unlock(privateMutex, animatingPr);
-}
 params->totalWorkCycles = totalWorkCycles;
 params->totalBadCycles = totalBadCycles;
 params->numGoodTasks   = numGoodTasks;
 params->totalSyncCycles = totalSyncCycles;
 /*
 params->totalSyncCycles = VMS__give_num_plugin_cycles();
 params->totalBadSyncCycles = 0;
 params->numGoodSyncs = VMS__give_num_plugin_animations();
 */
-//Wait for all threads to end
-barrier_wait(params->barrier, animatingPr);
 //Shutdown worker
 VPThread__dissipate_thread(animatingPr);
 //below return never reached --> there for gcc
 return (workspace1 + workspace2);  //to prevent gcc from optimizing work out
 /* this is run after the VMS is set up*/
 void benchmark(void *_params, VirtProcr *animatingPr)
 {
-int i, cpuID;
+int i, cpuID, idx;
-struct barrier_t  barr;
+struct barrier_t barr;
 BenchParams      *params;
 params = (BenchParams *)_params;
-barrier_init(&barr, num_threads+1, animatingPr);
+barrier_init(&barr, NUM_CORES+1, animatingPr);
+//Init central communication exchange
+lossyCom__exchange_t* centralMsgExchange = lossyCom__initialize(NUM_CORES);
 //prepare input
-for(i=0; i<num_threads; i++)
+for(i=0; i<NUM_CORES; i++)
 {
 workerParamsArray[i].barrier = &barr;
-}
+workerParamsArray[i].coreID = i;
+workerParamsArray[i].centralMsgExchange = centralMsgExchange;
+workerParamsArray[i].terminate = FALSE;
+}
+global_broadcast_counter = 0;
+// init random number generator for wait and msg content
+seed1 = rand()%1000;
+seed2 = rand()%1000;
 //save cycles before execution of threads, to get total exe cycles
 measurement_t *startExeCycles, *endExeCycles;
 startExeCycles = params->startExeCycles;
 sizeof(startExeCycles->cycles));
 if(nread<0) perror("Error reading cycles counter");
 #endif
 //create (which starts running) all threads
-for(i=0; i<num_threads; i++)
+for(i=NUM_CORES-1; i>=0; i--)
-{ VPThread__create_thread((VirtProcrFnPtr)worker_TLF, &(workerParamsArray[i]), animatingPr);
+{
-}
+VPThread__create_thread_with_affinity((VirtProcrFnPtr)worker_TLF,
-//wait for all threads to finish
+&(workerParamsArray[i]),
-barrier_wait(&barr, animatingPr);
+animatingPr,
+i);//schedule to core i
+}
 #ifdef MEASURE_PERF
 //endBarrierCycles read in barrier_wait()!  Merten, email me if want to chg
 params->endExeCycles->cycles = barr.endBarrierCycles.cycles;
 #endif
+barrier_wait(&barr, animatingPr);
+printf("Total broadcast count: %d\n", global_broadcast_counter);
+//print send msgs
+/*
+printf("sendMsgs = []\n");
+for(i = 0; i<NUM_CORES; i++)
+{
+printf("sendMsgs.append([");
+for(idx = 0; idx< workerParamsArray[i].sendMsgs->numInArray; idx++)
+{
+printf("(%lu, %lu),",
+(uint64_t)(workerParamsArray[i].ptrToArrayOfSendMsgs[idx]) & 0xFFFFFFFF,
+((uint64_t)(workerParamsArray[i].ptrToArrayOfSendMsgs[idx]) >> 32 ) & 0xFFFFFFFF);
+}
+printf("])\n");
+}
+//print received msgs
+printf("receivedMsgs = []\n");
+for(i = 0; i<NUM_CORES; i++)
+{
+printf("receivedMsgs.append([");
+for(idx = 0; idx< workerParamsArray[i].receivedMsgs->numInArray; idx++)
+{
+printf("(%lu, %lu),",
+(uint64_t)(workerParamsArray[i].ptrToArrayOfReceivedMsgs[idx]) & 0xFFFFFFFF,
+((uint64_t)(workerParamsArray[i].ptrToArrayOfReceivedMsgs[idx]) >> 32 ) & 0xFFFFFFFF);
+}
+printf("])\n");
+}*/
 /*
 uint64_t overallWorkCycles = 0;
 for(i=0; i<num_threads; i++){
 printf("WorkCycles: %lu\n",input[i].totalWorkCycles);
 {
 if(argv[i][0] == '-' && argv[i][2] == 0)
 {
 switch(argv[i][1])
 {
-case 't':
+case 'n':
 if(!isdigit(argv[++i][0]))
 {
-fprintf(stderr, "-t must be followed by the number of worker threads to spawn\n");
+fprintf(stderr, "-t must be followed by the number messages to send per core\n");
 return EXIT_FAILURE;
 }
-num_threads = atoi(argv[i]);
+num_msg_to_send = atoi(argv[i]);
-if(!num_threads)
+if(!num_msg_to_send)
 {
-fprintf(stderr, "invalid number of threads specified: %d\n", num_threads);
+fprintf(stderr, "invalid number of messages to send: %d\n", num_msg_to_send);
 return EXIT_FAILURE;
 }
 break;
-case 'o':
-if(!isdigit(argv[++i][0]))
-{
-fputs("-i must be followed by a number\n", stderr);
-return EXIT_FAILURE;
-}
-outer_iters = atoi(argv[i]);
-				break;
-case 'i':
-if(!isdigit(argv[++i][0]))
-{
-fputs("-o must be followed by a number (workload size)\n", stderr);
-return EXIT_FAILURE;
-}
-inner_iters = atoi(argv[i]);
-				break;
 case 'h':
 fputs(usage, stdout);
 return 0;
 default:
 fprintf(stderr, "unrecognized argument: %s\n", argv[i]);
 fputs(usage, stderr);
 return EXIT_FAILURE;
 }//switch
 }//if arg
 else
 {
-			fprintf(stderr, "unrecognized argument: %s\n", argv[i]);
+		fprintf(stderr, "unrecognized argument: %s\n", argv[i]);
-			fputs(usage, stderr);
+		fputs(usage, stderr);
-			return EXIT_FAILURE;
+		return EXIT_FAILURE;
 }
 }//for
 #ifdef MEASURE_PERF
 benchParams = malloc(sizeof(BenchParams));
 benchParams->startExeCycles = &startExeCycles;
 benchParams->endExeCycles   = &endExeCycles;
-workerParamsArray =  (WorkerParams *)malloc( (num_threads + 1) * sizeof(WorkerParams) );
+workerParamsArray =  (WorkerParams *)malloc( (NUM_CORES) * sizeof(WorkerParams) );
 if(workerParamsArray == NULL ) printf("error mallocing worker params array\n");
 //This is the transition to the VMS runtime
 VPThread__create_seed_procr_and_do_work( &benchmark, benchParams );
 //   printf("Sum across threads of Bad Sync cycles: %lu\n", totalBadSyncCyclesAcrossCores);
 printf("Overhead per sync: %f\n", (double)totalOverhead / (double)numSyncs );
 printf("ExeCycles/WorkCycles Ratio %f\n",
 (double)totalExeCycles / (double)totalWorkCyclesAcrossCores);
 #else
-printf("No measurement done!\n");
+printf("#No measurement done!\n");
 #endif
 return 0;
 }

Mercurial > cgi-bin > hgwebdir.cgi > PR > Applications > Vthread > Vthread__Best_Effort_Msg__Bench

comparison main.c @ 20:29b273cf3b1f