Add async ref count

Author: javalikescript

src/async.c

@@ -33,15 +33,24 @@ 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;
+  luv_ref_t *ref;
   int ret;
   luv_ctx_t* ctx = luv_context(L);
   luaL_checktype(L, 1, LUA_TFUNCTION);
-  handle = (uv_async_t*)luv_newuserdata(L, uv_handle_size(UV_ASYNC));
+  ref = (luv_ref_t*)luv_newuserdata(L, sizeof(luv_ref_t));
+  handle = &ref->handle.async;
+  ret = uv_mutex_init(&ref->mutex);
+  if (ret < 0) {
+    lua_pop(L, 1);
+    return luv_error(L, ret);
+  }
   ret = uv_async_init(ctx->loop, handle, luv_async_cb);
   if (ret < 0) {
+    uv_mutex_destroy(&ref->mutex);
     lua_pop(L, 1);
     return luv_error(L, ret);
   }
+  ref->count = 1;
   data = luv_setup_handle(L, ctx);
   data->extra = (luv_thread_arg_t*)malloc(sizeof(luv_thread_arg_t));
   data->extra_gc = free;
@@ -51,6 +60,22 @@ static int luv_new_async(lua_State* L) {
   return 1;
 }
 
+static int luv_handle_gc(lua_State* L);
+
+static int luv_async_gc(lua_State* L) {
+  luv_ref_t** udata = (luv_ref_t**)lua_touserdata(L, 1);
+  luv_ref_t* ref = *udata;
+  uv_mutex_t *mutex = &ref->mutex;
+  uv_mutex_lock(mutex);
+  ref->count--;
+  uv_mutex_unlock(mutex);
+  if (ref->count > 0) {
+    return 0;
+  }
+  uv_mutex_destroy(mutex);
+  return luv_handle_gc(L);
+}
+
 static int luv_async_send(lua_State* L) {
   int ret;
   uv_async_t* handle = luv_check_async(L, 1);
@@ -61,3 +86,10 @@ static int luv_async_send(lua_State* L) {
   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);
+}

src/lthreadpool.h

@@ -49,6 +49,14 @@ typedef struct {
   luv_val_t argv[LUV_THREAD_MAXNUM_ARG];
 } luv_thread_arg_t;
 
+typedef struct {
+  union {
+    uv_async_t async;
+  } handle;
+  uv_mutex_t mutex;
+  int count;
+} luv_ref_t;
+
 //luajit miss LUA_OK
 #ifndef LUA_OK
 #define LUA_OK 0

src/luv.c

@@ -892,6 +892,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

src/thread.c

@@ -189,15 +189,25 @@ static int luv_thread_arg_push(lua_State* L, luv_thread_arg_t* args, int flags)
     case LUA_TUSERDATA:
       if (arg->val.udata.size)
       {
+        int r = 1;
         char *p = lua_newuserdata(L, arg->val.udata.size);
         memcpy(p, arg->val.udata.data, arg->val.udata.size);
         if (arg->val.udata.metaname)
         {
           luaL_getmetatable(L, arg->val.udata.metaname);
           lua_setmetatable(L, -2);
+          if (strcmp(arg->val.udata.metaname, "uv_async") == 0) {
+            luv_ref_t* ref = *(luv_ref_t**)p;
+            uv_mutex_lock(&ref->mutex);
+            ref->count++;
+            uv_mutex_unlock(&ref->mutex);
+            r = 0;
+          }
+        }
+        if (r) {
+          lua_pushvalue(L, -1);
+          arg->ref[side] = luaL_ref(L, LUA_REGISTRYINDEX);
         }
-        lua_pushvalue(L, -1);
-        arg->ref[side] = luaL_ref(L, LUA_REGISTRYINDEX);
       }else{
         lua_pushlightuserdata(L, (void*)arg->val.udata.data);
       }

tests/test-async.lua

@@ -1,7 +1,6 @@
 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)
       p('in async notify callback')
@@ -11,22 +10,43 @@ return require('lib/tap')(function (test)
       assert(c==250)
       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)
+    uv.new_thread(function(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')
       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 ")
+      uv.run()
+    end, async):join()
+  end)
+
+  test("test pass back async between threads", function(p, p, expect, uv)
+    local async
+    async = uv.new_async(expect(function (a)
+      uv.close(async)
+      p('in async notify callback')
+      p(a)
+      assert(type(a)=='userdata')
+      local timer = uv.new_timer()
+      uv.timer_start(timer, 10, 0, expect(function()
+        uv.close(timer)
+        p("timeout", timer)
+        assert(uv.async_send(a,'a',true,250)==0)
+      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 (a,b,c)
+        uv.close(as)
+        assert(a=='a', 'bad string')
+        assert(b==true, 'bad boolean')
+        assert(c==250, 'bad number')
+      end)
+      assert(uv.async_send(asy,as)==0)
+      uv.run()
+    end, async)
+    uv.run()
+    t:join()
   end)
 
 end)