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