Wir lernen den VoIP-Engine Mediastreamer2. Teil 11

Der Artikelinhalt stammt von meinem Zen-Kanal.

Datenverschiebungsmechanismus

• Datenblock dblk_t
• Nachricht mblk_t
• Funktionen zur Arbeit mit Nachrichten mblk_t
• Warteschlange queue_t
• Funktionen zur Arbeit mit Warteschlangen queue_t
• Filterverbindungen
• Signalpunkt im Datenverarbeitungsgraphen
• Hintergrundaktivitäten des Tickers
• Puffer (MSBufferizer)
• Funktionen zur Arbeit mit MSBufferizer

Letzte Woche Artikel Wir haben unseren eigenen Filter entwickelt. In diesem Artikel widmen wir uns dem internen Mechanismus zur Verschiebung von Daten zwischen den Filtern des Mediastreamers. Dies wird es uns in Zukunft ermöglichen, ausgeklügelte Filter mit weniger Aufwand zu schreiben.

Datenverschiebungsmechanismus

Die Datenverschiebung im Mediastreamer erfolgt über Warteschlangen, die durch die Struktur queue_tbeschrieben werden. Über die Warteschlangen werden Serien von Nachrichten des Typs mblk_tübertragen, die selbst keine Signaldaten enthalten, sondern lediglich Verweise auf die vorherige, folgende Nachricht und auf den Datenblock. Außerdem möchte ich besonders betonen, dass es noch ein Feld für einen Verweis auf eine Nachricht desselben Typs gibt, das es ermöglicht, eine einfach verkettete Liste von Nachrichten zu organisieren. Eine Gruppe von Nachrichten, die durch solch eine Liste verbunden sind, nennen wir Tuppel. Somit kann jedes Element der Warteschlange eine einzelne Nachricht sein mblk_t, oder es kann der Kopf eines Tuppels von Nachrichten sein mblk_t. Jede Nachricht des Tuppels kann ihren eigenen zugehörigen Datenblock haben. Warum Tuplen benötigt werden, werden wir etwas später besprechen.

Wie bereits erwähnt, enthält eine Nachricht selbst keinen Datenblock, sondern nur einen Zeiger auf den Speicherbereich, in dem der Block gespeichert ist. In diesem Teil ähnelt das gesamte Bild der Arbeit des Mediastreamers einem Lager von Türen aus dem Zeichentrickfilm "Die Monster AG", in dem die Türen (Verweise auf Daten — Zimmer) sich mit wahnsinniger Geschwindigkeit auf hängenden Förderbändern bewegen, während die Zimmer selbst unbeweglich bleiben.

Nun, lassen Sie uns die hierarchische Struktur von unten nach oben betrachten und die genannten Entitäten des Datenübertragungsmechanismus im Mediastreamer im Detail untersuchen.

Datenblock dblk_t

Ein Datenblock besteht aus einem Header und einem Datenpuffer. Der Header wird durch die folgende Struktur beschrieben,

typedef struct datab
{
unsigned char *db_base; // Zeiger auf den Anfang des Datenpuffers.
unsigned char *db_lim;  // Zeiger auf das Ende des Datenpuffers.
void (*db_freefn)(void*); // Funktion zur Freigabe des Speichers beim Löschen des Blocks.
int db_ref; // Referenzzähler.
} dblk_t;

Die Strukturfelder enthalten Zeiger auf den Anfang des Puffers, das Ende des Puffers und die Funktion zum Löschen des Datenpuffers. Das letzte Element im Header db_ref — ist ein Referenzzähler; wenn er null erreicht, ist das ein Signal zum Entfernen dieses Blocks aus dem Speicher. Wenn der Datenblock mit der Funktion datab_alloc() erstellt wurde, wird der Datenpuffer sofort nach dem Header im Speicher angeordnet. In allen anderen Fällen kann sich der Puffer separat befinden. Im Datenpuffer werden die Signalwerte oder andere Daten gespeichert, die wir durch Filter verarbeiten möchten.

