<commit_before>/* *
 * This file is part of Feng
 *
 * Copyright (C) 2009 by LScube team <<EMAIL>>
 * See AUTHORS for more details
 *
 * feng is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * feng is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with feng; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 *
 * */

#define G_LOG_DOMAIN "bufferqueue"

#include <config.h>

#include "media/media.h"
#include "network/rtp.h"

#include <stdbool.h>
#include <stdio.h>
#include <string.h>

/* We don't enable this with !NDEBUG because it's _massive_! */
#ifdef FENG_BQ_DEBUG
# define bq_debug(fmt, ...) g_debug("[%s] " fmt, __PRETTY_FUNCTION__, __VA_ARGS__)
#else
# define bq_debug(...)
#endif

/**
 * @defgroup bufferqueue Buffer Queue
 *
 * @brief Producer/consumer buffer queue structure
 *
 * The buffer queue is a data structure that allows one producer
 * (i.e.: a demuxer) to read data and feed it to multiple consumers
 * (i.e.: the RTSP clients).
 *
 * The structure is implemented in a very simple way: there is one big
 * double-ended queue for each producer, the elements of this queue
 * are wrappers around the actual buffer structures. Each time the
 * producer reads a buffer, it is added to its queue, with a “seen
 * count” of zero.
 *
 * Each consumer gets, at any time, the pointer to the start of the
 * queue and as it moves on to the following element, the “seen count”
 * gets incremented. Once the seen count reaches the amount of
 * consumers, the element is deleted from the queue.
 *
 * @{
 */

static inline struct MParserBuffer *GLIST_TO_BQELEM(GList *pointer)
{
    return (struct MParserBuffer*)pointer->data;
}

/**
 * @brief Ensures the validity of the current element pointer for the consume
 *
 * @param consumer The consumer to verify the pointer of
 *
 * Call this function whenever current_element_pointer is going to be
 * used; if any condition happened that would make it unreliable, set
 * it to NULL.
 */
static void bq_consumer_confirm_pointer(RTP_session *consumer)
{
    Track *producer = consumer->track;
    if ( consumer->current_element_pointer == NULL )
        return;

    if ( producer->queue_serial != consumer->queue_serial ) {
        bq_debug("C:%p pointer %p reset PQS:%lu < CQS:%lu",
                consumer,
                consumer->current_element_pointer,
                producer->queue_serial,
                consumer->queue_serial);
        consumer->current_element_pointer = NULL;
        return;
    }

    if ( producer->queue->head &&
         consumer->last_element_serial < GLIST_TO_BQELEM(producer->queue->head)->seq_no ) {
        bq_debug("C:%p pointer %p reset LES:%lu:%u < PQHS:%lu:%u",
                consumer,
                consumer->current_element_pointer,
                consumer->queue_serial, consumer->last_element_serial,
                producer->queue_serial,
                GLIST_TO_BQELEM(producer->queue->head)->seq_no);
        consumer->current_element_pointer = NULL;
        return;
    }
}

static inline struct MParserBuffer *BQ_OBJECT(RTP_session *consumer)
{
    bq_consumer_confirm_pointer(consumer);

    if ( consumer->current_element_pointer == NULL )
        return NULL;

    return (struct MParserBuffer*)consumer->current_element_pointer->data;
}


static void mparser_buffer_free(struct MParserBuffer *buffer)
{
    g_free(buffer->data);
    g_slice_free(struct MParserBuffer, buffer);
}

/**
 * @brief Destroy one by one the elements in
 *        Track::queue.
 *
 * @param elem_generic Element to destroy
 * @param free_func_generic Function to use for destroying the
 *                          elements' payload.
 */
static void bq_element_free_internal(gpointer elem_generic,
                                     ATTR_UNUSED gpointer unused) {
    struct MParserBuffer *const element = (struct MParserBuffer*)elem_generic;

    bq_debug("Free object %p %lu",
            element,
            element->seen);

    mparser_buffer_free(element);
}

/**
 * @brief Resets a producer's queue (unlocked version)
 *
 * @param producer Producer to reset the queue of
 *
 * @internal This function does not lock the producer and should only
 *           be used by @ref add_track !
 * @note This function will require exclusive access to the producer,
 *       and will thus lock its mutex.
 *
 * This function will change the currently-used queue for the
 * producer, so that a discontinuity will allow the consumers not to
 * worry about getting old buffers.
 */
