X-Git-Url: http://demsky.eecs.uci.edu/git/?a=blobdiff_plain;f=mpmc-queue%2Fmpmc-queue.cc;h=7de14d589c5819b9055618928eae43accc2f15bd;hb=5ab6d890042234648c0baec333ed65e90ca055cf;hp=53bc6134ea028c4e9e6d3addcd1d8685922eb3d1;hpb=ca8555d0e02fb0b6e37bfbff4adee760c2248efb;p=model-checker-benchmarks.git

diff --git a/mpmc-queue/mpmc-queue.cc b/mpmc-queue/mpmc-queue.cc
index 53bc613..7de14d5 100644
--- a/mpmc-queue/mpmc-queue.cc
+++ b/mpmc-queue/mpmc-queue.cc
@@ -18,7 +18,19 @@ void threadA(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
 void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
 {
 	int32_t *bin;
-	while (bin = queue->read_fetch()) {
+	while ((bin = queue->read_fetch()) != NULL) {
+		printf("Read: %d\n", load_32(bin));
+		queue->read_consume();
+	}
+}
+
+void threadC(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
+{
+	int32_t *bin = queue->write_prepare();
+	store_32(bin, 1);
+	queue->write_publish();
+
+	while ((bin = queue->read_fetch()) != NULL) {
 		printf("Read: %d\n", load_32(bin));
 		queue->read_consume();
 	}
@@ -26,6 +38,7 @@ void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
 
 #define MAXREADERS 3
 #define MAXWRITERS 3
+#define MAXRDWR 3
 
 #ifdef CONFIG_MPMC_READERS
 #define DEFAULT_READERS (CONFIG_MPMC_READERS)
@@ -39,7 +52,13 @@ void threadB(struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> *queue)
 #define DEFAULT_WRITERS 2
 #endif
 
-int readers = DEFAULT_READERS, writers = DEFAULT_WRITERS;
+#ifdef CONFIG_MPMC_RDWR
+#define DEFAULT_RDWR (CONFIG_MPMC_RDWR)
+#else
+#define DEFAULT_RDWR 0
+#endif
+
+int readers = DEFAULT_READERS, writers = DEFAULT_WRITERS, rdwr = DEFAULT_RDWR;
 
 void print_usage()
 {
@@ -84,16 +103,19 @@ void process_params(int argc, char **argv)
 int user_main(int argc, char **argv)
 {
 	struct mpmc_boundq_1_alt<int32_t, sizeof(int32_t)> queue;
-	thrd_t A[MAXWRITERS], B[MAXREADERS];
+	thrd_t A[MAXWRITERS], B[MAXREADERS], C[MAXRDWR];
 
 	/* 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);
 
+#ifndef CONFIG_MPMC_NO_INITIAL_ELEMENT
+	printf("Adding initial element\n");
 	int32_t *bin = queue.write_prepare();
 	store_32(bin, 17);
 	queue.write_publish();
+#endif
 
 	printf("Start threads\n");
 
@@ -102,10 +124,15 @@ int user_main(int argc, char **argv)
 	for (int i = 0; i < readers; i++)
 		thrd_create(&B[i], (thrd_start_t)&threadB, &queue);
 
+	for (int i = 0; i < rdwr; i++)
+		thrd_create(&C[i], (thrd_start_t)&threadC, &queue);
+
 	for (int i = 0; i < writers; i++)
 		thrd_join(A[i]);
 	for (int i = 0; i < readers; i++)
 		thrd_join(B[i]);
+	for (int i = 0; i < rdwr; i++)
+		thrd_join(C[i]);
 
 	printf("Threads complete\n");