add CallbackManager

Author: matyhtf

config.m4

@@ -470,7 +470,6 @@ if test "$PHP_SWOOLE" != "no"; then
         src/protocol/ssl.c \
         src/protocol/websocket.c \
         src/reactor/base.cc \
-        src/reactor/defer_task.cc \
         src/reactor/epoll.c \
         src/reactor/kqueue.c \
         src/reactor/poll.c \

include/async.h

@@ -77,7 +77,6 @@ typedef struct
 
 extern swAsyncIO SwooleAIO;
 
-void swAio_free(void);
 int swAio_dispatch(const swAio_event *request);
 swAio_event* swAio_dispatch2(const swAio_event *request);
 int swAio_cancel(int task_id);

include/swoole.h

@@ -1664,7 +1664,7 @@ struct _swReactor
     void (*free)(swReactor *);
 
     void *defer_tasks;
-    void (*do_defer_tasks)(swReactor *);
+    void *destroy_callbacks;
 
     swDefer_callback idle_task;
     swDefer_callback future_task;
@@ -1891,6 +1891,7 @@ static sw_inline int swReactor_events(int fdtype)
 
 int swReactor_create(swReactor *reactor, int max_event);
 void swReactor_destory(swReactor *reactor);
+void swReactor_add_destroy_callback(swReactor *reactor, swCallback cb, void *data);
 
 static inline void swReactor_before_wait(swReactor *reactor)
 {
@@ -1902,9 +1903,6 @@ static inline void swReactor_before_wait(swReactor *reactor)
 
 int swReactor_empty(swReactor *reactor);
 
-void swReactor_defer_task_create(swReactor *reactor);
-void swReactor_defer_task_destroy(swReactor *reactor);
-
 static sw_inline swConnection* swReactor_get(swReactor *reactor, int fd)
 {
     swConnection *socket = reactor->thread ? &reactor->socket_list[fd] : (swConnection*) swArray_alloc(reactor->socket_array, fd);

include/swoole_cxx.h

@@ -2,12 +2,13 @@
 
 #include "swoole.h"
 
+#include <list>
 #include <memory>
 #include <string>
 #include <cstdio>
 
-namespace swoole
-{
+namespace swoole {
+//-------------------------------------------------------------------------------
 namespace cpp_string
 {
 template<typename ...Args>
@@ -30,4 +31,44 @@ inline std::string vformat(const char *format, va_list args)
     return std::string(buf.get(), buf.get() + size - 1); // We don't want the '\0' inside
 }
 }
+
+struct Callback
+{
+    swCallback callback;
+    void *private_data;
+
+    Callback(swCallback cb, void *_private_data)
+
+    {
+        callback = cb;
+        private_data = _private_data;
+    }
+};
+
+class CallbackManager
+{
+public:
+    inline void append(swCallback cb, void *private_data)
+    {
+        list_.push_back(new Callback(cb, private_data));
+    }
+    inline void prepend(swCallback cb, void *private_data)
+    {
+        list_.push_front(new Callback(cb, private_data));
+    }
+    inline void execute()
+    {
+        while (!list_.empty())
+        {
+            Callback *task = list_.front();
+            list_.pop_front();
+            task->callback(task->private_data);
+            delete task;
+        }
+    }
+protected:
+    std::list<Callback *> list_;
+};
+
+//-------------------------------------------------------------------------------
 }

src/network/async_thread.cc

@@ -30,6 +30,8 @@ typedef swAio_event async_event;
 
 swAsyncIO SwooleAIO;
 
+static void swAio_free(void *private_data);
+
 class async_event_queue
 {
 public:
@@ -317,6 +319,8 @@ static int swAio_init()
         SwooleAIO.max_thread_count = SwooleAIO.min_thread_count;
     }
 
+    swReactor_add_destroy_callback(SwooleG.main_reactor, swAio_free, nullptr);
+
     pool = new async_thread_pool(SwooleAIO.min_thread_count, SwooleAIO.min_thread_count);
     pool->start();
     SwooleAIO.init = 1;
@@ -345,7 +349,7 @@ swAio_event* swAio_dispatch2(const swAio_event *request)
     return pool->dispatch(request);
 }
 
-void swAio_free(void)
+static void swAio_free(void *private_data)
 {
     if (!SwooleAIO.init)
     {

src/reactor/base.cc

@@ -15,9 +15,12 @@
 */
 
 #include "swoole.h"
+#include "swoole_cxx.h"
 #include "connection.h"
 #include "async.h"
 
+using swoole::CallbackManager;
+
 #ifdef SW_USE_MALLOC_TRIM
 #ifdef __APPLE__
 #include <sys/malloc.h>
@@ -26,9 +29,11 @@
 #endif
 #endif
 
-static void swReactor_onTimeout(swReactor *reactor);
-static void swReactor_onFinish(swReactor *reactor);
-static void swReactor_onBegin(swReactor *reactor);
+static void reactor_timeout(swReactor *reactor);
+static void reactor_finish(swReactor *reactor);
+static void reactor_begin(swReactor *reactor);
+static void defer_task_do(swReactor *reactor);
+static void defer_task_add(swReactor *reactor, swCallback callback, void *data);
 
 int swReactor_create(swReactor *reactor, int max_event)
 {
@@ -47,13 +52,15 @@ int swReactor_create(swReactor *reactor, int max_event)
 
     reactor->running = 1;
 
-    reactor->onFinish = swReactor_onFinish;
-    reactor->onTimeout = swReactor_onTimeout;
+    reactor->onFinish = reactor_finish;
+    reactor->onTimeout = reactor_timeout;
     reactor->is_empty = swReactor_empty;
 
     reactor->write = swReactor_write;
     reactor->close = swReactor_close;
-    swReactor_defer_task_create(reactor);
+
+    reactor->defer = defer_task_add;
+    reactor->defer_tasks = nullptr;
 
     reactor->socket_array = swArray_new(1024, sizeof(swConnection));
     if (!reactor->socket_array)
@@ -132,15 +139,18 @@ int swReactor_empty(swReactor *reactor)
 /**
  * execute when reactor timeout and reactor finish
  */
-static void swReactor_onFinish(swReactor *reactor)
+static void reactor_finish(swReactor *reactor)
 {
     //check timer
     if (reactor->check_timer)
     {
         swTimer_select(&SwooleG.timer);
     }
     //defer tasks
-    reactor->do_defer_tasks(reactor);
+    if (reactor->defer_tasks)
+    {
+        defer_task_do(reactor);
+    }
     //callback at the end
     if (reactor->idle_task.callback)
     {
@@ -167,9 +177,9 @@ static void swReactor_onFinish(swReactor *reactor)
 #endif
 }
 
-static void swReactor_onTimeout(swReactor *reactor)
+static void reactor_timeout(swReactor *reactor)
 {
-    swReactor_onFinish(reactor);
+    reactor_finish(reactor);
 
     if (reactor->disable_accept)
     {
@@ -180,10 +190,10 @@ static void swReactor_onTimeout(swReactor *reactor)
 
 void swReactor_activate_future_task(swReactor *reactor)
 {
-    reactor->onBegin = swReactor_onBegin;
+    reactor->onBegin = reactor_begin;
 }
 
-static void swReactor_onBegin(swReactor *reactor)
+static void reactor_begin(swReactor *reactor)
 {
     if (reactor->future_task.callback)
     {
@@ -397,8 +407,50 @@ int swReactor_wait_write_buffer(swReactor *reactor, int fd)
     return SW_OK;
 }
 
+void swReactor_add_destroy_callback(swReactor *reactor, swCallback cb, void *data)
+{
+    CallbackManager *cm = (CallbackManager *) reactor->destroy_callbacks;
+    if (cm == nullptr)
+    {
+        cm = new CallbackManager;
+        reactor->destroy_callbacks = cm;
+    }
+    cm->append(cb, data);
+}
+
+void swReactor_defer_task_destroy(swReactor *reactor)
+{
+    CallbackManager *tasks = (CallbackManager *) reactor->defer_tasks;
+    delete tasks;
+}
+
+static void defer_task_do(swReactor *reactor)
+{
+    CallbackManager *cm = (CallbackManager *) reactor->defer_tasks;
+    cm->execute();
+    reactor->defer_tasks = nullptr;
+    delete cm;
+}
+
+static void defer_task_add(swReactor *reactor, swCallback callback, void *data)
+{
+    CallbackManager *cm = (CallbackManager *) reactor->defer_tasks;
+    if (cm == nullptr)
+    {
+        cm = new CallbackManager;
+        reactor->defer_tasks = cm;
+    }
+    cm->append(callback, data);
+}
+
 void swReactor_destory(swReactor *reactor)
 {
-    swAio_free();
+    if (reactor->destroy_callbacks)
+    {
+        CallbackManager *cm = (CallbackManager *) reactor->destroy_callbacks;
+        cm->execute();
+        reactor->destroy_callbacks = nullptr;
+        delete cm;
+    }
     reactor->free(reactor);
 }

src/reactor/defer_task.cc

@@ -28,21 +28,13 @@
 #include <ifaddrs.h>
 #include <sys/ioctl.h>
 
-#include <queue>
-
 #include "zend_exceptions.h"
 
 ZEND_DECLARE_MODULE_GLOBALS(swoole)
 
 extern sapi_module_struct sapi_module;
 
-struct rshutdown_func
-{
-    swCallback callback;
-    void *private_data;
-};
-
-std::queue<rshutdown_func *> rshutdown_functions;
+static swoole::CallbackManager rshutdown_callbacks;
 
 ZEND_BEGIN_ARG_INFO_EX(arginfo_swoole_void, 0, 0, 0)
 ZEND_END_ARG_INFO()
@@ -321,19 +313,12 @@ void php_swoole_register_shutdown_function_prepend(const char *function)
 
 void php_swoole_register_rshutdown_callback(swCallback cb, void *private_data)
 {
-    rshutdown_func *rf = (rshutdown_func*) emalloc(sizeof(rshutdown_func));
-    rshutdown_functions.push(rf);
+    rshutdown_callbacks.append(cb, private_data);
 }
 
 static void execute_rshutdown_callback()
 {
-    while(!rshutdown_functions.empty())
-    {
-        rshutdown_func *rf = rshutdown_functions.front();
-        rshutdown_functions.pop();
-        rf->callback(rf->private_data);
-        efree(rf);
-    }
+    rshutdown_callbacks.execute();
 }
 
 static void fatal_error(int code, const char *format, ...)