X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=tests%2Fbenchmark%2Ftest_urcu_workqueue.c;h=4c070861530c08ca2028cd0059abad5144c12d2f;hb=8a2c74fefc366e219bc09c4a4ebd7bacd5cb83e5;hp=9ac235067030a98b995e186d59929268fcb51703;hpb=1abec5a096ef4d74bea3782f7844e1f5cf0e90d6;p=userspace-rcu.git

diff --git a/tests/benchmark/test_urcu_workqueue.c b/tests/benchmark/test_urcu_workqueue.c
index 9ac2350..4c07086 100644
--- a/tests/benchmark/test_urcu_workqueue.c
+++ b/tests/benchmark/test_urcu_workqueue.c
@@ -52,7 +52,7 @@
 #include <urcu/wfstack.h>
 #include <urcu/workqueue-fifo.h>
 
-static volatile int test_go, test_stop_enqueue, test_stop_dequeue;
+static volatile int test_go, test_stop_enqueue;
 
 static unsigned long work_loops;
 
@@ -60,6 +60,8 @@ static unsigned long duration;
 
 static unsigned long dispatch_delay_loops;
 
+static unsigned long max_queue_len;
+
 static inline void loop_sleep(unsigned long loops)
 {
 	while (loops-- != 0)
@@ -119,11 +121,6 @@ static void set_affinity(void)
 /*
  * returns 0 if test should end.
  */
-static int test_duration_dequeue(void)
-{
-	return !test_stop_dequeue;
-}
-
 static int test_duration_enqueue(void)
 {
 	return !test_stop_enqueue;
@@ -158,10 +155,19 @@ static void *thr_dispatcher(void *_count)
 
 	for (;;) {
 		struct test_work *work = malloc(sizeof(*work));
+		enum urcu_enqueue_ret ret;
+
 		if (!work)
 			goto fail;
-		printf_verbose("queue work %p\n", work);
-		urcu_queue_work(&workqueue, &work->w);
+retry:
+		printf_verbose("attempt queue work %p\n", work);
+		ret = urcu_queue_work(&workqueue, &work->w);
+		if (ret == URCU_ENQUEUE_FULL) {
+			printf_verbose("queue work %p (queue full)\n", work);
+			(void) poll(NULL, 0, 10);
+			goto retry;
+		}
+		printf_verbose("queue work %p (ok)\n", work);
 		URCU_TLS(nr_enqueues)++;
 
 		if (caa_unlikely(dispatch_delay_loops))
@@ -185,7 +191,6 @@ static void *thr_worker(void *_count)
 	unsigned long long *count = _count;
 	unsigned int counter = 0;
 	struct urcu_worker worker;
-	int blocking = 1;
 
 	printf_verbose("thread_begin %s, tid %lu\n",
 			"worker", urcu_get_thread_id());
@@ -193,8 +198,8 @@ static void *thr_worker(void *_count)
 	set_affinity();
 
 	rcu_register_thread();
-	urcu_worker_init(&worker, URCU_WORKER_STEAL);
-	//urcu_worker_init(&worker, 0);
+	urcu_worker_init(&workqueue, &worker, URCU_WORKER_STEAL);
+	//urcu_worker_init(&workqueue, &worker, 0);
 	urcu_worker_register(&workqueue, &worker);
 
 	while (!test_go)
@@ -203,9 +208,11 @@ static void *thr_worker(void *_count)
 	cmm_smp_mb();
 
 	for (;;) {
-		int batch_work_count = 0;
+		enum urcu_accept_ret ret;
 
-		urcu_accept_work(&workqueue, &worker, blocking);
+		ret = urcu_accept_work(&worker);
+		if (ret == URCU_ACCEPT_SHUTDOWN)
+			break;
 		for (;;) {
 			struct urcu_work *work;
 			struct test_work *t;
@@ -215,17 +222,11 @@ static void *thr_worker(void *_count)
 				break;
 			t = caa_container_of(work, struct test_work, w);
 			printf_verbose("dequeue work %p\n", t);
-			batch_work_count++;
 			URCU_TLS(nr_dequeues)++;
 			if (caa_unlikely(work_loops))
 				loop_sleep(work_loops);
 			free(t);
 		}
-		if (!test_duration_dequeue())
-			blocking = 0;
-		if (caa_unlikely(!test_duration_dequeue()
-				&& !batch_work_count))
-			break;
 	}
 end:
 	urcu_worker_unregister(&workqueue, &worker);
@@ -249,6 +250,7 @@ static void show_usage(int argc, char **argv)
 	printf("	[-v] (verbose output)\n");
 	printf("	[-a cpu#] [-a cpu#]... (affinity)\n");
 	printf("	[-w] Wait for worker to empty stack\n");
+	printf("	[-m len] (Max queue length. 0 means infinite.))\n");
 	printf("\n");
 }
 
@@ -299,6 +301,13 @@ int main(int argc, char **argv)
 			use_affinity = 1;
 			printf_verbose("Adding CPU %d affinity\n", a);
 			break;
+		case 'm':
+			if (argc < i + 2) {
+				show_usage(argc, argv);
+				return -1;
+			}
+			max_queue_len = atol(argv[++i]);
+			break;
 		case 'c':
 			if (argc < i + 2) {
 				show_usage(argc, argv);
@@ -336,7 +345,7 @@ int main(int argc, char **argv)
 	tid_worker = calloc(nr_workers, sizeof(*tid_worker));
 	count_dispatcher = calloc(nr_dispatchers, sizeof(*count_dispatcher));
 	count_worker = calloc(nr_workers, sizeof(*count_worker));
-	urcu_workqueue_init(&workqueue);
+	urcu_workqueue_init(&workqueue, max_queue_len);
 
 	next_aff = 0;
 
@@ -373,10 +382,7 @@ int main(int argc, char **argv)
 			sleep(1);
 		}
 	}
-	test_stop_dequeue = 1;
-
-	/* Send finish to all workers */
-	urcu_workqueue_wakeup_all(&workqueue);
+	urcu_workqueue_shutdown(&workqueue);
 
 	for (i = 0; i < nr_dispatchers; i++) {
 		err = pthread_join(tid_dispatcher[i], &tret);
@@ -393,9 +399,9 @@ int main(int argc, char **argv)
 
 	printf("SUMMARY %-25s testdur %4lu nr_dispatchers %3u dispatch_delay_loops %6lu "
 		"work_loops %lu nr_workers %3u "
-		"nr_enqueues %12llu nr_dequeues %12llu\n",
+		"nr_enqueues %12llu nr_dequeues %12llu max_queue_len %lu\n",
 		argv[0], duration, nr_dispatchers, dispatch_delay_loops, work_loops,
-		nr_workers, tot_enqueues, tot_dequeues);
+		nr_workers, tot_enqueues, tot_dequeues, max_queue_len);
 	free(count_dispatcher);
 	free(count_worker);
 	free(tid_dispatcher);