🥇Zagłębiamy się w silnik VoIP Mediastreamer2. Część 11

Materiał artykułu pochodzi z mojego kanału Zen.

Mechanizm przesyłania danych

Blok danych dblk_t
Wiadomość mblk_t
Funkcje pracy z wiadomościami mblk_t
Kolejka queue_t
Funkcje pracy z kolejkami queue_t
Połączenie filtrów
Punkt sygnałowy grafu przetwarzania danych
Kulisy działania tickera
Bufor (MSBufferizer)
Funkcje pracy z MSBufferizer

W poprzedniej artykuł opracowaliśmy nasz własny filtr. Ten artykuł poświęcimy konstrukcji wewnętrznego mechanizmu przesyłania danych między filtrami mediastreamera. Pozwoli to w przyszłości pisać bardziej skomplikowane filtry przy mniejszych nakładach.

Mechanizm przesyłania danych

Przesyłanie danych w mediastreamerze odbywa się za pomocą kolejki opisanej przez struktury queue_t. Przez kolejki przesuwają się sekwencje wiadomości typu mblk_t, które same z siebie nie zawierają danych sygnału, a jedynie odniesienia do poprzedniej, następnej wiadomości oraz bloku danych. Ponadto, chcę szczególnie podkreślić, że istnieje pole do odniesienia do wiadomości tego samego typu, które pozwala zorganizować jednokierunkową listę wiadomości. Grupę wiadomości połączonych taką listą będziemy nazywać krotką. W ten sposób, każdy element kolejki może być pojedynczą wiadomością mblk_t, a może być także głową krotki wiadomości mblk_t. Każda wiadomość w krotce może mieć swój przypisany blok danych. Dlaczego potrzebne są krotki, omówimy trochę później.

Jak już wspomniano, samo w sobie wiadomość nie zawiera bloku danych, zamiast tego zawiera jedynie wskaźnik do obszaru pamięci, w którym przechowywany jest blok. W tej części ogólny obraz działania mediastreamera przypomina magazyn drzwi w filmie animowanym "Potwory i spółka", gdzie drzwi (odniesienia do danych – pokoje) poruszają się z oszałamiającą prędkością po wiszących taśmach transportowych, podczas gdy same pokoje pozostają nieruchome.

Teraz, przechodząc przez hierarchię od dołu do góry, przyjrzyjmy się dokładniej wymienionym bytom mechanizmu przesyłania danych w mediastreamerze.

Blok danych dblk_t

Blok danych składa się z nagłówka i bufora danych. Nagłówek opisany jest następującą strukturą,

typedef struct datab
{
unsigned char *db_base; // Wskaźnik na początek bufora danych.
unsigned char *db_lim;  // Wskaźnik na koniec bufora danych.
void (*db_freefn)(void*); // Funkcja zwalniająca pamięć przy usunięciu bloku.
int db_ref; // Licznik referencji.
} dblk_t;

Pola struktury zawierają wskaźniki na początek bufora, koniec bufora oraz funkcję usuwania bufora danych. Ostatni element w nagłówku db_ref — licznik referencji, gdy osiągnie zero, wskazuje to na potrzebę usunięcia bloku z pamięci. Jeśli blok danych został utworzony przez funkcję datab_alloc() , to bufor danych będzie umieszczony w pamięci tuż po nagłówku. W przeciwnym razie bufor może znajdować się gdzie indziej. W buforze danych będą przechowywane próbki sygnału lub inne dane, które chcemy przetwarzać filtrami.

Nowa instancja bloku danych jest tworzona za pomocą funkcji:

dblk_t *datab_alloc(int size);

Jako parametr wejściowy przekazywany jest rozmiar danych, które ma przechowywać blok. Przydzielana jest większa ilość pamięci, aby na początku przydzielonej pamięci umieścić nagłówek — strukturę datab. Ale przy użyciu innych funkcji zdarza się, że bufor danych może znajdować się niezależnie od nagłówka bloku danych. Pola struktury przy tworzeniu są konfigurowane w ten sposób, aby pole db_base wskazywało na początek obszaru danych, a db_lim na jego koniec. Licznik referencji db_ref ustawiany jest na jeden. Wskaźnik funkcji czyszczenia danych ustawiany jest na zero.

