[C-Api] async invoke feature

Add new callback and options to get the inference result asynchronously.

Signed-off-by: Jaeyun Jung <jy1210.jung@samsung.com>

Author: jaeyun-jung

c/include/nnstreamer-tizen-internal.h

@@ -19,6 +19,17 @@
 extern "C" {
 #endif /* __cplusplus */
 
+/**
+ * @brief Callback for tensor data stream of machine-learning API.
+ * @details Note that the buffer may be deallocated after the return and this is synchronously called. Thus, if you need the data afterwards, copy the data to another buffer and return fast. Do not spend too much time in the callback.
+ * @since_tizen 10.0
+ * @remarks The @a data can be used only in the callback. To use outside, make a copy.
+ * @param[in] data The handle of the tensor data (a single frame. tensor/tensors). You can get the information of given tensor data frame using ml_tensors_data_get_info().
+ * @param[in,out] user_data User application's private data.
+ * @return @c 0 on success. Otherwise a negative error value.
+ */
+typedef int (*ml_tensors_data_cb) (const ml_tensors_data_h data, void *user_data);
+
 /**
  * @brief Constructs the pipeline (GStreamer + NNStreamer).
  * @details This function is to construct the pipeline without checking the permission in platform internally. See ml_pipeline_construct() for the details.
@@ -38,7 +49,9 @@ typedef struct {
   char *custom_option;           /**< Custom option string for neural network framework. */
   char *fw_name;                 /**< The explicit framework name given by user */
   int invoke_dynamic;            /**< True for supporting invoke with flexible output. */
-  int invoke_async;              /**< The sub-plugin must support asynchronous output to use this option. If set to TRUE, the sub-plugin can generate multiple outputs asynchronously per single input. Otherwise, only synchronous single-output is expected and async callback/handle are ignored. */
+  int invoke_async;              /**< The sub-plugin must support asynchronous output to use this option. If set to TRUE, the sub-plugin can generate multiple outputs asynchronously per single input. Otherwise, only synchronous single-output is expected and async callback is ignored. */
+  ml_tensors_data_cb invoke_async_cb; /**< Callback function to be called when the sub-plugin generates an output asynchronously. This is only available when invoke_async is set to TRUE. */
+  void *invoke_async_pdata;           /**< Private data to be passed to async callback. */
 } ml_single_preset;
 
 /**

c/src/ml-api-inference-single.c

@@ -143,6 +143,8 @@ typedef struct
   GList *destroy_data_list;           /**< data to be freed by filter */
   gboolean invoke_dynamic;            /**< true to invoke flexible tensor */
   gboolean invoke_async;              /**< true to invoke and return result asynchronously */