Eine neue Instanz des Datenblocks wird mit der Funktion erstellt:

dblk_t *datab_alloc(int size);

Als Eingabeparameter wird die Größe der Daten übergeben, die der Block speichern wird. Es wird zusätzlicher Speicher reserviert, um am Anfang des reservierten Speichers den Header — die Struktur databunterzubringen. Aber bei der Verwendung anderer Funktionen ist dies nicht immer der Fall; in einigen Fällen kann der Datenpuffer getrennt vom Header des Datenblocks liegen. Die Strukturfelder werden bei der Erstellung so konfiguriert, dass ihr Feld db_base auf den Anfang des Datenbereichs und db_lim auf dessen Ende zeigt. Der Referenzzähler db_ref wird auf Eins gesetzt. Der Zeiger auf die Funktion zur Datenbereinigung wird auf null gesetzt.

Nachricht mblk_t

Wie bereits erwähnt, haben die Elemente der Warteschlange den Typ mblk_t, der wie folgt definiert ist:

typedef struct msgb
{
  struct msgb *b_prev;   // Zeiger auf das vorherige Element der Liste.
  struct msgb *b_next;   // Zeiger auf das nächste Element der Liste.
  struct msgb *b_cont;   // Zeiger, um an die Nachricht andere Nachrichten anzuhängen, um ein Nachrichten-Tuplex zu erstellen.
  struct datab *b_datap; // Zeiger auf die Struktur des Datenblocks.
  unsigned char *b_rptr; // Zeiger auf den Anfang des Datenbereichs zum Lesen der Daten des Puffers b_datap.
  unsigned char *b_wptr; // Zeiger auf den Anfang des Datenbereichs zum Schreiben der Daten des Puffers b_datap.
  uint32_t reserved1;    // Reserviertes Feld 1, Mediastreamer platziert dort Verwaltungsinformationen. 
  uint32_t reserved2;    // Reserviertes Feld 2, Mediastreamer platziert dort Verwaltungsinformationen.
  #if defined(ORTP_TIMESTAMP)
  struct timeval timestamp;
  #endif
  ortp_recv_addr_t recv_addr;
} mblk_t;

Struktur mblk_t enthält zu Beginn die Zeiger b_prev, b_next, die erforderlich sind, um eine doppelt verkettete Liste (die Warteschlange) zu organisieren. queue_t).

Dann folgt der Zeiger b_cont, der nur verwendet wird, wenn die Nachricht in das Tuple eingeht. Für die letzte Nachricht im Tuple bleibt dieser Zeiger null.

Als Nächstes sehen wir einen Zeiger auf einen Datenblock b_datap, wegen dem die Nachricht existiert. Darauf folgen Zeiger auf den Bereich innerhalb des Datenblocks. Das Feld b_rptr zeigt den Ort an, von dem die Daten aus dem Puffer gelesen werden. Das Feld b_wptr zeigt den Ort an, von dem in den Puffer geschrieben wird.

Die verbleibenden Felder sind dienstlicher Natur und beziehen sich nicht auf die Funktionsweise des Datenübertragungsmechanismus.

Nachfolgend ist eine einzelne Nachricht mit dem Namen m1 und einem Datenblock dargestellt. d1.

Im nächsten Bild ist ein Tuple aus drei Nachrichten dargestellt m1, m1_1, m1_2.

Funktionen zur Arbeit mit Nachrichten mblk_t

Eine neue Nachricht mblk_t wird durch die Funktion erstellt:

mblk_t *allocb(int size, int pri); 

sie reserviert im Speicher eine neue Nachricht mblk_t mit einem Datenblock der angegebenen Größe Größe, das zweite Argument — pri wird in der betrachteten Version der Bibliothek nicht verwendet. Es sollte null bleiben. Während der Ausführung der Funktion wird Speicher für die Struktur der neuen Nachricht zugewiesen und die Funktion mblk_init(), die alle Felder der erstellten Struktur auf null setzt, wird aufgerufen und dann, mithilfe des oben genannten datab_alloc(), wird ein Datenpuffer erstellt. Danach erfolgt die Konfiguration der Felder in der Struktur:

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

