Наши партнеры

Библиотека сайта rus-linux.net

Фреймворк GStreamer. Руководство разработчика приложений. Программные потоки

Оригинал: GStreamer Application Development Manual
Авторы: Wim Taymans, Steve Baker, Andy Wingo, Ronald S. Bultje, Stefan Kost
Дата публикации: 21 мая 2014 г.
Перевод: А.Панин
Дата перевода: 21 июня 2014 г.

Глава 17. Программные потоки

Фреймворк GStreamer является многопоточным и полностью потокобезопасным. Большая часть механизмов управления программными потоками скрыта от приложения с целью упрощения процесса его разработки. Однако, в некоторых случаях при разработке приложения может возникнуть необходимость в воздействии на ряд аспектов работы механизма управления программными потоками. Фреймворк GStreamer позволяет приложениям принудительно использовать множество программных потоков в некоторых частях конвейера. Обратитесь к разделу ниже под названием "Когда вы можете предпочесть принудительное использование программного потока?" для ознакомления с подробностями.

Фреймворк GStreamer также может уведомлять вас о создании программных потоков для того, чтобы вы имели возможность настроить такие параметры, как приоритет потоков или выбрать пул потоков для использования. Обратитесь к разделу "Конфигурация программных потоков в рамках фреймворка GStreamer" для ознакомления с подробностями.

17.1. Планирование работы элементов в рамках фреймворка GStreamer

Каждый элемент конвейера фреймворка GStreamer принимает решение о том, как будет планироваться его работа. Элементы могут принимать решения о том, будет ли осуществляться планирование работы их точек соединения на основе операций отправки или операций приема данных. Элемент может, к примеру, принять решение о создании программного потока для захвата данных посредством входной точки соединения или/и для передачи данных посредством выходной точки соединения. Элемент также может принять решение о том, использовать ли программные потоки предыдущего или следующего элемента конвейера для обработки своих данных в режимах отправки и приема соответственно. Фреймворк GStreamer не накладывает каких-либо ограничений на возможные решения элемента в отношении планирования своей работы. Обратитесь к Руководству разработчика плагинов для ознакомления с подробностями.

В любом случае некоторые элементы будут создавать программные потоки, называемые "программными потоками, предназначенными для обработки мультимедийных потоков" ("streaming threads") для своих целей, заключающихся в обработке мультимедийных данных. Программные потоки, предназначенные для обработки мультимедийных потоков или объекты задач типа GstTask создаются в рамках объекта пула задач типа GstTaskPool в момент, когда элементу необходим программный поток для обработки мультимедийного потока. В следующем разделе мы познакомимся с методом получения уведомлений о состоянии упомянутых объектов задач и объектов пулов задач.

17.2. Конфигурация программных потоков в рамках фреймворка GStreamer

Сообщение STREAM_STATUS передается с помощью шины сообщений для информирования вас о состоянии программных потоков, предназначенных для обработки мультимедийных потоков. Из упомянутого сообщения вы можете извлечь следующую информацию:

Перед созданием нового программного потока вы будете уведомлены с помощью сообщения типа GST_STREAM_STATUS_TYPE_CREATE. После его приема у вас появится возможность установки параметров объекта пула задач и объекта задачи типов GstTaskPool и GstTask соответственно. Специальный пул задач будет предоставлять специальные программные потоки, выполняющие функции обработки мультимедийных потоков.
Данное сообщение должно быть обработано синхронно в том случае, если вы хотите создать специальный пул задач. В том случае, если вы не настраиваете пул задач для создания программного потока, после выхода из обработчика данного сообщения объект задачи будет использовать стандартный пул.
Сообщение будет отправлено при входе в программный поток и выходе из него. Это именно тот момент, когда вам следует устанавливать приоритеты программных потоков. Вы также получите уведомление в момент уничтожения программного потока.
Вы получите сообщения в момент, когда программный поток начнет, приостановит или завершит свою работу. Эти сообщения могут использоваться для визуализации состояния программных потоков, предназначенных для обработки мультимедийных потоков, средствами приложения с графическим интерфейсом.

