Динамически (un) связывать элементы в бегущем (gstreamer) конвейере?

Ответ 1

• Моя любимая "концепция" для понимания связывания (и динамической компоновки), думает о трубопроводе как о реальной трубе с потоком воды через нее. Как только вы это сделаете, некоторые вещи станут очень очевидными. Например, "вы устанавливаете источник для PLAYING перед связыванием элемента?", Становится "вы включаете воду перед подключением шланга?" И сами ответы. Более того, с динамической связью, как бы вы гарантировали, что "утечка" воды (что плохо, "утечки" ) в GStreamer является эквивалентом получения GST_FLOW_NOT_LINKED, и остановит ваш источник и удовольствие) или засорится (может выпадение или перегрузка пакетов).

• Да. Многие. С небольшим отказом от ответственности, что я до сих пор работаю с 0.10, и некоторые из них, возможно, были исправлены с 1.0, к сожалению, очень сложно выполнять динамическое связывание и отключение от GStreamer 0.10. Позволь мне объяснить: Скажем, вы используете Tee, и хотите развязать одну ветку. Вы начнете с отпускания Tes srcpad (не считая его разблокировки, что происходит как часть освобождения пэда), и теперь вы должны надежно стереть элементы вниз по течению от этой панели. (Водный эквивалент заключается в том, что вы закрываете клапан после тройника и теперь должны иметь возможность демонтировать трубы после клапана, вы не начнете разбирать трубы, не закрывая сначала клапан, если вы не хотите промокнуть...) Это будет работать большую часть времени, но здесь есть гонка. Поскольку после того, как вы отпустили пэд, на этом пэде может остаться push или pad-alloc, и если теперь вы в своем коде начнете срывать нисходящие элементы, это может теперь сбой из-за гонки, которая существует в некоторых элементах, если они получают push или pad-alloc при срыве, или вы получаете GST_FLOW_WRONG_STATE или GST_FLOW_NOT_LINKED, и они вернутся к источнику, останавливающему поток для всех...

• Я много экспериментировал с этим, и обнаружил, что если вам нужна стабильность, а время сбоя/замораживания иногда не является вариантом, вам нужен элемент, который будет служить вашей динамической сетью безопасности. Элемент, который гарантирует, что абсолютно никакой активности не произойдет на пэде после того, как вы отпустите/отмените его. Единственный способ сделать это - разбить еще одну парадигму GStreamer на то, чтобы не нажимать, удерживая блокировку: вам нужно удерживать блокировку при нажатии/отправке событий/выделении пэдов. Я сделал это некоторое время назад здесь. (тестовый корпус является самым важным, конечно, поскольку он позволяет вам проверять свои собственные/другие элементы для их безопасности) Вы также можете представить элемент без блокировки, который проглотит все плохие FlowReturns и просто нарисует красивую картинку для своего восходящего потока, но тогда вам нужно быть абсолютно уверенным, что все ваши нисходящие элементы будут "push или pad-alloc" в то время как вы отключите "-safe", так как ваш элемент не сможет гарантировать, что когда-то "остановить поток" (release/отсоединение) будет выполнено, небольшое падение не будет выдавливаться.

Конечно, вы должны поместить некоторые из них в перспективе. Окно для этих ужасных условий гонки, о которых я говорю, на самом деле очень, очень мало, и может произойти только каждые 1000 или 10.000 раз, когда вы запускаете свою программу. Но для профессионального применения это, конечно, неприемлемо. Я поговорил, где я освещал некоторые из этих материалов здесь

Ответ 2

