impl scheduler.stop

5 files changed, 87 insertions, 26 deletions

diff --git a/core/src/xmake/io/poller_spank.c b/core/src/xmake/io/poller_spank.c
new file mode 100644
index 000000000..e5f6ba5ab
--- /dev/null
+++ b/core/src/xmake/io/poller_spank.c
@@ -0,0 +1,48 @@
+/*!A cross-platform build utility based on Lua
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Copyright (C) 2015 - 2019, TBOOX Open Source Group.
+ *
+ * @author      ruki
+ * @file        poller_spank.c
+ *
+ */
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * trace
+ */
+#define TB_TRACE_MODULE_NAME    "poller_spank"
+#define TB_TRACE_MODULE_DEBUG   (0)
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * includes
+ */
+#include "prefix.h"
+#include "poller.h"
+
+/* //////////////////////////////////////////////////////////////////////////////////////
+ * interfaces
+ */
+
+// io.poller_spank()
+tb_int_t xm_io_poller_spank(lua_State* lua)
+{
+    // check
+    tb_assert_and_check_return_val(lua, 0);
+
+    // spank the poller, break the tb_poller_wait() and return all events
+    tb_poller_spak(xm_io_poller());
+    return 0;
+}
+
diff --git a/core/src/xmake/machine.c b/core/src/xmake/machine.c
index 4fdc73204..28903ec3a 100644
--- a/core/src/xmake/machine.c
+++ b/core/src/xmake/machine.c
@@ -128,6 +128,7 @@ tb_int_t xm_io_socket_close(lua_State* lua);
 tb_int_t xm_io_poller_insert(lua_State* lua);
 tb_int_t xm_io_poller_modify(lua_State* lua);
 tb_int_t xm_io_poller_remove(lua_State* lua);
+tb_int_t xm_io_poller_spank(lua_State* lua);
 tb_int_t xm_io_poller_support(lua_State* lua);
 tb_int_t xm_io_poller_wait(lua_State* lua);
 
@@ -279,6 +280,7 @@ static luaL_Reg const g_io_functions[] =
 ,   { "poller_insert",      xm_io_poller_insert    }
 ,   { "poller_modify",      xm_io_poller_modify    }
 ,   { "poller_remove",      xm_io_poller_remove    }
+,   { "poller_spank",       xm_io_poller_spank     }
 ,   { "poller_support",     xm_io_poller_support   }
 ,   { "poller_wait",        xm_io_poller_wait      }
 ,   { tb_null,              tb_null                }
diff --git a/core/src/xmake/makefile b/core/src/xmake/makefile
index 67a340933..935e210a6 100644
--- a/core/src/xmake/makefile
+++ b/core/src/xmake/makefile
@@ -65,6 +65,7 @@ xmake_C_FILES 			+= \
 						io/poller_insert \
 						io/poller_remove \
 						io/poller_modify \
+						io/poller_spank \
 						io/poller_support \
 						io/socket_open \
 						io/socket_rawfd \
diff --git a/xmake/core/base/poller.lua b/xmake/core/base/poller.lua
index f964ae8bc..8c4085670 100644
--- a/xmake/core/base/poller.lua
+++ b/xmake/core/base/poller.lua
@@ -120,6 +120,11 @@ function poller:support(otype, events)
     return false, string.format("invalid poller object type(%d)!", otype)
 end
 
+-- spank poller to break the wait() and return all triggered events
+function poller:spank()
+    io.poller_spank()
+end
+
 -- insert object events to poller
 function poller:insert(otype, obj, events, udata)
     if otype == poller.OT_SOCK then
diff --git a/xmake/core/base/scheduler.lua b/xmake/core/base/scheduler.lua
index b8f3ca8c1..50e42c5b6 100644
--- a/xmake/core/base/scheduler.lua
+++ b/xmake/core/base/scheduler.lua
@@ -129,7 +129,7 @@ function scheduler:_sockevents_cb(sock, sockevents)
     local events_prev_save = bit.rshift(events_prev, 16)
 
     -- is waiting?
-    local running = self:_co_suspended(sock)
+    local running = self:_co_sock_suspended(sock)
     if running and running:is_suspended() then
     
         -- eof for edge trigger?
@@ -152,7 +152,7 @@ function scheduler:_sockevents_cb(sock, sockevents)
         end
 
         -- resume this coroutine task