Komunikat mblk_t

Jak już wspomniano, elementy kolejki mają typ mblk_t, zdefiniowany jest w następujący sposób:

typedef struct msgb
{
  struct msgb *b_prev;   // Wskaźnik na poprzedni element listy.
  struct msgb *b_next;   // Wskaźnik na następny element listy.
  struct msgb *b_cont;   // Wskaźnik do dołączania innych wiadomości do wiadomości, aby utworzyć krotkę wiadomości.
  struct datab *b_datap; // Wskaźnik na strukturę bloku danych.
  unsigned char *b_rptr; // Wskaźnik na początek obszaru danych do odczytu danych z bufora b_datap.
  unsigned char *b_wptr; // Wskaźnik na początek obszaru danych do zapisu danych z bufora b_datap.
  uint32_t reserved1;    // Zarezerwowane pole1, mediastreamer umieszcza tam informacje serwisowe.
  uint32_t reserved2;    // Zarezerwowane pole2, mediastreamer umieszcza tam informacje serwisowe.
  #if defined(ORTP_TIMESTAMP)
  struct timeval timestamp;
  #endif
  ortp_recv_addr_t recv_addr;
} mblk_t;

Struktura mblk_t na początku zawiera wskaźniki b_prev, b_next, które są niezbędne do organizacji dwukierunkowej listy (której jest kolejka queue_t).

Następnie znajduje się wskaźnik b_cont, który jest używany tylko wtedy, gdy wiadomość wchodzi do krotki. Dla ostatniej wiadomości w krotce ten wskaźnik pozostaje zerowy.

Następnie widzimy wskaźnik na blok danych b_datap, dla którego istnieje wiadomość. Następne są wskaźniki na obszar w obrębie bufora danych bloku. Pole b_rptr określa miejsce, z którego będą odczytywane dane z bufora. Pole b_wptr określa miejsce, z którego będą zapisywane dane do bufora.

Pozostałe pola mają charakter operacyjny i nie dotyczą działania mechanizmu przesyłania danych.

Poniżej przedstawiono pojedynczą wiadomość o nazwie m1 i bloku danych d1.

Na następnym rysunku przedstawiono krotkę z trzema wiadomościami m1, m1_1, m1_2.

Funkcje pracy z wiadomościami mblk_t

Nowa wiadomość mblk_t jest tworzona przez funkcję:

mblk_t *allocb(int size, int pri);

Tworzy nowe wiadomości w pamięci mblk_t z blokiem danych o określonym rozmiarze rozmiar, drugi argument — pri nie jest używany w omawianej wersji biblioteki. Powinien pozostać zerowy. W trakcie działania funkcji pamięć będzie przydzielana pod strukturę nowej wiadomości, a funkcja mblk_init(), która wyzeruje wszystkie pola nowo utworzonej struktury, a następnie, za pomocą wspomnianego wcześniej, datab_alloc(), utworzy bufor danych. Następnie zostaną ustawione pola w strukturze:

mp->b_datap=datab;
mp->b_rptr=mp->b_wptr=datab->db_base;
mp->b_next=mp->b_prev=mp->b_cont=NULL;

Na wyjściu otrzymujemy nową wiadomość z zainicjowanymi polami i pustym buforem danych. Aby dodać do wiadomości dane, należy skopiować je do bufora bloku danych:

memcpy(msg->b_rptr, data, size);

gdzie data — wskaźnik na źródło danych, a rozmiar — ich rozmiar.
Następnie należy zaktualizować wskaźnik do miejsca zapisu, aby ponownie wskazywał na początek wolnego obszaru w buforze:

msg->b_wptr = msg->b_wptr + size

Jeśli wymagana jest utworzenie wiadomości z już istniejącego bufora, bez kopiowania, używa się do tego funkcji:

mblk_t *esballoc(uint8_t *buf, int size, int pri, void (*freefn)(void*));

