Mercurial > cgi-bin > hgwebdir.cgi > PR > Applications > Vthread > Vthread__Best_Effort_Msg__Bench
view 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 |
line source
1 /*
2 *
3 */
4 #include <stdio.h>
5 #include <stdlib.h>
6 #include <string.h>
7 #include <math.h>
8 #include <ctype.h>
9 #include <errno.h>
10 #include <pthread.h>
11 #include <unistd.h>
12 #include "VMS_Implementations/Vthread_impl/VPThread.h"
13 #include "C_Libraries/Queue_impl/PrivateQueue.h"
14 #include "C_Libraries/DynArray/DynArray.h"
15 #include "C_Libraries/BestEffortMessaging/LossyCom.h"
17 #include <linux/perf_event.h>
18 #include <linux/prctl.h>
19 #include <sys/syscall.h>
21 #undef DEBUG
22 //#define DEBUG
24 //#define MEASURE_PERF
26 #if !defined(unix) && !defined(__unix__)
27 #ifdef __MACH__
28 #define unix 1
29 #define __unix__ 1
30 #endif /* __MACH__ */
31 #endif /* unix */
33 /* find the appropriate way to define explicitly sized types */
34 /* for C99 or GNU libc (also mach's libc) we can use stdint.h */
35 #if (__STDC_VERSION__ >= 199900) || defined(__GLIBC__) || defined(__MACH__)
36 #include <stdint.h>
37 #elif defined(unix) || defined(__unix__) /* some UNIX systems have them in sys/types.h */
38 #include <sys/types.h>
39 #elif defined(__WIN32__) || defined(WIN32) /* the nameless one */
40 typedef unsigned __int8 uint8_t;
41 typedef unsigned __int32 uint32_t;
42 #endif /* sized type detection */
44 /* provide a millisecond-resolution timer for each system */
45 #if defined(unix) || defined(__unix__)
46 #include <time.h>
47 #include <sys/time.h>
48 unsigned long get_msec(void) {
49 static struct timeval timeval, first_timeval;
51 gettimeofday(&timeval, 0);
52 if(first_timeval.tv_sec == 0) {
53 first_timeval = timeval;
54 return 0;
55 }
56 return (timeval.tv_sec - first_timeval.tv_sec) * 1000 + (timeval.tv_usec - first_timeval.tv_usec) / 1000;
57 }
58 #elif defined(__WIN32__) || defined(WIN32)
59 #include <windows.h>
60 unsigned long get_msec(void) {
61 return GetTickCount();
62 }
63 #else
64 //#error "I don't know how to measure time on your platform"
65 #endif
67 //======================== Defines =========================
68 typedef struct perfData measurement_t;
69 struct perfData{
70 uint64 cycles;
71 uint64 instructions;
72 };
74 const char *usage = {
75 "Usage: msg_passing_test [options]\n"
76 " Starts threads equal to the number of cores and sends\n"
77 " messages to random receivers\n\n"
78 "Options:\n"
79 " -n <num> This specifies the number of sends done by each thread.\n"
80 " -h this help screen\n\n"
81 };
83 /***************************
84 * Barrier Implementation
85 ***************************/
87 struct barrier_t
88 {
89 int counter;
90 int nthreads;
91 int32 mutex;
92 int32 cond;
93 measurement_t endBarrierCycles;
95 };
96 typedef struct barrier_t barrier;
98 void inline barrier_init(barrier *barr, int nthreads, VirtProcr *animatingPr)
99 {
100 barr->counter = 0;
101 barr->nthreads = nthreads;
102 barr->mutex = VPThread__make_mutex(animatingPr);
103 barr->cond = VPThread__make_cond(barr->mutex, animatingPr);
104 }
106 int cycles_counter_main_fd;
107 void inline barrier_wait(barrier *barr, VirtProcr *animatingPr)
108 { int i;
110 VPThread__mutex_lock(barr->mutex, animatingPr);
111 barr->counter++;
112 if(barr->counter == barr->nthreads)
113 {
114 #ifdef MEASURE_PERF
115 read(cycles_counter_main_fd, &(barr->endBarrierCycles.cycles), \
116 sizeof(barr->endBarrierCycles.cycles));
117 #endif
119 barr->counter = 0;
120 for(i=0; i < barr->nthreads; i++)
121 VPThread__cond_signal(barr->cond, animatingPr);
122 }
123 else
124 { VPThread__cond_wait(barr->cond, animatingPr);
125 }
126 VPThread__mutex_unlock(barr->mutex, animatingPr);
127 }
130 /**************************
131 * Worker Parameters
132 **************************/
133 typedef struct
134 { struct barrier_t* barrier;
135 uint64_t totalWorkCycles;
136 uint64_t totalBadCycles;
137 uint64_t totalSyncCycles;
138 uint64_t totalBadSyncCycles;
139 uint64 numGoodSyncs;
140 uint64 numGoodTasks;
141 uint64_t coreID;
142 lossyCom__endpoint_t* localEndpoint;
143 lossyCom__exchange_t* centralMsgExchange;
144 unsigned int receivedACKs;
145 unsigned int broadcasterStatus;
146 unsigned int terminate;
147 }
148 WorkerParams;
150 typedef struct
151 { measurement_t *startExeCycles;
152 measurement_t *endExeCycles;
153 }
154 BenchParams;
156 typedef struct
157 {
158 lossyCom__endpointID_t receiverID;
159 lossyCom__msgBody_t msg;
160 } savedMsg_t;
162 //======================== Globals =========================
163 char __ProgrammName[] = "overhead_test";
164 char __DataSet[255];
166 int num_msg_to_send;
167 size_t chunk_size = 0;
169 int cycles_counter_fd[NUM_CORES];
170 struct perf_event_attr* hw_event;
172 WorkerParams *workerParamsArray;
174 // init random number
175 uint32_t seed1;
176 uint32_t seed2;
178 //======================== App Code =========================
179 /*
180 * Workload
181 */
183 #define saveCyclesAndInstrs(core,cycles) do{ \
184 int cycles_fd = cycles_counter_fd[core]; \
185 int nread; \
186 \
187 nread = read(cycles_fd,&(cycles),sizeof(cycles)); \
188 if(nread<0){ \
189 perror("Error reading cycles counter"); \
190 cycles = 0; \
191 } \
192 } while (0) //macro magic for scoping
194 extern inline uint32_t
195 randomNumber(uint32_t* seed1, uint32_t* seed2);
197 #define BROADCAST BROADCAST_ID
198 #define BROADCAST_ACK BROADCAST_ID-1
199 #define TERMINATE BROADCAST_ID-2
201 #define RECEIVING_BROADCAST 0
202 #define BROADCASTING 1
203 #define RECEIVING_ACK 2
205 /*
206 * Message Handler Function
207 */
208 void msgHandler(lossyCom__endpointID_t senderID, lossyCom__msgBody_t msg, void* data)
209 {
210 WorkerParams* threadData = (WorkerParams*)data;
211 lossyCom__endpoint_t* comEndpoint = threadData->localEndpoint;
212 lossyCom__endpointID_t receiverID;
214 if(msg == BROADCAST_ID) //answer broadcast message
215 {
216 lossyCom__sendMsg(comEndpoint, senderID, BROADCAST_ACK);
217 return;
218 }
219 if(msg == (BROADCAST_ACK) && threadData->broadcasterStatus == RECEIVING_ACK)
220 {
221 threadData->receivedACKs++;
222 if(threadData->receivedACKs == NUM_CORES/2)//chose next broadcaster
223 {
224 do{
225 receiverID = randomNumber(&seed1, &seed2) % NUM_CORES;
226 }while(receiverID == comEndpoint->endpointID);
228 //send the receiverID to the receiver to notify him that he is next
229 lossyCom__sendMsg(comEndpoint, receiverID, receiverID);
230 threadData->broadcasterStatus = RECEIVING_BROADCAST;
231 }
232 return;
233 }
234 if(msg == TERMINATE) //termination message
235 {
236 printf("endpoint %d received termination request\n", comEndpoint->endpointID);
237 threadData->terminate = TRUE;
238 return;
239 }
240 //
241 threadData->broadcasterStatus = BROADCASTING;
242 }
244 unsigned int global_broadcast_counter;
246 double
247 worker_TLF(void* _params, VirtProcr* animatingPr)
248 {
249 unsigned int msgCounter;
250 unsigned int broadcaster;
251 uint32_t wait_iterations;
252 WorkerParams* params = (WorkerParams*)_params;
253 unsigned int totalWorkCycles = 0, totalBadCycles = 0;
254 unsigned int totalSyncCycles = 0, totalBadSyncCycles = 0;
255 unsigned int workspace1=0, numGoodSyncs = 0, numGoodTasks = 0;
256 double workspace2=0.0;
258 //core 0 always starts
259 params->broadcasterStatus = params->coreID==0?BROADCASTING:RECEIVING_BROADCAST;
261 /*
262 int32 privateMutex = VPThread__make_mutex(animatingPr);
264 int cpuid = sched_getcpu();
266 measurement_t startWorkload, endWorkload, startWorkload2, endWorkload2;
267 uint64 numCycles;
268 */
269 #ifdef MEASURE_PERF
270 saveCyclesAndInstrs(cpuid,startWorkload.cycles);
271 #endif
273 //initialize endpoint for communication
274 lossyCom__endpoint_t comEndpoint;
275 params->localEndpoint = &comEndpoint;
276 lossyCom__initialize_endpoint(&comEndpoint,
277 params->centralMsgExchange,
278 params->coreID,
279 msgHandler,
280 params);
282 lossyCom__endpointID_t receiverID;
283 msgCounter = 0;
284 while(msgCounter <= num_msg_to_send)
285 {
286 int i;
288 if(params->broadcasterStatus == BROADCASTING)
289 {
290 if(msgCounter == num_msg_to_send)//send termination msg
291 {
292 lossyCom__sendMsg(&comEndpoint,BROADCAST_ID, TERMINATE);
293 break;
294 }else{ //generate and send random message
295 params->receivedACKs = 0;
296 lossyCom__sendMsg(&comEndpoint, BROADCAST_ID, BROADCAST);
297 global_broadcast_counter++;
298 if(global_broadcast_counter % 1000 == 0){
299 printf("broadcast count: %d\n", global_broadcast_counter);
300 }
301 params->broadcasterStatus = RECEIVING_ACK; //mark msg as send
302 msgCounter++;
303 }
304 }
306 //check if the benchmark should terminate
307 if(params->terminate)
308 break;
310 //receive msg
311 lossyCom__receiveMsg(&comEndpoint);
312 }
315 #ifdef MEASURE_PERF
316 saveCyclesAndInstrs(cpuid,endWorkload.cycles);
317 numCycles = endWorkload.cycles - startWorkload.cycles;
318 //sanity check (400K is about 20K iters)
319 if( numCycles < 400000 ) {totalWorkCycles += numCycles; numGoodTasks++;}
320 else {totalBadCycles += numCycles; }
321 #endif
323 barrier_wait(params->barrier, animatingPr);
325 params->totalWorkCycles = totalWorkCycles;
326 params->totalBadCycles = totalBadCycles;
327 params->numGoodTasks = numGoodTasks;
328 params->totalSyncCycles = totalSyncCycles;
329 params->totalBadSyncCycles = totalBadSyncCycles;
330 params->numGoodSyncs = numGoodSyncs;
331 /*
332 params->totalSyncCycles = VMS__give_num_plugin_cycles();
333 params->totalBadSyncCycles = 0;
334 params->numGoodSyncs = VMS__give_num_plugin_animations();
335 */
336 //Shutdown worker
337 VPThread__dissipate_thread(animatingPr);
339 //below return never reached --> there for gcc
340 return (workspace1 + workspace2); //to prevent gcc from optimizing work out
341 }
344 /* this is run after the VMS is set up*/
345 void benchmark(void *_params, VirtProcr *animatingPr)
346 {
347 int i, cpuID, idx;
348 struct barrier_t barr;
349 BenchParams *params;
351 params = (BenchParams *)_params;
353 barrier_init(&barr, NUM_CORES+1, animatingPr);
355 //Init central communication exchange
356 lossyCom__exchange_t* centralMsgExchange = lossyCom__initialize(NUM_CORES);
358 //prepare input
359 for(i=0; i<NUM_CORES; i++)
360 {
361 workerParamsArray[i].barrier = &barr;
362 workerParamsArray[i].coreID = i;
363 workerParamsArray[i].centralMsgExchange = centralMsgExchange;
364 workerParamsArray[i].terminate = FALSE;
365 }
366 global_broadcast_counter = 0;
368 // init random number generator for wait and msg content
369 seed1 = rand()%1000;
370 seed2 = rand()%1000;
372 //save cycles before execution of threads, to get total exe cycles
373 measurement_t *startExeCycles, *endExeCycles;
374 startExeCycles = params->startExeCycles;
376 #ifdef MEASURE_PERF
377 int nread = read(cycles_counter_main_fd, &(startExeCycles->cycles),
378 sizeof(startExeCycles->cycles));
379 if(nread<0) perror("Error reading cycles counter");
380 #endif
382 //create (which starts running) all threads
383 for(i=NUM_CORES-1; i>=0; i--)
384 {
385 VPThread__create_thread_with_affinity((VirtProcrFnPtr)worker_TLF,
386 &(workerParamsArray[i]),
387 animatingPr,
388 i);//schedule to core i
389 }
391 #ifdef MEASURE_PERF
392 //endBarrierCycles read in barrier_wait()! Merten, email me if want to chg
393 params->endExeCycles->cycles = barr.endBarrierCycles.cycles;
394 #endif
396 barrier_wait(&barr, animatingPr);
397 printf("Total broadcast count: %d\n", global_broadcast_counter);
399 //print send msgs
400 /*
401 printf("sendMsgs = []\n");
402 for(i = 0; i<NUM_CORES; i++)
403 {
404 printf("sendMsgs.append([");
405 for(idx = 0; idx< workerParamsArray[i].sendMsgs->numInArray; idx++)
406 {
407 printf("(%lu, %lu),",
408 (uint64_t)(workerParamsArray[i].ptrToArrayOfSendMsgs[idx]) & 0xFFFFFFFF,
409 ((uint64_t)(workerParamsArray[i].ptrToArrayOfSendMsgs[idx]) >> 32 ) & 0xFFFFFFFF);
410 }
411 printf("])\n");
412 }
415 //print received msgs
416 printf("receivedMsgs = []\n");
417 for(i = 0; i<NUM_CORES; i++)
418 {
419 printf("receivedMsgs.append([");
420 for(idx = 0; idx< workerParamsArray[i].receivedMsgs->numInArray; idx++)
421 {
422 printf("(%lu, %lu),",
423 (uint64_t)(workerParamsArray[i].ptrToArrayOfReceivedMsgs[idx]) & 0xFFFFFFFF,
424 ((uint64_t)(workerParamsArray[i].ptrToArrayOfReceivedMsgs[idx]) >> 32 ) & 0xFFFFFFFF);
425 }
426 printf("])\n");
427 }*/
429 /*
430 uint64_t overallWorkCycles = 0;
431 for(i=0; i<num_threads; i++){
432 printf("WorkCycles: %lu\n",input[i].totalWorkCycles);
433 overallWorkCycles += input[i].totalWorkCycles;
434 }
436 printf("Sum across threads of work cycles: %lu\n", overallWorkCycles);
437 printf("Total Execution: %lu\n", endBenchTime.cycles-startBenchTime.cycles);
438 printf("Runtime/Workcycle Ratio %lu\n",
439 ((endBenchTime.cycles-startBenchTime.cycles)*100)/overallWorkCycles);
440 */
442 //======================================================
444 VPThread__dissipate_thread(animatingPr);
445 }
447 int main(int argc, char **argv)
448 {
449 int i;
451 //set global static variables, based on cmd-line args
452 for(i=1; i<argc; i++)
453 {
454 if(argv[i][0] == '-' && argv[i][2] == 0)
455 {
456 switch(argv[i][1])
457 {
458 case 'n':
459 if(!isdigit(argv[++i][0]))
460 {
461 fprintf(stderr, "-t must be followed by the number messages to send per core\n");
462 return EXIT_FAILURE;
463 }
464 num_msg_to_send = atoi(argv[i]);
465 if(!num_msg_to_send)
466 {
467 fprintf(stderr, "invalid number of messages to send: %d\n", num_msg_to_send);
468 return EXIT_FAILURE;
469 }
470 break;
471 case 'h':
472 fputs(usage, stdout);
473 return 0;
474 default:
475 fprintf(stderr, "unrecognized argument: %s\n", argv[i]);
476 fputs(usage, stderr);
477 return EXIT_FAILURE;
478 }//switch
479 }//if arg
480 else
481 {
482 fprintf(stderr, "unrecognized argument: %s\n", argv[i]);
483 fputs(usage, stderr);
484 return EXIT_FAILURE;
485 }
486 }//for
489 #ifdef MEASURE_PERF
490 //setup performance counters
491 hw_event = malloc(sizeof(struct perf_event_attr));
492 memset(hw_event,0,sizeof(struct perf_event_attr));
494 hw_event->type = PERF_TYPE_HARDWARE;
495 hw_event->size = sizeof(hw_event);
496 hw_event->disabled = 0;
497 hw_event->freq = 0;
498 hw_event->inherit = 1; /* children inherit it */
499 hw_event->pinned = 1; /* says this virt counter must always be on HW */
500 hw_event->exclusive = 0; /* only group on PMU */
501 hw_event->exclude_user = 0; /* don't count user */
502 hw_event->exclude_kernel = 1; /* don't count kernel */
503 hw_event->exclude_hv = 1; /* ditto hypervisor */
504 hw_event->exclude_idle = 1; /* don't count when idle */
505 hw_event->mmap = 0; /* include mmap data */
506 hw_event->comm = 0; /* include comm data */
508 hw_event->config = PERF_COUNT_HW_CPU_CYCLES; //cycles
510 int cpuID, retries;
512 for( cpuID = 0; cpuID < NUM_CORES; cpuID++ )
513 { retries = 0;
514 do
515 { retries += 1;
516 cycles_counter_fd[cpuID] =
517 syscall(__NR_perf_event_open, hw_event,
518 0,//pid_t: 0 is "pid of calling process"
519 cpuID,//int: cpu, the value returned by "CPUID" instr(?)
520 -1,//int: group_fd, -1 is "leader" or independent
521 0//unsigned long: flags
522 );
523 }
524 while(cycles_counter_fd[cpuID]<0 && retries < 100);
525 if(retries >= 100)
526 {
527 fprintf(stderr,"On core %d: ",cpuID);
528 perror("Failed to open cycles counter");
529 }
530 }
532 //Set up counter to accumulate total cycles to process, across all CPUs
534 retries = 0;
535 do
536 { retries += 1;
537 cycles_counter_main_fd =
538 syscall(__NR_perf_event_open, hw_event,
539 0,//pid_t: 0 is "pid of calling process"
540 -1,//int: cpu, -1 means accumulate from all cores
541 -1,//int: group_fd, -1 is "leader" == independent
542 0//unsigned long: flags
543 );
544 }
545 while(cycles_counter_main_fd<0 && retries < 100);
546 if(retries >= 100)
547 {
548 fprintf(stderr,"in main ");
549 perror("Failed to open cycles counter");
550 }
551 #endif
553 measurement_t startExeCycles, endExeCycles;
554 BenchParams *benchParams;
556 benchParams = malloc(sizeof(BenchParams));
558 benchParams->startExeCycles = &startExeCycles;
559 benchParams->endExeCycles = &endExeCycles;
561 workerParamsArray = (WorkerParams *)malloc( (NUM_CORES) * sizeof(WorkerParams) );
562 if(workerParamsArray == NULL ) printf("error mallocing worker params array\n");
565 //This is the transition to the VMS runtime
566 VPThread__create_seed_procr_and_do_work( &benchmark, benchParams );
568 #ifdef MEASURE_PERF
569 uint64_t totalWorkCyclesAcrossCores = 0, totalBadCyclesAcrossCores = 0;
570 uint64_t totalSyncCyclesAcrossCores = 0, totalBadSyncCyclesAcrossCores = 0;
571 for(i=0; i<num_threads; i++){
572 printf("WorkCycles: %lu\n",workerParamsArray[i].totalWorkCycles);
573 // printf("Num Good Tasks: %lu\n",workerParamsArray[i].numGoodTasks);
574 // printf("SyncCycles: %lu\n",workerParamsArray[i].totalSyncCycles);
575 // printf("Num Good Syncs: %lu\n",workerParamsArray[i].numGoodSyncs);
576 totalWorkCyclesAcrossCores += workerParamsArray[i].totalWorkCycles;
577 totalBadCyclesAcrossCores += workerParamsArray[i].totalBadCycles;
578 totalSyncCyclesAcrossCores += workerParamsArray[i].totalSyncCycles;
579 totalBadSyncCyclesAcrossCores += workerParamsArray[i].totalBadSyncCycles;
580 }
582 uint64_t totalExeCycles = endExeCycles.cycles - startExeCycles.cycles;
583 totalExeCycles -= totalBadCyclesAcrossCores;
584 uint64 totalOverhead = totalExeCycles - totalWorkCyclesAcrossCores;
585 int32 numSyncs = outer_iters * num_threads * 2;
586 printf("Total Execution Cycles: %lu\n", totalExeCycles);
587 printf("Sum across threads of work cycles: %lu\n", totalWorkCyclesAcrossCores);
588 printf("Sum across threads of bad work cycles: %lu\n", totalBadCyclesAcrossCores);
589 // printf("Sum across threads of Bad Sync cycles: %lu\n", totalBadSyncCyclesAcrossCores);
590 printf("Overhead per sync: %f\n", (double)totalOverhead / (double)numSyncs );
591 printf("ExeCycles/WorkCycles Ratio %f\n",
592 (double)totalExeCycles / (double)totalWorkCyclesAcrossCores);
593 #else
594 printf("#No measurement done!\n");
595 #endif
596 return 0;
597 }