В следующих разделах мы рассмотрим несколько примеров использования описанных сообщений.

17.2.1. Повышение приоритета программного потока

        .----------.    .----------.
        | faksesrc |    | fakesink |
        |        вых.->вход.       |
        '----------'    '----------'

Давайте рассмотрим изображенный выше простой конвейер. Нам хотелось бы повысить приоритет программного потока, предназначенного для обработки мультимедийного потока. В данном случае элемент fakesrc создает программный поток для обработки мультимедийных данных с целью генерации потока произвольных данных и передачи его связанному элементу для приема произвольных данных. Алгоритм изменения приоритета будет выглядеть аналогичным образом:

При переводе элемента fakesrc из состояния "готовности" (READY) в состояние "пауза" (PAUSED), ему потребуется программный поток для передачи данных элементу fakesink. Он отправит сообщение STREAM_STATUS, указывающее на его потребность в программном потоке для обработки мультимедийных данных.
Приложение будет обрабатывать сообщения STREAM_STATUS с помощью синхронного обработчика сообщений шины. После этого оно настроит объект специального пула задач типа GstTaskPool для размещения объекта задачи типа GstTask, передаваемой вместе с сообщением. Этот специальный пул задач отвечает за создание программных потоков. В данном примере мы будем создавать программный поток с повышенным приоритетом.
Альтернативным методом изменения приоритета или политики планирования исполнения текущего программного потока является обработка уведомлений о входе в программный поток и выходе из него (ENTER/LEAVE), так как синхронная обработка сообщения происходит в контексте программного потока.

На первом шаге мы должны реализовать специальный объект пула задач типа GstTaskPool, с помощью которого мы сможем настроить объект задачи. Ниже приведена реализация подкласса класса GstTaskPool, в которой используются функции библиотеки pthreads для создания потока реального времени с политикой планирования исполнения SCHED_RR. Учтите, что для создания программных потоков реального времени могут потребоваться дополнительные привилегии в системе.

#include <pthread.h>

typedef struct
{
  pthread_t thread;
} TestRTId;

G_DEFINE_TYPE (TestRTPool, test_rt_pool, GST_TYPE_TASK_POOL);

static void
default_prepare (GstTaskPool * pool, GError ** error)
{
  /* Здесь мы не выполняем никаких действий. В рамках данной функции мы могли бы  
    создать пул программных потоков для последующего использования, но не делаем этого */
}

static void
default_cleanup (GstTaskPool * pool)
{
}

static gpointer
default_push (GstTaskPool * pool, GstTaskPoolFunction func, gpointer data,
    GError ** error)
{
  TestRTId *tid;
  gint res;
  pthread_attr_t attr;
  struct sched_param param;

tid = g_slice_new0 (TestRTId);

pthread_attr_init (&attr);
  if ((res = pthread_attr_setschedpolicy (&attr, SCHED_RR)) != 0)
    g_warning ("Ошибка вызова функции setschedpolicy: %p", g_strerror (res));

param.sched_priority = 50;
  if ((res = pthread_attr_setschedparam (&attr, ¶m)) != 0)
    g_warning ("Ошибка вызова функции setschedparam: %p", g_strerror (res));

if ((res = pthread_attr_setinheritsched (&attr, PTHREAD_EXPLICIT_SCHED)) != 0)
    g_warning ("Ошибка вызова функции setinheritsched: %p", g_strerror (res));

res = pthread_create (&tid->thread, &attr, (void *(*)(void *)) func, data);

if (res != 0) {
    g_set_error (error, G_THREAD_ERROR, G_THREAD_ERROR_AGAIN,
        "Возникла ошибка при создании программного потока: %s", g_strerror (res));
    g_slice_free (TestRTId, tid);
    tid = NULL;
  }

return tid;
}

static void
default_join (GstTaskPool * pool, gpointer id)
{
  TestRTId *tid = (TestRTId *) id;

pthread_join (tid->thread, NULL);

g_slice_free (TestRTId, tid);
}

