fix server shutdown

Author: matyhtf

ext-src/swoole_server.cc

@@ -2641,31 +2641,25 @@ static PHP_METHOD(swoole_server, start) {
     ServerObject *server_object = server_fetch_object(Z_OBJ_P(php_swoole_server_zval_ptr(serv)));
 
 #ifdef SW_THREAD
+    zend_string *bootstrap = nullptr;
+    zend_string *thread_argv_serialized = nullptr;
+    zval thread_argv = {};
+
     if (serv->is_thread_mode()) {
         zval *_bootstrap = zend::object_get(ZEND_THIS, ZEND_STRL("bootstrap"));
-        zend_string *bootstrap = zend_string_dup(Z_STR_P(_bootstrap), 1);
-        zend_string *argv = nullptr;
-        zval thread_argv = {};
+        bootstrap = zend_string_dup(Z_STR_P(_bootstrap), 1);
 
         if (!ZVAL_IS_NULL(&server_object->init_arguments)) {
             call_user_function(NULL, NULL, &server_object->init_arguments, &thread_argv, 0, NULL);
-            argv = php_swoole_thread_serialize(&thread_argv);
+            thread_argv_serialized = php_swoole_thread_serialize(&thread_argv);
         }
 
-        serv->worker_thread_start = [bootstrap, argv](const WorkerFn &fn) {
+        serv->worker_thread_start = [bootstrap, thread_argv_serialized](const WorkerFn &fn) {
             worker_thread_fn = fn;
             zend_string *bootstrap_copy = zend_string_dup(bootstrap, 1);
-            zend_string *argv_copy = argv ? zend_string_dup(argv, 1) : nullptr;
+            zend_string *argv_copy = thread_argv_serialized ? zend_string_dup(thread_argv_serialized, 1) : nullptr;
             php_swoole_thread_start(bootstrap_copy, argv_copy);
         };
-
-        ON_SCOPE_EXIT {
-            zend_string_release(bootstrap);
-            if (argv) {
-                zend_string_release(argv);
-            }
-            zval_ptr_dtor(&thread_argv);
-        };
     }
 #endif
 
@@ -2675,6 +2669,17 @@ static PHP_METHOD(swoole_server, start) {
     if (serv->start() < 0) {
         php_swoole_fatal_error(E_ERROR, "failed to start server. Error: %s", sw_error);
     }
+
+#ifdef SW_THREAD
+    if (bootstrap) {
+        zend_string_release(bootstrap);
+    }
+    if (thread_argv_serialized) {
+        zend_string_release(thread_argv_serialized);
+    }
+    zval_ptr_dtor(&thread_argv);
+#endif
+
     RETURN_TRUE;
 }
 

include/swoole_server.h

@@ -160,6 +160,8 @@ struct ReactorThread {
 
     int init(Server *serv, Reactor *reactor, uint16_t reactor_id);
     void shutdown(Reactor *reactor);
+    int close_connection(Reactor *reactor, SessionId session_id);
+    void clean();
 };
 
 struct ServerPortGS {
@@ -1075,6 +1077,10 @@ class Server {
         }
     }
 
+    bool is_master_thread() {
+        return swoole_get_thread_type() == Server::THREAD_MASTER;
+    }
+
     bool is_hash_dispatch_mode() {
         return dispatch_mode == DISPATCH_FDMOD || dispatch_mode == DISPATCH_IPMOD ||
                dispatch_mode == DISPATCH_CO_CONN_LB;
@@ -1311,7 +1317,7 @@ class Server {
 
     void call_hook(enum HookType type, void *arg);
     void call_worker_start_callback(Worker *worker);
-    ResultCode call_command_handler(MessageBus &mb, uint16_t worker_id, network::Socket *sock);
+    void call_command_handler(MessageBus &mb, uint16_t worker_id, network::Socket *sock);
     std::string call_command_handler_in_master(int command_id, const std::string &msg);
     void call_command_callback(int64_t request_id, const std::string &result);
     void foreach_connection(const std::function<void(Connection *)> &callback);

src/server/master.cc

@@ -93,13 +93,13 @@ void Server::call_command_callback(int64_t request_id, const std::string &result
     iter->second(this, result);
 }
 
-ResultCode Server::call_command_handler(MessageBus &mb, uint16_t worker_id, Socket *sock) {
+void Server::call_command_handler(MessageBus &mb, uint16_t worker_id, Socket *sock) {
     PipeBuffer *buffer = mb.get_buffer();
     int command_id = buffer->info.server_fd;
     auto iter = command_handlers.find(command_id);
     if (iter == command_handlers.end()) {
         swoole_error_log(SW_LOG_ERROR, SW_ERROR_SERVER_INVALID_COMMAND, "Unknown command[command_id=%d]", command_id);
-        return SW_OK;
+        return;
     }
 
     Server::Command::Handler handler = iter->second;
@@ -114,7 +114,7 @@ ResultCode Server::call_command_handler(MessageBus &mb, uint16_t worker_id, Sock
     task.info.len = result.length();
     task.data = result.c_str();
 
-    return mb.write(sock, &task) ? SW_OK : SW_ERR;
+    mb.write(sock, &task);
 }
 
 std::string Server::call_command_handler_in_master(int command_id, const std::string &msg) {
@@ -877,7 +877,12 @@ bool Server::shutdown() {
     if (is_base_mode()) {
         pid = get_manager_pid() == 0 ? get_master_pid() : get_manager_pid();
     } else if (is_thread_mode()) {
-        return factory->shutdown();
+        if (is_master_thread()) {
+            stop_master_thread();
+        } else {
+            running = false;
+        }
+        return true;
     } else {
         pid = get_master_pid();
     }
@@ -943,6 +948,14 @@ void Server::stop_master_thread() {
         };
         reactor->set_exit_condition(Reactor::EXIT_CONDITION_FORCED_TERMINATION, fn);
     }
+    if (is_thread_mode()) {
+        SW_LOOP_N(reactor_num) {
+            auto thread = get_thread(i);
+            DataHead ev = {};
+            ev.type = SW_SERVER_EVENT_SHUTDOWN;
+            thread->notify_pipe->send_blocking((void *) &ev, sizeof(ev));
+        }
+    }
 }
 
 bool Server::signal_handler_shutdown() {
@@ -1004,7 +1017,7 @@ void Server::destroy() {
         if (task_worker_num > 0) {
             gs->task_workers.destroy();
         }
-    } else {
+    } else if (is_process_mode()) {
         swoole_trace_log(SW_TRACE_SERVER, "terminate reactor threads");
         /**
          * Wait until all the end of the thread
@@ -1646,6 +1659,11 @@ void Server::timer_callback(Timer *timer, TimerNode *tnode) {
     if (serv->hooks[Server::HOOK_MASTER_TIMER]) {
         serv->call_hook(Server::HOOK_MASTER_TIMER, serv);
     }
+
+    if (!serv->is_running()) {
+        sw_reactor()->running = false;
+        serv->stop_master_thread();
+    }
 }
 
 int Server::add_worker(Worker *worker) {
@@ -1664,7 +1682,7 @@ bool Server::add_command(const std::string &name, int accepted_process_types, co
     if (commands.find(name) != commands.end()) {
         return false;
     }
-    if (is_process_mode() && pipe_command == nullptr) {
+    if (!is_base_mode() && pipe_command == nullptr) {
         auto _pipe = new UnixSocket(false, SOCK_DGRAM);
         if (!_pipe->ready()) {
             delete _pipe;

src/server/reactor_thread.cc

@@ -329,8 +329,39 @@ void ReactorThread::shutdown(Reactor *reactor) {
     });
 
     reactor->set_wait_exit(true);
+}
 
-    message_bus.free_buffer();
+int ReactorThread::close_connection(Reactor *reactor, SessionId session_id) {
+    Server *serv = (Server *) reactor->ptr;
+    Connection *conn = serv->get_connection_verify_no_ssl(session_id);
+    if (!conn) {
+        swoole_error_log(SW_LOG_TRACE,
+                         SW_ERROR_SESSION_NOT_EXIST,
+                         "force close connection failed, session#%ld does not exist",
+                         session_id);
+        return SW_OK;
+    }
+
+    if (serv->disable_notify || conn->close_force) {
+        return Server::close_connection(reactor, conn->socket);
+    }
+
+#ifdef SW_USE_OPENSSL
+    /**
+     * SSL connections that have not completed the handshake,
+     * do not need to notify the workers, just close
+     */
+    if (conn->ssl && !conn->ssl_ready) {
+        return Server::close_connection(reactor, conn->socket);
+    }
+#endif
+    conn->close_force = 1;
+    Event _ev = {};
+    _ev.fd = conn->fd;
+    _ev.socket = conn->socket;
+    reactor->trigger_close_event(&_ev);
+
+    return SW_OK;
 }
 
 /**
@@ -350,49 +381,21 @@ static int ReactorThread_onPipeRead(Reactor *reactor, Event *ev) {
         if (resp->info.type == SW_SERVER_EVENT_INCOMING) {
             Connection *conn = serv->get_connection_verify_no_ssl(resp->info.fd);
             if (conn && serv->connection_incoming(reactor, conn) < 0) {
-                return reactor->close(reactor, conn->socket);
+                reactor->close(reactor, conn->socket);
             }
         } else if (resp->info.type == SW_SERVER_EVENT_COMMAND_REQUEST) {
-            return serv->call_command_handler(thread->message_bus, thread->id, thread->pipe_command);
+            serv->call_command_handler(thread->message_bus, thread->id, thread->pipe_command);
         } else if (resp->info.type == SW_SERVER_EVENT_COMMAND_RESPONSE) {
             auto packet = thread->message_bus.get_packet();
             serv->call_command_callback(resp->info.fd, std::string(packet.data, packet.length));
         } else if (resp->info.type == SW_SERVER_EVENT_SHUTDOWN) {
             thread->shutdown(reactor);
-            return SW_OK;
         } else if (resp->info.type == SW_SERVER_EVENT_FINISH) {
             serv->onFinish(serv, (EventData *) resp);
         } else if (resp->info.type == SW_SERVER_EVENT_PIPE_MESSAGE) {
             serv->onPipeMessage(serv, (EventData *) resp);
         } else if (resp->info.type == SW_SERVER_EVENT_CLOSE_FORCE) {
-            SessionId session_id = resp->info.fd;
-            Connection *conn = serv->get_connection_verify_no_ssl(session_id);
-            if (!conn) {
-                swoole_error_log(SW_LOG_TRACE,
-                                 SW_ERROR_SESSION_NOT_EXIST,
-                                 "force close connection failed, session#%ld does not exist",
-                                 session_id);
-                return SW_OK;
-            }
-
-            if (serv->disable_notify || conn->close_force) {
-                return Server::close_connection(reactor, conn->socket);
-            }
-
-#ifdef SW_USE_OPENSSL
-            /**
-             * SSL connections that have not completed the handshake,
-             * do not need to notify the workers, just close
-             */
-            if (conn->ssl && !conn->ssl_ready) {
-                return Server::close_connection(reactor, conn->socket);
-            }
-#endif
-            conn->close_force = 1;
-            Event _ev = {};
-            _ev.fd = conn->fd;
-            _ev.socket = conn->socket;
-            reactor->trigger_close_event(&_ev);
+            thread->close_connection(reactor, resp->info.fd);
         } else {
             PacketPtr packet = thread->message_bus.get_packet();
             _send.info = resp->info;
@@ -793,6 +796,15 @@ int ReactorThread::init(Server *serv, Reactor *reactor, uint16_t reactor_id) {
     return SW_OK;
 }
 
+void ReactorThread::clean() {
+    sw_free(pipe_sockets);
+    if (pipe_command) {
+        pipe_command->fd = -1;
+        delete pipe_command;
+    }
+    message_bus.free_buffer();
+}
+
 void Server::reactor_thread_main_loop(Server *serv, int reactor_id) {
     SwooleTG.id = reactor_id;
     SwooleTG.type = Server::THREAD_REACTOR;
@@ -824,11 +836,7 @@ void Server::reactor_thread_main_loop(Server *serv, int reactor_id) {
     if (serv->is_thread_mode()) {
         serv->worker_stop_callback(serv->get_worker(reactor_id));
     }
-    sw_free(thread->pipe_sockets);
-    if (thread->pipe_command) {
-        thread->pipe_command->fd = -1;
-        delete thread->pipe_command;
-    }
+    thread->clean();
 }
 
 static void ReactorThread_resume_data_receiving(Timer *timer, TimerNode *tnode) {

src/server/thread.cc

@@ -56,21 +56,16 @@ bool ThreadFactory::start() {
 }
 
 bool ThreadFactory::shutdown() {
-    server_->running = false;
-    SW_LOOP_N(server_->reactor_num) {
-        auto thread = server_->get_thread(i);
-        DataHead ev = {};
-        ev.type = SW_SERVER_EVENT_SHUTDOWN;
-        thread->notify_pipe->send_blocking((void *) &ev, sizeof(ev));
+    for (auto &thread : threads_) {
+        if (thread.joinable()) {
+            thread.join();
+        }
     }
-    server_->stop_master_thread();
     return true;
 }
 
 ThreadFactory::~ThreadFactory() {
-    for (auto &thread : threads_) {
-        thread.join();
-    }
+
 }
 
 void ThreadFactory::at_thread_exit(Worker *worker) {