You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2018/04/12 20:52:22 UTC
[1/4] qpid-proton git commit: NO-JIRA: [c] Minor comment & example
fixes, multi-thread API doc fixes
Repository: qpid-proton
Updated Branches:
refs/heads/master df467c4c4 -> 188ce2806
NO-JIRA: [c] Minor comment & example fixes, multi-thread API doc fixes
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/811d683a
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/811d683a
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/811d683a
Branch: refs/heads/master
Commit: 811d683a090f00bc732073624c36b8bc51de1c05
Parents: df467c4
Author: Alan Conway <ac...@redhat.com>
Authored: Fri Feb 16 09:12:13 2018 -0500
Committer: Alan Conway <ac...@redhat.com>
Committed: Wed Apr 11 18:20:31 2018 -0400
----------------------------------------------------------------------
c/examples/broker.c | 2 +-
c/include/proton/netaddr.h | 2 +-
c/include/proton/proactor.h | 7 +++----
cpp/examples/broker.cpp | 4 ++--
4 files changed, 7 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/811d683a/c/examples/broker.c
----------------------------------------------------------------------
diff --git a/c/examples/broker.c b/c/examples/broker.c
index 852fa3a..dcb0158 100644
--- a/c/examples/broker.c
+++ b/c/examples/broker.c
@@ -289,7 +289,7 @@ static void handle(broker_t* b, pn_event_t* e) {
switch (pn_event_type(e)) {
case PN_LISTENER_OPEN: {
- char port[256]; /* Get the listening port */
+ char port[PN_MAX_ADDR]; /* Get the listening port */
pn_netaddr_host_port(pn_listener_addr(pn_event_listener(e)), NULL, 0, port, sizeof(port));
printf("listening on %s\n", port);
fflush(stdout);
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/811d683a/c/include/proton/netaddr.h
----------------------------------------------------------------------
diff --git a/c/include/proton/netaddr.h b/c/include/proton/netaddr.h
index 4d422cf..1ac8c21 100644
--- a/c/include/proton/netaddr.h
+++ b/c/include/proton/netaddr.h
@@ -42,7 +42,7 @@ extern "C" {
typedef struct pn_netaddr_t pn_netaddr_t;
/**
- * Format a network address as a human-readable string in `buf`.
+ * Format a network address string in `buf`.
*
* @return the length of the string (excluding trailing '\0'), if >= size then
* the address was truncated.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/811d683a/c/include/proton/proactor.h
----------------------------------------------------------------------
diff --git a/c/include/proton/proactor.h b/c/include/proton/proactor.h
index be20ff1..77e0d84 100644
--- a/c/include/proton/proactor.h
+++ b/c/include/proton/proactor.h
@@ -74,7 +74,7 @@ extern "C" {
#define PN_MAX_ADDR 1060
/**
- * Format a host:port address string for pn_proactor_connect2() or pn_proactor_listen2()
+ * Format a host:port address string for pn_proactor_connect() or pn_proactor_listen()
*
* @param[out] addr address is copied to this buffer, with trailing '\0'
* @param[in] size size of addr buffer
@@ -234,8 +234,7 @@ PNP_EXTERN void pn_proactor_interrupt(pn_proactor_t *proactor);
* will be delivered to the next available thread.
*
* Calling pn_proactor_set_timeout() again before the PN_PROACTOR_TIMEOUT
- * is delivered will cancel the previous timeout and deliver an event only after
- * the new timeout.
+ * is delivered replaces the previous timeout value.
*
* @note Thread-safe
*/
@@ -256,7 +255,7 @@ PNP_EXTERN void pn_proactor_cancel_timeout(pn_proactor_t *proactor);
* and so on) remain intact, but the transport is closed and unbound. The
* proactor will not return any more events for this connection. The caller must
* call pn_connection_free(), either directly or indirectly by re-using @p
- * connection in another call to pn_proactor_connect2() or pn_proactor_listen2().
+ * connection in another call to pn_proactor_connect() or pn_proactor_listen().
*
* @note **Not thread-safe**. Call this function from a connection
* event handler.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/811d683a/cpp/examples/broker.cpp
----------------------------------------------------------------------
diff --git a/cpp/examples/broker.cpp b/cpp/examples/broker.cpp
index 65384b3..47d6109 100644
--- a/cpp/examples/broker.cpp
+++ b/cpp/examples/broker.cpp
@@ -52,8 +52,8 @@ int hardware_concurrency() {return 1;}
#include "fake_cpp11.hpp"
-// This is a simplified model for a message broker, that only allows for messages to go to a
-// single receiver.
+// This is a simplified model for a message broker, that only allows for
+// messages to go to a single receiver.
//
// This broker is multithread safe and if compiled with C++11 with a multithreaded Proton
// binding library will use as many threads as there are thread resources available (usually
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[3/4] qpid-proton git commit: PROTON-1771: [c] add missing lock
around wake_if_inactive
Posted by ac...@apache.org.
PROTON-1771: [c] add missing lock around wake_if_inactive
Fixes a race condition discovered by threaderciser.c
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/3e2f9b5f
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/3e2f9b5f
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/3e2f9b5f
Branch: refs/heads/master
Commit: 3e2f9b5f8164a6720740b49b72af6471b1081df9
Parents: 8a4fac6
Author: Alan Conway <ac...@redhat.com>
Authored: Thu Apr 5 17:57:49 2018 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Thu Apr 12 16:23:13 2018 -0400
----------------------------------------------------------------------
c/src/proactor/epoll.c | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/3e2f9b5f/c/src/proactor/epoll.c
----------------------------------------------------------------------
diff --git a/c/src/proactor/epoll.c b/c/src/proactor/epoll.c
index 8ff5831..752e6e0 100644
--- a/c/src/proactor/epoll.c
+++ b/c/src/proactor/epoll.c
@@ -1321,7 +1321,9 @@ void pn_proactor_connect2(pn_proactor_t *p, pn_connection_t *c, pn_transport_t *
} else {
psocket_gai_error(&pc->psocket, gai_error, "connect to ");
notify = wake(&pc->context);
+ lock(&p->context.mutex);
notify_proactor = wake_if_inactive(p);
+ unlock(&p->context.mutex);
}
}
/* We need to issue INACTIVE on immediate failure */
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[4/4] qpid-proton git commit: PROTON-1771: [c] locking around
epoll_extended_t
Posted by ac...@apache.org.
PROTON-1771: [c] locking around epoll_extended_t
Add locking for epoll_extended_t structure to fix race conditions revealed
by threaderciser.c
We pass pointers to this struct between threads via epoll, and epoll does not
guarantee that memory writes by one thread will be visible in the next thread
that gets the pointer. Since helgrind reports the races, it suggests that epoll
is not acting as a memory barrier in practice either.
These mutexes will not be contended so they are not expensive. We could use an
acquire/release memory barrier, but an un-contended mutex is basically exactly
that, and is portable.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/188ce280
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/188ce280
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/188ce280
Branch: refs/heads/master
Commit: 188ce28066df8f5e965fb63593f419f49c950760
Parents: 3e2f9b5
Author: Alan Conway <ac...@redhat.com>
Authored: Thu Apr 5 19:10:11 2018 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Thu Apr 12 16:23:13 2018 -0400
----------------------------------------------------------------------
c/src/proactor/epoll.c | 68 ++++++++++++++++++++++++++-------------------
1 file changed, 40 insertions(+), 28 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/188ce280/c/src/proactor/epoll.c
----------------------------------------------------------------------
diff --git a/c/src/proactor/epoll.c b/c/src/proactor/epoll.c
index 752e6e0..d0db0a7 100644
--- a/c/src/proactor/epoll.c
+++ b/c/src/proactor/epoll.c
@@ -128,20 +128,9 @@ typedef struct epoll_extended_t {
epoll_type_t type; // io/timer/wakeup
uint32_t wanted; // events to poll for
bool polling;
- pmutex barrier_mutex;
+ pmutex mutex;
} epoll_extended_t;
-/* epoll_ctl()/epoll_wait() do not form a memory barrier, so cached memory
- writes to struct epoll_extended_t in the EPOLL_ADD thread might not be
- visible to epoll_wait() thread. This function creates a memory barrier,
- called before epoll_ctl() and after epoll_wait()
-*/
-static void memory_barrier(epoll_extended_t *ee) {
- // Mutex lock/unlock has the side-effect of being a memory barrier.
- lock(&ee->barrier_mutex);
- unlock(&ee->barrier_mutex);
-}
-
/*
* This timerfd logic assumes EPOLLONESHOT and there never being two
* active timeout callbacks. There can be multiple (or zero)
@@ -287,28 +276,38 @@ PN_STRUCT_CLASSDEF(pn_proactor, CID_pn_proactor)
PN_STRUCT_CLASSDEF(pn_listener, CID_pn_listener)
static bool start_polling(epoll_extended_t *ee, int epollfd) {
- if (ee->polling)
+ lock(&ee->mutex);
+ if (ee->polling) {
+ unlock(&ee->mutex);
return false;
+ }
ee->polling = true;
struct epoll_event ev;
ev.data.ptr = ee;
ev.events = ee->wanted | EPOLLONESHOT;
- memory_barrier(ee);
- return (epoll_ctl(epollfd, EPOLL_CTL_ADD, ee->fd, &ev) == 0);
+ int fd = ee->fd;
+ unlock(&ee->mutex);
+ return (epoll_ctl(epollfd, EPOLL_CTL_ADD, fd, &ev) == 0);
}
static void stop_polling(epoll_extended_t *ee, int epollfd) {
// TODO: check for error, return bool or just log?
- if (ee->fd == -1 || !ee->polling || epollfd == -1)
+ lock(&ee->mutex);
+ if (ee->fd == -1 || !ee->polling || epollfd == -1) {
+ unlock(&ee->mutex);
return;
+ }
struct epoll_event ev;
ev.data.ptr = ee;
ev.events = 0;
- memory_barrier(ee);
- if (epoll_ctl(epollfd, EPOLL_CTL_DEL, ee->fd, &ev) == -1)
+ int fd = ee->fd;
+ unlock(&ee->mutex);
+ if (epoll_ctl(epollfd, EPOLL_CTL_DEL, fd, &ev) == -1)
EPOLL_FATAL("EPOLL_CTL_DEL", errno);
+ lock(&ee->mutex);
ee->fd = -1;
ee->polling = false;
+ unlock(&ee->mutex);
}
/*
@@ -682,10 +681,12 @@ static void psocket_gai_error(psocket_t *ps, int gai_err, const char* what) {
static void rearm(pn_proactor_t *p, epoll_extended_t *ee) {
struct epoll_event ev;
+ lock(&ee->mutex);
ev.data.ptr = ee;
ev.events = ee->wanted | EPOLLONESHOT;
- memory_barrier(ee);
- if (epoll_ctl(p->epollfd, EPOLL_CTL_MOD, ee->fd, &ev) == -1)
+ int fd = ee->fd;
+ unlock(&ee->mutex);
+ if (epoll_ctl(p->epollfd, EPOLL_CTL_MOD, fd, &ev) == -1)
EPOLL_FATAL("arming polled file descriptor", errno);
}
@@ -1229,6 +1230,7 @@ static void pconnection_start(pconnection_t *pc) {
(void)getsockname(pc->psocket.sockfd, (struct sockaddr*)&pc->local.ss, &len);
epoll_extended_t *ee = &pc->psocket.epoll_io;
+ lock(&ee->mutex);
if (ee->polling) { /* This is not the first attempt, stop polling and close the old FD */
int fd = ee->fd; /* Save fd, it will be set to -1 by stop_polling */
stop_polling(ee, efd);
@@ -1236,6 +1238,7 @@ static void pconnection_start(pconnection_t *pc) {
}
ee->fd = pc->psocket.sockfd;
pc->current_arm = ee->wanted = EPOLLIN | EPOLLOUT;
+ unlock(&ee->mutex);
start_polling(ee, efd); // TODO: check for error
}
@@ -1759,11 +1762,13 @@ void pn_listener_accept2(pn_listener_t *l, pn_connection_t *c, pn_transport_t *t
/* Set up an epoll_extended_t to be used for wakeup or interrupts */
static void epoll_wake_init(epoll_extended_t *ee, int eventfd, int epollfd) {
+ lock(&ee->mutex);
ee->psocket = NULL;
ee->fd = eventfd;
ee->type = WAKE;
ee->wanted = EPOLLIN;
ee->polling = false;
+ unlock(&ee->mutex);
start_polling(ee, epollfd); // TODO: check for error
}
@@ -1954,7 +1959,10 @@ static bool proactor_remove(pcontext_t *ctx) {
}
static pn_event_batch_t *process_inbound_wake(pn_proactor_t *p, epoll_extended_t *ee) {
- if (ee->fd == p->interruptfd) { /* Interrupts have their own dedicated eventfd */
+ lock(&ee->mutex);
+ int fd = ee->fd;
+ unlock(&ee->mutex);
+ if (fd == p->interruptfd) { /* Interrupts have their own dedicated eventfd */
(void)read_uint64(p->interruptfd);
rearm(p, &p->epoll_interrupt);
return proactor_process(p, PN_PROACTOR_INTERRUPT);
@@ -1999,25 +2007,29 @@ static pn_event_batch_t *proactor_do_epoll(struct pn_proactor_t* p, bool can_blo
}
assert(n == 1);
epoll_extended_t *ee = (epoll_extended_t *) ev.data.ptr;
- memory_barrier(ee);
- if (ee->type == WAKE) {
+ lock(&ee->mutex);
+ epoll_type_t type = ee->type;
+ struct psocket_t *psocket = ee->psocket;
+ unlock(&ee->mutex);
+
+ if (type == WAKE) {
batch = process_inbound_wake(p, ee);
- } else if (ee->type == PROACTOR_TIMER) {
+ } else if (type == PROACTOR_TIMER) {
batch = proactor_process(p, PN_PROACTOR_TIMEOUT);
} else {
- pconnection_t *pc = psocket_pconnection(ee->psocket);
+ pconnection_t *pc = psocket_pconnection(psocket);
if (pc) {
- if (ee->type == PCONNECTION_IO) {
+ if (type == PCONNECTION_IO) {
batch = pconnection_process(pc, ev.events, false, false);
} else {
- assert(ee->type == PCONNECTION_TIMER);
+ assert(type == PCONNECTION_TIMER);
batch = pconnection_process(pc, 0, true, false);
}
}
else {
// TODO: can any of the listener processing be parallelized like IOCP?
- batch = listener_process(ee->psocket, ev.events);
+ batch = listener_process(psocket, ev.events);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/4] qpid-proton git commit: PROTON-1771: [c] threaderciser thread
race test for C proactor
Posted by ac...@apache.org.
PROTON-1771: [c] threaderciser thread race test for C proactor
Take random actions in multiple threads, try to provoke race conditions under
helgrind or other race checker.
Actions can be enabled or disabled on command line to simplify the test.
threaderciser.supp has supressions for race conditions detected on Linux with
the epoll proactor.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/8a4fac69
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/8a4fac69
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/8a4fac69
Branch: refs/heads/master
Commit: 8a4fac69fa84111142351bb763c4835fc706a273
Parents: 811d683
Author: Alan Conway <ac...@redhat.com>
Authored: Wed Mar 7 16:08:06 2018 -0500
Committer: Alan Conway <ac...@redhat.com>
Committed: Thu Apr 12 16:22:08 2018 -0400
----------------------------------------------------------------------
CMakeLists.txt | 1 +
c/tests/CMakeLists.txt | 30 ++-
c/tests/thread.h | 81 ++++++
c/tests/threaderciser.c | 550 ++++++++++++++++++++++++++++++++++++++++
c/tests/threaderciser.supp | 5 +
5 files changed, 664 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/8a4fac69/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 4e05d48..a0fca0d 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -229,6 +229,7 @@ else ()
set (VALGRIND_ENV "VALGRIND=${VALGRIND_EXECUTABLE}" "VALGRIND_ARGS=${VALGRIND_OPTIONS} --suppressions=${VALGRIND_SUPPRESSIONS}")
separate_arguments(VALGRIND_OPTIONS_LIST UNIX_COMMAND ${VALGRIND_OPTIONS})
set (memcheck-cmd ${VALGRIND_EXECUTABLE} ${VALGRIND_OPTIONS_LIST} "--suppressions=${VALGRIND_SUPPRESSIONS}")
+ set (racecheck-cmd ${VALGRIND_EXECUTABLE} --tool=helgrind --error-exitcode=42 --quiet)
else ()
message(STATUS "Can't locate the valgrind command; no run-time error detection")
endif ()
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/8a4fac69/c/tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/c/tests/CMakeLists.txt b/c/tests/CMakeLists.txt
index 3b43fc8..7283c87 100644
--- a/c/tests/CMakeLists.txt
+++ b/c/tests/CMakeLists.txt
@@ -22,14 +22,19 @@ add_definitions(${COMPILE_WARNING_FLAGS} ${COMPILE_PLATFORM_FLAGS})
configure_file(test_config.h.in test_config.h)
include_directories(${CMAKE_CURRENT_BINARY_DIR})
-macro (pn_add_c_test test)
+macro (pn_build_c_test test)
add_executable (${test} ${ARGN})
target_link_libraries (${test} qpid-proton-core ${PLATFORM_LIBS})
if (BUILD_WITH_CXX)
set_source_files_properties (${ARGN} PROPERTIES LANGUAGE CXX)
endif (BUILD_WITH_CXX)
- add_test (NAME ${test}
- COMMAND ${PN_ENV_SCRIPT} -- "PATH=$<TARGET_FILE_DIR:qpid-proton-core>" ${memcheck-cmd} $<TARGET_FILE:${test}>)
+endmacro()
+
+set(test_env ${PN_ENV_SCRIPT} -- "PATH=$<TARGET_FILE_DIR:qpid-proton-core>")
+
+macro (pn_add_c_test test)
+ pn_build_c_test(${test} ${ARGN})
+ add_test (NAME ${test} COMMAND ${test_env} ${memcheck-cmd} $<TARGET_FILE:${test}>)
endmacro(pn_add_c_test)
pn_add_c_test (c-object-tests object.c)
@@ -49,6 +54,25 @@ if(HAS_PROACTOR)
pn_add_c_test (c-proactor-tests proactor.c)
target_link_libraries (c-proactor-tests qpid-proton-proactor)
+ # TODO Enable by default when races are cleared up
+ option(THRADERCISER "Run the threaderciser concurrency tests" OFF)
+ if (THREADERCISER)
+ pn_build_c_test(c-threaderciser threaderciser.c)
+ target_link_libraries (c-threaderciser pthread qpid-proton-proactor)
+ find_library(PTHREAD_LIBRARY pthread)
+ if (PTHREAD_LIBRARY)
+ target_link_libraries (c-threaderciser ${PTHREAD_LIBRARY})
+ endif()
+
+ # Always run with race checker, run with mem checker also if it is enabled
+ set(prog $<TARGET_FILE:c-threaderciser>)
+ set (opts --suppressions=${CMAKE_CURRENT_SOURCE_DIR}/threaderciser.supp)
+ add_test (NAME c-threaderciser-racecheck COMMAND ${test_env} ${racecheck-cmd} ${opts} ${prog})
+ if (memcheck-cmd)
+ add_test (NAME c-threaderciser-memcheck COMMAND ${test_env} ${memcheck-cmd} ${prog})
+ endif()
+ endif()
+
if(WIN32)
set(path "$<TARGET_FILE_DIR:c-broker>\\;$<TARGET_FILE_DIR:qpid-proton>")
else(WIN32)
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/8a4fac69/c/tests/thread.h
----------------------------------------------------------------------
diff --git a/c/tests/thread.h b/c/tests/thread.h
new file mode 100644
index 0000000..9e029fc
--- /dev/null
+++ b/c/tests/thread.h
@@ -0,0 +1,81 @@
+#ifndef TESTS_THREAD_H
+#define TESTS_THREAD_H
+
+/*
+ * 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.
+ */
+
+/* Simulate a subset of POSIX threads for windows */
+
+#ifdef _WIN32
+
+#include <windows.h>
+#include <process.h>
+
+typedef struct {
+ HANDLE handle;
+ void *(*func)(void *);
+ void *arg;
+} pthread_t;
+
+static unsigned __stdcall pthread_run(void *thr0) {
+ pthread_t *t = (pthread_t *) thr0;
+ t->func(t->arg);
+ return 0;
+}
+
+static int pthread_create(pthread_t *t, void *unused, void *(*f)(void *), void *arg) {
+ t->func = f;
+ t->arg = arg;
+ t->handle = (HANDLE) _beginthreadex(0, 0, &pthread_run, t, 0, 0);
+ if (t->handle) {
+ return 0;
+ }
+ return -1;
+}
+
+static int pthread_join(pthread_t t, void **unused) {
+ if (t.handle) {
+ WaitForSingleObject(t.handle, INFINITE);
+ CloseHandle(t.handle);
+ }
+ return 0;
+}
+
+typedef CRITICAL_SECTION pthread_mutex_t;
+#define pthread_mutex_init(m, unused) InitializeCriticalSectionAndSpinCount(m, 4000)
+#define pthread_mutex_destroy(m) DeleteCriticalSection(m)
+#define pthread_mutex_lock(m) EnterCriticalSection(m)
+#define pthread_mutex_unlock(m) LeaveCriticalSection(m)
+
+void millisleep(long ms) { Sleep(ms); }
+
+#else /* POSIX */
+
+#include <pthread.h>
+#include <unistd.h> /* For sleep() */
+
+void millisleep(long ms) {
+ struct timespec delay = {0};
+ delay.tv_sec = ms / 1000;
+ delay.tv_nsec = (ms % 1000) * 1000000;
+ nanosleep(&delay, NULL);
+}
+
+#endif
+
+#endif /*!TESTS_THREAD_H*/
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/8a4fac69/c/tests/threaderciser.c
----------------------------------------------------------------------
diff --git a/c/tests/threaderciser.c b/c/tests/threaderciser.c
new file mode 100644
index 0000000..82f39ea
--- /dev/null
+++ b/c/tests/threaderciser.c
@@ -0,0 +1,550 @@
+/*
+ * 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.
+ */
+
+/* This test is intended to be run under a race detector (e.g. helgrind,
+ libtsan) to uncover races by generating concurrent, pseudo-random activity.
+ It should also be run under memory checkers (e.g. valgrind memcheck, libasan)
+ for race conditions that leak memory.
+
+ The goal is to drive concurrent activity as hard as possible while staying
+ within the rules of the API. It may do things that an application is unlikely
+ to do (e.g. close a listener before it opens) but that are not disallowed by
+ the API rules - that's the goal.
+
+ It is impossible to get repeatable behaviour with multiple threads due to
+ unpredictable scheduling. Currently using plain old rand(), if quality of
+ randomness is problem we can upgrade.
+
+ TODO
+ - closing connections
+ - pn_proactor_release_connection and re-use with pn_proactor_connect/accept
+ - sending/receiving/tracking messages
+ - cancel timeout
+*/
+
+#include "thread.h"
+
+#include <proton/connection.h>
+#include <proton/event.h>
+#include <proton/listener.h>
+#include <proton/netaddr.h>
+#include <proton/proactor.h>
+
+#include <stdarg.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+
+#undef NDEBUG /* Enable assert even in release builds */
+#include <assert.h>
+
+#define BACKLOG 16 /* Listener backlog */
+#define TIMEOUT_MAX 100 /* Milliseconds */
+#define SLEEP_MAX 100 /* Milliseconds */
+
+/* Set of actions that can be enabled/disabled/counted */
+typedef enum { A_LISTEN, A_LCLOSE, A_CONNECT, A_CCLOSE, A_WAKE, A_TIMEOUT } action;
+const char* action_name[] = { "listen", "lclose", "connect", "cclose", "wake", "timeout" };
+#define action_size (sizeof(action_name)/sizeof(*action_name))
+bool action_enabled[action_size] = { 0 } ;
+
+static int assert_no_err(int n) { assert(n >= 0); return n; }
+
+static bool debug_enable = false;
+
+/*
+ NOTE: Printing to a single debug log from many threads creates artificial
+ before/after relationships that may mask race conditions (locks in fputs)
+
+ TODO: something better, e.g. per-thread timestamped logs that are only
+ collated and printed periodically or at end of process.
+*/
+#define debug(...) if (debug_enable) debug_impl(__VA_ARGS__)
+
+static void debug_impl(const char *fmt, ...) {
+ /* Collect thread-id and message in a single buffer to minimize mixed messages */
+ char msg[256];
+ char *i = msg;
+ char *end = i + sizeof(msg);
+ i += assert_no_err(snprintf(i, end-i, "(%lx) ", pthread_self()));
+ if (i < end) {
+ va_list ap;
+ va_start(ap, fmt);
+ i += assert_no_err(vsnprintf(i, end-i, fmt, ap));
+ va_end(ap);
+ }
+ fputs(msg, stderr);
+}
+
+/* Shorthand for debugging an action using id as identifier */
+static inline void debuga(action a, void *id) {
+ debug("[%p] %s", id, action_name[a]);
+}
+
+/* Thread safe pools of structs.
+
+ An element can be "picked" at random concurrently by many threads, and is
+ guaranteed not to be freed until "done" is called by all threads. Element
+ structs must have their own locks if picking threads can modify the element.
+*/
+
+struct pool;
+
+/* This must be the first field in structs that go in a pool */
+typedef struct pool_entry {
+ struct pool *pool;
+ int ref; /* threads using this entry */
+} pool_entry;
+
+typedef void (*pool_free_fn)(void*);
+
+typedef struct pool {
+ pthread_mutex_t lock;
+ pool_entry **entries;
+ size_t size, max;
+ pool_free_fn free_fn;
+} pool;
+
+
+void pool_init(pool *p, size_t max, pool_free_fn free_fn) {
+ pthread_mutex_init(&p->lock, NULL);
+ p->entries = (pool_entry**)calloc(max, sizeof(*p->entries));
+ p->size = 0;
+ p->max = max;
+ p->free_fn = free_fn;
+}
+
+void pool_destroy(pool *p) {
+ pthread_mutex_destroy(&p->lock);
+ free(p->entries);
+}
+
+bool pool_add(pool *p, pool_entry *entry) {
+ pthread_mutex_lock(&p->lock);
+ entry->pool = p;
+ entry->ref = 1;
+ bool ok = false;
+ if (p->size < p->max) {
+ p->entries[p->size++] = entry;
+ ok = true;
+ }
+ pthread_mutex_unlock(&p->lock);
+ return ok;
+}
+
+void pool_unref(pool_entry *entry) {
+ pool *p = (pool*)entry->pool;
+ pthread_mutex_lock(&p->lock);
+ if (--entry->ref == 0) {
+ size_t i = 0;
+ while (i < p->size && p->entries[i] != entry) ++i;
+ if (i < p->size) {
+ --p->size;
+ memmove(p->entries + i, p->entries + i + 1, (p->size - i)*sizeof(*p->entries));
+ (*p->free_fn)(entry);
+ }
+ }
+ pthread_mutex_unlock(&p->lock);
+}
+
+/* Pick a random element, NULL if empty. Call pool_unref(entry) when finished */
+pool_entry *pool_pick(pool *p) {
+ pthread_mutex_lock(&p->lock);
+ pool_entry *entry = NULL;
+ if (p->size) {
+ entry = p->entries[rand() % p->size];
+ ++entry->ref;
+ }
+ pthread_mutex_unlock(&p->lock);
+ return entry;
+}
+
+/* Macro for type-safe pools */
+#define DECLARE_POOL(P, T) \
+ typedef struct P { pool p; } P; \
+ void P##_init(P *p, size_t max, void (*free_fn)(T*)) { pool_init((pool*)p, max, (pool_free_fn) free_fn); } \
+ bool P##_add(P *p, T *entry) { return pool_add((pool*)p, &entry->entry); } \
+ void P##_unref(T *entry) { pool_unref(&entry->entry); } \
+ T *P##_pick(P *p) { return (T*)pool_pick((pool*) p); } \
+ void P##_destroy(P *p) { pool_destroy(&p->p); }
+
+
+/* Connection pool */
+
+typedef struct connection_ctx {
+ pool_entry entry;
+ pn_connection_t *pn_connection;
+ pthread_mutex_t lock; /* Lock PN_TRANSPORT_CLOSED vs. pn_connection_wake() */
+} connection_ctx;
+
+DECLARE_POOL(cpool, connection_ctx)
+
+static connection_ctx* connection_ctx_new(void) {
+ connection_ctx *ctx = (connection_ctx*)malloc(sizeof(*ctx));
+ ctx->pn_connection = pn_connection();
+ pthread_mutex_init(&ctx->lock, NULL);
+ pn_connection_set_context(ctx->pn_connection, ctx);
+ return ctx;
+}
+
+static void connection_ctx_free(connection_ctx* ctx) {
+ /* Don't free ctx->pn_connection, freed by proactor */
+ pthread_mutex_destroy(&ctx->lock);
+ free(ctx);
+}
+
+void cpool_connect(cpool *cp, pn_proactor_t *proactor, const char *addr) {
+ if (!action_enabled[A_CONNECT]) return;
+ connection_ctx *ctx = connection_ctx_new();
+ if (cpool_add(cp, ctx)) {
+ debuga(A_CONNECT, ctx->pn_connection);
+ pn_proactor_connect(proactor, ctx->pn_connection, addr);
+ } else {
+ pn_connection_free(ctx->pn_connection); /* Won't be freed by proactor */
+ connection_ctx_free(ctx);
+ }
+}
+
+void cpool_wake(cpool *cp) {
+ if (!action_enabled[A_WAKE]) return;
+ connection_ctx *ctx = cpool_pick(cp);
+ if (ctx) {
+ /* Required locking: application may not call wake on a freed connection */
+ pthread_mutex_lock(&ctx->lock);
+ if (ctx && ctx->pn_connection) {
+ debuga(A_WAKE, ctx->pn_connection);
+ pn_connection_wake(ctx->pn_connection);
+ }
+ pthread_mutex_unlock(&ctx->lock);
+ cpool_unref(ctx);
+ }
+}
+
+static void connection_ctx_on_close(connection_ctx *ctx) {
+ /* Required locking: mark connection (possibly) closed no more wake calls */
+ pthread_mutex_lock(&ctx->lock);
+ ctx->pn_connection = NULL;
+ pthread_mutex_unlock(&ctx->lock);
+ cpool_unref(ctx);
+}
+
+/* Listener pool */
+
+typedef struct listener_ctx {
+ pool_entry entry;
+ pn_listener_t *pn_listener;
+ pthread_mutex_t lock; /* Lock PN_LISTENER_CLOSE vs. pn_listener_close() */
+ char addr[PN_MAX_ADDR];
+} listener_ctx;
+
+DECLARE_POOL(lpool, listener_ctx)
+
+static listener_ctx* listener_ctx_new(void) {
+ listener_ctx *ctx = (listener_ctx*)malloc(sizeof(*ctx));
+ ctx->pn_listener = pn_listener();
+ pthread_mutex_init(&ctx->lock, NULL);
+ /* Use "invalid:address" because "" is treated as ":5672" */
+ strncpy(ctx->addr, "invalid:address", sizeof(ctx->addr));
+ pn_listener_set_context(ctx->pn_listener, ctx);
+ return ctx;
+}
+
+static void listener_ctx_free(listener_ctx *ctx) {
+ /* Don't free ctx->pn_listener, freed by proactor */
+ pthread_mutex_destroy(&ctx->lock);
+ free(ctx);
+}
+
+static void lpool_listen(lpool *lp, pn_proactor_t *proactor) {
+ if (!action_enabled[A_LISTEN]) return;
+ char a[PN_MAX_ADDR];
+ pn_proactor_addr(a, sizeof(a), "", "0");
+ listener_ctx *ctx = listener_ctx_new();
+ if (lpool_add(lp, ctx)) {
+ debuga(A_LISTEN, ctx->pn_listener);
+ pn_proactor_listen(proactor, ctx->pn_listener, a, BACKLOG);
+ } else {
+ pn_listener_free(ctx->pn_listener); /* Won't be freed by proactor */
+ listener_ctx_free(ctx);
+ }
+}
+
+/* Advertise address once open */
+static void listener_ctx_on_open(listener_ctx *ctx) {
+ pthread_mutex_lock(&ctx->lock);
+ if (ctx->pn_listener) {
+ pn_netaddr_str(pn_listener_addr(ctx->pn_listener), ctx->addr, sizeof(ctx->addr));
+ }
+ debug("[%p] listening on %s", ctx->pn_listener, ctx->addr);
+ pthread_mutex_unlock(&ctx->lock);
+}
+
+static void listener_ctx_on_close(listener_ctx *ctx) {
+ pthread_mutex_lock(&ctx->lock);
+ ctx->pn_listener = NULL;
+ pthread_mutex_unlock(&ctx->lock);
+ lpool_unref(ctx);
+}
+
+/* Pick a random listening address from the listener pool.
+ Returns "invalid:address" for no address.
+*/
+static void lpool_addr(lpool *lp, char* a, size_t s) {
+ strncpy(a, "invalid:address", s);
+ listener_ctx *ctx = lpool_pick(lp);
+ if (ctx) {
+ pthread_mutex_lock(&ctx->lock);
+ strncpy(a, ctx->addr, s);
+ pthread_mutex_unlock(&ctx->lock);
+ lpool_unref(ctx);
+ }
+}
+
+void lpool_close(lpool *lp) {
+ if (!action_enabled[A_LCLOSE]) return;
+ listener_ctx *ctx = lpool_pick(lp);
+ if (ctx) {
+ pthread_mutex_lock(&ctx->lock);
+ if (ctx->pn_listener) {
+ pn_listener_close(ctx->pn_listener);
+ debuga(A_LCLOSE, ctx->pn_listener);
+ }
+ pthread_mutex_unlock(&ctx->lock);
+ lpool_unref(ctx);
+ }
+}
+
+/* Global state */
+
+typedef struct {
+ pn_proactor_t *proactor;
+ int threads;
+ lpool listeners; /* waiting for PN_PROACTOR_LISTEN */
+ cpool connections_active; /* active in the proactor */
+ cpool connections_idle; /* released and can be reused */
+
+ pthread_mutex_t lock;
+ bool shutdown;
+} global;
+
+void global_init(global *g, int threads) {
+ memset(g, 0, sizeof(*g));
+ g->proactor = pn_proactor();
+ g->threads = threads;
+ lpool_init(&g->listeners, g->threads/2, listener_ctx_free);
+ cpool_init(&g->connections_active, g->threads/2, connection_ctx_free);
+ cpool_init(&g->connections_idle, g->threads/2, connection_ctx_free);
+ pthread_mutex_init(&g->lock, NULL);
+}
+
+void global_destroy(global *g) {
+ pn_proactor_free(g->proactor);
+ lpool_destroy(&g->listeners);
+ cpool_destroy(&g->connections_active);
+ cpool_destroy(&g->connections_idle);
+ pthread_mutex_destroy(&g->lock);
+}
+
+bool global_get_shutdown(global *g) {
+ pthread_mutex_lock(&g->lock);
+ bool ret = g->shutdown;
+ pthread_mutex_unlock(&g->lock);
+ return ret;
+}
+
+void global_set_shutdown(global *g, bool set) {
+ pthread_mutex_lock(&g->lock);
+ g->shutdown = set;
+ pthread_mutex_unlock(&g->lock);
+}
+
+void global_connect(global *g) {
+ char a[PN_MAX_ADDR];
+ lpool_addr(&g->listeners, a, sizeof(a));
+ cpool_connect(&g->connections_active, g->proactor, a);
+}
+
+/* Return true with given probability */
+static bool maybe(double probability) {
+ if (probability == 1.0) return true;
+ return rand() < (probability * RAND_MAX);
+}
+
+/* Run random activities that can be done from any thread. */
+static void global_do_stuff(global *g) {
+ if (maybe(0.5)) global_connect(g);
+ if (maybe(0.3)) lpool_listen(&g->listeners, g->proactor);
+ if (maybe(0.5)) cpool_wake(&g->connections_active);
+ if (maybe(0.5)) cpool_wake(&g->connections_idle);
+ if (maybe(0.1)) lpool_close(&g->listeners);
+ if (action_enabled[A_TIMEOUT] && maybe(0.5)) {
+ debuga(A_TIMEOUT, g->proactor);
+ pn_proactor_set_timeout(g->proactor, rand() % TIMEOUT_MAX);
+ }
+}
+
+static void* user_thread(void* void_g) {
+ debug("user_thread start");
+ global *g = (global*)void_g;
+ while (!global_get_shutdown(g)) {
+ global_do_stuff(g);
+ millisleep(rand() % SLEEP_MAX);
+ }
+ debug("user_thread end");
+ return NULL;
+}
+
+static bool handle(global *g, pn_event_t *e) {
+ switch (pn_event_type(e)) {
+
+ case PN_PROACTOR_TIMEOUT: {
+ if (global_get_shutdown(g)) return false;
+ global_do_stuff(g);
+ break;
+ }
+ case PN_LISTENER_OPEN: {
+ listener_ctx *ctx = (listener_ctx*)pn_listener_get_context(pn_event_listener(e));
+ listener_ctx_on_open(ctx);
+ cpool_connect(&g->connections_active, g->proactor, ctx->addr); /* Initial connection */
+ break;
+ }
+ case PN_LISTENER_CLOSE: {
+ listener_ctx_on_close((listener_ctx*)pn_listener_get_context(pn_event_listener(e)));
+ break;
+ }
+ case PN_TRANSPORT_CLOSED: {
+ connection_ctx_on_close((connection_ctx*)pn_connection_get_context(pn_event_connection(e)));
+ break;
+ }
+ case PN_PROACTOR_INACTIVE: /* Shutting down */
+ pn_proactor_interrupt(g->proactor); /* Interrupt remaining threads */
+ return false;
+
+ case PN_PROACTOR_INTERRUPT:
+ pn_proactor_interrupt(g->proactor); /* Pass the interrupt along */
+ return false;
+
+ /* FIXME aconway 2018-03-09: MORE */
+
+ default:
+ break;
+ }
+ return true;
+}
+
+static void* proactor_thread(void* void_g) {
+ debug("proactor_thread start");
+ global *g = (global*) void_g;
+ bool ok = true;
+ while (ok) {
+ pn_event_batch_t *events = pn_proactor_wait(g->proactor);
+ pn_event_t *e;
+ while (ok && (e = pn_event_batch_next(events))) {
+ ok = ok && handle(g, e);
+ }
+ pn_proactor_done(g->proactor, events);
+ }
+ debug("proactor_thread end");
+ return NULL;
+}
+
+/* Command line arguments */
+
+static const int default_runtime = 1;
+static const int default_threads = 8;
+
+void usage(const char **argv, const char **arg, const char **end) {
+ fprintf(stderr, "usage: %s [options]\n", argv[0]);
+ fprintf(stderr, " -time TIME: total run-time in seconds (default %d)\n", default_runtime);
+ fprintf(stderr, " -threads THREADS: total number of threads (default %d)\n", default_threads);
+ fprintf(stderr, " -debug: print debug messages\n");
+ fprintf(stderr, " ");
+ for (int i = 0; i < (int)action_size; ++i) fprintf(stderr, " -%s", action_name[i]);
+ fprintf(stderr, ": enable actions\n\n");
+
+ fprintf(stderr, "bad argument: ");
+ for (const char **a = argv+1; a < arg; ++a) fprintf(stderr, "%s ", *a);
+ fprintf(stderr, ">>> %s <<<", *arg++);
+ for (; arg < end; ++arg) fprintf(stderr, " %s", *arg);
+ exit(1);
+}
+
+int main(int argc, const char* argv[]) {
+ const char **arg = argv + 1;
+ const char **end = argv + argc;
+ int runtime = default_runtime;
+ int threads = default_threads;
+
+ while (arg < end) {
+ if (!strcmp(*arg, "-time") && ++arg < end) {
+ runtime = atoi(*arg);
+ if (runtime <= 0) usage(argv, arg, end);
+ }
+ else if (!strcmp(*arg, "-threads") && ++arg < end) {
+ threads = atoi(*arg);
+ if (threads <= 0) usage(argv, arg, end);
+ if (threads % 2) threads += 1; /* Round up to even: half proactor, half user */
+ }
+ else if (!strcmp(*arg, "-debug")) {
+ debug_enable = true;
+ }
+ else if (**arg == '-') {
+ size_t i = 0;
+ while (i < action_size && strcmp((*arg)+1, action_name[i])) ++i;
+ if (i == action_size) usage(argv, arg, end);
+ action_enabled[i] = true;
+ } else {
+ break;
+ }
+ ++arg;
+ }
+ int i = 0;
+ /* If no actions are requested on command line, enable them all */
+ while (i < (int)action_size && !action_enabled[i]) ++i;
+ if (i == action_size) {
+ for (i = 0; i < (int)action_size; ++i) action_enabled[i] = true;
+ }
+
+ /* Set up global state, start threads */
+ debug("threaderciser start threads=%d, time=%d\n", threads, runtime);
+ global g;
+ global_init(&g, threads);
+ lpool_listen(&g.listeners, g.proactor); /* Start initial listener */
+
+ pthread_t *user_threads = (pthread_t*)calloc(threads/2, sizeof(pthread_t));
+ pthread_t *proactor_threads = (pthread_t*)calloc(threads/2, sizeof(pthread_t));
+ for (i = 0; i < threads/2; ++i) {
+ pthread_create(&user_threads[i], NULL, user_thread, &g);
+ pthread_create(&proactor_threads[i], NULL, proactor_thread, &g);
+ }
+ millisleep(runtime*1000);
+
+ debug("shut down");
+ global_set_shutdown(&g, true);
+ void *ignore;
+ for (i = 0; i < threads/2; ++i) pthread_join(user_threads[i], &ignore);
+ debug("disconnect");
+ pn_proactor_disconnect(g.proactor, NULL);
+ for (i = 0; i < threads/2; ++i) pthread_join(proactor_threads[i], &ignore);
+
+ free(user_threads);
+ free(proactor_threads);
+ global_destroy(&g);
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/8a4fac69/c/tests/threaderciser.supp
----------------------------------------------------------------------
diff --git a/c/tests/threaderciser.supp b/c/tests/threaderciser.supp
new file mode 100644
index 0000000..2428038
--- /dev/null
+++ b/c/tests/threaderciser.supp
@@ -0,0 +1,5 @@
+{
+ IGNORE: benign race in pni_log_enabled
+ Helgrind:Race
+ fun:pni_log_enabled
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org