Optimize code, fix tests. Improve the log component to resolve data consistency issues in multithreaded mode. (#5754)

Author: matyhtf

core-tests/src/memory/global_memory.cpp

@@ -45,6 +45,5 @@ TEST(global_memory, alloc) {
     ASSERT_STREQ(ptr2, "hello, world, #2");
     ASSERT_STREQ(ptr3, "hello, world, #3");
 
-    pool->destroy();
     delete pool;
 }

core-tests/src/network/stream.cpp

@@ -83,8 +83,6 @@ TEST(stream, send) {
         swoole_event_wait();
 
         kill(getpid(), SIGTERM);
-
-        swoole_signal_unblock_all();
     });
 
     serv.onWorkerStart = [&lock](Server *serv, Worker *worker) { lock.unlock(); };

core-tests/src/os/process_pool.cpp

@@ -277,7 +277,6 @@ TEST(process_pool, listen) {
         c.close();
 
         kill(getpid(), SIGTERM);
-        swoole_signal_unblock_all();
     });
 
     ASSERT_EQ(pool.wait(), SW_OK);

core-tests/src/os/signal.cpp

@@ -1,4 +1,5 @@
 #include "test_core.h"
+#include "swoole_process_pool.h"
 #include "swoole_signal.h"
 
 #ifdef HAVE_SIGNALFD
@@ -34,3 +35,33 @@ TEST(os_signal, signalfd) {
     swoole_event_wait();
 }
 #endif
+
+TEST(os_signal, block) {
+    ASSERT_EQ(swoole::test::spawn_exec_and_wait([]() {
+                  sysv_signal(SIGIO, [](int signo) { exit(255); });
+
+                  std::thread t([] {
+                      swoole_signal_block_all();
+                      pthread_kill(pthread_self(), SIGIO);
+                  });
+                  t.join();
+              }),
+              0);
+}
+
+TEST(os_signal, unblock) {
+    auto status = swoole::test::spawn_exec_and_wait([]() {
+        sysv_signal(SIGIO, [](int signo) { exit(255); });
+
+        std::thread t([] {
+            swoole_signal_block_all();
+            pthread_kill(pthread_self(), SIGIO);
+            swoole_signal_unblock_all();
+        });
+        t.join();
+    });
+
+    auto exit_status = swoole::ExitStatus(getpid(), status);
+
+    ASSERT_EQ(exit_status.get_code(), 255);
+}

ext-src/swoole_coroutine.cc

