queue.c (3097B)
1 #define HEADER_IMPL 2 #include "queue.h" 3 4 #include <assert.h> 5 #include <pthread.h> 6 #include <stdio.h> 7 #include <string.h> 8 #include <time.h> 9 10 static const size_t limit = 1000000000; 11 static const size_t qsize = PAGESZ_2M; 12 13 static void * 14 writer(void *data) 15 { 16 struct spsc_queue *queue = data; 17 18 size_t *ptr; 19 for (size_t i = 0; i < limit; i++) { 20 do { 21 ptr = spsc_queue_write(queue, sizeof *ptr, alignof(*ptr)); 22 } while (!ptr); 23 24 *ptr = i; 25 26 spsc_queue_write_commit(queue, ptr, sizeof *ptr); 27 } 28 29 return (void *) (sizeof *ptr * limit); // return bytes written 30 } 31 32 static void * 33 reader(void *data) 34 { 35 struct spsc_queue *queue = data; 36 37 size_t *ptr; 38 for (size_t i = 0; i < limit; i++) { 39 do { 40 ptr = spsc_queue_read(queue, sizeof *ptr, alignof(*ptr)); 41 } while (!ptr); 42 43 ASSERT(*ptr == i); 44 45 spsc_queue_read_commit(queue, ptr, sizeof *ptr); 46 } 47 48 return (void *) (sizeof *ptr * limit); // return bytes read 49 } 50 51 static void 52 benchmark(void) 53 { 54 struct spsc_queue queue; 55 int res = spsc_queue_init(&queue, qsize, qsize); 56 assert(res == 0); 57 58 const uint64_t nsecs = 1000000000, usecs = 1000000, msecs = 1000; 59 struct timespec start, end; 60 61 pthread_t writer_thread, reader_thread; 62 pthread_create(&reader_thread, NULL, reader, &queue); 63 pthread_create(&writer_thread, NULL, writer, &queue); 64 65 clock_gettime(CLOCK_MONOTONIC_RAW, &start); 66 67 void *nbytes_written; 68 pthread_join(writer_thread, &nbytes_written); 69 70 void *nbytes_read; 71 pthread_join(reader_thread, &nbytes_read); 72 73 clock_gettime(CLOCK_MONOTONIC_RAW, &end); 74 75 uint64_t ns = ((end.tv_sec - start.tv_sec) * nsecs) + (end.tv_nsec - start.tv_nsec); 76 uint64_t bytes = (uint64_t) nbytes_written + (uint64_t) nbytes_read; 77 double gigs = bytes / (double) GiB(1); 78 79 printf("Iters: %zu\n", limit); 80 printf("Elapsed time: ms: %lu, ns: %lu, avg. ns per iter: %lu\n", 81 ns / usecs, ns, ns / limit); 82 83 printf("Ops/sec: %lu, bytes written: %lu, bytes read: %lu, total GiBps: %.03f\n", 84 (limit * nsecs) / ns, (uint64_t) nbytes_written, (uint64_t) nbytes_read, 85 (float) ((gigs * nsecs) / ns)); 86 } 87 88 int 89 main(void) 90 { 91 int *foo = mirrormap(NULL, PAGESZ_4K, PAGESZ_4K, 2); 92 int *bar = (void *) ((uintptr_t) foo + PAGESZ_4K); 93 94 *foo = 42; 95 assert(*bar == 42); 96 97 munmap(foo, 0); 98 99 struct queue queue; 100 int res = queue_init(&queue, PAGESZ_4K, PAGESZ_4K); 101 assert(res == 0); 102 103 assert(queue_length(queue.head, queue.tail) == 0); 104 assert(queue_capacity(queue.cap, queue.head, queue.tail) == PAGESZ_4K); 105 106 char buf[] = "Hello, World!"; 107 108 char *mydst = queue_write(&queue, sizeof buf, alignof(char)); 109 assert(mydst); 110 queue_write_commit(&queue, mydst, sizeof buf); 111 112 strcpy(mydst, buf); 113 114 assert(queue_length(queue.head, queue.tail) == sizeof buf); 115 assert(queue_capacity(queue.cap, queue.head, queue.tail) == (PAGESZ_4K - sizeof buf)); 116 117 char *mysrc = queue_read(&queue, sizeof buf, alignof(char)); 118 assert(mysrc); 119 queue_read_commit(&queue, mysrc, sizeof buf); 120 121 printf("%s\n", mysrc); 122 123 assert(queue_length(queue.head, queue.tail) == 0); 124 assert(queue_capacity(queue.cap, queue.head, queue.tail) == PAGESZ_4K); 125 126 queue_free(&queue); 127 128 benchmark(); 129 130 return 0; 131 }