ST: Use coroutine to execute valgrind check.

ossrs · Aug 20, 2024 · d7f45c5 · d7f45c5
1 parent f560dcd
commit d7f45c5
Show file tree

Hide file tree

Showing 7 changed files with 59 additions and 16 deletions.
diff --git a/trunk/src/app/srs_app_gb28181.hpp b/trunk/src/app/srs_app_gb28181.hpp
@@ -188,7 +188,7 @@ class SrsGbSession : public ISrsResource, public ISrsCoroutineHandler, public IS
     void on_media_transport(SrsSharedResource<SrsGbMediaTcpConn> media);
     // Get the candidate for SDP generation, the public IP address for device to connect to.
     std::string pip();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 private:
@@ -305,7 +305,7 @@ class SrsGbSipTcpConn : public ISrsResource, public ISrsCoroutineHandler, public
 public:
     virtual const SrsContextId& get_id();
     virtual std::string desc();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 private:
@@ -333,7 +333,7 @@ class SrsGbSipTcpReceiver : public ISrsStartable, public ISrsCoroutineHandler
 // Interface ISrsStartable
 public:
     virtual srs_error_t start();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 private:
@@ -362,7 +362,7 @@ class SrsGbSipTcpSender : public ISrsStartable, public ISrsCoroutineHandler
 // Interface ISrsStartable
 public:
     virtual srs_error_t start();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 private:
@@ -422,7 +422,7 @@ class SrsGbMediaTcpConn : public ISrsResource, public ISrsCoroutineHandler, publ
 public:
     virtual const SrsContextId& get_id();
     virtual std::string desc();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 private:

diff --git a/trunk/src/app/srs_app_http_api.cpp b/trunk/src/app/srs_app_http_api.cpp
@@ -1097,14 +1097,25 @@ srs_error_t SrsGoApiTcmalloc::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
 
 SrsGoApiValgrind::SrsGoApiValgrind()
 {
+    trd_ = NULL;
 }
 
 SrsGoApiValgrind::~SrsGoApiValgrind()
 {
+    srs_freep(trd_);
 }
 
 srs_error_t SrsGoApiValgrind::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r)
 {
+    srs_error_t err = srs_success;
+
+    if (!trd_) {
+        trd_ = new SrsSTCoroutine("valgrind", this, _srs_context->get_id());
+        if ((err = trd_->start()) != srs_success) {
+            return srs_error_wrap(err, "start");
+        }
+    }
+
     string check = r->query_get("check");
     srs_trace("query check=%s", check.c_str());
 
@@ -1125,26 +1136,52 @@ srs_error_t SrsGoApiValgrind::serve_http(ISrsHttpResponseWriter* w, ISrsHttpMess
     res->set("see", SrsJsonAny::str("https://valgrind.org/docs/manual/mc-manual.html"));
     obj->set("data", res);
 
-    // Does a full memory check right now.
+    // Does a memory check later.
     if (check == "full") {
         res->set("call", SrsJsonAny::str("VALGRIND_DO_LEAK_CHECK"));
-        VALGRIND_DO_LEAK_CHECK;
     } else if (check == "quick") {
         res->set("call", SrsJsonAny::str("VALGRIND_DO_QUICK_LEAK_CHECK"));
-        VALGRIND_DO_QUICK_LEAK_CHECK;
     } else if (check == "added") {
         res->set("call", SrsJsonAny::str("VALGRIND_DO_ADDED_LEAK_CHECK"));
-        VALGRIND_DO_ADDED_LEAK_CHECK;
     } else if (check == "changed") {
         res->set("call", SrsJsonAny::str("VALGRIND_DO_CHANGED_LEAK_CHECK"));
-        VALGRIND_DO_CHANGED_LEAK_CHECK;
     } else if (check == "new") {
         res->set("call", SrsJsonAny::str("VALGRIND_DO_NEW_LEAK_CHECK"));
-        VALGRIND_DO_NEW_LEAK_CHECK;
     }
+    task_ = check;
 
     return srs_api_response(w, r, obj->dumps());
 }
+
+srs_error_t SrsGoApiValgrind::cycle()
+{
+    srs_error_t err = srs_success;
+
+    while (true) {
+        if ((err = trd_->pull()) != srs_success) {
+            return srs_error_wrap(err, "pull");
+        }
+
+        std::string check = task_;
+        task_ = "";
+
+        if (check == "full") {
+            VALGRIND_DO_LEAK_CHECK;
+        } else if (check == "quick") {
+            VALGRIND_DO_QUICK_LEAK_CHECK;
+        } else if (check == "added") {
+            VALGRIND_DO_ADDED_LEAK_CHECK;
+        } else if (check == "changed") {
+            VALGRIND_DO_CHANGED_LEAK_CHECK;
+        } else if (check == "new") {
+            VALGRIND_DO_NEW_LEAK_CHECK;
+        }
+
+        srs_usleep(3 * SRS_UTIME_SECONDS);
+    }
+
+    return err;
+}
 #endif
 
 SrsGoApiMetrics::SrsGoApiMetrics()

diff --git a/trunk/src/app/srs_app_http_api.hpp b/trunk/src/app/srs_app_http_api.hpp
@@ -217,13 +217,19 @@ class SrsGoApiTcmalloc : public ISrsHttpHandler
 #endif
 
 #ifdef SRS_VALGRIND
-class SrsGoApiValgrind : public ISrsHttpHandler
+class SrsGoApiValgrind : public ISrsHttpHandler, public ISrsCoroutineHandler
 {
+private:
+    SrsCoroutine* trd_;
+    std::string task_;
 public:
     SrsGoApiValgrind();
     virtual ~SrsGoApiValgrind();
 public:
     virtual srs_error_t serve_http(ISrsHttpResponseWriter* w, ISrsHttpMessage* r);
+// Interface ISrsCoroutineHandler
+public:
+    virtual srs_error_t cycle();
 };
 #endif
 

diff --git a/trunk/src/app/srs_app_http_conn.hpp b/trunk/src/app/srs_app_http_conn.hpp
@@ -94,7 +94,7 @@ class SrsHttpConn : public ISrsConnection, public ISrsStartable, public ISrsCoro
 // Interface ISrsStartable
 public:
     virtual srs_error_t start();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 private:

diff --git a/trunk/src/app/srs_app_recv_thread.hpp b/trunk/src/app/srs_app_recv_thread.hpp
@@ -204,7 +204,7 @@ class SrsHttpRecvThread : public ISrsCoroutineHandler
     virtual srs_error_t start();
 public:
     virtual srs_error_t pull();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 };

diff --git a/trunk/src/app/srs_app_rtmp_conn.hpp b/trunk/src/app/srs_app_rtmp_conn.hpp
@@ -181,7 +181,7 @@ class SrsRtmpConn : public ISrsConnection, public ISrsStartable, public ISrsRelo
     // when client cycle thread stop, invoke the on_thread_stop(), which will use server
     // To remove the client by server->remove(this).
     virtual srs_error_t start();
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     // The thread cycle function,
     // when serve connection completed, terminate the loop which will terminate the thread,

diff --git a/trunk/src/app/srs_app_st.hpp b/trunk/src/app/srs_app_st.hpp
@@ -291,7 +291,7 @@ class SrsExecutorCoroutine : public ISrsResource, public ISrsStartable, public I
 public:
     virtual const SrsContextId& cid();
     virtual void set_cid(const SrsContextId& cid);
-// Interface ISrsOneCycleThreadHandler
+// Interface ISrsCoroutineHandler
 public:
     virtual srs_error_t cycle();
 // Interface ISrsResource