You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kg...@apache.org on 2021/05/03 14:35:37 UTC
[qpid-dispatch] branch main updated: DISPATCH-1679: Make
qd_timer_cancel and qd_timer_free synchronous
This is an automated email from the ASF dual-hosted git repository.
kgiusti pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
The following commit(s) were added to refs/heads/main by this push:
new 34b3c11 DISPATCH-1679: Make qd_timer_cancel and qd_timer_free synchronous
34b3c11 is described below
commit 34b3c116dc97f9a1d7caf00fcea0e5e5144e6c45
Author: Kenneth Giusti <kg...@apache.org>
AuthorDate: Wed Jun 10 14:25:49 2020 -0400
DISPATCH-1679: Make qd_timer_cancel and qd_timer_free synchronous
This patch makes the qd_timer_cancel() and qd_timer_free() routines
block if the timer callback is currently running on another thread.
This ensures that the callback is not executing when these calls
return.
It includes a fix to the HTTP1 adapter to avoid freeing its timer
while holding the adaptor lock.
This closes #1170
---
include/qpid/dispatch/timer.h | 41 +++-
src/adaptors/http1/http1_adaptor.c | 10 +-
src/adaptors/http1/http1_server.c | 29 +--
src/timer.c | 165 +++++++++++--
src/timer_private.h | 14 +-
tests/CMakeLists.txt | 6 +
tests/dummy.conf | 28 +++
tests/threaded_timer_test.c | 473 +++++++++++++++++++++++++++++++++++++
tests/tsan.supp | 3 +-
9 files changed, 704 insertions(+), 65 deletions(-)
diff --git a/include/qpid/dispatch/timer.h b/include/qpid/dispatch/timer.h
index e9c9511..1d84d94 100644
--- a/include/qpid/dispatch/timer.h
+++ b/include/qpid/dispatch/timer.h
@@ -42,7 +42,8 @@ typedef int64_t qd_duration_t;
/**
* Timer Callback
*
- * Callback invoked after a timer's interval expires and the timer fires.
+ * Callback invoked after a timer's interval expires and the timer fires. This
+ * may be invoked on any server thread.
*
* @param context The context supplied in qd_timer
*/
@@ -51,7 +52,7 @@ typedef void (*qd_timer_cb_t)(void* context);
/**
* Create a new timer object.
- *
+ *
* @param qd Pointer to the dispatch instance.
* @param cb The callback function to be invoked when the timer expires.
* @param context An opaque, user-supplied context to be passed into the callback.
@@ -61,9 +62,17 @@ qd_timer_t *qd_timer(qd_dispatch_t *qd, qd_timer_cb_t cb, void* context);
/**
- * Free the resources for a timer object. If the timer was scheduled, it will be canceled
- * prior to freeing. After this function returns, the callback will not be invoked for this
- * timer.
+ * Free a timer.
+ *
+ * If the timer was scheduled, it will be canceled prior to freeing. If the
+ * callback is currently running on another server thread the caller will block
+ * until the callback completes. On return the caller is guaranteed that the
+ * timer callback will never run again. This may be called during the timer
+ * callback - in that case the call will not block.
+ *
+ * Note Well:
+ * Since this call may block until the timer callback completes, it is critical
+ * that the caller MUST NOT be holding any lock that the timer callback takes.
*
* @param timer Pointer to the timer object returned by qd_timer.
*/
@@ -73,9 +82,10 @@ void qd_timer_free(qd_timer_t *timer);
/**
* Schedule a timer to fire in the future.
*
- * Note that the timer callback will never be invoked synchronously during the execution
- * of qd_timer_schedule. Even if the interval is immediate (0), the callback invocation will
- * be asynchronous and after the return of this function.
+ * Note that the timer callback will never be invoked synchronously during the
+ * execution of qd_timer_schedule. Even if the interval is immediate (0), the
+ * callback invocation will be asynchronous and after the return of this
+ * function. This may be called during the timer callback.
*
* @param timer Pointer to the timer object returned by qd_timer.
* @param msec The minimum number of milliseconds of delay until the timer fires.
@@ -85,9 +95,18 @@ void qd_timer_schedule(qd_timer_t *timer, qd_duration_t msec);
/**
- * Attempt to cancel a scheduled timer. Since the timer callback can be invoked on any
- * server thread, it is always possible that a last-second cancel attempt may arrive too late
- * to stop the timer from firing (i.e. the cancel is concurrent with the fire callback).
+ * Cancel a scheduled timer.
+ *
+ * If the timer is scheduled it will be canceled and the callback will not be
+ * invoked. If the timer callback is currently executing on another thread
+ * this call will block until the callback completes. On return the caller is
+ * guaranteed that the callback will not execute unless it is rescheduled using
+ * qd_timer_schedule. This call must not be invoked from within the callback
+ * itself.
+ *
+ * Note Well:
+ * Since this call may block until the timer callback completes, it is critical
+ * that the caller MUST NOT be holding any lock that the timer callback takes.
*
* @param timer Pointer to the timer object returned by qd_timer.
*/
diff --git a/src/adaptors/http1/http1_adaptor.c b/src/adaptors/http1/http1_adaptor.c
index 8536131..0903529 100644
--- a/src/adaptors/http1/http1_adaptor.c
+++ b/src/adaptors/http1/http1_adaptor.c
@@ -82,16 +82,16 @@ void qdr_http1_connection_free(qdr_http1_connection_t *hconn)
{
if (hconn) {
pn_raw_connection_t *rconn = 0;
+ qd_timer_t *timer = 0;
// prevent core from activating this connection while it is being torn
// down. Also prevent timer callbacks from running. see
- // _core_connection_activate_CT, and _do_reconnect/_do_activate in
- // http1_server.c
+ // _core_connection_activate_CT and _do_reconnect in http1_server.c
//
sys_mutex_lock(qdr_http1_adaptor->lock);
{
DEQ_REMOVE(qdr_http1_adaptor->connections, hconn);
- qd_timer_free(hconn->server.reconnect_timer);
+ timer = hconn->server.reconnect_timer;
hconn->server.reconnect_timer = 0;
rconn = hconn->raw_conn;
hconn->raw_conn = 0;
@@ -105,6 +105,10 @@ void qdr_http1_connection_free(qdr_http1_connection_t *hconn)
}
sys_mutex_unlock(qdr_http1_adaptor->lock);
+ // must free timer outside of lock since callback
+ // attempts to take lock:
+ qd_timer_free(timer);
+
// cleanup outstanding requests
//
if (hconn->type == HTTP1_CONN_SERVER)
diff --git a/src/adaptors/http1/http1_server.c b/src/adaptors/http1/http1_server.c
index fb7188f..cdc137a 100644
--- a/src/adaptors/http1/http1_server.c
+++ b/src/adaptors/http1/http1_server.c
@@ -416,24 +416,21 @@ static void _do_reconnect(void *context)
}
sys_mutex_unlock(qdr_http1_adaptor->lock);
- if (hconn->qdr_conn) {
-
- // handle any qdr_connection_t processing requests that occurred since
- // this raw connection dropped.
- while (qdr_connection_process(hconn->qdr_conn))
- ;
-
- if (!hconn->qdr_conn) {
- // the qdr_connection_t has been closed
- qd_log(qdr_http1_adaptor->log, QD_LOG_DEBUG,
- "[C%"PRIu64"] HTTP/1.x server connection closed", hconn->conn_id);
- qdr_http1_connection_free(hconn);
- return;
- }
+ // handle any qdr_connection_t processing requests that occurred since
+ // this raw connection dropped.
+ while (hconn->qdr_conn && qdr_connection_process(hconn->qdr_conn))
+ ;
- _process_request((_server_request_t*) DEQ_HEAD(hconn->requests));
+ if (!hconn->qdr_conn) {
+ // the qdr_connection_t has been closed
+ qd_log(qdr_http1_adaptor->log, QD_LOG_DEBUG,
+ "[C%"PRIu64"] HTTP/1.x server connection closed", hconn->conn_id);
+ qdr_http1_connection_free(hconn);
+ return;
}
+ _process_request((_server_request_t*) DEQ_HEAD(hconn->requests));
+
// Do not attempt to re-connect if the current request is still in
// progress. This happens when the server has closed the connection before
// the request message has fully arrived (!rx_complete).
@@ -597,7 +594,7 @@ static void _handle_connection_events(pn_event_t *e, qd_server_t *qd_server, voi
// prevent core activation
sys_mutex_lock(qdr_http1_adaptor->lock);
hconn->raw_conn = 0;
- if (reconnect)
+ if (reconnect && hconn->server.reconnect_timer)
qd_timer_schedule(hconn->server.reconnect_timer, hconn->server.reconnect_pause);
sys_mutex_unlock(qdr_http1_adaptor->lock);
diff --git a/src/timer.c b/src/timer.c
index 9041f93..71698d1 100644
--- a/src/timer.c
+++ b/src/timer.c
@@ -25,15 +25,59 @@
#include "qpid/dispatch/alloc.h"
#include "qpid/dispatch/ctools.h"
-#include "qpid/dispatch/threading.h"
+#include "qpid/dispatch/atomic.h"
#include <assert.h>
#include <stdio.h>
#include <time.h>
+
+// timer state machine
+//
+// IDLE: initial state or state immediately after the callback finishes
+// running.
+//
+// SCHEDULED: timer has been scheduled to run. It is on the scheduled_timer
+// list. Valid next states are IDLE (if timer canceled), RUNNING, or DELETED.
+//
+// RUNNING: the timer callback is executing. Valid next states are IDLE,
+// SCHEDULED, BLOCKED or DELETED. The address of the thread executing the
+// callback is available in callback_thread
+//
+// BLOCKED: the callback is executing and another thread is blocked waiting for
+// the callback to complete. This state is used when cancelling or freeing a
+// running timer.
+//
+// DELETED: final state. There are no valid next states.
+//
+typedef enum {
+ QD_TIMER_STATE_IDLE,
+ QD_TIMER_STATE_SCHEDULED,
+ QD_TIMER_STATE_RUNNING,
+ QD_TIMER_STATE_BLOCKED,
+ QD_TIMER_STATE_DELETED
+} qd_timer_state_t;
+
+
+struct qd_timer_t {
+ DEQ_LINKS(qd_timer_t);
+ qd_server_t *server;
+ qd_timer_cb_t handler;
+ void *context;
+ sys_cond_t *condition;
+ sys_atomic_t ref_count; // referenced by user and when on scheduled list
+ qd_timestamp_t delta_time;
+ qd_timer_state_t state;
+};
+
+DEQ_DECLARE(qd_timer_t, qd_timer_list_t);
+
static sys_mutex_t *lock = NULL;
-static qd_timer_list_t idle_timers = {0};
static qd_timer_list_t scheduled_timers = {0};
+
+// thread currently running timer callbacks or 0 if callbacks are not running
+static sys_thread_t *callback_thread = 0;
+
/* Timers have relative delta_time measured from the previous timer.
* The delta_time of the first timer on the queue is measured from timer_base.
*/
@@ -49,17 +93,20 @@ sys_mutex_t* qd_timer_lock() { return lock; }
// Private static functions
//=========================================================================
-static void timer_cancel_LH(qd_timer_t *timer)
+// returns true if timer removed from scheduled list
+static bool timer_cancel_LH(qd_timer_t *timer)
{
- if (timer->scheduled) {
+ if (timer->state == QD_TIMER_STATE_SCHEDULED) {
if (timer->next)
timer->next->delta_time += timer->delta_time;
DEQ_REMOVE(scheduled_timers, timer);
- DEQ_INSERT_TAIL(idle_timers, timer);
- timer->scheduled = false;
+ timer->state = QD_TIMER_STATE_IDLE;
+ return true;
}
+ return false;
}
+
/* Adjust timer's time_base and delays for the current time. */
static void timer_adjust_now_LH()
{
@@ -81,6 +128,18 @@ static void timer_adjust_now_LH()
}
+static void timer_decref_LH(qd_timer_t *timer)
+{
+ assert(sys_atomic_get(&timer->ref_count) > 0);
+ if (sys_atomic_dec(&timer->ref_count) == 1) {
+ assert(timer->state != QD_TIMER_STATE_SCHEDULED);
+ sys_cond_free(timer->condition);
+ sys_atomic_destroy(&timer->ref_count);
+ free_qd_timer_t(timer);
+ }
+}
+
+
//=========================================================================
// Public Functions from timer.h
//=========================================================================
@@ -92,16 +151,21 @@ qd_timer_t *qd_timer(qd_dispatch_t *qd, qd_timer_cb_t cb, void* context)
if (!timer)
return 0;
+ sys_cond_t *cond = sys_cond();
+ if (!cond) {
+ free_qd_timer_t(timer);
+ return 0;
+ }
+
DEQ_ITEM_INIT(timer);
timer->server = qd ? qd->server : 0;
timer->handler = cb;
timer->context = context;
timer->delta_time = 0;
- timer->scheduled = false;
- sys_mutex_lock(lock);
- DEQ_INSERT_TAIL(idle_timers, timer);
- sys_mutex_unlock(lock);
+ timer->condition = cond;
+ timer->state = QD_TIMER_STATE_IDLE;
+ sys_atomic_init(&timer->ref_count, 1);
return timer;
}
@@ -110,16 +174,37 @@ qd_timer_t *qd_timer(qd_dispatch_t *qd, qd_timer_cb_t cb, void* context)
void qd_timer_free(qd_timer_t *timer)
{
if (!timer) return;
+
sys_mutex_lock(lock);
- timer_cancel_LH(timer);
- DEQ_REMOVE(idle_timers, timer);
+
+ assert(timer->state != QD_TIMER_STATE_DELETED); // double free!!!
+
+ if (timer->state == QD_TIMER_STATE_RUNNING) {
+ if (sys_thread_self() != callback_thread) {
+ // Another thread is running the callback (see qd_timer_visit())
+ // Wait until the callback finishes
+ timer->state = QD_TIMER_STATE_BLOCKED;
+ sys_cond_wait(timer->condition, lock);
+ }
+ }
+
+ // we can safely free the timer since the callback is not running
+
+ if (timer_cancel_LH(timer)) {
+ // removed from scheduled_timers, so drop ref_count
+ assert(sys_atomic_get(&timer->ref_count) > 1); // expect caller holds a ref_count
+ timer_decref_LH(timer);
+ }
+
+ timer->state = QD_TIMER_STATE_DELETED;
+ timer_decref_LH(timer); // now drop caller ref_count
sys_mutex_unlock(lock);
- free_qd_timer_t(timer);
}
__attribute__((noinline)) // permit replacement by dummy implementation in unit_tests
-qd_timestamp_t qd_timer_now() {
+qd_timestamp_t qd_timer_now()
+{
struct timespec tv;
clock_gettime(CLOCK_MONOTONIC, &tv);
return ((qd_timestamp_t)tv.tv_sec) * 1000 + tv.tv_nsec / 1000000;
@@ -129,8 +214,9 @@ qd_timestamp_t qd_timer_now() {
void qd_timer_schedule(qd_timer_t *timer, qd_duration_t duration)
{
sys_mutex_lock(lock);
- timer_cancel_LH(timer); // Timer is now on the idle list
- DEQ_REMOVE(idle_timers, timer);
+
+ assert(timer->state != QD_TIMER_STATE_DELETED);
+ const bool was_scheduled = timer_cancel_LH(timer);
//
// Find the insert point in the schedule.
@@ -157,7 +243,12 @@ void qd_timer_schedule(qd_timer_t *timer, qd_duration_t duration)
else
DEQ_INSERT_HEAD(scheduled_timers, timer);
}
- timer->scheduled = true;
+
+ timer->state = QD_TIMER_STATE_SCHEDULED;
+ if (!was_scheduled) {
+ // scheduled_timers list reference:
+ sys_atomic_inc(&timer->ref_count);
+ }
qd_timer_t *first = DEQ_HEAD(scheduled_timers);
qd_server_timeout(first->server, first->delta_time);
@@ -168,7 +259,19 @@ void qd_timer_schedule(qd_timer_t *timer, qd_duration_t duration)
void qd_timer_cancel(qd_timer_t *timer)
{
sys_mutex_lock(lock);
- timer_cancel_LH(timer);
+
+ if (timer->state == QD_TIMER_STATE_RUNNING) {
+ assert(sys_thread_self() != callback_thread); // cancel within callback not allowed
+ timer->state = QD_TIMER_STATE_BLOCKED;
+ sys_cond_wait(timer->condition, lock);
+ }
+
+ // timer may have been resheduled before wait returns
+ const bool need_decref = timer_cancel_LH(timer);
+ timer->state = QD_TIMER_STATE_IDLE;
+ if (need_decref) // was on scheduled list
+ timer_decref_LH(timer);
+
sys_mutex_unlock(lock);
}
@@ -181,7 +284,6 @@ void qd_timer_cancel(qd_timer_t *timer)
void qd_timer_initialize(sys_mutex_t *server_lock)
{
lock = server_lock;
- DEQ_INIT(idle_timers);
DEQ_INIT(scheduled_timers);
time_base = 0;
}
@@ -197,18 +299,39 @@ void qd_timer_finalize(void)
void qd_timer_visit()
{
sys_mutex_lock(lock);
+ callback_thread = sys_thread_self();
timer_adjust_now_LH();
qd_timer_t *timer = DEQ_HEAD(scheduled_timers);
while (timer && timer->delta_time == 0) {
- timer_cancel_LH(timer); /* Removes timer from scheduled_timers */
+ // Remove timer from scheduled_timers but keep ref_count
+ assert(timer->state == QD_TIMER_STATE_SCHEDULED);
+ // note: still holding scheduled_timers refcount
+ timer_cancel_LH(timer);
+ timer->state = QD_TIMER_STATE_RUNNING;
sys_mutex_unlock(lock);
- timer->handler(timer->context); /* Call the handler outside the lock, may re-schedule */
+
+ /* The callback may reschedule or delete the timer while the lock is
+ * dropped. Attempting to delete the timer now will cause the caller to
+ * block until the callback is done.
+ */
+ timer->handler(timer->context);
+
sys_mutex_lock(lock);
+ if (timer->state == QD_TIMER_STATE_BLOCKED) {
+ sys_cond_signal(timer->condition);
+ // expect blocked caller sets timer->state
+ } else if (timer->state == QD_TIMER_STATE_RUNNING) {
+ timer->state = QD_TIMER_STATE_IDLE;
+ }
+
+ // now drop scheduled_timers reference:
+ timer_decref_LH(timer);
timer = DEQ_HEAD(scheduled_timers);
}
qd_timer_t *first = DEQ_HEAD(scheduled_timers);
if (first) {
qd_server_timeout(first->server, first->delta_time);
}
+ callback_thread = 0;
sys_mutex_unlock(lock);
}
diff --git a/src/timer_private.h b/src/timer_private.h
index 88397b1..830c8ff 100644
--- a/src/timer_private.h
+++ b/src/timer_private.h
@@ -19,20 +19,8 @@
* under the License.
*/
-#include "qpid/dispatch/ctools.h"
-#include "qpid/dispatch/threading.h"
#include "qpid/dispatch/timer.h"
-
-struct qd_timer_t {
- DEQ_LINKS(qd_timer_t);
- qd_server_t *server;
- qd_timer_cb_t handler;
- void *context;
- qd_timestamp_t delta_time;
- bool scheduled; /* true means on scheduled list, false on idle list */
-};
-
-DEQ_DECLARE(qd_timer_t, qd_timer_list_t);
+#include "qpid/dispatch/threading.h"
void qd_timer_initialize(sys_mutex_t *server_lock);
void qd_timer_finalize(void);
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index a64819f..5aec28d 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -63,6 +63,9 @@ target_link_libraries(test-receiver ${Proton_LIBRARIES})
add_executable(clogger clogger.c)
target_link_libraries(clogger ${Proton_LIBRARIES})
+add_executable(threaded_timer_test threaded_timer_test.c)
+target_link_libraries(threaded_timer_test qpid-dispatch)
+
# Bubblewrap is an unprivileged sandboxing tool for Linux. Setting --unshare-net allows
# running tests in parallel (the ctest -j option) without port clashes
set(USE_BWRAP OFF CACHE BOOL "Wrap test executions with bwrap (https://github.com/containers/bubblewrap)")
@@ -83,6 +86,9 @@ add_test(unit_tests_size_2 ${TEST_WRAP} unit_tests_size 2)
add_test(unit_tests_size_1 ${TEST_WRAP} unit_tests_size 1)
add_test(unit_tests ${TEST_WRAP} unit_tests ${CMAKE_CURRENT_SOURCE_DIR}/threads4.conf)
+# stand alone tests
+add_test(threaded_timer_test ${TEST_WRAP} threaded_timer_test ${CMAKE_CURRENT_SOURCE_DIR}/dummy.conf)
+
# Unit test python modules
foreach(py_test_module
router_engine_test
diff --git a/tests/dummy.conf b/tests/dummy.conf
new file mode 100644
index 0000000..0fbf350
--- /dev/null
+++ b/tests/dummy.conf
@@ -0,0 +1,28 @@
+##
+## Licensed to the Apache Software Foundation (ASF) under one
+## or more contributor license agreements. See the NOTICE file
+## distributed with this work for additional information
+## regarding copyright ownership. The ASF licenses this file
+## to you under the Apache License, Version 2.0 (the
+## "License"); you may not use this file except in compliance
+## with the License. You may obtain a copy of the License at
+##
+## http://www.apache.org/licenses/LICENSE-2.0
+##
+## Unless required by applicable law or agreed to in writing,
+## software distributed under the License is distributed on an
+## "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+## KIND, either express or implied. See the License for the
+## specific language governing permissions and limitations
+## under the License
+##
+
+
+router {
+ id : QDR
+}
+
+log {
+ module: DEFAULT
+ enable: none
+}
diff --git a/tests/threaded_timer_test.c b/tests/threaded_timer_test.c
new file mode 100644
index 0000000..2cab8a2
--- /dev/null
+++ b/tests/threaded_timer_test.c
@@ -0,0 +1,473 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <stdio.h>
+#include <limits.h>
+#include <sys/select.h>
+#include <qpid/dispatch/timer.h>
+#include "dispatch_private.h"
+#include <qpid/dispatch/alloc.h>
+#include "timer_private.h"
+#include "test_case.h"
+#include <qpid/dispatch/atomic.h>
+#include <qpid/dispatch/threading.h>
+
+
+// test the interaction of threads and timers
+//
+// This test cannot be part of the unit_test executable since unit_test
+// overrides some of the timer-private functions and this test needs to
+// exercise the "real" versions of those functions.
+
+
+// overide qd_server_timeout - this test uses ticker_thread instead of proactor
+// so this is not needed
+void qd_server_timeout(qd_server_t *server, qd_duration_t duration)
+{
+}
+
+
+// Timer thread - polls the timers every 10 msecs
+//
+static sys_mutex_t *ticker_lock = 0;
+static bool done = false;
+static void *ticker_thread(void *arg)
+{
+ struct timeval period;
+
+ sys_mutex_lock(ticker_lock);
+ while (!done) {
+ // 10 msec sleep
+ sys_mutex_unlock(ticker_lock);
+ period.tv_sec = 0;
+ period.tv_usec = 10 * 1000;
+ select(0, 0, 0, 0, &period);
+ qd_timer_visit();
+ sys_mutex_lock(ticker_lock);
+ }
+ sys_mutex_unlock(ticker_lock);
+ return 0;
+}
+
+
+// global event for synchronizing with timer callback
+//
+static struct event_t {
+ sys_mutex_t *m;
+ sys_cond_t *c;
+} event;
+
+static void test_setup()
+{
+ event.m = sys_mutex();
+ event.c = sys_cond();
+}
+
+static void test_cleanup()
+{
+ sys_mutex_free(event.m);
+ event.m = 0;
+ sys_cond_free(event.c);
+ event.c = 0;
+}
+
+
+//
+// simple set and expire test
+//
+
+static void test_simple_cb(void *context)
+{
+ int *iptr = (int *)context;
+
+ sys_mutex_lock(event.m); // block until test_simple waits
+ *iptr = 2;
+ sys_cond_signal(event.c);
+ sys_mutex_unlock(event.m);
+}
+
+static int test_simple(const char *name)
+{
+ int i = 1;
+ int result = 0;
+
+ test_setup();
+
+ qd_timer_t *t = qd_timer(0, test_simple_cb, (void *)&i);
+ sys_mutex_lock(event.m);
+
+ qd_timestamp_t start = qd_timer_now();
+ qd_timer_schedule(t, 100);
+ sys_cond_wait(event.c, event.m); // wait for cb to finish
+ qd_timestamp_t stop = qd_timer_now();
+ if (i != 2) {
+ fprintf(stderr, "%s failed: expected timer to run\n", name);
+ result = 1;
+ }
+
+ if ((stop - start) < 100) {
+ fprintf(stderr, "%s failed: expected timer ran too soon\n", name);
+ result = 1;
+ }
+ sys_mutex_unlock(event.m);
+ qd_timer_free(t);
+
+ test_cleanup();
+ fprintf(stderr, "Test test_simple: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test rescheduling from within the callback
+//
+
+static qd_timer_t *reschedule_timer;
+
+static void test_reschedule_internal_cb(void *context)
+{
+ int *iptr = (int *)context;
+
+ switch (*iptr) {
+ case 1: // first time in
+ (*iptr) += 1;
+ qd_timer_schedule(reschedule_timer, 100);
+ break;
+ case 2:
+ (*iptr) += 1;
+ sys_mutex_lock(event.m);
+ sys_cond_signal(event.c);
+ sys_mutex_unlock(event.m);
+ break;
+ }
+}
+
+static int test_reschedule_internal(const char *name)
+{
+ int i = 1;
+ int result = 0;
+
+ test_setup();
+ reschedule_timer = qd_timer(0, test_reschedule_internal_cb, (void *)&i);
+ sys_mutex_lock(event.m);
+ qd_timestamp_t start = qd_timer_now();
+ qd_timer_schedule(reschedule_timer, 100);
+ sys_cond_wait(event.c, event.m); // wait for cb to finish
+ qd_timestamp_t stop = qd_timer_now();
+
+ if (i != 3) {
+ fprintf(stderr, "%s failed: expected timer to reschedule\n", name);
+ result = 1;
+ }
+
+ if ((stop - start) < 200) {
+ fprintf(stderr, "%s failed: timer expired too soon\n", name);
+ result = 1;
+ }
+ sys_mutex_unlock(event.m);
+ qd_timer_free(reschedule_timer);
+
+ test_cleanup();
+ fprintf(stderr, "Test reschedule_internal: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test freeing timer from within the callback
+//
+
+static qd_timer_t *free_timer;
+
+static void test_free_internal_cb(void *context)
+{
+ sys_mutex_lock(event.m);
+ qd_timer_free(free_timer);
+ free_timer = 0;
+
+ sys_cond_signal(event.c);
+ sys_mutex_unlock(event.m);
+}
+
+static int test_free_internal(const char *name)
+{
+ int result = 0;
+
+ test_setup();
+ free_timer = qd_timer(0, test_free_internal_cb, 0);
+ sys_mutex_lock(event.m);
+ qd_timestamp_t start = qd_timer_now();
+ qd_timer_schedule(free_timer, 100);
+ sys_cond_wait(event.c, event.m); // wait for cb to finish
+ qd_timestamp_t stop = qd_timer_now();
+ if ((stop - start) < 100) {
+ fprintf(stderr, "%s failed: timer expired too soon\n", name);
+ result = 1;
+ }
+ sys_mutex_unlock(event.m);
+ if (free_timer != 0) {
+ fprintf(stderr, "%s failed: timer free failed\n", name);
+ result = 1;
+ }
+
+ test_cleanup();
+ fprintf(stderr, "Test free_internal: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test freeing timer before it can run
+//
+
+static void test_prefree_cb(void *context)
+{
+ abort(); // should never run
+}
+
+static int test_prefree(const char *name)
+{
+ int result = 0;
+
+ test_setup();
+ qd_timer_t *t = qd_timer(0, test_prefree_cb, 0);
+ qd_timestamp_t start = qd_timer_now();
+ qd_timer_schedule(t, 1000 * 60); // looong time
+ qd_timer_free(t);
+ qd_timestamp_t stop = qd_timer_now();
+ if ((stop - start) > 1000 * 60) {
+ fprintf(stderr, "%s failed: free blocked\n", name);
+ result = 1;
+ }
+
+ test_cleanup();
+ fprintf(stderr, "Test prefree: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test canceling timer before it can run
+//
+
+static void test_early_cancel_cb(void *context)
+{
+ abort(); // should never run
+}
+
+static int test_early_cancel(const char *name)
+{
+ int result = 0;
+
+ test_setup();
+ qd_timer_t *t = qd_timer(0, test_early_cancel_cb, 0);
+ qd_timestamp_t start = qd_timer_now();
+ qd_timer_schedule(t, 1000 * 60); // looong time
+ qd_timer_cancel(t);
+ qd_timestamp_t stop = qd_timer_now();
+ if ((stop - start) > 1000 * 60) {
+ fprintf(stderr, "%s failed: free blocked\n", name);
+ result = 1;
+ }
+ qd_timer_free(t);
+
+ test_cleanup();
+ fprintf(stderr, "Test early cancel: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test cancel then re-arm timer
+//
+
+static void test_rerun_cb(void *context)
+{
+ int *iptr = (int *)context;
+ if (*iptr != 2) {
+ abort(); // should never run
+ }
+ (*iptr) = 3;
+ sys_mutex_lock(event.m);
+ sys_cond_signal(event.c);
+ sys_mutex_unlock(event.m);
+}
+
+static int test_rerun(const char *name)
+{
+ int result = 0;
+ int i = 1;
+
+ test_setup();
+ qd_timer_t *t = qd_timer(0, test_rerun_cb, &i);
+ qd_timestamp_t start = qd_timer_now();
+ qd_timer_schedule(t, 1000 * 60); // looong time
+ qd_timer_cancel(t);
+ qd_timestamp_t stop = qd_timer_now();
+ if ((stop - start) > 1000 * 60) {
+ fprintf(stderr, "%s failed: free blocked\n", name);
+ result = 1;
+ }
+
+ sys_mutex_lock(event.m);
+
+ i = 2;
+ start = qd_timer_now();
+ qd_timer_schedule(t, 100);
+ sys_cond_wait(event.c, event.m); // wait for cb to finish
+ stop = qd_timer_now();
+ if ((stop - start) < 100) {
+ fprintf(stderr, "%s failed: expected timer ran too soon\n", name);
+ result = 1;
+ }
+ if (i != 3) {
+ fprintf(stderr, "%s failed: expected callback to finish\n", name);
+ result = 1;
+ }
+ sys_mutex_unlock(event.m);
+ qd_timer_free(t);
+
+ test_cleanup();
+ fprintf(stderr, "Test rerun: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test that canceling a running timer will block until callback completes
+//
+
+static void long_running_cb(void *context)
+{
+ struct timeval period = {.tv_sec = 0,
+ .tv_usec = 2000 * 1000}; // 2 sec
+
+ // wake caller
+ sys_mutex_lock(event.m);
+ sys_cond_signal(event.c);
+ sys_mutex_unlock(event.m);
+
+ // sleep for 2 second, cancel should block for this
+ select(0, 0, 0, 0, &period);
+ sys_atomic_inc((sys_atomic_t *)context);
+}
+
+static int test_sync_cancel(const char *name)
+{
+ int result = 0;
+ sys_atomic_t flag;
+ sys_atomic_init(&flag, 1);
+
+ test_setup();
+ qd_timer_t *t = qd_timer(0, long_running_cb, (void *)&flag);
+
+ sys_mutex_lock(event.m);
+ qd_timer_schedule(t, 10);
+ sys_cond_wait(event.c, event.m); // wait for cb to start
+ qd_timer_cancel(t); // expected to block until cb finishes
+ if (sys_atomic_get(&flag) != 2) {
+ fprintf(stderr, "%s failed: callback still running\n", name);
+ result = 1;
+ }
+ sys_mutex_unlock(event.m);
+ qd_timer_free(t);
+
+ test_cleanup();
+ fprintf(stderr, "Test sync cancel: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+//
+// Test that freeing a running timer will block until callback completes
+//
+
+static int test_sync_free(const char *name)
+{
+ int result = 0;
+ sys_atomic_t flag;
+ sys_atomic_init(&flag, 1);
+
+ test_setup();
+ qd_timer_t *t = qd_timer(0, long_running_cb, (void *)&flag);
+
+ sys_mutex_lock(event.m);
+ qd_timer_schedule(t, 10);
+ sys_cond_wait(event.c, event.m); // wait for cb to start
+ qd_timer_free(t); // expected to block until cb finishes
+ if (sys_atomic_get(&flag) != 2) {
+ fprintf(stderr, "%s failed: callback still running\n", name);
+ result = 1;
+ }
+ sys_mutex_unlock(event.m);
+
+ test_cleanup();
+ fprintf(stderr, "Test sync free: %s\n", result ? "FAILED" : "ok");
+ return result;
+}
+
+
+int main(int argc, char *argv[])
+{
+ int result = 0;
+
+ if (argc != 2) {
+ fprintf(stderr, "usage: %s <config-file>\n", argv[0]);
+ exit(1);
+ }
+
+ // Call qd_dispatch() first initialize allocator used by other tests.
+ qd_dispatch_t *qd = qd_dispatch(0, false);
+
+ qd_dispatch_validate_config(argv[1]);
+ if (qd_error_code()) {
+ printf("Config failed: %s\n", qd_error_message());
+ return 1;
+ }
+
+ qd_dispatch_load_config(qd, argv[1]);
+ if (qd_error_code()) {
+ printf("Config failed: %s\n", qd_error_message());
+ return 1;
+ }
+
+ ticker_lock = sys_mutex();
+
+ sys_thread_t *ticker = sys_thread(ticker_thread, 0);
+
+ result = test_simple(argv[0]);
+ result += test_reschedule_internal(argv[0]);
+ result += test_free_internal(argv[0]);
+ result += test_prefree(argv[0]);
+ result += test_early_cancel(argv[0]);
+ result += test_rerun(argv[0]);
+ result += test_sync_cancel(argv[0]);
+ result += test_sync_free(argv[0]);
+
+ sys_mutex_lock(ticker_lock);
+ done = true;
+ sys_mutex_unlock(ticker_lock);
+
+ sys_thread_join(ticker);
+ sys_thread_free(ticker);
+
+ qd_dispatch_free(qd); // dispatch_free last.
+
+ return result;
+}
diff --git a/tests/tsan.supp b/tests/tsan.supp
index 68d89d8..4248dc4 100644
--- a/tests/tsan.supp
+++ b/tests/tsan.supp
@@ -14,7 +14,8 @@ race:qd_entity_refresh_allocator
mutex:qd_router_timer_handler
race:qdr_process_tick_CT
-deadlock:qd_timer
+deadlock:qd_timer_schedule
+deadlock:qd_timer_visit
race:qdr_connection_process
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org