Modify flush method of filter graph

use av_buffersrc_close for better compliance with common
standards and resolve the frame loss issue when use fps
filter.

fix

Author: HuHeng

bmf/c_modules/include/ffmpeg_filter.h

@@ -38,6 +38,9 @@ class CFFFilter : public Module {
     std::vector<bool> in_eof_;
     std::vector<bool> out_eof_;
     std::vector<bool> out_eof_send_;
+    std::vector<bool> push_eof_;
+
+
     std::map<int, FilterConfig> config_;
     int64_t stream_start_time_;
     int64_t stream_first_dts_;
@@ -79,6 +82,7 @@ class CFFFilter : public Module {
 
     bmf_sdk::Packet convert_avframe_to_packet(AVFrame *frame, int index);
     bool check_finished();
+    bool check_input_finished();
     int get_cache_frame(int index, AVFrame *&frame, int &choose_index);
 };
 

bmf/c_modules/src/ffmpeg_filter.cpp

@@ -433,6 +433,8 @@ int CFFFilter::process_filter_graph(Task &task) {
                     ret = filter_graph_->push_frame(frame, choose_index);
                     if (frame) {
                         av_frame_free(&frame);
+                    } else if(choose_index >=0 && choose_index < num_input_streams_){
+                        push_eof_[choose_index] = true;
                     }
                     if (ret < 0) {
                         BMFLOG_NODE(BMF_INFO, node_id_) << "init push frame, ret: " << ret;
@@ -449,6 +451,7 @@ int CFFFilter::process_filter_graph(Task &task) {
             int ret =
                 filter_graph_->reap_filters(output_frames, push_frame_flag);
             push_frame_flag = 0;
+            int get_num = output_frames.size();
             for (auto output_frame : output_frames) {
                 for (int index = 0; index < output_frame.second.size();
                      index++) {
@@ -465,6 +468,14 @@ int CFFFilter::process_filter_graph(Task &task) {
                     task.fill_output_packet(output_frame.first, packet);
                 }
             }
+
+            if (check_input_finished() && get_num == 0) {
+                for (int i = 0; i < num_output_streams_; i++) {
+                    out_eof_[i] = true;
+                }
+                return 0;
+            }
+
             if (check_finished()) {
                 return 0;
             }
@@ -474,7 +485,7 @@ int CFFFilter::process_filter_graph(Task &task) {
             if (ret != AVERROR(EAGAIN) && ret < 0) {
                 return ret;
             }
-            int index = filter_graph_->get_the_most_failed_nb_request();
+            int index = filter_graph_->get_the_most_failed_nb_request(push_eof_);
             if (index >= 0) {
                 AVFrame *frame;
                 int choose_index;
@@ -491,13 +502,18 @@ int CFFFilter::process_filter_graph(Task &task) {
                     push_frame_flag = 1;
                     if (frame) {
                         av_frame_free(&frame);
+
+                    } else if(choose_index >=0 && choose_index < num_input_streams_){
+                        push_eof_[choose_index] = true;
                     }
+
                     if (ret < 0) {
                         BMFLOG_NODE(BMF_INFO, node_id_) << "push frame, choose_index: " << choose_index << " ret: " << ret;
                         return ret;
                     }
                 } else if (in_eof_[index]) {
                     filter_graph_->push_frame(NULL, index);
+                    push_eof_[index] = true;
                     push_frame_flag = 1;
                 } else {
                     break;
@@ -535,6 +551,15 @@ bool CFFFilter::check_finished() {
     return true;
 }
 
+bool CFFFilter::check_input_finished() {
+    for (int i = 0; i < num_input_streams_; i++) {
+        if (!in_eof_[i] || !push_eof_[i]) {
+            return false;
+        }
+    }
+    return true;
+}
+
 int CFFFilter::process(Task &task) {
     std::lock_guard<std::mutex> lk(cache_mutex_);
 
@@ -558,6 +583,11 @@ int CFFFilter::process(Task &task) {
                 in_eof_.push_back(false);
             }
         }
+        if (push_eof_.size() == 0) {
+            for (int i = 0; i < num_input_streams_; i++) {
+                push_eof_.push_back(false);
+            }
+        }
     }
     if (!num_output_streams_) {
         num_output_streams_ = task.get_outputs().size();
@@ -642,6 +672,7 @@ int CFFFilter::process(Task &task) {
 int CFFFilter::reset() {
     for (int i = 0; i < num_input_streams_; i++) {
         in_eof_[i] = false;
+        push_eof_[i] = false;
     }
     for (int i = 0; i < num_output_streams_; i++) {
         out_eof_[i] = false;

bmf/sdk/cpp_sdk/include/bmf/sdk/filter_graph.h

@@ -62,6 +62,7 @@ class FilterGraph {
     std::map<int, FilterConfig> in_configs_;
     std::map<int, FilterConfig> out_configs_;
     bool b_init_;
+    std::map<int, int64_t> next_pts_map;
 
   public:
     AVFilterGraph *filter_graph_;
@@ -439,8 +440,18 @@ class FilterGraph {
             return -1;
         }
 
-        ret = av_buffersrc_add_frame_flags(buffer_src_ctx_[in_idx], frame,
+        if (frame == NULL && next_pts_map.find(in_idx) != next_pts_map.end()) {
+            ret = av_buffersrc_close(buffer_src_ctx_[in_idx], next_pts_map[in_idx], AV_BUFFERSRC_FLAG_PUSH);
+
+        } else {
+            ret = av_buffersrc_add_frame_flags(buffer_src_ctx_[in_idx], frame,
                                            AV_BUFFERSRC_FLAG_PUSH);
+            if (frame) {
+                int64_t duration = frame->pkt_duration > 0 ? frame->pkt_duration : 1;
+                next_pts_map[in_idx] = frame->pts + duration;
+            }
+        }
+
         if (ret < 0) {
             av_log(NULL, AV_LOG_ERROR, "Error while feeding the filtergraph\n");
             return ret;
@@ -536,10 +547,13 @@ class FilterGraph {
         return 0;
     };
 
-    int get_the_most_failed_nb_request() {
-        int index = -1;
-        int most_failed_request = 0;
+    int get_the_most_failed_nb_request(std::vector<bool>& push_eofs) {
+        int index = 0;
+        int most_failed_request = -1;
         for (int i = 0; i < buffer_src_ctx_.size(); i++) {
+            if (push_eofs[i]) {
+                continue;
+            }
             int failed_request =
                 av_buffersrc_get_nb_failed_requests(buffer_src_ctx_[i]);
             if (failed_request > most_failed_request) {
@@ -603,9 +617,18 @@ class FilterGraph {
     int push_frame(AVFrame *frame, int index) {
         int ret = 0;
         if ((ret = check_input_property(frame, index)) >= 0) {
-            ret = av_buffersrc_add_frame_flags(buffer_src_ctx_[index], frame,
-                                               AV_BUFFERSRC_FLAG_PUSH |
-                                                   AV_BUFFERSRC_FLAG_KEEP_REF);
+            if (frame == NULL && next_pts_map.find(index) != next_pts_map.end()) {
+                ret = av_buffersrc_close(buffer_src_ctx_[index], next_pts_map[index], AV_BUFFERSRC_FLAG_PUSH | AV_BUFFERSRC_FLAG_KEEP_REF);
+
+            } else {
+                ret = av_buffersrc_add_frame_flags(buffer_src_ctx_[index], frame,
+                                                   AV_BUFFERSRC_FLAG_PUSH |
+                                                       AV_BUFFERSRC_FLAG_KEEP_REF);
+                if (frame) {
+                    int64_t duration = frame->pkt_duration > 0 ? frame->pkt_duration : 1;
+                    next_pts_map[index] = frame->pts + duration;
+                }
+            }
             if (ret < 0) {
                 if (ret != AVERROR_EOF) {
                     BMFLOG(BMF_ERROR) << "add frame error: " << ret;