Изучаване на VoIP движка Mediastreamer2. Част 11

Материалът на статията е взет от моя дзен-канал.

Механизъм за прехвърляне на данни

• Данни блок 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 с блок данни с посочения размер размер, вторият аргумент — 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 — указател към източника на данните, а размер — тяхната големина.
След това е необходимо да се актуализира указателят на записа, така че отново да сочи към началото на свободната област в буфера:

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 — указател към блока, копията на който ще бъде добавена в съобщението;
размер — размер на данните;
pad — флаг, че размерът на заделената памет трябва да бъде подравнен по границата от 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);

ако аргументът 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).

Свързване на филтри

Описаната по-горе опашка се използва за предаване на съобщения от един филтър на друг или от един на няколко филтри. Филтрите и техните връзки образуват насочена графика. Входът или изходът на филтъра ще наречем общото наименование "пин". За описание на реда на свързване на филтрите помежду им, в медийната стрийминг система се използва понятието "сигнална точка". Сигналната точка е структура _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, те могат да бъдат преместени и променени по всяко време*/
    MSList *notify_callbacks; /* Списък на обратни повиквания, използвани за обработка на събития от филтъра. */
    uint32_t last_tick;       /* Номер на последната такт, когато е извикана функцията process(). */
    MSFilterStats *stats;     /* Статистика на работата на филтъра.*/
    int postponed_task; /*Брой отложени задачи. Някои филтри могат да отлагат обработката на данни (извикване на process()) за няколко такта.*/
    bool_t seen;  /* Флаг, който тикерът използва, за да обозначи, че този екземпляр на филтъра вече е обслужен на този такт.*/
};
typedef struct _MSFilter MSFilter;

След като в C-програмата свързахме филтрите в съответствие с нашия замисъл (но не свързахме тикера), ние по този начин създадохме ориентиран граф, чийто върхове са екземпляри на структурата MSFilter, а ребрата са екземпляри на връзките. MSQueue.

Задкулисна дейност на тикера

Когато ви казах, че тикерът е източник на тактове за филтър, това не беше цялата истина за него. Тикерът е обект, който по график изпълнява функции process() на всички филтри в схемата (графа), към която е свързан. Когато в C-програмата свързваме тикера с филтъра на графа, ние показваме на тикера графа, който от този момент той ще управлява, докато не го отключим. След свързването тикерът започва да преглежда поверения му граф, съставяйки списък на филтрите, в които влиза. За да не "преброи" един и същ филтър два пъти, той помага откритите филтри, като задава флаг 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, полето размер се задава на нула.

Добавяне на съобщение:

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, който извършва преоредяване на байтовете в отсчетите от мрежов ред в реда на хоста: l16.c

В следващата статия ще разгледаме въпроса за оценка на натоварването на тикера и начините за справяне с прекомерната изчислителна натовареност в медиастримера.

Източник: habr.com