+  ml_tensors_data_cb invoke_async_cb; /**< Callback function to be called when the sub-plugin generates an output asynchronously. */
+  void *invoke_async_pdata;           /**< Private data to be passed to async callback. */
 } ml_single;
 
 /**
@@ -568,6 +570,72 @@ invoke_thread (void *arg)
   return NULL;
 }
 
+/**
+ * @brief Internal function to get the asynchronous invoke.
+ */
+static int
+ml_single_async_cb (GstTensorMemory * data, GstTensorsInfo * info,
+    void *user_data)
+{
+  ml_single_h single = (ml_single_h) user_data;
+  ml_single *single_h;
+  ml_tensors_info_h _info = NULL;
+  ml_tensors_data_h _data = NULL;
+  unsigned int i;
+  int ret = ML_ERROR_NONE;
+
+  ML_SINGLE_GET_VALID_HANDLE_LOCKED (single_h, single, 0);
+
+  if (!single_h->invoke_async_cb) {
+    /* No callback, do nothing. Internal state changing? */
+    goto done;
+  }
+
+  ret = _ml_tensors_info_create_from_gst (&_info, info);
+  if (ret != ML_ERROR_NONE) {
+    _ml_error_report
+        ("Cannot handle tensor data stream. Failed to create ml information.");
+    goto done;
+  }
+
+  ret = ml_tensors_data_create (_info, &_data);
+  if (ret != ML_ERROR_NONE) {
+    _ml_error_report
+        ("Cannot handle tensor data stream. Failed to create ml data.");
+    goto done;
+  }
+
+  for (i = 0; i < info->num_tensors; ++i) {
+    ret = ml_tensors_data_set_tensor_data (_data, i,
+        data[i].data, data[i].size);
+    if (ret != ML_ERROR_NONE) {
+      _ml_error_report
+          ("Cannot handle tensor data stream. Failed to update ml data of index %u, size is %zu.",
+          i, data[i].size);
+      goto done;
+    }
+  }
+
+  ret = single_h->invoke_async_cb (_data, single_h->invoke_async_pdata);
+  if (ret != ML_ERROR_NONE) {
+    _ml_error_report
+        ("Cannot handle tensor data stream. The callback function returns error '%d'.",
+        ret);
+  }
+
+done:
+  if (_info) {
+    ml_tensors_info_destroy (_info);
+  }
+
+  if (_data) {
+    ml_tensors_data_destroy (_data);
+  }
+
+  ML_SINGLE_HANDLE_UNLOCK (single_h);
+  return (ret == ML_ERROR_NONE) ? 0 : -1;
+}
+
 /**
  * @brief Sets the information (tensor dimension, type, name and so on) of required input data for the given model, and get updated output data information.
  * @details Note that a model/framework may not support setting such information.
@@ -922,6 +990,10 @@ _ml_single_open_custom_validate_arguments (ml_single_h * single,
     _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
         "The parameter, 'info' (ml_single_preset *), is not valid. Its models entry if NULL (info->models is NULL).");
 
+  if (info->invoke_async && !info->invoke_async_cb)
+    _ml_error_report_return (ML_ERROR_INVALID_PARAMETER,
+        "The parameter, 'info' (ml_single_preset *), is not valid. It has 'invoke_async' entry but its callback 'invoke_async_cb' is NULL");
+
   return ML_ERROR_NONE;
 }
 
@@ -1016,6 +1088,8 @@ ml_single_open_custom (ml_single_h * single, ml_single_preset * info)
 
   single_h->invoke_dynamic = info->invoke_dynamic;
   single_h->invoke_async = info->invoke_async;
+  single_h->invoke_async_cb = info->invoke_async_cb;
+  single_h->invoke_async_pdata = info->invoke_async_pdata;
 
   filter_obj = G_OBJECT (single_h->filter);
 
@@ -1095,6 +1169,12 @@ ml_single_open_custom (ml_single_h * single, ml_single_preset * info)
     g_object_set (filter_obj, "custom", info->custom_option, NULL);
   }
 
+  /* Set async callback. */
+  if (single_h->invoke_async) {
+    single_h->klass->set_invoke_async_callback (single_h->filter,
+        ml_single_async_cb, single_h);
+  }
+
   /* 4. Start the nnfw to get inout configurations if needed */
   if (!single_h->klass->start (single_h->filter)) {
     _ml_error_report
@@ -1232,6 +1312,12 @@ ml_single_open_with_option (ml_single_h * single, const ml_option_h option)
     if (g_ascii_strcasecmp ((gchar *) value, "true") == 0)
       info.invoke_async = TRUE;
   }
+  if (ML_ERROR_NONE == ml_option_get (option, "async_callback", &value)) {
+    info.invoke_async_cb = (ml_tensors_data_cb) value;
+  }
+  if (ML_ERROR_NONE == ml_option_get (option, "async_data", &value)) {
+    info.invoke_async_pdata = value;
+  }
 
   return ml_single_open_custom (single, &info);
 }
@@ -1259,6 +1345,9 @@ ml_single_close (ml_single_h single)
 
   ML_SINGLE_GET_VALID_HANDLE_LOCKED (single_h, single, 1);
 
+  /* First, clear all callbacks. */
+  single_h->invoke_async_cb = NULL;
+
   single_h->state = JOIN_REQUESTED;
   g_cond_broadcast (&single_h->cond);
   invoking = single_h->invoking;

c/src/ml-api-service-extension.c

@@ -67,6 +67,17 @@ typedef struct
   GHashTable *node_table;
 } ml_extension_s;
 
+/**
+ * @brief Internal function to handle the asynchronous invoke.
+ */
+static int
+_ml_extension_async_cb (const ml_tensors_data_h data, void *user_data)
+{
+  ml_service_s *mls = (ml_service_s *) user_data;
+
+  return _ml_service_invoke_event_new_data (mls, NULL, data);
+}
+
 /**
  * @brief Internal function to create node info in pipeline.
  */
@@ -355,6 +366,11 @@ _ml_extension_conf_parse_single (ml_service_s * mls, JsonObject * single)
 
     if (STR_IS_VALID (invoke_async)) {
       ml_option_set (option, "invoke_async", g_strdup (invoke_async), g_free);
+
+      if (g_ascii_strcasecmp (invoke_async, "true") == 0) {
+        ml_option_set (option, "async_callback", _ml_extension_async_cb, NULL);
+        ml_option_set (option, "async_data", mls, NULL);
+      }
     }
   }