Merge pull request #815 from bytecodealliance/main

Implement async termination of blocking thread (bytecodealliance#2516)

Author: wenyongh

build-scripts/config_common.cmake

@@ -273,6 +273,13 @@ else ()
     add_definitions (-DWASM_DISABLE_STACK_HW_BOUND_CHECK=0)
   endif ()
 endif ()
+if (WAMR_DISABLE_WAKEUP_BLOCKING_OP EQUAL 1)
+  add_definitions (-DWASM_DISABLE_WAKEUP_BLOCKING_OP=1)
+  message ("     Wakeup of blocking operations disabled")
+else ()
+  add_definitions (-DWASM_DISABLE_WAKEUP_BLOCKING_OP=0)
+  message ("     Wakeup of blocking operations enabled")
+endif ()
 if (WAMR_BUILD_SIMD EQUAL 1)
   if (NOT WAMR_BUILD_TARGET MATCHES "RISCV64.*")
     add_definitions (-DWASM_ENABLE_SIMD=1)

core/iwasm/common/wasm_blocking_op.c

@@ -0,0 +1,92 @@
+/*
+ * Copyright (C) 2023 Midokura Japan KK.  All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
+ */
+
+#include "wasm_runtime_common.h"
+
+#include "bh_platform.h"
+#include "bh_common.h"
+#include "bh_assert.h"
+
+#if WASM_ENABLE_THREAD_MGR != 0 && defined(OS_ENABLE_WAKEUP_BLOCKING_OP)
+
+#define LOCK(env) WASM_SUSPEND_FLAGS_LOCK((env)->wait_lock)
+#define UNLOCK(env) WASM_SUSPEND_FLAGS_UNLOCK((env)->wait_lock)
+
+#define ISSET(env, bit)                                                       \
+    ((WASM_SUSPEND_FLAGS_GET((env)->suspend_flags) & WASM_SUSPEND_FLAG_##bit) \
+     != 0)
+#define SET(env, bit) \
+    WASM_SUSPEND_FLAGS_FETCH_OR((env)->suspend_flags, WASM_SUSPEND_FLAG_##bit)
+#define CLR(env, bit) \
+    WASM_SUSPEND_FLAGS_FETCH_AND((env)->suspend_flags, ~WASM_SUSPEND_FLAG_##bit)
+
+bool
+wasm_runtime_begin_blocking_op(wasm_exec_env_t env)
+{
+    LOCK(env);
+    bh_assert(!ISSET(env, BLOCKING));
+    SET(env, BLOCKING);
+    if (ISSET(env, TERMINATE)) {
+        CLR(env, BLOCKING);
+        UNLOCK(env);
+        return false;
+    }
+    UNLOCK(env);
+    os_begin_blocking_op();
+    return true;
+}
+
+void
+wasm_runtime_end_blocking_op(wasm_exec_env_t env)
+{
+    int saved_errno = errno;
+    LOCK(env);
+    bh_assert(ISSET(env, BLOCKING));
+    CLR(env, BLOCKING);
+    UNLOCK(env);
+    os_end_blocking_op();
+    errno = saved_errno;
+}
+
+void
+wasm_runtime_interrupt_blocking_op(wasm_exec_env_t env)
+{
+    /*
+     * ISSET(BLOCKING) here means that the target thread
+     * is in somewhere between wasm_begin_blocking_op and
+     * wasm_end_blocking_op.
+     * keep waking it up until it reaches wasm_end_blocking_op,
+     * which clears the BLOCKING bit.
+     *
+     * this dumb loop is necessary because posix doesn't provide
+     * a way to unmask signal and block atomically.
+     */
+
+    LOCK(env);
+    SET(env, TERMINATE);
+    while (ISSET(env, BLOCKING)) {
+        UNLOCK(env);
+        os_wakeup_blocking_op(env->handle);
+
+        /* relax a bit */
+        os_usleep(50 * 1000);
+        LOCK(env);
+    }
+    UNLOCK(env);
+}
+
+#else /* WASM_ENABLE_THREAD_MGR && OS_ENABLE_WAKEUP_BLOCKING_OP */
+
+bool
+wasm_runtime_begin_blocking_op(wasm_exec_env_t env)
+{
+    return true;
+}
+
+void
+wasm_runtime_end_blocking_op(wasm_exec_env_t env)
+{}
+
+#endif /* WASM_ENABLE_THREAD_MGR && OS_ENABLE_WAKEUP_BLOCKING_OP */

core/iwasm/common/wasm_runtime_common.c

@@ -457,8 +457,21 @@ wasm_runtime_env_init()
     }
 #endif
 
+#if WASM_ENABLE_THREAD_MGR != 0 && defined(OS_ENABLE_WAKEUP_BLOCKING_OP)
+    if (os_blocking_op_init() != BHT_OK) {
+        goto fail11;
+    }
+    os_end_blocking_op();
+#endif
+
     return true;
 
+#if WASM_ENABLE_THREAD_MGR != 0 && defined(OS_ENABLE_WAKEUP_BLOCKING_OP)
+fail11:
+#if WASM_ENABLE_JIT != 0 || WASM_ENABLE_WAMR_COMPILER != 0
+    aot_compiler_destroy();
+#endif
+#endif
 #if WASM_ENABLE_JIT != 0 || WASM_ENABLE_WAMR_COMPILER != 0
 fail10:
 #if WASM_ENABLE_FAST_JIT != 0
@@ -1392,6 +1405,10 @@ wasm_runtime_init_thread_env(void)
     }
 #endif
 
+#if WASM_ENABLE_THREAD_MGR != 0 && defined(OS_ENABLE_WAKEUP_BLOCKING_OP)
+    os_end_blocking_op();
+#endif
+
     return true;
 }
 

core/iwasm/common/wasm_runtime_common.h

@@ -1040,6 +1040,15 @@ WASM_RUNTIME_API_EXTERN bool
 wasm_runtime_is_import_global_linked(const char *module_name,
                                      const char *global_name);
 
+WASM_RUNTIME_API_EXTERN bool
+wasm_runtime_begin_blocking_op(WASMExecEnv *exec_env);
+
+WASM_RUNTIME_API_EXTERN void
+wasm_runtime_end_blocking_op(WASMExecEnv *exec_env);
+
+void
+wasm_runtime_interrupt_blocking_op(WASMExecEnv *exec_env);
+
 #ifdef __cplusplus
 }
 #endif

