Fix core tests & optimize code (#5800)

matyhtf · web-flow · commit 5b9d1602114b · 2025-06-11T19:59:45.000+08:00
diff --git a/.github/workflows/core.yml b/.github/workflows/core.yml
@@ -47,7 +47,7 @@ jobs:
 
       - name: build
         run: |
-          cmake . -D CODE_COVERAGE=ON -D enable_thread=1 -D verbose=1 -D enable_asan=1 || exit 1
+          cmake . -D CODE_COVERAGE=ON -D enable_thread=1 -D verbose=1 || exit 1
           make VERBOSE=1 -j $(nproc) core-tests || exit 1
 
       - name: run tests
diff --git a/.github/workflows/unit.yml b/.github/workflows/unit.yml
@@ -11,7 +11,7 @@ jobs:
       matrix:
         php: ['8.1', '8.2', '8.3', '8.4']
         os: [ ubuntu-24.04, ubuntu-24.04-arm ]
-    name: ${{ matrix.php }}-${{ matrix.os }}-test
+    name: ${{ matrix.php }}-${{ matrix.os }}
     runs-on: ${{ matrix.os }}
     steps:
     - uses: actions/checkout@v4
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -120,6 +120,7 @@ endif()
 if (DEFINED enable_asan)
 	set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=address -fno-omit-frame-pointer")
     set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -fsanitize=address -fno-omit-frame-pointer")
+    add_definitions(-DSW_USE_ASAN)
 endif()
 
 if (DEFINED enable_thread)
diff --git a/core-tests/src/os/timer.cpp b/core-tests/src/os/timer.cpp
@@ -148,7 +148,7 @@ TEST(timer, error) {
     SwooleTG.timer = nullptr;
 
     swoole_timer_free();
-    ASSERT_EQ(swoole_timer_select(), SW_ERR);
+    swoole_timer_select(); // no effect
     ASSERT_EQ(swoole_timer_get(1), nullptr);
     ASSERT_FALSE(swoole_timer_clear(1));
     ASSERT_FALSE(swoole_timer_exists(1));
diff --git a/core-tests/src/reactor/base.cpp b/core-tests/src/reactor/base.cpp
@@ -527,6 +527,7 @@ static void test_error_event(Reactor::Type type, int retval) {
     ASSERT_TRUE(p.ready());
 
     Reactor *reactor = new Reactor(1024, type);
+    SwooleTG.reactor = reactor;
 
     reactor->ptr = &retval;
 
@@ -542,6 +543,7 @@ static void test_error_event(Reactor::Type type, int retval) {
     p.close(SW_PIPE_CLOSE_WORKER);
     ASSERT_EQ(reactor->wait(), SW_OK);
     delete reactor;
+    SwooleTG.reactor = nullptr;
 }
 
 TEST(reactor, error_event) {
diff --git a/core-tests/src/server/server.cpp b/core-tests/src/server/server.cpp
@@ -697,6 +697,7 @@ TEST(server, thread) {
     ASSERT_EQ(test::counter_get(4), 2);  // onPipeMessage called
 }
 
+#ifndef SW_USE_ASAN
 TEST(server, task_thread) {
     DEBUG() << "new server\n";
     Server serv(Server::MODE_THREAD);
@@ -972,6 +973,7 @@ TEST(server, reload_thread_3) {
     test::wait_all_child_processes();
 }
 #endif
+#endif
 
 TEST(server, reload_all_workers) {
     Server serv(Server::MODE_PROCESS);
@@ -1027,6 +1029,8 @@ TEST(server, reload_all_workers) {
                 kill(serv->gs->master_pid, SIGTERM);
             }
         }
+
+        DEBUG() << "onWorkerStart: id=" << worker->id << "\n";
     };
 
     ASSERT_EQ(serv.start(), 0);
@@ -1075,6 +1079,8 @@ TEST(server, reload_all_workers2) {
                 kill(serv->gs->master_pid, SIGTERM);
             }
         }
+
+        DEBUG() << "onWorkerStart: id=" << worker->id << "\n";
     };
 
     serv.onBeforeReload = [](Server *serv) {
@@ -2126,12 +2132,14 @@ static void test_task_ipc(Server &serv) {
     serv.onTask = [](Server *serv, EventData *task) -> int {
         EXPECT_EQ(string(task->data, task->info.len), string(packet));
         EXPECT_TRUE(serv->finish(task->data, task->info.len, 0, task));
+        DEBUG() << "onTask: " << task->info.len << " bytes\n";
         return 0;
     };
 
     serv.onFinish = [](Server *serv, EventData *task) -> int {
         EXPECT_EQ(string(task->data, task->info.len), string(packet));
         usleep(100000);
+        DEBUG() << "onFinish: " << task->info.len << " bytes\n";
         serv->shutdown();
         return 0;
     };
@@ -2146,6 +2154,12 @@ static void test_task_ipc(Server &serv) {
             buf.info.ext_flags |= (SW_TASK_NONBLOCK | SW_TASK_CALLBACK);
             EXPECT_TRUE(serv->task(&buf, &_dst_worker_id));
         }
+
+        DEBUG() << "onWorkerStart: id=" << worker->id << "\n";
+    };
+
+    serv.onBeforeShutdown = [](Server *serv) {
+        DEBUG() << "onBeforeShutdown\n";
     };
 
     ASSERT_EQ(serv.start(), 0);
diff --git a/include/swoole.h b/include/swoole.h
@@ -602,11 +602,7 @@ int swoole_set_cpu_affinity(cpu_set_t *set);
 int swoole_get_cpu_affinity(cpu_set_t *set);
 #endif
 
-void swoole_clock_realtime(timespec *time);
-timespec swoole_time_until(time_t milliseconds);
-
 namespace swoole {
-
 typedef long SessionId;
 typedef long TaskId;
 typedef uint8_t ReactorId;
@@ -781,6 +777,8 @@ std::string dirname(const std::string &file);
 int hook_add(void **hooks, int type, const Callback &func, int push_back);
 void hook_call(void **hooks, int type, void *arg);
 double microtime();
+void realtime_get(timespec *time);
+void realtime_add(timespec *time, int64_t add_msec);
 }  // namespace swoole
 
 extern swoole::Global SwooleG;                      // Local Global Variable
diff --git a/include/swoole_api.h b/include/swoole_api.h
@@ -47,7 +47,7 @@ SW_API void swoole_timer_delay(swoole::TimerNode *tnode, long delay_ms);
 SW_API swoole::TimerNode *swoole_timer_get(long timer_id);
 SW_API bool swoole_timer_clear(long timer_id);
 SW_API void swoole_timer_free();
-SW_API int swoole_timer_select();
+SW_API void swoole_timer_select();
 SW_API int64_t swoole_timer_get_next_msec();
 SW_API bool swoole_timer_is_available();
 
diff --git a/include/swoole_process_pool.h b/include/swoole_process_pool.h
@@ -437,7 +437,8 @@ struct ProcessPool {
     static int run_with_stream_protocol(ProcessPool *pool, Worker *worker);
     static int run_with_message_protocol(ProcessPool *pool, Worker *worker);
     static int run_async(ProcessPool *pool, Worker *worker);
-    static void create_sync_timer();
+    void at_worker_enter(Worker *worker);
+    void at_worker_exit(Worker *worker);
 
     bool wait_detached_worker(std::unordered_set<pid_t> &detached_workers, pid_t pid);
 };
diff --git a/include/swoole_timer.h b/include/swoole_timer.h
@@ -80,7 +80,7 @@ class Timer {
     ~Timer();
 
     int64_t get_relative_msec() const {
-        return time<std::chrono::milliseconds>(true) - base_time;
+        return get_absolute_msec() - base_time;
     }
 
     int64_t get_next_msec() const {
@@ -106,7 +106,7 @@ class Timer {
         update(tnode);
     }
     void reinit(bool manually_trigger = false);
-    int select();
+    void select();
 
     TimerNode *get(long id) {
         auto it = map.find(id);
diff --git a/src/core/base.cc b/src/core/base.cc
@@ -1007,11 +1007,9 @@ std::string intersection(const std::vector<std::string> &vec1, std::set<std::str
     return "";
 }
 
-double microtime(void) {
-    struct timeval t;
+double microtime() {
+    timeval t;
     gettimeofday(&t, nullptr);
     return (double) t.tv_sec + ((double) t.tv_usec / 1000000);
 }
-
-//-------------------------------------------------------------------------------
 };  // namespace swoole
diff --git a/src/core/timer.cc b/src/core/timer.cc
@@ -89,16 +89,11 @@ TimerNode *Timer::add(long _msec, bool persistent, void *data, const TimerCallba
         return nullptr;
     }
 
-    int64_t now_msec = get_relative_msec();
-    if (sw_unlikely(now_msec < 0)) {
-        return nullptr;
-    }
-
     auto *tnode = new TimerNode();
     tnode->id = _next_id++;
     tnode->data = data;
     tnode->type = TimerNode::TYPE_KERNEL;
-    tnode->exec_msec = now_msec + _msec;
+    tnode->exec_msec = get_relative_msec() + _msec;
     tnode->interval = persistent ? _msec : 0;
     tnode->removed = false;
     tnode->callback = callback;
@@ -156,19 +151,15 @@ bool Timer::remove(TimerNode *tnode) {
     return true;
 }
 
-int Timer::select() {
-    int64_t now_msec = get_relative_msec();
-    if (sw_unlikely(now_msec < 0)) {
-        return SW_ERR;
-    }
-
+void Timer::select() {
+    const int64_t now_msec = get_relative_msec();
     TimerNode *tnode = nullptr;
     HeapNode *tmp;
 
     swoole_trace_log(SW_TRACE_TIMER, "select begin: now_msec=%" PRId64 ", round=%" PRId64, now_msec, round);
 
     while ((tmp = heap.top())) {
-        tnode = (TimerNode *) tmp->data;
+        tnode = static_cast<TimerNode *>(tmp->data);
         if (tnode->exec_msec > now_msec || tnode->round == round) {
             break;
         }
@@ -212,7 +203,5 @@ int Timer::select() {
         set(this, next_msec_);
     }
     round++;
-
-    return SW_OK;
 }
 };  // namespace swoole
diff --git a/src/coroutine/base.cc b/src/coroutine/base.cc
@@ -310,26 +310,3 @@ swoole::Coroutine *swoole_coroutine_iterator_each() {
     ++gdb_iterator_;
     return co;
 }
-
-void swoole_clock_realtime(timespec *time) {
-    auto now = std::chrono::system_clock::now();
-    auto ns = std::chrono::duration_cast<std::chrono::nanoseconds>(now.time_since_epoch());
-    time->tv_sec = ns.count() / SW_NUM_BILLION;
-    time->tv_nsec = ns.count() % SW_NUM_BILLION;
-}
-
-timespec swoole_time_until(time_t milliseconds) {
-    timespec t;
-    swoole_clock_realtime(&t);
-
-    const time_t sec_increment = milliseconds / 1000;
-    const time_t nsec_increment = (milliseconds % 1000) * SW_NUM_MILLION;
-
-    t.tv_sec += sec_increment;
-    t.tv_nsec += nsec_increment;
-
-    t.tv_sec += t.tv_nsec / SW_NUM_BILLION;
-    t.tv_nsec %= SW_NUM_BILLION;
-
-    return t;
-}
diff --git a/src/lock/mutex.cc b/src/lock/mutex.cc
@@ -92,7 +92,9 @@ int Mutex::trylock_rd() {
 
 #ifdef HAVE_MUTEX_TIMEDLOCK
 int Mutex::lock_wait(int timeout_msec) {
-    timespec timeo = swoole_time_until(timeout_msec);
+    timespec timeo;
+    realtime_get(&timeo);
+    realtime_add(&timeo, timeout_msec);
     return pthread_mutex_timedlock(&impl->lock_, &timeo);
 }
 #else
diff --git a/src/network/socket.cc b/src/network/socket.cc
@@ -457,7 +457,9 @@ void Socket::free() {
     delete out_buffer;
 
     if (swoole_event_is_available()) {
-        removed = 1;
+        if (!removed) {
+            swoole_event_del(this);
+        }
         swoole_event_defer(socket_free_defer, this);
     } else {
         socket_free_defer(this);
diff --git a/src/os/process_pool.cc b/src/os/process_pool.cc
@@ -524,11 +524,22 @@ bool ProcessPool::is_worker_running(Worker *worker) {
     return running && !worker->is_shutdown() && !worker->has_exceeded_max_request();
 }
 
-void ProcessPool::create_sync_timer() {
-    if (SwooleTG.timer) {
-        SwooleTG.timer->reinit(true);
-    } else {
-        swoole_timer_create(true);
+void ProcessPool::at_worker_enter(Worker *worker) {
+    if (worker->pipe_worker) {
+        worker->pipe_worker->dont_restart = 1;
+    }
+    if (ipc_mode == SW_IPC_UNIXSOCK) {
+        if (swoole_timer_is_available()) {
+            sw_timer()->reinit(true);
+        } else {
+            swoole_timer_create(true);
+        }
+    }
+}
+
+void ProcessPool::at_worker_exit(Worker *worker) {
+    if (swoole_timer_is_available()) {
+        swoole_timer_free();
     }
 }
 
@@ -540,9 +551,6 @@ int ProcessPool::run_with_task_protocol(ProcessPool *pool, Worker *worker) {
 
     ssize_t n = 0;
 
-    /**
-     * Use from_fd save the task_worker->id
-     */
     out.buf.info.server_fd = worker->id;
 
     if (pool->schedule_by_sysvmsg) {
@@ -551,11 +559,7 @@ int ProcessPool::run_with_task_protocol(ProcessPool *pool, Worker *worker) {
         out.mtype = worker->id + 1;
     }
 
-    if (pool->ipc_mode == SW_IPC_UNIXSOCK) {
-        create_sync_timer();
-        worker->pipe_worker->dont_restart = 1;
-    }
-
+    pool->at_worker_enter(worker);
     while (pool->is_worker_running(worker)) {
         /**
          * fetch task
@@ -614,6 +618,7 @@ int ProcessPool::run_with_task_protocol(ProcessPool *pool, Worker *worker) {
     _end:
         worker_end_callback();
     }
+    pool->at_worker_exit(worker);
 
     return SW_OK;
 }
@@ -677,11 +682,7 @@ int ProcessPool::run_with_stream_protocol(ProcessPool *pool, Worker *worker) {
     auto *outbuf = reinterpret_cast<QueueNode *>(pool->packet_buffer);
     outbuf->mtype = 0;
 
-    if (pool->ipc_mode == SW_IPC_UNIXSOCK) {
-        create_sync_timer();
-        worker->pipe_worker->dont_restart = 1;
-    }
-
+    pool->at_worker_enter(worker);
     while (pool->is_worker_running(worker)) {
         /**
          * fetch task
@@ -763,6 +764,7 @@ int ProcessPool::run_with_stream_protocol(ProcessPool *pool, Worker *worker) {
     _end:
         worker_end_callback();
     }
+    pool->at_worker_exit(worker);
 
     return SW_OK;
 }
@@ -796,9 +798,7 @@ int ProcessPool::run_with_message_protocol(ProcessPool *pool, Worker *worker) {
         pool->create_message_bus();
     }
 
-    create_sync_timer();
-    worker->pipe_worker->dont_restart = 1;
-
+    pool->at_worker_enter(worker);
     while (pool->is_worker_running(worker)) {
         switch (fn()) {
         case 0:
@@ -813,6 +813,7 @@ int ProcessPool::run_with_message_protocol(ProcessPool *pool, Worker *worker) {
             break;
         }
     }
+    pool->at_worker_exit(worker);
 
     return SW_OK;
 }
diff --git a/src/os/timer.cc b/src/os/timer.cc
diff --git a/src/reactor/base.cc b/src/reactor/base.cc
diff --git a/src/wrapper/timer.cc b/src/wrapper/timer.cc

Original file line number	Diff line number	Diff line change
`@@ -80,7 +80,7 @@ class Timer {`
`80`	`80`	`~Timer();`
`81`	`81`
`82`	`82`	`int64_t get_relative_msec() const {`
`83`		`- return time<std::chrono::milliseconds>(true) - base_time;`
	`83`	`+ return get_absolute_msec() - base_time;`
`84`	`84`	`}`
`85`	`85`
`86`	`86`	`int64_t get_next_msec() const {`
`@@ -106,7 +106,7 @@ class Timer {`
`106`	`106`	`update(tnode);`
`107`	`107`	`}`
`108`	`108`	`void reinit(bool manually_trigger = false);`
`109`		`- int select();`
	`109`	`+ void select();`
`110`	`110`
`111`	`111`	`TimerNode *get(long id) {`
`112`	`112`	`auto it = map.find(id);`