Am Ende erhalten wir eine neue Nachricht mit initialisierten Feldern und einem leeren Datenpuffer. Um Daten zur Nachricht hinzuzufügen, müssen sie in den Datenblockpuffer kopiert werden:

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

wo data — ein Zeiger auf die Datenquelle, und Größe — deren Größe.
Dann muss der Zeiger auf den Schreibpunkt aktualisiert werden, damit er wieder auf den Anfang des freien Bereichs im Puffer zeigt:

msg->b_wptr = msg->b_wptr + size

Wenn eine Nachricht aus einem bereits vorhandenen Puffer erstellt werden soll, ohne Kopieren, wird dafür die Funktion verwendet:

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

Die Funktion wird nach der Erstellung der Nachricht und der Struktur des Datenblocks die Zeiger auf die Daten an die Adresse buf. Das heißt, in diesem Fall befindet sich der Datenpuffer nicht unmittelbar nach den Feldern des Datenblock-Headers, wie es bei der Erstellung des Datenblocks durch die Funktion der Fall war. datab_alloc()Der an die Funktion übergebene Datenpuffer bleibt an der Stelle, wo er war, wird aber mithilfe von Zeigern auf den gerade erstellten Blockheader gelenkt, und dieser wiederum auf die Nachricht.

Zu einer Nachricht mblk_t können mehrere Datenblöcke sequenziell angehängt werden. Dies geschieht durch die Funktion:

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

mp — die Nachricht, zu der ein weiterer Datenblock hinzugefügt werden soll;
data — ein Zeiger auf den Block, dessen Kopie zur Nachricht hinzugefügt wird;
Größe — die Größe der Daten;
pad — ein Flag, das anzeigt, dass die Größe des zugewiesenen Speichers auf eine Grenze von 4 Bytes ausgerichtet werden soll (die Auffüllung erfolgt mit Nullen).

Wenn im vorhandenen Datenpuffer der Nachricht genügend Platz ist, werden die neuen Daten an die bereits dort vorhandenen Daten angehängt. Sollte der freie Platz im Datenpuffer der Nachricht geringer sein als Größe, wird eine neue Nachricht erstellt, mit ausreichend großem Puffer, und die Daten werden in ihren Puffer kopiert. Diese neue Nachricht wird mit der ursprünglichen über einen Zeiger verbunden b_cont. In diesem Fall verwandelt sich die Nachricht in ein Tupel.

Wenn in das Tupel ein weiterer Datenblock hinzugefügt werden soll, muss die Funktion verwendet werden:

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

Sie sucht die letzte Nachricht im Tupel (die hat b_cont wird null sein) und ruft für diese Nachricht die Funktion appendb().

Um die Größe der Daten in der Nachricht oder dem Tupel zu ermitteln, kann die Funktion verwendet werden:

int msgdsize(const mblk_t *mp);

Diese Durchläuft alle Nachrichten des Tupels und gibt die Gesamtanzahl der Daten in den Datenpuffern dieser Nachrichten zurück. Für jede Nachricht wird die Datenanzahl so berechnet:

 mp->b_wptr - mp->b_rptr

Um zwei Tupel zu verbinden, wird die Funktion verwendet:

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

Sie fügt das Tupel newm an das Ende des Tupels mp an und gibt einen Zeiger auf die letzte Nachricht des entstandenen Tupels zurück.

Falls nötig, kann das Tupel in eine einzige Nachricht mit einem zusammenhängenden Datenblock verwandelt werden, dies geschieht durch die Funktion:

void msgpullup(mblk_t *mp,int len);

