gsttensor_mux.c

Go to the documentation of this file.

 
#ifdef HAVE_CONFIG_H
#include <config.h>
#endif
 
#include <string.h>
#include <gst/gst.h>
#include <glib.h>
#include <nnstreamer_util.h>
 
#include "gsttensor_mux.h"
 
GST_DEBUG_CATEGORY_STATIC (gst_tensor_mux_debug);
#define GST_CAT_DEFAULT gst_tensor_mux_debug
 
#ifndef DBG
#define DBG (!tensor_mux->silent)
#endif
 
enum
{
  PROP_0,
  PROP_SILENT,
  PROP_SYNC_MODE,
  PROP_SYNC_OPTION,
};
 
#define CAPS_STRING_SINK GST_TENSOR_CAP_DEFAULT ";" GST_TENSORS_CAP_MAKE ("{ static, flexible }")
 
#define CAPS_STRING_SRC GST_TENSORS_CAP_MAKE ("{ static, flexible }")
 
static GstStaticPadTemplate src_templ = GST_STATIC_PAD_TEMPLATE ("src",
    GST_PAD_SRC,
    GST_PAD_ALWAYS,
    GST_STATIC_CAPS (CAPS_STRING_SRC)
    );
 
static GstStaticPadTemplate sink_templ = GST_STATIC_PAD_TEMPLATE ("sink_%u",
    GST_PAD_SINK,
    GST_PAD_REQUEST,
    GST_STATIC_CAPS (CAPS_STRING_SINK)
    );
 
static gboolean gst_tensor_mux_src_event (GstPad * pad, GstObject * parent,
    GstEvent * event);
static GstPad *gst_tensor_mux_request_new_pad (GstElement * element,
    GstPadTemplate * templ, const gchar * name, const GstCaps * caps);
static GstStateChangeReturn gst_tensor_mux_change_state (GstElement * element,
    GstStateChange transition);
static gboolean gst_tensor_mux_sink_event (GstCollectPads * pads,
    GstCollectData * data, GstEvent * event, GstTensorMux * tensor_mux);
static GstFlowReturn gst_tensor_mux_collected (GstCollectPads * pads,
    GstTensorMux * tensor_mux);
static GstFlowReturn gst_tensor_mux_do_clip (GstCollectPads * pads,
    GstCollectData * data, GstBuffer * buffer, GstBuffer ** out,
    GstTensorMux * tensor_mux);
 
static void gst_tensor_mux_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec);
static void gst_tensor_mux_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec);
static void gst_tensor_mux_finalize (GObject * object);
 
#define gst_tensor_mux_parent_class parent_class
G_DEFINE_TYPE (GstTensorMux, gst_tensor_mux, GST_TYPE_ELEMENT);
 