static void bq_producer_reset_queue_internal(Track *producer) {
    bq_debug("Producer %p queue %p queue_serial %lu",
            producer,
            producer->queue,
            producer->queue_serial);

    if ( producer->queue ) {
        g_queue_foreach(producer->queue,
                        bq_element_free_internal,
                        NULL);
        g_queue_clear(producer->queue);
        g_queue_free(producer->queue);
    }

    producer->queue = g_queue_new();
    producer->queue_serial++;
    producer->next_serial = 1;
}

/**
 * @brief Resets a producer's queue
 *
 * @param producer Producer to reset the queue of
 *
 * @note This function will require exclusive access to the producer,
 *       and will thus lock its mutex.
 *
 * This function will change the currently-used queue for the
 * producer, so that a discontinuity will allow the consumers not to
 * worry about getting old buffers.
 */
void track_reset_queue(Track *producer) {
    bq_debug("Producer %p",
            producer);

    /* Ensure we have the exclusive access */
    g_mutex_lock(producer->lock);

    g_assert(!producer->stopped);

    bq_producer_reset_queue_internal(producer);

    /* Leave the exclusive access */
    g_mutex_unlock(producer->lock);
}

/**
 * @brief Destroy the head of the producer queue
 *
 * @param producer Producer to destroy the head queue of
 * @param element element to destroy; this is a safety check
 */
static void bq_producer_destroy_head(Track *producer, GList *pointer) {
    struct MParserBuffer *elem = pointer->data;

    bq_debug("P:%p PQH:%p pointer %p",
             producer,
             producer->queue->head,
             pointer);

    /* We can only remove the head of the queue, if we're doing
     * anything else, something is funky. Ibid if it hasn't been seen
     * yet.
     */
    g_assert(pointer == producer->queue->head);
    g_assert(elem->seen == producer->consumers);

    /* Remove the element from the queue */
    g_queue_pop_head(producer->queue);

    /* If we reached the end of the queue, consider like it was a new
     * one */
    if ( g_queue_get_length(producer->queue) == 0 ) {
        producer->queue_serial++;
        producer->next_serial = 1;
    }

    bq_element_free_internal(elem, NULL);
}

/**
 * @brief Remove reference from the current element
 *
 * @param consumer The consumer that has seen the element
 *<commit_msg>track: avoid calling bq_element_free_internal directly.<commit_after>/* *
 * This file is part of Feng
 *
 * Copyright (C) 2009 by LScube team <<EMAIL>>
 * See AUTHORS for more details
 *
 * feng is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * feng is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with feng; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 *
 * */

#define G_LOG_DOMAIN "bufferqueue"

#include <config.h>

#include "media/media.h"
#include "network/rtp.h"

#include <stdbool.h>
#include <stdio.h>
#include <string.h>

/* We don't enable this with !NDEBUG because it's _massive_! */
#ifdef FENG_BQ_DEBUG
# define bq_debug(fmt, ...) g_debug("[%s] " fmt, __PRETTY_FUNCTION__, __VA_ARGS__)
#else
# define bq_debug(...)
#endif

/**
 * @defgroup bufferqueue Buffer Queue
 *
 * @brief Producer/consumer buffer queue structure
 *
 * The buffer queue is a data structure that allows one producer
 * (i.e.: a demuxer) to read data and feed it to multiple consumers
 * (i.e.: the RTSP clients).
 *
 * The structure is implemented in a very simple way: there is one big
 * double-ended queue for each producer, the elements of this queue
 * are wrappers around the actual buffer structures. Each time the
 * producer reads a buffer, it is added to its queue, with a “seen
 * count” of zero.
 *
 * Each consumer gets, at any time, the pointer to the start of the
 * queue and as it moves on to the following element, the “seen count”
 * gets incremented. Once the seen count reaches the amount of
 * consumers, the element is deleted from the queue.
 *
 * @{
 */

static inline struct MParserBuffer *GLIST_TO_BQELEM(GList *pointer)
{
    return (struct MParserBuffer*)pointer->data;
}

/**
 * @brief Ensures the validity of the current element pointer for the consume
 *
 * @param consumer The consumer to verify the pointer of
 *
 * Call this function whenever current_element_pointer is going to be
 * used; if any condition happened that would make it unreliable, set
 * it to NULL.
 */
