aboutsummaryrefslogtreecommitdiffhomepage
path: root/test/core/support/sync_test.c
diff options
context:
space:
mode:
Diffstat (limited to 'test/core/support/sync_test.c')
-rw-r--r--test/core/support/sync_test.c451
1 files changed, 451 insertions, 0 deletions
diff --git a/test/core/support/sync_test.c b/test/core/support/sync_test.c
new file mode 100644
index 0000000000..93f9c4cc8d
--- /dev/null
+++ b/test/core/support/sync_test.c
@@ -0,0 +1,451 @@
+/*
+ *
+ * Copyright 2014, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/* Test of gpr synchronization support. */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/thd.h>
+#include <grpc/support/time.h>
+#include "test/core/util/test_config.h"
+
+/* ==================Example use of interface===================
+
+ A producer-consumer queue of up to N integers,
+ illustrating the use of the calls in this interface. */
+
+#define N 4
+
+typedef struct queue {
+ gpr_cv non_empty; /* Signalled when length becomes non-zero. */
+ gpr_cv non_full; /* Signalled when length becomes non-N. */
+ gpr_mu mu; /* Protects all fields below.
+ (That is, except during initialization or
+ destruction, the fields below should be accessed
+ only by a thread that holds mu.) */
+ int head; /* Index of head of queue 0..N-1. */
+ int length; /* Number of valid elements in queue 0..N. */
+ int elem[N]; /* elem[head .. head+length-1] are queue elements. */
+} queue;
+
+/* Initialize *q. */
+void queue_init(queue *q) {
+ gpr_mu_init(&q->mu);
+ gpr_cv_init(&q->non_empty);
+ gpr_cv_init(&q->non_full);
+ q->head = 0;
+ q->length = 0;
+}
+
+/* Free storage associated with *q. */
+void queue_destroy(queue *q) {
+ gpr_mu_destroy(&q->mu);
+ gpr_cv_destroy(&q->non_empty);
+ gpr_cv_destroy(&q->non_full);
+}
+
+/* Wait until there is room in *q, then append x to *q. */
+void queue_append(queue *q, int x) {
+ gpr_mu_lock(&q->mu);
+ /* To wait for a predicate without a deadline, loop on the negation of the
+ predicate, and use gpr_cv_wait(..., gpr_inf_future) inside the loop
+ to release the lock, wait, and reacquire on each iteration. Code that
+ makes the condition true should use gpr_cv_broadcast() on the
+ corresponding condition variable. The predicate must be on state
+ protected by the lock. */
+ while (q->length == N) {
+ gpr_cv_wait(&q->non_full, &q->mu, gpr_inf_future);
+ }
+ if (q->length == 0) { /* Wake threads blocked in queue_remove(). */
+ /* It's normal to use gpr_cv_broadcast() or gpr_signal() while
+ holding the lock. */
+ gpr_cv_broadcast(&q->non_empty);
+ }
+ q->elem[(q->head + q->length) % N] = x;
+ q->length++;
+ gpr_mu_unlock(&q->mu);
+}
+
+/* If it can be done without blocking, append x to *q and return non-zero.
+ Otherwise return 0. */
+int queue_try_append(queue *q, int x) {
+ int result = 0;
+ if (gpr_mu_trylock(&q->mu)) {
+ if (q->length != N) {
+ if (q->length == 0) { /* Wake threads blocked in queue_remove(). */
+ gpr_cv_broadcast(&q->non_empty);
+ }
+ q->elem[(q->head + q->length) % N] = x;
+ q->length++;
+ result = 1;
+ }
+ gpr_mu_unlock(&q->mu);
+ }
+ return result;
+}
+
+/* Wait until the *q is non-empty or deadline abs_deadline passes. If the
+ queue is non-empty, remove its head entry, place it in *head, and return
+ non-zero. Otherwise return 0. */
+int queue_remove(queue *q, int *head, gpr_timespec abs_deadline) {
+ int result = 0;
+ gpr_mu_lock(&q->mu);
+ /* To wait for a predicate with a deadline, loop on the negation of the
+ predicate or until gpr_cv_wait() returns true. Code that makes
+ the condition true should use gpr_cv_broadcast() on the corresponding
+ condition variable. The predicate must be on state protected by the
+ lock. */
+ while (q->length == 0 && !gpr_cv_wait(&q->non_empty, &q->mu, abs_deadline)) {
+ }
+ if (q->length != 0) { /* Queue is non-empty. */
+ result = 1;
+ if (q->length == N) { /* Wake threads blocked in queue_append(). */
+ gpr_cv_broadcast(&q->non_full);
+ }
+ *head = q->elem[q->head];
+ q->head = (q->head + 1) % N;
+ q->length--;
+ } /* else deadline exceeded */
+ gpr_mu_unlock(&q->mu);
+ return result;
+}
+
+/* ------------------------------------------------- */
+/* Tests for gpr_mu and gpr_cv, and the queue example. */
+struct test {
+ int threads; /* number of threads */
+
+ gpr_int64 iterations; /* number of iterations per thread */
+ gpr_int64 counter;
+ int thread_count; /* used to allocate thread ids */
+ int done; /* threads not yet completed */
+
+ gpr_mu mu; /* protects iterations, counter, thread_count, done */
+
+ gpr_cv cv; /* signalling depends on test */
+
+ gpr_cv done_cv; /* signalled when done == 0 */
+
+ queue q;
+
+ gpr_stats_counter stats_counter;
+
+ gpr_refcount refcount;
+ gpr_refcount thread_refcount;
+ gpr_event event;
+};
+
+/* Return pointer to a new struct test. */
+static struct test *test_new(int threads, gpr_int64 iterations) {
+ struct test *m = gpr_malloc(sizeof(*m));
+ m->threads = threads;
+ m->iterations = iterations;
+ m->counter = 0;
+ m->thread_count = 0;
+ m->done = threads;
+ gpr_mu_init(&m->mu);
+ gpr_cv_init(&m->cv);
+ gpr_cv_init(&m->done_cv);
+ queue_init(&m->q);
+ gpr_stats_init(&m->stats_counter, 0);
+ gpr_ref_init(&m->refcount, 0);
+ gpr_ref_init(&m->thread_refcount, threads);
+ gpr_event_init(&m->event);
+ return m;
+}
+
+/* Return pointer to a new struct test. */
+static void test_destroy(struct test *m) {
+ gpr_mu_destroy(&m->mu);
+ gpr_cv_destroy(&m->cv);
+ gpr_cv_destroy(&m->done_cv);
+ queue_destroy(&m->q);
+ gpr_free(m);
+}
+
+/* Create m->threads threads, each running (*body)(m) */
+static void test_create_threads(struct test *m, void (*body)(void *arg)) {
+ gpr_thd_id id;
+ int i;
+ for (i = 0; i != m->threads; i++) {
+ GPR_ASSERT(gpr_thd_new(&id, body, m, NULL));
+ }
+}
+
+/* Wait until all threads report done. */
+static void test_wait(struct test *m) {
+ gpr_mu_lock(&m->mu);
+ while (m->done != 0) {
+ gpr_cv_wait(&m->done_cv, &m->mu, gpr_inf_future);
+ }
+ gpr_mu_unlock(&m->mu);
+}
+
+/* Get an integer thread id in the raneg 0..threads-1 */
+static int thread_id(struct test *m) {
+ int id;
+ gpr_mu_lock(&m->mu);
+ id = m->thread_count++;
+ gpr_mu_unlock(&m->mu);
+ return id;
+}
+
+/* Indicate that a thread is done, by decrementing m->done
+ and signalling done_cv if m->done==0. */
+static void mark_thread_done(struct test *m) {
+ gpr_mu_lock(&m->mu);
+ GPR_ASSERT(m->done != 0);
+ m->done--;
+ if (m->done == 0) {
+ gpr_cv_signal(&m->done_cv);
+ }
+ gpr_mu_unlock(&m->mu);
+}
+
+/* Test several threads running (*body)(struct test *m) for increasing settings
+ of m->iterations, until about timeout_s to 2*timeout_s seconds have elapsed.
+ If extra!=NULL, run (*extra)(m) in an additional thread. */
+static void test(const char *name, void (*body)(void *m),
+ void (*extra)(void *m), int timeout_s) {
+ gpr_int64 iterations = 1024;
+ struct test *m;
+ gpr_timespec start = gpr_now();
+ gpr_timespec time_taken;
+ gpr_timespec deadline =
+ gpr_time_add(start, gpr_time_from_micros(timeout_s * 1000000));
+ fprintf(stderr, "%s:", name);
+ while (gpr_time_cmp(gpr_now(), deadline) < 0) {
+ iterations <<= 1;
+ fprintf(stderr, " %ld", (long)iterations);
+ m = test_new(10, iterations);
+ if (extra != NULL) {
+ gpr_thd_id id;
+ GPR_ASSERT(gpr_thd_new(&id, extra, m, NULL));
+ m->done++; /* one more thread to wait for */
+ }
+ test_create_threads(m, body);
+ test_wait(m);
+ if (m->counter != m->threads * m->iterations) {
+ fprintf(stderr, "counter %ld threads %d iterations %ld\n",
+ (long)m->counter, m->threads, (long)m->iterations);
+ GPR_ASSERT(0);
+ }
+ test_destroy(m);
+ }
+ time_taken = gpr_time_sub(gpr_now(), start);
+ fprintf(stderr, " done %ld.%09d s\n", (long)time_taken.tv_sec,
+ (int)time_taken.tv_nsec);
+}
+
+/* Increment m->counter on each iteration; then mark thread as done. */
+static void inc(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ for (i = 0; i != m->iterations; i++) {
+ gpr_mu_lock(&m->mu);
+ m->counter++;
+ gpr_mu_unlock(&m->mu);
+ }
+ mark_thread_done(m);
+}
+
+/* Increment m->counter under lock acquired with trylock, m->iterations times;
+ then mark thread as done. */
+static void inctry(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ for (i = 0; i != m->iterations;) {
+ if (gpr_mu_trylock(&m->mu)) {
+ m->counter++;
+ gpr_mu_unlock(&m->mu);
+ i++;
+ }
+ }
+ mark_thread_done(m);
+}
+
+/* Increment counter only when (m->counter%m->threads)==m->thread_id; then mark
+ thread as done. */
+static void inc_by_turns(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ int id = thread_id(m);
+ for (i = 0; i != m->iterations; i++) {
+ gpr_mu_lock(&m->mu);
+ while ((m->counter % m->threads) != id) {
+ gpr_cv_wait(&m->cv, &m->mu, gpr_inf_future);
+ }
+ m->counter++;
+ gpr_cv_broadcast(&m->cv);
+ gpr_mu_unlock(&m->mu);
+ }
+ mark_thread_done(m);
+}
+
+/* Wait a millisecond and increment counter on each iteration;
+ then mark thread as done. */
+static void inc_with_1ms_delay(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ for (i = 0; i != m->iterations; i++) {
+ gpr_timespec deadline;
+ gpr_mu_lock(&m->mu);
+ deadline = gpr_time_add(gpr_now(), gpr_time_from_micros(1000));
+ while (!gpr_cv_wait(&m->cv, &m->mu, deadline)) {
+ }
+ m->counter++;
+ gpr_mu_unlock(&m->mu);
+ }
+ mark_thread_done(m);
+}
+
+/* Wait a millisecond and increment counter on each iteration, using an event
+ for timing; then mark thread as done. */
+static void inc_with_1ms_delay_event(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ for (i = 0; i != m->iterations; i++) {
+ gpr_timespec deadline;
+ deadline = gpr_time_add(gpr_now(), gpr_time_from_micros(1000));
+ GPR_ASSERT(gpr_event_wait(&m->event, deadline) == NULL);
+ gpr_mu_lock(&m->mu);
+ m->counter++;
+ gpr_mu_unlock(&m->mu);
+ }
+ mark_thread_done(m);
+}
+
+/* Produce m->iterations elements on queue m->q, then mark thread as done.
+ Even threads use queue_append(), and odd threads use queue_try_append()
+ until it succeeds. */
+static void many_producers(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ int x = thread_id(m);
+ if ((x & 1) == 0) {
+ for (i = 0; i != m->iterations; i++) {
+ queue_append(&m->q, 1);
+ }
+ } else {
+ for (i = 0; i != m->iterations; i++) {
+ while (!queue_try_append(&m->q, 1)) {
+ }
+ }
+ }
+ mark_thread_done(m);
+}
+
+/* Consume elements from m->q until m->threads*m->iterations are seen,
+ wait an extra second to confirm that no more elements are arriving,
+ then mark thread as done. */
+static void consumer(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 n = m->iterations * m->threads;
+ gpr_int64 i;
+ int value;
+ for (i = 0; i != n; i++) {
+ queue_remove(&m->q, &value, gpr_inf_future);
+ }
+ gpr_mu_lock(&m->mu);
+ m->counter = n;
+ gpr_mu_unlock(&m->mu);
+ GPR_ASSERT(
+ !queue_remove(&m->q, &value,
+ gpr_time_add(gpr_now(), gpr_time_from_micros(1000000))));
+ mark_thread_done(m);
+}
+
+/* Increment m->stats_counter m->iterations times, transfer counter value to
+ m->counter, then mark thread as done. */
+static void statsinc(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ for (i = 0; i != m->iterations; i++) {
+ gpr_stats_inc(&m->stats_counter, 1);
+ }
+ gpr_mu_lock(&m->mu);
+ m->counter = gpr_stats_read(&m->stats_counter);
+ gpr_mu_unlock(&m->mu);
+ mark_thread_done(m);
+}
+
+/* Increment m->refcount m->iterations times, decrement m->thread_refcount
+ once, and if it reaches zero, set m->event to (void*)1; then mark thread as
+ done. */
+static void refinc(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 i;
+ for (i = 0; i != m->iterations; i++) {
+ gpr_ref(&m->refcount);
+ }
+ if (gpr_unref(&m->thread_refcount)) {
+ gpr_event_set(&m->event, (void *)1);
+ }
+ mark_thread_done(m);
+}
+
+/* Wait until m->event is set to (void *)1, then decrement m->refcount
+ m->stats_counter m->iterations times, and ensure that the last decrement
+ caused the counter to reach zero, then mark thread as done. */
+static void refcheck(void *v /*=m*/) {
+ struct test *m = v;
+ gpr_int64 n = m->iterations * m->threads;
+ gpr_int64 i;
+ GPR_ASSERT(gpr_event_wait(&m->event, gpr_inf_future) == (void *)1);
+ GPR_ASSERT(gpr_event_get(&m->event) == (void *)1);
+ for (i = 1; i != n; i++) {
+ GPR_ASSERT(!gpr_unref(&m->refcount));
+ m->counter++;
+ }
+ GPR_ASSERT(gpr_unref(&m->refcount));
+ m->counter++;
+ mark_thread_done(m);
+}
+
+/* ------------------------------------------------- */
+
+int main(int argc, char *argv[]) {
+ grpc_test_init(argc, argv);
+ test("mutex", &inc, NULL, 1);
+ test("mutex try", &inctry, NULL, 1);
+ test("cv", &inc_by_turns, NULL, 1);
+ test("timedcv", &inc_with_1ms_delay, NULL, 1);
+ test("queue", &many_producers, &consumer, 10);
+ test("stats_counter", &statsinc, NULL, 1);
+ test("refcount", &refinc, &refcheck, 1);
+ test("timedevent", &inc_with_1ms_delay_event, NULL, 1);
+ return 0;
+}
generated by cgit v1.2.3 (git 2.39.1) at 2024-05-24 12:38:28 +0000