For #906, #902, use coroutine for reusable2 thread

trunk/src/app/srs_app_edge.cpp

@@ -47,15 +47,9 @@ using namespace std;
 #include <srs_kernel_balance.hpp>
 #include <srs_app_rtmp_conn.hpp>
 
-// when error, edge ingester sleep for a while and retry.
-#define SRS_EDGE_INGESTER_CIMS (3*1000)
-
 // when edge timeout, retry next.
 #define SRS_EDGE_INGESTER_TMMS (5*1000)
 
-// when error, edge ingester sleep for a while and retry.
-#define SRS_EDGE_FORWARDER_CIMS (3*1000)
-
 // when edge error, wait for quit
 #define SRS_EDGE_FORWARDER_TMMS (150)
 
@@ -172,7 +166,7 @@ SrsEdgeIngester::SrsEdgeIngester()
 
     upstream = new SrsEdgeRtmpUpstream(redirect);
     lb = new SrsLbRoundRobin();
-    pthread = new SrsReusableThread2("edge-igs", this, SRS_EDGE_INGESTER_CIMS);
+    trd = NULL;
 }
 
 SrsEdgeIngester::~SrsEdgeIngester()
@@ -181,7 +175,7 @@ SrsEdgeIngester::~SrsEdgeIngester()
 
     srs_freep(upstream);
     srs_freep(lb);
-    srs_freep(pthread);
+    srs_freep(trd);
 }
 
 int SrsEdgeIngester::initialize(SrsSource* s, SrsPlayEdge* e, SrsRequest* r)
@@ -204,12 +198,14 @@ int SrsEdgeIngester::start()
         return ret;
     }
 
-    return pthread->start();
+    srs_freep(trd);
+    trd = new SrsCoroutine("edge-igs", this);
+    return trd->start();
 }
 
 void SrsEdgeIngester::stop()
 {
-    pthread->stop();
+    trd->stop();
     upstream->close();
 
     // notice to unpublish.
@@ -223,11 +219,30 @@ string SrsEdgeIngester::get_curr_origin()
     return lb->selected();
 }
 