wenn das Argument len gleich -1 ist, wird die Größe des zugewiesenen Puffers automatisch bestimmt. Wenn len Eine positive Zahl wird ein Puffer dieser Größe erstellen, in den die Daten der Nachrichten des Tuples kopiert werden. Wenn der Puffer voll ist, wird das Kopieren eingestellt. Die erste Nachricht des Tuples erhält einen Puffer neuer Größe mit den kopierten Daten. Die anderen Nachrichten werden gelöscht und der Speicher wird in den Heap zurückgegeben.

Beim Löschen der Struktur mblk_t wird der Referenzzähler des Datenblocks berücksichtigt, wenn bei der Aufforderung freeb() er gleich Null ist, werden die Datenpuffer zusammen mit dem zuweisenden Exemplar gelöscht. mblk_t, auf das er zeigt.

Initialisierung der Felder der neuen Nachricht:

void mblk_init(mblk_t *mp);

Hinzufügen einer weiteren Datenportion zur Nachricht:

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

Wenn die neuen Daten nicht in den freien Platz des Datenpuffers der Nachricht passen, wird der Nachricht ein separat erstelltes Nachricht mit einem Puffer der passenden Größe angehängt (im ersten Nachricht wird ein Zeiger auf die hinzugefügte Nachricht gesetzt), die Nachricht wird zu einem Tuple.

Hinzufügen einer Datenportion zum Tuple:

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

Die Funktion ruft appendb() in einer Schleife auf.

Vereinigen von zwei Tuples in ein einzelnes:

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

Nachricht newm wird angefügt an mp.

Erstellen einer Kopie einer einzelnen Nachricht:

mblk_t *copyb(const mblk_t *mp);

Vollständige Kopierung des Tuples mit allen Datenblöcken:

mblk_t *copymsg(const mblk_t *mp);

Dabei werden die Elemente des Tuples von der Funktion copyb().

Erstellen einer leichten Kopie der Nachricht mblk_t. Dabei wird der Datenblock nicht kopiert, sondern der Referenzzähler erhöht. db_ref:

mblk_t *dupb(mblk_t *mp);

Erstellen einer leichten Kopie des Tuples. Die Datenblöcke werden nicht kopiert, nur ihre Referenzzähler werden erhöht. db_ref:

mblk_t *dupmsg(mblk_t* m);

Zusammenfügen aller Nachrichten des Tuples zu einer Nachricht:

void msgpullup(mblk_t *mp,size_t len);

Wenn das Argument len gleich -1 ist, wird die Größe des zugewiesenen Puffers automatisch bestimmt.

Löschen einer Nachricht, des Tuples:

void freemsg(mblk_t *mp);

Der Referenzzähler des Datenblocks wird um eins verringert. Wenn er dabei Null erreicht, wird auch der Datenblock gelöscht.

Zählen des gesamten Datenvolumens in der Nachricht oder im Tuple.

size_t msgdsize(const mblk_t *mp);

Extrahieren der Nachricht aus dem Ende der Warteschlange:

mblk_t *ms_queue_peek_last (q);

Kopieren des Inhalts der reservierten Felder einer Nachricht in eine andere Nachricht (in diesen Feldern befinden sich tatsächlich Flags, die vom Medienstreamer verwendet werden):

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

Warteschlange queue_t

Die Warteschlange in dem Medienstreamer ist als ringförmige doppeltverkettete Liste implementiert. Jedes Element der Liste enthält einen Zeiger auf einen Datenblock mit Signalabtastungen. Das bedeutet, dass nur die Zeiger auf den Datenblock nacheinander verschoben werden, während die Daten selbst stationär bleiben. Das heißt, es werden nur die Links auf diese verschoben.
Die Struktur, die die Warteschlange beschreibt queue_t, ist unten dargestellt:

typedef struct _queue
{
   mblk_t _q_stopper; /* "Leeres" Element der Warteschlange, zeigt nicht auf Daten, wird nur zur Steuerung der Warteschlange verwendet. Bei der Initialisierung der Warteschlange (qinit()) werden seine Zeiger so konfiguriert, dass sie auf sich selbst zeigen. */
   int q_mcount;        // Anzahl der Elemente in der Warteschlange.
} queue_t;

