Support multiple threads by thread pool. v5.0.32

trunk/src/app/srs_app_threads.cpp

@@ -30,6 +30,20 @@
 #include <string>
 using namespace std;
 
+#include <unistd.h>
+#include <fcntl.h>
+
+#ifdef SRS_OSX
+pid_t gettid() {
+    return 0;
+}
+#else
+#if __GLIBC__ == 2 && __GLIBC_MINOR__ < 30
+        #include <sys/syscall.h>
+        #define gettid() syscall(SYS_gettid)
+    #endif
+#endif
+
 extern ISrsLog* _srs_log;
 extern ISrsContext* _srs_context;
 extern SrsConfig* _srs_config;
@@ -276,7 +290,7 @@ srs_error_t SrsCircuitBreaker::on_timer(srs_utime_t interval)
 SrsCircuitBreaker* _srs_circuit_breaker = NULL;
 SrsAsyncCallWorker* _srs_dvr_async = NULL;
 
-srs_error_t srs_thread_initialize()
+srs_error_t srs_global_initialize()
 {
     srs_error_t err = srs_success;
 
@@ -292,11 +306,6 @@ srs_error_t srs_thread_initialize()
     _srs_pps_cids_get = new SrsPps();
     _srs_pps_cids_set = new SrsPps();
 
-    // Initialize ST, which depends on pps cids.
-    if ((err = srs_st_init()) != srs_success) {
-        return srs_error_wrap(err, "initialize st failed");
-    }
-
     // The global objects which depends on ST.
     _srs_hybrid = new SrsHybridServer();
     _srs_sources = new SrsLiveSourceManager();
@@ -465,3 +474,312 @@ void SrsThreadMutex::unlock()
     srs_assert(!r0);
 }
 
