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