Die Struktur enthält ein Feld – einen Zeiger _q_stopper vom Typ *mblk_t, der auf das erste Element (Nachricht) in der Warteschlange zeigt. Das zweite Feld der Struktur ist der Zähler der in der Warteschlange befindlichen Nachrichten.
Im folgenden Bild ist die Warteschlange mit dem Namen q1 dargestellt, die 4 Nachrichten m1, m2, m3, m4 enthält.

Im nächsten Bild ist die Warteschlange mit dem Namen q1 dargestellt, die 4 Nachrichten m1, m2, m3, m4 enthält. Die Nachricht m2 ist der Kopf des Tupels, in das noch zwei Nachrichten m2_1 und m2_2 eingefügt werden.

Funktionen zur Arbeit mit Warteschlangen queue_t

Initialisierung der Warteschlange:

void qinit(queue_t *q);

Feld _q_stopper (ab jetzt werden wir sie "Stopper" nennen) wird mittels der Funktion initialisiert mblk_init(), sein Zeiger auf das vorhergehende Element und das nächste Element werden so konfiguriert, dass sie auf sich selbst zeigen. Der Zähler der Warteschlangenelemente wird auf Null gesetzt.

Hinzufügen eines neuen Elements (Nachricht):

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

Das neue Element m wird ans Ende der Liste hinzugefügt, die Zeiger des Elements werden so konfiguriert, dass der Stopper für es das nächste Element wird und es für den Stopper das vorhergehende. Der Zähler der Warteschlangenelemente wird erhöht.

Entnahme eines Elements aus der Warteschlange:

mblk_t * getq(queue_t *q); 

es wird die Nachricht entnommen, die nach dem Stopper steht, der Zähler der Elemente wird vermindert. Wenn in der Warteschlange außer dem Stopper keine Elemente vorhanden sind, wird 0 zurückgegeben.

Einfügen einer Nachricht in die Warteschlange:

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

Das Element mp wird vor dem Element emp. Wenn empgesetzt, ist es gleich 0, wird die Nachricht ans Ende der Warteschlange hinzugefügt.

Entnahme der Nachricht vom Kopf der Warteschlange:

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

Der Zähler der Elemente wird vermindert.

Lesen des Zeigers auf das erste Element in der Warteschlange:

mblk_t * peekq(queue_t *q); 

Entfernen aller Elemente aus der Warteschlange mit gleichzeitiger Entfernung der Elemente selbst:

void flushq(queue_t *q, int how);

Argument how wird nicht verwendet. Der Zähler der Elemente in der Warteschlange wird auf null gesetzt.

Makro zur Ablesung des Zeigers auf das letzte Element der Warteschlange:

mblk_t * qlast(queue_t *q);

Bei der Arbeit mit Nachrichtenwarteschlangen sollte beachtet werden, dass beim Aufruf ms_queue_put(q, m) mit einem Nullzeiger auf die Nachricht die Funktion in eine Endlosschleife gerät. Ihr Programm wird hängen bleiben. Ähnlich verhält sich ms_queue_next(q, m).

Filterverbindungen

Die oben beschriebene Warteschlange wird verwendet, um Nachrichten von einem Filter zu einem anderen oder von einem zu mehreren Filtern zu übertragen. Filter und deren Verbindungen bilden einen gerichteten Graphen. Der Eingang oder Ausgang eines Filters wird als allgemeiner Begriff "Pin" bezeichnet. Zur Beschreibung der Reihenfolge der Verbindungen zwischen Filtern wird im Mediastreamer der Begriff "Signalpunkt" verwendet. Ein Signalpunkt ist eine Struktur _MSCPoint, die einen Zeiger auf den Filter und die Nummer eines seiner Pins enthält; dementsprechend beschreibt sie die Verbindung eines der Eingänge oder Ausgänge des Filters.

