From b2e083b00d0f9f5936c4091745555ee3cbb3998f Mon Sep 17 00:00:00 2001 From: winlin Date: Wed, 29 Jun 2022 20:15:44 +0800 Subject: [PATCH] Support multiple threads by thread pool. v5.0.32 --- trunk/conf/full.conf | 18 +- trunk/doc/CHANGELOG.md | 1 + trunk/src/app/srs_app_config.cpp | 24 +- trunk/src/app/srs_app_config.hpp | 1 + trunk/src/app/srs_app_server.cpp | 77 ------ trunk/src/app/srs_app_server.hpp | 1 - trunk/src/app/srs_app_threads.cpp | 330 +++++++++++++++++++++++++- trunk/src/app/srs_app_threads.hpp | 78 +++++- trunk/src/core/srs_core_version5.hpp | 2 +- trunk/src/kernel/srs_kernel_error.hpp | 2 + trunk/src/main/srs_main_server.cpp | 41 +++- trunk/src/utest/srs_utest.cpp | 8 +- 12 files changed, 479 insertions(+), 104 deletions(-) diff --git a/trunk/conf/full.conf b/trunk/conf/full.conf index f5a5a4d66..1b85d6864 100644 --- a/trunk/conf/full.conf +++ b/trunk/conf/full.conf @@ -7,12 +7,11 @@ # for example, 192.168.1.100:1935 10.10.10.100:1935 # where the ip is optional, default to 0.0.0.0, that is 1935 equals to 0.0.0.0:1935 listen 1935; -# the pid file -# to ensure only one process can use a pid file -# and provides the current running process id, for script, -# for example, init.d script to manage the server. -# default: ./objs/srs.pid -pid ./objs/srs.pid; +# The pid file to write the pid, for managing the SRS process and avoiding duplicated processes. +# If need to run multiple processes, please change this pid file to another one. +# Note: Do not support reloading, for SRS5+ +# Default: ./objs/srs.pid +pid ./objs/srs.pid; # the default chunk size is 128, max is 65536, # some client does not support chunk size change, # however, most clients support it and it can improve @@ -127,6 +126,13 @@ tcmalloc_release_rate 0.8; # Default: on query_latest_version on; +# For thread pool. +threads { + # The thread pool manager cycle interval, in seconds. + # Default: 5 + interval 5; +} + # For system circuit breaker. circuit_breaker { # Whether enable the circuit breaker. diff --git a/trunk/doc/CHANGELOG.md b/trunk/doc/CHANGELOG.md index 6a548f1b0..1118741b4 100644 --- a/trunk/doc/CHANGELOG.md +++ b/trunk/doc/CHANGELOG.md @@ -8,6 +8,7 @@ The changelog for SRS. ## SRS 5.0 Changelog +* v5.0, 2022-06-29, Support multiple threads by thread pool. v5.0.32 * v5.0, 2022-06-28, ST: Support thread-local for multiple threads. v5.0.31 * v5.0, 2022-06-17, Merge [#3010](https://github.com/ossrs/srs/pull/3010): SRT: Support Coroutine Native SRT over ST. (#3010). v5.0.30 * v5.0, 2022-06-15, For [#3058](https://github.com/ossrs/srs/pull/3058): Docker: Support x86_64, armv7 and aarch64 docker image (#3058). v5.0.29 diff --git a/trunk/src/app/srs_app_config.cpp b/trunk/src/app/srs_app_config.cpp index 2f89c2ef0..c877a4feb 100644 --- a/trunk/src/app/srs_app_config.cpp +++ b/trunk/src/app/srs_app_config.cpp @@ -2389,7 +2389,7 @@ srs_error_t SrsConfig::check_normal_config() && n != "ff_log_level" && n != "grace_final_wait" && n != "force_grace_quit" && n != "grace_start_wait" && n != "empty_ip_ok" && n != "disable_daemon_for_docker" && n != "inotify_auto_reload" && n != "auto_reload_for_docker" && n != "tcmalloc_release_rate" - && n != "query_latest_version" + && n != "query_latest_version" && n != "threads" && n != "circuit_breaker" && n != "is_full" && n != "in_docker" ) { return srs_error_new(ERROR_SYSTEM_CONFIG_INVALID, "illegal directive %s", n.c_str()); @@ -3155,6 +3155,28 @@ double SrsConfig::tcmalloc_release_rate() return trr; } +srs_utime_t SrsConfig::get_threads_interval() +{ + static srs_utime_t DEFAULT = 5 * SRS_UTIME_SECONDS; + + SrsConfDirective* conf = root->get("threads"); + if (!conf) { + return DEFAULT; + } + + conf = conf->get("interval"); + if (!conf || conf->arg0().empty()) { + return DEFAULT; + } + + int v = ::atoi(conf->arg0().c_str()); + if (v <= 0) { + return DEFAULT; + } + + return v * SRS_UTIME_SECONDS; +} + bool SrsConfig::get_circuit_breaker() { static bool DEFAULT = true; diff --git a/trunk/src/app/srs_app_config.hpp b/trunk/src/app/srs_app_config.hpp index 845bf1573..409de2fd7 100644 --- a/trunk/src/app/srs_app_config.hpp +++ b/trunk/src/app/srs_app_config.hpp @@ -449,6 +449,7 @@ public: virtual double tcmalloc_release_rate(); // Thread pool section. public: + virtual srs_utime_t get_threads_interval(); virtual bool get_circuit_breaker(); virtual int get_high_threshold(); virtual int get_high_pulse(); diff --git a/trunk/src/app/srs_app_server.cpp b/trunk/src/app/srs_app_server.cpp index 95c4b3325..5a5034562 100644 --- a/trunk/src/app/srs_app_server.cpp +++ b/trunk/src/app/srs_app_server.cpp @@ -687,79 +687,6 @@ srs_error_t SrsServer::initialize_signal() return err; } -srs_error_t SrsServer::acquire_pid_file() -{ - srs_error_t err = srs_success; - - // when srs in dolphin mode, no need the pid file. - if (_srs_config->is_dolphin()) { - return srs_success; - } - - std::string pid_file = _srs_config->get_pid_file(); - - // Try to create dir for pid file. - string pid_dir = srs_path_dirname(pid_file); - if (!srs_path_exists(pid_dir)) { - if ((err = srs_create_dir_recursively(pid_dir)) != srs_success) { - return srs_error_wrap(err, "create %s", pid_dir.c_str()); - } - } - - // -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 SrsServer::listen() { srs_error_t err = srs_success; @@ -1637,10 +1564,6 @@ srs_error_t SrsServerAdapter::run(SrsWaitGroup* wg) return srs_error_wrap(err, "initialize st"); } - if ((err = srs->acquire_pid_file()) != srs_success) { - return srs_error_wrap(err, "acquire pid file"); - } - if ((err = srs->initialize_signal()) != srs_success) { return srs_error_wrap(err, "initialize signal"); } diff --git a/trunk/src/app/srs_app_server.hpp b/trunk/src/app/srs_app_server.hpp index e3d7fd2db..7d9ab9a71 100644 --- a/trunk/src/app/srs_app_server.hpp +++ b/trunk/src/app/srs_app_server.hpp @@ -250,7 +250,6 @@ public: virtual srs_error_t initialize(ISrsServerCycle* ch); virtual srs_error_t initialize_st(); virtual srs_error_t initialize_signal(); - virtual srs_error_t acquire_pid_file(); virtual srs_error_t listen(); virtual srs_error_t register_signal(); virtual srs_error_t http_handle(); diff --git a/trunk/src/app/srs_app_threads.cpp b/trunk/src/app/srs_app_threads.cpp index 4e1a696fb..890c62be7 100644 --- a/trunk/src/app/srs_app_threads.cpp +++ b/trunk/src/app/srs_app_threads.cpp @@ -30,6 +30,20 @@ #include using namespace std; +#include +#include + +#ifdef SRS_OSX +pid_t gettid() { + return 0; +} +#else +#if __GLIBC__ == 2 && __GLIBC_MINOR__ < 30 + #include + #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 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 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 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(); + diff --git a/trunk/src/app/srs_app_threads.hpp b/trunk/src/app/srs_app_threads.hpp index 6fbaa35ed..a3e43c31a 100644 --- a/trunk/src/app/srs_app_threads.hpp +++ b/trunk/src/app/srs_app_threads.hpp @@ -13,6 +13,9 @@ #include +class SrsThreadPool; +class SrsProcSelfStat; + // Protect server in high load. class SrsCircuitBreaker : public ISrsFastTimer { @@ -48,8 +51,8 @@ private: extern SrsCircuitBreaker* _srs_circuit_breaker; -// Initialize global or thread-local variables. -extern srs_error_t srs_thread_initialize(); +// Initialize global shared variables cross all threads. +extern srs_error_t srs_global_initialize(); // The thread mutex wrapper, without error. class SrsThreadMutex @@ -84,5 +87,76 @@ public: } }; +// The information for a thread. +class SrsThreadEntry +{ +public: + SrsThreadPool* pool; + std::string label; + std::string name; + srs_error_t (*start)(void* arg); + void* arg; + int num; + // @see https://man7.org/linux/man-pages/man2/gettid.2.html + pid_t tid; +public: + // The thread object. + pthread_t trd; + // The exit error of thread. + srs_error_t err; +public: + SrsThreadEntry(); + virtual ~SrsThreadEntry(); +}; + +// Allocate a(or almost) fixed thread poll to execute tasks, +// so that we can take the advantage of multiple CPUs. +class SrsThreadPool +{ +private: + SrsThreadEntry* entry_; + srs_utime_t interval_; +private: + SrsThreadMutex* lock_; + std::vector threads_; +private: + // The hybrid server entry, the cpu percent used for circuit breaker. + SrsThreadEntry* hybrid_; + std::vector hybrids_; +private: + // The pid file fd, lock the file write when server is running. + // @remark the init.d script should cleanup the pid file, when stop service, + // for the server never delete the file; when system startup, the pid in pid file + // maybe valid but the process is not SRS, the init.d script will never start server. + int pid_fd; +public: + SrsThreadPool(); + virtual ~SrsThreadPool(); +public: + // Setup the thread-local variables. + static srs_error_t setup_thread_locals(); + // Initialize the thread pool. + srs_error_t initialize(); +private: + // Require the PID file for the whole process. + virtual srs_error_t acquire_pid_file(); +public: + // Execute start function with label in thread. + srs_error_t execute(std::string label, srs_error_t (*start)(void* arg), void* arg); + // Run in the primordial thread, util stop or quit. + srs_error_t run(); + // Stop the thread pool and quit the primordial thread. + void stop(); +public: + SrsThreadEntry* self(); + SrsThreadEntry* hybrid(); + std::vector hybrids(); +private: + static void* start(void* arg); +}; + +// It MUST be thread-safe, global and shared object. +extern SrsThreadPool* _srs_thread_pool; + #endif diff --git a/trunk/src/core/srs_core_version5.hpp b/trunk/src/core/srs_core_version5.hpp index 2309ed682..7193212fb 100644 --- a/trunk/src/core/srs_core_version5.hpp +++ b/trunk/src/core/srs_core_version5.hpp @@ -9,6 +9,6 @@ #define VERSION_MAJOR 5 #define VERSION_MINOR 0 -#define VERSION_REVISION 31 +#define VERSION_REVISION 32 #endif diff --git a/trunk/src/kernel/srs_kernel_error.hpp b/trunk/src/kernel/srs_kernel_error.hpp index 4f6aa7727..a296d4c47 100644 --- a/trunk/src/kernel/srs_kernel_error.hpp +++ b/trunk/src/kernel/srs_kernel_error.hpp @@ -97,6 +97,8 @@ #define ERROR_SOCKET_SETREUSEADDR 1079 #define ERROR_SOCKET_SETCLOSEEXEC 1080 #define ERROR_SOCKET_ACCEPT 1081 +#define ERROR_THREAD_CREATE 1082 +#define ERROR_THREAD_FINISHED 1083 /////////////////////////////////////////////////////// // RTMP protocol error. diff --git a/trunk/src/main/srs_main_server.cpp b/trunk/src/main/srs_main_server.cpp index c69da1517..d21555fb8 100644 --- a/trunk/src/main/srs_main_server.cpp +++ b/trunk/src/main/srs_main_server.cpp @@ -50,12 +50,13 @@ using namespace std; // pre-declare srs_error_t run_directly_or_daemon(); +srs_error_t run_in_thread_pool(); srs_error_t srs_detect_docker(); -srs_error_t run_hybrid_server(); void show_macro_features(); // @global log and context. ISrsLog* _srs_log = NULL; +// It SHOULD be thread-safe, because it use thread-local thread private data. ISrsContext* _srs_context = NULL; // @global config object for app module. SrsConfig* _srs_config = NULL; @@ -73,8 +74,12 @@ srs_error_t do_main(int argc, char** argv) { srs_error_t err = srs_success; - // Initialize global or thread-local variables. - if ((err = srs_thread_initialize()) != srs_success) { + // Initialize global and thread-local variables. + if ((err = srs_global_initialize()) != srs_success) { + return srs_error_wrap(err, "global init"); + } + + if ((err = SrsThreadPool::setup_thread_locals()) != srs_success) { return srs_error_wrap(err, "thread init"); } @@ -404,8 +409,8 @@ srs_error_t run_directly_or_daemon() // If not daemon, directly run hybrid server. if (!run_as_daemon) { - if ((err = run_hybrid_server()) != srs_success) { - return srs_error_wrap(err, "run hybrid"); + if ((err = run_in_thread_pool()) != srs_success) { + return srs_error_wrap(err, "run thread pool"); } return srs_success; } @@ -441,14 +446,34 @@ srs_error_t run_directly_or_daemon() // son srs_trace("son(daemon) process running."); - if ((err = run_hybrid_server()) != srs_success) { - return srs_error_wrap(err, "daemon run hybrid"); + if ((err = run_in_thread_pool()) != srs_success) { + return srs_error_wrap(err, "daemon run thread pool"); } return err; } -srs_error_t run_hybrid_server() +srs_error_t run_hybrid_server(void* arg); +srs_error_t run_in_thread_pool() +{ + srs_error_t err = srs_success; + + // Initialize the thread pool. + if ((err = _srs_thread_pool->initialize()) != srs_success) { + return srs_error_wrap(err, "init thread pool"); + } + + // Start the hybrid service worker thread, for RTMP and RTC server, etc. + if ((err = _srs_thread_pool->execute("hybrid", run_hybrid_server, (void*)NULL)) != srs_success) { + return srs_error_wrap(err, "start hybrid server thread"); + } + + srs_trace("Pool: Start threads primordial=1, hybrids=1 ok"); + + return _srs_thread_pool->run(); +} + +srs_error_t run_hybrid_server(void* /*arg*/) { srs_error_t err = srs_success; diff --git a/trunk/src/utest/srs_utest.cpp b/trunk/src/utest/srs_utest.cpp index d0f98f6e3..d369234fd 100644 --- a/trunk/src/utest/srs_utest.cpp +++ b/trunk/src/utest/srs_utest.cpp @@ -42,8 +42,12 @@ bool _srs_in_docker = false; srs_error_t prepare_main() { srs_error_t err = srs_success; - if ((err = srs_thread_initialize()) != srs_success) { - return srs_error_wrap(err, "init st"); + if ((err = srs_global_initialize()) != srs_success) { + return srs_error_wrap(err, "init global"); + } + + if ((err = SrsThreadPool::setup_thread_locals()) != srs_success) { + return srs_error_wrap(err, "init thread"); } srs_freep(_srs_log);