improve poller

6 files changed, 154 insertions, 69 deletions

diff --git a/core/src/xmake/io/poller_wait.c b/core/src/xmake/io/poller_wait.c
index 2589025c2..eceb44cbf 100644
--- a/core/src/xmake/io/poller_wait.c
+++ b/core/src/xmake/io/poller_wait.c
@@ -32,6 +32,13 @@
 #include "poller.h"
 
 /* //////////////////////////////////////////////////////////////////////////////////////
+ * globals
+ */
+
+// we need only one global poller and main thread, so it is thread-safe.
+static tb_int_t g_events_count = 0;
+
+/* //////////////////////////////////////////////////////////////////////////////////////
  * private implementation
  */
 static tb_void_t xm_io_poller_event(tb_poller_ref_t poller, tb_socket_ref_t sock, tb_size_t events, tb_cpointer_t priv)
@@ -46,6 +53,7 @@ static tb_void_t xm_io_poller_event(tb_poller_ref_t poller, tb_socket_ref_t sock
     lua_rawseti(lua, -2, 1);
     lua_pushinteger(lua, (tb_int_t)events);
     lua_rawseti(lua, -2, 2);
+    lua_rawseti(lua, -2, ++g_events_count);
 }
 
 /* //////////////////////////////////////////////////////////////////////////////////////
@@ -63,10 +71,10 @@ tb_int_t xm_io_poller_wait(lua_State* lua)
 
     // wait it
     lua_newtable(lua);
+    g_events_count = 0;
     tb_long_t count = tb_poller_wait(xm_io_poller(), xm_io_poller_event, timeout);
     if (count > 0)
     {
-        lua_rawseti(lua, -2, (tb_int_t)count);
         lua_pushinteger(lua, (tb_int_t)count);
         return 2;
     } 
diff --git a/tests/modules/socket/sched_tcp/echo_client.lua b/tests/modules/socket/sched_tcp/echo_client.lua
index 50e8c6dcd..f9c993729 100644
--- a/tests/modules/socket/sched_tcp/echo_client.lua
+++ b/tests/modules/socket/sched_tcp/echo_client.lua
@@ -22,7 +22,7 @@ end
 function main(count)
     count = count and tonumber(count) or 1
     for i = 1, count do
-        scheduler.run(_session, "127.0.0.1", 9001)
+        scheduler.co_start(_session, "127.0.0.1", 9001)
     end
     scheduler.runloop()
 end
diff --git a/tests/modules/socket/sched_tcp/echo_server.lua b/tests/modules/socket/sched_tcp/echo_server.lua
index 4ae2119cc..18f3d990f 100644
--- a/tests/modules/socket/sched_tcp/echo_server.lua
+++ b/tests/modules/socket/sched_tcp/echo_server.lua
@@ -28,13 +28,13 @@ function _listen(addr, port)
         local sock_client = sock:accept()
         if sock_client then
             print("%s: accepted", sock_client) 
-            scheduler.run(_session, sock_client) 
+            scheduler.co_start(_session, sock_client) 
         end
     end
     sock:close()
 end
 
 function main()
-    scheduler.run(_listen, "127.0.0.1", 9001)
+    scheduler.co_start(_listen, "127.0.0.1", 9001)
     scheduler.runloop()
 end
diff --git a/xmake/core/base/poller.lua b/xmake/core/base/poller.lua
index 20f3280ba..291125ec2 100644
--- a/xmake/core/base/poller.lua
+++ b/xmake/core/base/poller.lua
@@ -25,13 +25,18 @@ local poller = poller or {}
 local io     = require("base/io")
 local string = require("base/string")
 
+-- the poller object type
+poller.OT_SOCK = 1
+poller.OT_PROC = 2
+poller.OT_PIPE = 3
+
 -- get socket wait data 
-function poller:_waitdata(sock)
+function poller:_sockdata(sock)
     return self._CACHE and self._CACHE[sock] or nil
 end
 
 -- set socket wait data
-function poller:_waitdata_set(sock, data)
+function poller:_sockdata_set(sock, data)
     local cache = self._CACHE 
     if not cache then
         cache = {}
@@ -41,7 +46,7 @@ function poller:_waitdata_set(sock, data)
 end
 
 -- insert socket events to poller
-function poller:insert(sock, events)
+function poller:_insert_sock(sock, events, udata)
 
     -- ensure opened
     local ok, errors = sock:_ensure_opened()
@@ -50,17 +55,17 @@ function poller:insert(sock, events)
     end
 
     -- insert it
-    if not io.poller_insert(sock._SOCK, events) then
-        return false, string.format("insert %s events(%d) to poller failed!", sock, events)
+    if not io.poller_insert(sock:csock(), events) then
+        return false, string.format("%s: insert events(%d) to poller failed!", sock, events)
     end
 
-    -- save wait data and save sock/ref for gc
-    self:_waitdata_set(sock._SOCK, {sock, events})
+    -- save socket data and save sock/ref for gc
+    self:_sockdata_set(sock:csock(), udata)
     return true
 end
 
 -- modify socket events in poller
-function poller:modify(sock, events)
+function poller:_modify_sock(sock, events, udata)
 
     -- ensure opened
     local ok, errors = sock:_ensure_opened()
@@ -69,17 +74,17 @@ function poller:modify(sock, events)
     end
 
     -- modify it
-    if not io.poller_modify(sock._SOCK, events) then
-        return false, string.format("modify %s events(%d) to poller failed!", sock, events)
+    if not io.poller_modify(sock:csock(), events) then
+        return false, string.format("%s: modify events(%d) to poller failed!", sock, events)
     end
 
-    -- update wait data for this socket
-    self:_waitdata_set(sock._SOCK, {sock, events})
+    -- update socket data for this socket
+    self:_sockdata_set(sock:csock(), udata)
     return true
 end
 
 -- remove socket from poller
-function poller:remove(sock)
+function poller:_remove_sock(sock)
 
     -- ensure opened
     local ok, errors = sock:_ensure_opened()
@@ -88,34 +93,56 @@ function poller:remove(sock)
     end
 
     -- remove it
-    if not io.poller_remove(sock._SOCK) then
-        return false, string.format("remove %s from poller failed!", sock)
+    if not io.poller_remove(sock:csock()) then
+        return false, string.format("%s: remove events from poller failed!", sock)
     end
 
-    -- remove wait data for this socket
-    self:_waitdata_set(sock, nil)
+    -- remove socket data for this socket
+    self:_sockdata_set(sock, nil)
     return true
 end
 
+-- insert object events to poller
+function poller:insert(otype, obj, events, udata)
+    if otype == poller.OT_SOCK then
+        return self:_insert_sock(obj, events, udata)
+    end
+    return false, string.format("invalid poller object type(%d)!", otype)
+end
+
+-- modify object events in poller
+function poller:modify(otype, obj, events, udata)
+    if otype == poller.OT_SOCK then
+        return self:_modify_sock(obj, events, udata)
+    end
+    return false, string.format("invalid poller object type(%d)!", otype)
+end
+
+-- remove socket from poller
+function poller:remove(otype, obj)
+    if otype == poller.OT_SOCK then
+        return self:_remove_sock(obj)
+    end
+    return false, string.format("invalid poller object type(%d)!", otype)
+end
+
 -- wait socket events in poller
 function poller:wait(timeout)
 
     -- wait it
-    local sockevents, count = io.poller_wait(timeout or -1) 
+    local events, count = io.poller_wait(timeout or -1) 
     if count < 0 then
         return -1, "wait events in poller failed!"
     end
 
     -- wrap socket 
     local results = {}
-    if sockevents then
-        for _, v in ipairs(sockevents) do
-            local sock     = v[1]
-            local events   = v[2]
-            local waitdata = self:_waitdata(sock)
-            if waitdata then
-                results[waitdata[1]] = events
-            end
+    if events then
+        for _, v in ipairs(events) do
+            -- TODO only socket events now. It will be proc/pipe events in the future
+            local csock      = v[1]
+            local sockevents = v[2]
+            table.insert(results, {poller.OT_SOCK,  sockevents, self:_sockdata(csock)})
         end
     end
     return count, results
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index 52502c066..5cd4ba16c 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -94,16 +94,6 @@ function scheduler:_timer()
     return t
 end
 
--- wait the current coroutine
-function scheduler:_co_wait(...)
-    return self:co_suspend(...)
-end
-
--- wake the given coroutine
-function scheduler:_co_wake(co, ...)
-    return self:co_resume(co, ...)
-end
-
 -- start a new coroutine task
 function scheduler:co_start(cotask, ...)
     return self:co_start_named(nil, cotask, ...)
@@ -150,19 +140,61 @@ end
 -- wait socket events
 function scheduler:waitsock(sock, events, timeout)
 
+    -- get the running coroutine
+    local running = self:co_running()
+    if not running then
+        return -1, "we must call waitsock() in coroutine with scheduler!"
+    end
+
+    io.print("wait sock: %s events: %d timeout: %d", sock, events, timeout)
+
+    -- the socket events callback
+    local function sockevents_cb(sockevents)
+        -- TODO
+        self:co_resume(running, sockevents)
+    end
+
+    -- add socket events to poller
     -- TODO
-    return 0
+    
+    -- insert socket to poller for waiting events
+    local ok, errors = poller:insert(poller.OT_SOCK, sock, events, sockevents_cb)
+    if not ok then
+        return -1, errors
+    end
+
+    -- register timeout task to timer
+    if timeout > 0 then
+        self:_timer():post(function (cancel) 
+            -- TODO
+            self:co_resume(running, 0)
+        end, timeout)
+    end
+
+    -- wait
+    return self:co_suspend()
 end
 
 -- sleep some times (ms)
 function scheduler:sleep(ms)
+
+    -- we need not do sleep 
+    if ms == 0 then
+        return true
+    end
+
+    -- get the running coroutine
     local running = self:co_running()
     if not running then
         return false, "we must call sleep() in coroutine with scheduler!"
     end
+
+    -- register timeout task to timer
     self:_timer():post(function (cancel) 
         self:co_resume(running)
     end, ms)
+
+    -- wait
     self:co_suspend()
     return true
 end
@@ -199,11 +231,24 @@ function scheduler:runloop()
             break
         end
 
+        -- resume all suspended tasks with events
+        for _, e in ipairs(events) do
+            local otype = e[1]
+            if otype == poller.OT_SOCK then
+                local sockevents = e[2]
+                local sockfunc   = e[3]
+                if sockfunc then
+                    sockfunc(sockevents)
+                end
+            else
+                ok = false
+                errors = string.format("invalid poller object type(%d)", otype)
+                break
+            end
+        end
+
         -- spank the timer and trigger all timeout tasks
         self:_timer():next()
-
-        -- resume all suspended tasks with events
-        -- TODO
     end
 
     -- mark the loop as stopped first
diff --git a/xmake/core/base/socket.lua b/xmake/core/base/socket.lua
index ff362a132..4b459cc39 100644
--- a/xmake/core/base/socket.lua
+++ b/xmake/core/base/socket.lua
@@ -64,6 +64,11 @@ function _instance:family()
     return self._FAMILY
 end
 
+-- get cdata socket
+function _instance:csock()
+    return self._SOCK
+end
+
 -- get socket rawfd
 function _instance:rawfd()
 
@@ -74,7 +79,7 @@ function _instance:rawfd()
     end
 
     -- get rawfd
-    local result, errors = io.socket_rawfd(self._SOCK)
+    local result, errors = io.socket_rawfd(self:csock())
     if not result and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -91,7 +96,7 @@ function _instance:bind(addr, port)
     end
 
     -- bind it
-    local ok, errors = io.socket_bind(self._SOCK, addr, port, self:family())
+    local ok, errors = io.socket_bind(self:csock(), addr, port, self:family())
     if not ok and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -108,7 +113,7 @@ function _instance:listen(backlog)
     end
 
     -- listen it
-    local ok, errors = io.socket_listen(self._SOCK, backlog or 10)
+    local ok, errors = io.socket_listen(self:csock(), backlog or 10)
     if not ok and errors then
         errors = string.format("%s: %s", self, errors)
     end
@@ -125,12 +130,12 @@ function _instance:accept(opt)
     end
 
     -- accept it
-    local sock, errors = io.socket_accept(self._SOCK)
+    local sock, errors = io.socket_accept(self:csock())
     if not sock and not errors then
         opt = opt or {}
         local events, waiterrs = self:wait(socket.EV_ACPT, opt.timeout or -1)
         if events == socket.EV_ACPT then
-            sock, errors = io.socket_accept(self._SOCK)
+            sock, errors = io.socket_accept(self:csock())
         else
             errors = waiterrs
         end
@@ -154,12 +159,12 @@ function _instance:connect(addr, port, opt)
     end
 
     -- connect it
-    local ok, errors = io.socket_connect(self._SOCK, addr, port, self:family())
+    local ok, errors = io.socket_connect(self:csock(), addr, port, self:family())
     if ok == 0 then
         opt = opt or {}
         local events, waiterrs = self:wait(socket.EV_CONN, opt.timeout or -1)
         if events == socket.EV_CONN then
-            ok, errors = io.socket_connect(self._SOCK, addr, port, self:family())
+            ok, errors = io.socket_connect(self:csock(), addr, port, self:family())
         else
             errors = waiterrs
         end
@@ -204,7 +209,7 @@ function _instance:send(data, opt)
     if opt.block then
         local size = last + 1 - start
         while start <= last do
-            real, errors = io.socket_send(self._SOCK, data, start, last)
+            real, errors = io.socket_send(self:csock(), data, start, last)
             if real > 0 then
                 send = send + real
                 start = start + real
@@ -225,7 +230,7 @@ function _instance:send(data, opt)
             send = -1
         end
     else
-        send, errors = io.socket_send(self._SOCK, data, start, last)
+        send, errors = io.socket_send(self:csock(), data, start, last)
         if send < 0 and errors then
             errors = string.format("%s: %s", self, errors)
         end
@@ -266,7 +271,7 @@ function _instance:sendfile(file, opt)
     if opt.block then
         local size = last + 1 - start
         while start <= last do
-            real, errors = io.socket_sendfile(self._SOCK, file._FILE, start, last)
+            real, errors = io.socket_sendfile(self:csock(), file._FILE, start, last)
             if real > 0 then
                 send = send + real
                 start = start + real
@@ -287,7 +292,7 @@ function _instance:sendfile(file, opt)
             send = -1
         end
     else
-        send, errors = io.socket_sendfile(self._SOCK, file._FILE, start, last)
+        send, errors = io.socket_sendfile(self:csock(), file._FILE, start, last)
         if send < 0 and errors then
             errors = string.format("%s: %s", self, errors)
         end
@@ -320,7 +325,7 @@ function _instance:recv(size, opt)
     if opt.block then
         local results = {}
         while recv < size do
-            real, data_or_errors = io.socket_recv(self._SOCK, size - recv)
+            real, data_or_errors = io.socket_recv(self:csock(), size - recv)
             if real > 0 then
                 recv = recv + real
                 wait = false
@@ -343,7 +348,7 @@ function _instance:recv(size, opt)
             recv = -1
         end
     else
-        recv, data_or_errors = io.socket_recv(self._SOCK, size)
+        recv, data_or_errors = io.socket_recv(self:csock(), size)
         if recv > 0 then
             data_or_errors = bytes(data_or_errors)
         end
@@ -385,7 +390,7 @@ function _instance:sendto(data, addr, port, opt)
     local errors = nil
     if opt.block then
         while true do
-            send, errors = io.socket_sendto(self._SOCK, data, addr, port, self:family())
+            send, errors = io.socket_sendto(self:csock(), data, addr, port, self:family())
             if send == 0 and not wait then
                 local events, waiterrs = self:wait(socket.EV_SEND, opt.timeout or -1)
                 if events == socket.EV_SEND then
@@ -399,7 +404,7 @@ function _instance:sendto(data, addr, port, opt)
             end
         end
     else
-        send, errors = io.socket_sendto(self._SOCK, data, addr, port, self:family())
+        send, errors = io.socket_sendto(self:csock(), data, addr, port, self:family())
         if send < 0 and errors then
             errors = string.format("%s: %s", self, errors)
         end
@@ -435,7 +440,7 @@ function _instance:recvfrom(size, opt)
     local data_or_errors = nil
     if opt.block then
         while true do
-            recv, data_or_errors, addr, port = io.socket_recvfrom(self._SOCK, size)
+            recv, data_or_errors, addr, port = io.socket_recvfrom(self:csock(), size)
             if recv > 0 then
                 data_or_errors = bytes(data_or_errors)
                 break
@@ -453,7 +458,7 @@ function _instance:recvfrom(size, opt)
             end
         end
     else
-        recv, data_or_errors, addr, port = io.socket_recvfrom(self._SOCK, size)
+        recv, data_or_errors, addr, port = io.socket_recvfrom(self:csock(), size)
         if recv > 0 then
             data_or_errors = bytes(data_or_errors)
         end
@@ -474,17 +479,17 @@ function _instance:wait(events, timeout)
     end
 
     -- wait events
-    local events = -1
+    local result = -1
     local errors = nil
     if scheduler:co_running() then
-        events, errors = scheduler:waitsock(self._SOCK, events, timeout or -1)
+        result, errors = scheduler:waitsock(self, events, timeout or -1)
     else
-        events, errors = io.socket_wait(self._SOCK, events, timeout or -1)
+        result, errors = io.socket_wait(self:csock(), events, timeout or -1)
     end
-    if events < 0 and errors then
+    if result < 0 and errors then
         errors = string.format("%s: %s", self, errors)
     end
-    return events, errors
+    return result, errors
 end
 
 -- close socket
@@ -497,7 +502,7 @@ function _instance:close()
     end
 
     -- close it
-    ok = io.socket_close(self._SOCK)
+    ok = io.socket_close(self:csock())
     if ok then
         self._SOCK = nil
     end
@@ -506,7 +511,7 @@ end
 
 -- ensure the socket is opened
 function _instance:_ensure_opened()
-    if not self._SOCK then
+    if not self:csock() then
         return false, string.format("%s: has been closed!", self)
     end
     return true
@@ -521,7 +526,7 @@ end
 
 -- gc(socket)
 function _instance:__gc()
-    if self._SOCK and io.socket_close(self._SOCK) then
+    if self:csock() and io.socket_close(self:csock()) then
         self._SOCK = nil
     end
 end