Implement ethread events with timeout

4 files changed, 73 insertions, 19 deletions

diff --git a/erts/include/internal/ethr_internal.h b/erts/include/internal/ethr_internal.h
index c9b1db5b46..65195145af 100644
--- a/erts/include/internal/ethr_internal.h
+++ b/erts/include/internal/ethr_internal.h
@@ -57,6 +57,33 @@ ETHR_PROTO_NORETURN__ ethr_abort__(void);
 int ethr_win_get_errno__(void);
 #endif
 
+#ifdef ETHR_INCLUDE_MONOTONIC_CLOCK__
+#undef ETHR_HAVE_ETHR_GET_MONOTONIC_TIME
+#if defined(ETHR_HAVE_CLOCK_GETTIME_MONOTONIC)		\
+    || defined(ETHR_HAVE_MACH_CLOCK_GET_TIME)		\
+    || defined(ETHR_HAVE_GETHRTIME)
+#ifdef ETHR_TIME_WITH_SYS_TIME
+#  include <time.h>
+#  include <sys/time.h>
+#else
+#  ifdef ETHR_HAVE_SYS_TIME_H
+#    include <sys/time.h>
+#  else
+#    include <time.h>
+#  endif
+#endif
+#ifdef ETHR_HAVE_MACH_CLOCK_GET_TIME
+#include <mach/clock.h>
+#include <mach/mach.h>
+#endif
+#define ETHR_HAVE_ETHR_GET_MONOTONIC_TIME
+ethr_sint64_t ethr_get_monotonic_time(void);
+int ethr_get_monotonic_time_is_broken(void);
+#endif
+#endif /* ETHR_INCLUDE_MONOTONIC_CLOCK__ */
+
+void ethr_init_event__(void);
+
 /* implemented in lib_src/common/ethread_aux.c */
 int ethr_init_common__(ethr_init_data *id);
 int ethr_late_init_common__(ethr_late_init_data *lid);
diff --git a/erts/include/internal/ethread_header_config.h.in b/erts/include/internal/ethread_header_config.h.in
index ffd9d350e2..618fcba380 100644
--- a/erts/include/internal/ethread_header_config.h.in
+++ b/erts/include/internal/ethread_header_config.h.in
@@ -250,5 +250,5 @@
 /* Define to the monotonic clock id to use */
 #undef ETHR_MONOTONIC_CLOCK_ID
 
-
-
+/* Define if pthread_cond_timedwait() can be used with a monotonic clock */
+#undef ETHR_HAVE_PTHREAD_COND_TIMEDWAIT_MONOTONIC
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);
diff --git a/erts/include/internal/win/ethr_event.h b/erts/include/internal/win/ethr_event.h
index 6363174a74..95e681983f 100644
--- a/erts/include/internal/win/ethr_event.h
+++ b/erts/include/internal/win/ethr_event.h
@@ -58,6 +58,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);