Я стараюсь использовать селектор вывода или селектора ввода в зависимости от ситуации, а не сложности блокировки клавиатуры (я ответил на блокировку клавиатуры в другом сообщении http://gstreamer-devel.966125.n4.nabble.com/Dynamically-adding-and-removing-branches-of-a-tee-td973635.html#a4656812). И подключите селектор к fakesrc или fakesink, если он не используется. В приведенном ниже примере, если вы используете GTK, можно заменить строку g_timeout_add (SWITCH_TIMEOUT, switch_cb, osel); на gtk_toggle_button и поместить весь код в настоящее время в функцию switch_cb в функцию обратного вызова кнопки переключения. В этом коде можно переключаться между двумя изображениями. Я бы заменил один снимок изображения на fakesink, чтобы поддерживать работу на конвейере, в случае, если я хочу добавить tee в будущем с файловым окном, где я хочу записывать видео, но предоставляю плеер возможность включить (селектор на изображение)/off (селектор на fakesink) дисплей. Это позволяет добавлять/удалять бункеры во время выполнения с помощью селектора.

#include <gst/gst.h>

#define SWITCH_TIMEOUT 1000
#define NUM_VIDEO_BUFFERS 500

static GMainLoop *loop;

/* Output selector src pads */
static GstPad *osel_src1 = NULL;
static GstPad *osel_src2 = NULL;

static gboolean
my_bus_callback (GstBus * bus, GstMessage * message, gpointer data)
{
  g_print ("Got %s message\n", GST_MESSAGE_TYPE_NAME (message));

  switch (GST_MESSAGE_TYPE (message)) {
    case GST_MESSAGE_ERROR:{
      GError *err;
      gchar *debug;

      gst_message_parse_error (message, &err, &debug);
      g_print ("Error: %s\n", err->message);
      g_error_free (err);
      g_free (debug);

      g_main_loop_quit (loop);
      break;
    }
    case GST_MESSAGE_EOS:
      /* end-of-stream */
      g_main_loop_quit (loop);
      break;
    default:
      /* unhandled message */
      break;
  }
  /* we want to be notified again the next time there is a message
   * on the bus, so returning TRUE (FALSE means we want to stop watching
   * for messages on the bus and our callback should not be called again)
   */
  return TRUE;
}

static gboolean
switch_cb (gpointer user_data)
{
  GstElement *sel = GST_ELEMENT (user_data);
  GstPad *old_pad, *new_pad = NULL;

  g_object_get (G_OBJECT (sel), "active-pad", &old_pad, NULL);

  if (old_pad == osel_src1)
    new_pad = osel_src2;
  else
    new_pad = osel_src1;

  g_object_set (G_OBJECT (sel), "active-pad", new_pad, NULL);

  g_print ("switched from %s:%s to %s:%s\n", GST_DEBUG_PAD_NAME (old_pad),
      GST_DEBUG_PAD_NAME (new_pad));

  gst_object_unref (old_pad);

  return TRUE;

}

gint
main (gint argc, gchar * argv[])
{
  GstElement *pipeline, *src, *toverlay, *osel, *sink1, *sink2, *convert;
  GstPad *sinkpad1;
  GstPad *sinkpad2;
  GstBus *bus;

  /* init GStreamer */
  gst_init (&argc, &argv);
  loop = g_main_loop_new (NULL, FALSE);

  /* create elements */
  pipeline = gst_element_factory_make ("pipeline", "pipeline");
  src = gst_element_factory_make ("videotestsrc", "src");
  toverlay = gst_element_factory_make ("timeoverlay", "timeoverlay");
  osel = gst_element_factory_make ("output-selector", "osel");
  convert = gst_element_factory_make ("ffmpegcolorspace", "convert");
  sink1 = gst_element_factory_make ("xvimagesink", "sink1");
  sink2 = gst_element_factory_make ("ximagesink", "sink2");

  if (!pipeline || !src || !toverlay || !osel || !convert || !sink1 || !sink2) {
    g_print ("missing element\n");
    return -1;
  }

  /* add them to bin */
  gst_bin_add_many (GST_BIN (pipeline), src, toverlay, osel, convert, sink1,
      sink2, NULL);

  /* set properties */
  g_object_set (G_OBJECT (src), "is-live", TRUE, NULL);
  g_object_set (G_OBJECT (src), "do-timestamp", TRUE, NULL);
  g_object_set (G_OBJECT (src), "num-buffers", NUM_VIDEO_BUFFERS, NULL);
  g_object_set (G_OBJECT (sink1), "sync", FALSE, "async", FALSE, NULL);
  g_object_set (G_OBJECT (sink2), "sync", FALSE, "async", FALSE, NULL);
  g_object_set (G_OBJECT (osel), "resend-latest", TRUE, NULL);

  /* link src ! timeoverlay ! osel */
  if (!gst_element_link_many (src, toverlay, osel, NULL)) {
    g_print ("linking failed\n");
    return -1;
  }

  /* link output 1 */
  sinkpad1 = gst_element_get_static_pad (sink1, "sink");
  osel_src1 = gst_element_get_request_pad (osel, "src%d");
  if (gst_pad_link (osel_src1, sinkpad1) != GST_PAD_LINK_OK) {
    g_print ("linking output 1 failed\n");
    return -1;
  }
  gst_object_unref (sinkpad1);

  /* link output 2 */
  sinkpad2 = gst_element_get_static_pad (convert, "sink");
  osel_src2 = gst_element_get_request_pad (osel, "src%d");
  if (gst_pad_link (osel_src2, sinkpad2) != GST_PAD_LINK_OK) {
    g_print ("linking output 2 failed\n");
    return -1;
  }
  gst_object_unref (sinkpad2);

  if (!gst_element_link (convert, sink2)) {
    g_print ("linking output 2 failed\n");
    return -1;
  }

  /* add switch callback */
  g_timeout_add (SWITCH_TIMEOUT, switch_cb, osel);

  /* change to playing */
  bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
  gst_bus_add_watch (bus, my_bus_callback, loop);
  gst_object_unref (bus);

  gst_element_set_state (pipeline, GST_STATE_PLAYING);

  /* now run */
  g_main_loop_run (loop);

  /* also clean up */
  gst_element_set_state (pipeline, GST_STATE_NULL);
  gst_element_release_request_pad (osel, osel_src1);
  gst_element_release_request_pad (osel, osel_src2);
  gst_object_unref (GST_OBJECT (pipeline));

  return 0;
}

Ответ 3

На самом деле я пытаюсь сделать то же самое. Не слишком много удачи: (

Я получил следующую ссылку, запросив канал #gstreamer IRC: http://cgit.freedesktop.org/gstreamer/gstreamer/tree/docs/design/part-dynamic.txt

Возможно, намек в правильном направлении.

Пожалуйста, дайте мне знать, когда вы найдете другую документацию...

Ответ 4

Это сообщение появилось первым, когда я искал динамическое изменение любого конвейера gstreamer. Нашел несколько ссылок, но теперь он хорошо документирован в руководстве: http://gstreamer.freedesktop.org/data/doc/gstreamer/head/manual/html/section-dynamic-pipelines.html

Ответ 5

Я не достигал создания полных разборчивых файлов мультиплексирования для Gstreamer 0.10 поверх мультификсации или селектора вывода.

После анализа множества альтернатив мое решение принимает в качестве базы кода пример, изображенный в: http://gstreamer.freedesktop.org/data/doc/gstreamer/head/manual/html/section-dynamic-pipelines.html

API-интерфейс зондов был немного изменен с 0.10 до 1.0, но нижеприведенное решение работает для создания каждого N секундных файлов MP4:

static GstElement *pipeline = NULL;

// Pipeline -> src                    -> dynamic pipeline
// Pipeline -> capsfilter(f264file)   -> mp4mux(mux0)                                 -> filesink(fsink0)
// Pipeline -> elem_before||blockpad| -> |elem_cur_sinkpad||elem_cur||elem_cur_srcpad -> |elem_after_sinkpad||elem_after
static gulong probe_id;             // probe ID
static GstElement *elem_before;     // SRC of dynamic pipeline
static GstElement *elem_after;      // SINK of dynamic pipeline
static GstElement *elem_cur;        // Main element of dynamic pipeline
static GstPad *blockpad;            // SRC pad to be blocked
static GstPad *elem_cur_srcpad;     // SRC pad where check EOS
static GstPad *elem_cur_sinkpad;    // SINK of dynamic pipeline
static GstPad *elem_after_sinkpad;  // SINK of SINK element

// Last Buffer Timestamp
static GstClockTime last_ts = 0;

typedef enum {
  NO_NEW_FILE,  // Keep current file destination
  NEW_FILE,     // Switch file destination
} NewFileStatus;
static NewFileStatus newfile = NO_NEW_FILE; // Switch File Flag

static int counter = 1; // Index filename

// EOS listener to switch to other file destination
static gboolean
event_probe_cb (GstPad * pad, GstEvent * event, gpointer user_data)
{
  g_print ("INSIDE event_probe_cb:%d type:%s\n",probe_id,
      GST_EVENT_TYPE (event)==GST_EVENT_EOS?"EOS":GST_EVENT_TYPE (event)==GST_EVENT_NEWSEGMENT?"NEWSEGMENT":"OTHER");

  if (GST_EVENT_TYPE (event) != GST_EVENT_EOS)
  {
    // Push the event in the pipe flow (false DROP)
    return TRUE;
  }

  // remove the probe first
  gst_pad_remove_event_probe (pad, probe_id);

  gst_object_unref (elem_cur_srcpad);
  gst_object_unref (elem_after_sinkpad);
  gst_element_release_request_pad(elem_cur, elem_cur_sinkpad);

  gst_element_set_state (elem_cur, GST_STATE_NULL);
  gst_element_set_state (elem_after, GST_STATE_NULL);

  // remove unlinks automatically
  GST_DEBUG_OBJECT (pipeline, "removing %" GST_PTR_FORMAT, elem_cur);
  gst_bin_remove (GST_BIN (pipeline), elem_cur);
  GST_DEBUG_OBJECT (pipeline, "removing %" GST_PTR_FORMAT, elem_after);
  gst_bin_remove (GST_BIN (pipeline), elem_after);

  GstElement * mux0 = gst_element_factory_make("mp4mux", "mux0");
  GstElement * fsink0 = gst_element_factory_make("filesink", "fsink0");
  elem_cur = mux0;
  elem_after = fsink0;

  if(!mux0 || !fsink0)
  {
    printf("mising elements\n");
  }

  GST_DEBUG_OBJECT (pipeline, "adding   %" GST_PTR_FORMAT, elem_cur);
  gst_bin_add (GST_BIN (pipeline), elem_cur);
  GST_DEBUG_OBJECT (pipeline, "adding   %" GST_PTR_FORMAT, elem_after);
  gst_bin_add (GST_BIN (pipeline), elem_after);

  char buffer[128];
  sprintf(buffer, "test_%d.mp4", counter++);
  g_print ("File Switching %s\n", buffer);
  g_object_set(G_OBJECT(elem_after), "location", buffer, NULL);

  GST_DEBUG_OBJECT (pipeline, "linking..");
  elem_cur_srcpad = gst_element_get_static_pad (elem_cur, "src");
  elem_cur_sinkpad = gst_element_get_request_pad (elem_cur, "video_%d");
  elem_after_sinkpad = gst_element_get_static_pad (elem_after, "sink");

  if(gst_pad_link(blockpad, elem_cur_sinkpad) != GST_PAD_LINK_OK)
  {
    printf("linking output 0 failed\n");
    return -1;
  }
  if(gst_pad_link(elem_cur_srcpad, elem_after_sinkpad) != GST_PAD_LINK_OK)
  {
    printf("linking output 1 failed\n");
    return -1;
  }

  g_print ("Moving to PLAYING\n");
  gst_element_set_state (elem_cur, GST_STATE_PLAYING);
  gst_element_set_state (elem_after, GST_STATE_PLAYING);

  GST_DEBUG_OBJECT (pipeline, "done");

  newfile = NO_NEW_FILE;
  // Push the event in the pipe flow (false DROP)
  return TRUE;
}

// Check if Buffer contains a KEY FRAME
static gboolean
is_sync_frame (GstBuffer * buffer)
{
  if (GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_DELTA_UNIT)) 
  {
    return FALSE;
  }
  else if (!GST_BUFFER_FLAG_IS_SET(buffer, GST_BUFFER_FLAG_IN_CAPS)) 
  {
    return TRUE;
  }
}

// Block source and launch EOS to MUXER to achieve a full muxed file
static gboolean
pad_probe_cb (GstPad * pad, GstBuffer * buffer, gpointer user_data)
{
  g_print ("\n\tINSIDE pad_probe_cb:%d %s %s\n",probe_id, (newfile?"newfile":"thesame"), 
      (is_sync_frame (buffer)?"KEYframe":"frame"));
  GST_DEBUG_OBJECT (pad, "pad is blocked now");

  last_ts = GST_BUFFER_TIMESTAMP(buffer);
  if(!GST_CLOCK_TIME_IS_VALID(last_ts))
      last_ts=0;

  if((newfile==NO_NEW_FILE) || !is_sync_frame (buffer))
    return TRUE;

  /* remove the probe first */
  gst_pad_remove_buffer_probe (pad, probe_id);

  /* install new probe for EOS */
  probe_id = gst_pad_add_event_probe (elem_after_sinkpad, G_CALLBACK(event_probe_cb), user_data);

  /* push EOS into the element, the probe will be fired when the
   * EOS leaves the effect and it has thus drained all of its data */
  gst_pad_send_event (elem_cur_sinkpad, gst_event_new_eos ());

  // Wait til the EOS have been processed the Buffer with the Key frame will be the FIRST
  while(newfile != NO_NEW_FILE)
      Sleep(1);

  // Push the buffer in the pipe flow (false DROP)
  return TRUE;
}

// this timeout is periodically run as part of the mainloop
static gboolean timeout (gpointer user_data)
{
  g_print ("TIMEOUT\n");
  if(!playing)
      return false;
  newfile = NEW_FILE;
  /* install new probe for Keyframe and New File */
  probe_id = gst_pad_add_buffer_probe (blockpad, G_CALLBACK(pad_probe_cb), pipeline);
  return true;
}

код >