Signalpunkt im Datenverarbeitungsgraphen

typedef struct _MSCPoint{
struct _MSFilter *filter; \\ Zeiger auf den Mediastreamer-Filter.
int pin;                        \\ Nummer eines der Eingänge oder Ausgänge des Filters, d.h. Pin.
} MSCPoint;

Die Pins der Filter werden beginnend bei null nummeriert.

Die Verbindung zweier Pins über eine Nachrichtenwarteschlange wird durch die Struktur _MSQueue, die die Nachrichtenwarteschlange und Zeiger auf zwei Signalpunkte, die sie verbindet, enthält:

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

Wir nennen diese Struktur Signal-Link. Jeder Mediastreamer-Filter enthält eine Tabelle der Eingangslinks und eine Tabelle der Ausgangslinks (MSQueue). Die Größe der Tabellen wird bei der Erstellung des Filters festgelegt; das haben wir bereits mit Hilfe der exportierten Variablen vom Typ MSFilterDesc, als wir unseren eigenen Filter entwickelt haben. Unten ist die Struktur dargestellt, die jeden Filter im Mediastreamer beschreibt, MSFilter:

struct _MSFilter{
    MSFilterDesc *desc;    
    /* Zeiger auf die Filterbeschreibung. */
    /* Geschützte Attribute, sie können nicht verschoben oder entfernt werden, da sonst die Funktionsweise mit Plugins beeinträchtigt wird. */
    ms_mutex_t lock;      /* Semaphore. */
    MSQueue **inputs;     /* Tabelle der Eingabelinks. */
    MSQueue **outputs;    /* Tabelle der Ausgabelinks. */
    struct _MSFactory *factory; /* Zeiger auf die Fabrik, die diese Filterinstanz erstellt hat. */
    void *padding;              /* Nicht verwendet, wird verwendet, wenn geschützte Felder hinzugefügt werden. */
    void *data;                 /* Zeiger auf eine beliebige Struktur zur Speicherung von Daten des internen Zustands des Filters und Zwischenergebnissen. */
    struct _MSTicker *ticker;   /* Zeiger auf das Tickobjekt, welches nicht null sein darf, wenn die Funktion process() aufgerufen wird. */
    /*private attributes, they can be moved and changed at any time*/
    MSList *notify_callbacks; /* Liste von Rückrufmethoden, die zur Behandlung von Filterereignissen verwendet werden. */
    uint32_t last_tick;       /* Nummer des letzten Taktes, in dem die Funktion process() aufgerufen wurde. */
    MSFilterStats *stats;     /* Statistik der Filteroperation.*/
    int postponed_task; /* Anzahl der aufgeschobenen Aufgaben. Einige Filter können die Verarbeitung von Daten (den Aufruf von process()) um mehrere Takte verschieben.*/
    bool_t seen;  /* Flag, das vom Ticker verwendet wird, um zu kennzeichnen, dass diese Filterinstanz bereits in diesem Takt verarbeitet wurde.*/
};
typedef struct _MSFilter MSFilter;

Nachdem wir die Filter in unserem C-Programm gemäß unserem Konzept verbunden haben (aber den Ticker nicht angeschlossen haben), haben wir somit einen gerichteten Graphen erstellt, dessen Knoten Instanzen der Struktur sind, MSFilter, und die Kanten sind Linkinstanzen. MSQueue.

Hintergrundaktivitäten des Tickers

