Threads-Hybrid: Start multiple hybrid threads

winlinvip · winlinvip · commit babe8e67876a · 2021-04-26T08:14:12.000+08:00
diff --git a/trunk/src/app/srs_app_hybrid.cpp b/trunk/src/app/srs_app_hybrid.cpp
@@ -146,6 +146,8 @@ SrsHybridServer::SrsHybridServer()
     timer_ = NULL;
 
     clock_monitor_ = new SrsClockWallMonitor();
+
+    stream_index_ = -1;
 }
 
 SrsHybridServer::~SrsHybridServer()
@@ -444,5 +446,5 @@ srs_error_t SrsHybridServer::on_thread_message(SrsThreadMessage* msg, SrsThreadP
     return err;
 }
 
-SrsHybridServer* _srs_hybrid = new SrsHybridServer();
+ __thread SrsHybridServer* _srs_hybrid = NULL;
 
diff --git a/trunk/src/app/srs_app_hybrid.hpp b/trunk/src/app/srs_app_hybrid.hpp
@@ -56,11 +56,17 @@ class SrsHybridServer : public ISrsFastTimer, public ISrsThreadResponder
     std::vector<ISrsHybridServer*> servers;
     SrsFastTimer* timer_;
     SrsClockWallMonitor* clock_monitor_;
+private:
+    // The config index for hybrid/stream server.
+    int stream_index_;
 public:
     SrsHybridServer();
     virtual ~SrsHybridServer();
 public:
     virtual void register_server(ISrsHybridServer* svr);
+public:
+    int stream_index() { return stream_index_; } // SrsHybridServer::stream_index()
+    void set_stream_index(int v) { stream_index_ = v; } // SrsHybridServer::set_stream_index()
 public:
     virtual srs_error_t initialize();
     virtual srs_error_t run();
@@ -74,6 +80,6 @@ class SrsHybridServer : public ISrsFastTimer, public ISrsThreadResponder
     srs_error_t on_thread_message(SrsThreadMessage* msg, SrsThreadPipeChannel* channel);
 };
 
-extern SrsHybridServer* _srs_hybrid;
+extern __thread SrsHybridServer* _srs_hybrid;
 
 #endif
diff --git a/trunk/src/app/srs_app_rtc_dtls.cpp b/trunk/src/app/srs_app_rtc_dtls.cpp
@@ -246,9 +246,6 @@ srs_error_t SrsDtlsCertificate::initialize()
     // OPENSSL_init_ssl();
 #endif
 
-    // Initialize SRTP first.
-    srs_assert(srtp_init() == 0);
-
     // Whether use ECDSA certificate.
     ecdsa_mode = _srs_config->get_rtc_server_ecdsa();
 
diff --git a/trunk/src/app/srs_app_server.cpp b/trunk/src/app/srs_app_server.cpp
@@ -945,7 +945,8 @@ srs_error_t SrsServer::register_signal()
 srs_error_t SrsServer::ingest()
 {
     srs_error_t err = srs_success;
-    
+
+    // TODO: FIXME: Should move from hybrid to api threads.
     if ((err = ingester->start()) != srs_success) {
         return srs_error_wrap(err, "ingest start");
     }
diff --git a/trunk/src/app/srs_app_threads.cpp b/trunk/src/app/srs_app_threads.cpp
@@ -29,6 +29,7 @@
 #include <srs_core_autofree.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_app_utility.hpp>
+#include <srs_app_hybrid.hpp>
 
 #include <unistd.h>
 
@@ -505,13 +506,20 @@ srs_error_t SrsThreadPool::setup()
         return srs_error_wrap(err, "init st");
     }
 
+    // Create the hybrid RTMP/HTTP/RTC server.
+    _srs_hybrid = new SrsHybridServer();
+
     return err;
 }
 
 srs_error_t SrsThreadPool::initialize()
 {
     srs_error_t err = srs_success;
 
+    // Initialize global shared SRTP once.
+    srs_assert(srtp_init() == 0);
+
+    // Initialize the master primordial thread.
     SrsThreadEntry* entry = (SrsThreadEntry*)entry_;
 #ifndef SRS_OSX
     // Load CPU affinity from config.
diff --git a/trunk/src/main/srs_main_server.cpp b/trunk/src/main/srs_main_server.cpp
@@ -530,12 +530,17 @@ srs_error_t run_in_thread_pool()
         return srs_error_wrap(err, "start api server thread");
     }
 
-    // Start the hybrid service worker thread, for RTMP and RTC server, etc.
-    if ((err = _srs_thread_pool->execute("hybrid", run_hybrid_server, NULL)) != srs_success) {
-        return srs_error_wrap(err, "start hybrid server thread");
+    // Start a number of hybrid service threads.
+    int hybrids = _srs_config->get_threads_hybrids();
+    for (int stream_index = 0; stream_index < hybrids; stream_index++) {
+        // TODO: FIXME: Change the thread name for debugging?
+        // Start the hybrid service worker thread, for RTMP and RTC server, etc.
+        if ((err = _srs_thread_pool->execute("hybrid", run_hybrid_server, (void*)(uint64_t)stream_index)) != srs_success) {
+            return srs_error_wrap(err, "start hybrid server %d thread", stream_index);
+        }
     }
 
-    srs_trace("Pool: Start threads srtp=%d, recv=%d, send=%d", srtps, recvs, sends);
+    srs_trace("Pool: Start threads hybrids=%d, srtp=%d, recv=%d, send=%d", hybrids, srtps, recvs, sends);
 
     return _srs_thread_pool->run();
 }
@@ -545,6 +550,10 @@ srs_error_t run_hybrid_server(void* arg)
 {
     srs_error_t err = srs_success;
 
+    // The config index for hybrid/stream server.
+    int stream_index = (int)(uint64_t)arg;
+    _srs_hybrid->set_stream_index(stream_index);
+
     // Create servers and register them.
     _srs_hybrid->register_server(new SrsServerAdapter());
 

Original file line number	Diff line number	Diff line change
`@@ -146,6 +146,8 @@ SrsHybridServer::SrsHybridServer()`
`146`	`146`	`timer_ = NULL;`
`147`	`147`
`148`	`148`	`clock_monitor_ = new SrsClockWallMonitor();`
	`149`	`+`
	`150`	`+ stream_index_ = -1;`
`149`	`151`	`}`
`150`	`152`
`151`	`153`	`SrsHybridServer::~SrsHybridServer()`
`@@ -444,5 +446,5 @@ srs_error_t SrsHybridServer::on_thread_message(SrsThreadMessage* msg, SrsThreadP`
`444`	`446`	`return err;`
`445`	`447`	`}`
`446`	`448`
`447`		`-SrsHybridServer* _srs_hybrid = new SrsHybridServer();`
	`449`	`+ __thread SrsHybridServer* _srs_hybrid = NULL;`
`448`	`450`
Original file line number	Diff line number	Diff line change
`@@ -945,7 +945,8 @@ srs_error_t SrsServer::register_signal()`
`945`	`945`	`srs_error_t SrsServer::ingest()`
`946`	`946`	`{`
`947`	`947`	`srs_error_t err = srs_success;`
`948`		`-`
	`948`	`+`
	`949`	`+ // TODO: FIXME: Should move from hybrid to api threads.`
`949`	`950`	`if ((err = ingester->start()) != srs_success) {`
`950`	`951`	`return srs_error_wrap(err, "ingest start");`
`951`	`952`	`}`