Implement ethread events with timeout

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);