From c4cb0e5e6939b84bfe135bb37db7087d6e0be7e8 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Sun, 17 Nov 2024 15:21:48 +0100 Subject: [PATCH 1/9] Use a mutex to protect async arguments --- src/async.c | 67 +++++++++++++++++++++---- src/luv.c | 1 + src/thread.c | 112 +++++++++++++++++++++++++----------------- tests/test-async.lua | 114 +++++++++++++++++++++++++++++++++++-------- 4 files changed, 220 insertions(+), 74 deletions(-) diff --git a/src/async.c b/src/async.c index db25ce45..f0935196 100644 --- a/src/async.c +++ b/src/async.c @@ -16,6 +16,13 @@ */ #include "private.h" +typedef struct { + luv_thread_arg_t targ; + uv_mutex_t mutex; +} luv_async_arg_t; + +#define luv_get_async_arg_from_handle(H) ((luv_async_arg_t *) ((luv_handle_t*) (H)->data)->extra) + static uv_async_t* luv_check_async(lua_State* L, int index) { uv_async_t* handle = (uv_async_t*)luv_checkudata(L, index, "uv_async"); luaL_argcheck(L, handle->type == UV_ASYNC && handle->data, index, "Expected uv_async_t"); @@ -25,9 +32,19 @@ static uv_async_t* luv_check_async(lua_State* L, int index) { static void luv_async_cb(uv_async_t* handle) { luv_handle_t* data = (luv_handle_t*)handle->data; lua_State* L = data->ctx->L; - int n = luv_thread_arg_push(L, (luv_thread_arg_t*)data->extra, LUVF_THREAD_SIDE_MAIN); - luv_call_callback(L, data, LUV_ASYNC, n); - luv_thread_arg_clear(L, (luv_thread_arg_t*)data->extra, LUVF_THREAD_SIDE_MAIN); + luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); + uv_mutex_t *argmutex = &asarg->mutex; + luv_thread_arg_t targcpy; + int n; + uv_mutex_lock(argmutex); + targcpy = asarg->targ; // work on a copy of the arguments + asarg->targ.argc = 0; // empty the original, nothing to clear + uv_mutex_unlock(argmutex); + n = luv_thread_arg_push(L, &targcpy, LUVF_THREAD_SIDE_MAIN); + if (n >= 0) { + luv_call_callback(L, data, LUV_ASYNC, n); + } + luv_thread_arg_clear(L, &targcpy, LUVF_THREAD_SIDE_MAIN); // clear the copy } static int luv_new_async(lua_State* L) { @@ -43,21 +60,53 @@ static int luv_new_async(lua_State* L) { return luv_error(L, ret); } data = luv_setup_handle(L, ctx); - data->extra = (luv_thread_arg_t*)malloc(sizeof(luv_thread_arg_t)); + luv_async_arg_t* asarg = (luv_async_arg_t*)malloc(sizeof(luv_async_arg_t)); + memset(asarg, 0, sizeof(luv_async_arg_t)); + ret = uv_mutex_init(&asarg->mutex); + if (ret < 0) { // unlikely + abort(); + } + data->extra = asarg; data->extra_gc = free; - memset(data->extra, 0, sizeof(luv_thread_arg_t)); handle->data = data; luv_check_callback(L, (luv_handle_t*)handle->data, LUV_ASYNC, 1); return 1; } + +static int luv_handle_gc(lua_State* L); + +static int luv_async_gc(lua_State* L) { + uv_async_t* handle = *(uv_async_t**)lua_touserdata(L, 1); + luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); + uv_mutex_t *argmutex = &asarg->mutex; + uv_mutex_lock(argmutex); + luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_CHILD); // in case of a pending send, set side to avoid unref + uv_mutex_unlock(argmutex); + uv_mutex_destroy(argmutex); + return luv_handle_gc(L); +} + static int luv_async_send(lua_State* L) { int ret; uv_async_t* handle = luv_check_async(L, 1); - luv_thread_arg_t* arg = (luv_thread_arg_t *)((luv_handle_t*) handle->data)->extra; - - luv_thread_arg_set(L, arg, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); + luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); + uv_mutex_t *argmutex = &asarg->mutex; + int n; + uv_mutex_lock(argmutex); + luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_CHILD); // in case of a pending send + n = luv_thread_arg_set(L, &asarg->targ, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); + uv_mutex_unlock(argmutex); + if (n < 0) { + return luv_thread_arg_error(L); + } ret = uv_async_send(handle); - luv_thread_arg_clear(L, arg, LUVF_THREAD_SIDE_CHILD); return luv_result(L, ret); } + +static void luv_async_init(lua_State* L) { + luaL_getmetatable(L, "uv_async"); + lua_pushcfunction(L, luv_async_gc); + lua_setfield(L, -2, "__gc"); + lua_pop(L, 1); +} diff --git a/src/luv.c b/src/luv.c index 18c01936..c196099c 100644 --- a/src/luv.c +++ b/src/luv.c @@ -908,6 +908,7 @@ LUALIB_API int luaopen_luv (lua_State* L) { luv_req_init(L); luv_handle_init(L); + luv_async_init(L); #if LUV_UV_VERSION_GEQ(1, 28, 0) luv_dir_init(L); #endif diff --git a/src/thread.c b/src/thread.c index 3c21364a..67c4e037 100644 --- a/src/thread.c +++ b/src/thread.c @@ -66,7 +66,11 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int int i; int side = LUVF_THREAD_SIDE(flags); int async = LUVF_THREAD_ASYNC(flags); - + /* + * thread works by reference. + * async works by copy, a use case consists in sending just before the thread ends, + * it results that the callback will be called after the thread/async life. + */ idx = idx > 0 ? idx : 1; i = idx; args->flags = flags; @@ -91,11 +95,12 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int } break; case LUA_TSTRING: - if (async) - { - const char* p = lua_tolstring(L, i, &arg->val.str.len); - arg->val.str.base = malloc(arg->val.str.len); - memcpy((void*)arg->val.str.base, p, arg->val.str.len); + if (async) { + size_t l = 0; + const char* p = lua_tolstring(L, i, &l); + void* b = malloc(l + 1); + arg->val.str.base = memcpy((void*)b, p, l + 1); + arg->val.str.len = l; } else { arg->val.str.base = lua_tolstring(L, i, &arg->val.str.len); lua_pushvalue(L, i); @@ -103,13 +108,27 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int } break; case LUA_TUSERDATA: - arg->val.udata.data = lua_topointer(L, i); - arg->val.udata.size = lua_rawlen(L, i); - arg->val.udata.metaname = luv_getmtname(L, i); - - if (arg->val.udata.size) { - lua_pushvalue(L, i); - arg->ref[side] = luaL_ref(L, LUA_REGISTRYINDEX); + { + const void* p = lua_topointer(L, i); + size_t l = lua_rawlen(L, i); + const char* mtname = luv_getmtname(L, i); + if (async) { + if (l > 0) { + void* b = malloc(l); + p = (const void*)memcpy(b, p, l); + } + if (mtname != NULL) { + size_t ml = strlen(mtname) + 1; + char* b = malloc(ml); + mtname = (const void*)memcpy(b, mtname, ml); + } + } else { + lua_pushvalue(L, i); + arg->ref[side] = luaL_ref(L, LUA_REGISTRYINDEX); + } + arg->val.udata.data = p; + arg->val.udata.size = l; + arg->val.udata.metaname = mtname; } break; default: @@ -127,43 +146,46 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags) { int i; int side = LUVF_THREAD_SIDE(flags); - int set = LUVF_THREAD_SIDE(args->flags); + int setside = LUVF_THREAD_SIDE(args->flags); int async = LUVF_THREAD_ASYNC(args->flags); - - if (args->argc == 0) - return; - + /* + * clear is safe to be called multiple times, values are set to LUA_NOREF or NULL, argc is preserved. + * thread unrefs per side. + * async frees values on the first calling side. + */ for (i = 0; i < args->argc; i++) { luv_val_t* arg = args->argv + i; switch (arg->type) { case LUA_TSTRING: - if (arg->ref[side] != LUA_NOREF) - { + if (arg->ref[side] != LUA_NOREF) { luaL_unref(L, LUA_REGISTRYINDEX, arg->ref[side]); arg->ref[side] = LUA_NOREF; - } else { - if(async && set!=side) - { - free((void*)arg->val.str.base); - arg->val.str.base = NULL; - arg->val.str.len = 0; - } + } + if (async) { + free((void*)arg->val.str.base); + arg->val.str.base = NULL; } break; case LUA_TUSERDATA: - if (arg->ref[side]!=LUA_NOREF) - { - if (side != set) - { + if (arg->ref[side] != LUA_NOREF) { + if (side != setside) { // avoid custom gc lua_rawgeti(L, LUA_REGISTRYINDEX, arg->ref[side]); lua_pushnil(L); lua_setmetatable(L, -2); - lua_pop(L, -1); + lua_pop(L, 1); } luaL_unref(L, LUA_REGISTRYINDEX, arg->ref[side]); arg->ref[side] = LUA_NOREF; } + if (async) { + if (arg->val.udata.size > 0) { + free((void*)arg->val.udata.data); + arg->val.udata.data = NULL; + } + free((void*)arg->val.udata.metaname); + arg->val.udata.metaname = NULL; + } break; default: break; @@ -171,7 +193,6 @@ static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags } } -// called only in thread static int luv_thread_arg_push(lua_State* L, luv_thread_arg_t* args, int flags) { int i = 0; int side = LUVF_THREAD_SIDE(flags); @@ -196,18 +217,16 @@ static int luv_thread_arg_push(lua_State* L, luv_thread_arg_t* args, int flags) lua_pushlstring(L, arg->val.str.base, arg->val.str.len); break; case LUA_TUSERDATA: - if (arg->val.udata.size) - { + if (arg->val.udata.size > 0) { char *p = lua_newuserdata(L, arg->val.udata.size); memcpy(p, arg->val.udata.data, arg->val.udata.size); - if (arg->val.udata.metaname) - { + if (arg->val.udata.metaname != NULL) { luaL_getmetatable(L, arg->val.udata.metaname); lua_setmetatable(L, -2); } lua_pushvalue(L, -1); arg->ref[side] = luaL_ref(L, LUA_REGISTRYINDEX); - }else{ + } else { lua_pushlightuserdata(L, (void*)arg->val.udata.data); } break; @@ -306,9 +325,11 @@ static void luv_thread_cb(void* varg) { static void luv_thread_notify_close_cb(uv_handle_t *handle) { luv_thread_t *thread = handle->data; - if (thread->handle != 0) - uv_thread_join(&thread->handle); - + uv_thread_t uvt = thread->handle; + if (uvt != 0) { + thread->handle = 0; + uv_thread_join(&uvt); + } luaL_unref(thread->L, LUA_REGISTRYINDEX, thread->ref); thread->ref = LUA_NOREF; thread->L = NULL; @@ -494,9 +515,12 @@ static int luv_thread_setpriority(lua_State* L) { static int luv_thread_join(lua_State* L) { luv_thread_t* tid = luv_check_thread(L, 1); - int ret = uv_thread_join(&tid->handle); - if (ret < 0) return luv_error(L, ret); - tid->handle = 0; + uv_thread_t uvt = tid->handle; + if (uvt != 0) { + tid->handle = 0; + int ret = uv_thread_join(&uvt); + if (ret < 0) return luv_error(L, ret); + } lua_pushboolean(L, 1); return 1; } diff --git a/tests/test-async.lua b/tests/test-async.lua index e6ba667b..4b75e1ac 100644 --- a/tests/test-async.lua +++ b/tests/test-async.lua @@ -1,32 +1,104 @@ return require('lib/tap')(function (test) test("test pass async between threads", function(p, p, expect, uv) - local before = os.time() local async - async = uv.new_async(expect(function (a,b,c) + async = uv.new_async(expect(function (s, b, i, n, u) p('in async notify callback') - p(a,b,c) - assert(a=='a') - assert(b==true) - assert(c==250) + p(s, b, i, n, u) + assert(s=='a', 'bad string') + assert(b==true, 'bad boolean') + assert(i==250, 'bad integer') + assert(n==3.14, 'bad number') + assert(type(u)=='userdata', 'bad userdata') uv.close(async) end)) - local args = {500, 'string', nil, false, 5, "helloworld",async} - local unpack = unpack or table.unpack - uv.new_thread(function(num,s,null,bool,five,hw,asy) - local uv = require'luv' - assert(type(num) == "number") - assert(type(s) == "string") - assert(null == nil) - assert(bool == false) - assert(five == 5) - assert(hw == 'helloworld') + uv.new_thread(function(asy) assert(type(asy)=='userdata') - assert(uv.async_send(asy,'a',true,250)==0) - uv.sleep(1000) - end, unpack(args)):join() - local elapsed = (os.time() - before) * 1000 - assert(elapsed >= 1000, "elapsed should be at least delay ") + assert(asy:send('a', true, 250, 3.14, io.stderr)==0) + require('luv').sleep(10) + end, async):join() + end) + + test("test async multiple send", function(p, p, expect, uv) + local async + async = uv.new_async(expect(function (v) + p('in async notify callback') + assert(v=='ok') + async:close() + end)) + uv.new_thread(function(asy) + assert(type(asy)=='userdata') + assert(asy:send('not ok')==0) -- will be ignored but its ok + assert(asy:send('ok')==0) + require('luv').sleep(10) + end, async):join() + end) + + test("test async send from same thread", function(p, p, expect, uv) + local async + async = uv.new_async(expect(function (v) + p('in async notify callback') + assert(v=='ok') + async:close() + end)) + assert(async:send('not ok')==0) -- will be ignored but its ok + assert(async:send('ok')==0) + uv.run() + end) + + test("test async send during callback", function(p, p, expect, uv) + local async + async = uv.new_async(expect(function (d, v) + p('in async notify callback', d, v) + assert(v=='ok') + if d > 0 then + uv.sleep(d) + else + async:close() + end + end, 2)) + local t = uv.new_thread(function(asy) + local uv = require('luv') + assert(type(asy)=='userdata') + assert(asy:send(100, 'ok')==0) + uv.sleep(10) -- let async callback starts + assert(asy:send(0, 'ok')==0) + uv.sleep(10) + end, async) + uv.run() + t:join() + end) + + test("test pass back async between threads", function(p, p, expect, uv) + local async + async = uv.new_async(expect(function (asy) + async:close() + p('in async notify callback') + assert(type(asy)=='userdata') + assert(debug.getmetatable(asy)) + assert(asy:send('Hi\0', true, 250)==0) -- only working inside callback + local timer = uv.new_timer() + timer:start(10, 0, expect(function() + timer:close() + p('timeout') + assert(not debug.getmetatable(asy)) -- outside callback the userdata loose its metatable + end)) + end)) + local t = uv.new_thread(function(asy) + local uv = require('luv') + assert(type(asy)=='userdata', 'bad aync type') + local as + as = uv.new_async(function (s, b, i) + as:close() + assert(s=='Hi\0', 'bad string') + assert(b==true, 'bad boolean') + assert(i==250, 'bad integer') + end) + assert(asy:send(as)==0) + uv.run() + end, async) + uv.run() + t:join() end) end) From d6d27bb6a6d76502419f268401dd6c547bd6cddc Mon Sep 17 00:00:00 2001 From: javalikescript Date: Mon, 18 Nov 2024 22:22:13 +0100 Subject: [PATCH 2/9] Fix work --- src/async.c | 4 ++-- src/thread.c | 6 +++--- src/work.c | 6 +++--- 3 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/async.c b/src/async.c index f0935196..23b2ce06 100644 --- a/src/async.c +++ b/src/async.c @@ -81,7 +81,7 @@ static int luv_async_gc(lua_State* L) { luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); uv_mutex_t *argmutex = &asarg->mutex; uv_mutex_lock(argmutex); - luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_CHILD); // in case of a pending send, set side to avoid unref + luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_MAIN); // in case of a pending send uv_mutex_unlock(argmutex); uv_mutex_destroy(argmutex); return luv_handle_gc(L); @@ -94,7 +94,7 @@ static int luv_async_send(lua_State* L) { uv_mutex_t *argmutex = &asarg->mutex; int n; uv_mutex_lock(argmutex); - luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_CHILD); // in case of a pending send + luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_MAIN); // in case of a pending send n = luv_thread_arg_set(L, &asarg->targ, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); uv_mutex_unlock(argmutex); if (n < 0) { diff --git a/src/thread.c b/src/thread.c index 67c4e037..fcfcfdf6 100644 --- a/src/thread.c +++ b/src/thread.c @@ -151,7 +151,7 @@ static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags /* * clear is safe to be called multiple times, values are set to LUA_NOREF or NULL, argc is preserved. * thread unrefs per side. - * async frees values on the first calling side. + * async frees values on the push side. */ for (i = 0; i < args->argc; i++) { luv_val_t* arg = args->argv + i; @@ -161,7 +161,7 @@ static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags luaL_unref(L, LUA_REGISTRYINDEX, arg->ref[side]); arg->ref[side] = LUA_NOREF; } - if (async) { + if (async && side != setside) { free((void*)arg->val.str.base); arg->val.str.base = NULL; } @@ -178,7 +178,7 @@ static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags luaL_unref(L, LUA_REGISTRYINDEX, arg->ref[side]); arg->ref[side] = LUA_NOREF; } - if (async) { + if (async && side != setside) { if (arg->val.udata.size > 0) { free((void*)arg->val.udata.data); arg->val.udata.data = NULL; diff --git a/src/work.c b/src/work.c index b51bfd11..8b973021 100644 --- a/src/work.c +++ b/src/work.c @@ -113,7 +113,7 @@ static int luv_work_cb(lua_State* L) { return luv_thread_arg_error(L); } lua_pop(L, i); // pop all returned value - luv_thread_arg_clear(L, &work->rets, LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); + luv_thread_arg_clear(L, &work->rets, LUVF_THREAD_SIDE_CHILD); } luv_thread_arg_clear(L, &work->args, LUVF_THREAD_SIDE_CHILD); } else { @@ -154,7 +154,7 @@ static void luv_work_cb_wrapper(uv_work_t* req) { // uv__threadpool_cleanup, so exit is not called in luv_cfpcall. int i = lctx->thrd_cpcall(L, luv_work_cb, (void*)req, LUVF_CALLBACK_NOEXIT); if (i != LUA_OK) { - luv_thread_arg_clear(L, &work->rets, LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); + luv_thread_arg_clear(L, &work->rets, LUVF_THREAD_SIDE_CHILD); luv_thread_arg_clear(L, &work->args, LUVF_THREAD_SIDE_CHILD); } } @@ -177,7 +177,7 @@ static void luv_after_work_cb(uv_work_t* req, int status) { work->ref = LUA_NOREF; luv_thread_arg_clear(L, &work->args, LUVF_THREAD_SIDE_MAIN); - luv_thread_arg_clear(L, &work->rets, LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_MAIN); + luv_thread_arg_clear(L, &work->rets, LUVF_THREAD_SIDE_MAIN); free(work); } From 84436f38ac5a82f2c53439f8712f01fcae4b047b Mon Sep 17 00:00:00 2001 From: javalikescript Date: Wed, 20 Nov 2024 19:19:41 +0100 Subject: [PATCH 3/9] Split clear and free --- src/async.c | 8 +++++--- src/thread.c | 52 ++++++++++++++++++++++++++++++---------------------- 2 files changed, 35 insertions(+), 25 deletions(-) diff --git a/src/async.c b/src/async.c index 23b2ce06..17799a19 100644 --- a/src/async.c +++ b/src/async.c @@ -73,15 +73,17 @@ static int luv_new_async(lua_State* L) { return 1; } - +// From handle.c static int luv_handle_gc(lua_State* L); +// From thread.c +static void luv_thread_arg_free(luv_thread_arg_t* args); static int luv_async_gc(lua_State* L) { uv_async_t* handle = *(uv_async_t**)lua_touserdata(L, 1); luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); uv_mutex_t *argmutex = &asarg->mutex; uv_mutex_lock(argmutex); - luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_MAIN); // in case of a pending send + luv_thread_arg_free(&asarg->targ); // in case of a pending send uv_mutex_unlock(argmutex); uv_mutex_destroy(argmutex); return luv_handle_gc(L); @@ -94,7 +96,7 @@ static int luv_async_send(lua_State* L) { uv_mutex_t *argmutex = &asarg->mutex; int n; uv_mutex_lock(argmutex); - luv_thread_arg_clear(L, &asarg->targ, LUVF_THREAD_SIDE_MAIN); // in case of a pending send + luv_thread_arg_free(&asarg->targ); // in case of a pending send n = luv_thread_arg_set(L, &asarg->targ, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); uv_mutex_unlock(argmutex); if (n < 0) { diff --git a/src/thread.c b/src/thread.c index fcfcfdf6..7e752eb6 100644 --- a/src/thread.c +++ b/src/thread.c @@ -66,11 +66,7 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int int i; int side = LUVF_THREAD_SIDE(flags); int async = LUVF_THREAD_ASYNC(flags); - /* - * thread works by reference. - * async works by copy, a use case consists in sending just before the thread ends, - * it results that the callback will be called after the thread/async life. - */ + // thread works by reference, async works by copy. idx = idx > 0 ? idx : 1; i = idx; args->flags = flags; @@ -143,16 +139,37 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int return args->argc; } +static void luv_thread_arg_free(luv_thread_arg_t* args) { + int i; + for (i = 0; i < args->argc; i++) { + luv_val_t* arg = args->argv + i; + switch (arg->type) { + case LUA_TSTRING: + free((void*)arg->val.str.base); + arg->val.str.base = NULL; + break; + case LUA_TUSERDATA: + if (arg->val.udata.size > 0) { + free((void*)arg->val.udata.data); + arg->val.udata.data = NULL; + } + free((void*)arg->val.udata.metaname); + arg->val.udata.metaname = NULL; + break; + } + } +} + static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags) { int i; int side = LUVF_THREAD_SIDE(flags); int setside = LUVF_THREAD_SIDE(args->flags); int async = LUVF_THREAD_ASYNC(args->flags); - /* - * clear is safe to be called multiple times, values are set to LUA_NOREF or NULL, argc is preserved. - * thread unrefs per side. - * async frees values on the push side. - */ + + // clear is safe to be called multiple times from multiple sides. + if (args->argc <= 0) { + return; + } for (i = 0; i < args->argc; i++) { luv_val_t* arg = args->argv + i; switch (arg->type) { @@ -161,10 +178,6 @@ static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags luaL_unref(L, LUA_REGISTRYINDEX, arg->ref[side]); arg->ref[side] = LUA_NOREF; } - if (async && side != setside) { - free((void*)arg->val.str.base); - arg->val.str.base = NULL; - } break; case LUA_TUSERDATA: if (arg->ref[side] != LUA_NOREF) { @@ -178,19 +191,14 @@ static void luv_thread_arg_clear(lua_State* L, luv_thread_arg_t* args, int flags luaL_unref(L, LUA_REGISTRYINDEX, arg->ref[side]); arg->ref[side] = LUA_NOREF; } - if (async && side != setside) { - if (arg->val.udata.size > 0) { - free((void*)arg->val.udata.data); - arg->val.udata.data = NULL; - } - free((void*)arg->val.udata.metaname); - arg->val.udata.metaname = NULL; - } break; default: break; } } + if (async && side != setside) { + luv_thread_arg_free(args); + } } static int luv_thread_arg_push(lua_State* L, luv_thread_arg_t* args, int flags) { From 8a68607d100fc147f01637f3ca351e800d5d4ab4 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Wed, 20 Nov 2024 20:34:06 +0100 Subject: [PATCH 4/9] Add FIFO queue option --- src/async.c | 97 ++++++++++++++++++++++++++++++++++++++------ tests/test-async.lua | 21 ++++++++++ 2 files changed, 105 insertions(+), 13 deletions(-) diff --git a/src/async.c b/src/async.c index 17799a19..b19d1196 100644 --- a/src/async.c +++ b/src/async.c @@ -16,9 +16,18 @@ */ #include "private.h" +typedef struct luv_async_send_s { + luv_thread_arg_t targ; + struct luv_async_send_s* next; +} luv_async_send_t; + typedef struct { luv_thread_arg_t targ; uv_mutex_t mutex; + int max; // FIFO queue in case of max > 1 + int count; + luv_async_send_t* first; + luv_async_send_t* last; } luv_async_arg_t; #define luv_get_async_arg_from_handle(H) ((luv_async_arg_t *) ((luv_handle_t*) (H)->data)->extra) @@ -29,22 +38,63 @@ static uv_async_t* luv_check_async(lua_State* L, int index) { return handle; } +#define luv_is_async_queue(AA) ((AA)->max > 1) + +static luv_async_send_t* luv_async_pop(luv_async_arg_t* asarg) { + luv_async_send_t* sendarg = asarg->first; + if (sendarg != NULL) { + asarg->count--; + asarg->first = sendarg->next; + if (asarg->first == NULL) { + asarg->last = NULL; + } + } + return sendarg; +} + +static luv_async_send_t* luv_async_push(luv_async_arg_t* asarg) { + luv_async_send_t* sendarg = (luv_async_send_t*)malloc(sizeof(luv_async_send_t)); + memset(sendarg, 0, sizeof(luv_async_send_t)); + asarg->count++; + if (asarg->last != NULL) { + asarg->last->next = sendarg; + } + asarg->last = sendarg; + if (asarg->first == NULL) { + asarg->first = sendarg; + } + return sendarg; +} + static void luv_async_cb(uv_async_t* handle) { luv_handle_t* data = (luv_handle_t*)handle->data; lua_State* L = data->ctx->L; luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); uv_mutex_t *argmutex = &asarg->mutex; - luv_thread_arg_t targcpy; + luv_thread_arg_t targcpy; // work on a copy of the arguments int n; - uv_mutex_lock(argmutex); - targcpy = asarg->targ; // work on a copy of the arguments - asarg->targ.argc = 0; // empty the original, nothing to clear - uv_mutex_unlock(argmutex); - n = luv_thread_arg_push(L, &targcpy, LUVF_THREAD_SIDE_MAIN); - if (n >= 0) { - luv_call_callback(L, data, LUV_ASYNC, n); - } - luv_thread_arg_clear(L, &targcpy, LUVF_THREAD_SIDE_MAIN); // clear the copy + int q = luv_is_async_queue(asarg); + do { + uv_mutex_lock(argmutex); + if (q) { + luv_async_send_t* sendarg = luv_async_pop(asarg); + if (sendarg == NULL) { + uv_mutex_unlock(argmutex); + return; + } + targcpy = sendarg->targ; + free(sendarg); + } else { + targcpy = asarg->targ; + asarg->targ.argc = 0; // empty the shared original, nothing to clear + } + uv_mutex_unlock(argmutex); + n = luv_thread_arg_push(L, &targcpy, LUVF_THREAD_SIDE_MAIN); + if (n >= 0) { + luv_call_callback(L, data, LUV_ASYNC, n); + } + luv_thread_arg_clear(L, &targcpy, LUVF_THREAD_SIDE_MAIN); // clear the copy + } while (q); } static int luv_new_async(lua_State* L) { @@ -52,6 +102,7 @@ static int luv_new_async(lua_State* L) { luv_handle_t* data; int ret; luv_ctx_t* ctx = luv_context(L); + int max = luaL_optinteger(L, 2, 0); luaL_checktype(L, 1, LUA_TFUNCTION); handle = (uv_async_t*)luv_newuserdata(L, uv_handle_size(UV_ASYNC)); ret = uv_async_init(ctx->loop, handle, luv_async_cb); @@ -62,6 +113,7 @@ static int luv_new_async(lua_State* L) { data = luv_setup_handle(L, ctx); luv_async_arg_t* asarg = (luv_async_arg_t*)malloc(sizeof(luv_async_arg_t)); memset(asarg, 0, sizeof(luv_async_arg_t)); + asarg->max = max; ret = uv_mutex_init(&asarg->mutex); if (ret < 0) { // unlikely abort(); @@ -83,7 +135,15 @@ static int luv_async_gc(lua_State* L) { luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); uv_mutex_t *argmutex = &asarg->mutex; uv_mutex_lock(argmutex); - luv_thread_arg_free(&asarg->targ); // in case of a pending send + if (luv_is_async_queue(asarg)) { + luv_async_send_t* sendarg; + while ((sendarg = luv_async_pop(asarg)) != NULL) { + luv_thread_arg_free(&sendarg->targ); + free(sendarg); + } + } else { + luv_thread_arg_free(&asarg->targ); // in case of a pending send + } uv_mutex_unlock(argmutex); uv_mutex_destroy(argmutex); return luv_handle_gc(L); @@ -94,10 +154,21 @@ static int luv_async_send(lua_State* L) { uv_async_t* handle = luv_check_async(L, 1); luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); uv_mutex_t *argmutex = &asarg->mutex; + luv_thread_arg_t* args; int n; uv_mutex_lock(argmutex); - luv_thread_arg_free(&asarg->targ); // in case of a pending send - n = luv_thread_arg_set(L, &asarg->targ, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); + if (luv_is_async_queue(asarg)) { + if (asarg->count >= asarg->max) { + uv_mutex_unlock(argmutex); + return luv_error(L, UV_ENOSPC); + } + luv_async_send_t* sendarg = luv_async_push(asarg); + args = &sendarg->targ; + } else { + luv_thread_arg_free(&asarg->targ); // in case of a pending send + args = &asarg->targ; + } + n = luv_thread_arg_set(L, args, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); uv_mutex_unlock(argmutex); if (n < 0) { return luv_thread_arg_error(L); diff --git a/tests/test-async.lua b/tests/test-async.lua index 4b75e1ac..23c44d9c 100644 --- a/tests/test-async.lua +++ b/tests/test-async.lua @@ -34,6 +34,27 @@ return require('lib/tap')(function (test) end, async):join() end) + test("test async queue send", function(p, p, expect, uv) + local async + async = uv.new_async(expect(function (v) + p('in async notify callback') + if v == 'close' then + async:close() + else + assert(v=='ok') + end + end, 3), 3) + uv.new_thread(function(asy) + local uv = require('luv') + assert(type(asy)=='userdata') + assert(asy:send('ok')==0) + assert(asy:send('ok')==0) + assert(asy:send('close')==0) + assert(select(3, asy:send('not ok'))=='ENOSPC') + uv.sleep(10) + end, async):join() + end) + test("test async send from same thread", function(p, p, expect, uv) local async async = uv.new_async(expect(function (v) From 98e8fa9e27b543e3ff77e791c9db8a6801f036a5 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Fri, 22 Nov 2024 22:07:50 +0100 Subject: [PATCH 5/9] Allow queue with only one item --- docs.md | 7 ++++++- src/async.c | 4 ++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/docs.md b/docs.md index ef74a499..193b3240 100644 --- a/docs.md +++ b/docs.md @@ -1016,15 +1016,19 @@ end) async:send() ``` -### `uv.new_async(callback)` +### `uv.new_async(callback, [size])` **Parameters:** - `callback`: `callable` - `...`: `threadargs` passed to/from `uv.async_send(async, ...)` +- `size`: `integer` or `nil` (default: `0`) Creates and initializes a new `uv_async_t`. Returns the Lua userdata wrapping it. +If size is omitted (or 0), each call to send discards any pending one; otherwise, a call to send could fail with `ENOSPC` when there are already `size` pending send. + + **Returns:** `uv_async_t userdata` or `fail` **Note**: Unlike other handle initialization functions, this immediately starts @@ -1050,6 +1054,7 @@ every call to it will yield an execution of the callback. For example: if `uv.async_send()` is called 5 times in a row before the callback is called, the callback will only be called once. If `uv.async_send()` is called again after the callback was called, it will be called again. +When specifying a `size` greater than 0, this function will fail when there are pending calls. ## `uv_poll_t` — Poll handle diff --git a/src/async.c b/src/async.c index b19d1196..5c2f190a 100644 --- a/src/async.c +++ b/src/async.c @@ -24,7 +24,7 @@ typedef struct luv_async_send_s { typedef struct { luv_thread_arg_t targ; uv_mutex_t mutex; - int max; // FIFO queue in case of max > 1 + int max; // FIFO queue in case of max > 0 int count; luv_async_send_t* first; luv_async_send_t* last; @@ -38,7 +38,7 @@ static uv_async_t* luv_check_async(lua_State* L, int index) { return handle; } -#define luv_is_async_queue(AA) ((AA)->max > 1) +#define luv_is_async_queue(AA) ((AA)->max > 0) static luv_async_send_t* luv_async_pop(luv_async_arg_t* asarg) { luv_async_send_t* sendarg = asarg->first; From 7f4f31f9e88ad64c36aaeceec6b57dad533a0ee4 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Sun, 24 Nov 2024 11:35:47 +0100 Subject: [PATCH 6/9] Enhance error handling --- src/async.c | 48 ++++++++++++++++++++++++++------------------ src/thread.c | 27 ++++++++++++++++++++----- tests/test-async.lua | 5 +++-- 3 files changed, 54 insertions(+), 26 deletions(-) diff --git a/src/async.c b/src/async.c index 5c2f190a..061e9f2c 100644 --- a/src/async.c +++ b/src/async.c @@ -54,14 +54,16 @@ static luv_async_send_t* luv_async_pop(luv_async_arg_t* asarg) { static luv_async_send_t* luv_async_push(luv_async_arg_t* asarg) { luv_async_send_t* sendarg = (luv_async_send_t*)malloc(sizeof(luv_async_send_t)); - memset(sendarg, 0, sizeof(luv_async_send_t)); - asarg->count++; - if (asarg->last != NULL) { - asarg->last->next = sendarg; - } - asarg->last = sendarg; - if (asarg->first == NULL) { - asarg->first = sendarg; + if (sendarg != NULL) { + memset(sendarg, 0, sizeof(luv_async_send_t)); + asarg->count++; + if (asarg->last != NULL) { + asarg->last->next = sendarg; + } + asarg->last = sendarg; + if (asarg->first == NULL) { + asarg->first = sendarg; + } } return sendarg; } @@ -154,25 +156,33 @@ static int luv_async_send(lua_State* L) { uv_async_t* handle = luv_check_async(L, 1); luv_async_arg_t* asarg = luv_get_async_arg_from_handle(handle); uv_mutex_t *argmutex = &asarg->mutex; - luv_thread_arg_t* args; - int n; + luv_thread_arg_t targcpy; + ret = luv_thread_arg_set(L, &targcpy, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); + if (ret < 0) { + luv_thread_arg_free(&targcpy); + return luv_thread_arg_error(L); + } uv_mutex_lock(argmutex); if (luv_is_async_queue(asarg)) { - if (asarg->count >= asarg->max) { + luv_async_send_t* sendarg = NULL; + ret = UV_ENOSPC; + if (asarg->count < asarg->max) { + sendarg = luv_async_push(asarg); + if (sendarg == NULL) { + ret = UV_ENOMEM; + } + } + if (sendarg == NULL) { uv_mutex_unlock(argmutex); - return luv_error(L, UV_ENOSPC); + luv_thread_arg_free(&targcpy); + return luv_error(L, ret); } - luv_async_send_t* sendarg = luv_async_push(asarg); - args = &sendarg->targ; + sendarg->targ = targcpy; } else { luv_thread_arg_free(&asarg->targ); // in case of a pending send - args = &asarg->targ; + asarg->targ = targcpy; } - n = luv_thread_arg_set(L, args, 2, lua_gettop(L), LUVF_THREAD_MODE_ASYNC|LUVF_THREAD_SIDE_CHILD); uv_mutex_unlock(argmutex); - if (n < 0) { - return luv_thread_arg_error(L); - } ret = uv_async_send(handle); return luv_result(L, ret); } diff --git a/src/thread.c b/src/thread.c index 7e752eb6..d8021d92 100644 --- a/src/thread.c +++ b/src/thread.c @@ -62,6 +62,13 @@ static const char* luv_getmtname(lua_State *L, int idx) { return name; } +static int luv_thread_arg_set_error(lua_State* L, luv_thread_arg_t* args, int type, int index) { + args->argc = index; + lua_pushinteger(L, type); + lua_pushinteger(L, index + 1); + return -1; +} + static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int top, int flags) { int i; int side = LUVF_THREAD_SIDE(flags); @@ -72,7 +79,8 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int args->flags = flags; while (i <= top && i < LUV_THREAD_MAXNUM_ARG + idx) { - luv_val_t *arg = args->argv + (i - idx); + int ii = i - idx; + luv_val_t *arg = args->argv + ii; arg->type = lua_type(L, i); arg->ref[0] = arg->ref[1] = LUA_NOREF; switch (arg->type) @@ -95,6 +103,9 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int size_t l = 0; const char* p = lua_tolstring(L, i, &l); void* b = malloc(l + 1); + if (b == NULL) { + return luv_thread_arg_set_error(L, args, LUA_TNONE, ii); + } arg->val.str.base = memcpy((void*)b, p, l + 1); arg->val.str.len = l; } else { @@ -111,11 +122,17 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int if (async) { if (l > 0) { void* b = malloc(l); + if (b == NULL) { + return luv_thread_arg_set_error(L, args, LUA_TNONE, ii); + } p = (const void*)memcpy(b, p, l); } if (mtname != NULL) { size_t ml = strlen(mtname) + 1; char* b = malloc(ml); + if (b == NULL) { + return luv_thread_arg_set_error(L, args, LUA_TNONE, ii); + } mtname = (const void*)memcpy(b, mtname, ml); } } else { @@ -128,10 +145,7 @@ static int luv_thread_arg_set(lua_State* L, luv_thread_arg_t* args, int idx, int } break; default: - args->argc = i - idx; - lua_pushinteger(L, arg->type); - lua_pushinteger(L, i - idx + 1); - return -1; + return luv_thread_arg_set_error(L, args, arg->type, ii); } i++; } @@ -251,6 +265,9 @@ static int luv_thread_arg_error(lua_State *L) { int type = lua_tointeger(L, -2); int pos = lua_tointeger(L, -1); lua_pop(L, 2); + if (type == LUA_TNONE) { + return luaL_error(L, "Error: thread arg failure at %d", pos); + } return luaL_error(L, "Error: thread arg not support type '%s' at %d", lua_typename(L, type), pos); } diff --git a/tests/test-async.lua b/tests/test-async.lua index 23c44d9c..d057ea53 100644 --- a/tests/test-async.lua +++ b/tests/test-async.lua @@ -28,7 +28,7 @@ return require('lib/tap')(function (test) end)) uv.new_thread(function(asy) assert(type(asy)=='userdata') - assert(asy:send('not ok')==0) -- will be ignored but its ok + assert(asy:send('not ok')==0) -- will be ignored but it's ok assert(asy:send('ok')==0) require('luv').sleep(10) end, async):join() @@ -62,8 +62,9 @@ return require('lib/tap')(function (test) assert(v=='ok') async:close() end)) - assert(async:send('not ok')==0) -- will be ignored but its ok + assert(async:send('not ok')==0) -- will be ignored but it's ok assert(async:send('ok')==0) + assert(pcall(uv.async_send, 'not ok', function() end)==false) -- will fail uv.run() end) From 9fa42ef602913926b2fef0752b5d0ee8325f96e8 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Fri, 20 Dec 2024 08:53:38 +0100 Subject: [PATCH 7/9] Handle uv_mutex_init failure --- src/async.c | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/src/async.c b/src/async.c index 061e9f2c..1ee47570 100644 --- a/src/async.c +++ b/src/async.c @@ -102,10 +102,15 @@ static void luv_async_cb(uv_async_t* handle) { static int luv_new_async(lua_State* L) { uv_async_t* handle; luv_handle_t* data; + uv_mutex_t argmutex; int ret; luv_ctx_t* ctx = luv_context(L); int max = luaL_optinteger(L, 2, 0); luaL_checktype(L, 1, LUA_TFUNCTION); + ret = uv_mutex_init(&argmutex); + if (ret < 0) { + return luv_error(L, ret); + } handle = (uv_async_t*)luv_newuserdata(L, uv_handle_size(UV_ASYNC)); ret = uv_async_init(ctx->loop, handle, luv_async_cb); if (ret < 0) { @@ -116,10 +121,7 @@ static int luv_new_async(lua_State* L) { luv_async_arg_t* asarg = (luv_async_arg_t*)malloc(sizeof(luv_async_arg_t)); memset(asarg, 0, sizeof(luv_async_arg_t)); asarg->max = max; - ret = uv_mutex_init(&asarg->mutex); - if (ret < 0) { // unlikely - abort(); - } + asarg->mutex = argmutex; data->extra = asarg; data->extra_gc = free; handle->data = data; From 582728191499582a9483f74c36dbc17d5422f673 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Tue, 24 Dec 2024 12:04:39 +0100 Subject: [PATCH 8/9] Use ubuntu 24 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ca030617..c915af08 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,7 +55,7 @@ jobs: run: ./build/lua -e "collectgarbage('incremental', 0, 10000000000000)" tests/run.lua valgrind: - runs-on: ubuntu-latest + runs-on: ubuntu-24.04 env: BUILD_TYPE: Debug WITH_LUA_ENGINE: Lua From 090a0a76a0a3917eb3a34c443850a54eb85a34f0 Mon Sep 17 00:00:00 2001 From: javalikescript Date: Sat, 25 Jan 2025 08:50:20 +0100 Subject: [PATCH 9/9] Revert "Use ubuntu 24" This reverts commit 582728191499582a9483f74c36dbc17d5422f673. --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c915af08..ca030617 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -55,7 +55,7 @@ jobs: run: ./build/lua -e "collectgarbage('incremental', 0, 10000000000000)" tests/run.lua valgrind: - runs-on: ubuntu-24.04 + runs-on: ubuntu-latest env: BUILD_TYPE: Debug WITH_LUA_ENGINE: Lua