Threads: Support multiple SRTP/SEND/RECV threads.

winlinvip · winlinvip · commit 28d8f1a72c91 · 2021-04-26T08:13:30.000+08:00
1. Move received and cooked packets queue to thread entry, that is, each thread has its own queue.
2. In RECV thread, push received packet to queue of source thread, in thread listener.
3. In SRTP thread, push cooked packet to queue of source thread, in asyn SRTP task.
4. In hybrid thread, directly and only consume the packets of self thread.
5. Sync between SRTP task by lock.
diff --git a/trunk/conf/full.conf b/trunk/conf/full.conf
@@ -121,15 +121,18 @@ threads {
     # The thread pool manager cycle interval, in seconds.
     # Default: 5
     interval 5;
-    # Whether enable the ASYNC SRTP, codec in dedicate threads.
-    # Default: off
-    async_srtp off;
-    # Whether enable the ASYNC RECV, recv udp packets in dedicate threads.
-    # Default: off
-    async_recv off;
-    # Whether enable the ASYNC SEND, send udp packets in dedicate threads.
-    # Default: off
-    async_send off;
+    # The number of threads for SRTP, codec in dedicate threads.
+    # Note that 0 to disable it. Max to 64 threads.
+    # Default: 1
+    async_srtp 1;
+    # The number of threads for RECV, recv udp packets in dedicate threads.
+    # Note that 0 to disable it. Max to 64 threads.
+    # Default: 1
+    async_recv 1;
+    # The number of threads for SEND, send udp packets in dedicate threads.
+    # Note that 0 to disable it. Max to 64 threads.
+    # Default: 1
+    async_send 1;
     # Whether enable the tunnel, to consume packets between srtp/recv/send threads,
     # without proxy by hybrid(except the few head packets).
     # Default: off
diff --git a/trunk/src/app/srs_app_config.cpp b/trunk/src/app/srs_app_config.cpp
@@ -4127,9 +4127,9 @@ srs_utime_t SrsConfig::get_threads_interval()
     return v * SRS_UTIME_SECONDS;
 }
 
-bool SrsConfig::get_threads_async_srtp()
+int SrsConfig::get_threads_async_srtp()
 {
-    static bool DEFAULT = false;
+    static int DEFAULT = 1;
 
     SrsConfDirective* conf = root->get("threads");
     if (!conf) {
@@ -4141,12 +4141,16 @@ bool SrsConfig::get_threads_async_srtp()
         return DEFAULT;
     }
 
-    return SRS_CONF_PERFER_FALSE(conf->arg0());
+    int v = ::atoi(conf->arg0().c_str());
+    if (v < 0 || v > 64) {
+        return DEFAULT;
+    }
+    return v;
 }
 
-bool SrsConfig::get_threads_async_recv()
+int SrsConfig::get_threads_async_recv()
 {
-    static bool DEFAULT = false;
+    static int DEFAULT = 1;
 
     SrsConfDirective* conf = root->get("threads");
     if (!conf) {
@@ -4158,12 +4162,16 @@ bool SrsConfig::get_threads_async_recv()
         return DEFAULT;
     }
 
-    return SRS_CONF_PERFER_FALSE(conf->arg0());
+    int v = ::atoi(conf->arg0().c_str());
+    if (v < 0 || v > 64) {
+        return DEFAULT;
+    }
+    return v;
 }
 
-bool SrsConfig::get_threads_async_send()
+int SrsConfig::get_threads_async_send()
 {
-    static bool DEFAULT = false;
+    static int DEFAULT = 1;
 
     SrsConfDirective* conf = root->get("threads");
     if (!conf) {
@@ -4175,7 +4183,11 @@ bool SrsConfig::get_threads_async_send()
         return DEFAULT;
     }
 
-    return SRS_CONF_PERFER_FALSE(conf->arg0());
+    int v = ::atoi(conf->arg0().c_str());
+    if (v < 0 || v > 64) {
+        return DEFAULT;
+    }
+    return v;
 }
 
 bool SrsConfig::get_threads_async_tunnel()