Funkcja po utworzeniu wiadomości i struktury bloku danych ustawi jej wskaźniki na dane pod adresem buf. Tzn. w tym przypadku bufor danych nie znajduje się tuż za polami nagłówka bloku danych, jak miało to miejsce przy tworzeniu bloku danych przez funkcję. datab_alloc()Przekazany do funkcji bufor danych pozostanie w swojej pierwotnej lokalizacji, ale dzięki wskaźnikom zostanie skierowany do właśnie utworzonego nagłówka bloku danych, a ten z kolei do wiadomości.

Do jednej wiadomości mblk_t może być dołączonych kilka bloków danych. Robi się to za pomocą funkcji:

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

mp — wiadomość, do której zostanie dodany kolejny blok danych;
data — wskaźnik na blok, kopia którego zostanie dodana do wiadomości;
rozmiar — rozmiar danych;
pad — flaga wskazująca, czy rozmiar przydzielanej pamięci ma być wyrównany do granicy 4 bajtów (uzupełnienie będzie wykonane zerami).

Jeżeli w istniejącym buforze danych wiadomości jest wystarczająco miejsca, nowe dane zostaną dołączone za danymi, które już się tam znajdują. Jeśli w buforze danych wiadomości zwolnionego miejsca jest mniej niż rozmiar, tworzona jest nowa wiadomość z wystarczającym rozmiarem bufora, a dane są kopiowane do jej bufora. Ta nowa wiadomość jest dołączana do oryginalnej za pomocą wskaźnika b_cont. W takim przypadku wiadomość przekształca się w krotkę.

Jeżeli do krotki wymaga się dołączenia kolejnego bloku danych, należy użyć funkcji:

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

ona odszuka ostatnią wiadomość w krotce (ma ona b_cont będzie zerowa) i wywoła dla tej wiadomości funkcję appendb().

Aby poznać rozmiar danych w wiadomości lub krotce, można użyć funkcji:

int msgdsize(const mblk_t *mp);

ona przejdzie przez wszystkie wiadomości krotki i zwróci całkowitą ilość danych w buforach danych tych wiadomości. Dla każdej wiadomości liczba danych jest obliczana w następujący sposób:

 mp->b_wptr - mp->b_rptr

Aby połączyć dwie krotki, stosuje się funkcję:

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

ona dołącza krotkę newm na końcu krotki mp i zwraca wskaźnik do ostatniej wiadomości otrzymanej krotki.

W razie potrzeby, krotkę można przekształcić w jedną wiadomość z pojedynczym blokiem danych, robi się to za pomocą funkcji:

void msgpullup(mblk_t *mp,int len);

jeżeli argument len jest równy -1, to rozmiar przydzielanego bufora określany jest automatycznie. Jeśli len Jeśli jest to liczba dodatnia, utworzony zostanie bufor o tym rozmiarze, do którego skopiowane zostaną dane wiadomości krotki. Jeśli bufor się zapełni, kopiowanie zostanie przerwane. Pierwsza wiadomość krotki otrzyma bufor nowego rozmiaru z skopiowanymi danymi. Pozostałe wiadomości zostaną usunięte, a pamięć zwrócona do sterty.

Podczas usuwania struktury mblk_t uwzględniany jest licznik referencji bloku danych, jeśli podczas wywoływania freeb() wynosi zero, to bufor danych zostaje usunięty wraz z instancją mblk_t, na którą wskazuje.

Inicjalizacja pól nowej wiadomości:

void mblk_init(mblk_t *mp);

Dodanie do wiadomości kolejnej porcji danych:

mblk_t * appendb(mblk_t *mp, const char *data, size_t size, bool_t pad);

Jeśli nowe dane nie mieszczą się w wolnym miejscu bufora danych wiadomości, do wiadomości dołączana jest osobno utworzona wiadomość z buforem odpowiedniego rozmiaru (w pierwszej wiadomości ustawiany jest wskaźnik na dodaną wiadomość), co przekształca wiadomość w krotkę.

Dodanie porcji danych do krotki:

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

Funkcja wywołuje appendb() w pętli.

Scalanie dwóch krotek w jedną:

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

Komunikat newm zostanie dołączone do mp.

Utworzenie kopii pojedynczej wiadomości:

mblk_t *copyb(const mblk_t *mp);

Pełne skopiowanie krotki ze wszystkimi blokami danych:

mblk_t *copymsg(const mblk_t *mp);