Als ich Ihnen sagte, dass der Ticker ein Quelle für Taktsignale ist, war das nicht die ganze Wahrheit über ihn. Der Ticker ist ein Objekt, das zeitgesteuert Funktionen ausführt process() aller Filter im Schema (Graphen), an das er angeschlossen ist. Wenn wir im C-Programm den Ticker mit einem Filter des Graphen verbinden, zeigen wir dem Ticker den Graphen, den er von diesem Moment an verwalten wird, bis wir ihn wieder trennen. Nach dem Anschluss beginnt der Ticker, den ihm anvertrauten Graphen zu durchsuchen und eine Liste der Filter zu erstellen, die ihn umfasst. Um nicht "dasselbe" Filter zweimal zu zählen, kennzeichnet er die entdeckten Filter, indem er das Flag seensetzt. Die Suche erfolgt in den Linktabellen, die jeder Filter hat.

Während seiner Einführungstour überprüft der Ticker, ob unter den Filtern mindestens einer vorhanden ist, der die Rolle einer Datenquelle erfüllt. Wenn keine gefunden wird, gilt das Diagramm als fehlerhaft, und der Ticker bricht die Ausführung ab.

Wenn das Diagramm "richtig" ist, wird für jeden gefundenen Filter zur Initialisierung die Funktion preprocess()aufgerufen. Sobald der Moment für den nächsten Verarbeitungsschritt kommt (standardmäßig alle 10 Millisekunden), ruft der Ticker die Funktion process() für alle zuvor gefundenen Quellfilter auf, gefolgt von den restlichen Filtern in der Liste. Wenn der Filter Eingangslinks hat, wird der Aufruf der Funktion process() so lange wiederholt, bis die Warteschlangen der Eingangslinks leer sind. Danach wechselt er zum nächsten Filter in der Liste und "scrollt" ihn durch, bis die Eingangslinks von Nachrichten befreit sind. Der Ticker wechselt von Filter zu Filter, bis die Liste beendet ist. An dieser Stelle endet der Verarbeitungsschritt.

Jetzt kehren wir zu den Tuplen zurück und sprechen darüber, warum eine solche Entität im Mediastreamer hinzugefügt wurde. Allgemein benötigt der Algorithmus, der innerhalb des Filters arbeitet, andere Datenmengen als die Größe der Datenpuffer, die am Eingang ankommen. Nehmen wir an, wir schreiben einen Filter, der eine schnelle Fourier-Transformation durchführt, die definitionsgemäß nur Datenblöcke verarbeiten kann, deren Größe eine Potenz von zwei ist. Lassen Sie uns 512 Abtastwerte annehmen. Wenn die Daten über einen Telefonkanal generiert werden, wird der Datenpuffer jeder Nachricht am Eingang 160 Abtastwerte des Signals liefern. Es gibt den Anreiz, die Daten vom Eingang nicht abzurufen, bis die benötigte Datenmenge erreicht ist. In diesem Fall würde es jedoch zu einem Konflikt mit dem Ticker kommen, der erfolglos versucht, den Filter zu scrollen, bis die Eingangslinks entleert sind. Wir haben diese Regel zuvor als drittes Prinzip der Funktionsweise des Filters bezeichnet. Gemäß diesem Prinzip muss die Funktion process() des Filters alle Daten aus den Eingangswarteschlangen abholen.

Darüber hinaus können am Eingang nur 512 Messwerte entnommen werden, da die Abholung nur in ganzen Blöcken möglich ist, d.h. der Filter muss 640 Messwerte entnehmen und verwendet 512 davon, während der Rest für das Ansammeln neuer Daten reserviert bleibt. So muss unser Filter neben seiner Hauptfunktion auch unterstützende Maßnahmen zur Zwischenlagerung der Eingangsdaten gewährleisten. Die Entwickler des Mediasreamers und der Lösung dieses allgemeinen Problems haben ein spezielles Objekt – MSBufferizer (Puffer) – entwickelt, das diese Aufgabe mit Hilfe von Tupeln löst.

Puffer (MSBufferizer)