static void
test_rt_pool_class_init (TestRTPoolClass * klass)
{
  GstTaskPoolClass *gsttaskpool_class;

gsttaskpool_class = (GstTaskPoolClass *) klass;

gsttaskpool_class->prepare = default_prepare;
  gsttaskpool_class->cleanup = default_cleanup;
  gsttaskpool_class->push = default_push;
  gsttaskpool_class->join = default_join;
}

static void
test_rt_pool_init (TestRTPool * pool)
{
}

GstTaskPool *
test_rt_pool_new (void)
{
  GstTaskPool *pool;

pool = g_object_new (TEST_TYPE_RT_POOL, NULL);

return pool;
}

При реализации класса пула программных потоков особенно важной является функция с именем "push", предназначенная для добавления программного потока в пул. Реализация класса должна создавать программный поток, который вызовет заданную функцию. Более сложные реализации класса могут поддерживать работу нескольких программных потоков в рамках одного пула, так как операции создания и уничтожения программных потоков не всегда являются самыми быстрыми.

На следующем шаге мы должны будем произвести настройку нашего специального пула программных потоков в момент, когда он потребуется элементу fakesrc. Для этого мы будем перехватывать сообщения STREAM_STATUS с помощью синхронного обработчика.

static GMainLoop* loop;

static void
on_stream_status (GstBus     *bus,
                  GstMessage *message,
                  gpointer    user_data)
{
  GstStreamStatusType type;
  GstElement *owner;
  const GValue *val;
  GstTask *task = NULL;

gst_message_parse_stream_status (message, &type, &owner);

val = gst_message_get_stream_status_object (message);

/* Определимся, можем ли мы работать с данным объектом */
  if (G_VALUE_TYPE (val) == GST_TYPE_TASK) {
    task = g_value_get_object (val);
  }

switch (type) {
    case GST_STREAM_STATUS_TYPE_CREATE:
      if (task) {
        GstTaskPool *pool;

pool = test_rt_pool_new();

gst_task_set_pool (task, pool);
      }
      break;
    default:
      break;
  }
}

static void
on_error (GstBus     *bus,
          GstMessage *message,
          gpointer    user_data)
{
  g_message ("Принято сообщение об ошибке (ERROR)");
  g_main_loop_quit (loop);
}

static void
on_eos (GstBus     *bus,
        GstMessage *message,
        gpointer    user_data)
{
  g_main_loop_quit (loop);
}

int
main (int argc, char *argv[])
{
  GstElement *bin, *fakesrc, *fakesink;
  GstBus *bus;
  GstStateChangeReturn ret;
  
  gst_init (&argc, &argv);

/* Создание нового контейнера для хранения элементов */
  bin = gst_pipeline_new ("pipeline");
  g_assert (bin);

/* Создание элемента для  ввода данных*/
  fakesrc = gst_element_factory_make ("fakesrc", "fakesrc");
  g_assert (fakesrc);
  g_object_set (fakesrc, "num-buffers", 50, NULL);

/* И элемента для вывода данных */
  fakesink = gst_element_factory_make ("fakesink", "fakesink");
  g_assert (fakesink);

/* Добавление объектов в состав главного конвейера */
  gst_bin_add_many (GST_BIN (bin), fakesrc, fakesink, NULL);

/* Связывание элементов */
  gst_element_link (fakesrc, fakesink);

loop = g_main_loop_new (NULL, FALSE);

/* Получение объекта шины, для сообщений с которой мы должны установить синхронный обработчик */
  bus = gst_pipeline_get_bus (GST_PIPELINE (bin));
  gst_bus_enable_sync_message_emission (bus);
  gst_bus_add_signal_watch (bus);

g_signal_connect (bus, "sync-message::stream-status",
      (GCallback) on_stream_status, NULL);
  g_signal_connect (bus, "message::error",
      (GCallback) on_error, NULL);
  g_signal_connect (bus, "message::eos",
      (GCallback) on_eos, NULL);

/* Начало воспроизведения */
  ret = gst_element_set_state (bin, GST_STATE_PLAYING);
  if (ret != GST_STATE_CHANGE_SUCCESS) {
    g_message ("Не удалось изменить состояние конвейера");
    return -1;
  }

/* Запуск главного цикла обработки событий, который будет обрабатывать сообщения,
     переданные с помощью шины сообщений до момента приема сообщения  об окончании
     потока (EOS) или ошибке (ERROR) */
  g_main_loop_run (loop);

/* Остановка элементов контейнера */
  gst_element_set_state (bin, GST_STATE_NULL);
  gst_object_unref (bus);
  g_main_loop_unref (loop);

return 0;
}

