refactor thread barrier

Author: matyhtf

include/swoole.h

@@ -176,7 +176,7 @@ typedef unsigned long ulong_t;
 #define SW_ASSERT(e)
 #define SW_ASSERT_1BYTE(v)
 #endif
-#define SW_START_SLEEP usleep(100000)  // sleep 1s,wait fork and pthread_create
+#define SW_START_SLEEP usleep(100000)  // sleep 0.1s, wait fork and pthread_create
 
 #ifdef SW_THREAD
 #define SW_THREAD_LOCAL thread_local

include/swoole_lock.h

@@ -105,4 +105,23 @@ class SpinLock : public Lock {
     int trylock() override;
 };
 #endif
+
+#if defined(HAVE_PTHREAD_BARRIER) && !(defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__))
+#define SW_USE_PTHREAD_BARRIER
+#endif
+
+struct Barrier {
+#ifdef SW_USE_PTHREAD_BARRIER
+    pthread_barrier_t barrier_;
+    pthread_barrierattr_t barrier_attr_;
+    bool shared_;
+#else
+    sw_atomic_t count_;
+    sw_atomic_t barrier_;
+#endif
+    void init(bool shared, int count);
+    void wait();
+    void destroy();
+};
+
 }  // namespace swoole

include/swoole_server.h