static void
gst_tensor_mux_class_init (GstTensorMuxClass * klass)
{
  GObjectClass *gobject_class;
  GstElementClass *gstelement_class;
 
  GST_DEBUG_CATEGORY_INIT (gst_tensor_mux_debug, "tensor_mux", 0,
      "Element to merge tensor stream to tensors stream");
 
  gobject_class = (GObjectClass *) klass;
  gstelement_class = (GstElementClass *) klass;
 
  parent_class = g_type_class_peek_parent (klass);
 
  gobject_class->finalize = gst_tensor_mux_finalize;
  gobject_class->get_property = gst_tensor_mux_get_property;
  gobject_class->set_property = gst_tensor_mux_set_property;
 
  g_object_class_install_property (gobject_class, PROP_SILENT,
      g_param_spec_boolean ("silent", "Silent", "Produce verbose output ?",
          TRUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
  g_object_class_install_property (gobject_class, PROP_SYNC_MODE,
      g_param_spec_string ("sync-mode", "Sync Mode",
          "Time synchronization mode?", "", G_PARAM_READWRITE));
 
  g_object_class_install_property (gobject_class, PROP_SYNC_OPTION,
      g_param_spec_string ("sync-option", "Sync Option",
          "Option for the time synchronization mode ?", "", G_PARAM_READWRITE));
 
  gstelement_class->request_new_pad =
      GST_DEBUG_FUNCPTR (gst_tensor_mux_request_new_pad);
  gstelement_class->change_state =
      GST_DEBUG_FUNCPTR (gst_tensor_mux_change_state);
 
  gst_element_class_add_pad_template (gstelement_class,
      gst_static_pad_template_get (&sink_templ));
  gst_element_class_add_pad_template (gstelement_class,
      gst_static_pad_template_get (&src_templ));
 
  gst_element_class_set_details_simple (gstelement_class,
      "TensorMux",
      "Muxer/Tensor",
      "Merge multiple tensor stream to tensors stream",
      "Jijoong Moon <jijoong.moon@samsung.com>");
 
}
 
static void
gst_tensor_mux_init (GstTensorMux * tensor_mux)
{
  GstElementClass *klass = GST_ELEMENT_GET_CLASS (tensor_mux);
 
  tensor_mux->srcpad =
      gst_pad_new_from_template (gst_element_class_get_pad_template (klass,
          "src"), "src");
  gst_pad_set_event_function (tensor_mux->srcpad, gst_tensor_mux_src_event);
 
  gst_element_add_pad (GST_ELEMENT (tensor_mux), tensor_mux->srcpad);
 
  tensor_mux->collect = gst_collect_pads_new ();
  gst_collect_pads_set_event_function (tensor_mux->collect,
      (GstCollectPadsEventFunction)
      GST_DEBUG_FUNCPTR (gst_tensor_mux_sink_event), tensor_mux);
  gst_collect_pads_set_function (tensor_mux->collect,
      (GstCollectPadsFunction) GST_DEBUG_FUNCPTR (gst_tensor_mux_collected),
      tensor_mux);
  gst_collect_pads_set_clip_function (tensor_mux->collect,
      (GstCollectPadsClipFunction) GST_DEBUG_FUNCPTR (gst_tensor_mux_do_clip),
      tensor_mux);
 
  tensor_mux->silent = TRUE;
  tensor_mux->sync.mode = SYNC_SLOWEST;
  tensor_mux->sync.option = NULL;
  tensor_mux->current_time = 0;
  tensor_mux->need_set_time = TRUE;
  gst_tensors_config_init (&tensor_mux->tensors_config);
}
 
static void
gst_tensor_mux_finalize (GObject * object)
{
  GstTensorMux *tensor_mux;
 
  tensor_mux = GST_TENSOR_MUX (object);
 
  if (tensor_mux->collect) {
    gst_tensor_time_sync_flush (tensor_mux->collect);
    gst_object_unref (tensor_mux->collect);
    tensor_mux->collect = NULL;
  }
 
  if (tensor_mux->sync.option) {
    g_free (tensor_mux->sync.option);
    tensor_mux->sync.option = NULL;
  }
 
  gst_tensors_config_free (&tensor_mux->tensors_config);
 
  G_OBJECT_CLASS (parent_class)->finalize (object);
}
 
static GstPad *
gst_tensor_mux_request_new_pad (GstElement * element, GstPadTemplate * templ,
    const gchar * req_name, const GstCaps * caps)
{
  GstPad *newpad;
  GSList *walk = NULL;
  GstTensorMux *tensor_mux;
  gchar *name;
  UNUSED (req_name);
  UNUSED (caps);
 
  g_return_val_if_fail (templ != NULL, NULL);
  g_return_val_if_fail (GST_IS_TENSOR_MUX (element), NULL);
 
  tensor_mux = GST_TENSOR_MUX (element);
  walk = tensor_mux->collect->data;
 
  name = g_strdup_printf ("sink_%u", g_slist_length (walk));
  newpad = gst_pad_new_from_template (templ, name);
  g_free (name);
 
  if (newpad) {
    GstTensorCollectPadData *tensormuxpad;
    gboolean locked, waiting;
 
    locked = waiting = TRUE;
 
    if (tensor_mux->sync.mode == SYNC_REFRESH) {
      locked = waiting = FALSE;
    }
 
    tensormuxpad = (GstTensorCollectPadData *)
        gst_collect_pads_add_pad (tensor_mux->collect, newpad,
        sizeof (GstTensorCollectPadData), NULL, locked);
 
    /* NOTE: if locked is TRUE, waiting flag is not effective */
    gst_collect_pads_set_waiting (tensor_mux->collect,
        (GstCollectData *) tensormuxpad, waiting);
 
    gst_pad_set_element_private (newpad, tensormuxpad);
    gst_element_add_pad (element, newpad);
  } else {
    GST_WARNING_OBJECT (tensor_mux, "failed to create request pad");
  }
  return newpad;
}
 
static gboolean
gst_tensor_mux_src_event (GstPad * pad, GstObject * parent, GstEvent * event)
{
  g_return_val_if_fail (event != NULL, FALSE);
 
  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_SEEK:
      gst_event_unref (event);
      return FALSE;
    default:
      break;
  }
 
  return gst_pad_event_default (pad, parent, event);
}
 
static void
gst_tensor_mux_set_waiting (GstTensorMux * tensor_mux, gboolean waiting)
{
  if (tensor_mux->sync.mode == SYNC_REFRESH) {
    GstCollectPads *pads = tensor_mux->collect;
    GSList *walk = pads->data;
 
    while (walk) {
      gst_collect_pads_set_waiting (pads, walk->data, waiting);
      walk = g_slist_next (walk);
    }
  }
}
 
static gboolean
gst_tensor_mux_sink_event (GstCollectPads * pads, GstCollectData * data,
    GstEvent * event, GstTensorMux * tensor_mux)
{
  g_return_val_if_fail (event != NULL, FALSE);
 
  switch (GST_EVENT_TYPE (event)) {
    case GST_EVENT_FLUSH_STOP:
      tensor_mux->need_segment = TRUE;
      tensor_mux->need_set_time = TRUE;
      gst_tensor_time_sync_flush (tensor_mux->collect);
      break;
    case GST_EVENT_EOS:
      gst_tensor_mux_set_waiting (tensor_mux, FALSE);
      break;
    default:
      break;
  }
 
  return gst_collect_pads_event_default (pads, data, event, FALSE);
}
 
static gboolean
gst_tensor_mux_collect_buffer (GstTensorMux * tensor_mux,
    GstBuffer * tensors_buf, gboolean * is_eos)
{
  if (tensor_mux->need_set_time) {
    if (gst_tensor_time_sync_get_current_time (tensor_mux->collect,
            &tensor_mux->sync, &tensor_mux->current_time, tensors_buf)) {
      /* end-of-stream */
      *is_eos = TRUE;
      return FALSE;
    }
 
    tensor_mux->need_set_time = FALSE;
    silent_debug (tensor_mux, "Current Time : %" GST_TIME_FORMAT,
        GST_TIME_ARGS (tensor_mux->current_time));
  }
 
  return gst_tensor_time_sync_buffer_from_collectpad (tensor_mux->collect,
      &tensor_mux->sync, tensor_mux->current_time, tensors_buf,
      &tensor_mux->tensors_config, is_eos);
}
 
static gboolean
gst_tensor_mux_set_src_caps (GstTensorMux * tensor_mux)
{
  if (!tensor_mux->negotiated) {
    GstCaps *caps;
 
    if (gst_tensors_config_validate (&tensor_mux->tensors_config)) {
      caps = gst_tensor_pad_caps_from_config (tensor_mux->srcpad,
          &tensor_mux->tensors_config);
 
      if (gst_pad_set_caps (tensor_mux->srcpad, caps)) {
        tensor_mux->negotiated = TRUE;
      }
 
      gst_caps_unref (caps);
    }
  }
 
  if (!tensor_mux->negotiated) {
    GST_WARNING_OBJECT (tensor_mux, "failed to set caps");
    GST_ELEMENT_ERROR (tensor_mux, CORE, NEGOTIATION, (NULL), (NULL));
  }
 
  return tensor_mux->negotiated;
}
 
static void
gst_tensor_mux_send_segment_event (GstTensorMux * tensor_mux,
    GstClockTime pts, GstClockTime dts)
{
  if (tensor_mux->need_segment) {
    GstSegment segment;
    GstClockTime time = 0;
 
    if (GST_CLOCK_TIME_IS_VALID (dts)) {
      time = dts;
    } else if (GST_CLOCK_TIME_IS_VALID (pts)) {
      time = pts;
    }
 
    gst_segment_init (&segment, GST_FORMAT_TIME);
    segment.start = time;
    gst_pad_push_event (tensor_mux->srcpad, gst_event_new_segment (&segment));
    tensor_mux->need_segment = FALSE;
  }
}
 
static GstBuffer *
gst_tensor_mux_chain_flex_tensor (GstTensorMux * tensor_mux, GstBuffer * buf)
{
  GstBuffer *buffer;
  GstMemory *mem, *new_mem;
  GstTensorsInfo *info;
  GstTensorInfo *_info;
  GstTensorMetaInfo meta;
  guint i;
 
  /* If input is flexible, do nothing. It is already flexible tensor. */
  if (gst_tensors_config_is_flexible (&tensor_mux->tensors_config))
    return buf;
 
  info = &tensor_mux->tensors_config.info;
  buffer = gst_buffer_new ();
 
  for (i = 0; i < info->num_tensors; i++) {
    mem = gst_tensor_buffer_get_nth_memory (buf, i);
 
    /* append header */
    _info = gst_tensors_info_get_nth_info (info, i);
    gst_tensor_info_convert_to_meta (_info, &meta);
    new_mem = gst_tensor_meta_info_append_header (&meta, mem);
    gst_memory_unref (mem);
 
    gst_tensor_buffer_append_memory (buffer, new_mem, _info);
  }
 
  gst_buffer_copy_into (buffer, buf, GST_BUFFER_COPY_METADATA, 0, -1);
  gst_buffer_unref (buf);
  return buffer;
}
 
static GstFlowReturn
gst_tensor_mux_collected (GstCollectPads * pads, GstTensorMux * tensor_mux)
{
  GstFlowReturn ret = GST_FLOW_OK;
  GstBuffer *tensors_buf;
  gboolean isEOS = FALSE;
  gboolean buf_collected = FALSE;
 
  GST_DEBUG_OBJECT (tensor_mux, " all pads are collected ");
 
  if (tensor_mux->need_stream_start) {
    GstCollectData *data = (GstCollectData *) pads->data->data;
    g_autofree gchar *sink_stream_id = gst_pad_get_stream_id (data->pad);
    g_autofree gchar *element_name = gst_element_get_name (tensor_mux);
    g_autofree gchar *pad_name = gst_pad_get_name (tensor_mux->srcpad);
    g_autofree gchar *stream_id = g_strdup_printf ("%s-%s-nnsmux-%s-%08x",
        GST_STR_NULL (sink_stream_id), element_name, pad_name, g_random_int ());
 
    gst_pad_push_event (tensor_mux->srcpad, gst_event_new_stream_start (stream_id));
    tensor_mux->need_stream_start = FALSE;
  }
 
  if ((tensors_buf = gst_buffer_new ()) == NULL) {
    ml_logf ("gst_buffer_new() returns NULL. Out of memory?\n");
    return GST_FLOW_ERROR;
  }
 
  buf_collected =
      gst_tensor_mux_collect_buffer (tensor_mux, tensors_buf, &isEOS);
 
  gst_tensor_mux_set_waiting (tensor_mux, TRUE);
 
  if (!buf_collected) {
    if (isEOS) {
      gst_pad_push_event (tensor_mux->srcpad, gst_event_new_eos ());
      ret = GST_FLOW_EOS;
    }
 
    gst_buffer_unref (tensors_buf);
    return ret;
  }
 
  if (!gst_tensor_mux_set_src_caps (tensor_mux)) {
    gst_buffer_unref (tensors_buf);
    return GST_FLOW_NOT_NEGOTIATED;
  }
 
  gst_tensor_mux_send_segment_event (tensor_mux, GST_BUFFER_PTS (tensors_buf),
      GST_BUFFER_DTS (tensors_buf));
 
  /* add header if output is flexible */
  if (gst_tensor_pad_caps_is_flexible (tensor_mux->srcpad))
    tensors_buf = gst_tensor_mux_chain_flex_tensor (tensor_mux, tensors_buf);
 
  ret = gst_pad_push (tensor_mux->srcpad, tensors_buf);
  tensor_mux->need_set_time = TRUE;
 
  if (ret != GST_FLOW_OK) {
    GST_WARNING_OBJECT (tensor_mux, "pushed outbuf, result = %s",
        gst_flow_get_name (ret));
  }
 
  return ret;
}
 
static GstFlowReturn
gst_tensor_mux_do_clip (GstCollectPads * pads, GstCollectData * data,
    GstBuffer * buffer, GstBuffer ** out, GstTensorMux * tensor_mux)
{
  UNUSED (pads);
  UNUSED (data);
  gst_tensor_mux_set_waiting (tensor_mux, FALSE);
  *out = buffer;
  return GST_FLOW_OK;
}
 
static void
gst_tensor_mux_ready_to_paused (GstTensorMux * tensor_mux)
{
  tensor_mux->need_stream_start = TRUE;
  tensor_mux->need_segment = TRUE;
  tensor_mux->negotiated = FALSE;
  gst_collect_pads_start (tensor_mux->collect);
}
 
static GstStateChangeReturn
gst_tensor_mux_change_state (GstElement * element, GstStateChange transition)
{
  GstTensorMux *tensor_mux;
  GstStateChangeReturn ret;
  tensor_mux = GST_TENSOR_MUX (element);
  switch (transition) {
    case GST_STATE_CHANGE_READY_TO_PAUSED:
      gst_tensor_mux_ready_to_paused (tensor_mux);
      break;
    case GST_STATE_CHANGE_PAUSED_TO_READY:
      gst_collect_pads_stop (tensor_mux->collect);
      break;
    default:
      break;
  }
 
  ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
  if (ret == GST_STATE_CHANGE_FAILURE)
    return ret;
  switch (transition) {
    case GST_STATE_CHANGE_PAUSED_TO_READY:
      break;
    default:
      break;
  }
 
  return ret;
}
 
static void
gst_tensor_mux_set_property (GObject * object, guint prop_id,
    const GValue * value, GParamSpec * pspec)
{
  GstTensorMux *tensor_mux = GST_TENSOR_MUX (object);
  switch (prop_id) {
    case PROP_SILENT:
      tensor_mux->silent = g_value_get_boolean (value);
      break;
    case PROP_SYNC_MODE:
      tensor_mux->sync.mode =
          gst_tensor_time_sync_get_mode (g_value_get_string (value));
      if (tensor_mux->sync.mode == SYNC_END) {
        tensor_mux->sync.mode = SYNC_SLOWEST;
      }
      silent_debug (tensor_mux, "Mode = %d(%s)\n", tensor_mux->sync.mode,
          gst_tensor_time_sync_get_mode_string (tensor_mux->sync.mode));
      gst_tensor_time_sync_set_option_data (&tensor_mux->sync);
      break;
    case PROP_SYNC_OPTION:
      tensor_mux->sync.option = g_value_dup_string (value);
      silent_debug (tensor_mux, "Option = %s\n", tensor_mux->sync.option);
      gst_tensor_time_sync_set_option_data (&tensor_mux->sync);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}
 
static void
gst_tensor_mux_get_property (GObject * object, guint prop_id,
    GValue * value, GParamSpec * pspec)
{
  GstTensorMux *tensor_mux = GST_TENSOR_MUX (object);
  switch (prop_id) {
    case PROP_SILENT:
      g_value_set_boolean (value, tensor_mux->silent);
      break;
    case PROP_SYNC_MODE:
      g_value_set_string (value,
          gst_tensor_time_sync_get_mode_string (tensor_mux->sync.mode));
      break;
    case PROP_SYNC_OPTION:
      g_value_set_string (value, tensor_mux->sync.option);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
  }
}