diff --git a/trunk/src/app/srs_app_config.hpp b/trunk/src/app/srs_app_config.hpp
@@ -478,9 +478,9 @@ class SrsConfig
 // Thread pool section.
 public:
     virtual srs_utime_t get_threads_interval();
-    virtual bool get_threads_async_srtp();
-    virtual bool get_threads_async_recv();
-    virtual bool get_threads_async_send();
+    virtual int get_threads_async_srtp();
+    virtual int get_threads_async_recv();
+    virtual int get_threads_async_send();
     virtual bool get_threads_async_tunnel();
     virtual bool get_threads_cpu_affinity(std::string label, int* start, int* end);
     virtual int get_threads_max_recv_queue();
diff --git a/trunk/src/app/srs_app_hybrid.cpp b/trunk/src/app/srs_app_hybrid.cpp
@@ -215,17 +215,22 @@ srs_error_t SrsHybridServer::run()
         }
     }
 
+    // Get the entry of self thread.
+    SrsThreadEntry* entry = _srs_thread_pool->self();
+
     // Consume the async UDP/SRTP packets.
     while (true) {
         int consumed = 0;
 
         // Consume the received UDP packets.
-        if ((err = _srs_async_recv->consume(&consumed)) != srs_success) {
+        // Note that this might run in multiple threads, but it's ok.
+        if ((err = _srs_async_recv->consume(entry, &consumed)) != srs_success) {
             srs_error_reset(err); // Ignore any error.
         }
 
         // Consume the cooked SRTP packets.
-        if ((err = _srs_async_srtp->consume(&consumed)) != srs_success) {
+        // Note that this might run in multiple threads, but it's ok.
+        if ((err = _srs_async_srtp->consume(entry, &consumed)) != srs_success) {
             srs_error_reset(err); // Ignore any error.
         }
 
diff --git a/trunk/src/app/srs_app_listener.cpp b/trunk/src/app/srs_app_listener.cpp
@@ -669,7 +669,7 @@ srs_error_t SrsUdpMuxListener::cycle()
     set_socket_buffer();
 
     // Sleep infinite if use async_recv.
-    if (_srs_config->get_threads_async_recv()) {
+    if (_srs_config->get_threads_async_recv() > 0) {
         SrsThreadUdpListener* listener = new SrsThreadUdpListener(lfd, handler);
         _srs_async_recv->add_listener(listener);
 
diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp
@@ -100,8 +100,8 @@ SrsSecurityTransport::SrsSecurityTransport(SrsRtcConnection* s)
 
     dtls_ = new SrsDtls((ISrsDtlsCallback*)this);
 
-    bool async_srtp = _srs_config->get_threads_async_srtp();
-    if (!async_srtp) {
+    int async_srtp = _srs_config->get_threads_async_srtp();
+    if (async_srtp <= 0) {
         srtp_ = new SrsSRTP();
     } else {
         srtp_ = new SrsAsyncSRTP(this);
diff --git a/trunk/src/app/srs_app_threads.cpp b/trunk/src/app/srs_app_threads.cpp
diff --git a/trunk/src/app/srs_app_threads.hpp b/trunk/src/app/srs_app_threads.hpp
diff --git a/trunk/src/main/srs_main_server.cpp b/trunk/src/main/srs_main_server.cpp

Original file line number	Diff line number	Diff line change
`@@ -4127,9 +4127,9 @@ srs_utime_t SrsConfig::get_threads_interval()`
`4127`	`4127`	`return v * SRS_UTIME_SECONDS;`
`4128`	`4128`	`}`
`4129`	`4129`
`4130`		`-bool SrsConfig::get_threads_async_srtp()`
	`4130`	`+int SrsConfig::get_threads_async_srtp()`
`4131`	`4131`	`{`
`4132`		`- static bool DEFAULT = false;`
	`4132`	`+ static int DEFAULT = 1;`
`4133`	`4133`
`4134`	`4134`	`SrsConfDirective* conf = root->get("threads");`
`4135`	`4135`	`if (!conf) {`
`@@ -4141,12 +4141,16 @@ bool SrsConfig::get_threads_async_srtp()`
`4141`	`4141`	`return DEFAULT;`
`4142`	`4142`	`}`
`4143`	`4143`
`4144`		`- return SRS_CONF_PERFER_FALSE(conf->arg0());`
	`4144`	`+ int v = ::atoi(conf->arg0().c_str());`
	`4145`	`+ if (v < 0 \|\| v > 64) {`
	`4146`	`+ return DEFAULT;`
	`4147`	`+ }`
	`4148`	`+ return v;`
`4145`	`4149`	`}`
`4146`	`4150`
`4147`		`-bool SrsConfig::get_threads_async_recv()`
	`4151`	`+int SrsConfig::get_threads_async_recv()`
`4148`	`4152`	`{`
`4149`		`- static bool DEFAULT = false;`
	`4153`	`+ static int DEFAULT = 1;`
`4150`	`4154`
`4151`	`4155`	`SrsConfDirective* conf = root->get("threads");`
`4152`	`4156`	`if (!conf) {`
`@@ -4158,12 +4162,16 @@ bool SrsConfig::get_threads_async_recv()`
`4158`	`4162`	`return DEFAULT;`
`4159`	`4163`	`}`
`4160`	`4164`
`4161`		`- return SRS_CONF_PERFER_FALSE(conf->arg0());`
	`4165`	`+ int v = ::atoi(conf->arg0().c_str());`
	`4166`	`+ if (v < 0 \|\| v > 64) {`
	`4167`	`+ return DEFAULT;`
	`4168`	`+ }`
	`4169`	`+ return v;`
`4162`	`4170`	`}`
`4163`	`4171`
`4164`		`-bool SrsConfig::get_threads_async_send()`
	`4172`	`+int SrsConfig::get_threads_async_send()`
`4165`	`4173`	`{`
`4166`		`- static bool DEFAULT = false;`
	`4174`	`+ static int DEFAULT = 1;`
`4167`	`4175`
`4168`	`4176`	`SrsConfDirective* conf = root->get("threads");`
`4169`	`4177`	`if (!conf) {`
`@@ -4175,7 +4183,11 @@ bool SrsConfig::get_threads_async_send()`
`4175`	`4183`	`return DEFAULT;`
`4176`	`4184`	`}`
`4177`	`4185`
`4178`		`- return SRS_CONF_PERFER_FALSE(conf->arg0());`
	`4186`	`+ int v = ::atoi(conf->arg0().c_str());`
	`4187`	`+ if (v < 0 \|\| v > 64) {`
	`4188`	`+ return DEFAULT;`
	`4189`	`+ }`
	`4190`	`+ return v;`
`4179`	`4191`	`}`
`4180`	`4192`
`4181`	`4193`	`bool SrsConfig::get_threads_async_tunnel()`
Original file line number	Diff line number	Diff line change
`@@ -215,17 +215,22 @@ srs_error_t SrsHybridServer::run()`
`215`	`215`	`}`
`216`	`216`	`}`
`217`	`217`
	`218`	`+ // Get the entry of self thread.`
	`219`	`+ SrsThreadEntry* entry = _srs_thread_pool->self();`
	`220`	`+`
`218`	`221`	`// Consume the async UDP/SRTP packets.`
`219`	`222`	`while (true) {`
`220`	`223`	`int consumed = 0;`
`221`	`224`
`222`	`225`	`// Consume the received UDP packets.`
`223`		`- if ((err = _srs_async_recv->consume(&consumed)) != srs_success) {`
	`226`	`+ // Note that this might run in multiple threads, but it's ok.`
	`227`	`+ if ((err = _srs_async_recv->consume(entry, &consumed)) != srs_success) {`
`224`	`228`	`srs_error_reset(err); // Ignore any error.`
`225`	`229`	`}`
`226`	`230`
`227`	`231`	`// Consume the cooked SRTP packets.`
`228`		`- if ((err = _srs_async_srtp->consume(&consumed)) != srs_success) {`
	`232`	`+ // Note that this might run in multiple threads, but it's ok.`
	`233`	`+ if ((err = _srs_async_srtp->consume(entry, &consumed)) != srs_success) {`
`229`	`234`	`srs_error_reset(err); // Ignore any error.`
`230`	`235`	`}`
`231`	`236`