data_buffer.c

Go to the documentation of this file.

/*
 * Asterisk -- An open source telephony toolkit.
 *
 * Copyright (C) 2018, Digium, Inc.
 *
 * Joshua Colp <[email protected]>
 * Ben Ford <[email protected]>
 *
 * See http://www.asterisk.org for more information about
 * the Asterisk project. Please do not directly contact
 * any of the maintainers of this project for assistance;
 * the project provides a web site, mailing lists and IRC
 * channels for your use.
 *
 * This program is free software, distributed under the terms of
 * the GNU General Public License Version 2. See the LICENSE file
 * at the top of the source tree.
 */

/*! \file
 *
 * \brief Data Buffer API
 *
 * \author Joshua Colp <[email protected]>
 * \author Ben Ford <[email protected]>
 */

/*** MODULEINFO
   <support_level>core</support_level>
 ***/

#include "asterisk.h"

#include "asterisk/logger.h"
#include "asterisk/strings.h"
#include "asterisk/data_buffer.h"
#include "asterisk/linkedlists.h"

/*!
 * \brief The number of payloads to increment the cache by
 */
#define CACHED_PAYLOAD_MAX 5

/*!
 * \brief Payload entry placed inside of the data buffer list
 */
struct data_buffer_payload_entry {
   /*! \brief The payload for this position */
   void *payload;
   /*! \brief The provided position for this */
   size_t pos;
   /*! \brief Linked list information */
   AST_LIST_ENTRY(data_buffer_payload_entry) list;
};

/*!
 * \brief Data buffer containing fixed number of data payloads
 */
struct ast_data_buffer {
   /*! \brief Callback function to free a data payload */
   ast_data_buffer_free_callback free_fn;
   /*! \brief A linked list of data payloads */
   AST_LIST_HEAD_NOLOCK(, data_buffer_payload_entry) payloads;
   /*! \brief A linked list of unused cached data payloads */
   AST_LIST_HEAD_NOLOCK(, data_buffer_payload_entry) cached_payloads;
   /*! \brief The current number of data payloads in the buffer */
   size_t count;
   /*! \brief Maximum number of data payloads in the buffer */
   size_t max;
   /*! \brief The current number of data payloads in the cache */
   size_t cache_count;
};

static void free_fn_do_nothing(void *data)
{
   return;
}

/*!
 * \brief Helper function to allocate a data payload
 */
static struct data_buffer_payload_entry *data_buffer_payload_alloc(void *payload, size_t pos)
{
   struct data_buffer_payload_entry *data_payload;

   data_payload = ast_calloc(1, sizeof(*data_payload));
   if (!data_payload) {
      return NULL;
   }

   data_payload->payload = payload;
   data_payload->pos = pos;

   return data_payload;
}

/*!
 * \brief Helper function that sets the cache to its maximum number of payloads
 */
static void ast_data_buffer_cache_adjust(struct ast_data_buffer *buffer)
{
   int buffer_space;

   ast_assert(buffer != NULL);

   buffer_space = buffer->max - buffer->count;

   if (buffer->cache_count == buffer_space) {
      return;
   }

   if (buffer->cache_count < buffer_space) {
      /* Add payloads to the cache, if able */
      while (buffer->cache_count < CACHED_PAYLOAD_MAX && buffer->cache_count < buffer_space) {
         struct data_buffer_payload_entry *buffer_payload;

         buffer_payload = data_buffer_payload_alloc(NULL, -1);
         if (buffer_payload) {
            AST_LIST_INSERT_TAIL(&buffer->cached_payloads, buffer_payload, list);
            buffer->cache_count++;
            continue;
         }

         ast_log(LOG_ERROR, "Failed to allocate memory to the cache.");
         break;
      }
   } else if (buffer->cache_count > buffer_space) {
      /* Remove payloads from the cache */
      while (buffer->cache_count > buffer_space) {
         struct data_buffer_payload_entry *buffer_payload;

         buffer_payload = AST_LIST_REMOVE_HEAD(&buffer->cached_payloads, list);
         if (buffer_payload) {
            ast_free(buffer_payload);
            buffer->cache_count--;
            continue;
         }

         ast_log(LOG_ERROR, "Failed to remove memory from the cache.");
         break;
      }
   }
}

struct ast_data_buffer *ast_data_buffer_alloc(ast_data_buffer_free_callback free_fn, size_t size)
{
   struct ast_data_buffer *buffer;

   ast_assert(size != 0);

   buffer = ast_calloc(1, sizeof(*buffer));
   if (!buffer) {
      return NULL;
   }

   AST_LIST_HEAD_INIT_NOLOCK(&buffer->payloads);
   AST_LIST_HEAD_INIT_NOLOCK(&buffer->cached_payloads);

   /* If free_fn is NULL, just use free_fn_do_nothing as a default */
   buffer->free_fn = free_fn ? free_fn : free_fn_do_nothing;
   buffer->max = size;

   ast_data_buffer_cache_adjust(buffer);

   return buffer;
}

void ast_data_buffer_resize(struct ast_data_buffer *buffer, size_t size)
{
   struct data_buffer_payload_entry *existing_payload;

   ast_assert(buffer != NULL);

   /* The buffer must have at least a size of 1 */
   ast_assert(size > 0);

   if (buffer->max == size) {
      return;
   }

   /* If the size is decreasing, some payloads will need to be freed */
   if (buffer->max > size) {
      int remove = buffer->max - size;

      AST_LIST_TRAVERSE_SAFE_BEGIN(&buffer->payloads, existing_payload, list) {
         if (remove) {
            AST_LIST_REMOVE_HEAD(&buffer->payloads, list);
            buffer->free_fn(existing_payload->payload);
            ast_free(existing_payload);
            buffer->count--;
            remove--;
            continue;
         }
         break;
      }
      AST_LIST_TRAVERSE_SAFE_END;
   }

   buffer->max = size;
   ast_data_buffer_cache_adjust(buffer);
}

