From 30ee992762972e3dbf8c8440b9f352b33b79f38f Mon Sep 17 00:00:00 2001 From: "liang.he" Date: Wed, 9 Nov 2022 12:50:58 +0800 Subject: [PATCH] Add wasm_module_obtain() to clone wasm_module_t (#1660) Also add support for wasm_shared_module_t, and add lock for wasm_module_t related operations. And add wasm-c-api sample threads, update sample clone. --- core/iwasm/common/wasm_c_api.c | 235 ++++++++++++++++++------ core/iwasm/common/wasm_c_api_internal.h | 4 +- core/iwasm/include/wasm_c_api.h | 6 + samples/wasm-c-api/CMakeLists.txt | 1 + samples/wasm-c-api/src/clone.c | 86 +++++---- samples/wasm-c-api/src/threads.c | 185 +++++++++++++++++++ samples/wasm-c-api/src/threads.wat | 5 + 7 files changed, 425 insertions(+), 97 deletions(-) create mode 100644 samples/wasm-c-api/src/threads.c create mode 100644 samples/wasm-c-api/src/threads.wat diff --git a/core/iwasm/common/wasm_c_api.c b/core/iwasm/common/wasm_c_api.c index 9077b9f3e..a38958364 100644 --- a/core/iwasm/common/wasm_c_api.c +++ b/core/iwasm/common/wasm_c_api.c @@ -17,10 +17,24 @@ #endif /*WASM_ENABLE_JIT != 0 && WASM_ENABLE_LAZY_JIT == 0*/ #endif /*WASM_ENABLE_AOT != 0*/ +/* + * Thread Model: + * - Only one wasm_engine_t in one process + * - One wasm_store_t is only accessed by one thread. wasm_store_t can't be + * shared in threads + * - wasm_module_t can be shared in threads + * - wasm_instance_t can not be shared in threads + */ + #define ASSERT_NOT_IMPLEMENTED() bh_assert(!"not implemented") #define UNREACHABLE() bh_assert(!"unreachable") -typedef struct wasm_module_ex_t wasm_module_ex_t; +typedef struct wasm_module_ex_t { + struct WASMModuleCommon *module_comm_rt; + wasm_byte_vec_t *binary; + korp_mutex lock; + uint32 ref_count; +} wasm_module_ex_t; static void wasm_module_delete_internal(wasm_module_t *); @@ -261,7 +275,24 @@ static void wasm_engine_delete_internal(wasm_engine_t *engine) { if (engine) { - DEINIT_VEC(engine->stores, wasm_store_vec_delete); + if (engine->stores) { + bh_vector_destroy((Vector *)engine->stores); + wasm_runtime_free(engine->stores); + } + + /* clean all created wasm_module_t and their locks */ + { + unsigned i; + for (i = 0; i < engine->modules.num_elems; i++) { + wasm_module_ex_t *module; + if (bh_vector_get(&engine->modules, i, &module)) { + os_mutex_destroy(&module->lock); + wasm_runtime_free(module); + } + } + bh_vector_destroy(&engine->modules); + } + wasm_runtime_free(engine); } @@ -319,9 +350,14 @@ wasm_engine_new_internal(mem_alloc_type_t type, const MemAllocOption *opts) goto failed; } - /* create wasm_store_vec_t */ INIT_VEC(engine->stores, wasm_store_vec_new_uninitialized, 1); + if (!bh_vector_init(&engine->modules, DEFAULT_VECTOR_INIT_SIZE, + sizeof(wasm_module_ex_t *), true)) + goto failed; + + engine->ref_count = 1; + RETURN_OBJ(engine, wasm_engine_delete_internal) } @@ -345,13 +381,10 @@ wasm_engine_new_with_args(mem_alloc_type_t type, const MemAllocOption *opts) os_mutex_lock(&engine_lock); #endif - if (!singleton_engine) { + if (!singleton_engine) singleton_engine = wasm_engine_new_internal(type, opts); - } - - if (singleton_engine) { + else singleton_engine->ref_count++; - } #if defined(OS_THREAD_MUTEX_INITIALIZER) os_mutex_unlock(&engine_lock); @@ -376,23 +409,27 @@ wasm_engine_new_with_config(own wasm_config_t *config) void wasm_engine_delete(wasm_engine_t *engine) { + if (!engine) + return; + #if defined(OS_THREAD_MUTEX_INITIALIZER) os_mutex_lock(&engine_lock); #endif - if (!engine || !singleton_engine || engine != singleton_engine) { + if (!singleton_engine) { #if defined(OS_THREAD_MUTEX_INITIALIZER) os_mutex_unlock(&engine_lock); #endif return; } - if (singleton_engine->ref_count > 0) { - singleton_engine->ref_count--; - if (singleton_engine->ref_count == 0) { - wasm_engine_delete_internal(engine); - singleton_engine = NULL; - } + bh_assert(engine == singleton_engine); + bh_assert(singleton_engine->ref_count > 0); + + singleton_engine->ref_count--; + if (singleton_engine->ref_count == 0) { + wasm_engine_delete_internal(engine); + singleton_engine = NULL; } #if defined(OS_THREAD_MUTEX_INITIALIZER) @@ -446,29 +483,10 @@ failed: void wasm_store_delete(wasm_store_t *store) { - size_t i, store_count; - if (!store) { return; } - /* remove it from the list in the engine */ - store_count = bh_vector_size((Vector *)singleton_engine->stores); - for (i = 0; i != store_count; ++i) { - wasm_store_t *tmp; - - if (!bh_vector_get((Vector *)singleton_engine->stores, (uint32)i, - &tmp)) { - break; - } - - if (tmp == store) { - bh_vector_remove((Vector *)singleton_engine->stores, (uint32)i, - NULL); - break; - } - } - DEINIT_VEC(store->modules, wasm_module_vec_delete); DEINIT_VEC(store->instances, wasm_instance_vec_delete); if (store->foreigns) { @@ -477,7 +495,6 @@ wasm_store_delete(wasm_store_t *store) } wasm_runtime_free(store); - wasm_runtime_destroy_thread_env(); } @@ -1858,11 +1875,6 @@ wasm_foreign_delete(wasm_foreign_t *foreign) } } -struct wasm_module_ex_t { - struct WASMModuleCommon *module_comm_rt; - wasm_byte_vec_t *binary; -}; - static inline wasm_module_t * module_ext_to_module(wasm_module_ex_t *module_ex) { @@ -1888,20 +1900,18 @@ wasm_module_new(wasm_store_t *store, const wasm_byte_vec_t *binary) { char error_buf[128] = { 0 }; wasm_module_ex_t *module_ex = NULL; - PackageType pkg_type; bh_assert(singleton_engine); - if (!store || !binary || binary->size > UINT32_MAX) { - LOG_ERROR("%s failed", __FUNCTION__); - return NULL; - } - - pkg_type = get_package_type((uint8 *)binary->data, (uint32)binary->size); + if (!store || !binary || binary->size == 0 || binary->size > UINT32_MAX) + goto quit; /* whether the combination of compilation flags are compatable with the * package type */ { + PackageType pkg_type; + pkg_type = + get_package_type((uint8 *)binary->data, (uint32)binary->size); bool result = false; #if WASM_ENABLE_INTERP != 0 result = (pkg_type == Wasm_Module_Bytecode); @@ -1913,34 +1923,58 @@ wasm_module_new(wasm_store_t *store, const wasm_byte_vec_t *binary) if (!result) { LOG_VERBOSE("current building isn't compatiable with the module," "may need recompile"); + goto quit; } } module_ex = malloc_internal(sizeof(wasm_module_ex_t)); - if (!module_ex) { - goto failed; - } + if (!module_ex) + goto quit; - INIT_VEC(module_ex->binary, wasm_byte_vec_new, binary->size, binary->data); + module_ex->binary = malloc_internal(sizeof(wasm_byte_vec_t)); + if (!module_ex->binary) + goto free_module; + + wasm_byte_vec_copy(module_ex->binary, binary); + if (!module_ex->binary->data) + goto free_binary; module_ex->module_comm_rt = wasm_runtime_load( (uint8 *)module_ex->binary->data, (uint32)module_ex->binary->size, error_buf, (uint32)sizeof(error_buf)); if (!(module_ex->module_comm_rt)) { LOG_ERROR(error_buf); - goto failed; + goto free_vec; } - /* add it to a watching list in store */ - if (!bh_vector_append((Vector *)store->modules, &module_ex)) { - goto failed; - } + /* append it to a watching list in store */ + if (!bh_vector_append((Vector *)store->modules, &module_ex)) + goto unload; + if (os_mutex_init(&module_ex->lock) != BHT_OK) + goto remove_last; + + if (!bh_vector_append(&singleton_engine->modules, &module_ex)) + goto destroy_lock; + + module_ex->ref_count = 1; return module_ext_to_module(module_ex); -failed: +destroy_lock: + os_mutex_destroy(&module_ex->lock); +remove_last: + bh_vector_remove((Vector *)store->modules, store->modules->num_elems - 1, + NULL); +unload: + wasm_runtime_unload(module_ex->module_comm_rt); +free_vec: + wasm_byte_vec_delete(module_ex->binary); +free_binary: + wasm_runtime_free(module_ex->binary); +free_module: + wasm_runtime_free(module_ex); +quit: LOG_ERROR("%s failed", __FUNCTION__); - wasm_module_delete_internal(module_ext_to_module(module_ex)); return NULL; } @@ -1978,6 +2012,16 @@ wasm_module_delete_internal(wasm_module_t *module) } module_ex = module_to_module_ext(module); + + os_mutex_lock(&module_ex->lock); + + /* N -> N-1 -> 0 -> UINT32_MAX */ + module_ex->ref_count--; + if (module_ex->ref_count > 0) { + os_mutex_unlock(&module_ex->lock); + return; + } + DEINIT_VEC(module_ex->binary, wasm_byte_vec_delete); if (module_ex->module_comm_rt) { @@ -1985,13 +2029,14 @@ wasm_module_delete_internal(wasm_module_t *module) module_ex->module_comm_rt = NULL; } - wasm_runtime_free(module_ex); + os_mutex_unlock(&module_ex->lock); } void wasm_module_delete(wasm_module_t *module) { /* the module will be released when releasing the store */ + (void)module; } void @@ -2007,6 +2052,9 @@ wasm_module_imports(const wasm_module_t *module, own wasm_importtype_vec_t *out) return; } + if (((const wasm_module_ex_t *)(module))->ref_count == 0) + return; + #if WASM_ENABLE_INTERP != 0 if ((*module)->module_type == Wasm_Module_Bytecode) { import_func_count = MODULE_INTERP(module)->import_function_count; @@ -2243,6 +2291,9 @@ wasm_module_exports(const wasm_module_t *module, wasm_exporttype_vec_t *out) return; } + if (((const wasm_module_ex_t *)(module))->ref_count == 0) + return; + #if WASM_ENABLE_INTERP != 0 if ((*module)->module_type == Wasm_Module_Bytecode) { export_count = MODULE_INTERP(module)->export_count; @@ -2425,6 +2476,9 @@ wasm_module_serialize(wasm_module_t *module, own wasm_byte_vec_t *out) if (!module || !out) return; + if (((const wasm_module_ex_t *)(module))->ref_count == 0) + return; + module_ex = module_to_module_ext(module); comp_ctx = ((WASMModule *)(module_ex->module_comm_rt))->comp_ctx; comp_data = ((WASMModule *)(module_ex->module_comm_rt))->comp_data; @@ -2446,6 +2500,69 @@ wasm_module_deserialize(wasm_store_t *store, const wasm_byte_vec_t *binary) } #endif +wasm_module_t * +wasm_module_obtain(wasm_store_t *store, wasm_shared_module_t *shared_module) +{ + wasm_module_ex_t *module_ex = NULL; + + if (!store || !shared_module) + return NULL; + + module_ex = (wasm_module_ex_t *)shared_module; + + os_mutex_lock(&module_ex->lock); + + /* deleting the module... */ + if (module_ex->ref_count == 0) { + LOG_WARNING("wasm_module_obtain re-enter a module under deleting."); + os_mutex_unlock(&module_ex->lock); + return NULL; + } + + /* add it to a watching list in store */ + if (!bh_vector_append((Vector *)store->modules, &module_ex)) { + os_mutex_unlock(&module_ex->lock); + return NULL; + } + + module_ex->ref_count++; + os_mutex_unlock(&module_ex->lock); + + return (wasm_module_t *)shared_module; +} + +wasm_shared_module_t * +wasm_module_share(wasm_module_t *module) +{ + wasm_module_ex_t *module_ex = NULL; + + if (!module) + return NULL; + + module_ex = (wasm_module_ex_t *)module; + + os_mutex_lock(&module_ex->lock); + + /* deleting the module... */ + if (module_ex->ref_count == 0) { + LOG_WARNING("wasm_module_share re-enter a module under deleting."); + os_mutex_unlock(&module_ex->lock); + return NULL; + } + + module_ex->ref_count++; + + os_mutex_unlock(&module_ex->lock); + + return (wasm_shared_module_t *)module; +} + +void +wasm_shared_module_delete(own wasm_shared_module_t *shared_module) +{ + return wasm_module_delete_internal((wasm_module_t *)shared_module); +} + static wasm_func_t * wasm_func_new_basic(wasm_store_t *store, const wasm_functype_t *type, wasm_func_callback_t func_callback) diff --git a/core/iwasm/common/wasm_c_api_internal.h b/core/iwasm/common/wasm_c_api_internal.h index ac1c93313..2733fa143 100644 --- a/core/iwasm/common/wasm_c_api_internal.h +++ b/core/iwasm/common/wasm_c_api_internal.h @@ -25,12 +25,14 @@ WASM_DECLARE_VEC(store, *) /* Runtime Environment */ struct wasm_engine_t { - /* support one store for now */ wasm_store_vec_t *stores; uint32 ref_count; + /* list of wasm_module_ex_t */ + Vector modules; }; struct wasm_store_t { + /* maybe should remove the list */ wasm_module_vec_t *modules; wasm_instance_vec_t *instances; Vector *foreigns; diff --git a/core/iwasm/include/wasm_c_api.h b/core/iwasm/include/wasm_c_api.h index 62113b20c..39ee01a2d 100644 --- a/core/iwasm/include/wasm_c_api.h +++ b/core/iwasm/include/wasm_c_api.h @@ -464,6 +464,7 @@ struct WASMModuleCommon; typedef struct WASMModuleCommon *wasm_module_t; #endif + WASM_API_EXTERN own wasm_module_t* wasm_module_new( wasm_store_t*, const wasm_byte_vec_t* binary); @@ -477,6 +478,11 @@ WASM_API_EXTERN void wasm_module_exports(const wasm_module_t*, own wasm_exportty WASM_API_EXTERN void wasm_module_serialize(wasm_module_t*, own wasm_byte_vec_t* out); WASM_API_EXTERN own wasm_module_t* wasm_module_deserialize(wasm_store_t*, const wasm_byte_vec_t*); +typedef wasm_module_t wasm_shared_module_t; +WASM_API_EXTERN own wasm_shared_module_t* wasm_module_share(wasm_module_t*); +WASM_API_EXTERN own wasm_module_t* wasm_module_obtain(wasm_store_t*, wasm_shared_module_t*); +WASM_API_EXTERN void wasm_shared_module_delete(own wasm_shared_module_t*); + // Function Instances diff --git a/samples/wasm-c-api/CMakeLists.txt b/samples/wasm-c-api/CMakeLists.txt index 5dc770060..867e4723e 100644 --- a/samples/wasm-c-api/CMakeLists.txt +++ b/samples/wasm-c-api/CMakeLists.txt @@ -134,6 +134,7 @@ set(EXAMPLES memory reflect table + threads trap ) diff --git a/samples/wasm-c-api/src/clone.c b/samples/wasm-c-api/src/clone.c index 29534f30f..bfe621758 100644 --- a/samples/wasm-c-api/src/clone.c +++ b/samples/wasm-c-api/src/clone.c @@ -20,6 +20,7 @@ typedef struct _vm { wasm_engine_t *engine; wasm_store_t *store; wasm_module_t *module; + wasm_shared_module_t *shared_module; wasm_instance_t *instance; wasm_func_t **function_list; wasm_memory_t *memory; @@ -72,10 +73,49 @@ fail: return NULL; } +wasm_vm_t * +vm_release(wasm_vm_t *vm) +{ + if (!vm) + return NULL; + + if (vm->function_list) { + free(vm->function_list); + vm->function_list = NULL; + } + + vm->memory = NULL; + + if (vm->exports) { + wasm_extern_vec_delete(vm->exports); + free(vm->exports); + vm->exports = NULL; + } + + wasm_instance_delete(vm->instance); + vm->instance = NULL; + + wasm_shared_module_delete(vm->shared_module); + vm->shared_module = NULL; + + wasm_module_delete(vm->module); + vm->module = NULL; + + wasm_store_delete(vm->store); + vm->store = NULL; + + wasm_engine_delete(vm->engine); + vm->engine = NULL; + + free(vm); + return NULL; +} + bool vm_load(wasm_vm_t *vm, const wasm_byte_vec_t *binary) { vm->module = wasm_module_new(vm->store, binary); + vm->shared_module = wasm_module_share(vm->module); return vm->module != NULL; } @@ -127,7 +167,10 @@ vm_clone_from_module(const wasm_vm_t *base) secondary = vm_new(); if (secondary) { printf("Reuse module and bypass vm_load()..."); - secondary->module = base->module; + secondary->module = + wasm_module_obtain(base->store, base->shared_module); + if (!secondary->module) + secondary = vm_release(secondary); } return secondary; @@ -163,39 +206,6 @@ vm_clone(const wasm_vm_t *base, clone_level level) return vm_clone_from_instance(base); } -wasm_vm_t * -vm_release(wasm_vm_t *vm) -{ - if (!vm) - return NULL; - - if (vm->function_list) { - free(vm->function_list); - vm->function_list = NULL; - } - - if (vm->exports) { - wasm_extern_vec_delete(vm->exports); - free(vm->exports); - vm->exports = NULL; - } - - wasm_instance_delete(vm->instance); - vm->instance = NULL; - - wasm_module_delete(vm->module); - vm->module = NULL; - - wasm_store_delete(vm->store); - vm->store = NULL; - - wasm_engine_delete(vm->engine); - vm->engine = NULL; - - free(vm); - return NULL; -} - bool vm_memory_set_byte(const wasm_vm_t *vm, uint32_t offset, uint8_t byte) { @@ -444,6 +454,8 @@ static void * thrd_func(void *arg) { thread_arg_t *thrd_arg = (thread_arg_t *)arg; + + sleep(rand() % 5); printf("Running warm start at %s...\n", thrd_arg->name); pthread_setspecific(name_key, thrd_arg->name); @@ -483,7 +495,6 @@ main() run_test(base_vm); printf("Running warm start at other threads...\n"); - pthread_mutex_trylock(&ready_go_lock); pthread_t tids[WORKER_NUMBER] = { 0 }; thread_arg_t thrd_args[WORKER_NUMBER] = { 0 }; @@ -501,8 +512,8 @@ main() break; } - sleep(5); - + sleep(1); + pthread_mutex_trylock(&ready_go_lock); ready_go_flag = true; pthread_mutex_unlock(&ready_go_lock); pthread_cond_broadcast(&ready_go_cond); @@ -511,6 +522,7 @@ main() if (tids[i] != 0) pthread_join(tids[i], NULL); } + vm_release(base_vm); ret = EXIT_SUCCESS; quit: diff --git a/samples/wasm-c-api/src/threads.c b/samples/wasm-c-api/src/threads.c new file mode 100644 index 000000000..d7ae48939 --- /dev/null +++ b/samples/wasm-c-api/src/threads.c @@ -0,0 +1,185 @@ +#include +#include +#include +#include +#include +#include + +#include "wasm_c_api.h" + +#define own + +const int N_THREADS = 10; +const int N_REPS = 3; + +// A function to be called from Wasm code. +own wasm_trap_t * +callback(const wasm_val_vec_t *args, wasm_val_vec_t *results) +{ + assert(args->data[0].kind == WASM_I32); + printf("> Thread %d running\n", args->data[0].of.i32); + return NULL; +} + +typedef struct { + wasm_engine_t *engine; + wasm_shared_module_t *module; + int id; +} thread_args; + +void * +run(void *args_abs) +{ + thread_args *args = (thread_args *)args_abs; + + // Rereate store and module. + own wasm_store_t *store = wasm_store_new(args->engine); + own wasm_module_t *module = wasm_module_obtain(store, args->module); + + // Run the example N times. + for (int i = 0; i < N_REPS; ++i) { + usleep(100000); + + // Create imports. + own wasm_functype_t *func_type = + wasm_functype_new_1_0(wasm_valtype_new_i32()); + own wasm_func_t *func = wasm_func_new(store, func_type, callback); + wasm_functype_delete(func_type); + + wasm_val_t val = WASM_I32_VAL((int32_t)args->id); + own wasm_globaltype_t *global_type = + wasm_globaltype_new(wasm_valtype_new_i32(), WASM_CONST); + own wasm_global_t *global = wasm_global_new(store, global_type, &val); + wasm_globaltype_delete(global_type); + + // Instantiate. + wasm_extern_t *externs[] = { + wasm_func_as_extern(func), + wasm_global_as_extern(global), + }; + wasm_extern_vec_t imports = WASM_ARRAY_VEC(externs); + own wasm_instance_t *instance = + wasm_instance_new(store, module, &imports, NULL); + if (!instance) { + printf("> Error instantiating module!\n"); + return NULL; + } + + wasm_func_delete(func); + wasm_global_delete(global); + + // Extract export. + own wasm_extern_vec_t exports; + wasm_instance_exports(instance, &exports); + if (exports.size == 0) { + printf("> Error accessing exports!\n"); + return NULL; + } + const wasm_func_t *run_func = wasm_extern_as_func(exports.data[0]); + if (run_func == NULL) { + printf("> Error accessing export!\n"); + return NULL; + } + + wasm_instance_delete(instance); + + // Call. + wasm_val_vec_t empty = WASM_EMPTY_VEC; + if (wasm_func_call(run_func, &empty, &empty)) { + printf("> Error calling function!\n"); + return NULL; + } + + wasm_extern_vec_delete(&exports); + } + + wasm_module_delete(module); + wasm_store_delete(store); + + free(args_abs); + + return NULL; +} + +int +main(int argc, const char *argv[]) +{ + // Initialize. + wasm_engine_t *engine = wasm_engine_new(); + + // Load binary. +#if WASM_ENABLE_AOT != 0 && WASM_ENABLE_INTERP == 0 + FILE *file = fopen("threads.aot", "rb"); +#else + FILE *file = fopen("threads.wasm", "rb"); +#endif + if (!file) { + printf("> Error loading module!\n"); + return 1; + } + + int ret = fseek(file, 0L, SEEK_END); + if (ret == -1) { + printf("> Error loading module!\n"); + fclose(file); + return 1; + } + + long file_size = ftell(file); + if (file_size == -1) { + printf("> Error loading module!\n"); + fclose(file); + return 1; + } + + ret = fseek(file, 0L, SEEK_SET); + if (ret == -1) { + printf("> Error loading module!\n"); + fclose(file); + return 1; + } + + wasm_byte_vec_t binary; + wasm_byte_vec_new_uninitialized(&binary, file_size); + if (fread(binary.data, file_size, 1, file) != 1) { + printf("> Error loading module!\n"); + return 1; + } + fclose(file); + + // Compile and share. + own wasm_store_t *store = wasm_store_new(engine); + own wasm_module_t *module = wasm_module_new(store, &binary); + if (!module) { + printf("> Error compiling module!\n"); + return 1; + } + + wasm_byte_vec_delete(&binary); + + own wasm_shared_module_t *shared = wasm_module_share(module); + + wasm_module_delete(module); + wasm_store_delete(store); + + // Spawn threads. + pthread_t threads[N_THREADS]; + for (int i = 0; i < N_THREADS; i++) { + thread_args *args = malloc(sizeof(thread_args)); + args->id = i; + args->engine = engine; + args->module = shared; + printf("Initializing thread %d...\n", i); + pthread_create(&threads[i], NULL, &run, args); + } + + for (int i = 0; i < N_THREADS; i++) { + printf("Waiting for thread: %d\n", i); + pthread_join(threads[i], NULL); + } + + wasm_shared_module_delete(shared); + wasm_engine_delete(engine); + + return 0; +} diff --git a/samples/wasm-c-api/src/threads.wat b/samples/wasm-c-api/src/threads.wat new file mode 100644 index 000000000..29a3bbcc1 --- /dev/null +++ b/samples/wasm-c-api/src/threads.wat @@ -0,0 +1,5 @@ +(module + (func $message (import "" "hello") (param i32)) + (global $id (import "" "id") i32) + (func (export "run") (call $message (global.get $id))) +)