Учтите, что этой программе, скорее всего, потребуются привилегии пользователя root для создания потоков реального времени. В том случае, если не удастся создать программный поток, вызов функции изменения состояния будет завершаться неудачно, причем этот результат вызова функции обрабатывается нами в коде приложения, приведенном выше.

В том случае, если для работы конвейера используется множество программных потоков, вы будете получать множество сообщений STREAM_STATUS. В такой ситуации вам следует использовать информацию о владельце сообщения, которым, скорее всего, будет точка соединения или элемент, который инициировал запуск программного потока, для того, чтобы понять, каково назначение рассматриваемого программного потока в контексте данного приложения.

17.3. Когда вы можете предпочесть принудительное использование программного потока?

Мы увидели, что программные потоки создаются элементами, но также возможна вставка элементов в конвейер с единственной целью принудительного создания нового программного потока в рамках конвейера.

Ниже приведены некоторые причины, по которым вы можете принудительно создавать программные потоки. Однако, по причинам, связанным с производительностью, вам наверняка не захочется создавать по потоку для каждого существующего элемента, так как это приведет чрезмерным затратам ресурсов. Давайте рассмотрим несколько ситуаций, в которых использование программных потоков может быть особенно полезным:

Буферизация данных, например, при работе с сетевыми мультимедийными потоками или при записи данных из такого источника мультимедийного потока реального времени, как видеокарта или аудиокарта. В этом случае кратковременные задержки в каких-либо частях конвейера не приведут к потере данных. Также имеет смысл обратиться к разделу с названием "Буферизация потока" для ознакомления с механизмом буферизации потока на основе элемента queue2.
Рисунок 17.1. Буферизация потока данных из сетевого источника
Синхронизация устройств для вывода данных, например, в процессе проигрывания потока, содержащего как видео-, так и аудиоданные. При использовании программных потоков для обоих устройств, предназначенных для вывода данных, последние будут работать независимо друг от друга и их синхронизация будет осуществляться лучшим образом.
Рисунок 17.2. Синхронизация устройств для вывода аудио- и видеоданных

В тексте выше мы несколько раз упоминали элемент очереди "queue". Элемент очереди является разделительным элементом, с помощью которого вы можете принудительно использовать программные потоки. Он выполняет свои функции, используя классическую модель "поставщик/потребитель", которая изучается на занятиях по теории потоков в университетах по всему миру. Участвуя в работе конвейера, он также позволяет сделать процесс передачи данных между программными потоками потокобезопасным и выступает в качестве буфера данных. Элементы очередей имеют несколько свойств объектной модели GObject, значения которых могут изменяться при специфических способах их использования. Например, вы можете установить верхний и нижний пределы объема данных для элемента. В том случае, если объем данных меньше нижнего предела (который по умолчанию отключен), элемент заблокирует их вывод. В том случае, если объем данных больше верхнего предела, элемент заблокирует их ввод или (если он настроен соответствующим образом) отбросит данные.

Для использования элемента очереди (и, следовательно, для принудительного использования двух отдельных программных потоков в рамках конвейера), можно просто создать элемент "queue" и использовать его в качестве элемента конвейера. Внутренние механизмы фреймворка GStreamer позаботятся о корректной настройке всех параметров программных потоков.

Следующий раздел : Автоматическая подгрузка плагинов