+SrsThreadEntry::SrsThreadEntry()
+{
+    pool = NULL;
+    start = NULL;
+    arg = NULL;
+    num = 0;
+    tid = 0;
+
+    err = srs_success;
+}
+
+SrsThreadEntry::~SrsThreadEntry()
+{
+    srs_freep(err);
+
+    // TODO: FIXME: Should dispose trd.
+}
+
+SrsThreadPool::SrsThreadPool()
+{
+    entry_ = NULL;
+    lock_ = new SrsThreadMutex();
+    hybrid_ = NULL;
+
+    // Add primordial thread, current thread itself.
+    SrsThreadEntry* entry = new SrsThreadEntry();
+    threads_.push_back(entry);
+    entry_ = entry;
+
+    entry->pool = this;
+    entry->label = "primordial";
+    entry->start = NULL;
+    entry->arg = NULL;
+    entry->num = 1;
+    entry->trd = pthread_self();
+    entry->tid = gettid();
+
+    char buf[256];
+    snprintf(buf, sizeof(buf), "srs-master-%d", entry->num);
+    entry->name = buf;
+
+    pid_fd = -1;
+}
+
+// TODO: FIMXE: If free the pool, we should stop all threads.
+SrsThreadPool::~SrsThreadPool()
+{
+    srs_freep(lock_);
+
+    if (pid_fd > 0) {
+        ::close(pid_fd);
+        pid_fd = -1;
+    }
+}
+
+// Setup the thread-local variables, MUST call when each thread starting.
+srs_error_t SrsThreadPool::setup_thread_locals()
+{
+    srs_error_t err = srs_success;
+
+    // Initialize ST, which depends on pps cids.
+    if ((err = srs_st_init()) != srs_success) {
+        return srs_error_wrap(err, "initialize st failed");
+    }
+
+    return err;
+}
+
+srs_error_t SrsThreadPool::initialize()
+{
+    srs_error_t err = srs_success;
+
+    if ((err = acquire_pid_file()) != srs_success) {
+        return srs_error_wrap(err, "acquire pid file");
+    }
+
+    // Initialize the master primordial thread.
+    SrsThreadEntry* entry = (SrsThreadEntry*)entry_;
+
+    interval_ = _srs_config->get_threads_interval();
+
+    srs_trace("Thread #%d(%s): init name=%s, interval=%dms", entry->num, entry->label.c_str(), entry->name.c_str(), srsu2msi(interval_));
+
+    return err;
+}
+
+srs_error_t SrsThreadPool::acquire_pid_file()
+{
+    std::string pid_file = _srs_config->get_pid_file();
+
+    // -rw-r--r--
+    // 644
+    int mode = S_IRUSR | S_IWUSR |  S_IRGRP | S_IROTH;
+
+    int fd;
+    // open pid file
+    if ((fd = ::open(pid_file.c_str(), O_WRONLY | O_CREAT, mode)) == -1) {
+        return srs_error_new(ERROR_SYSTEM_PID_ACQUIRE, "open pid file=%s", pid_file.c_str());
+    }
+
+    // require write lock
+    struct flock lock;
+
+    lock.l_type = F_WRLCK; // F_RDLCK, F_WRLCK, F_UNLCK
+    lock.l_start = 0; // type offset, relative to l_whence
+    lock.l_whence = SEEK_SET;  // SEEK_SET, SEEK_CUR, SEEK_END
+    lock.l_len = 0;
+
+    if (fcntl(fd, F_SETLK, &lock) == -1) {
+        if(errno == EACCES || errno == EAGAIN) {
+            ::close(fd);
+            srs_error("srs is already running!");
+            return srs_error_new(ERROR_SYSTEM_PID_ALREADY_RUNNING, "srs is already running");
+        }
+        return srs_error_new(ERROR_SYSTEM_PID_LOCK, "access to pid=%s", pid_file.c_str());
+    }
+
+    // truncate file
+    if (ftruncate(fd, 0) != 0) {
+        return srs_error_new(ERROR_SYSTEM_PID_TRUNCATE_FILE, "truncate pid file=%s", pid_file.c_str());
+    }
+
+    // write the pid
+    string pid = srs_int2str(getpid());
+    if (write(fd, pid.c_str(), pid.length()) != (int)pid.length()) {
+        return srs_error_new(ERROR_SYSTEM_PID_WRITE_FILE, "write pid=%s to file=%s", pid.c_str(), pid_file.c_str());
+    }
+
+    // auto close when fork child process.
+    int val;
+    if ((val = fcntl(fd, F_GETFD, 0)) < 0) {
+        return srs_error_new(ERROR_SYSTEM_PID_GET_FILE_INFO, "fcntl fd=%d", fd);
+    }
+    val |= FD_CLOEXEC;
+    if (fcntl(fd, F_SETFD, val) < 0) {
+        return srs_error_new(ERROR_SYSTEM_PID_SET_FILE_INFO, "lock file=%s fd=%d", pid_file.c_str(), fd);
+    }
+
+    srs_trace("write pid=%s to %s success!", pid.c_str(), pid_file.c_str());
+    pid_fd = fd;
+
+    return srs_success;
+}
+
+srs_error_t SrsThreadPool::execute(string label, srs_error_t (*start)(void* arg), void* arg)
+{
+    srs_error_t err = srs_success;
+
+    SrsThreadEntry* entry = new SrsThreadEntry();
+
+    // Update the hybrid thread entry for circuit breaker.
+    if (label == "hybrid") {
+        hybrid_ = entry;
+        hybrids_.push_back(entry);
+    }
+
+    // To protect the threads_ for executing thread-safe.
+    if (true) {
+        SrsThreadLocker(lock_);
+        threads_.push_back(entry);
+    }
+
+    entry->pool = this;
+    entry->label = label;
+    entry->start = start;
+    entry->arg = arg;
+
+    // The id of thread, should equal to the debugger thread id.
+    // For gdb, it's: info threads
+    // For lldb, it's: thread list
+    static int num = entry_->num + 1;
+    entry->num = num++;
+
+    char buf[256];
+    snprintf(buf, sizeof(buf), "srs-%s-%d", entry->label.c_str(), entry->num);
+    entry->name = buf;
+
+    // https://man7.org/linux/man-pages/man3/pthread_create.3.html
+    pthread_t trd;
+    int r0 = pthread_create(&trd, NULL, SrsThreadPool::start, entry);
+    if (r0 != 0) {
+        entry->err = srs_error_new(ERROR_THREAD_CREATE, "create thread %s, r0=%d", label.c_str(), r0);
+        return srs_error_copy(entry->err);
+    }
+
+    entry->trd = trd;
+
+    return err;
+}
+
+srs_error_t SrsThreadPool::run()
+{
+    srs_error_t err = srs_success;
+
+    while (true) {
+        vector<SrsThreadEntry*> threads;
+        if (true) {
+            SrsThreadLocker(lock_);
+            threads = threads_;
+        }
+
+        // Check the threads status fastly.
+        int loops = (int)(interval_ / SRS_UTIME_SECONDS);
+        for (int i = 0; i < loops; i++) {
+            for (int i = 0; i < (int)threads.size(); i++) {
+                SrsThreadEntry* entry = threads.at(i);
+                if (entry->err != srs_success) {
+                    // Quit with success.
+                    if (srs_error_code(entry->err) == ERROR_THREAD_FINISHED) {
+                        srs_trace("quit for thread #%d(%s) finished", entry->num, entry->label.c_str());
+                        srs_freep(err);
+                        return srs_success;
+                    }
+
+                    // Quit with specified error.
+                    err = srs_error_copy(entry->err);
+                    err = srs_error_wrap(err, "thread #%d(%s)", entry->num, entry->label.c_str());
+                    return err;
+                }
+            }
+
+            srs_usleep(1 * SRS_UTIME_SECONDS);
+        }
+
+        // Show statistics for RTC server.
+        SrsProcSelfStat* u = srs_get_self_proc_stat();
+        // Resident Set Size: number of pages the process has in real memory.
+        int memory = (int)(u->rss * 4 / 1024);
+
+        srs_trace("Process: cpu=%.2f%%,%dMB, threads=%d", u->percent * 100, memory, (int)threads_.size());
+    }
+
+    return err;
+}
+
+void SrsThreadPool::stop()
+{
+    // TODO: FIXME: Should notify other threads to do cleanup and quit.
+}
+
+SrsThreadEntry* SrsThreadPool::self()
+{
+    std::vector<SrsThreadEntry*> threads;
+
+    if (true) {
+        SrsThreadLocker(lock_);
+        threads = threads_;
+    }
+
+    for (int i = 0; i < (int)threads.size(); i++) {
+        SrsThreadEntry* entry = threads.at(i);
+        if (entry->trd == pthread_self()) {
+            return entry;
+        }
+    }
+
+    return NULL;
+}
+
+SrsThreadEntry* SrsThreadPool::hybrid()
+{
+    return hybrid_;
+}
+
+vector<SrsThreadEntry*> SrsThreadPool::hybrids()
+{
+    return hybrids_;
+}
+
+void* SrsThreadPool::start(void* arg)
+{
+    srs_error_t err = srs_success;
+
+    SrsThreadEntry* entry = (SrsThreadEntry*)arg;
+
+    // Initialize thread-local variables.
+    if ((err = SrsThreadPool::setup_thread_locals()) != srs_success) {
+        entry->err = err;
+        return NULL;
+    }
+
+    // Set the thread local fields.
+    entry->tid = gettid();
+
+#ifndef SRS_OSX
+    // https://man7.org/linux/man-pages/man3/pthread_setname_np.3.html
+    pthread_setname_np(pthread_self(), entry->name.c_str());
+#else
+    pthread_setname_np(entry->name.c_str());
+#endif
+
+    srs_trace("Thread #%d: run with tid=%d, entry=%p, label=%s, name=%s", entry->num, (int)entry->tid, entry, entry->label.c_str(), entry->name.c_str());
+
+    if ((err = entry->start(entry->arg)) != srs_success) {
+        entry->err = err;
+    }
+
+    // We use a special error to indicates the normally done.
+    if (entry->err == srs_success) {
+        entry->err = srs_error_new(ERROR_THREAD_FINISHED, "finished normally");
+    }
+
+    // We do not use the return value, the err has been set to entry->err.
+    return NULL;
+}
+
+// It MUST be thread-safe, global and shared object.
+SrsThreadPool* _srs_thread_pool = new SrsThreadPool();
+