Матеріал статті взято з мого .
Механізм переміщення даних
- Блок даних dblk_t
- Повідомлення mblk_t
- Функції роботи із повідомленнями mblk_t
- Черга queue_t
- Функції роботи з чергами queue_t
- З'єднання фільтрів
- Сигнальна точка графа обробки даних
- Закулісна діяльність тикера
- Буферизатор (MSBufferizer)
- Функції роботи з MSBufferizer
У минулій ми розробили власний фільтр. Цю статтю ми присвітимо влаштуванню внутрішнього механізму переміщення даних між фільтрами медіастрімера. Це дозволить надалі писати витончені фільтри з меншими зусиллями.
Механізм переміщення даних
Переміщення даних у медіастримері виконується за допомогою черг, що описуються структурою queue_t. По чергах переміщуються низки повідомлень типу mblk_t, які власними силами не містять даних сигналу, лише посилання на попереднє, наступне повідомлення і блок даних. Крім цього, хочу наголосити особливо, є ще поле для посилання на повідомлення такого ж типу, яке дозволяє організувати однозв'язний список із повідомлень. Групу повідомлень, об'єднаних таким списком, називатимемо кортежем. Таким чином, будь-який елемент черги може виявитися поодиноким повідомленням mblk_t, а може і головою кортежу повідомлень mblk_t. Кожне повідомлення кортежу може мати свій підопічний блок даних. Навіщо потрібні кортежі ми обговоримо трохи згодом.
Як було сказано вище, саме собою повідомлення не містить блок даних, натомість воно містить тільки покажчик на область пам'яті де зберігається блок. У цій частині загальна картина роботи медіастрімера нагадує склад дверей у мультфільмі "Корпорація монстрів", де двері (посилання на дані - кімнати) з шаленою швидкістю рухаються підвісними конвеєрами, при цьому самі кімнати залишаються нерухомими.
Тепер, рухаючись ієрархією знизу вгору, розглянемо детально перераховані сутності механізму передачі даних в медіастрімері.
Блок даних dblk_t
Блок даних складається з заголовка та буфера даних. Заголовок описується наступною структурою,
typedef struct datab
{
unsigned char *db_base; // Указатель на начало буфер данных.
unsigned char *db_lim; // Указатель на конец буфер данных.
void (*db_freefn)(void*); // Функция освобождения памяти при удалении блока.
int db_ref; // Счетчик ссылок.
} dblk_t;Поля структури містять покажчики початку буфера, кінець буфера, функцію видалення буфера даних. Останній елемент у заголовку db_ref - Лічильник посилань, якщо він досягає нуля, це служить сигналом до видалення даного блоку з пам'яті. Якщо блок даних створено функцією datab_alloc() , то буфер даних буде розміщено в пам'яті відразу після заголовка. В інших випадках буфер може розташовуватися десь окремо. У буфері даних розташовуватимуться відліки сигналу або інші дані, які ми хочемо обробляти фільтрами.
Новий екземпляр блоку даних створюється за допомогою функції:
dblk_t *datab_alloc(int size);Як вхідний параметр їй передається розмір даних, які зберігатиме блок. Пам'яті виділяється більше, щоб на початку виділеної пам'яті розмістити заголовок - структуру datab. Але при використанні інших функцій так відбувається не завжди, в деяких випадках буфер даних може розташовуватися окремо від заголовка даних. Поля структури при створенні налаштовуються так, щоб її поле db_base вказувало на початок області даних, а db_lim на її кінець. Лічильник посилань db_ref встановлюється за одиницю. Вказівник функції очищення даних встановлюється в нуль.
Повідомлення mblk_t
Як було сказано, елементи черги мають тип mblk_t, він визначений так:
typedef struct msgb
{
struct msgb *b_prev; // Указатель на предыдущий элемент списка.
struct msgb *b_next; // Указатель на следующий элемент списка.
struct msgb *b_cont; // Указатель для подклейки к сообщению других сообщений, для создания кортежа сообщений.
struct datab *b_datap; // Указатель на структуру блока данных.
unsigned char *b_rptr; // Указатель на начало области данных для чтения данных буфера b_datap.
unsigned char *b_wptr; // Указатель на начало области данных для записи данных буфера b_datap.
uint32_t reserved1; // Зарезервированное поле1, медиастример помещает туда служебную информацию.
uint32_t reserved2; // Зарезервированное поле2, медиастример помещает туда служебную информацию.
#if defined(ORTP_TIMESTAMP)
struct timeval timestamp;
#endif
ortp_recv_addr_t recv_addr;
} mblk_t;Структура mblk_t на початку містить вказівники b_prev, b_next, які необхідні для організації двозв'язкового списку (яким є черга queue_t).
Потім йде вказівник b_cont, який використовується лише тоді, коли повідомлення входить до кортежу. Для останнього повідомлення в кортежі цей покажчик залишається нульовим.
Далі ми бачимо покажчик на блок даних b_datap, Заради якого і існує повідомлення. За ним ідуть покажчики, на область усередині буфера даних блоку. Поле b_rptr вказує місце, з якого читатимуться дані з буфера. Поле b_wptr вказує місце, з якого виконуватимуться запис у буфер.
Поля, що залишилися, носять службовий характер і не відносяться до роботи механізму передачі даних.
Нижче показано одиночне повідомлення з ім'ям m1 та блоком даних d1.
На наступному малюнку зображено кортеж із трьох повідомлень m1, m1_1, m1_2.
Функції роботи з повідомленнями mblk_t
нове повідомлення mblk_t створюється функцією:
mblk_t *allocb(int size, int pri); вона розміщує у пам'яті нове повідомлення mblk_t з блоком даних вказаного розміру розмір, другий аргумент - при не використовується в версії бібліотеки. Він має залишатися нульовим. Під час роботи функції буде виділено пам'ять під структуру нового повідомлення та викликано функцію mblk_init()яка обнулює всі поля створеного екземпляра структури і потім, за допомогою згаданої вище datab_alloc()створить буфер даних. Після чого буде виконано налаштування полів у структурі:
mp->b_datap=datab;
mp->b_rptr=mp->b_wptr=datab->db_base;
mp->b_next=mp->b_prev=mp->b_cont=NULL;На виході отримуємо нове повідомлення з ініціалізованими полями та порожнім буфером даних. Щоб додати до повідомлення дані, потрібно виконати їхнє копіювання в буфер блоку даних:
memcpy(msg->b_rptr, data, size);де дані - покажчик на джерело даних, а розмір - Їх розмір.
потім потрібно оновити покажчик на точку запису, щоб він знову вказував початок вільної області в буфері:
msg->b_wptr = msg->b_wptr + sizeЯкщо потрібно створити повідомлення з наявного буфера, без копіювання, то для цього використовується функція:
mblk_t *esballoc(uint8_t *buf, int size, int pri, void (*freefn)(void*)); Функція після створення повідомлення та структури блоку даних налаштує його покажчики на дані за адресою buf. Тобто. в даному випадку буфер даних не розташовується слідом за полями заголовка блоку даних, як це було при створенні блоку даних функцією datab_alloc(). Переданий функції буфер з даними залишиться там де був, але за допомогою покажчиків буде підстебнутий до щойно створеного заголовка блоку даних, а той відповідно до повідомлення.
До одного повідомлення mblk_t можуть бути послідовно причеплені кілька блоків даних. Це робиться функцією:
mblk_t * appendb(mblk_t *mp, const char *data, int size, bool_t pad); mp - Повідомлення до якого буде додано ще один блок даних;
дані - покажчик на блок, копію якого буде додано до повідомлення;
розмір - Розмір даних;
майданчик — прапор того, що розмір пам'яті, що виділяється, повинен бути вирівняний по межі 4 байт (додаток буде виконано нулями).
Якщо в буфері даних повідомлення достатньо місця, то нові дані будуть підклеєні за вже наявними там даними. Якщо вільного місця в буфері даних повідомлення виявиться менше, ніж розмір, то створюється нове повідомлення з достатнім розміром буфера і дані копіюються в його буфер. Це нове повідомлення, що підчіпляється до вихідного за допомогою покажчика b_cont. У цьому випадку повідомлення перетворюється на кортеж.
Якщо до кортежу потрібно додати ще один блок даних, то потрібно використовувати функцію:
void msgappend(mblk_t *mp, const char *data, int size, bool_t pad);вона знайде останнє повідомлення у кортежі (у нього b_cont буде нульовим) та викличе для цього повідомлення функцію appendb().
Дізнатися розмір даних у повідомленні або кортежі можна за допомогою функції:
int msgdsize(const mblk_t *mp);вона пробіжиться всіма повідомленнями кортежу і поверне сумарну кількість даних у буферах даних цих повідомлень. Для кожного повідомлення кількість даних обчислюється так:
mp->b_wptr - mp->b_rptrЩоб об'єднати два кортежі застосовується функція:
mblk_t *concatb(mblk_t *mp, mblk_t *newm);вона приєднує кортеж newm у хвіст кортежу mp і повертає покажчик на останнє повідомлення кортежу, що вийшов.
При необхідності, кортеж можна перетворити на одне повідомлення з єдиним блоком даних, це робиться функцією:
void msgpullup(mblk_t *mp,int len);якщо аргумент довжина дорівнює -1, то розмір буфера, що відводиться визначається автоматично. Якщо довжина позитивне число буде створено буфер цього розміру і в нього будуть скопійовані дані повідомлень кортежу. Якщо буфер закінчиться, то копіювання буде припинено. Перше повідомлення кортежу отримає буфер нового розміру із скопійованими даними. Інші повідомлення будуть видалені, а пам'ять повернуто до купи.
При видаленні структури mblk_t враховується лічильник посилань блоку даних, якщо під час виклику freeb() він виявляється дорівнює нулю, то буфер даних видаляється разом з екземпляром mblk_t, що на нього вказує.
Ініціалізація полів нового повідомлення:
void mblk_init(mblk_t *mp);Додавання до повідомлення ще однієї порції даних:
mblk_t * appendb(mblk_t *mp, const char *data, size_t size, bool_t pad);Якщо нові дані не розміщуються у вільне місце буфера даних повідомлення, то до повідомлення причіпляється окремо створене повідомлення з буфером потрібного розміру (у першому повідомленні встановлюється покажчик на додане повідомлення) повідомлення перетворюється на кортеж.
Додавання порції даних до кортежу:
void msgappend(mblk_t *mp, const char *data, size_t size, bool_t pad); Функція викликає appendb() у циклі.
Об'єднання двох кортежів в один:
mblk_t *concatb(mblk_t *mp, mblk_t *newm);Повідомлення newm буде приєднано до mp.
Створення копії одиночного повідомлення:
mblk_t *copyb(const mblk_t *mp);Повне копіювання кортежу з усіма блоками даних:
mblk_t *copymsg(const mblk_t *mp);Елементи кортежу при цьому копіюються функцією copyb().
Створення легкої копії повідомлення mblk_t. При цьому блок даних не копіюється, а збільшується лічильник його посилань db_ref:
mblk_t *dupb(mblk_t *mp);Створення легкої копії кортежу. Блоки даних не копіюються, тільки збільшуються їх лічильники посилань db_ref:
mblk_t *dupmsg(mblk_t* m);Склейка всіх повідомлень кортежу в одне повідомлення:
void msgpullup(mblk_t *mp,size_t len);Якщо аргумент довжина дорівнює -1, то розмір буфера, що відводиться визначається автоматично.
Видалення повідомлення, кортежу:
void freemsg(mblk_t *mp);Лічильник посилань блоку даних зменшується на одиницю. Якщо він при цьому досягає нуля, блок даних теж видаляється.
Підрахунок сумарного обсягу даних у повідомленні або кортежі.
size_t msgdsize(const mblk_t *mp);Вилучення повідомлення з хвоста черги:
mblk_t *ms_queue_peek_last (q);Копіювання вмісту зарезервованих полів одного повідомлення в інше повідомлення (насправді в цих полях знаходяться прапори, які використовуються медіастрімером):
mblk_meta_copy(const mblk_t *source, mblk *dest);черга queue_t
Чергу повідомлень у медіастрімері реалізовано як кільцевий двозв'язковий список. Кожен елемент списку містить покажчик блок даних з відліками сигналу. Виходить, що по черзі переміщуються лише покажчики на блок даних, тоді як дані залишаються нерухомими. Тобто. переміщуються лише посилання ними.
Структура описує чергу queue_t, показано нижче:
typedef struct _queue
{
mblk_t _q_stopper; /* "Холостой" элемент очереди, не указывает на данные, используется только для управления очередью. При инициализации очереди (qinit()) его указатели настраиваются так, чтобы они указывали на него самого. */
int q_mcount; // Количество элементов в очереди.
} queue_t;Структура містить поле - покажчик _q_stopper типу *mblk_t, він вказує на перший елемент (повідомлення) у черзі. Друге поле структури - це лічильник повідомлень, що знаходяться в черзі.
На малюнку нижче показана черга з ім'ям q1, що містить 4 повідомлення m1, m2, m3, m4.
На наступному малюнку показана черга з ім'ям q1, що містить 4 повідомлення m1, m2, m3, m4. Повідомлення m2 є головою кортежу, в який водять ще два повідомлення m2_1 та m2_2.
Функції роботи з чергами queue_t
Ініціалізацію черги:
void qinit(queue_t *q);Поле _q_stopper (далі називатимемо його "стопор") ініціалізується функцією mblk_init(), його покажчик попереднього елемента та наступного елемента налаштовуються так, щоб вони показували на нього самого. Лічильник елементів черги обнулюється.
Додавання нового елемента (повідомлення):
void putq(queue_t *q, mblk_t *m);Новий елемент m додається в кінець списку, покажчики елемента налаштовуються так, щоб стопор ставав йому наступним елементом, а він для стопора попереднім. Інкрементується лічильник елементів черги.
Вилучення елемента з черги:
mblk_t * getq(queue_t *q); витягується повідомлення, яке стоїть після стопора, лічильник елементів декрементується. Якщо черги, крім стопора, елементів немає, то повертається 0.
Вставка повідомлення в чергу:
void insq(queue_t *q, mblk_t *emp, mblk_t *mp); елемент mp вставляється перед елементом emp. Якщо emp=0, то повідомлення додається у хвіст черги.
Вилучення повідомлення з голови черги:
void remq(queue_t *q, mblk_t *mp); Лічильник елементів декрементується.
Читання покажчика на перший елемент у черзі:
mblk_t * peekq(queue_t *q); Видалення всіх елементів із черги з видаленням самих елементів:
void flushq(queue_t *q, int how);аргумент як не використовується. Лічильник елементів черги встановлюється у нуль.
Макрос читання покажчика на останній елемент черги:
mblk_t * qlast(queue_t *q);Під час роботи з чергами повідомлень слід пам'ятати, що з виклику ms_queue_put(q, m) з нульовим покажчиком повідомлення, функція зациклюється. Ваша програма зависне. Аналогічно поводиться ms_queue_next(q, m).
З'єднання фільтрів
Описана черга використовуються для передачі повідомлень від одного фільтра до іншого або від одного до кількох фільтрів. Фільтри та їх сполуки між собою утворюють спрямований граф. Вхід або вихід фільтра називатимемо узагальнюючим словом "пін". Для опису порядку з'єднань фільтрів між собою, у медіастрімері використовується поняття "сигнальної точки". Сигнальна точка це структура _MSCPoint, Що містить покажчик на фільтр і номер одного з його пінів, відповідно вона описує з'єднання одного з входів або виходів фільтра.
Сигнальна точка графа обробки даних
typedef struct _MSCPoint{
struct _MSFilter *filter; // Указатель на фильтр медиастримера.
int pin; // Номер одного из входов или выходов фильтра, т.е. пин.
} MSCPoint;
Піни фільтрів нумеруються з нуля.
З'єднання двох пінів чергою повідомлень описується структурою _MSQueue, яка містить чергу повідомлень та покажчики на дві сигнальні точки, які вона з'єднує:
typedef struct _MSQueue
{
queue_t q;
MSCPoint prev;
MSCPoint next;
}MSQueue;
Будемо називати цю структуру сигнальним лінком. Кожен фільтр медіастрімера містить таблицю вхідних лінків і таблицю вихідних лінків (MSQueue). Розмір таблиць задається при створенні фільтра, ми це вже робили за допомогою експортованої змінної типу MSFilterDescколи розробляли наш власний фільтр. Нижче показана структура, яка описує будь-який фільтр в медіастримері, MSFilter:
struct _MSFilter{
MSFilterDesc *desc; /* Указатель на дескриптор фильтра. */
/* Защищенные атрибуты, их нельзя сдвигать или убирать иначе будет нарушена работа с плагинами. */
ms_mutex_t lock; /* Семафор. */
MSQueue **inputs; /* Таблица входных линков. */
MSQueue **outputs; /* Таблица выходных линков. */
struct _MSFactory *factory; /* Указатель на фабрику, которая создала данный экземпляр фильтра. */
void *padding; /* Не используется, будет задействован если добавятся защищенные поля. */
void *data; /* Указатель на произвольную структуру для хранения данных внутреннего состояния фильтра и промежуточных вычислений. */
struct _MSTicker *ticker; /* Указатель на объект тикера, который не должен быть нулевым когда вызывается функция process(). */
/*private attributes, they can be moved and changed at any time*/
MSList *notify_callbacks; /* Список обратных вызовов, используемых для обработки событий фильтра. */
uint32_t last_tick; /* Номер последнего такта, когда выполнялся вызов process(). */
MSFilterStats *stats; /* Статистика работы фильтра.*/
int postponed_task; /*Количество отложенных задач. Некоторые фильтры могут откладывать обработку данных (вызов process()) на несколько тактов.*/
bool_t seen; /* Флаг, который использует тикер, чтобы помечать что этот экземпляр фильтра он уже обслужил на данном такте.*/
};
typedef struct _MSFilter MSFilter;
Після того, як ми в Сі-програмі з'єднали фільтри відповідно до нашого задуму (але не підключили тікер), ми тим самим створили спрямований граф, вузли якого це екземпляри структури MSFilter, а ребра це екземпляри лінків MSQueue.
Закулісна діяльність тикера
Коли я казав вам, що тикер — це фільтр джерело тактів, то була не вся правда про нього. Тікер це об'єкт, який щогодини виконує запуск функцій процес() всіх фільтрів схеми (графа), до якої він підключений. Коли ми в Сі-програмі підключаємо тикер до фільтра графа, ми показуємо тикеру граф, яким він з цього моменту керуватиме, поки ми його не відключимо. Після підключення, тикер починає оглядати довірений йому під опікою граф, складаючи список фільтрів в які до нього входять. Щоб не "порахувати" один і той же фільтр двічі він позначає виявлені фільтри, встановлюючи в них прапорець бачив. Пошук здійснюється за таблицями лінків, які є у кожного фільтра.
Під час своєї ознайомчої екскурсії графом, тикер перевіряє чи є серед фільтрів, хоча б один, який виконує роль джерела блоків даних. Якщо таких немає, то граф визнається неправильним і тикер аварійно завершує роботу.
Якщо граф виявився "правильним", у кожного знайденого фільтра для ініціалізації викликається функція preprocess(). Як тільки настає момент для чергового такту обробки (за замовчуванням кожні 10 мілісекунд), тикер викликає функцію процес() всім знайдених раніше фільтрів джерел, та був і інших фільтрів списку. Якщо фільтр має вхідні лінки, то запуск функції процес() повторюється до того часу, поки черги вхідних лінків не спорожніють. Після цього він переходить до наступного фільтра списку і "прокручує" його до звільнення вхідних лінків від повідомлень. Тікер переходить від фільтра до фільтра доки закінчиться список. У цьому обробка такту закінчується.
Тепер ми повернемося до кортежів і поговоримо про те, для чого в медіастрімер була додана така сутність. У загальному випадку, обсяг даних, необхідний алгоритму, що працює всередині фільтра не збігається і не кратний, розмір буферів даних, що надходять на вході. Наприклад, ми пишемо фільтр, який виконує швидке перетворення Фур'є, яке за визначенням може обробляти тільки блоки даних, чий розмір дорівнює ступеню двійки. Нехай це буде 512 відліків. Якщо дані генеруються телефонним каналом, буфер даних кожного повідомлення на вході буде приносити нам по 160 відліків сигналу. Є спокуса не забирати дані зі входу, доки там не виявиться необхідна кількість даних. Але в цьому випадку виникне колізія з тикером, який безуспішно намагатиметься прокрутити фільтр до спустошення вхідного лінка. Раніше ми окреслили це правило як третій принцип роботи фільтра. Відповідно до цього принципу, функція process() фільтра має забрати всі дані із вхідних черг.
Крім цього з входу не можна буде забрати лише 512 відліків, оскільки забирати можна цілими блоками, тобто. фільтру доведеться забрати 640 відліків та використавши 512 з них, залишок до накопичення нової порції даних. Таким чином, наш фільтр, крім основної своєї роботи, повинен забезпечити допоміжні дії з проміжного зберігання вхідних даних. Розробники медіастрімера і вирішення цієї спільної задачі розробили спеціальний об'єкт - MSBufferizer (буферизатор), який вирішує це завдання за допомогою кортежів.
Буферизатор (MSBufferizer)
Це об'єкт, який накопичуватиме вхідні дані всередині фільтра і почне віддавати їх в обробку, як тільки кількість інформації виявиться достатньою для прогортання алгоритму фільтра. Поки буферизатор накопичує дані, фільтр працюватиме у холостому режимі, не витрачаючи обчислювальної потужності процесора. Але як тільки функція читання з буферизатора поверне значення від нуля, функція process() фільтра починає забирати і обробляти з буферизатора дані порціями потрібного розміру, до вичерпання.
Незатребувані поки що дані залишаються в буферизаторі як перший елемент кортежу, до якого причіплюються наступні блоки вхідних даних.
Структура, яка описує буферизатор:
struct _MSBufferizer{
queue_t q; /* Очередь сообщений. */
int size; /* Суммарный размер данных находящихся в буферизаторе в данный момент. */
};
typedef struct _MSBufferizer MSBufferizer;Функції роботи з MSBufferizer
Створення нового екземпляра буферизатора:
MSBufferizer * ms_bufferizer_new(void);Виділяється пам'ять, ініціалізується в ms_bufferizer_init() та повертається покажчик.
Функція ініціалізації:
void ms_bufferizer_init(MSBufferizer *obj); Ініціалізується черга q, поле розмір встановлюється у нуль.
Додавання повідомлення:
void ms_bufferizer_put(MSBufferizer *obj, mblk_t *m); Повідомлення m додається до черги. Обчислений розмір блоків даних додається до розмір.
Перекладка в буферизатор всіх повідомлень черги даних лінка q:
void ms_bufferizer_put_from_queue(MSBufferizer *obj, MSQueue *q); Перенесення повідомлень з лінку q у буферизатор виконується за допомогою функції ms_bufferizer_put().
Читання з буферизатора:
int ms_bufferizer_read(MSBufferizer *obj, uint8_t *data, int datalen); Якщо розмір накопичених у буферизаторі даних виявляється меншим від запитаного (datalen), то функція повертає нуль, копіювання даних у data не виконується. В іншому випадку виконується послідовне копіювання даних з кортежів, що знаходяться в буферизаторі. Після копіювання кортеж видаляється та пам'ять звільняється. Копіювання закінчується в той момент, коли буде скопійовано datalen байтів. Якщо місце закінчується посеред блоку даних, то цьому повідомленні, блок даних буде скорочено до залишилася некопійованої частини. Під час наступного дзвінка копіювання продовжиться з цього місця.
Читання кількості даних, які доступні в буферизаторі:
int ms_bufferizer_get_avail(MSBufferizer *obj); Повертає поле розмір буферизатор.
Відкидання частини даних, що знаходяться у буферизаторі:
void ms_bufferizer_skip_bytes(MSBufferizer *obj, int bytes);Вказана кількість байтів даних виймається та відкидається. Відкидаються найстаріші дані.
Видалення всіх повідомлень, що знаходяться в буферизаторі:
void ms_bufferizer_flush(MSBufferizer *obj); Лічильник даних скидається у нуль.
Видалення всіх повідомлень, що знаходяться в буферизаторі:
void ms_bufferizer_uninit(MSBufferizer *obj); Обнулення лічильника не виконується.
Видалення буферизатора та звільнення пам'яті:
void ms_bufferizer_destroy(MSBufferizer *obj); Приклади використання буферизатора можна знайти у вихідному коді кількох фільтрів медіастрімера. Наприклад, у фільтрі MS_L16_ENC, який виконує перестановку байтів у відліках з мережевого порядку, в порядок хоста:
У наступній статті ми розглянемо питання оцінки навантаження на тикер і способи боротьби з надмірним обчислювальним навантаженням в медіастрімері.
Джерело: habr.com