Elementy krotki są kopiowane przez funkcję copyb().

Utworzenie lekkiej kopii wiadomości mblk_t. Przy tym blok danych nie jest kopiowany, lecz zwiększa się licznik jego referencji db_ref:

mblk_t *dupb(mblk_t *mp);

Utworzenie lekkiej kopii krotki. Bloki danych nie są kopiowane, tylko zwiększają się ich liczniki referencji db_ref:

mblk_t *dupmsg(mblk_t* m);

Scalenie wszystkich wiadomości krotki w jedną wiadomość:

void msgpullup(mblk_t *mp,size_t len);

Jeśli argument len wynosi -1, to rozmiar wydzielanego bufora określany jest automatycznie.

Usunięcie wiadomości, krotki:

void freemsg(mblk_t *mp);

Licznik referencji bloku danych zmniejsza się o jeden. Jeśli osiągnie zero, blok danych również zostaje usunięty.

Policzenie całkowitej objętości danych w wiadomości lub krotce.

size_t msgdsize(const mblk_t *mp);

Wyciągnięcie wiadomości z końca kolejki:

mblk_t *ms_queue_peek_last (q);

Kopiowanie zawartości zarezerwowanych pól jednej wiadomości do innej wiadomości (w rzeczywistości w tych polach znajdują się flagi używane przez strumień mediów):

mblk_meta_copy(const mblk_t *source, mblk *dest);

Kolejka queue_t

Kolejka wiadomości w mediastreamerze jest zrealizowana jako okrągła lista dwukierunkowa. Każdy element listy zawiera wskaźnik na blok danych z pomiarami sygnału. Oznacza to, że w kolejce poruszają się jedynie wskaźniki do bloku danych, podczas gdy same dane pozostają nieruchome. Tzn. poruszają się tylko odnośniki do nich.
Struktura opisująca kolejkę queue_t, pokazana poniżej:

typedef struct _queue
{
   mblk_t _q_stopper; 
   /* "Pusty" element kolejki, nie wskazuje na dane, służy tylko do zarządzania kolejką. Podczas inicjalizacji kolejki (qinit()) jego wskaźniki są ustawione tak, aby wskazywały na niego samego. */
   int q_mcount;        // Liczba elementów w kolejce.
} queue_t;

Struktura zawiera pole — wskaźnik _q_stopper typu *mblk_t, wskazuje na pierwszy element (wiadomość) w kolejce. Drugie pole struktury to licznik wiadomości znajdujących się w kolejce.
Na ilustracji poniżej przedstawiona jest kolejka o nazwie q1, zawierająca 4 wiadomości m1, m2, m3, m4.

Na następnej ilustracji przedstawiona jest kolejka o nazwie q1, zawierająca 4 wiadomości m1, m2, m3, m4. Wiadomość m2 jest głową krotki, do której prowadzą jeszcze dwie wiadomości m2_1 i m2_2.

Funkcje pracy z kolejkami queue_t

Inicjalizacja kolejki:

void qinit(queue_t *q);

Pole _q_stopper (dalej nazywane "stoporem") jest inicjalizowane przez funkcję mblk_init(), jego wskaźniki do poprzedniego i następnego elementu są ustawiane tak, aby wskazywały na niego samego. Licznik elementów w kolejce jest zerowany.

Dodanie nowego elementu (wiadomości):

void putq(queue_t *q, mblk_t *m);

Nowy element m jest dodawany na koniec listy, wskaźniki elementu są ustawiane tak, aby stopor stawał się jego następnym elementem, a on dla stopora poprzednim. Licznik elementów w kolejce jest inkrementowany.

Wyciąganie elementu z kolejki:

mblk_t * getq(queue_t *q);

wyciągane jest to wiadomość, która znajduje się po stoporze, licznik elementów jest dekrementowany. Jeśli w kolejce, poza stoporem, nie ma żadnych elementów, zwracane jest 0.

Wstawienie wiadomości do kolejki:

void insq(queue_t *q, mblk_t *emp, mblk_t *mp);

Element mp jest wstawiany przed elementem emp. Jeśli emp=0, to wiadomość jest dodawana do końca kolejki.

Wyciąganie wiadomości z głowy kolejki:

