diff mbox series

[FFmpeg-devel,06/10] lavfi/dnn_backend_tf: Request-based Execution

Message ID 20210528092454.31874-6-shubhanshu.e01@gmail.com
State Accepted
Headers show
Series [FFmpeg-devel,01/10] lavfi/dnn: Extract TaskItem and InferenceItem from OpenVino Backend
Related show

Checks

Context Check Description
andriy/x86_make success Make finished
andriy/x86_make_fate success Make fate finished
andriy/PPC64_make success Make finished
andriy/PPC64_make_fate success Make fate finished

Commit Message

Shubhanshu Saxena May 28, 2021, 9:24 a.m. UTC
This commit adds RequestItem and rearranges the existing sync
execution mechanism to use request-based execution. It will help
in adding async functionality to the TensorFlow backend later.

Signed-off-by: Shubhanshu Saxena <shubhanshu.e01@gmail.com>
---
 libavfilter/dnn/dnn_backend_tf.c | 297 +++++++++++++++++++++----------
 1 file changed, 206 insertions(+), 91 deletions(-)
diff mbox series

Patch

diff --git a/libavfilter/dnn/dnn_backend_tf.c b/libavfilter/dnn/dnn_backend_tf.c
index 4c16c2bdb0..793b108e55 100644
--- a/libavfilter/dnn/dnn_backend_tf.c
+++ b/libavfilter/dnn/dnn_backend_tf.c
@@ -35,10 +35,13 @@ 
 #include "dnn_backend_native_layer_maximum.h"
 #include "dnn_io_proc.h"
 #include "dnn_backend_common.h"