static void bq_consumer_confirm_pointer(RTP_session *consumer)
{
    Track *producer = consumer->track;
    if ( consumer->current_element_pointer == NULL )
        return;

    if ( producer->queue_serial != consumer->queue_serial ) {
        bq_debug("C:%p pointer %p reset PQS:%lu < CQS:%lu",
                consumer,
                consumer->current_element_pointer,
                producer->queue_serial,
                consumer->queue_serial);
        consumer->current_element_pointer = NULL;
        return;
    }

    if ( producer->queue->head &&
         consumer->last_element_serial < GLIST_TO_BQELEM(producer->queue->head)->seq_no ) {
        bq_debug("C:%p pointer %p reset LES:%lu:%u < PQHS:%lu:%u",
                consumer,
                consumer->current_element_pointer,
                consumer->queue_serial, consumer->last_element_serial,
                producer->queue_serial,
                GLIST_TO_BQELEM(producer->queue->head)->seq_no);
        consumer->current_element_pointer = NULL;
        return;
    }
}

static inline struct MParserBuffer *BQ_OBJECT(RTP_session *consumer)
{
    bq_consumer_confirm_pointer(consumer);

    if ( consumer->current_element_pointer == NULL )
        return NULL;

    return (struct MParserBuffer*)consumer->current_element_pointer->data;
}


static void mparser_buffer_free(struct MParserBuffer *buffer)
{
    bq_debug("Free object %p %lu",
             buffer,
             buffer->seen);

    g_free(buffer->data);
    g_slice_free(struct MParserBuffer, buffer);
}

/**
 * @brief Destroy one by one the elements in
 *        Track::queue.
 *
 * @param elem_generic Element to destroy
 * @param free_func_generic Function to use for destroying the
 *                          elements' payload.
 */
static void bq_element_free_internal(gpointer elem_generic,
                                     ATTR_UNUSED gpointer unused) {
    mparser_buffer_free((struct MParserBuffer*)elem_generic);
}

/**
 * @brief Resets a producer's queue (unlocked version)
 *
 * @param producer Producer to reset the queue of
 *
 * @internal This function does not lock the producer and should only
 *           be used by @ref add_track !
 * @note This function will require exclusive access to the producer,
 *       and will thus lock its mutex.
 *
 * This function will change the currently-used queue for the
 * producer, so that a discontinuity will allow the consumers not to
 * worry about getting old buffers.
 */
static void bq_producer_reset_queue_internal(Track *producer) {
    bq_debug("Producer %p queue %p queue_serial %lu",
            producer,
            producer->queue,
            producer->queue_serial);

    if ( producer->queue ) {
        g_queue_foreach(producer->queue,
                        bq_element_free_internal,
                        NULL);
        g_queue_clear(producer->queue);
        g_queue_free(producer->queue);
    }

    producer->queue = g_queue_new();
    producer->queue_serial++;
    producer->next_serial = 1;
}

/**
 * @brief Resets a producer's queue
 *
 * @param producer Producer to reset the queue of
 *
 * @note This function will require exclusive access to the producer,
 *       and will thus lock its mutex.
 *
 * This function will change the currently-used queue for the
 * producer, so that a discontinuity will allow the consumers not to
 * worry about getting old buffers.
 */
void track_reset_queue(Track *producer) {
    bq_debug("Producer %p",
            producer);

    /* Ensure we have the exclusive access */
    g_mutex_lock(producer->lock);

    g_assert(!producer->stopped);

    bq_producer_reset_queue_internal(producer);

    /* Leave the exclusive access */
    g_mutex_unlock(producer->lock);
}

/**
 * @brief Destroy the head of the producer queue
 *
 * @param producer Producer to destroy the head queue of
 * @param element element to destroy; this is a safety check
 */
static void bq_producer_destroy_head(Track *producer, GList *pointer) {
    struct MParserBuffer *elem = pointer->data;

    bq_debug("P:%p PQH:%p pointer %p",
             producer,
             producer->queue->head,
             pointer);

    /* We can only remove the head of the queue, if we're doing
     * anything else, something is funky. Ibid if it hasn't been seen
     * yet.
     */
    g_assert(pointer == producer->queue->head);
    g_assert(elem->seen == producer->consumers);

    /* Remove the element from the queue */
    g_queue_pop_head(producer->queue);

    /* If we reached the end of the queue, consider like it was a new
     * one */
    if ( g_queue_get_length(producer->queue) == 0 ) {
        producer->queue_serial++;
        producer->next_serial = 1;
    }

    mparser_buffer_free(elem);
}

/**
 * @brief Remove reference from the current element
 *
 * @param consumer The consumer that has seen the element
 *
{'fraction_non_alphanumeric': 0.0912096066923025, 'fraction_numerical': 0.0026985090737367606, 'mean_word_length': 3.0064864864864864, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 117, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '64851', 'n_tokens_mistral': 4283, 'n_tokens_neox': 4163, 'n_words': 1504}