@@ -1116,19 +1116,20 @@ PHP_FUNCTION(swoole_coroutine_defer) {
 
 static PHP_METHOD(swoole_coroutine, stats) {
     array_init(return_value);
+
     add_assoc_long_ex(return_value, ZEND_STRL("event_num"), sw_reactor() ? sw_reactor()->get_event_num() : 0);
-    add_assoc_long_ex(
-        return_value, ZEND_STRL("signal_listener_num"), SwooleTG.signal_listener_num + SwooleTG.co_signal_listener_num);
+    add_assoc_long_ex(return_value, ZEND_STRL("signal_listener_num"), swoole_signal_get_listener_num());
 
-    if (SwooleTG.async_threads) {
-        add_assoc_long_ex(return_value, ZEND_STRL("aio_task_num"), SwooleTG.async_threads->get_task_num());
-        add_assoc_long_ex(return_value, ZEND_STRL("aio_worker_num"), SwooleTG.async_threads->get_worker_num());
-        add_assoc_long_ex(return_value, ZEND_STRL("aio_queue_size"), SwooleTG.async_threads->get_queue_size());
+    if (sw_async_threads()) {
+        add_assoc_long_ex(return_value, ZEND_STRL("aio_task_num"), sw_async_threads()->get_task_num());
+        add_assoc_long_ex(return_value, ZEND_STRL("aio_worker_num"), sw_async_threads()->get_worker_num());
+        add_assoc_long_ex(return_value, ZEND_STRL("aio_queue_size"), sw_async_threads()->get_queue_size());
     } else {
         add_assoc_long_ex(return_value, ZEND_STRL("aio_task_num"), 0);
         add_assoc_long_ex(return_value, ZEND_STRL("aio_worker_num"), 0);
         add_assoc_long_ex(return_value, ZEND_STRL("aio_queue_size"), 0);
     }
+
     add_assoc_long_ex(return_value, ZEND_STRL("c_stack_size"), Coroutine::get_stack_size());
     add_assoc_long_ex(return_value, ZEND_STRL("coroutine_num"), Coroutine::count());
     add_assoc_long_ex(return_value, ZEND_STRL("coroutine_peak_num"), Coroutine::get_peak_num());

ext-src/swoole_process.cc

@@ -486,7 +486,7 @@ static PHP_METHOD(swoole_process, signal) {
 #endif
             signal_fci_caches[signo] = nullptr;
             swoole_event_defer(sw_callable_free, fci_cache);
-            SwooleTG.signal_listener_num--;
+            SwooleG.signal_listener_num--;
             RETURN_TRUE;
         } else {
             php_swoole_error(E_WARNING, "unable to find the callback of signal [" ZEND_LONG_FMT "]", signo);
@@ -506,7 +506,7 @@ static PHP_METHOD(swoole_process, signal) {
         if (signal_fci_caches[signo]) {
             sw_callable_free(signal_fci_caches[signo]);
         } else {
-            SwooleTG.signal_listener_num++;
+            SwooleG.signal_listener_num++;
         }
         signal_fci_caches[signo] = fci_cache;
 #ifdef SW_USE_THREAD_CONTEXT
@@ -521,15 +521,15 @@ static PHP_METHOD(swoole_process, signal) {
     if (!SwooleTG.reactor->isset_exit_condition(Reactor::EXIT_CONDITION_SIGNAL_LISTENER)) {
         SwooleTG.reactor->set_exit_condition(Reactor::EXIT_CONDITION_SIGNAL_LISTENER,
                                              [](Reactor *reactor, size_t &event_num) -> bool {
-                                                 return SwooleTG.signal_listener_num == 0 or !SwooleG.wait_signal;
+                                                 return SwooleG.signal_listener_num == 0 or !SwooleG.wait_signal;
                                              });
     }
 
     if (signal_fci_caches[signo]) {
         // free the old fci_cache
         swoole_event_defer(sw_callable_free, signal_fci_caches[signo]);
     } else {
-        SwooleTG.signal_listener_num++;
+        SwooleG.signal_listener_num++;
     }
     signal_fci_caches[signo] = fci_cache;
 

ext-src/swoole_timer.cc

@@ -183,6 +183,10 @@ static void timer_add(INTERNAL_FUNCTION_PARAMETERS, bool persistent) {
         php_swoole_check_reactor();
     }
 
+    /**
+     * In certain systems, such as macOS, zend_long is the long long type,
+     * and it must be explicitly converted to long.
+     */
     tnode = swoole_timer_add((long) ms, persistent, timer_callback, fci);
     if (UNEXPECTED(!tnode)) {
         php_swoole_fatal_error(E_WARNING, "add timer failed");

include/swoole.h

@@ -219,6 +219,7 @@ struct TimerNode;
 struct Event;
 class Pipe;
 class MessageBus;
+class Server;
 namespace network {
 struct Socket;
 struct Address;
@@ -690,23 +691,21 @@ struct RecvData {
 struct ThreadGlobal {
     uint16_t id;
     uint8_t type;
+    int32_t error;
 #ifdef SW_THREAD
     uint8_t process_type;
     uint32_t process_id;
 #endif
     String *buffer_stack;
     Reactor *reactor;
-    Logger *logger;
     Timer *timer;
     TimerScheduler timer_scheduler;
     MessageBus *message_bus;
     AsyncThreads *async_threads;
 #ifdef SW_USE_IOURING
     Iouring *iouring;
 #endif
-    uint32_t signal_listener_num;
-    uint32_t co_signal_listener_num;
-    int error;
+    bool signal_blocking_all;
 };
 
 struct Allocator {
@@ -762,6 +761,8 @@ struct Global {
     int signal_fd;
     bool signal_alarm;
     bool signal_dispatch;
+    uint32_t signal_listener_num;
+    uint32_t signal_async_listener_num;
 
     long trace_flags;
 
@@ -776,15 +777,18 @@ struct Global {
     MemoryPool *memory_pool;
     Allocator std_allocator;
     std::string task_tmpfile;
-    //-----------------------[DNS]--------------------------
+    //------------------[Single Instance]----------------------
+    Logger *logger;
+    Server *server;
+    //-----------------------[DNS]-----------------------------
     std::string dns_server_host;
     int dns_server_port;
     double dns_cache_refresh_time;
     int dns_tries;
     std::string dns_resolvconf_path;
     std::string dns_hosts_path;
     std::list<NameResolver> name_resolvers;
-    //-----------------------[AIO]--------------------------
+    //-----------------------[AIO]----------------------------
     uint32_t aio_core_worker_num;
     uint32_t aio_worker_num;
 #ifdef SW_USE_IOURING

include/swoole_async.h

@@ -136,3 +136,5 @@ void handler_getaddrinfo(AsyncEvent *event);
 
 }  // namespace async
 };  // namespace swoole
+
+swoole::AsyncThreads *sw_async_threads();

include/swoole_log.h

@@ -58,13 +58,20 @@ class Logger {
     std::string date_format = SW_LOG_DEFAULT_DATE_FORMAT;
     std::string log_file = "";
     std::string log_real_file;
+    std::mutex lock;
     int log_rotation = SW_LOG_ROTATION_SINGLE;
 
+    void reopen_without_lock();
+
   public:
     bool open(const char *logfile);
+    void reopen();
     void set_stream(FILE *stream);
+    /**
+     * Only the `put` and `reopen` functions are thread-safe,
+     * other functions must be used in a single-threaded environment.
+     */
     void put(int level, const char *content, size_t length);
-    void reopen();
     void close(void);
     void reset();
     void set_level(int lv);