benchmark/mpmc-queue/mpmc-queue.cc

   1 #include <inttypes.h>
   2 #include <threads.h>
   3 #include <stdio.h>
   4 #include <unistd.h>
   5 #include <stdlib.h>
   6
   7 #include <librace.h>
   8
   9 #include "mpmc-queue.h"
  10
  11 void threadA(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
  12 {
  13         int32_t *bin = queue->write_prepare();
  14         store_32(bin, 1);
  15         printf("write_bin %d, val %d\n", bin, 1);
  16         queue->write_publish(bin);
  17 }
  18
  19 void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
  20 {
  21         int32_t *bin;
  22         while (bin = queue->read_fetch()) {
  23                 printf("Read: %d\n", load_32(bin));
  24                 printf("read_bin %d, val %d\n", bin, load_32(bin));
  25                 queue->read_consume();
  26         }
  27 }
  28
  29 void threadC(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
  30 {
  31         int32_t *bin = queue->write_prepare();
  32         store_32(bin, 1);
  33         queue->write_publish(bin);
  34
  35         while (bin = queue->read_fetch()) {
  36                 printf("Read: %d\n", load_32(bin));
  37                 queue->read_consume();
  38         }
  39 }
  40
  41 #define MAXREADERS 3
  42 #define MAXWRITERS 3
  43 #define MAXRDWR 3
  44
  45 #ifdef CONFIG_MPMC_READERS
  46 #define DEFAULT_READERS (CONFIG_MPMC_READERS)
  47 #else
  48 #define DEFAULT_READERS 2
  49 #endif
  50
  51 #ifdef CONFIG_MPMC_WRITERS
  52 #define DEFAULT_WRITERS (CONFIG_MPMC_WRITERS)
  53 #else
  54 #define DEFAULT_WRITERS 2
  55 #endif
  56
  57 #ifdef CONFIG_MPMC_RDWR
  58 #define DEFAULT_RDWR (CONFIG_MPMC_RDWR)
  59 #else
  60 #define DEFAULT_RDWR 0
  61 #endif
  62
  63 int readers = DEFAULT_READERS, writers = DEFAULT_WRITERS, rdwr = DEFAULT_RDWR;
  64
  65 void print_usage()
  66 {
  67         printf("Error: use the following options\n"
  68                 " -r <num>              Choose number of reader threads\n"
  69                 " -w <num>              Choose number of writer threads\n");
  70         exit(EXIT_FAILURE);
  71 }
  72
  73 void process_params(int argc, char **argv)
  74 {
  75         const char *shortopts = "hr:w:";
  76         int opt;
  77         bool error = false;
  78
  79         while (!error && (opt = getopt(argc, argv, shortopts)) != -1) {
  80                 switch (opt) {
  81                 case 'h':
  82                         print_usage();
  83                         break;
  84                 case 'r':
  85                         readers = atoi(optarg);
  86                         break;
  87                 case 'w':
  88                         writers = atoi(optarg);
  89                         break;
  90                 default: /* '?' */
  91                         error = true;
  92                         break;
  93                 }
  94         }
  95
  96         if (writers < 1 || writers > MAXWRITERS)
  97                 error = true;
  98         if (readers < 1 || readers > MAXREADERS)
  99                 error = true;
 100
 101         if (error)
 102                 print_usage();
 103 }
 104
 105 int user_main(int argc, char **argv)
 106 {
 107         struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> queue;
 108         thrd_t A[MAXWRITERS], B[MAXREADERS], C[MAXRDWR];
 109
 110         /* Note: optarg() / optind is broken in model-checker - workaround is
 111          * to just copy&paste this test a few times */
 112         //process_params(argc, argv);
 113         printf("%d reader(s), %d writer(s)\n", readers, writers);
 114
 115 #ifndef CONFIG_MPMC_NO_INITIAL_ELEMENT
 116         printf("Adding initial element\n");
 117         int32_t *bin = queue.write_prepare();
 118         store_32(bin, 17);
 119         printf("init_write_bin %d, val %d\n", bin, 17);
 120         queue.write_publish(bin);
 121 #endif
 122
 123         printf("Start threads\n");
 124
 125         for (int i = 0; i < writers; i++)
 126                 thrd_create(&A[i], (thrd_start_t)&threadA, &queue);
 127         for (int i = 0; i < readers; i++)
 128                 thrd_create(&B[i], (thrd_start_t)&threadB, &queue);
 129
 130         for (int i = 0; i < rdwr; i++)
 131                 thrd_create(&C[i], (thrd_start_t)&threadC, &queue);
 132
 133         for (int i = 0; i < writers; i++)
 134                 thrd_join(A[i]);
 135         for (int i = 0; i < readers; i++)
 136                 thrd_join(B[i]);
 137         for (int i = 0; i < rdwr; i++)
 138                 thrd_join(C[i]);
 139
 140         printf("Threads complete\n");
 141
 142         return 0;
 143 }