void remq(queue_t *q, mblk_t *mp);

Licznik elementów jest dekrementowany.

Odczyt wskaźnika na pierwszy element w kolejce:

mblk_t * peekq(queue_t *q);

Usunięcie wszystkich elementów z kolejki z usunięciem samych elementów:

void flushq(queue_t *q, int how);

Argument how nie jest używane. Licznik elementów w kolejce zostaje ustawiony na zero.

Makro odczytu wskaźnika na ostatni element kolejki:

mblk_t * qlast(queue_t *q);

Przy pracy z kolejkami wiadomości należy pamiętać, że wywołanie ms_queue_put(q, m) z zerowym wskaźnikiem na wiadomość powoduje zapętlenie. Twoja aplikacja zawiesi się. Podobne zachowanie ma ms_queue_next(q, m).

Połączenie filtrów

Opisana powyżej kolejka jest używana do przesyłania wiadomości między jednym filtrem a drugim lub od jednego do wielu filtrów. Filtry i ich połączenia tworzą graf skierowany. Wejście lub wyjście filtra nazywamy ogólnym terminem „pin”. Do opisu porządku połączeń filtrów używa się pojęcia „punkt sygnałowy”. Punkt sygnałowy to struktura _MSCPoint, która zawiera wskaźnik do filtra i numer jednego z jego pinów, w związku z tym opisuje połączenie jednego z wejść lub wyjść filtra.

Punkt sygnałowy grafu przetwarzania danych

typedef struct _MSCPoint{
struct _MSFilter *filter; // Wskaźnik do filtra mediastreamera.
int pin; // Numer jednego z wejść lub wyjść filtra, tzn. pin.
} MSCPoint;

Piny filtrów numerowane są od zera.

Połączenie dwóch pinów za pomocą kolejki wiadomości opisuje struktura _MSQueue, która zawiera kolejkę wiadomości oraz wskaźniki do dwóch punktów sygnałowych, które łączy:

typedef struct _MSQueue
{
queue_t q;
MSCPoint prev;
MSCPoint next;
}MSQueue;

Nazwijmy tę strukturę połączeniem sygnałowym. Każdy filtr mediastreamera zawiera tabelę połączeń wejściowych oraz tabelę połączeń wyjściowych (MSQueue). Rozmiar tabel jest ustalany podczas tworzenia filtra, co już zrobiliśmy przy pomocy eksportowanej zmiennej typu MSFilterDesc, gdy opracowywaliśmy nasz własny filtr. Poniżej przedstawiona jest struktura opisująca każdy filtr w mediastreamerze, MSFilter:


struct _MSFilter{
    MSFilterDesc *desc;    
    /* Wskaźnik na deskryptor filtra. */
    /* Zabezpieczone atrybuty, ich nie można przesuwać ani usuwać, w przeciwnym razie praca z wtyczkami będzie naruszona. */
    ms_mutex_t lock;      /* Semafor. */
    MSQueue **inputs;     /* Tabela wejściowych łączy. */
    MSQueue **outputs;    /* Tabela wyjściowych łączy. */
    struct _MSFactory *factory; /* Wskaźnik na fabrykę, która stworzyła dany egzemplarz filtra. */
    void *padding;              /* Nie używane, zostanie użyte, jeśli zostaną dodane zabezpieczone pola. */
    void *data;                 /* Wskaźnik na dowolną strukturę do przechowywania danych stanu wewnętrznego filtra i obliczeń pośrednich. */
    struct _MSTicker *ticker;   /* Wskaźnik na obiekt tickera, który nie powinien być zerowy, gdy wywoływana jest funkcja process(). */
    /*prywatne atrybuty, mogą być przenoszone i zmieniane w każdej chwili*/
    MSList *notify_callbacks; /* Lista callbacków używanych do przetwarzania zdarzeń filtra. */
    uint32_t last_tick;       /* Numer ostatniego taktu, gdy wywoływana była funkcja process(). */
    MSFilterStats *stats;     /* Statystyka pracy filtra.*/
    int postponed_task; /*Liczba odłożonych zadań. Niektóre filtry mogą odkładać przetwarzanie danych (wywołanie process()) na kilka taktów.*/
    bool_t seen;  /* Flaga, którą używa ticker, aby oznaczyć, że ten egzemplarz filtra został już obsłużony w danym takcie.*/
};
typedef struct _MSFilter MSFilter;