@@ -394,10 +394,7 @@ struct ServerGS {
 
     sw_atomic_t spinlock;
 
-#ifdef HAVE_PTHREAD_BARRIER
-    pthread_barrier_t manager_barrier;
-    pthread_barrierattr_t manager_barrier_attr;
-#endif
+    Barrier manager_barrier;
 
     ProcessPool task_workers;
     ProcessPool event_workers;
@@ -858,9 +855,7 @@ class Server {
     std::shared_ptr<std::vector<std::string>> http_index_files = nullptr;
     std::shared_ptr<std::unordered_set<std::string>> http_compression_types = nullptr;
 
-#ifdef HAVE_PTHREAD_BARRIER
-    pthread_barrier_t reactor_thread_barrier = {};
-#endif
+    Barrier reactor_thread_barrier = {};
 
     /**
      * temporary directory for HTTP uploaded file.

src/lock/barrier.cc

@@ -0,0 +1,63 @@
+/*
+  +----------------------------------------------------------------------+
+  | Swoole                                                               |
+  +----------------------------------------------------------------------+
+  | This source file is subject to version 2.0 of the Apache license,    |
+  | that is bundled with this package in the file LICENSE, and is        |
+  | available through the world-wide-web at the following url:           |
+  | http://www.apache.org/licenses/LICENSE-2.0.html                      |
+  | If you did not receive a copy of the Apache2.0 license and are unable|
+  | to obtain it through the world-wide-web, please send a note to       |
+  | [email protected] so we can mail you a copy immediately.            |
+  +----------------------------------------------------------------------+
+  | Author: Tianfeng Han  <[email protected]>                             |
+  +----------------------------------------------------------------------+
+*/
+
+#include "swoole.h"
+#include "swoole_lock.h"
+
+namespace swoole {
+
+#define BARRIER_USEC 10000
+
+void Barrier::init(bool shared, int count) {
+#ifdef SW_USE_PTHREAD_BARRIER
+    if (shared) {
+        pthread_barrierattr_setpshared(&barrier_attr_, PTHREAD_PROCESS_SHARED);
+        pthread_barrier_init(&barrier_, &barrier_attr_, count);
+    } else {
+        pthread_barrier_init(&barrier_, nullptr, count);
+    }
+    shared_ = shared;
+#else
+    barrier_ = 0;
+    count_ = count;
+#endif
+}
+
+void Barrier::wait() {
+#ifdef SW_USE_PTHREAD_BARRIER
+    pthread_barrier_wait(&barrier_);
+#else
+    sw_atomic_add_fetch(&barrier_, 1);
+    SW_LOOP {
+        if (*barrier_ == count_) {
+            break;
+        }
+        usleep(BARRIER_USEC);
+        sw_atomic_memory_barrier();
+    }
+#endif
+}
+
+void Barrier::destroy() {
+#ifdef SW_USE_PTHREAD_BARRIER
+    pthread_barrier_destroy(&barrier_);
+    if (shared_) {
+        pthread_barrierattr_destroy(&barrier_attr_);
+    }
+#endif
+}
+
+};  // namespace swoole

src/server/manager.cc

@@ -192,12 +192,7 @@ void Manager::wait(Server *_server) {
         int sigid = SIGTERM;
         procctl(P_PID, 0, PROC_PDEATHSIG_CTL, &sigid);
 #endif
-
-#if defined(HAVE_PTHREAD_BARRIER) && !(defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__))
-        pthread_barrier_wait(&_server->gs->manager_barrier);
-#else
-        SW_START_SLEEP;
-#endif
+        _server->gs->manager_barrier.wait();
     }
 
     if (_server->isset_hook(Server::HOOK_MANAGER_START)) {

src/server/master.cc

@@ -434,21 +434,12 @@ int Server::start_master_thread(Reactor *reactor) {
         return SW_ERR;
     }
 
-#ifdef HAVE_PTHREAD_BARRIER
     if (!single_thread) {
-        pthread_barrier_wait(&reactor_thread_barrier);
+        reactor_thread_barrier.wait();
     }
-#if defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__)
-    SW_START_SLEEP;
-#else
     if (is_process_mode()) {
-        pthread_barrier_wait(&gs->manager_barrier);
+        gs->manager_barrier.wait();
     }
-#endif
-#else
-    SW_START_SLEEP;
-#endif
-
     gs->master_pid = getpid();
 
     if (isset_hook(HOOK_MASTER_START)) {
@@ -835,15 +826,10 @@ int Server::create() {
         return SW_ERR;
     }
 
-#ifdef HAVE_PTHREAD_BARRIER
     if (is_process_mode() || is_thread_mode()) {
-        pthread_barrier_init(&reactor_thread_barrier, nullptr, reactor_num + 1);
-#if !(defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__))
-        pthread_barrierattr_setpshared(&gs->manager_barrier_attr, PTHREAD_PROCESS_SHARED);
-        pthread_barrier_init(&gs->manager_barrier, &gs->manager_barrier_attr, 2);
-#endif
+        reactor_thread_barrier.init(false, reactor_num + 1);
+        gs->manager_barrier.init(true, 2);
     }
-#endif
 
     if (swoole_isset_hook(SW_GLOBAL_HOOK_AFTER_SERVER_CREATE)) {
         swoole_call_hook(SW_GLOBAL_HOOK_AFTER_SERVER_CREATE, this);
@@ -1060,16 +1046,10 @@ void Server::destroy() {
             delete l;
         }
     }
-#ifdef HAVE_PTHREAD_BARRIER
     if (is_process_mode()) {
-        pthread_barrier_destroy(&reactor_thread_barrier);
-#if !(defined(__FreeBSD__) || defined(__OpenBSD__) || defined(__NetBSD__))
-        pthread_barrier_destroy(&gs->manager_barrier);
-        pthread_barrierattr_destroy(&gs->manager_barrier_attr);
-#endif
+        reactor_thread_barrier.destroy();
+        gs->manager_barrier.destroy();
     }
-#endif
-
     if (is_base_mode()) {
         destroy_base_factory();
     } else if (is_thread_mode()) {

src/server/reactor_thread.cc

@@ -826,11 +826,7 @@ void Server::reactor_thread_main_loop(Server *serv, int reactor_id) {
     }
 
     // wait other thread
-#ifdef HAVE_PTHREAD_BARRIER
-    pthread_barrier_wait(&serv->reactor_thread_barrier);
-#else
-    SW_START_SLEEP;
-#endif
+    serv->reactor_thread_barrier.wait();
     // main loop
     swoole_event_wait();
     if (serv->is_thread_mode()) {