Dies ist ein Objekt, das die Eingangsdaten innerhalb des Filters ansammeln wird und mit der Verarbeitung beginnen wird, sobald die Menge der Daten ausreichend ist, um den Filteralgorithmus auszuführen. Solange der Puffer Daten ansammelt, wird der Filter im Leerlaufmodus arbeiten, ohne Rechenleistung des Prozessors zu verbrauchen. Sobald jedoch die Lese-Funktion des Puffers einen Wert ungleich Null zurückgibt, beginnt die Funktion process() des Filters, die im Puffer gespeicherten Daten in Portionen der benötigten Größe zu entnehmen und zu verarbeiten, bis alle Daten verbraucht sind.
Vorläufig nicht benötigte Daten bleiben im Puffer als erstes Element des Tupels, an das nachfolgende Blöcke von Eingangsdaten angehängt werden.

Die Struktur, die den Puffer beschreibt:

struct _MSBufferizer{
queue_t q; /* Nachrichtenwarteschlange. */
int size; /* Gesamter Datenumfang, der sich derzeit im Puffer befindet. */
};
typedef struct _MSBufferizer MSBufferizer;

Funktionen zur Arbeit mit MSBufferizer

Erstellung eines neuen Puffermoduls:

MSBufferizer * ms_bufferizer_new(void);

Speicher wird zugewiesen, wird in ms_bufferizer_init() initialisiert und ein Zeiger wird zurückgegeben.

Funktion zur Initialisierung:

void ms_bufferizer_init(MSBufferizer *obj); 

Die Warteschlange q, das Feld Größe wird auf Null gesetzt.

Nachricht hinzufügen:

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

Die Nachricht m wird in die Warteschlange eingefügt. Die berechnete Größe der Datenblöcke wird zu Größe.

Umsetzung in den Puffer aller Nachrichten aus der Datenwarteschlange q:

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

Der Transfer von Nachrichten aus der Warteschlange q in den Puffer erfolgt durch die Funktion ms_bufferizer_put().

Lesen aus dem Puffer:

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

Wenn die Größe der im Puffer gesammelten Daten kleiner ist als angefordert (datalen), die Funktion gibt null zurück, das Kopieren von Daten in data wird nicht durchgeführt. Andernfalls erfolgt die sequentielle Kopie von Daten aus den Tuples, die sich im Puffer befinden. Nach dem Kopieren wird das Tuple gelöscht und der Speicher wird freigegeben. Das Kopieren endet, sobald datalen Bytes kopiert wurden. Wenn der Speicher mitten im Datenblock endet, wird in dieser Meldung der Datenblock auf den verbleibenden, nicht kopierten Teil verkürzt. Beim nächsten Aufruf wird das Kopieren an dieser Stelle fortgesetzt.

Lesen der Menge an Daten, die derzeit im Puffer verfügbar sind:

int ms_bufferizer_get_avail(MSBufferizer *obj); 

Gibt das Feld zurück Größe des Puffers.

Abwerfen eines Teils der Daten, die sich im Puffer befinden:

void ms_bufferizer_skip_bytes(MSBufferizer *obj, int bytes);

Die angegebene Anzahl von Bytes an Daten wird extrahiert und verworfen. Die ältesten Daten werden verworfen.

Löschen aller Nachrichten, die sich im Puffer befinden:

void ms_bufferizer_flush(MSBufferizer *obj); 

Der Datenzähler wird auf null zurückgesetzt.

Löschen aller Nachrichten, die sich im Puffer befinden:

void ms_bufferizer_uninit(MSBufferizer *obj); 

Das Zurücksetzen des Zählers erfolgt nicht.

Löschen des Puffers und Freigabe von Speicher:

void ms_bufferizer_destroy(MSBufferizer *obj);  

Beispiele für die Verwendung des Puffers finden Sie im Quellcode mehrerer Streaming-Filter. Zum Beispiel im Filter MS_L16_ENC, der die Bytes in den Proben von Netzwerk- in Host-Reihenfolge umsortiert: l16.c

Im nächsten Artikel betrachten wir die Frage der Lastbewertung auf dem Ticker und Möglichkeiten zur Bekämpfung übermäßiger Rechenlast im Medienstreamer.

Quelle: habr.com