aboutsummaryrefslogtreecommitdiffstats
path: root/erts/include/internal/pthread
diff options
context:
space:
mode:
authorRickard Green <[email protected]>2015-02-13 03:18:16 +0100
committerRickard Green <[email protected]>2015-03-20 15:28:52 +0100
commitfa7b2c00cbf9212c4a3551980939b92fc6606510 (patch)
treee3978ac46d61aff7b6f66e67fd763de57a4414bd /erts/include/internal/pthread
parent6487aac5977cf470bc6a2cd0964da2850ee38717 (diff)
downloadotp-fa7b2c00cbf9212c4a3551980939b92fc6606510.tar.gz
otp-fa7b2c00cbf9212c4a3551980939b92fc6606510.tar.bz2
otp-fa7b2c00cbf9212c4a3551980939b92fc6606510.zip
Implement ethread events with timeout
Diffstat (limited to 'erts/include/internal/pthread')
-rw-r--r--erts/include/internal/pthread/ethr_event.h59
1 files changed, 42 insertions, 17 deletions
diff --git a/erts/include/internal/pthread/ethr_event.h b/erts/include/internal/pthread/ethr_event.h
index d0a77990cc..f67bac858b 100644
--- a/erts/include/internal/pthread/ethr_event.h
+++ b/erts/include/internal/pthread/ethr_event.h
@@ -46,12 +46,12 @@ typedef struct {
ethr_atomic32_t futex;
} ethr_event;
-#define ETHR_FUTEX__(FTX, OP, VAL) \
+#define ETHR_FUTEX__(FTX, OP, VAL, TIMEOUT) \
(-1 == syscall(__NR_futex, \
(void *) ethr_atomic32_addr((FTX)), \
(OP), \
(int) (VAL), \
- NULL, \
+ (TIMEOUT), \
NULL, \
0) \
? errno : 0)
@@ -64,7 +64,7 @@ ETHR_INLINE_FUNC_NAME_(ethr_event_set)(ethr_event *e)
ethr_sint32_t val;
val = ethr_atomic32_xchg_mb(&e->futex, ETHR_EVENT_ON__);
if (val == ETHR_EVENT_OFF_WAITER__) {
- int res = ETHR_FUTEX__(&e->futex, ETHR_FUTEX_WAKE__, 1);
+ int res = ETHR_FUTEX__(&e->futex, ETHR_FUTEX_WAKE__, 1, NULL);
if (res != 0)
ETHR_FATAL_ERROR__(res);
}
@@ -80,35 +80,58 @@ ETHR_INLINE_FUNC_NAME_(ethr_event_reset)(ethr_event *e)
#endif
#elif defined(ETHR_PTHREADS)
-/* --- Posix mutex/cond implementation of events ---------------------------- */
+/* --- Posix mutex/cond pipe/select implementation of events ---------------- */
+
typedef struct {
ethr_atomic32_t state;
pthread_mutex_t mtx;
pthread_cond_t cnd;
+ int fd[2];
} ethr_event;
-#define ETHR_EVENT_OFF_WAITER__ -1L
-#define ETHR_EVENT_OFF__ 1L
-#define ETHR_EVENT_ON__ 0L
+#define ETHR_EVENT_OFF_WAITER_SELECT__ ((ethr_sint32_t) -2)
+#define ETHR_EVENT_OFF_WAITER__ ((ethr_sint32_t) -1)
+#define ETHR_EVENT_OFF__ ((ethr_sint32_t) 1)
+#define ETHR_EVENT_ON__ ((ethr_sint32_t) 0)
+
+#define ETHR_EVENT_IS_WAITING__(VAL) ((VAL) < 0)
#if defined(ETHR_TRY_INLINE_FUNCS) || defined(ETHR_EVENT_IMPL__)
+#ifndef ETHR_HAVE_PTHREAD_TIMED_COND_MONOTONIC
+#include <unistd.h>
+#include <errno.h>
+#endif
+
static void ETHR_INLINE
ETHR_INLINE_FUNC_NAME_(ethr_event_set)(ethr_event *e)
{
ethr_sint32_t val;
val = ethr_atomic32_xchg_mb(&e->state, ETHR_EVENT_ON__);
- if (val == ETHR_EVENT_OFF_WAITER__) {
- int res = pthread_mutex_lock(&e->mtx);
- if (res != 0)
- ETHR_FATAL_ERROR__(res);
- res = pthread_cond_signal(&e->cnd);
- if (res != 0)
- ETHR_FATAL_ERROR__(res);
- res = pthread_mutex_unlock(&e->mtx);
- if (res != 0)
- ETHR_FATAL_ERROR__(res);
+ if (ETHR_EVENT_IS_WAITING__(val)) {
+ int res;
+ if (val == ETHR_EVENT_OFF_WAITER_SELECT__) {
+ ssize_t wres;
+ int fd = e->fd[1];
+ ETHR_ASSERT(fd >= 0);
+ do {
+ wres = write(fd, "!", 1);
+ } while (wres < 0 && errno == EINTR);
+ if (wres < 0 && errno != EAGAIN && errno != EWOULDBLOCK)
+ ETHR_FATAL_ERROR__(errno);
+ }
+ else {
+ res = pthread_mutex_lock(&e->mtx);
+ if (res != 0)
+ ETHR_FATAL_ERROR__(res);
+ res = pthread_cond_signal(&e->cnd);
+ if (res != 0)
+ ETHR_FATAL_ERROR__(res);
+ res = pthread_mutex_unlock(&e->mtx);
+ if (res != 0)
+ ETHR_FATAL_ERROR__(res);
+ }
}
}
@@ -127,6 +150,8 @@ int ethr_event_init(ethr_event *e);
int ethr_event_destroy(ethr_event *e);
int ethr_event_wait(ethr_event *e);
int ethr_event_swait(ethr_event *e, int spincount);
+int ethr_event_twait(ethr_event *e, ethr_sint64_t timeout);
+int ethr_event_stwait(ethr_event *e, int spincount, ethr_sint64_t timeout);
#if !defined(ETHR_TRY_INLINE_FUNCS) || defined(ETHR_EVENT_IMPL__)
void ethr_event_set(ethr_event *e);
void ethr_event_reset(ethr_event *e);