+// when error, edge ingester sleep for a while and retry.
+#define SRS_EDGE_INGESTER_CIMS (3*1000)
+
 int SrsEdgeIngester::cycle()
 {
     int ret = ERROR_SUCCESS;
 
-    for (;;) {
+    while (!trd->pull()) {
+        if ((ret = do_cycle()) != ERROR_SUCCESS) {
+            srs_warn("EdgeIngester: Ignore error, ret=%d", ret);
+        }
+
+        if (!trd->pull()) {
+            st_usleep(SRS_EDGE_INGESTER_CIMS * 1000);
+        }
+    }
+    return ret;
+}
+
+int SrsEdgeIngester::do_cycle()
+{
+    int ret = ERROR_SUCCESS;
+
+    while (!trd->pull()) {
         srs_freep(upstream);
         upstream = new SrsEdgeRtmpUpstream(redirect);
 
@@ -275,7 +290,7 @@ int SrsEdgeIngester::ingest()
     // set to larger timeout to read av data from origin.
     upstream->set_recv_timeout(SRS_EDGE_INGESTER_TMMS);
 
-    while (!pthread->interrupted()) {
+    while (!trd->pull()) {
         pprint->elapse();
 
         // pithy print
@@ -408,7 +423,7 @@ SrsEdgeForwarder::SrsEdgeForwarder()
 
     sdk = NULL;
     lb = new SrsLbRoundRobin();
-    pthread = new SrsReusableThread2("edge-fwr", this, SRS_EDGE_FORWARDER_CIMS);
+    trd = NULL;
     queue = new SrsMessageQueue();
 }
 
@@ -417,7 +432,7 @@ SrsEdgeForwarder::~SrsEdgeForwarder()
     stop();
 
     srs_freep(lb);
-    srs_freep(pthread);
+    srs_freep(trd);
     srs_freep(queue);
 }
 
@@ -478,30 +493,50 @@ int SrsEdgeForwarder::start()
         return ret;
     }
 
-    return pthread->start();
+    trd = new SrsCoroutine("edge-fwr", this);
+    return trd->start();
 }
 
 void SrsEdgeForwarder::stop()
 {
-    pthread->stop();
+    trd->stop();
     queue->clear();
     srs_freep(sdk);
 }
 
-#define SYS_MAX_EDGE_SEND_MSGS 128
+// when error, edge ingester sleep for a while and retry.
+#define SRS_EDGE_FORWARDER_CIMS (3*1000)
 
 int SrsEdgeForwarder::cycle()
 {
     int ret = ERROR_SUCCESS;
 
+    while (!trd->pull()) {
+        if ((ret = do_cycle()) != ERROR_SUCCESS) {
+            srs_warn("EdgeForwarder: Ignore error, ret=%d", ret);
+        }
+
+        if (!trd->pull()) {
+            st_usleep(SRS_EDGE_FORWARDER_CIMS * 1000);
+        }
+    }
+    return ret;
+}
+
+#define SYS_MAX_EDGE_SEND_MSGS 128
+
+int SrsEdgeForwarder::do_cycle()
+{
+    int ret = ERROR_SUCCESS;
+
     sdk->set_recv_timeout(SRS_CONSTS_RTMP_PULSE_TMMS);
 
     SrsPithyPrint* pprint = SrsPithyPrint::create_edge();
     SrsAutoFree(SrsPithyPrint, pprint);
 
     SrsMessageArray msgs(SYS_MAX_EDGE_SEND_MSGS);
 
-    while (!pthread->interrupted()) {
+    while (!trd->pull()) {
         if (send_error_code != ERROR_SUCCESS) {
             st_usleep(SRS_EDGE_FORWARDER_TMMS * 1000);
             continue;

trunk/src/app/srs_app_edge.hpp

@@ -114,13 +114,13 @@ class SrsEdgeRtmpUpstream : public SrsEdgeUpstream
 /**
  * edge used to ingest stream from origin.
  */
-class SrsEdgeIngester : public ISrsReusableThread2Handler
+class SrsEdgeIngester : public ISrsCoroutineHandler
 {
 private:
     SrsSource* source;
     SrsPlayEdge* edge;
     SrsRequest* req;
-    SrsReusableThread2* pthread;
+    SrsCoroutine* trd;
     SrsLbRoundRobin* lb;
     SrsEdgeUpstream* upstream;
     // for RTMP 302 redirect.
@@ -136,6 +136,8 @@ class SrsEdgeIngester : public ISrsReusableThread2Handler
 // interface ISrsReusableThread2Handler
 public:
     virtual int cycle();
+private:
+    virtual int do_cycle();
 private:
     virtual int ingest();
     virtual int process_publish_message(SrsCommonMessage* msg);
@@ -144,13 +146,13 @@ class SrsEdgeIngester : public ISrsReusableThread2Handler
 /**
  * edge used to forward stream to origin.
  */
-class SrsEdgeForwarder : public ISrsReusableThread2Handler
+class SrsEdgeForwarder : public ISrsCoroutineHandler
 {
 private:
     SrsSource* source;
     SrsPublishEdge* edge;
     SrsRequest* req;
-    SrsReusableThread2* pthread;
+    SrsCoroutine* trd;
     SrsSimpleRtmpClient* sdk;
     SrsLbRoundRobin* lb;
     /**
@@ -176,6 +178,8 @@ class SrsEdgeForwarder : public ISrsReusableThread2Handler
 // interface ISrsReusableThread2Handler
 public:
     virtual int cycle();
+private:
+    virtual int do_cycle();
 public:
     virtual int proxy(SrsCommonMessage* msg);
 };

trunk/src/app/srs_app_encoder.cpp

@@ -100,7 +100,10 @@ int SrsEncoder::cycle()
         if ((ret = do_cycle()) != ERROR_SUCCESS) {
             srs_warn("Encoder: Ignore error, ret=%d", ret);
         }
-        st_usleep(SRS_RTMP_ENCODER_CIMS * 1000);
+
+        if (!trd->pull()) {
+            st_usleep(SRS_RTMP_ENCODER_CIMS * 1000);
+        }
     }
 
     // kill ffmpeg when finished and it alive

trunk/src/app/srs_app_forward.cpp

@@ -47,9 +47,6 @@ using namespace std;
 #include <srs_kernel_utility.hpp>
 #include <srs_app_rtmp_conn.hpp>
 
-// when error, forwarder sleep for a while and retry.
-#define SRS_FORWARDER_CIMS (3000)
-
 SrsForwarder::SrsForwarder(SrsOriginHub* h)
 {
     hub = h;
@@ -58,15 +55,15 @@ SrsForwarder::SrsForwarder(SrsOriginHub* h)
     sh_video = sh_audio = NULL;
 
     sdk = NULL;
-    pthread = new SrsReusableThread2("forward", this, SRS_FORWARDER_CIMS);
+    trd = NULL;
     queue = new SrsMessageQueue();
     jitter = new SrsRtmpJitter();
 }
 
 SrsForwarder::~SrsForwarder()
 {
     srs_freep(sdk);
-    srs_freep(pthread);
+    srs_freep(trd);
     srs_freep(queue);
     srs_freep(jitter);
 
@@ -138,18 +135,19 @@ int SrsForwarder::on_publish()
               source_ep.c_str(), dest_ep.c_str(), tcUrl.c_str(),
               req->stream.c_str());
 
-    if ((ret = pthread->start()) != ERROR_SUCCESS) {
+    srs_freep(trd);
+    trd = new SrsCoroutine("forward", this);
+    if ((ret = trd->start()) != ERROR_SUCCESS) {
         srs_error("start srs thread failed. ret=%d", ret);
         return ret;
     }
-    srs_trace("forward thread cid=%d, current_cid=%d", pthread->cid(), _srs_context->get_id());
 
     return ret;
 }
 
 void SrsForwarder::on_unpublish()
 {
-    pthread->stop();
+    trd->stop();
     sdk->close();
 }
 
@@ -220,10 +218,30 @@ int SrsForwarder::on_video(SrsSharedPtrMessage* shared_video)
     return ret;
 }
 
+// when error, forwarder sleep for a while and retry.
+#define SRS_FORWARDER_CIMS (3000)
+
 int SrsForwarder::cycle()
 {
     int ret = ERROR_SUCCESS;
 
+    while (!trd->pull()) {
+        if ((ret = do_cycle()) != ERROR_SUCCESS) {
+            srs_warn("Forwarder: Ignore error, ret=%d", ret);
+        }
+
+        if (!trd->pull()) {
+            st_usleep(SRS_FORWARDER_CIMS * 1000);
+        }
+    }
+
+    return ret;
+}
+
+int SrsForwarder::do_cycle()
+{
+    int ret = ERROR_SUCCESS;
+
     std::string url;
     if (true) {
         std::string server;
@@ -289,7 +307,7 @@ int SrsForwarder::forward()
         }
     }
 
-    while (!pthread->interrupted()) {
+    while (!trd->pull()) {
         pprint->elapse();
 
         // read from client.

trunk/src/app/srs_app_forward.hpp

@@ -47,14 +47,14 @@ class SrsSimpleRtmpClient;
  * forward the stream to other servers.
  */
 // TODO: FIXME: refine the error log, comments it.
-class SrsForwarder : public ISrsReusableThread2Handler
+class SrsForwarder : public ISrsCoroutineHandler
 {
 private:
     // the ep to forward, server[:port].
     std::string ep_forward;
     SrsRequest* req;
 private:
-    SrsReusableThread2* pthread;
+    SrsCoroutine* trd;
 private:
     SrsOriginHub* hub;
     SrsSimpleRtmpClient* sdk;
@@ -93,6 +93,8 @@ class SrsForwarder : public ISrsReusableThread2Handler
 // interface ISrsReusableThread2Handler.
 public:
     virtual int cycle();
+private:
+    virtual int do_cycle();
 private:
     virtual int forward();
 };

trunk/src/app/srs_app_ingest.cpp

@@ -181,7 +181,10 @@ int SrsIngester::cycle()
         if ((ret = do_cycle()) != ERROR_SUCCESS) {
             srs_warn("Ingester: Ignore error, ret=%d", ret);
         }
-        st_usleep(SRS_AUTO_INGESTER_CIMS * 1000);
+
+        if (!trd->pull()) {
+            st_usleep(SRS_AUTO_INGESTER_CIMS * 1000);
+        }
     }
 
     return ret;

trunk/src/app/srs_app_kafka.cpp

@@ -501,7 +501,10 @@ int SrsKafkaProducer::cycle()
         if ((ret = do_cycle()) != ERROR_SUCCESS) {
             srs_warn("ignore kafka error. ret=%d", ret);
         }
-        st_usleep(SRS_KAKFA_CIMS * 1000);
+
+        if (!trd->pull()) {
+            st_usleep(SRS_KAKFA_CIMS * 1000);
+        }
     }
 
     return ret;

trunk/src/app/srs_app_ng_exec.cpp

@@ -86,7 +86,10 @@ int SrsNgExec::cycle()
         if ((ret = do_cycle()) != ERROR_SUCCESS) {
             srs_warn("EXEC: Ignore error, ret=%d", ret);
         }
-        st_usleep(SRS_RTMP_EXEC_CIMS * 1000);
+
+        if (!trd->pull()) {
+            st_usleep(SRS_RTMP_EXEC_CIMS * 1000);
+        }
     }
 
     std::vector<SrsProcess*>::iterator it;