Po połączeniu filtrów w programie C zgodnie z naszym zamysłem (ale bez podłączenia tickera), stworzyliśmy w ten sposób graf skierowany, którego węzły to egzemplarze struktury MSFilter, a krawędzie to egzemplarze łączy MSQueue.

Kulisy działania tickera

Kiedy mówiłem, że ticker to filtr źródłowy taktów, nie była to cała prawda o nim. Ticker to obiekt, który regularnie uruchamia funkcje process() wszystkich filtrów schemy (g grafu), do której jest podłączony. Kiedy w programie C podłączamy ticker do filtra w grafie, pokazujemy tickerowi graf, którym od tej pory będzie zarządzać, dopóki go nie odłączymy. Po podłączeniu ticker zaczyna przeszukiwać zaufany mu graf, sporządzając listę filtrów, które do niego wchodzą. Aby nie "zliczać" tego samego filtra dwa razy, oznacza wykryte filtry, ustawiając w nich znacznik seen. Przeszukiwanie odbywa się po tabelach łączy, które ma każdy filtr.

Podczas swojej wstępnej wizyty w grafie, ticker sprawdza, czy wśród filtrów znajduje się przynajmniej jeden, który pełni rolę źródła bloków danych. Jeśli taki nie zostanie znaleziony, graf uznaje się za niewłaściwy, a ticker kończy działanie awaryjnie.

Jeśli graf okazał się "poprawny", dla każdego znalezionego filtra, w celu inicjalizacji, wywoływana jest funkcja preprocess(). Gdy nadejdzie czas na kolejny krok przetwarzania (domyślnie co 10 milisekund), ticker wywołuje funkcję process() dla wszystkich wcześniej znalezionych filtrów źródłowych, a następnie dla pozostałych filtrów z listy. Jeśli filtr ma wejściowe łącza, uruchomienie funkcji process() jest powtarzane, aż kolejki wejściowych łączy się opróżnią. Następnie przechodzi do następnego filtra na liście i "przewija" go, aż wejściowe łącza zostaną zwolnione od wiadomości. Ticker przechodzi od filtra do filtra, aż lista się wyczerpie. Na tym kończy się przetwarzanie kroku.

Teraz wrócimy do krotek i porozmawiamy o tym, po co w mediastreamerze dodano taki byt. W ogólnym przypadku, objętość danych, potrzebna algorytmowi działającemu w filtrze, nie zgadza się i nie jest wielokrotnością rozmiaru buforów danych wchodzących. Na przykład piszemy filtr, który wykonuje szybkie przekształcenie Fouriera, które z definicji może przetwarzać tylko bloki danych, których rozmiar jest potęgą liczby dwa. Przyjmijmy, że to 512 próbek. Jeśli dane są generowane przez kanał telefoniczny, to bufor danych każdej wiadomości na wejściu przynosi nam po 160 próbek sygnału. Jest pokusa, by nie zabierać danych z wejścia, dopóki nie uzbiera się wymagana liczba danych. Ale w takim przypadku wystąpi kolizja z tickerem, który bezskutecznie będzie próbował przewinąć filtr do opróżnienia wejściowego łącza. Wcześniej oznaczyliśmy tę zasadę jako trzecią zasadę działania filtra. Zgodnie z tą zasadą funkcja process() filtra powinna pobrać wszystkie dane z kolejek wejściowych.

Oprócz tego z wejścia nie można będzie pobrać jedynie 512 próbek, ponieważ można je pobierać tylko w całości, tzn. filtr musi pobrać 640 próbek, a wykorzystując 512 z nich, pozostałość do zgromadzenia nowej porcji danych. W ten sposób nasz filtr, oprócz swojej głównej pracy, musi zapewnić dodatkowe działania związane z tymczasowym przechowywaniem danych wejściowych. Twórcy mediastreamera do rozwiązania tego wspólnego problemu opracowali specjalny obiekt — MSBufferizer (bufor), który rozwiązuje tę kwestię za pomocą krotek.

