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