Проучване на VoIP двигателя на Mediastreamer2. част 11

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

Механизъм за движение на данни

• Блок данни dblk_t
• Съобщение mblk_t
• Функции за работа със съобщения mblk_t
• Опашка queue_t
• Функции за работа с опашки queue_t
• Свързващи филтри
• Сигнална точка на графика за обработка на данни
• Зад кулисите дейности на тикер
• Буферизатор (MSBufferizer)
• Функции за работа с MSBufferizer

В последния Статия разработихме собствен филтър. Тази статия ще се съсредоточи върху вътрешния механизъм за преместване на данни между филтрите за медийни стриймъри. Това ще ви позволи да пишете сложни филтри с по-малко усилия в бъдеще.

Механизъм за движение на данни

Движението на данни в медийния стример се извършва с помощта на опашки, описани от структурата queue_t. Поредици от съобщения като mblk_t, които сами по себе си не съдържат сигнални данни, а само връзки към предишното, следващото съобщение и към блока с данни. Освен това искам специално да подчертая, че има и поле за връзка към съобщение от същия тип, което ви позволява да организирате единично свързан списък от съобщения. Ще наричаме група съобщения, обединени от такъв списък, кортеж. Така всеки елемент от опашката може да бъде едно съобщение mblk_tи може би главата на кортеж от съобщения mblk_t. Всяко съобщение от кортеж може да има свой собствен блок с данни на отделенията. Ще обсъдим защо са необходими кортежи малко по-късно.

Както бе споменато по-горе, самото съобщение не съдържа блок от данни; вместо това съдържа само указател към областта на паметта, където се съхранява блокът. В тази част цялостната картина на работата на медийния стример напомня на склада за врати в анимационния филм „Monsters, Inc.“, където вратите (връзки към данни - стаи) се движат с безумна скорост по горни конвейери, докато самите стаи останете неподвижни.

Сега, като се движим по йерархията отдолу нагоре, нека разгледаме подробно изброените обекти на механизма за предаване на данни в медийния стример.

Блок данни 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_пред, b_следващ, които са необходими за организиране на двойно свързан списък (който е опашка queue_t).

След това идва показалецът b_продължение, който се използва само когато съобщението е част от кортеж. За последното съобщение в кортежа този указател остава нула.

След това виждаме указател към блок с данни 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_продължение. В този случай съобщението се превръща в кортеж.

Ако трябва да добавите друг блок от данни към кортежа, тогава трябва да използвате функцията:

void msgappend(mblk_t *mp, const char *data, int size, bool_t pad);

тя ще намери последното съобщение в кортежа (той има b_продължение ще бъде нула) и ще извика функцията за това съобщение appendb().

Можете да разберете размера на данните в съобщение или кортеж, като използвате функцията:

int msgdsize(const mblk_t *mp);

той ще премине през всички съобщения в кортежа и ще върне общото количество данни в буферите за данни на тези съобщения. За всяко съобщение количеството данни се изчислява, както следва:

 mp->b_wptr - mp->b_rptr

За да комбинирате два кортежа, използвайте функцията:

mblk_t *concatb(mblk_t *mp, mblk_t *newm);

тя добавя кортежа новост към опашката на кортежа mp и връща указател към последното съобщение от получения кортеж.

Ако е необходимо, кортежът може да се превърне в едно съобщение с един блок от данни; това се прави от функцията:

void msgpullup(mblk_t *mp,int len);

ако аргумент дъл е равно на -1, тогава размерът на разпределения буфер се определя автоматично. Ако дъл е положително число, ще бъде създаден буфер с този размер и данните от кортежното съобщение ще бъдат копирани в него. Ако буферът се изчерпи, копирането ще спре там. Първото съобщение от кортежа ще получи буфер с нов размер с копираните данни. Останалите съобщения ще бъдат изтрити и паметта ще бъде върната в купчината.

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

Съобщение новост ще бъде приложен към 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_запушалка тип *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_запушалка (по-нататък ще го наричаме „стопер“) се инициализира от функцията 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).

Свързващи филтри

Описаната по-горе опашка се използва за предаване на съобщения от един филтър към друг или от един към няколко филтъра. Филтрите и техните връзки образуват насочена графа. Входът или изходът на филтъра ще се нарича общата дума "pin". За да опише реда, в който филтрите са свързани един с друг, медийният стример използва концепцията за „сигнална точка“. Сигналната точка е структура _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;

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

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

Когато ви казах, че тикерът е филтър за източника на тиковете, това не беше цялата истина за това. Тикерът е обект, който изпълнява функции на часовника процес() всички филтри на веригата (графа), към която е свързан. Когато свържем тикер към графичен филтър в C програма, ние показваме на тикера графиката, която той ще контролира от сега нататък, докато не го изключим. След като се свърже, тикерът започва да изследва графиката, поверена на неговите грижи, съставяйки списък с филтри, които го включват. За да не „брои“ един и същи филтър два пъти, той маркира откритите филтри, като поставя отметка в тях видян. Търсенето се извършва с помощта на таблиците с връзки, които има всеки филтър.

По време на въвеждащата си обиколка на графиката тикерът проверява дали сред филтрите има поне един, който действа като източник на блокове с данни. Ако няма такива, тогава графиката се счита за неправилна и тикерът се срива.

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

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

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