Bufor (MSBufferizer)

Jest to obiekt, który będzie gromadził dane wejściowe wewnątrz filtra i zacznie je wydawać do przetwarzania, gdy tylko ilość informacji okaże się wystarczająca do przeprowadzenia algorytmu filtra. Podczas gdy bufor gromadzi dane, filtr będzie działał w trybie jałowym, nie zużywając mocy obliczeniowej procesora. Ale gdy tylko funkcja odczytu z bufora zwróci wartość różną od zera, funkcja process() filtra zaczyna pobierać i przetwarzać dane z bufora w porcjach odpowiedniego rozmiaru, aż do ich wyczerpania.
Niepotrzebne na razie dane pozostają w buforze jako pierwszy element krotki, do której dołączają następne bloki danych wejściowych.

Struktura opisująca bufor:

struct _MSBufferizer{
queue_t q; /* Kolejka komunikatów. */
int size; /* Łączny rozmiar danych znajdujących się w buforze w danym momencie. */
};
typedef struct _MSBufferizer MSBufferizer;

Funkcje pracy z MSBufferizer

Tworzenie nowej instancji bufora:

MSBufferizer * ms_bufferizer_new(void);

Przydzielana jest pamięć, inicjalizowany w ms_bufferizer_init() i zwracany jest wskaźnik.

Funkcja inicjalizacji:

void ms_bufferizer_init(MSBufferizer *obj);

Inicjalizowana jest kolejka q, pole rozmiar ustawiane jest na zero.

Dodawanie komunikatu:

void ms_bufferizer_put(MSBufferizer *obj, mblk_t *m);

Komunikat m jest dodawany do kolejki. Obliczony rozmiar bloków danych jest dodawany do rozmiar.

Zrzut wszystkich komunikatów z kolejki danych łącza do bufora q:

void ms_bufferizer_put_from_queue(MSBufferizer *obj, MSQueue *q);

Przenoszenie komunikatów z łącza q do bufora odbywa się za pomocą funkcji ms_bufferizer_put().

Odczyt z bufora:

int ms_bufferizer_read(MSBufferizer *obj, uint8_t *data, int datalen);

Jeśli rozmiar zgromadzonych w buforze danych okaże się mniejszy niż żądany (datalen), funkcja zwraca zero, a kopiowanie danych do data nie jest wykonywane. W przeciwnym razie następuje sekwencyjne kopiowanie danych z krotek znajdujących się w buforze. Po skopiowaniu krotka jest usuwana, a pamięć zwalniana. Kopiowanie kończy się w momencie skopiowania datalen bajtów. Jeśli miejsce kończy się w połowie bloku danych, w tym komunikacie blok danych jest skracany do pozostałej, nie skopiowanej części. Przy następnym wywołaniu, kopiowanie będzie kontynuowane od tego miejsca.

Odczyt liczby danych, które są dostępne w danym momencie w buforze:

int ms_bufferizer_get_avail(MSBufferizer *obj);

Zwraca pole rozmiar bufora.

Odrzucanie części danych znajdujących się w buforze:

void ms_bufferizer_skip_bytes(MSBufferizer *obj, int bytes);

Określona liczba bajtów danych jest ekstrakowana i odrzucana. Odrzucane są najstarsze dane.

Usunięcie wszystkich wiadomości znajdujących się w buforze:

void ms_bufferizer_flush(MSBufferizer *obj);

Licznik danych jest resetowany do zera.

Usunięcie wszystkich wiadomości znajdujących się w buforze:

void ms_bufferizer_uninit(MSBufferizer *obj);

Resetowanie licznika nie jest wykonywane.

Usunięcie buforu i zwolnienie pamięci:

void ms_bufferizer_destroy(MSBufferizer *obj);

Przykłady użycia buforu można znaleźć w kodzie źródłowym kilku filtrów mediastreamera. Na przykład w filtrze MS_L16_ENC, który wykonuje przestawienie bajtów w pomiarach z porządku sieciowego do porządku hosta: l16.c

W następnym artykule omówimy kwestię oceny obciążenia na tickerze i sposoby radzenia sobie z nadmiernym obciążeniem obliczeniowym w mediastreamerze.

Źródło: habr.com