core/iwasm/common/wasm_suspend_flags.h

@@ -20,6 +20,8 @@ extern "C" {
 #define WASM_SUSPEND_FLAG_BREAKPOINT 0x4
 /* Return from pthread_exit */
 #define WASM_SUSPEND_FLAG_EXIT 0x8
+/* The thread might be blocking */
+#define WASM_SUSPEND_FLAG_BLOCKING 0x10
 
 typedef union WASMSuspendFlags {
     bh_atomic_32_t flags;

core/iwasm/include/wasm_export.h

@@ -1546,6 +1546,50 @@ wasm_runtime_set_context_spread(wasm_module_inst_t inst, void *key,
 WASM_RUNTIME_API_EXTERN void *
 wasm_runtime_get_context(wasm_module_inst_t inst, void *key);
 
+/*
+ * wasm_runtime_begin_blocking_op/wasm_runtime_end_blocking_op
+ *
+ * These APIs are intended to be used by the implementations of
+ * host functions. It wraps an operation which possibly blocks for long
+ * to prepare for async termination.
+ *
+ * eg.
+ *
+ *   if (!wasm_runtime_begin_blocking_op(exec_env)) {
+ *       return EINTR;
+ *   }
+ *   ret = possibly_blocking_op();
+ *   wasm_runtime_end_blocking_op(exec_env);
+ *   return ret;
+ *
+ * If threading support (WASM_ENABLE_THREAD_MGR) is not enabled,
+ * these functions are no-op.
+ *
+ * If the underlying platform support (OS_ENABLE_WAKEUP_BLOCKING_OP) is
+ * not available, these functions are no-op. In that case, the runtime
+ * might not terminate a blocking thread in a timely manner.
+ *
+ * If the underlying platform support is available, it's used to wake up
+ * the thread for async termination. The expectation here is that a
+ * `os_wakeup_blocking_op` call makes the blocking operation
+ * (`possibly_blocking_op` in the above example) return in a timely manner.
+ *
+ * The actual wake up mechanism used by `os_wakeup_blocking_op` is
+ * platform-dependent. It might impose some platform-dependent restrictions
+ * on the implementation of the blocking opearation.
+ *
+ * For example, on POSIX-like platforms, a signal (by default SIGUSR1) is
+ * used. The signal delivery configurations (eg. signal handler, signal mask,
+ * etc) for the signal are set up by the runtime. You can change the signal
+ * to use for this purpose by calling os_set_signal_number_for_blocking_op
+ * before the runtime initialization.
+ */
+WASM_RUNTIME_API_EXTERN bool
+wasm_runtime_begin_blocking_op(wasm_exec_env_t exec_env);
+
+WASM_RUNTIME_API_EXTERN void
+wasm_runtime_end_blocking_op(wasm_exec_env_t exec_env);
+
 /* clang-format on */
 
 #ifdef __cplusplus