Threads: Refine circuit breaker

winlinvip · winlinvip · commit 6c83e89ca634 · 2021-04-26T09:38:19.000+08:00
diff --git a/trunk/src/app/srs_app_rtc_conn.cpp b/trunk/src/app/srs_app_rtc_conn.cpp
@@ -1321,7 +1321,7 @@ srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket2*& pkt, SrsBuf
     }
 
     // If circuit-breaker is enabled, disable nack.
-    if (_srs_thread_pool->hybrid_critical_water_level()) {
+    if (_srs_circuit_breaker->hybrid_critical_water_level()) {
         ++_srs_pps_snack4->sugar;
         return err;
     }
@@ -1573,7 +1573,7 @@ srs_error_t SrsRtcPublishStream::notify(int type, srs_utime_t interval, srs_utim
         ++_srs_pps_twcc->sugar;
 
         // If circuit-breaker is dropping packet, disable TWCC.
-        if (_srs_thread_pool->hybrid_critical_water_level()) {
+        if (_srs_circuit_breaker->hybrid_critical_water_level()) {
             ++_srs_pps_snack4->sugar;
             return err;
         }
@@ -2432,7 +2432,7 @@ srs_error_t SrsRtcConnection::notify(int type, srs_utime_t interval, srs_utime_t
     // For publisher to send NACK.
     if (type == SRS_TICKID_SEND_NACKS) {
         // If circuit-breaker is enabled, disable nack.
-        if (_srs_thread_pool->hybrid_critical_water_level()) {
+        if (_srs_circuit_breaker->hybrid_critical_water_level()) {
             ++_srs_pps_snack4->sugar;
             return err;
         }
diff --git a/trunk/src/app/srs_app_rtc_queue.cpp b/trunk/src/app/srs_app_rtc_queue.cpp
@@ -235,7 +235,7 @@ SrsRtpNackForReceiver::~SrsRtpNackForReceiver()
 void SrsRtpNackForReceiver::insert(uint16_t first, uint16_t last)
 {
     // If circuit-breaker is enabled, disable nack.
-    if (_srs_thread_pool->hybrid_high_water_level()) {
+    if (_srs_circuit_breaker->hybrid_high_water_level()) {
         ++_srs_pps_snack4->sugar;
         return;
     }
@@ -272,7 +272,7 @@ void SrsRtpNackForReceiver::check_queue_size()
 void SrsRtpNackForReceiver::get_nack_seqs(SrsRtcpNack& seqs, uint32_t& timeout_nacks)
 {
     // If circuit-breaker is enabled, disable nack.
-    if (_srs_thread_pool->hybrid_high_water_level()) {
+    if (_srs_circuit_breaker->hybrid_high_water_level()) {
         queue_.clear();
         ++_srs_pps_snack4->sugar;
         return;
diff --git a/trunk/src/app/srs_app_rtc_source.cpp b/trunk/src/app/srs_app_rtc_source.cpp
@@ -571,7 +571,7 @@ srs_error_t SrsRtcStream::on_rtp(SrsRtpPacket2* pkt)
     srs_error_t err = srs_success;
 
     // If circuit-breaker is dying, drop packet.
-    if (_srs_thread_pool->hybrid_dying_water_level()) {
+    if (_srs_circuit_breaker->hybrid_dying_water_level()) {
         _srs_pps_aloss2->sugar += (int64_t)consumers.size();
         return err;
     }
diff --git a/trunk/src/app/srs_app_threads.cpp b/trunk/src/app/srs_app_threads.cpp
@@ -93,6 +93,120 @@ uint64_t srs_covert_cpuset(cpu_set_t v)
 #endif
 }
 
+SrsCircuitBreaker::SrsCircuitBreaker()
+{
+    hybrid_high_water_level_ = 0;
+    hybrid_critical_water_level_ = 0;
+    hybrid_dying_water_level_ = 0;
+
+    high_threshold_ = 0;
+    high_pulse_ = 0;
+    critical_threshold_ = 0;
+    critical_pulse_ = 0;
+    dying_threshold_ = 0;
+    dying_pulse_ = 0;
+}
+
+SrsCircuitBreaker::~SrsCircuitBreaker()
+{
+}
+
+srs_error_t SrsCircuitBreaker::initialize()
+{
+    srs_error_t err = srs_success;
+
+    // Start a timer to stat the data for circuit breaker.
+    _srs_hybrid->timer()->subscribe(1 * SRS_UTIME_SECONDS, this);
+
+    high_threshold_ = _srs_config->get_high_threshold();
+    high_pulse_ = _srs_config->get_high_pulse();
+    critical_threshold_ = _srs_config->get_critical_threshold();
+    critical_pulse_ = _srs_config->get_critical_pulse();
+    dying_threshold_ = _srs_config->get_dying_threshold();
+    dying_pulse_ = _srs_config->get_dying_pulse();
+
+    srs_trace("CircuitBreaker: high=%dx%d, critical=%dx%d, dying=%dx%d",
+        high_pulse_, high_threshold_, critical_pulse_, critical_threshold_, dying_pulse_, dying_threshold_);
+
+    return err;
+}
+
+bool SrsCircuitBreaker::hybrid_high_water_level()
+{
+    return hybrid_critical_water_level() || hybrid_high_water_level_;
+}
+
+bool SrsCircuitBreaker::hybrid_critical_water_level()
+{
+    return hybrid_dying_water_level() || hybrid_critical_water_level_;
+}
+
+bool SrsCircuitBreaker::hybrid_dying_water_level()
+{
+    return dying_pulse_ && hybrid_dying_water_level_ >= dying_pulse_;
+}
+
+srs_error_t SrsCircuitBreaker::on_timer(srs_utime_t interval, srs_utime_t tick)
+{
+    srs_error_t err = srs_success;
+
+    SrsThreadEntry* entry = _srs_thread_pool->self();
+    if (!entry->stat) {
+        return err;
+    }
+
+    // For Circuit-Breaker to update the SNMP, ASAP.
+    srs_update_udp_snmp_statistic();
+
+    // Update thread CPUs per 1s.
+    srs_update_thread_proc_stat(entry->stat, entry->tid);
+
+    // Update the Circuit-Breaker by water-level.
+    // Reset the high water-level when CPU is low for N times.
+    if (entry->stat->percent * 100 > high_threshold_) {
+        hybrid_high_water_level_ = high_pulse_;
+    } else if (hybrid_high_water_level_ > 0) {
+        hybrid_high_water_level_--;
+    }
+
+    // Reset the critical water-level when CPU is low for N times.
+    if (entry->stat->percent * 100 > critical_threshold_) {
+        hybrid_critical_water_level_ = critical_pulse_;
+    } else if (hybrid_critical_water_level_ > 0) {
+        hybrid_critical_water_level_--;
+    }
+
+    // Reset the dying water-level when CPU is low for N times.
+    if (entry->stat->percent * 100 > dying_threshold_) {
+        hybrid_dying_water_level_ = srs_min(dying_pulse_ + 1, hybrid_dying_water_level_ + 1);
+    } else if (hybrid_dying_water_level_ > 0) {
+        hybrid_dying_water_level_ = 0;
+    }
+
+    // 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);
+
+    // The hybrid thread cpu and memory.
+    float thread_percent = entry->stat->percent * 100;
+
+    string circuit_breaker;
+    if (true || hybrid_high_water_level() || hybrid_critical_water_level() || _srs_pps_aloss->r1s() || _srs_pps_rloss->r1s() || _srs_pps_snack2->r10s()) {
+        srs_trace("CircuitBreaker(%s): cpu=%.2f%%,%dMB, thread=%.2f%%, break=%d,%d,%d, cond=%d,%d,%.2f%%, snk=%d,%d,%d",
+            entry->label.c_str(), u->percent * 100, memory, thread_percent,
+            hybrid_high_water_level(), hybrid_critical_water_level(), hybrid_dying_water_level(), // Whether Circuit-Break is enable.
+            _srs_pps_rloss->r1s(), _srs_pps_aloss->r1s(), thread_percent, // The conditions to enable Circuit-Breaker.
+            _srs_pps_snack2->r10s(), _srs_pps_snack3->r10s(), // NACK packet,seqs sent.
+            _srs_pps_snack4->r10s() // NACK drop by Circuit-Break.
+        );
+    }
+
+    return err;
+}
+
+__thread SrsCircuitBreaker* _srs_circuit_breaker = NULL;
+
 SrsPipe::SrsPipe()
 {
     pipes_[0] = pipes_[1] = -1;
@@ -451,16 +565,6 @@ SrsThreadPool::SrsThreadPool()
     entry_ = NULL;
     lock_ = new SrsThreadMutex();
     hybrid_ = NULL;
-    hybrid_high_water_level_ = 0;
-    hybrid_critical_water_level_ = 0;
-    hybrid_dying_water_level_ = 0;
-
-    high_threshold_ = 0;
-    high_pulse_ = 0;
-    critical_threshold_ = 0;
-    critical_pulse_ = 0;
-    dying_threshold_ = 0;
-    dying_pulse_ = 0;
 
     // Add primordial thread, current thread itself.
     SrsThreadEntry* entry = new SrsThreadEntry();
@@ -493,21 +597,6 @@ SrsThreadPool::~SrsThreadPool()
     }
 }
 
-bool SrsThreadPool::hybrid_high_water_level()
-{
-    return hybrid_critical_water_level() || hybrid_high_water_level_;
-}
-
-bool SrsThreadPool::hybrid_critical_water_level()
-{
-    return hybrid_dying_water_level() || hybrid_critical_water_level_;
-}
-
-bool SrsThreadPool::hybrid_dying_water_level()
-{
-    return dying_pulse_ && hybrid_dying_water_level_ >= dying_pulse_;
-}
-
 // Thread local objects.
 extern const int LOG_MAX_SIZE;
 extern __thread char* _srs_log_data;
@@ -618,6 +707,9 @@ srs_error_t SrsThreadPool::setup()
     // Create the hybrid RTMP/HTTP/RTC server.
     _srs_hybrid = new SrsHybridServer();
 
+    // Create the circuit breaker for each thread.
+    _srs_circuit_breaker = new SrsCircuitBreaker();
+
     // Create the source manager for server.
     _srs_sources = new SrsSourceManager();
 
@@ -780,12 +872,6 @@ srs_error_t SrsThreadPool::initialize()
 #endif
 
     interval_ = _srs_config->get_threads_interval();
-    high_threshold_ = _srs_config->get_high_threshold();
-    high_pulse_ = _srs_config->get_high_pulse();
-    critical_threshold_ = _srs_config->get_critical_threshold();
-    critical_pulse_ = _srs_config->get_critical_pulse();
-    dying_threshold_ = _srs_config->get_dying_threshold();
-    dying_pulse_ = _srs_config->get_dying_pulse();
     int async_srtp = _srs_config->get_threads_async_srtp();
 
     int recv_queue = _srs_config->get_threads_max_recv_queue();
@@ -798,10 +884,9 @@ srs_error_t SrsThreadPool::initialize()
     _srs_async_recv->set_tunnel_enabled(async_tunnel);
     _srs_async_srtp->set_tunnel_enabled(async_tunnel);
 
-    srs_trace("Thread #%d(%s): init name=%s, interval=%dms, async_srtp=%d, cpuset=%d/%d-0x%" PRIx64 "/%d-0x%" PRIx64 ", water_level=%dx%d,%dx%d,%dx%d recvQ=%d, aSend=%d, tunnel=%d",
+    srs_trace("Thread #%d(%s): init name=%s, interval=%dms, async_srtp=%d, cpuset=%d/%d-0x%" PRIx64 "/%d-0x%" PRIx64 ", recvQ=%d, aSend=%d, tunnel=%d",
         entry->num, entry->label.c_str(), entry->name.c_str(), srsu2msi(interval_), async_srtp,
         entry->cpuset_ok, r0, srs_covert_cpuset(entry->cpuset), r1, srs_covert_cpuset(entry->cpuset2),
-        high_pulse_, high_threshold_, critical_pulse_, critical_threshold_, dying_pulse_, dying_threshold_,
         recv_queue, async_send, async_tunnel);
 
     return err;
@@ -943,44 +1028,6 @@ srs_error_t SrsThreadPool::run()
                 }
             }
 
-            // For Circuit-Breaker to update the SNMP, ASAP.
-            srs_update_udp_snmp_statistic();
-
-            // Update thread CPUs per 1s.
-            for (int i = 0; i < (int)threads.size(); i++) {
-                SrsThreadEntry* entry = threads.at(i);
-                if (!entry->tid) {
-                    continue;
-                }
-
-                srs_update_thread_proc_stat(entry->stat, entry->tid);
-            }
-
-            // Update the Circuit-Breaker by water-level.
-            // TODO: FIXME: Should stat all hybrid servers.
-            if (hybrid_ && hybrid_->stat) {
-                // Reset the high water-level when CPU is low for N times.
-                if (hybrid_->stat->percent * 100 > high_threshold_) {
-                    hybrid_high_water_level_ = high_pulse_;
-                } else if (hybrid_high_water_level_ > 0) {
-                    hybrid_high_water_level_--;
-                }
-
-                // Reset the critical water-level when CPU is low for N times.
-                if (hybrid_->stat->percent * 100 > critical_threshold_) {
-                    hybrid_critical_water_level_ = critical_pulse_;
-                } else if (hybrid_critical_water_level_ > 0) {
-                    hybrid_critical_water_level_--;
-                }
-
-                // Reset the dying water-level when CPU is low for N times.
-                if (hybrid_->stat->percent * 100 > dying_threshold_) {
-                    hybrid_dying_water_level_ = srs_min(dying_pulse_ + 1, hybrid_dying_water_level_ + 1);
-                } else if (hybrid_dying_water_level_ > 0) {
-                    hybrid_dying_water_level_ = 0;
-                }
-            }
-
             srs_usleep(1 * SRS_UTIME_SECONDS);
         }
 
@@ -1011,16 +1058,8 @@ srs_error_t SrsThreadPool::run()
             tunnel_desc = buf;
         }
 
-        // 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);
-
         // The hybrid thread cpu and memory.
-        float thread_percent = 0.0f, top_percent = 0.0f;
-        if (hybrid_ && hybrid_->stat) {
-            thread_percent = hybrid_->stat->percent * 100;
-        }
+        float top_percent = 0.0f;
         for (int i = 0; i < (int)threads.size(); i++) {
             SrsThreadEntry* entry = threads.at(i);
             if (!entry->stat || entry->stat->percent <= 0) {
@@ -1029,20 +1068,14 @@ srs_error_t SrsThreadPool::run()
             top_percent = srs_max(top_percent, entry->stat->percent * 100);
         }
 
-        string circuit_breaker;
-        if (hybrid_high_water_level() || hybrid_critical_water_level() || _srs_pps_aloss->r1s() || _srs_pps_rloss->r1s() || _srs_pps_snack2->r10s()) {
-            snprintf(buf, sizeof(buf), ", break=%d,%d,%d, cond=%d,%d,%.2f%%, snk=%d,%d,%d",
-                hybrid_high_water_level(), hybrid_critical_water_level(), hybrid_dying_water_level(), // Whether Circuit-Break is enable.
-                _srs_pps_rloss->r1s(), _srs_pps_aloss->r1s(), thread_percent, // The conditions to enable Circuit-Breaker.
-                _srs_pps_snack2->r10s(), _srs_pps_snack3->r10s(), // NACK packet,seqs sent.
-                _srs_pps_snack4->r10s() // NACK drop by Circuit-Break.
-            );
-            circuit_breaker = buf;
-        }
+        // 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,%.2f%%,%.2f%%%s%s%s%s%s",
-            u->percent * 100, memory, (int)threads_.size(), top_percent, thread_percent,
-            async_logs.c_str(), sync_desc.c_str(), queue_desc.c_str(), circuit_breaker.c_str(),
+        srs_trace("Process: cpu=%.2f%%,%dMB, threads=%d,%.2f%%%%s%s%s%s",
+            u->percent * 100, memory, (int)threads_.size(), top_percent,
+            async_logs.c_str(), sync_desc.c_str(), queue_desc.c_str(),
             tunnel_desc.c_str());
     }
 
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
`@@ -1321,7 +1321,7 @@ srs_error_t SrsRtcPublishStream::do_on_rtp_plaintext(SrsRtpPacket2*& pkt, SrsBuf`
`1321`	`1321`	`}`
`1322`	`1322`
`1323`	`1323`	`// If circuit-breaker is enabled, disable nack.`
`1324`		`- if (_srs_thread_pool->hybrid_critical_water_level()) {`
	`1324`	`+ if (_srs_circuit_breaker->hybrid_critical_water_level()) {`
`1325`	`1325`	`++_srs_pps_snack4->sugar;`
`1326`	`1326`	`return err;`
`1327`	`1327`	`}`
`@@ -1573,7 +1573,7 @@ srs_error_t SrsRtcPublishStream::notify(int type, srs_utime_t interval, srs_utim`
`1573`	`1573`	`++_srs_pps_twcc->sugar;`
`1574`	`1574`
`1575`	`1575`	`// If circuit-breaker is dropping packet, disable TWCC.`
`1576`		`- if (_srs_thread_pool->hybrid_critical_water_level()) {`
	`1576`	`+ if (_srs_circuit_breaker->hybrid_critical_water_level()) {`
`1577`	`1577`	`++_srs_pps_snack4->sugar;`
`1578`	`1578`	`return err;`
`1579`	`1579`	`}`
`@@ -2432,7 +2432,7 @@ srs_error_t SrsRtcConnection::notify(int type, srs_utime_t interval, srs_utime_t`
`2432`	`2432`	`// For publisher to send NACK.`
`2433`	`2433`	`if (type == SRS_TICKID_SEND_NACKS) {`
`2434`	`2434`	`// If circuit-breaker is enabled, disable nack.`
`2435`		`- if (_srs_thread_pool->hybrid_critical_water_level()) {`
	`2435`	`+ if (_srs_circuit_breaker->hybrid_critical_water_level()) {`
`2436`	`2436`	`++_srs_pps_snack4->sugar;`
`2437`	`2437`	`return err;`
`2438`	`2438`	`}`
Original file line number	Diff line number	Diff line change
`@@ -235,7 +235,7 @@ SrsRtpNackForReceiver::~SrsRtpNackForReceiver()`
`235`	`235`	`void SrsRtpNackForReceiver::insert(uint16_t first, uint16_t last)`
`236`	`236`	`{`
`237`	`237`	`// If circuit-breaker is enabled, disable nack.`
`238`		`- if (_srs_thread_pool->hybrid_high_water_level()) {`
	`238`	`+ if (_srs_circuit_breaker->hybrid_high_water_level()) {`
`239`	`239`	`++_srs_pps_snack4->sugar;`
`240`	`240`	`return;`
`241`	`241`	`}`
`@@ -272,7 +272,7 @@ void SrsRtpNackForReceiver::check_queue_size()`
`272`	`272`	`void SrsRtpNackForReceiver::get_nack_seqs(SrsRtcpNack& seqs, uint32_t& timeout_nacks)`
`273`	`273`	`{`
`274`	`274`	`// If circuit-breaker is enabled, disable nack.`
`275`		`- if (_srs_thread_pool->hybrid_high_water_level()) {`
	`275`	`+ if (_srs_circuit_breaker->hybrid_high_water_level()) {`
`276`	`276`	`queue_.clear();`
`277`	`277`	`++_srs_pps_snack4->sugar;`
`278`	`278`	`return;`
Original file line number	Diff line number	Diff line change
`@@ -571,7 +571,7 @@ srs_error_t SrsRtcStream::on_rtp(SrsRtpPacket2* pkt)`
`571`	`571`	`srs_error_t err = srs_success;`
`572`	`572`
`573`	`573`	`// If circuit-breaker is dying, drop packet.`
`574`		`- if (_srs_thread_pool->hybrid_dying_water_level()) {`
	`574`	`+ if (_srs_circuit_breaker->hybrid_dying_water_level()) {`
`575`	`575`	`_srs_pps_aloss2->sugar += (int64_t)consumers.size();`
`576`	`576`	`return err;`
`577`	`577`	`}`