From 41d403a758afbfbe951e96ebb9461eb4b245363d Mon Sep 17 00:00:00 2001
From: Brian Norris <banorris@uci.edu>
Date: Fri, 2 Nov 2012 16:23:10 -0700
Subject: [PATCH] mpmc-queue: expand to 2R2W, 2R1W, 1R2W tests
I rewrote the test to allow a configurable number of readers and
writers, but then I figured out that our model-checker doesn't support
optarg() and its global optind properly, when using mmap()/mprotect()
snapshotting (it doesn't add the optarg library to be snapshotted...)
So, this test grows to 3 separate source files, at least for now. The
only change is the number of reader and writer threads.
---
mpmc-queue/.gitignore | 2 +
mpmc-queue/Makefile | 5 +-
mpmc-queue/mpmc-1r2w.cc | 101 +++++++++++++++++++++++++++++++++++++++
mpmc-queue/mpmc-2r1w.cc | 101 +++++++++++++++++++++++++++++++++++++++
mpmc-queue/mpmc-queue.cc | 71 +++++++++++++++++++++++----
5 files changed, 269 insertions(+), 11 deletions(-)
create mode 100644 mpmc-queue/mpmc-1r2w.cc
create mode 100644 mpmc-queue/mpmc-2r1w.cc
diff --git a/mpmc-queue/.gitignore b/mpmc-queue/.gitignore
index 4b0d836..6166337 100644
--- a/mpmc-queue/.gitignore
+++ b/mpmc-queue/.gitignore
@@ -1 +1,3 @@
/mpmc-queue
+/mpmc-1r2w
+/mpmc-2r1w
diff --git a/mpmc-queue/Makefile b/mpmc-queue/Makefile
index 07e2540..6636fcd 100644
--- a/mpmc-queue/Makefile
+++ b/mpmc-queue/Makefile
@@ -1,10 +1,11 @@
include ../benchmarks.mk
TESTNAME = mpmc-queue
+TESTS = mpmc-queue mpmc-1r2w mpmc-2r1w
-all: $(TESTNAME)
+all: $(TESTS)
-$(TESTNAME): $(TESTNAME).cc $(TESTNAME).h
+%: %.cc $(TESTNAME).h
$(CXX) -o $@ $< $(CPPFLAGS) $(LDFLAGS)
clean:
diff --git a/mpmc-queue/mpmc-1r2w.cc b/mpmc-queue/mpmc-1r2w.cc
new file mode 100644
index 0000000..78c9597
--- /dev/null
+++ b/mpmc-queue/mpmc-1r2w.cc
@@ -0,0 +1,101 @@
+#include <inttypes.h>
+#include <threads.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <stdlib.h>
+
+#include <librace.h>
+
+#include "mpmc-queue.h"
+
+void threadA(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+ int32_t *bin = queue->write_prepare();
+ store_32(bin, 1);
+ queue->write_publish();
+}
+
+void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+ int32_t *bin;
+ while (bin = queue->read_fetch()) {
+ printf("Read: %d\n", load_32(bin));
+ queue->read_consume();
+ }
+}
+
+#define MAXREADERS 3
+#define MAXWRITERS 3
+
+int readers = 1, writers = 2;
+
+void print_usage()
+{
+ printf("Error: use the following options\n"
+ " -r <num> Choose number of reader threads\n"
+ " -w <num> Choose number of writer threads\n");
+ exit(EXIT_FAILURE);
+}
+
+void process_params(int argc, char **argv)
+{
+ const char *shortopts = "hr:w:";
+ int opt;
+ bool error = false;
+
+ while (!error && (opt = getopt(argc, argv, shortopts)) != -1) {
+ switch (opt) {
+ case 'h':
+ print_usage();
+ break;
+ case 'r':
+ readers = atoi(optarg);
+ break;
+ case 'w':
+ writers = atoi(optarg);
+ break;
+ default: /* '?' */
+ error = true;
+ break;
+ }
+ }
+
+ if (writers < 1 || writers > MAXWRITERS)
+ error = true;
+ if (readers < 1 || readers > MAXREADERS)
+ error = true;
+
+ if (error)
+ print_usage();
+}
+
+int user_main(int argc, char **argv)
+{
+ struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> queue;
+ thrd_t A[MAXWRITERS], B[MAXREADERS];
+
+ /* Note: optarg() / optind is broken in model-checker - workaround is
+ * to just copy&paste this test a few times */
+ //process_params(argc, argv);
+ printf("%d reader(s), %d writer(s)\n", readers, writers);
+
+ int32_t *bin = queue.write_prepare();
+ store_32(bin, 17);
+ queue.write_publish();
+
+ printf("Start threads\n");
+
+ for (int i = 0; i < writers; i++)
+ thrd_create(&A[i], (thrd_start_t)&threadA, &queue);
+ for (int i = 0; i < readers; i++)
+ thrd_create(&B[i], (thrd_start_t)&threadB, &queue);
+
+ for (int i = 0; i < writers; i++)
+ thrd_join(A[i]);
+ for (int i = 0; i < readers; i++)
+ thrd_join(B[i]);
+
+ printf("Threads complete\n");
+
+ return 0;
+}
diff --git a/mpmc-queue/mpmc-2r1w.cc b/mpmc-queue/mpmc-2r1w.cc
new file mode 100644
index 0000000..8fc4652
--- /dev/null
+++ b/mpmc-queue/mpmc-2r1w.cc
@@ -0,0 +1,101 @@
+#include <inttypes.h>
+#include <threads.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <stdlib.h>
+
+#include <librace.h>
+
+#include "mpmc-queue.h"
+
+void threadA(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+ int32_t *bin = queue->write_prepare();
+ store_32(bin, 1);
+ queue->write_publish();
+}
+
+void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+ int32_t *bin;
+ while (bin = queue->read_fetch()) {
+ printf("Read: %d\n", load_32(bin));
+ queue->read_consume();
+ }
+}
+
+#define MAXREADERS 3
+#define MAXWRITERS 3
+
+int readers = 2, writers = 1;
+
+void print_usage()
+{
+ printf("Error: use the following options\n"
+ " -r <num> Choose number of reader threads\n"
+ " -w <num> Choose number of writer threads\n");
+ exit(EXIT_FAILURE);
+}
+
+void process_params(int argc, char **argv)
+{
+ const char *shortopts = "hr:w:";
+ int opt;
+ bool error = false;
+
+ while (!error && (opt = getopt(argc, argv, shortopts)) != -1) {
+ switch (opt) {
+ case 'h':
+ print_usage();
+ break;
+ case 'r':
+ readers = atoi(optarg);
+ break;
+ case 'w':
+ writers = atoi(optarg);
+ break;
+ default: /* '?' */
+ error = true;
+ break;
+ }
+ }
+
+ if (writers < 1 || writers > MAXWRITERS)
+ error = true;
+ if (readers < 1 || readers > MAXREADERS)
+ error = true;
+
+ if (error)
+ print_usage();
+}
+
+int user_main(int argc, char **argv)
+{
+ struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> queue;
+ thrd_t A[MAXWRITERS], B[MAXREADERS];
+
+ /* Note: optarg() / optind is broken in model-checker - workaround is
+ * to just copy&paste this test a few times */
+ //process_params(argc, argv);
+ printf("%d reader(s), %d writer(s)\n", readers, writers);
+
+ int32_t *bin = queue.write_prepare();
+ store_32(bin, 17);
+ queue.write_publish();
+
+ printf("Start threads\n");
+
+ for (int i = 0; i < writers; i++)
+ thrd_create(&A[i], (thrd_start_t)&threadA, &queue);
+ for (int i = 0; i < readers; i++)
+ thrd_create(&B[i], (thrd_start_t)&threadB, &queue);
+
+ for (int i = 0; i < writers; i++)
+ thrd_join(A[i]);
+ for (int i = 0; i < readers; i++)
+ thrd_join(B[i]);
+
+ printf("Threads complete\n");
+
+ return 0;
+}
diff --git a/mpmc-queue/mpmc-queue.cc b/mpmc-queue/mpmc-queue.cc
index 741811a..d6271be 100644
--- a/mpmc-queue/mpmc-queue.cc
+++ b/mpmc-queue/mpmc-queue.cc
@@ -1,6 +1,8 @@
#include <inttypes.h>
#include <threads.h>
#include <stdio.h>
+#include <unistd.h>
+#include <stdlib.h>
#include <librace.h>
@@ -22,10 +24,60 @@ void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
}
}
+#define MAXREADERS 3
+#define MAXWRITERS 3
+
+int readers = 2, writers = 2;
+
+void print_usage()
+{
+ printf("Error: use the following options\n"
+ " -r <num> Choose number of reader threads\n"
+ " -w <num> Choose number of writer threads\n");
+ exit(EXIT_FAILURE);
+}
+
+void process_params(int argc, char **argv)
+{
+ const char *shortopts = "hr:w:";
+ int opt;
+ bool error = false;
+
+ while (!error && (opt = getopt(argc, argv, shortopts)) != -1) {
+ switch (opt) {
+ case 'h':
+ print_usage();
+ break;
+ case 'r':
+ readers = atoi(optarg);
+ break;
+ case 'w':
+ writers = atoi(optarg);
+ break;
+ default: /* '?' */
+ error = true;
+ break;
+ }
+ }
+
+ if (writers < 1 || writers > MAXWRITERS)
+ error = true;
+ if (readers < 1 || readers > MAXREADERS)
+ error = true;
+
+ if (error)
+ print_usage();
+}
+
int user_main(int argc, char **argv)
{
struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> queue;
- thrd_t A1, A2, B1, B2;
+ thrd_t A[MAXWRITERS], B[MAXREADERS];
+
+ /* Note: optarg() / optind is broken in model-checker - workaround is
+ * to just copy&paste this test a few times */
+ //process_params(argc, argv);
+ printf("%d reader(s), %d writer(s)\n", readers, writers);
int32_t *bin = queue.write_prepare();
store_32(bin, 17);
@@ -33,14 +85,15 @@ int user_main(int argc, char **argv)
printf("Start threads\n");
- thrd_create(&A1, (thrd_start_t)&threadA, &queue);
- thrd_create(&A2, (thrd_start_t)&threadA, &queue);
- thrd_create(&B1, (thrd_start_t)&threadB, &queue);
- thrd_create(&B2, (thrd_start_t)&threadB, &queue);
- thrd_join(A1);
- thrd_join(A2);
- thrd_join(B1);
- thrd_join(B2);
+ for (int i = 0; i < writers; i++)
+ thrd_create(&A[i], (thrd_start_t)&threadA, &queue);
+ for (int i = 0; i < readers; i++)
+ thrd_create(&B[i], (thrd_start_t)&threadB, &queue);
+
+ for (int i = 0; i < writers; i++)
+ thrd_join(A[i]);
+ for (int i = 0; i < readers; i++)
+ thrd_join(B[i]);
printf("Threads complete\n");
--
2.34.1