Материалът на статията е взет от моя .
Механизм за прехвърляне на данни
- Данни блок 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 с блок данни с указан размер size, вторият аргумент — pri не се използва в разглежданата версия на библиотеката. Той трябва да остане нулев. По време на изпълнението на функцията ще бъде заделена памет за структурата на новото съобщение и ще бъде извикана функцията 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);където 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 — съобщението, към което ще бъде добавен нов блок данни;
data — указател на блока, чиято копия ще бъде добавена в съобщението;
size — размер на данните;
pad — флаг, който указва, че размерът на заделената памет трябва да бъде подравнен по граница от 4 байта (допълнението ще бъде извършено с нули).
Ако в наличния буфер на съобщението има достатъчно място, новите данни ще бъдат добавени след вече съществуващите там данни. Ако свободното място в буфера на съобщението е по-малко от size, то се създава ново съобщение, с достатъчен размер на буфера и данните се копират в неговия буфер. Това ново съобщение се прикрепя към оригиналното с помощта на указател 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);ако аргументът len е равен на -1, то размерът на заделената памет се определя автоматично. Ако len Ако е положително число, ще бъде създаден буфер с този размер и в него ще бъдат копирани данните на съобщенията от кортежа. Ако буферът свърши, копирането ще прекрати. Първото съобщение от кортежа ще получи буфер с нов размер с копираните данни. Останалите съобщения ще бъдат изтрити, а паметта ще бъде върната в купа.
При изтриване на структурата 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);Ако аргументът 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);Аргумент how не се използва. Броячът на елементите в опашката се задава на нула.
Макрос за четене на указателя към последния елемент на опашката:
mblk_t * qlast(queue_t *q);При работа с опашки от съобщения трябва да се има предвид, че при извикване на ms_queue_put(q, m) с нулев указател за съобщение, функцията ще влезе в безкраен цикъл. Вашата програма ще зацикли. По подобен начин се държи ms_queue_next(q, m).
Свързване на филтри
Описаната по-горе опашка се използва за предаване на съобщения от един филтър към друг или от един към flera филтри. Филтрите и техните свързвания образуват насочена графична структура. Входът или изходът на филтъра се нарича обобщаваща дума "пин". За описване на реда на свързванията между филтрите, в медиастримера се използва концепцията "сигнална точка". Сигналната точка е структура _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.
Задкулисна дейност на тикера
Когато ви казах, че тикерът е филтър, източник на тактове, не беше цялата истина за него. Тикерът е обект, който по часовник извършва стартиране на функции process() на всички филтри от схемата (графа), към която е свързан. Когато в Си-програмата свързваме тикера с филтъра на графа, ние показваме на тикера графа, който той ще управлява от този момент нататък, докато не го отключим. След свързването тикерът започва да преглежда графа, на който е поверен, и съставя списък на филтрите, в които е включен. За да не "преброява" един и същ филтър два пъти, той маркира откритите филтри, задавайки им флаг seen. Търсенето се извършва по таблиците на линковете, които всеки филтър има.
По време на своята запознаваща екскурзия по графа, тикер проверява дали сред филтри има поне един, който изпълнява роля на източник на блокове данни. Ако не се намери такъв, графът се признава за неправилен и тикерът аварийно прекратява работа.
Ако графът се е оказал "правилен", за всеки намерен филтър, за инициализация, се извиква функцията preprocess(). В момента, в който дойде ред за следващия такт на обработка (по подразбиране на всеки 10 милисекунди), тикерът извиква функцията process() за всички предварително намерени филтри източници, а след това и за останалите филтри в списъка. Ако филтърът има входни линкове, то изпълнението на функцията process() се повтаря, докато входните линкове не се изпразнят. След това, той преминава към следващия филтър в списъка и "издърпва" входните линкове от съобщения. Тикерът преминава от един филтър на друг, докато не приключи списъкът. С това обработката на такта завършва.
Сега ще се върнем към кортежите и ще обсъдим защо в медиастриймера е добавена такава същност. В общия случай, обемът данни, необходим на алгоритъма, работещ в рамките на филтъра, не съвпада и не е кратен на размера на буферите данни, постъпващи на входа. Например, пишем филтър, който извършва бързо преобразование на Фурие, което по определение може да обработва само блокове данни, чийто размер е равен на степен на двойка. Да предположим, че това е 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, поле size и се задава на нула.
Добавяне на съобщение:
void ms_bufferizer_put(MSBufferizer *obj, mblk_t *m); Съобщението m се добавя в опашката. Изчислената големина на блоковете данни се добавя към size.
Прехвърляне на всички съобщения от опашката на данни на връзката в буферизатора 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); Връща полето size на буферизатора.
Пропускане на част от данните, намиращи се в буферизатора:
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
