{"id":79540,"date":"2020-04-28T07:42:01","date_gmt":"2020-04-28T05:42:01","guid":{"rendered":"https:\/\/prohoster.info\/blog\/administrirovanie\/izuchaem-voip-dvizhok-mediastreamer2-chast-11"},"modified":"2020-04-28T07:42:01","modified_gmt":"2020-04-28T05:42:01","slug":"izuchaem-voip-dvizhok-mediastreamer2-chast-11","status":"publish","type":"post","link":"https:\/\/prohoster.info\/en\/blog\/administrirovanie\/izuchaem-voip-dvizhok-mediastreamer2-chast-11","title":{"rendered":"Exploring the VoIP engine Mediastreamer2. Part 11","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"<p>The article material is taken from my <noindex><a rel=\"nofollow\" href=\"https:\/\/zen.yandex.ru\/profile\/editor\/id\/5e3f8e0751f5346faab4fc7a\">Zen channel<\/a><\/noindex>.<\/p>\n<p><\/p>\n<p><img decoding=\"async\" alt=\"Exploring the VoIP engine Mediastreamer2. Part 11\" src=\"\/wp-content\/uploads\/2020\/04\/b66fab43eb45c626c50f0383dcfa7cca.png\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<p><\/p>\n<p>Data Transfer Mechanism<\/p>\n<p><\/p>\n<ul>\n<li>Data Block dblk_t<\/li>\n<li>Message mblk_t<\/li>\n<li>Functions for working with messages mblk_t<\/li>\n<li>Queue queue_t<\/li>\n<li>Functions for working with queues queue_t<\/li>\n<li>Filter Connections<\/li>\n<li>Signal Point of the Data Processing Graph<\/li>\n<li>Behind-the-Scenes Activity of the Ticker<\/li>\n<li>Buffer (MSBufferizer)<\/li>\n<li>Functions for working with MSBufferizer<\/li>\n<\/ul>\n<p><\/p>\n<p>Previously <noindex><a rel=\"nofollow\" href=\"https:\/\/habr.com\/ru\/post\/497932\/\">article<\/a><\/noindex> we developed our own filter. In this article, we will dedicate ourselves to the inner mechanism of data transfer between media streamer filters. This will allow us to write sophisticated filters with less effort in the future.<\/p>\n<p><noindex><a rel=\"nofollow\" name=\"habracut\"><\/a><\/noindex><\/p>\n<h2 id=\"mehanizm-peremescheniya-dannyh\">Data Transfer Mechanism<\/h2>\n<p><\/p>\n<p>Data transfer in the media streamer is performed using queues described by the structure <em>queue_t<\/em>. Data is transferred through queues in streams of messages of type <em>mblk_t<\/em>, which themselves do not contain signal data but only references to the previous and next message and to the data block. Additionally, I want to emphasize that there is a field to reference another message of the same type, which allows us to organize a singly linked list of messages. We will call a group of messages linked in this way a tuple. Thus, any element of the queue can be a single message <em>mblk_t<\/em>, or it can be the head of a tuple of messages <em>mblk_t<\/em>. Each message in a tuple may have its own associated data block. We will discuss the purpose of tuples a bit later.<\/p>\n<p><\/p>\n<p>As mentioned above, the message itself does not contain a data block; instead, it contains only a pointer to the memory area where the block is stored. At this point, the overall picture of how the media streamer works resembles a warehouse of doors from the animated film \"Monsters, Inc.\", where the doors (links to data \u2014 rooms) move at an insane speed along overhead conveyors, while the rooms themselves remain stationary. <\/p>\n<p><\/p>\n<p>Now, moving up the hierarchy from bottom to top, let us examine in detail the entities of the data transfer mechanism in the media streamer.<\/p>\n<p><\/p>\n<h3 id=\"blok--dannyh-dblk_t\">Data Block <em>dblk_t<\/em><\/h3>\n<p><\/p>\n<p>A data block consists of a header and a data buffer. The header is described by the following structure,<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">typedef struct datab\n{\nunsigned char *db_base; \/\/ Pointer to the start of the data buffer.\nunsigned char *db_lim;  \/\/ Pointer to the end of the data buffer.\nvoid (*db_freefn)(void*); \/\/ Memory release function when deleting the block.\nint db_ref; \/\/ Reference counter.\n} dblk_t;<\/code><\/pre>\n<p><\/p>\n<p>The structure fields contain pointers to the start of the buffer, the end of the buffer, and the function for deleting the data buffer. The last element in the header <em>db_ref<\/em> \u2014 reference counter; when it reaches zero, it signals the deletion of this block from memory. If the data block was created by the function <em>datab_alloc() <\/em>, then the data buffer will be located in memory immediately after the header. In all other cases, the buffer may be located separately. The data buffer will contain signal counts or other data that we want to process with filters.<\/p>\n<p><\/p>\n<p>A new instance of a data block is created using the function:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">dblk_t *datab_alloc(int size);<\/code><\/pre>\n<p><\/p>\n<p>The size of the data that the block will store is passed to it as an input parameter. More memory is allocated so that at the beginning of the allocated memory, the header\u2014 the structure <em>datab<\/em>, can be placed. However, when using other functions, this does not always happen; in some cases, the data buffer may be located separately from the data block header. The structure fields are set during creation so that its field <em>db_base<\/em> points to the start of the data area, and <em>db_lim<\/em> to its end. The reference counter <em>db_ref<\/em> is set to one. The data cleanup function pointer is set to zero.<\/p>\n<p><\/p>\n<h3 id=\"soobschenie-mblk_t\">Message <em>mblk_t<\/em><\/h3>\n<p><\/p>\n<p>As mentioned, queue elements are of type <em>mblk_t,<\/em> which is defined as follows:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">typedef struct msgb\n{\n  struct msgb *b_prev;   \/\/ Pointer to the previous list element.\n  struct msgb *b_next;   \/\/ Pointer to the next list element.\n  struct msgb *b_cont;   \/\/ Pointer to attach other messages to the message, to create a tuple of messages.\n  struct datab *b_datap; \/\/ Pointer to the data block structure.\n  unsigned char *b_rptr; \/\/ Pointer to the start of the data area for reading data from the b_datap buffer.\n  unsigned char *b_wptr; \/\/ Pointer to the start of the data area for writing data to the b_datap buffer.\n  uint32_t reserved1;    \/\/ Reserved field 1, the media streamer places control information there.\n  uint32_t reserved2;    \/\/ Reserved field 2, the media streamer places control information there.\n  #if defined(ORTP_TIMESTAMP)\n  struct timeval timestamp;\n  #endif\n  ortp_recv_addr_t recv_addr;\n} mblk_t;<\/code><\/pre>\n<p><\/p>\n<p>Structure <em>mblk_t<\/em> at the beginning contains pointers <em>b_prev<\/em>, <em>b_next<\/em>, which are necessary for organizing a doubly linked list (which is the queue <em>queue_t<\/em>). <\/p>\n<p><\/p>\n<p>Then there is the pointer <em>b_cont<\/em>, which is only used when the message enters the tuple. For the last message in the tuple, this pointer remains null. <\/p>\n<p><\/p>\n<p>Next, we see a pointer to the data block <em>b_datap<\/em>, for which the message exists. Following this are pointers to the area within the data buffer of the block. The field <em>b_rptr<\/em> indicates the location from which data will be read from the buffer. The field <em>b_wptr<\/em> indicates the location from which data will be written to the buffer. <\/p>\n<p><\/p>\n<p>The remaining fields are for internal use and do not relate to the operation of the data transfer mechanism.<\/p>\n<p><\/p>\n<p>Below is a single message named <em>m1<\/em> and a data block <em>d1<\/em>.<br \/>\n<img decoding=\"async\" alt=\"Exploring the VoIP engine Mediastreamer2. Part 11\" src=\"\/wp-content\/uploads\/2020\/04\/e6a726b13491832258c1228f03bde802.png\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nThe next figure shows a tuple of three messages <em>m1<\/em>, <em>m1_1<\/em>, <em>m1_2<\/em>.<br \/>\n<img decoding=\"async\" alt=\"Exploring the VoIP engine Mediastreamer2. Part 11\" src=\"\/wp-content\/uploads\/2020\/04\/9613feefe9684b9551cd98115faa9a08.png\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<p><\/p>\n<h3 id=\"funkcii-raboty-s-soobscheniyami--mblk_t\">Message handling functions <em>mblk_t<\/em><\/h3>\n<p><\/p>\n<p>A new message <em>mblk_t<\/em> is created by the function:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t *allocb(int size, int pri); <\/code><\/pre>\n<p><\/p>\n<p>it allocates a new message in memory <em>mblk_t<\/em> with a data block of the specified size <em>size<\/em>, the second argument \u2014 <em>pri<\/em> is not used in this version of the library. It should remain null. During the function's operation, memory will be allocated for the new message structure, and the function <em>mblk_init()<\/em>, which will reset all fields of the created structure instance, will then, using the aforementioned <em>datab_alloc()<\/em>, create the data buffer. After that, the fields in the structure will be set up: <\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mp-&gt;b_datap=datab;\nmp-&gt;b_rptr=mp-&gt;b_wptr=datab-&gt;db_base;\nmp-&gt;b_next=mp-&gt;b_prev=mp-&gt;b_cont=NULL;<\/code><\/pre>\n<p><\/p>\n<p>The output is a new message with initialized fields and an empty data buffer. To add data to the message, it is necessary to copy them into the data block's buffer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">memcpy(msg-&gt;b_rptr, data, size);<\/code><\/pre>\n<p><\/p>\n<p>where <em>data<\/em> \u2014 pointer to the data source, and <em>size<\/em> \u2014 their size.<br \/>\nThen, it is necessary to update the write pointer so that it again points to the start of the free area in the buffer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">msg-&gt;b_wptr = msg-&gt;b_wptr + size<\/code><\/pre>\n<p><\/p>\n<p>If it is required to create a message from an already existing buffer, without copying, then the function used is:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t *esballoc(uint8_t *buf, int size, int pri, void (*freefn)(void*)); <\/code><\/pre>\n<p><\/p>\n<p>The function, after creating the message and the data block structure, will adjust its pointers to the data at the address <em>buf<\/em>. That is, in this case, the data buffer does not follow the fields of the data block header, as it did when the data block was created by the function. <em>datab_alloc()<\/em>The buffer passed to the function will stay where it is, but with the help of pointers, it will be linked to the newly created header block, which in turn links to the message.<\/p>\n<p><\/p>\n<p>A single message <em>mblk_t<\/em> can have multiple data blocks consecutively attached to it. This is done using the function:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t * appendb(mblk_t *mp, const char *data, int size, bool_t pad); <\/code><\/pre>\n<p><\/p>\n<p><em>mp<\/em> \u2014 the message to which an additional data block will be added;<br \/>\n<em>data<\/em> \u2014 a pointer to the block, a copy of which will be added to the message;<br \/>\n<em>size<\/em> \u2014 the size of the data;<br \/>\n<em>pad<\/em> \u2014 a flag indicating that the allocated memory size should be aligned to a 4-byte boundary (padding will be performed with zeros). <\/p>\n<p><\/p>\n<p>If there is enough space in the existing message data buffer, the new data will be appended behind the existing data there. If there is less free space in the message data buffer than <em>size<\/em>, a new message is created with a sufficient buffer size, and the data is copied into its buffer. This new message is linked to the original message through the pointer <em>b_cont<\/em>. In this case, the message turns into a tuple. <\/p>\n<p><\/p>\n<p>If another data block needs to be added to the tuple, the function should be used:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void msgappend(mblk_t *mp, const char *data, int size, bool_t pad);<\/code><\/pre>\n<p><\/p>\n<p>it will locate the last message in the tuple (which will have <em>b_cont<\/em> zero) and call the function for that message <em>appendb()<\/em>.<\/p>\n<p><\/p>\n<p>The data size in a message or a tuple can be determined using the function:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">int msgdsize(const mblk_t *mp);<\/code><\/pre>\n<p><\/p>\n<p>it will traverse all messages in the tuple and return the total amount of data in the data buffers of these messages. For each message, the amount of data is calculated as follows:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\"> mp-&gt;b_wptr - mp-&gt;b_rptr<\/code><\/pre>\n<p><\/p>\n<p>To concatenate two tuples, the function is used:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t *concatb(mblk_t *mp, mblk_t *newm);<\/code><\/pre>\n<p><\/p>\n<p>it attaches the tuple <em>newm<\/em> to the end of the tuple <em>mp<\/em> and returns a pointer to the last message of the resulting tuple.<\/p>\n<p><\/p>\n<p>If necessary, a tuple can be turned into a single message with one data block using the function:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void msgpullup(mblk_t *mp,int len);<\/code><\/pre>\n<p><\/p>\n<p>if the argument <em>len<\/em> is -1, then the size of the allocated buffer is determined automatically. If <em>len<\/em> If a positive number is provided, a buffer of that size will be created, and the data from the tuple messages will be copied into it. If the buffer runs out, the copying will stop. The first message of the tuple will receive a new size buffer with the copied data. The remaining messages will be discarded, and the memory will be returned to the heap.<\/p>\n<p><\/p>\n<p>When deleting the structure <em>mblk_t<\/em> the reference counter of the data block is considered if upon calling <em>freeb()<\/em> it equals zero, the data buffer is deleted along with the instance <em>mblk_t<\/em>, to which it points.<\/p>\n<p><\/p>\n<p>Initialization of the new message fields:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void mblk_init(mblk_t *mp);<\/code><\/pre>\n<p><\/p>\n<p>Adding another portion of data to the message: <\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t * appendb(mblk_t *mp, const char *data, size_t size, bool_t pad);<\/code><\/pre>\n<p><\/p>\n<p>If the new data does not fit in the free space of the message data buffer, a separately created message with the required buffer size is attached to the message (a pointer to the added message is set in the first message), turning it into a tuple.<\/p>\n<p><\/p>\n<p>Adding a portion of data to the tuple:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void msgappend(mblk_t *mp, const char *data, size_t size, bool_t pad); <\/code><\/pre>\n<p><\/p>\n<p>The function calls appendb() in a loop.<\/p>\n<p><\/p>\n<p>Merging two tuples into one:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t *concatb(mblk_t *mp, mblk_t *newm);<\/code><\/pre>\n<p><\/p>\n<p>Message <em>newm<\/em> will be appended to <em>mp<\/em>.<\/p>\n<p><\/p>\n<p>Creating a copy of a single message:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t *copyb(const mblk_t *mp);<\/code><\/pre>\n<p><\/p>\n<p>Complete copying of the tuple with all data blocks:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t *copymsg(const mblk_t *mp);<\/code><\/pre>\n<p><\/p>\n<p>The elements of the tuple are copied by the function <em>copyb()<\/em>.<\/p>\n<p><\/p>\n<p>Creating a lightweight copy of the message. In this case, the data block is not copied, but its reference counter is increased. <em>mblk_t<\/em>mblk_t *dupb(mblk_t *mp); <em>db_ref<\/em>:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">Creating a lightweight copy of the tuple. Data blocks are not copied, only their reference counters are increased.<\/code><\/pre>\n<p><\/p>\n<p>mblk_t *dupmsg(mblk_t* m); <em>db_ref<\/em>:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">Concatenating all messages of the tuple into a single message:<\/code><\/pre>\n<p><\/p>\n<p>void msgpullup(mblk_t *mp,size_t len);<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">If the argument<\/code><\/pre>\n<p><\/p>\n<p>equals -1, the size of the allocated buffer is determined automatically. <em>len<\/em> Deletion of the message, tuple:<\/p>\n<p><\/p>\n<p>void freemsg(mblk_t *mp);<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">The reference counter of the data block is decreased by one. If it reaches zero, the data block is also deleted.<\/code><\/pre>\n<p><\/p>\n<p>Counting the total volume of data in the message or tuple.<\/p>\n<p><\/p>\n<p>size_t msgdsize(const mblk_t *mp);<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">Extracting a message from the tail of the queue:<\/code><\/pre>\n<p><\/p>\n<p>mblk_t *ms_queue_peek_last (q);<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">Copying the contents of the reserved fields from one message to another (actually, these fields contain flags used by the media streamer):<\/code><\/pre>\n<p><\/p>\n<p>mblk_meta_copy(const mblk_t *source, mblk *dest);<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">Queue<\/code><\/pre>\n<p><\/p>\n<h3 id=\"ochered-queue_t\">Queue <em>queue_t<\/em><\/h3>\n<p><\/p>\n<p>The message queue in the media streamer is implemented as a circular doubly linked list. Each element of the list contains a pointer to a data block with signal readings. This means that only the pointers to the data blocks are moved in sequence, while the actual data remains stationary. In other words, only the references to them are moved.<br \/>\nStructure describing the queue <em>queue_t<\/em>, shown below:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">typedef struct _queue\n{\n   mblk_t _q_stopper; \/* \"Dummy\" queue element, does not point to data, used solely for queue management. During queue initialization (qinit()), its pointers are set to point to itself. *\/\n   int q_mcount;        \/\/ Number of elements in the queue.\n} queue_t;<\/code><\/pre>\n<p><\/p>\n<p>The structure contains a field \u2014 pointer <em>_q_stopper<\/em> of type *mblk_t, which points to the first element (message) in the queue. The second field of the structure is a counter of messages in the queue.<br \/>\nThe figure below shows the queue named q1, containing 4 messages m1, m2, m3, m4.<br \/>\n<img decoding=\"async\" alt=\"Exploring the VoIP engine Mediastreamer2. Part 11\" src=\"\/wp-content\/uploads\/2020\/04\/07fc043861edf15e5eb40e49c8ec5e3e.png\" style=\"display:block;margin: 0 auto;\" \/><br \/>\nThe next figure shows the queue named q1, containing 4 messages m1, m2, m3, m4. Message m2 is the head of the tuple, which includes two more messages m2_1 and m2_2.<\/p>\n<p><\/p>\n<p><img decoding=\"async\" alt=\"Exploring the VoIP engine Mediastreamer2. Part 11\" src=\"\/wp-content\/uploads\/2020\/04\/8fa7667f3690674a2ee321b4a6c39e1d.png\" style=\"display:block;margin: 0 auto;\" \/><\/p>\n<p><\/p>\n<h3 id=\"funkcii-raboty-s-ocheredyami-queue_t\">Functions for working with queues queue_t<\/h3>\n<p><\/p>\n<p>Initializing the queue:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void qinit(queue_t *q);<\/code><\/pre>\n<p><\/p>\n<p>Field <em>_q_stopper<\/em> (hereinafter referred to as \"stopper\") is initialized by a function <em>mblk_init()<\/em>, its previous and next element pointers are configured to point to itself. The element count in the queue is reset to zero.<\/p>\n<p><\/p>\n<p>Adding a new element (message):<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void putq(queue_t *q, mblk_t *m);<\/code><\/pre>\n<p><\/p>\n<p>The new element <em>m<\/em> is added to the end of the list, the element pointers are configured so that the stopper becomes the next element for it, and it becomes the previous one for the stopper. The element count in the queue is incremented.<\/p>\n<p><\/p>\n<p>Extracting an element from the queue:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t * getq(queue_t *q); <\/code><\/pre>\n<p><\/p>\n<p>the message that stands after the stopper is extracted, and the element count is decremented. If there are no elements in the queue apart from the stopper, 0 is returned.<\/p>\n<p><\/p>\n<p>Inserting a message into the queue:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void insq(queue_t *q, mblk_t *emp, mblk_t *mp); <\/code><\/pre>\n<p><\/p>\n<p>Element <em>mp<\/em> is inserted before the element <em>emp<\/em>. If <em>emp<\/em>=0, then the message is added to the tail of the queue.<\/p>\n<p><\/p>\n<p>Extracting a message from the head of the queue:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void remq(queue_t *q, mblk_t *mp); <\/code><\/pre>\n<p><\/p>\n<p>The element count is decremented.<\/p>\n<p><\/p>\n<p>Reading the pointer to the first element in the queue:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t * peekq(queue_t *q); <\/code><\/pre>\n<p><\/p>\n<p>Removing all elements from the queue with the removal of the elements themselves:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void flushq(queue_t *q, int how);<\/code><\/pre>\n<p><\/p>\n<p>Argument <em>how<\/em> is not used. The queue element counter is set to zero.<\/p>\n<p><\/p>\n<p>Macro for reading the pointer to the last element of the queue:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">mblk_t * qlast(queue_t *q);<\/code><\/pre>\n<p><\/p>\n<p>When working with message queues, it should be noted that when calling <em>ms_queue_put(q, m)<\/em> with a null pointer for the message, the function enters an infinite loop. Your program will hang. The same behavior occurs with <em>ms_queue_next(q, m)<\/em>.<\/p>\n<p><\/p>\n<h3 id=\"soedinenie-filtrov\">Filter Connections<\/h3>\n<p><\/p>\n<p>The queue described above is used to transfer messages from one filter to another or from one to several filters. Filters and their connections form a directed graph. The input or output of a filter will be referred to by the general term \"pin\". To describe the order of connections between filters, the media streamer uses the concept of a \"signal point\". A signal point is a structure <em>_MSCPoint<\/em>, which contains a pointer to the filter and the number of one of its pins; thus, it describes the connection of one of the filter's inputs or outputs. <\/p>\n<p><\/p>\n<h4 id=\"signalnaya-tochka-grafa-obrabotki-dannyh\">Signal Point of the Data Processing Graph<\/h4>\n<p><\/p>\n<pre><code class=\"cpp\">typedef struct _MSCPoint{\nstruct _MSFilter *filter; \/\/ Pointer to the media streamer filter.\nint pin;                        \/\/ The number of one of the filter's inputs or outputs, i.e., the pin.\n} MSCPoint;\n<\/code><\/pre>\n<p><\/p>\n<p>The pins of the filters are numbered starting from zero. <\/p>\n<p><\/p>\n<p>The connection of two pins by a message queue is described by the structure <em>_MSQueue<\/em>, which contains the message queue and pointers to two signal points that it connects:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">typedef struct _MSQueue\n{\nqueue_t q;\nMSCPoint prev;\nMSCPoint next;\n}MSQueue;\n<\/code><\/pre>\n<p><\/p>\n<p>We will call this structure a signal link. Each media streamer filter contains a table of input links and a table of output links (<em>MSQueue<\/em>). The size of the tables is specified when creating the filter, which we already did using the exported variable of type <em>MSFilterDesc<\/em>, when developing our own filter. Below is the structure describing any filter in the media streamer, <em>MSFilter<\/em>:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">\nstruct _MSFilter{\n    MSFilterDesc *desc;    \/* Pointer to filter descriptor. *\/\n    \/* Protected attributes, they cannot be moved or removed, otherwise it will interfere with plugin operation. *\/\n    ms_mutex_t lock;      \/* Semaphore. *\/\n    MSQueue **inputs;     \/* Table of input links. *\/\n    MSQueue **outputs;    \/* Table of output links. *\/\n    struct _MSFactory *factory; \/* Pointer to the factory that created this filter instance. *\/\n    void *padding;              \/* Not used, will be utilized if protected fields are added. *\/\n    void *data;                 \/* Pointer to arbitrary structure for storing filter's internal state data and intermediate computations. *\/\n    struct _MSTicker *ticker;   \/* Pointer to ticker object, which must not be NULL when process() is called. *\/\n    \/*private attributes, they can be moved and changed at any time*\/\n    MSList *notify_callbacks; \/* List of callbacks used for handling filter events. *\/\n    uint32_t last_tick;       \/* Number of the last tick when process() was called. *\/\n    MSFilterStats *stats;     \/* Filter operation statistics.*\/\n    int postponed_task; \/* Number of postponed tasks. Some filters may delay data processing (process() call) for several ticks.*\/\n    bool_t seen;  \/* Flag used by ticker to mark that this filter instance has already been serviced in this tick.*\/\n};\ntypedef struct _MSFilter MSFilter;\n<\/code><\/pre>\n<p><\/p>\n<p>Once we have connected the filters in our C program according to our design (but not connected the ticker), we have thereby created a directed graph, where the nodes are instances of the structure <em>MSFilter<\/em>, and the edges are instances of the links <em>MSQueue<\/em>. <\/p>\n<p><\/p>\n<h3 id=\"zakulisnaya-deyatelnost-tikera\">Behind-the-Scenes Activity of the Ticker<\/h3>\n<p><\/p>\n<p>When I told you that the ticker is a source tick filter, that was not the whole truth about it. The ticker is an object that triggers the execution of functions <em>process()<\/em> of all filters in the scheme (graph) to which it is connected. When we connect a ticker to a graph filter in the C program, we show the ticker the graph it will manage from that moment until we disconnect it. After the connection, the ticker begins to inspect the entrusted graph, compiling a list of the filters into which it is included. To avoid \"counting\" the same filter twice, it marks the discovered filters by setting a flag in them <em>seen<\/em>. The search is conducted through the link tables that each filter has. <\/p>\n<p><\/p>\n<p>During its introductory tour of the graph, the ticker checks if there is at least one filter among the filters that acts as a source of data blocks. If none are found, the graph is deemed incorrect, and the ticker terminates its operation.<\/p>\n<p><\/p>\n<p>If the graph turns out to be \"correct\", the initialization function is called for each found filter <em>preprocess()<\/em>. Once the time arrives for the next processing cycle (default every 10 milliseconds), the ticker calls the function <em>process()<\/em> for all previously found source filters, and then for the remaining filters in the list. If a filter has input links, the function invocation <em>process()<\/em> repeats until the input links of the queues are empty. After that, it moves on to the next filter in the list and \"scrolls\" it until the input links are cleared of messages. The ticker moves from filter to filter until the list ends. This concludes the tick processing. <\/p>\n<p><\/p>\n<p>Now let\u2019s return to tuples and discuss why such an entity was added to the media streamer. In general, the volume of data required by the algorithm operating inside the filter does not match and is not a multiple of the size of the data buffers arriving at input. For example, we're writing a filter that performs a fast Fourier transform, which by definition can only process data blocks whose size is a power of two. Let\u2019s say this is 512 samples. If the data is generated by a telephone line, the data buffer of each incoming message will bring us 160 samples of the signal. There is a temptation not to retrieve data from the input until the necessary amount is available. But in this case, there will be a collision with the ticker, which will unsuccessfully attempt to poll the filter until the input link is exhausted. We previously outlined this rule as the third principle of the filter's operation. According to this principle, the filter's process() function must retrieve all data from the input queues.<\/p>\n<p><\/p>\n<p>In addition, only 512 samples can be retrieved from the input, as samples can only be taken in whole blocks, i.e., the filter will have to retrieve 640 samples and after using 512 of them, the remainder will wait until new data is collected. Thus, our filter, aside from its primary function, must provide auxiliary actions for temporarily storing incoming data. The developers of the media streamer and the solution to this common task designed a special object \u2014 MSBufferizer, which tackles this using tuples. <\/p>\n<p><\/p>\n<h3 id=\"buferizator-msbufferizer\">Buffer (MSBufferizer)<\/h3>\n<p><\/p>\n<p>This is an object that will accumulate incoming data within the filter and will start delivering it for processing as soon as the amount of information is sufficient to run the filter algorithm. While the bufferizer is accumulating data, the filter will run in idle mode, not expending CPU processing power. But as soon as the read function from the bufferizer returns a non-zero value, the filter's process() function begins retrieving and processing data from the bufferizer in the required size batches until they are depleted.<br \/>\nThe currently unused data remains in the bufferizer as the first element of the tuple, to which subsequent blocks of incoming data are attached. <\/p>\n<p><\/p>\n<p>The structure that describes the bufferizer: <\/p>\n<p><\/p>\n<pre><code class=\"cpp\">struct _MSBufferizer{\nqueue_t q; \/* Message queue. *\/\nint size; \/* Total size of data in the bufferizer at this moment. *\/\n};\ntypedef struct _MSBufferizer MSBufferizer;<\/code><\/pre>\n<p><\/p>\n<h3 id=\"funkcii-raboty-s-msbufferizer\">Functions for working with MSBufferizer<\/h3>\n<p><\/p>\n<p>Creating a new instance of the bufferizer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">MSBufferizer * ms_bufferizer_new(void);<\/code><\/pre>\n<p><\/p>\n<p>Memory is allocated, initialized in <em>ms_bufferizer_init()<\/em> and a pointer is returned.<\/p>\n<p><\/p>\n<p>Initialization function:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_init(MSBufferizer *obj); <\/code><\/pre>\n<p><\/p>\n<p>The queue <em>q<\/em>, field <em>size<\/em> is set to zero.<\/p>\n<p><\/p>\n<p>Adding a message:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_put(MSBufferizer *obj, mblk_t *m); <\/code><\/pre>\n<p><\/p>\n<p>Message m is added to the queue. The calculated size of data blocks is added to <em>size<\/em>.<\/p>\n<p><\/p>\n<p>Transferring all messages from the link's data queue to the bufferizer <em>q<\/em>:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_put_from_queue(MSBufferizer *obj, MSQueue *q);   <\/code><\/pre>\n<p><\/p>\n<p>Messages from the link <em>q<\/em> to the bufferizer are transferred using the function <em>ms_bufferizer_put()<\/em>.<\/p>\n<p><\/p>\n<p>Reading from the bufferizer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">int ms_bufferizer_read(MSBufferizer *obj, uint8_t *data, int datalen); <\/code><\/pre>\n<p><\/p>\n<p>If the size of the accumulated data in the bufferizer is less than the requested amount (<em>datalen<\/em>), the function returns zero, and data copying to data is not performed. Otherwise, sequential data copying occurs from the tuples present in the buffer. After copying, the tuple is removed and memory is freed. The copying ends when datalen bytes have been copied. If space runs out in the middle of a data block, the data block will be truncated to the remaining uncopied portion. The next call will continue copying from this point.<\/p>\n<p><\/p>\n<p>Reading the amount of data currently available in the buffer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">int ms_bufferizer_get_avail(MSBufferizer *obj); <\/code><\/pre>\n<p><\/p>\n<p>Returns the field <em>size<\/em> of the buffer.<\/p>\n<p><\/p>\n<p>Discarding part of the data in the buffer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_skip_bytes(MSBufferizer *obj, int bytes);<\/code><\/pre>\n<p><\/p>\n<p>The specified number of bytes of data is extracted and discarded. The oldest data is discarded.<\/p>\n<p><\/p>\n<p>Removing all messages in the buffer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_flush(MSBufferizer *obj); <\/code><\/pre>\n<p><\/p>\n<p>The data counter is reset to zero.<\/p>\n<p><\/p>\n<p>Removing all messages in the buffer:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_uninit(MSBufferizer *obj); <\/code><\/pre>\n<p><\/p>\n<p>Resetting the counter is not performed.<\/p>\n<p><\/p>\n<p>Removing the buffer and freeing memory:<\/p>\n<p><\/p>\n<pre><code class=\"cpp\">void ms_bufferizer_destroy(MSBufferizer *obj);  <\/code><\/pre>\n<p><\/p>\n<p>Examples of using the buffer can be found in the source code of several media streamer filters. For example, in the filter MS_L16_ENC, which performs byte rearrangement in the samples from network order to host order:<noindex><a rel=\"nofollow\" href=\"https:\/\/github.com\/BelledonneCommunications\/mediastreamer2\/blob\/1a9e8dec369a65ed10db975f5b77b92dc999b096\/src\/audiofilters\/l16.c\"> l16.c<\/a><\/noindex><\/p>\n<p><\/p>\n<p>In the next article, we will discuss load assessment on the ticker and ways to combat excessive computational load in the media streamer.<\/p>\n<p>Source: <a content=\"nofollow\" rel=\"nofollow\" href=\"https:\/\/habr.com\/ru\/post\/499010\/\">habr.com<\/a> <\/p>","protected":false,"gt_translate_keys":[{"key":"rendered","format":"html"}]},"excerpt":{"rendered":"<p>\u041c\u0430\u0442\u0435\u0440\u0438\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 \u0432\u0437\u044f\u0442 \u0441 \u043c\u043e\u0435\u0433\u043e \u0434\u0437\u0435\u043d-\u043a\u0430\u043d\u0430\u043b\u0430. \u041c\u0435\u0445\u0430\u043d\u0438\u0437\u043c \u043f\u0435\u0440\u0435\u043c\u0435\u0449\u0435\u043d\u0438\u044f \u0434\u0430\u043d\u043d\u044b\u0445 \u0411\u043b\u043e\u043a \u0434\u0430\u043d\u043d\u044b\u0445 dblk_t \u0421\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u0435 mblk_t \u0424\u0443\u043d\u043a\u0446\u0438\u0438 \u0440\u0430\u0431\u043e\u0442\u044b \u0441 \u0441\u043e\u043e\u0431\u0449\u0435\u043d\u0438\u044f\u043c\u0438 mblk_t \u041e\u0447\u0435\u0440\u0435\u0434\u044c queue_t \u0424\u0443\u043d\u043a\u0446\u0438\u0438 \u0440\u0430\u0431\u043e\u0442\u044b \u0441 \u043e\u0447\u0435\u0440\u0435\u0434\u044f\u043c\u0438 queue_t \u0421\u043e\u0435\u0434\u0438\u043d\u0435\u043d\u0438\u0435 \u0444\u0438\u043b\u044c\u0442\u0440\u043e\u0432 \u0421\u0438\u0433\u043d\u0430\u043b\u044c\u043d\u0430\u044f \u0442\u043e\u0447\u043a\u0430 \u0433\u0440\u0430\u0444\u0430 \u043e\u0431\u0440\u0430\u0431\u043e\u0442\u043a\u0438 \u0434\u0430\u043d\u043d\u044b\u0445 \u0417\u0430\u043a\u0443\u043b\u0438\u0441\u043d\u0430\u044f \u0434\u0435\u044f\u0442\u0435\u043b\u044c\u043d\u043e\u0441\u0442\u044c \u0442\u0438\u043a\u0435\u0440\u0430 \u0411\u0443\u0444\u0435\u0440\u0438\u0437\u0430\u0442\u043e\u0440 (MSBufferizer) \u0424\u0443\u043d\u043a\u0446\u0438\u0438 \u0440\u0430\u0431\u043e\u0442\u044b \u0441 MSBufferizer \u0412 \u043f\u0440\u043e\u0448\u043b\u043e\u0439 \u0441\u0442\u0430\u0442\u044c\u0435 \u043c\u044b \u0440\u0430\u0437\u0440\u0430\u0431\u043e\u0442\u0430\u043b\u0438 \u0441\u0432\u043e\u0439 \u0441\u043e\u0431\u0441\u0442\u0432\u0435\u043d\u043d\u044b\u0439 \u0444\u0438\u043b\u044c\u0442\u0440. \u042d\u0442\u0443 \u0441\u0442\u0430\u0442\u044c\u044e \u043c\u044b \u043f\u043e\u0441\u0432\u0435\u0442\u0438\u043c \u0443\u0441\u0442\u0440\u043e\u0439\u0441\u0442\u0432\u0443 [&hellip;]<\/p>\n","protected":false,"gt_translate_keys":[{"key":"rendered","format":"html"}]},"author":1,"featured_media":79541,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[688],"tags":[],"class_list":["post-79540","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-administrirovanie"],"aioseo_notices":[],"aioseo_head":"\n\t\t<!-- All in One SEO 5.0.2 - aioseo.com -->\n\t<meta name=\"description\" content=\"\u041c\u0430\u0442\u0435\u0440\u0438\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 \u0432\u0437\u044f\u0442 \u0441 \u043c\u043e\u0435\u0433\u043e \u0434\u0437\u0435\u043d-\u043a\u0430\u043d\u0430\u043b\u0430.\" \/>\n\t<meta name=\"robots\" content=\"max-image-preview:large\" \/>\n\t<meta name=\"author\" content=\"Yuri Gagarin\"\/>\n\t<link rel=\"canonical\" href=\"https:\/\/prohoster.info\/en\/blog\/administrirovanie\/izuchaem-voip-dvizhok-mediastreamer2-chast-11\" \/>\n\t<meta name=\"generator\" content=\"All in One SEO (AIOSEO) 5.0.2\" \/>\n\t\t<meta property=\"og:locale\" content=\"en_US\" \/>\n\t\t<meta property=\"og:site_name\" content=\"ProHoster | \u041a\u0443\u043f\u0438\u0442\u044c \u043d\u0430\u0434\u0435\u0436\u043d\u044b\u0439 \u0445\u043e\u0441\u0442\u0438\u043d\u0433 \u0434\u043b\u044f \u0441\u0430\u0439\u0442\u043e\u0432 \u0441 \u0437\u0430\u0449\u0438\u0442\u043e\u0439 \u043e\u0442 DDoS, VPS VDS \u0441\u0435\u0440\u0432\u0435\u0440\u044b\" \/>\n\t\t<meta property=\"og:type\" content=\"article\" \/>\n\t\t<meta property=\"og:title\" content=\"\ud83e\udd47\u0418\u0437\u0443\u0447\u0430\u0435\u043c VoIP-\u0434\u0432\u0438\u0436\u043e\u043a Mediastreamer2. \u0427\u0430\u0441\u0442\u044c 11 | ProHoster\" \/>\n\t\t<meta property=\"og:description\" content=\"\u041c\u0430\u0442\u0435\u0440\u0438\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 \u0432\u0437\u044f\u0442 \u0441 \u043c\u043e\u0435\u0433\u043e \u0434\u0437\u0435\u043d-\u043a\u0430\u043d\u0430\u043b\u0430.\" \/>\n\t\t<meta property=\"og:url\" content=\"https:\/\/prohoster.info\/en\/blog\/administrirovanie\/izuchaem-voip-dvizhok-mediastreamer2-chast-11\" \/>\n\t\t<meta property=\"og:image\" content=\"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg\" \/>\n\t\t<meta property=\"og:image:secure_url\" content=\"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg\" \/>\n\t\t<meta property=\"og:image:width\" content=\"350\" \/>\n\t\t<meta property=\"og:image:height\" content=\"350\" \/>\n\t\t<meta property=\"article:published_time\" content=\"2020-04-28T05:42:01+00:00\" \/>\n\t\t<meta property=\"article:modified_time\" content=\"2020-04-28T05:42:01+00:00\" \/>\n\t\t<meta property=\"article:publisher\" content=\"https:\/\/www.facebook.com\/prohoster\" \/>\n\t\t<meta property=\"article:author\" content=\"https:\/\/www.facebook.com\/prohoster\" \/>\n\t\t<!-- All in One SEO -->\n\n","aioseo_head_json":{"title":"\ud83e\udd47Exploring the VoIP engine Mediastreamer2. Part 11 | ProHoster","description":"The content of the article is taken from my Zen channel.","canonical_url":"https:\/\/prohoster.info\/en\/blog\/administrirovanie\/izuchaem-voip-dvizhok-mediastreamer2-chast-11","robots":"max-image-preview:large","keywords":"","webmasterTools":{"miscellaneous":""},"schema":null,"og:locale":"en_US","og:site_name":"ProHoster | \u041a\u0443\u043f\u0438\u0442\u044c \u043d\u0430\u0434\u0435\u0436\u043d\u044b\u0439 \u0445\u043e\u0441\u0442\u0438\u043d\u0433 \u0434\u043b\u044f \u0441\u0430\u0439\u0442\u043e\u0432 \u0441 \u0437\u0430\u0449\u0438\u0442\u043e\u0439 \u043e\u0442 DDoS, VPS VDS \u0441\u0435\u0440\u0432\u0435\u0440\u044b","og:type":"article","og:title":"\ud83e\udd47\u0418\u0437\u0443\u0447\u0430\u0435\u043c VoIP-\u0434\u0432\u0438\u0436\u043e\u043a Mediastreamer2. \u0427\u0430\u0441\u0442\u044c 11 | ProHoster","og:description":"\u041c\u0430\u0442\u0435\u0440\u0438\u0430\u043b \u0441\u0442\u0430\u0442\u044c\u0438 \u0432\u0437\u044f\u0442 \u0441 \u043c\u043e\u0435\u0433\u043e \u0434\u0437\u0435\u043d-\u043a\u0430\u043d\u0430\u043b\u0430.","og:url":"https:\/\/prohoster.info\/en\/blog\/administrirovanie\/izuchaem-voip-dvizhok-mediastreamer2-chast-11","og:image":"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg","og:image:secure_url":"https:\/\/prohoster.info\/wp-content\/uploads\/2021\/11\/logo-350.jpg","og:image:width":350,"og:image:height":350,"article:published_time":"2020-04-28T05:42:01+00:00","article:modified_time":"2020-04-28T05:42:01+00:00","article:publisher":"https:\/\/www.facebook.com\/prohoster","article:author":"https:\/\/www.facebook.com\/prohoster"},"aioseo_meta_data":{"post_id":"79540","title":null,"description":null,"keywords":null,"keyphrases":null,"primary_term":null,"canonical_url":null,"og_title":null,"og_description":null,"og_object_type":"default","og_image_type":"default","og_image_url":null,"og_image_width":null,"og_image_height":null,"og_image_custom_url":null,"og_image_custom_fields":null,"og_video":null,"og_custom_url":null,"og_article_section":null,"og_article_tags":null,"twitter_use_og":false,"twitter_card":"default","twitter_image_type":"default","twitter_image_url":null,"twitter_image_custom_url":null,"twitter_image_custom_fields":null,"twitter_title":null,"twitter_description":null,"schema":{"blockGraphs":[],"customGraphs":[],"default":{"data":{"Article":[],"Course":[],"Dataset":[],"FAQPage":[],"Movie":[],"Person":[],"Product":[],"ProductReview":[],"Car":[],"Recipe":[],"Service":[],"SoftwareApplication":[],"WebPage":[]},"graphName":"","isEnabled":true},"graphs":[]},"schema_type":null,"schema_type_options":null,"pillar_content":false,"robots_default":true,"robots_noindex":false,"robots_noarchive":false,"robots_nosnippet":false,"robots_nofollow":false,"robots_noimageindex":false,"robots_noodp":false,"robots_notranslate":false,"robots_max_snippet":null,"robots_max_videopreview":null,"robots_max_imagepreview":"large","priority":null,"frequency":null,"local_seo":null,"seo_analyzer_scan_date":null,"breadcrumb_settings":null,"limit_modified_date":false,"reviewed_by":null,"ai":null,"created":"2021-02-28 12:40:33","updated":"2022-10-06 16:26:59","focus_keyword":null,"additional_keywords":null,"truseo_locale":null},"gt_translate_keys":[{"key":"link","format":"url"}],"_links":{"self":[{"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/posts\/79540","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/comments?post=79540"}],"version-history":[{"count":0,"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/posts\/79540\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/media\/79541"}],"wp:attachment":[{"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/media?parent=79540"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/categories?post=79540"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/prohoster.info\/en\/wp-json\/wp\/v2\/tags?post=79540"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}