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