[6.0]Server for thread mode

ext-src/swoole_http_server.cc

@@ -32,8 +32,8 @@ namespace WebSocket = swoole::websocket;
 zend_class_entry *swoole_http_server_ce;
 zend_object_handlers swoole_http_server_handlers;
 
-static std::queue<HttpContext *> queued_http_contexts;
-static std::unordered_map<SessionId, zend::Variable> client_ips;
+static SW_THREAD_LOCAL std::queue<HttpContext *> queued_http_contexts;
+static SW_THREAD_LOCAL std::unordered_map<SessionId, zend::Variable> client_ips;
 
 static bool http_context_send_data(HttpContext *ctx, const char *data, size_t length);
 static bool http_context_sendfile(HttpContext *ctx, const char *file, uint32_t l_file, off_t offset, size_t length);

ext-src/swoole_server.cc

@@ -1868,11 +1868,14 @@ static PHP_METHOD(swoole_server, __construct) {
     Z_PARAM_LONG(serv_mode)
     Z_PARAM_LONG(sock_type)
     ZEND_PARSE_PARAMETERS_END_EX(RETURN_FALSE);
-
+#ifndef SW_THREAD
     if (serv_mode != Server::MODE_BASE && serv_mode != Server::MODE_PROCESS) {
         zend_throw_error(NULL, "invalid $mode parameters %d", (int) serv_mode);
         RETURN_FALSE;
     }
+#else
+    serv_mode = Server::MODE_THREAD;
+#endif
 
     serv = new Server((enum Server::Mode) serv_mode);
     serv->private_data_2 = sw_zval_dup(zserv);
@@ -2577,6 +2580,10 @@ static PHP_METHOD(swoole_server, start) {
     server_object->register_callback();
     server_object->on_before_start();
 
+#ifdef SW_THREAD
+    serv->worker_num = 1;
+#endif
+
     if (serv->start() < 0) {
         php_swoole_fatal_error(E_ERROR, "failed to start server. Error: %s", sw_error);
     }

include/swoole.h

@@ -177,6 +177,14 @@ typedef unsigned long ulong_t;
 #endif
 #define SW_START_SLEEP usleep(100000)  // sleep 1s,wait fork and pthread_create
 
+#ifdef SW_THREAD
+#define SW_THREAD_LOCAL thread_local
+#include "swoole_lock.h"
+extern swoole::Mutex thread_lock;
+#else
+#define SW_THREAD_LOCAL
+#endif
+
 /*-----------------------------------Memory------------------------------------*/
 void *sw_malloc(size_t size);
 void sw_free(void *ptr);

include/swoole_coroutine.h

@@ -41,12 +41,6 @@ typedef std::chrono::microseconds seconds_type;
 #define CALC_EXECUTE_USEC(yield_coroutine, resume_coroutine)
 #endif
 
-#ifdef SW_THREAD
-#define SW_THREAD_LOCAL thread_local
-#else
-#define SW_THREAD_LOCAL
-#endif
-
 namespace swoole {
 class Coroutine {
   public:

include/swoole_process_pool.h

@@ -325,5 +325,5 @@ static sw_inline int swoole_kill(pid_t __pid, int __sig) {
     return kill(__pid, __sig);
 }
 
-extern swoole::WorkerGlobal SwooleWG;  // Worker Global Variable
+extern SW_THREAD_LOCAL swoole::WorkerGlobal SwooleWG;  // Worker Global Variable
 typedef swoole::ProtocolType swProtocolType;

include/swoole_server.h

@@ -27,6 +27,9 @@
 #include "swoole_pipe.h"
 #include "swoole_channel.h"
 #include "swoole_message_bus.h"
+#ifdef SW_THREAD
+#include "swoole_lock.h"
+#endif
 
 #ifdef SW_USE_OPENSSL
 #include "swoole_dtls.h"
@@ -488,6 +491,9 @@ class Server {
     enum Mode {
         MODE_BASE = 1,
         MODE_PROCESS = 2,
+#ifdef SW_THREAD
+        MODE_THREAD = 3,
+#endif
     };
 
     enum TaskIpcMode {
@@ -1000,9 +1006,19 @@ class Server {
     }
 
     bool is_base_mode() {
+#ifndef SW_THREAD
         return mode_ == MODE_BASE;
+#else
+        return mode_ == MODE_BASE || mode_ == MODE_THREAD;
+#endif
     }
 
+#ifdef SW_THREAD
+    bool is_thread_mode() {
+        return mode_ == MODE_THREAD;
+    }
+#endif
+
     bool is_enable_coroutine() {
         if (is_task_worker()) {
             return task_enable_coroutine;
@@ -1451,7 +1467,7 @@ typedef swoole::Server swServer;
 typedef swoole::ListenPort swListenPort;
 typedef swoole::RecvData swRecvData;
 
-extern swoole::Server *g_server_instance;
+extern SW_THREAD_LOCAL swoole::Server *g_server_instance;
 
 static inline swoole::Server *sw_server() {
     return g_server_instance;

src/core/base.cc

@@ -100,7 +100,7 @@ swoole::Global SwooleG = {};
 __thread swoole::ThreadGlobal SwooleTG = {};
 
 static std::unordered_map<std::string, void *> functions;
-static swoole::Logger *g_logger_instance = nullptr;
+static SW_THREAD_LOCAL swoole::Logger *g_logger_instance = nullptr;
 
 #ifdef __MACH__
 static __thread char _sw_error_buf[SW_ERROR_MSG_SIZE];
@@ -152,7 +152,20 @@ static void bug_report_message_init() {
 }
 
 void swoole_init(void) {
+#ifdef SW_THREAD
+    thread_lock.lock();
+#endif
     if (SwooleG.init) {
+#ifdef SW_THREAD
+        thread_lock.unlock();
+        SwooleTG.buffer_stack = new swoole::String(SW_STACK_BUFFER_SIZE);
+        g_logger_instance = new swoole::Logger;
+#ifdef SW_DEBUG
+        sw_logger()->set_level(0);
+#else
+        sw_logger()->set_level(SW_LOG_INFO);
+#endif
+#endif
         return;
     }
 
@@ -211,6 +224,10 @@ void swoole_init(void) {
 
     // init bug report message
     bug_report_message_init();
+
+#ifdef SW_THREAD
+    thread_lock.unlock();
+#endif
 }
 
 SW_EXTERN_C_BEGIN

src/server/master.cc

@@ -16,7 +16,9 @@
 
 #include "swoole_server.h"
 #include "swoole_memory.h"
+#ifndef SW_THREAD
 #include "swoole_lock.h"
+#endif
 #include "swoole_util.h"
 
 #include <assert.h>
@@ -25,7 +27,11 @@
 using swoole::network::SendfileTask;
 using swoole::network::Socket;
 
-swoole::Server *g_server_instance = nullptr;
+SW_THREAD_LOCAL swoole::Server *g_server_instance = nullptr;
+#ifdef SW_THREAD
+swoole::Mutex thread_lock(0);
+static std::vector<swListenPort *> listen_ports;
+#endif
 
 namespace swoole {
 
@@ -696,10 +702,10 @@
         file_put_contents(pid_file, sw_tg_buffer()->str, n);
     }
     int ret;
-    if (is_base_mode()) {
-        ret = start_reactor_processes();
-    } else {
+    if (is_process_mode()) {
         ret = start_reactor_threads();
+    } else {
+        ret = start_reactor_processes();
     }
     // failed to start
     if (ret < 0) {
@@ -851,12 +857,12 @@
     }
 
     int retval;
-    if (is_base_mode()) {
-        factory = new BaseFactory(this);
-        retval = create_reactor_processes();
-    } else {
+    if (is_process_mode()) {
         factory = new ProcessFactory(this);
         retval = create_reactor_threads();
+    } else {
+        factory = new BaseFactory(this);
+        retval = create_reactor_processes();
     }
 
 #ifdef HAVE_PTHREAD_BARRIER
@@ -978,7 +984,7 @@
         join_reactor_thread();
     }
 
-	release_pipe_buffers();
+    release_pipe_buffers();
 
     for (auto port : ports) {
         port->close();
@@ -1515,7 +1521,7 @@
                          "sendfile name[%.8s...] length %u is exceed the max name len %u",
                          file,
                          l_file,
-                         (uint32_t)(SW_IPC_BUFFER_SIZE - sizeof(SendfileTask) - 1));
+                         (uint32_t) (SW_IPC_BUFFER_SIZE - sizeof(SendfileTask) - 1));
         return false;
     }
     // string must be zero termination (for `state` system call)
@@ -1769,7 +1775,7 @@
 
 #ifdef SW_USE_OPENSSL
     if (type & SW_SOCK_SSL) {
-        type = (SocketType)(type & (~SW_SOCK_SSL));
+        type = (SocketType) (type & (~SW_SOCK_SSL));
         ls->type = type;
         ls->ssl = 1;
         ls->ssl_context = new SSLContext();
@@ -1793,24 +1799,53 @@
     }
 #endif
 
-    ls->socket = make_socket(
-        ls->type, ls->is_dgram() ? SW_FD_DGRAM_SERVER : SW_FD_STREAM_SERVER, SW_SOCK_CLOEXEC | SW_SOCK_NONBLOCK);
-    if (ls->socket == nullptr) {
-        swoole_set_last_error(errno);
-        return nullptr;
+#ifdef SW_THREAD
+    thread_lock.lock();
+    if (listen_ports.size() > 0) {
+        for (auto listen_port : listen_ports) {
+            if (listen_port->type == type && listen_port->port == port && listen_port->host == host) {
+                ls->socket = listen_port->socket;
+                thread_lock.unlock();
+                break;
+            }
+        }
     }
+
+    if (ls->socket == nullptr) {
+#endif
+        ls->socket = make_socket(
+            ls->type, ls->is_dgram() ? SW_FD_DGRAM_SERVER : SW_FD_STREAM_SERVER, SW_SOCK_CLOEXEC | SW_SOCK_NONBLOCK);
+        if (ls->socket == nullptr) {
+            swoole_set_last_error(errno);
+#ifdef SW_THREAD
+            thread_lock.unlock();
+#endif
+            return nullptr;
+        }
+
 #if defined(SW_SUPPORT_DTLS) && defined(HAVE_KQUEUE)
-    if (ls->is_dtls()) {
-        ls->socket->set_reuse_port();
-    }
+        if (ls->is_dtls()) {
+            ls->socket->set_reuse_port();
+        }
 #endif
 
-    if (ls->socket->bind(ls->host, &ls->port) < 0) {
-        swoole_set_last_error(errno);
-        ls->socket->free();
-        return nullptr;
+        if (ls->socket->bind(ls->host, &ls->port) < 0) {
+            swoole_set_last_error(errno);
+            ls->socket->free();
+#ifdef SW_THREAD
+            thread_lock.unlock();
+#endif
+            return nullptr;
+        }
+
+        ls->socket->info.assign(ls->type, ls->host, ls->port);
+
+#ifdef SW_THREAD
+        listen_ports.push_back(ls);
+        thread_lock.unlock();
     }
-    ls->socket->info.assign(ls->type, ls->host, ls->port);
+#endif
+
     check_port_type(ls);
     ptr.release();
     ports.push_back(ls);
@@ -2016,7 +2051,7 @@
 }
 
 void Server::release_pipe_buffers() {
-	message_bus.free_buffer();
+    message_bus.free_buffer();
 }
 
 int Server::get_idle_worker_num() {

src/server/reactor_process.cc

@@ -30,8 +30,12 @@
 #endif
 
 static bool Server_is_single(Server *serv) {
-    return serv->worker_num == 1 && serv->task_worker_num == 0 && serv->max_request == 0 &&
-           serv->user_worker_list.empty();
+    return
+#ifdef SW_THREAD
+        serv->is_thread_mode() ||
+#endif
+        (serv->worker_num == 1 && serv->task_worker_num == 0 && serv->max_request == 0 &&
+         serv->user_worker_list.empty());
 }
 
 int Server::create_reactor_processes() {
@@ -48,32 +52,48 @@
     sw_free(connection_list);
 }
 
+#ifdef SW_THREAD
+static bool one_off_task = false;
+#endif
+
 int Server::start_reactor_processes() {
     single_thread = 1;
 
     // listen TCP
     if (have_stream_sock == 1) {
-        for (auto ls : ports) {
-            if (ls->is_dgram()) {
-                continue;
-            }
-#ifdef HAVE_REUSEPORT
-            if (enable_reuse_port) {
-                if (::close(ls->socket->fd) < 0) {
-                    swoole_sys_warning("close(%d) failed", ls->socket->fd);
+#ifdef SW_THREAD
+        thread_lock.lock();
+        if (!one_off_task) {
+#endif
+            for (auto ls : ports) {
+                if (ls->is_dgram()) {
+                    continue;
                 }
-                delete ls->socket;
-                ls->socket = nullptr;
-                continue;
-            } else
+#ifdef HAVE_REUSEPORT
+                if (enable_reuse_port) {
+                    if (::close(ls->socket->fd) < 0) {
+                        swoole_sys_warning("close(%d) failed", ls->socket->fd);
+                    }
+                    delete ls->socket;
+                    ls->socket = nullptr;
+                    continue;
+                } else
 #endif
-            {
-                // listen server socket
-                if (ls->listen() < 0) {
-                    return SW_ERR;
+                {
+                    // listen server socket
+                    if (ls->listen() < 0) {
+#ifdef SW_THREAD
+                        thread_lock.unlock();
+#endif
+                        return SW_ERR;
+                    }
                 }
             }
+#ifdef SW_THREAD
+            one_off_task = true;
         }
+        thread_lock.unlock();
+#endif
     }
 
     ProcessPool *pool = &gs->event_workers;