-        self:_co_suspended_set(sock, nil)
+        self:_co_sock_suspended_set(sock, nil)
         self:co_resume(running, (bit.band(sockevents, poller.EV_SOCK_ERROR) ~= 0) and -1 or sockevents)
     else
         -- cache socket events
@@ -162,18 +162,32 @@ function scheduler:_sockevents_cb(sock, sockevents)
 end
 
 -- get the suspended coroutine task
-function scheduler:_co_suspended(key)
-    return self._CO_SUSPENDED_TASKS and self._CO_SUSPENDED_TASKS[key] or nil
+function scheduler:_co_sock_suspended(sock)
+    return self._CO_SOCK_SUSPENDED_TASKS and self._CO_SOCK_SUSPENDED_TASKS[sock] or nil
 end
 
 -- set the suspended coroutine task
-function scheduler:_co_suspended_set(key, co)
-    local co_suspended_tasks = self._CO_SUSPENDED_TASKS 
-    if not co_suspended_tasks then
-        co_suspended_tasks = {}
-        self._CO_SUSPENDED_TASKS = co_suspended_tasks
+function scheduler:_co_sock_suspended_set(sock, co)
+    local co_sock_suspended_tasks = self._CO_SOCK_SUSPENDED_TASKS 
+    if not co_sock_suspended_tasks then
+        co_sock_suspended_tasks = {}
+        self._CO_SOCK_SUSPENDED_TASKS = co_sock_suspended_tasks
+    end
+    co_sock_suspended_tasks[sock] = co
+end
+
+-- cancel and resume all suspended socket tasks after stopping scheduler
+-- we cannot suspend them forever, all tasks will be exited directly and free all resources.
+function scheduler:_co_sock_suspended_cancel_all()
+    local co_sock_suspended_tasks = self._CO_SOCK_SUSPENDED_TASKS
+    if co_sock_suspended_tasks then
+        for _, co in pairs(co_sock_suspended_tasks) do
+            local ok, errors = self:co_resume(co, -1) 
+            if not ok then
+                return false, errors
+            end
+        end
     end
-    co_suspended_tasks[key] = co
 end
 
 -- start a new coroutine task
@@ -298,7 +312,7 @@ function scheduler:sock_wait(sock, events, timeout)
     if timeout > 0 then
         timer_task = self:_timer():post(function (cancel) 
             if not cancel and running:is_suspended() then
-                self:_co_suspended_set(sock, nil)
+                self:_co_sock_suspended_set(sock, nil)
                 self:co_resume(running, 0)
             end
         end, timeout)
@@ -309,7 +323,7 @@ function scheduler:sock_wait(sock, events, timeout)
     self:_sockevents_set(sock:csock(), events)
 
     -- save the suspended coroutine
-    self:_co_suspended_set(sock, running)
+    self:_co_sock_suspended_set(sock, running)
 
     -- wait
     return self:co_suspend()
@@ -328,7 +342,7 @@ function scheduler:sock_cancel(sock)
             return false, errors
         end
         self:_sockevents_set(sock:csock(), 0)
-        self:_co_suspended_set(sock, nil)
+        self:_co_sock_suspended_set(sock, nil)
     end
     return true
 end
@@ -366,9 +380,9 @@ end
 
 -- stop the scheduler loop
 function scheduler:stop()
-    -- TODO post a kill signal to poller
-    -- stop timer and cancel all tasks
+    -- mark scheduler status as stopped and spank the poller:wait()
     self._STARTED = false
+    poller:spank()
     return true
 end
 
@@ -434,17 +448,8 @@ function scheduler:runloop()
     -- mark the loop as stopped first
     self._STARTED = false
 
-    -- resume all suspended tasks after stopping scheduler
-    -- we cannot suspend them now, all tasks will be exited directly and free all resources.
-    local co_suspended_tasks = self._CO_SUSPENDED_TASKS
-    if co_suspended_tasks then
-        for _, co in pairs(co_suspended_tasks) do
-            local ok, errors = self:co_resume(co)
-            if not ok then
-                return false, errors
-            end
-        end
-    end
+    -- cancel all suspended tasks after stopping scheduler
+    self:_co_sock_suspended_cancel_all()
 
     -- cancel all timeout tasks and trigger them
     self:_timer():kill()