+#include "safe_queue.h"
+#include "queue.h"
 #include <tensorflow/c/c_api.h>
 
 typedef struct TFOptions{
     char *sess_config;
+    uint32_t nireq;
 } TFOptions;
 
 typedef struct TFContext {
@@ -52,26 +55,79 @@  typedef struct TFModel{
     TF_Graph *graph;
     TF_Session *session;
     TF_Status *status;
+    SafeQueue *request_queue;
+    Queue *inference_queue;
 } TFModel;
 
+typedef struct tf_infer_request {
+    TF_Output *tf_outputs;
+    TF_Tensor **output_tensors;
+    TF_Output *tf_input;
+    TF_Tensor *input_tensor;
+} tf_infer_request;
+
+typedef struct RequestItem {
+    tf_infer_request *infer_request;
+    InferenceItem *inference;
+    // further properties will be added later for async
+} RequestItem;
+
 #define OFFSET(x) offsetof(TFContext, x)
 #define FLAGS AV_OPT_FLAG_FILTERING_PARAM
 static const AVOption dnn_tensorflow_options[] = {
     { "sess_config", "config for SessionOptions", OFFSET(options.sess_config), AV_OPT_TYPE_STRING, { .str = NULL }, 0, 0, FLAGS },
+    { "nireq",  "number of request",   OFFSET(options.nireq),       AV_OPT_TYPE_INT,    { .i64 = 0 },     0, INT_MAX, FLAGS },
     { NULL }
 };
 
 AVFILTER_DEFINE_CLASS(dnn_tensorflow);
 
-static DNNReturnType execute_model_tf(const DNNModel *model, const char *input_name, AVFrame *in_frame,
-                                      const char **output_names, uint32_t nb_output, AVFrame *out_frame,
-                                      int do_ioproc);
+static DNNReturnType execute_model_tf(RequestItem *request, Queue *inference_queue);
 
 static void free_buffer(void *data, size_t length)
 {
     av_freep(&data);
 }
 
+static void tf_free_request(tf_infer_request *request)
+{
+    if (!request)
+        return;
+    if (request->input_tensor) {
+        TF_DeleteTensor(request->input_tensor);
+        request->input_tensor = NULL;
+    }
+    av_freep(&request->tf_input);
+    av_freep(&request->tf_outputs);
+    av_freep(&request->output_tensors);
+}
+
+static tf_infer_request* tf_create_inference_request(void)
+{
+    tf_infer_request* infer_request = av_malloc(sizeof(tf_infer_request));
+    infer_request->tf_outputs = NULL;
+    infer_request->tf_input = NULL;
+    infer_request->input_tensor = NULL;
+    infer_request->output_tensors = NULL;
+    return infer_request;
+}
+
+static DNNReturnType extract_inference_from_task(TaskItem *task, Queue *inference_queue)
+{
+    InferenceItem *inference = av_malloc(sizeof(*inference));
+    if (!inference) {
+        return DNN_ERROR;
+    }
+    task->inference_todo = 1;
+    task->inference_done = 0;
+    inference->task = task;
+    if (ff_queue_push_back(inference_queue, inference) < 0) {
+        av_freep(&inference);
+        return DNN_ERROR;
+    }
+    return DNN_SUCCESS;
+}
+
 static TF_Buffer *read_graph(const char *model_filename)
 {
     TF_Buffer *graph_buf;
@@ -171,6 +227,8 @@  static DNNReturnType get_output_tf(void *model, const char *input_name, int inpu
     TFContext *ctx = &tf_model->ctx;
     AVFrame *in_frame = av_frame_alloc();
     AVFrame *out_frame = NULL;
+    TaskItem task;
+    RequestItem *request;
 
     if (!in_frame) {
         av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for input frame\n");
@@ -187,7 +245,27 @@  static DNNReturnType get_output_tf(void *model, const char *input_name, int inpu
     in_frame->width = input_width;
     in_frame->height = input_height;
 
-    ret = execute_model_tf(tf_model->model, input_name, in_frame, &output_name, 1, out_frame, 0);
+    task.do_ioproc = 0;
+    task.async = 0;
+    task.input_name = input_name;
+    task.in_frame = in_frame;
+    task.output_names = &output_name;
+    task.out_frame = out_frame;
+    task.model = tf_model;
+    task.nb_output = 1;
+
+    if (extract_inference_from_task(&task, tf_model->inference_queue) != DNN_SUCCESS) {
+        av_log(ctx, AV_LOG_ERROR, "unable to extract inference from task.\n");
+        return DNN_ERROR;
+    }
+
+    request = ff_safe_queue_pop_front(tf_model->request_queue);
+    if (!request) {
+        av_log(ctx, AV_LOG_ERROR, "unable to get infer request.\n");
+        return DNN_ERROR;
+    }
+
+    ret = execute_model_tf(request, tf_model->inference_queue);
     *output_width = out_frame->width;
     *output_height = out_frame->height;
 
@@ -691,6 +769,7 @@  DNNModel *ff_dnn_load_model_tf(const char *model_filename, DNNFunctionType func_
 {
     DNNModel *model = NULL;
     TFModel *tf_model = NULL;
+    TFContext *ctx = NULL;
 
     model = av_mallocz(sizeof(DNNModel));
     if (!model){
@@ -704,10 +783,11 @@  DNNModel *ff_dnn_load_model_tf(const char *model_filename, DNNFunctionType func_
     }
     tf_model->ctx.class = &dnn_tensorflow_class;
     tf_model->model = model;
+    ctx = &tf_model->ctx;
 
     //parse options
-    av_opt_set_defaults(&tf_model->ctx);
-    if (av_opt_set_from_string(&tf_model->ctx, options, NULL, "=", "&") < 0) {
+    av_opt_set_defaults(&ctx);
+    if (av_opt_set_from_string(&ctx, options, NULL, "=", "&") < 0) {
         av_log(&tf_model->ctx, AV_LOG_ERROR, "Failed to parse options \"%s\"\n", options);
         av_freep(&tf_model);
         av_freep(&model);
@@ -723,6 +803,19 @@  DNNModel *ff_dnn_load_model_tf(const char *model_filename, DNNFunctionType func_
         }
     }
 
+    if (ctx->options.nireq <= 0) {
+        ctx->options.nireq = av_cpu_count() / 2 + 1;
+    }
+
+    tf_model->request_queue = ff_safe_queue_create();
+
+    for (int i = 0; i < ctx->options.nireq; i++) {
+        RequestItem *item = av_mallocz(sizeof(*item));
+        item->infer_request = tf_create_inference_request();
+        ff_safe_queue_push_back(tf_model->request_queue, item);
+    }
+
+    tf_model->inference_queue = ff_queue_create();
     model->model = tf_model;
     model->get_input = &get_input_tf;
     model->get_output = &get_output_tf;
@@ -733,168 +826,176 @@  DNNModel *ff_dnn_load_model_tf(const char *model_filename, DNNFunctionType func_
     return model;
 }
 
-static DNNReturnType execute_model_tf(const DNNModel *model, const char *input_name, AVFrame *in_frame,
-                                      const char **output_names, uint32_t nb_output, AVFrame *out_frame,
-                                      int do_ioproc)
+static DNNReturnType execute_model_tf(RequestItem *request, Queue *inference_queue)
 {
-    TF_Output *tf_outputs;
-    TFModel *tf_model = model->model;
-    TFContext *ctx = &tf_model->ctx;
+    TFModel *tf_model;
+    TFContext *ctx;
+    tf_infer_request *infer_request;
+    InferenceItem *inference;
+    TaskItem *task;
     DNNData input, *outputs;
-    TF_Tensor **output_tensors;
-    TF_Output tf_input;
-    TF_Tensor *input_tensor;
 
-    if (get_input_tf(tf_model, &input, input_name) != DNN_SUCCESS)
+    inference = ff_queue_pop_front(inference_queue);
+    av_assert0(inference);
+    task = inference->task;
+    tf_model = task->model;
+    ctx = &tf_model->ctx;
+    request->inference = inference;
+
+    if (get_input_tf(tf_model, &input, task->input_name) != DNN_SUCCESS)
         return DNN_ERROR;
-    input.height = in_frame->height;
-    input.width = in_frame->width;
 
-    tf_input.oper = TF_GraphOperationByName(tf_model->graph, input_name);
-    if (!tf_input.oper){
+    infer_request = request->infer_request;
+    input.height = task->in_frame->height;
+    input.width = task->in_frame->width;
+
+    infer_request->tf_input = av_malloc(sizeof(TF_Output));
+    infer_request->tf_input->oper = TF_GraphOperationByName(tf_model->graph, task->input_name);
+    if (!infer_request->tf_input->oper){
         av_log(ctx, AV_LOG_ERROR, "Could not find \"%s\" in model\n", input_name);
         return DNN_ERROR;
     }
-    tf_input.index = 0;
-    input_tensor = allocate_input_tensor(&input);
-    if (!input_tensor){
+    infer_request->tf_input->index = 0;
+    infer_request->input_tensor = allocate_input_tensor(&input);
+    if (!infer_request->input_tensor){
         av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for input tensor\n");
         return DNN_ERROR;
     }
-    input.data = (float *)TF_TensorData(input_tensor);
+    input.data = (float *)TF_TensorData(infer_request->input_tensor);
 
     switch (tf_model->model->func_type) {
     case DFT_PROCESS_FRAME:
-        if (do_ioproc) {
+        if (task->do_ioproc) {
             if (tf_model->model->frame_pre_proc != NULL) {
-                tf_model->model->frame_pre_proc(in_frame, &input, tf_model->model->filter_ctx);
+                tf_model->model->frame_pre_proc(task->in_frame, &input, tf_model->model->filter_ctx);
             } else {
-                ff_proc_from_frame_to_dnn(in_frame, &input, ctx);
+                ff_proc_from_frame_to_dnn(task->in_frame, &input, ctx);
             }
         }
         break;
     case DFT_ANALYTICS_DETECT:
-        ff_frame_to_dnn_detect(in_frame, &input, ctx);
+        ff_frame_to_dnn_detect(task->in_frame, &input, ctx);
         break;
     default:
         avpriv_report_missing_feature(ctx, "model function type %d", tf_model->model->func_type);
         break;
     }
 
-    tf_outputs = av_malloc_array(nb_output, sizeof(*tf_outputs));
-    if (tf_outputs == NULL) {
-        TF_DeleteTensor(input_tensor);
-        av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for *tf_outputs\n"); \
+    infer_request->tf_outputs = av_malloc_array(task->nb_output, sizeof(TF_Output));
+    if (infer_request->tf_outputs == NULL) {
+        av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for *tf_outputs\n");
         return DNN_ERROR;
     }
 
-    output_tensors = av_mallocz_array(nb_output, sizeof(*output_tensors));
-    if (!output_tensors) {
-        TF_DeleteTensor(input_tensor);
-        av_freep(&tf_outputs);
-        av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for output tensor\n"); \
+    infer_request->output_tensors = av_mallocz_array(task->nb_output, sizeof(*infer_request->output_tensors));
+    if (!infer_request->output_tensors) {
+        av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for output tensor\n");
         return DNN_ERROR;
     }
 
-    for (int i = 0; i < nb_output; ++i) {
-        tf_outputs[i].oper = TF_GraphOperationByName(tf_model->graph, output_names[i]);
-        if (!tf_outputs[i].oper) {
-            TF_DeleteTensor(input_tensor);
-            av_freep(&tf_outputs);
-            av_freep(&output_tensors);
-            av_log(ctx, AV_LOG_ERROR, "Could not find output \"%s\" in model\n", output_names[i]); \
+    for (int i = 0; i < task->nb_output; ++i) {
+        infer_request->tf_outputs[i].oper = TF_GraphOperationByName(tf_model->graph, task->output_names[i]);
+        if (!infer_request->tf_outputs[i].oper) {
+            av_log(ctx, AV_LOG_ERROR, "Could not find output \"%s\" in model\n", task->output_names[i]);
             return DNN_ERROR;
         }
-        tf_outputs[i].index = 0;
+        infer_request->tf_outputs[i].index = 0;
     }
 
     TF_SessionRun(tf_model->session, NULL,
-                  &tf_input, &input_tensor, 1,
-                  tf_outputs, output_tensors, nb_output,
-                  NULL, 0, NULL, tf_model->status);
+                    infer_request->tf_input, &infer_request->input_tensor, 1,
+                    infer_request->tf_outputs, infer_request->output_tensors,
+                    task->nb_output, NULL, 0, NULL,
+                    tf_model->status);
     if (TF_GetCode(tf_model->status) != TF_OK) {
-        TF_DeleteTensor(input_tensor);
-        av_freep(&tf_outputs);
-        av_freep(&output_tensors);
-        av_log(ctx, AV_LOG_ERROR, "Failed to run session when executing model\n");
-        return DNN_ERROR;
+            tf_free_request(infer_request);
+            av_log(ctx, AV_LOG_ERROR, "Failed to run session when executing model\n");
+            return DNN_ERROR;
     }
 
-    outputs = av_malloc_array(nb_output, sizeof(*outputs));
+    outputs = av_malloc_array(task->nb_output, sizeof(*outputs));
     if (!outputs) {
-        TF_DeleteTensor(input_tensor);
-        av_freep(&tf_outputs);
-        av_freep(&output_tensors);
-        av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for *outputs\n"); \
+        tf_free_request(infer_request);
+        av_log(ctx, AV_LOG_ERROR, "Failed to allocate memory for *outputs\n");
         return DNN_ERROR;
     }
 
-    for (uint32_t i = 0; i < nb_output; ++i) {
-        outputs[i].height = TF_Dim(output_tensors[i], 1);
-        outputs[i].width = TF_Dim(output_tensors[i], 2);
-        outputs[i].channels = TF_Dim(output_tensors[i], 3);
-        outputs[i].data = TF_TensorData(output_tensors[i]);
-        outputs[i].dt = TF_TensorType(output_tensors[i]);
+    for (uint32_t i = 0; i < task->nb_output; ++i) {
+        outputs[i].height = TF_Dim(infer_request->output_tensors[i], 1);
+        outputs[i].width = TF_Dim(infer_request->output_tensors[i], 2);
+        outputs[i].channels = TF_Dim(infer_request->output_tensors[i], 3);
+        outputs[i].data = TF_TensorData(infer_request->output_tensors[i]);
+        outputs[i].dt = TF_TensorType(infer_request->output_tensors[i]);
     }
-    switch (model->func_type) {
+    switch (tf_model->model->func_type) {
     case DFT_PROCESS_FRAME:
         //it only support 1 output if it's frame in & frame out
-        if (do_ioproc) {
+        if (task->do_ioproc) {
             if (tf_model->model->frame_post_proc != NULL) {
-                tf_model->model->frame_post_proc(out_frame, outputs, tf_model->model->filter_ctx);
+                tf_model->model->frame_post_proc(task->out_frame, outputs, tf_model->model->filter_ctx);
             } else {
-                ff_proc_from_dnn_to_frame(out_frame, outputs, ctx);
+                ff_proc_from_dnn_to_frame(task->out_frame, outputs, ctx);
             }
         } else {
-            out_frame->width = outputs[0].width;
-            out_frame->height = outputs[0].height;
+            task->out_frame->width = outputs[0].width;
+            task->out_frame->height = outputs[0].height;
         }
         break;
     case DFT_ANALYTICS_DETECT:
-        if (!model->detect_post_proc) {
+        if (!tf_model->model->detect_post_proc) {
             av_log(ctx, AV_LOG_ERROR, "Detect filter needs provide post proc\n");
             return DNN_ERROR;
         }
-        model->detect_post_proc(out_frame, outputs, nb_output, model->filter_ctx);
+        tf_model->model->detect_post_proc(task->out_frame, outputs, task->nb_output, tf_model->model->filter_ctx);
         break;
     default:
-        for (uint32_t i = 0; i < nb_output; ++i) {
-            if (output_tensors[i]) {
-                TF_DeleteTensor(output_tensors[i]);
+        for (uint32_t i = 0; i < task->nb_output; ++i) {
+            if (infer_request->output_tensors[i]) {
+                TF_DeleteTensor(infer_request->output_tensors[i]);
             }
         }
-        TF_DeleteTensor(input_tensor);
-        av_freep(&output_tensors);
-        av_freep(&tf_outputs);
-        av_freep(&outputs);
-
         av_log(ctx, AV_LOG_ERROR, "Tensorflow backend does not support this kind of dnn filter now\n");
         return DNN_ERROR;
     }
-
-    for (uint32_t i = 0; i < nb_output; ++i) {
-        if (output_tensors[i]) {
-            TF_DeleteTensor(output_tensors[i]);
+    for (uint32_t i = 0; i < task->nb_output; ++i) {
+        if (infer_request->output_tensors[i]) {
+            TF_DeleteTensor(infer_request->output_tensors[i]);
         }
     }
-    TF_DeleteTensor(input_tensor);
-    av_freep(&output_tensors);
-    av_freep(&tf_outputs);
+    task->inference_done++;
+    tf_free_request(infer_request);
     av_freep(&outputs);
-    return DNN_SUCCESS;
+    ff_safe_queue_push_back(tf_model->request_queue, request);
+    return (task->inference_done == task->inference_todo) ? DNN_SUCCESS : DNN_ERROR;
 }
 
 DNNReturnType ff_dnn_execute_model_tf(const DNNModel *model, DNNExecBaseParams *exec_params)
 {
     TFModel *tf_model = model->model;
     TFContext *ctx = &tf_model->ctx;
+    TaskItem task;
+    RequestItem *request;
 
     if (ff_check_exec_params(ctx, DNN_TF, model->func_type, exec_params) != 0) {
-         return DNN_ERROR;
+        return DNN_ERROR;
+    }
+
+    if (ff_dnn_fill_task(&task, exec_params, tf_model, 0, 1) != DNN_SUCCESS) {
+        return DNN_ERROR;
+    }
+
+    if (extract_inference_from_task(&task, tf_model->inference_queue) != DNN_SUCCESS) {
+        av_log(ctx, AV_LOG_ERROR, "unable to extract inference from task.\n");
+        return DNN_ERROR;
     }
 
-    return execute_model_tf(model, exec_params->input_name, exec_params->in_frame,
-                            exec_params->output_names, exec_params->nb_output, exec_params->out_frame, 1);
+    request = ff_safe_queue_pop_front(tf_model->request_queue);
+    if (!request) {
+        av_log(ctx, AV_LOG_ERROR, "unable to get infer request.\n");
+        return DNN_ERROR;
+    }
+
+    return execute_model_tf(request, tf_model->inference_queue);
 }
 
 void ff_dnn_free_model_tf(DNNModel **model)
@@ -903,6 +1004,20 @@  void ff_dnn_free_model_tf(DNNModel **model)
 
     if (*model){
         tf_model = (*model)->model;
+        while (ff_safe_queue_size(tf_model->request_queue) != 0) {
+            RequestItem *item = ff_safe_queue_pop_front(tf_model->request_queue);
+            tf_free_request(item->infer_request);
+            av_freep(&item->infer_request);
+            av_freep(&item);
+        }
+        ff_safe_queue_destroy(tf_model->request_queue);
+
+        while (ff_queue_size(tf_model->inference_queue) != 0) {
+            InferenceItem *item = ff_queue_pop_front(tf_model->inference_queue);
+            av_freep(&item);
+        }
+        ff_queue_destroy(tf_model->inference_queue);
+
         if (tf_model->graph){
             TF_DeleteGraph(tf_model->graph);
         }