For #1601, flush async on_dvr/on_hls events before stop. 3.0.118

trunk/src/app/srs_app_async_call.cpp

@@ -40,6 +40,7 @@ SrsAsyncCallWorker::SrsAsyncCallWorker()
 {
     trd = new SrsDummyCoroutine();
     wait = srs_cond_new();
+    lock = srs_mutex_new();
 }
 
 SrsAsyncCallWorker::~SrsAsyncCallWorker()
@@ -54,6 +55,7 @@ SrsAsyncCallWorker::~SrsAsyncCallWorker()
     tasks.clear();
     
     srs_cond_destroy(wait);
+    srs_mutex_destroy(lock);
 }
 
 srs_error_t SrsAsyncCallWorker::execute(ISrsAsyncCallTask* t)
@@ -87,6 +89,7 @@ srs_error_t SrsAsyncCallWorker::start()
 
 void SrsAsyncCallWorker::stop()
 {
+    flush_tasks();
     srs_cond_signal(wait);
     trd->stop();
 }
@@ -103,23 +106,40 @@ srs_error_t SrsAsyncCallWorker::cycle()
         if (tasks.empty()) {
             srs_cond_wait(wait);
         }
-        
-        std::vector<ISrsAsyncCallTask*> copy = tasks;
-        tasks.clear();
-        
-        std::vector<ISrsAsyncCallTask*>::iterator it;
-        for (it = copy.begin(); it != copy.end(); ++it) {
-            ISrsAsyncCallTask* task = *it;
-            
-            if ((err = task->call()) != srs_success) {
-                srs_warn("ignore task failed %s", srs_error_desc(err).c_str());
-                srs_freep(err);
-            }
-            srs_freep(task);
-        }
+
+        flush_tasks();
     }
     
     return err;
 }
 
+void SrsAsyncCallWorker::flush_tasks()
+{
+    srs_error_t err = srs_success;
+
+    // Avoid the async call blocking other coroutines.
+    std::vector<ISrsAsyncCallTask*> copy;
+    if (true) {
+        SrsLocker(lock);
+
+        if (tasks.empty()) {
+            return;
+        }
+
+        copy = tasks;
+        tasks.clear();
+    }
+
+    std::vector<ISrsAsyncCallTask*>::iterator it;
+    for (it = copy.begin(); it != copy.end(); ++it) {
+        ISrsAsyncCallTask* task = *it;
+
+        if ((err = task->call()) != srs_success) {
+            srs_warn("ignore task failed %s", srs_error_desc(err).c_str());
+            srs_freep(err);
+        }
+        srs_freep(task);
+    }
+}
+
 

trunk/src/app/srs_app_async_call.hpp

@@ -61,6 +61,7 @@ private:
 protected:
     std::vector<ISrsAsyncCallTask*> tasks;
     srs_cond_t wait;
+    srs_mutex_t lock;
 public:
     SrsAsyncCallWorker();
     virtual ~SrsAsyncCallWorker();
@@ -73,6 +74,8 @@ public:
 // Interface ISrsReusableThreadHandler
 public:
     virtual srs_error_t cycle();
+private:
+    virtual void flush_tasks();
 };
 
 #endif

trunk/src/app/srs_app_dvr.cpp

@@ -740,8 +740,6 @@ srs_error_t SrsDvrSessionPlan::on_publish()
 
 void SrsDvrSessionPlan::on_unpublish()
 {
-    SrsDvrPlan::on_unpublish();
-
     // support multiple publish.
     if (!dvr_enabled) {
         return;
@@ -754,6 +752,10 @@ void SrsDvrSessionPlan::on_unpublish()
     }
     
     dvr_enabled = false;
+
+    // We should notify the on_dvr, then stop the async.
+    // @see https://github.com/ossrs/srs/issues/1601
+    SrsDvrPlan::on_unpublish();
 }
 
 SrsDvrSegmentPlan::SrsDvrSegmentPlan()
@@ -815,14 +817,16 @@ void SrsDvrSegmentPlan::on_unpublish()
 {
     srs_error_t err = srs_success;
 
-    SrsDvrPlan::on_unpublish();
-
     if ((err = segment->close()) != srs_success) {
         srs_warn("ignore err %s", srs_error_desc(err).c_str());
         srs_freep(err);
     }
 
     dvr_enabled = false;
+
+    // We should notify the on_dvr, then stop the async.
+    // @see https://github.com/ossrs/srs/issues/1601
+    SrsDvrPlan::on_unpublish();
 }
 
 srs_error_t SrsDvrSegmentPlan::on_audio(SrsSharedPtrMessage* shared_audio, SrsFormat* format)

trunk/src/app/srs_app_rtmp_conn.cpp

@@ -406,7 +406,7 @@ srs_error_t SrsRtmpConn::service_cycle()
             rtmp->set_send_timeout(SRS_REPUBLISH_RECV_TIMEOUT);
             rtmp->set_recv_timeout(SRS_REPUBLISH_SEND_TIMEOUT);
             
-            srs_trace("rtmp: retry for republish");
+            srs_info("rtmp: retry for republish");
             srs_freep(err);
             continue;
         }