async call worker fast execute tasks.

@@ -257,7 +257,9 @@ echo "\"" >> $SRS_AUTO_HEADERS_H
 # new empty line to auto headers file.
 # new empty line to auto headers file.
 echo "" >> $SRS_AUTO_HEADERS_H
 echo "" >> $SRS_AUTO_HEADERS_H
 
 
+#####################################################################################
 # auto header EOF.
 # auto header EOF.
+#####################################################################################
 echo "#endif" >> $SRS_AUTO_HEADERS_H
 echo "#endif" >> $SRS_AUTO_HEADERS_H
 echo "" >> $SRS_AUTO_HEADERS_H
 echo "" >> $SRS_AUTO_HEADERS_H
 
 

@@ -42,6 +42,7 @@ ISrsAsyncCallTask::~ISrsAsyncCallTask()
 SrsAsyncCallWorker::SrsAsyncCallWorker()
 SrsAsyncCallWorker::SrsAsyncCallWorker()
 {
 {
     pthread = new SrsReusableThread("async", this, SRS_AUTO_ASYNC_CALLBACL_SLEEP_US);
     pthread = new SrsReusableThread("async", this, SRS_AUTO_ASYNC_CALLBACL_SLEEP_US);
+    wait = st_cond_new();
 }
 }
 
 
 SrsAsyncCallWorker::~SrsAsyncCallWorker()
 SrsAsyncCallWorker::~SrsAsyncCallWorker()
@@ -54,6 +55,8 @@ SrsAsyncCallWorker::~SrsAsyncCallWorker()
         srs_freep(task);
         srs_freep(task);
     }
     }
     tasks.clear();
     tasks.clear();
+
+    st_cond_destroy(wait);
 }
 }
 
 
 int SrsAsyncCallWorker::execute(ISrsAsyncCallTask* t)
 int SrsAsyncCallWorker::execute(ISrsAsyncCallTask* t)
@@ -61,10 +64,16 @@ int SrsAsyncCallWorker::execute(ISrsAsyncCallTask* t)
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
 
 
     tasks.push_back(t);
     tasks.push_back(t);
+    st_cond_signal(wait);
 
 
     return ret;
     return ret;
 }
 }
 
 
+int SrsAsyncCallWorker::count()
+{
+    return (int)tasks.size();
+}
+
 int SrsAsyncCallWorker::start()
 int SrsAsyncCallWorker::start()
 {
 {
     return pthread->start();
     return pthread->start();
@@ -72,23 +81,30 @@ int SrsAsyncCallWorker::start()
 
 
 void SrsAsyncCallWorker::stop()
 void SrsAsyncCallWorker::stop()
 {
 {
+    st_cond_signal(wait);
     pthread->stop();
     pthread->stop();
 }
 }
 
 
 int SrsAsyncCallWorker::cycle()
 int SrsAsyncCallWorker::cycle()
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;
-    
-    std::vector<ISrsAsyncCallTask*> copies = tasks;
-    tasks.clear();
 
 
-    std::vector<ISrsAsyncCallTask*>::iterator it;
-    for (it = copies.begin(); it != copies.end(); ++it) {
-        ISrsAsyncCallTask* task = *it;
-        if ((ret = task->call()) != ERROR_SUCCESS) {
-            srs_warn("ignore async callback %s, ret=%d", task->to_string().c_str(), ret);
+    while (pthread->can_loop()) {
+        if (tasks.empty()) {
+            st_cond_wait(wait);
+        }
+
+        std::vector<ISrsAsyncCallTask*> copies = tasks;
+        tasks.clear();
+
+        std::vector<ISrsAsyncCallTask*>::iterator it;
+        for (it = copies.begin(); it != copies.end(); ++it) {
+            ISrsAsyncCallTask* task = *it;
+            if ((ret = task->call()) != ERROR_SUCCESS) {
+                srs_warn("ignore async callback %s, ret=%d", task->to_string().c_str(), ret);
+            }
+            srs_freep(task);
         }
         }
-        srs_freep(task);
     }
     }
 
 
     return ret;
     return ret;

@@ -70,12 +70,15 @@ class SrsAsyncCallWorker : public ISrsReusableThreadHandler
 {
 {
 private:
 private:
     SrsReusableThread* pthread;
     SrsReusableThread* pthread;
+protected:
     std::vector<ISrsAsyncCallTask*> tasks;
     std::vector<ISrsAsyncCallTask*> tasks;
+    st_cond_t wait;
 public:
 public:
     SrsAsyncCallWorker();
     SrsAsyncCallWorker();
     virtual ~SrsAsyncCallWorker();
     virtual ~SrsAsyncCallWorker();
 public:
 public:
     virtual int execute(ISrsAsyncCallTask* t);
     virtual int execute(ISrsAsyncCallTask* t);
+    virtual int count();
 public:
 public:
     virtual int start();
     virtual int start();
     virtual void stop();
     virtual void stop();