cdschecker_modified_benchmarks/mpmc-queue/mpmc-queue.cc

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