int ast_data_buffer_put(struct ast_data_buffer *buffer, size_t pos, void *payload)
{
   struct data_buffer_payload_entry *buffer_payload = NULL;
   struct data_buffer_payload_entry *existing_payload;
   int inserted = 0;

   ast_assert(buffer != NULL);
   ast_assert(payload != NULL);

   /* If the data buffer has reached its maximum size then the head goes away and
    * we will reuse its buffer payload
    */
   if (buffer->count == buffer->max) {
      buffer_payload = AST_LIST_REMOVE_HEAD(&buffer->payloads, list);
      buffer->free_fn(buffer_payload->payload);
      buffer->count--;

      /* Update this buffer payload with its new information */
      buffer_payload->payload = payload;
      buffer_payload->pos = pos;
   }
   if (!buffer_payload) {
      if (!buffer->cache_count) {
         ast_data_buffer_cache_adjust(buffer);
      }
      buffer_payload = AST_LIST_REMOVE_HEAD(&buffer->cached_payloads, list);
      buffer->cache_count--;

      /* Update the payload from the cache with its new information */
      buffer_payload->payload = payload;
      buffer_payload->pos = pos;
   }
   if (!buffer_payload) {
      return -1;
   }

   /* Given the position find its ideal spot within the buffer */
   AST_LIST_TRAVERSE_SAFE_BEGIN(&buffer->payloads, existing_payload, list) {
      /* If it's already in the buffer, drop it */
      if (existing_payload->pos == pos) {
         ast_debug(3, "Packet with position %zu is already in buffer. Not inserting.\n", pos);
         inserted = -1;
         break;
      }

      if (existing_payload->pos > pos) {
         AST_LIST_INSERT_BEFORE_CURRENT(buffer_payload, list);
         inserted = 1;
         break;
      }
   }
   AST_LIST_TRAVERSE_SAFE_END;

   if (inserted == -1) {
      return -1;
   }

   if (!inserted) {
      AST_LIST_INSERT_TAIL(&buffer->payloads, buffer_payload, list);
   }

   buffer->count++;

   return 0;
}

void *ast_data_buffer_get(const struct ast_data_buffer *buffer, size_t pos)
{
   struct data_buffer_payload_entry *buffer_payload;

   ast_assert(buffer != NULL);

   AST_LIST_TRAVERSE(&buffer->payloads, buffer_payload, list) {
      if (buffer_payload->pos == pos) {
         return buffer_payload->payload;
      }
   }

   return NULL;
}

static void data_buffer_free_buffer_payload(struct ast_data_buffer *buffer,
   struct data_buffer_payload_entry *buffer_payload)
{
   buffer_payload->payload = NULL;
   buffer->count--;

   if (buffer->cache_count < CACHED_PAYLOAD_MAX
         && buffer->cache_count < (buffer->max - buffer->count)) {
      AST_LIST_INSERT_TAIL(&buffer->cached_payloads, buffer_payload, list);
      buffer->cache_count++;
   } else {
      ast_free(buffer_payload);
   }
}

void *ast_data_buffer_remove(struct ast_data_buffer *buffer, size_t pos)
{
   struct data_buffer_payload_entry *buffer_payload;

   ast_assert(buffer != NULL);

   AST_LIST_TRAVERSE_SAFE_BEGIN(&buffer->payloads, buffer_payload, list) {
      if (buffer_payload->pos == pos) {
         void *payload = buffer_payload->payload;

         AST_LIST_REMOVE_CURRENT(list);
         data_buffer_free_buffer_payload(buffer, buffer_payload);

         return payload;
      }
   }
   AST_LIST_TRAVERSE_SAFE_END;

   return NULL;
}

void *ast_data_buffer_remove_head(struct ast_data_buffer *buffer)
{
   ast_assert(buffer != NULL);

   if (buffer->count > 0) {
      struct data_buffer_payload_entry *buffer_payload;
      void *payload;

      buffer_payload = AST_LIST_REMOVE_HEAD(&buffer->payloads, list);
      payload = buffer_payload->payload;
      data_buffer_free_buffer_payload(buffer, buffer_payload);

      return payload;
   }

   return NULL;
}

void ast_data_buffer_free(struct ast_data_buffer *buffer)
{
   struct data_buffer_payload_entry *buffer_payload;

   ast_assert(buffer != NULL);

   while ((buffer_payload = AST_LIST_REMOVE_HEAD(&buffer->payloads, list))) {
      buffer->free_fn(buffer_payload->payload);
      ast_free(buffer_payload);
   }

   while ((buffer_payload = AST_LIST_REMOVE_HEAD(&buffer->cached_payloads, list))) {
      ast_free(buffer_payload);
   }

   ast_free(buffer);
}

size_t ast_data_buffer_count(const struct ast_data_buffer *buffer)
{
   ast_assert(buffer != NULL);

   return buffer->count;
}

size_t ast_data_buffer_max(const struct ast_data_buffer *buffer)
{
   ast_assert(buffer != NULL);

   return buffer->max;
}