/src/freeradius-server/src/lib/io/message.c

Source
/*
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program 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 General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */

/**
 * $Id: 942c5dd3333163dfce530ae8c179cb7dae61d1a6 $
 *
 * @brief Messages for inter-thread communication
 * @file io/message.c
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSID("$Id: 942c5dd3333163dfce530ae8c179cb7dae61d1a6 $")

#include <freeradius-devel/io/message.h>
#include <freeradius-devel/util/strerror.h>


/*
 *  Debugging, mainly for message_set_test
 */
#if 0
#define MPRINT(...) fprintf(stderr, __VA_ARGS__)
#else
#define MPRINT(...)
#endif

#define MSG_ARRAY_SIZE (16)

#define CACHE_ALIGN(_x) do { _x += 63; _x &= ~(size_t) 63; } while (0)

/** A Message set, composed of message headers and ring buffer data.
 *
 *  A message set is intended to send short-lived messages.  The
 *  message headers are fixed in size, and allocated from an array
 *  which is treated like a circular buffer.  Message bodies (i.e. raw
 *  packets) are variable in size, and live in a separate ring buffer.
 *
 *  The message set starts off with a small array of message headers,
 *  and a small ring buffer.  If an array/buffer fills up, a new one
 *  is allocated at double the size of the previous one.
 *
 * The array / buffers are themselves kept in fixed-size arrays, of
 *  MSG_ARRAY_SIZE.  The reason is that the memory for fr_message_set_t
 *  should be contiguous, and not in a linked list scattered in
 *  memory.
 *
 *  The originator allocates a message, and sends it to a recipient.
 *  The recipient (usually in another thread) uses the message, and
 *  marks it as FR_MESSAGE_DONE.  The originator then asynchronously
 *  cleans up the message.
 *
 *  This asynchronous cleanup is done via self-clocking.  If there is
 *  no need to clean up the messages, it isn't done.  Only when we run
 *  out of space to store messages (or packets) is the cleanup done.
 *
 *  This cleanup latency ensures that we don't have cache line
 *  bouncing, where the originator sends the message, and then while
 *  the recipieent is reading it... thrashes the cache line with
 *  checks for "are you done?  Are you done?"
 *
 *  If there are more than one used entry in either array, we then try
 *  to coalesce the buffers on cleanup.  If we discover that one array
 *  is empty, we discard it, and move the used array entries into it's
 *  place.
 *
 *  This process ensures that we don't have too many buffers in
 *  progress.  It is better to have a few large buffers than many
 *  small ones.
 *
 *  MSG_ARRAY_SIZE is defined to be large (16 doublings) to allow for
 *  the edge case where messages are stuck for long periods of time.
 *
 *  With an initial message array size of 64, this gives us room for
 *  2M packets, if *all* of the mr_array entries have packets stuck in
 *  them that aren't cleaned up for extended periods of time.
 *
 *  @todo Add a flag for UDP-style protocols, where we can put the
 *  message into the ring buffer.  This helps with locality of
 *  reference, and removes the need to track two separate things.
 */
struct fr_message_set_s {
  int     mr_current; //!< current used message ring entry
  int     mr_max;   //!< max used message ring entry

  size_t      message_size; //!< size of the callers message, including fr_message_t

  int     mr_cleaned; //!< where we last cleaned

  int     rb_current; //!< current used ring buffer entry
  int     rb_max;   //!< max used ring buffer entry

  size_t      max_allocation; //!< maximum allocation size

  int     allocated;
  int     freed;

  fr_ring_buffer_t  *mr_array[MSG_ARRAY_SIZE]; //!< array of message arrays

  fr_ring_buffer_t  *rb_array[MSG_ARRAY_SIZE]; //!< array of ring buffers
};


/** Create a message set
 *
 * @param[in] ctx the context for talloc
 * @param[in] num_messages size of the initial message array.  MUST be a power of 2.
 * @param[in] message_size the size of each message, INCLUDING fr_message_t, which MUST be at the start of the struct
 * @param[in] ring_buffer_size of the ring buffer.  MUST be a power of 2.
 * @param[in] unlimited_size allow any message size to be allocated.  If false it is limited to ring_buffer_size / 2.
 * @return
 *  - NULL on error
 *  - newly allocated fr_message_set_t on success
 */
fr_message_set_t *fr_message_set_create(TALLOC_CTX *ctx, int num_messages, size_t message_size, size_t ring_buffer_size,
          bool unlimited_size)
{
  fr_message_set_t *ms;

  /*
   *  Too small, or not a power of 2.
   */
  if (num_messages < 8) num_messages = 8;

  if ((num_messages & (num_messages - 1)) != 0) {
    fr_strerror_const("Number of messages must be a power of 2");
    return NULL;
  }

  if (message_size < sizeof(fr_message_t)) {
    fr_strerror_printf("Message size must be at least %zd", sizeof(fr_message_t));
    return NULL;
  }

  if (message_size > 1024) {
    fr_strerror_const("Message size must be no larger than 1024");
    return NULL;
  }

  ms = talloc_zero(ctx, fr_message_set_t);
  if (!ms) {
    fr_strerror_const("Failed allocating memory");
    return NULL;
  }

  CACHE_ALIGN(message_size);
  ms->message_size = message_size;

  ms->rb_array[0] = fr_ring_buffer_create(ms, ring_buffer_size);
  if (!ms->rb_array[0]) {
    talloc_free(ms);
    return NULL;
  }
  ms->rb_max = 0;

  ms->mr_array[0] = fr_ring_buffer_create(ms, num_messages * message_size);
  if (!ms->mr_array[0]) {
    talloc_free(ms);
    return NULL;
  }

  /*
   * If unlimited size is allowed, set the max allocation to 1 << 29
   * which is based on the maximum ring buffer reservation of 1 << 30
   */
  ms->max_allocation = unlimited_size ? 1 << 29 : ring_buffer_size / 2;

  return ms;
}


/** Mark a message as done
 *
 *  Note that this call is usually done from a thread OTHER than the
 *  originator of the message.  As such, the message is NOT actually
 *  freed.  Instead, it is just marked as freed.
 *
 * @param[in] m the message to make as done.
 * @return
 *  - <0 on error
 *  - 0 on success
 */
int fr_message_done(fr_message_t *m)
{
  fr_assert(m->status != FR_MESSAGE_FREE);
  fr_assert(m->status != FR_MESSAGE_DONE);

  /*
   *  Mark a message as freed.  The originator will take
   *  care of cleaning it up.
   */
  if (m->status == FR_MESSAGE_USED) {
    m->status = FR_MESSAGE_DONE;
    return 0;
  }

  /*
   *  This message was localized, so we can free it via
   *  talloc.
   */
  if (m->status == FR_MESSAGE_LOCALIZED) {
    talloc_free(m);
    return 0;
  }

  /*
   *  A catastrophic error.
   */
  fr_assert(0 == 1);

  fr_strerror_const("Failed marking message as done");
  return -1;
}


/** Localize a message by copying it to local storage
 *
 *  This function "localizes" a message by copying it to local
 *  storage.  In the case where the recipient of a message has to sit
 *  on it for a while, that blocks the originator from cleaning up the
 *  message.  The recipient can then copy the message to local
 *  storage, so that the originator can clean it up.
 *
 *  The localized message is marked as FR_MESSAGE_LOCALIZED, so that
 *  the recipient can call the normal fr_message_done() function to
 *  free it.
 *
 * @param[in] ctx the talloc context to use for localization
 * @param[in] m the message to be localized
 * @param[in] message_size the size of the message, including the fr_message_t
 * @return
 *  - NULL on allocation error
 *  - a newly localized message
 */
fr_message_t *fr_message_localize(TALLOC_CTX *ctx, fr_message_t *m, size_t message_size)
{
  fr_message_t *l;

  if (m->status != FR_MESSAGE_USED) {
    fr_strerror_const("Cannot localize message unless it is in use");
    return NULL;
  }

  if (message_size < sizeof(fr_message_t)) {
    fr_strerror_const("Message size is too small");
    return NULL;
  }

  l = talloc_memdup(ctx, m, message_size);
  if (!l) {
  nomem:
    fr_strerror_const("Failed allocating memory");
    return NULL;
  }

  l->data = NULL;

  if (l->data_size) {
    l->data = talloc_memdup(l, m->data, l->data_size);
    if (!l->data) {
      talloc_free(l);
      goto nomem;
    }
  }

  l->status = FR_MESSAGE_LOCALIZED;

  /*
   *  After this change, "m" should not be used for
   *  anything.
   */
  m->status = FR_MESSAGE_DONE;

  /*
   *  Now clean up the other fields of the newly localized
   *  message.
   */
  l->rb = NULL;
  l->rb_size = 0;

  return l;
}


/** Clean up messages in a message ring.
 *
 *  Find the oldest messages which are marked FR_MESSAGE_DONE,
 *  and mark them FR_MESSAGE_FREE.
 *
 *  FIXME: If we care, track which ring buffer is in use, and how
 *  many contiguous chunks we can free.  Then, free the chunks at
 *  once, instead of piecemeal.  Realistically tho... this will
 *  probably make little difference.
 *
 * @param[in] ms the message set
 * @param[in] mr the message ring
 * @param[in] max_to_clean maximum number of messages to clean at a time.
 */
static int message_ring_gc(fr_message_set_t *ms, fr_ring_buffer_t *mr, int max_to_clean)
{
  int messages_cleaned = 0;
  size_t size;
  fr_message_t *m;

  while (true) {
    (void) fr_ring_buffer_start(mr, (uint8_t **) &m, &size);
    if (size == 0) break;

    fr_assert(m != NULL);
    fr_assert(size >= ms->message_size);

    fr_assert(m->status != FR_MESSAGE_FREE);
    if (m->status != FR_MESSAGE_DONE) break;

    messages_cleaned++;
    m->status = FR_MESSAGE_FREE;
    ms->freed++;

    if (m->rb) {
      (void) fr_ring_buffer_free(m->rb, m->rb_size);
#ifndef NDEBUG
      memset(m, 0, ms->message_size);
#endif
    }

    fr_ring_buffer_free(mr, ms->message_size);

    if (messages_cleaned >= max_to_clean) break;
  }

  MPRINT("CLEANED %d (%p) left\n", messages_cleaned, mr);
  return messages_cleaned;
}


/** Garbage collect "done" messages.
 *
 *  Called only from the originating thread.  We also clean a limited
 *  number of messages at a time, so that we don't have sudden latency
 *  spikes when cleaning 1M messages.
 *
 * @param[in] ms the message set
 * @param[in] max_to_clean the maximum number of messages to clean
 */
static void message_gc(fr_message_set_t *ms, int max_to_clean)
{
  int i;
  int arrays_freed, arrays_used, empty_slot;
  int largest_free_slot;
  int total_cleaned;
  size_t largest_free_size;

  /*
   *  Clean up "done" messages.
   */
  total_cleaned = 0;

  /*
   *  Garbage collect the smaller buffers first.
   */
  for (i = 0; i <= ms->mr_max; i++) {
    int cleaned;

    cleaned = message_ring_gc(ms, ms->mr_array[i], max_to_clean - total_cleaned);
    total_cleaned += cleaned;
    fr_assert(total_cleaned <= max_to_clean);

    /*
     *  Stop when we've reached our GC limit.
     */
    if (total_cleaned == max_to_clean) break;
  }

  /*
   *  Couldn't GC anything.  Don't do more work.
   */
  if (total_cleaned == 0) return;

  arrays_freed = 0;
  arrays_used = 0;

  /*
   *  Keep the two largest message buffers (used or not),
   *  and free all smaller ones which are empty.
   */
  for (i = ms->mr_max; i >= 0; i--) {
    fr_assert(ms->mr_array[i] != NULL);

    if (arrays_used < 2) {
      MPRINT("\tleaving entry %d alone\n", i);
      arrays_used++;
      continue;
    }

    /*
     *  If the message ring buffer is empty, check if
     *  we should perhaps delete it.
     */
    if (fr_ring_buffer_used(ms->mr_array[i]) == 0) {
      MPRINT("\tfreeing entry %d\n", i);
      TALLOC_FREE(ms->mr_array[i]);
      arrays_freed++;
      continue;
    }

    MPRINT("\tstill in use entry %d\n", i);
  }

  /*
   *  Some entries have been freed.  We need to coalesce the
   *  remaining entries.
   */
  if (arrays_freed) {
    MPRINT("TRYING TO PACK from %d free arrays out of %d\n", arrays_freed, ms->rb_max + 1);

    empty_slot = -1;

    /*
     *  Pack the rb array by moving used entries to
     *  the bottom of the array.
     */
    for (i = 0; i <= ms->mr_max; i++) {
      int j;

      /*
       *  Skip over empty entries, but set
       *  "empty_slot" to the first empty on we
       *  found.
       */
      if (!ms->mr_array[i]) {
        if (empty_slot < 0) empty_slot = i;

        continue;
      }

      /*
       *  This array entry is used, but there is
       *  no empty slot to put it into.  Ignore
       *  it, and continue
       */
      if (empty_slot < 0) continue;

      fr_assert(ms->mr_array[empty_slot] == NULL);

      ms->mr_array[empty_slot] = ms->mr_array[i];
      ms->mr_array[i] = NULL;

      /*
       *  Find the next empty slot which is
       *  greater than the one we just used.
       */
      for (j = empty_slot + 1; j <= i; j++) {
        if (!ms->mr_array[j]) {
          empty_slot = j;
          break;
        }
      }
    }

    /*
     *  Lower max, and set current to the largest
     *  array, whether or not it's used.
     */
    ms->mr_max -= arrays_freed;
    ms->mr_current = ms->mr_max;

#ifndef NDEBUG
    MPRINT("NUM RB ARRAYS NOW %d\n", ms->mr_max + 1);
    for (i = 0; i <= ms->mr_max; i++) {
      MPRINT("\t%d %p\n", i, ms->mr_array[i]);
      fr_assert(ms->mr_array[i] != NULL);
    }
#endif
  }

  /*
   *  And now we do the same thing for the ring buffers.
   *  Except that freeing the messages above also cleaned up
   *  the contents of each ring buffer, so all we need to do
   *  is find the largest empty ring buffer.
   *
   *  We do this by keeping the two largest ring buffers
   *  (used or not), and then freeing all smaller ones which
   *  are empty.
   */
  arrays_used = 0;
  arrays_freed = 0;
  MPRINT("TRYING TO FREE ARRAYS %d\n", ms->rb_max);
  for (i = ms->rb_max; i >= 0; i--) {
    fr_assert(ms->rb_array[i] != NULL);

    if (arrays_used < 2) {
      MPRINT("\tleaving entry %d alone\n", i);
      arrays_used++;
      continue;
    }

    if (fr_ring_buffer_used(ms->rb_array[i]) == 0) {
      MPRINT("\tfreeing entry %d\n", i);
      TALLOC_FREE(ms->rb_array[i]);
      arrays_freed++;
      continue;
    }

    MPRINT("\tstill in use entry %d\n", i);
  }

  /*
   *  Pack the array entries back down.
   */
  if (arrays_freed > 0) {
    MPRINT("TRYING TO PACK from %d free arrays out of %d\n", arrays_freed, ms->rb_max + 1);

    empty_slot = -1;

    /*
     *  Pack the rb array by moving used entries to
     *  the bottom of the array.
     */
    for (i = 0; i <= ms->rb_max; i++) {
      int j;

      /*
       *  Skip over empty entries, but set
       *  "empty_slot" to the first empty on we
       *  found.
       */
      if (!ms->rb_array[i]) {
        if (empty_slot < 0) empty_slot = i;

        continue;
      }

      /*
       *  This array entry is used, but there is
       *  no empty slot to put it into.  Ignore
       *  it, and continue
       */
      if (empty_slot < 0) continue;

      fr_assert(ms->rb_array[empty_slot] == NULL);

      ms->rb_array[empty_slot] = ms->rb_array[i];
      ms->rb_array[i] = NULL;

      /*
       *  Find the next empty slot which is
       *  greater than the one we just used.
       */
      for (j = empty_slot + 1; j <= i; j++) {
        if (!ms->rb_array[j]) {
          empty_slot = j;
          break;
        }
      }
    }

    /*
     *  Lower max, and set current to the largest
     *  array, whether or not it's used.
     */
    ms->rb_max -= arrays_freed;
    ms->rb_current = ms->rb_max;

#ifndef NDEBUG
    MPRINT("NUM RB ARRAYS NOW %d\n", ms->rb_max + 1);
    for (i = 0; i <= ms->rb_max; i++) {
      MPRINT("\t%d %p\n", i, ms->rb_array[i]);
      fr_assert(ms->rb_array[i] != NULL);
    }
#endif
  }

  /*
   *  Set the current ring buffer to the one with the
   *  largest free space in it.
   *
   *  This is different from the allocation strategy for
   *  messages.
   */
  if (!fr_cond_assert(ms->rb_array[ms->rb_max] != NULL)) return;

  largest_free_slot = ms->rb_max;
  largest_free_size = (fr_ring_buffer_size(ms->rb_array[ms->rb_max]) -
           fr_ring_buffer_used(ms->rb_array[ms->rb_max]));

  for (i = 0; i < ms->rb_max; i++) {
    size_t free_size;

    fr_assert(ms->rb_array[i] != NULL);

    free_size = (fr_ring_buffer_size(ms->rb_array[i]) -
           fr_ring_buffer_used(ms->rb_array[i]));
    if (largest_free_size < free_size) {
      largest_free_slot = i;
      largest_free_size = free_size;
    }
  }

  ms->rb_current = largest_free_slot;
  fr_assert(ms->rb_current >= 0);
  fr_assert(ms->rb_current <= ms->rb_max);
}

typedef uint8_t *(*fr_ring_buffer_op_t)(fr_ring_buffer_t *rb, size_t size);

/** Find a message ring with space, performing GC and expansion as needed.
 *
 *  The caller passes in the operation to perform on the ring buffer
 *  (fr_ring_buffer_alloc or fr_ring_buffer_reserve).  This function
 *  handles GC, searching across all message rings, and doubling the
 *  ring size when all are full.
 *
 * @param[out] p_cleaned a flag to indicate if we cleaned the message array
 * @param[in] ms the message set
 * @param[in] op the ring buffer operation (alloc or reserve)
 * @return
 *  - NULL on error
 *      - pointer to message-sized region on success
 */
static inline CC_HINT(always_inline)
uint8_t *_message_ring_find(bool *p_cleaned, fr_message_set_t *ms, fr_ring_buffer_op_t op)
{
  int i;
  uint8_t *p;
  fr_ring_buffer_t *mr;

  *p_cleaned = false;

  /*
   *  Grab the current message array.  In the general case,
   *  there's room, so we grab a message and go find a ring
   *  buffer.
   */
  mr = ms->mr_array[ms->mr_current];
  p = op(mr, ms->message_size);
  if (p) {
    MPRINT("RING FIND normal\n");
    return p;
  }

  MPRINT("CLEANING UP (%d - %d = %d)\n", ms->allocated, ms->freed,
    ms->allocated - ms->freed);

  /*
   *  Else the buffer is full.  Do a global cleanup.
   */
  message_gc(ms, 128);
  *p_cleaned = true;

  /*
   *  If we're lucky, the cleanup has given us a new
   *  "current" buffer, which is empty.  If so, use it.
   *
   *  Do a small per-ring GC first to free up the oldest
   *  entries.  This results in a small amount of cache line
   *  thrashing, but if the buffer is full, it's likely
   *  that the oldest entry can be freed.
   */
  mr = ms->mr_array[ms->mr_current];
  (void) message_ring_gc(ms, mr, 4);
  p = op(mr, ms->message_size);
  if (p) {
    MPRINT("RING FIND after cleanup\n");
    return p;
  }

  /*
   *  We've tried two allocations, and both failed.  Brute
   *  force over all arrays, trying to allocate one
   *  somewhere... anywhere.  We start from the largest
   *  array, because that is the one we want to use the
   *  most.
   *
   *  We want to avoid allocations in the smallest array,
   *  because that array will quickly wrap, and will cause
   *  us to do cleanups more often.  That also lets old
   *  entries in the smallest array age out, so that we can
   *  free the smallest arrays.
   */
  for (i = ms->mr_max; i >= 0; i--) {
    mr = ms->mr_array[i];

    (void) message_ring_gc(ms, mr, 4);
    p = op(mr, ms->message_size);
    if (p) {
      ms->mr_current = i;
      MPRINT("RING FIND from changed ring buffer\n");
      MPRINT("SET MR to changed %d\n", ms->mr_current);
      return p;
    }
  }

  /*
   *  All of the arrays are full.  If we don't have
   *  room to allocate another array, we're dead.
   */
  if ((ms->mr_max + 1) >= MSG_ARRAY_SIZE) {
    fr_strerror_const("All message arrays are full");
    return NULL;
  }

  /*
   *  Allocate another message ring, double the size
   *  of the previous maximum.
   */
  mr = fr_ring_buffer_create(ms, fr_ring_buffer_size(ms->mr_array[ms->mr_max]) * 2);
  if (!mr) {
    fr_strerror_const_push("Failed allocating ring buffer");
    return NULL;
  }

  /*
   *  Set the new one as current for all new
   *  allocations, and perform the operation on the
   *  entirely empty message ring.
   */
  ms->mr_max++;
  ms->mr_current = ms->mr_max;
  ms->mr_array[ms->mr_max] = mr;

  MPRINT("SET MR to doubled %d\n", ms->mr_current);

  return op(mr, ms->message_size);
}

/** Allocate a fr_message_t, WITHOUT a ring buffer
 *
 * @param[in] ms the message set
 * @param[out] p_cleaned a flag to indicate if we cleaned the message array
 * @return
 *      - NULL on error
 *  - fr_message_t* on success
 */
static inline CC_HINT(always_inline) fr_message_t *message_alloc(fr_message_set_t *ms, bool *p_cleaned)
{
  fr_message_t *m;

  m = (fr_message_t *)_message_ring_find(p_cleaned, ms, fr_ring_buffer_alloc);
  if (!m) return NULL;

  memset(m, 0, ms->message_size);
  m->status = FR_MESSAGE_USED;
  ms->allocated++;

  return m;
}

/** Reserve a fr_message_t, WITHOUT a ring buffer
 *
 * @note This only _reserves_ the message, i.e. checks there's space, and initialises the message
 *   memory, it does not perform the actual allocation.  This is useful if there are error
 *   paths which could result in the message needing to be returned to the message set without
 *   being used.  Multiple pending reservations are NOT permitted.
 *
 * @param[in] ms the message set to reserve the message in.
 * @param[out] p_cleaned a flag to indicate if we cleaned the message array
 * @return
 *      - NULL on error
 *  - fr_message_t* on success
 */
static inline CC_HINT(always_inline) fr_message_t *message_reserve(fr_message_set_t *ms, bool *p_cleaned)
{
  fr_message_t *m;

  m = (fr_message_t *)_message_ring_find(p_cleaned, ms, fr_ring_buffer_reserve);
  if (!m) return NULL;

  memset(m, 0, ms->message_size);
  m->status = FR_MESSAGE_USED;

  return m;
}

/** Finalise a reserved message allocation
 *
 * @note This finalises the allocation of a previous message.  No additional reserve calls can be made
 *   between the one that initialised memory for m, and this finalise call.
 *
 * @param[in] ms the message set to finalise the allocation in.
 * @param[in] m the message to perform the allocation for.
 * @return
 *      - <0 on error
 *  - 0 on success.
 */
static inline CC_HINT(always_inline) int message_finalise(fr_message_set_t *ms, fr_message_t *m)
{
  bool cleaned = false;
  fr_message_t *nm;

  /*
   *  This _should_ use the same ring as before, unless someone has
   *  interleaved _another_ reserve call between the initial reserve
   *  call and this finalisation.
   */
  nm = (fr_message_t *)_message_ring_find(&cleaned, ms, fr_ring_buffer_alloc);
  if (!fr_cond_assert_msg(nm == m, "Alloc interleaved between reserving and finalising message allocation %p", m)) {
    return -1;
  }

  ms->allocated++;

  return 0;
}

/** Reserve data in a ring buffer for this message
 *
 * @param[in] ms the message set
 * @param[in] m the message
 * @param[in] cleaned_up whether the message set was partially garbage collected
 * @return
 *  - NULL on error, and m is deallocated
 *  - m on success
 */
static fr_message_t *message_data_reserve(fr_message_set_t *ms, fr_message_t *m, bool cleaned_up)
{
  int i;
  fr_ring_buffer_t *rb;
  size_t alloc_size;

  /*
   *  And... we go through a bunch of hoops, all over again.
   */
  m->rb = ms->rb_array[ms->rb_current];
  fr_assert(m->rb != NULL);
  m->data = fr_ring_buffer_reserve(m->rb, m->rb_size);
  if (m->data) return m;

  /*
   *  When the simple allocation fails, ensure we don't do
   *  the cleanup twice in one allocation.
   */
  if (!cleaned_up) {
    /*
     *  If we run out of room in the current ring
     *  buffer, AND it's our only one, then just
     *  double it in size.
     */
    if (ms->rb_max == 0) goto alloc_rb;

    /*
     *  We're using multiple ring buffers, and we
     *  haven't already done a cleanup.  Force a
     *  cleanup.
     */
    MPRINT("CLEANED UP BECAUSE OF RING BUFFER (%d - %d = %d)\n", ms->allocated, ms->freed,
      ms->allocated - ms->freed);

    message_gc(ms, 128);

    /*
     *  Try to allocate the packet from the newly current ring buffer.
     */
    m->rb = ms->rb_array[ms->rb_current];
    fr_assert(m->rb != NULL);
    m->data = fr_ring_buffer_reserve(m->rb, m->rb_size);
    if (m->data) return m;

    MPRINT("CLEANUP RING BUFFER FAILED\n");
  }

  /*
   *  We've tried two allocations, and both failed.  Brute
   *  force over all arrays, trying to allocate one
   *  somewhere... anywhere.  We start from the largest
   *  array, because that is the one we want to use the
   *  most.
   *
   *  We want to avoid allocations in the smallest array,
   *  because that array will quickly wrap, and will cause
   *  us to do cleanups more often.  That also lets old
   *  entries in the smallest array age out, so that we can
   *  free the smallest arrays.
   */
  for (i = ms->rb_max; i >= 0; i--) {
    m->rb = ms->rb_array[i];
    fr_assert(m->rb != NULL);
    m->data = fr_ring_buffer_reserve(m->rb, m->rb_size);
    if (m->data) {
      MPRINT("MOVED TO RING BUFFER %d\n", i);
      ms->rb_current = i;
      return m;
    }
  }

  /*
   *  All of the arrays are full.  If we don't have
   *  room to allocate another array, we're dead.
   */
  if ((ms->rb_max + 1) >= MSG_ARRAY_SIZE) {
    fr_strerror_const("Message arrays are full");
    goto cleanup;
  }

alloc_rb:
  /*
   *  Allocate another message ring, double the size
   *  of the previous maximum, or large enough to hold
   *  twice the requested size if that is larger.
   *  fr_ring_buffer_create will round to the next power of 2.
   */
  alloc_size = fr_ring_buffer_size(ms->rb_array[ms->rb_max]) * 2;
  if (alloc_size < m->rb_size) {
    alloc_size = m->rb_size * 2;
  }
  rb = fr_ring_buffer_create(ms, alloc_size);
  if (!rb) {
    fr_strerror_const_push("Failed allocating ring buffer");
    goto cleanup;
  }

  MPRINT("RING BUFFER DOUBLES\n");

  /*
   *  Set the new one as current for all new
   *  allocations, allocate a message, and go try to
   *  reserve room for the raw packet data.
   */
  ms->rb_max++;
  ms->rb_current = ms->rb_max;
  ms->rb_array[ms->rb_current] = rb;

  /*
   *  And we should now have an entirely empty message ring.
   */
  m->rb = rb;
  m->data = fr_ring_buffer_reserve(m->rb, m->rb_size);
  if (m->data) return m;

cleanup:
  MPRINT("OUT OF MEMORY\n");

  m->rb = NULL;
  m->status = FR_MESSAGE_DONE;
  return NULL;
}


/** Reserve a message
 *
 *  A later call to fr_message_and_data_commit() will commit the reservation.
 *  This call just reserves a message header and space for the packet data.
 *
 *  If the caller later decides that the message is not needed, he
 *  should call fr_message_free() to free the message.
 *
 *  We assume that the caller will call fr_message_and_data_reserve(), and then
 *  almost immediately fr_message_and_data_commit().  Multiple calls in series
 *  to fr_message_and_data_reserve() MUST NOT be done.  The caller could also
 *  just call fr_ring_buffer_alloc(m->rb, size) if they wanted, and
 *  then update m->data_size by hand...
 *
 *  The message is returned
 *
 * @param[in] ms the message set
 * @param[in] reserve_size to reserve
 * @return
 *      - NULL on error
 *  - fr_message_t* on success
 */
fr_message_t *fr_message_and_data_reserve(fr_message_set_t *ms, size_t reserve_size)
{
  bool cleaned_up;
  fr_message_t *m;

  (void) talloc_get_type_abort(ms, fr_message_set_t);

  if (reserve_size > ms->max_allocation) {
    fr_strerror_printf("Cannot reserve %zd > max allocation %zd\n", reserve_size, ms->max_allocation);
    return NULL;
  }

  /*
   *  Reserve a bare message.
   */
  m = message_reserve(ms, &cleaned_up);
  if (!m) {
    MPRINT("Failed to reserve message\n");
    return NULL;
  }

  /*
   *  If the caller is not allocating any packet data, just
   *  return the empty message.
   */
  if (!reserve_size) return m;

  /*
   *  We leave m->data_size as zero, and m->rb_size as the
   *  reserved size.  This indicates that the message has
   *  reserved room for the packet data, but nothing has
   *  been allocated.
   */
  CACHE_ALIGN(reserve_size);
  m->rb_size = reserve_size;

  return message_data_reserve(ms, m, cleaned_up);
}

/** Cancel a reservation made by fr_message_and_data_reserve(), returning the slot to the set.
 *
 *  Both rings are uncommitted at this point (write_offset was not advanced).
 *  Clearing the message fields is sufficient: the next reserve overwrites the
 *  slot, and the data ring's reserved field is replaced by the new reservation size.
 *
 *  Callers MUST verify m->data_size == 0 before calling this.  If a concurrent
 *  fr_message_and_data_alloc() aliased this reservation (same ring slot, same
 *  data address), m->data_size will be non-zero and the slot is already committed;
 *  calling this function in that case corrupts the committed message.
 *
 * @param[in] ms  the message set the reservation was made against.
 * @param[in] m   the previously reserved message to cancel.
 */
void fr_message_and_data_reset(fr_message_set_t *ms, fr_message_t *m)
{
  (void) talloc_get_type_abort(ms, fr_message_set_t);
  fr_assert(m->status == FR_MESSAGE_USED);
  m->status = FR_MESSAGE_FREE;
  m->rb = NULL;
  m->data = NULL;
  m->data_size = 0;
  m->rb_size = 0;
}

/** Commit a previously reserved message, allocating exactly total_size bytes of packet data.
 *
 *  Commits both the message ring slot and the data ring buffer.  total_size is the
 *  final packet size; it may be less than the original reservation.  For streaming
 *  reads where m->data_size is already non-zero (leftover bytes from a prior read),
 *  total_size must include those bytes.
 *
 * @param[in] ms the message set
 * @param[in] m the previously reserved message (m is NOT talloc'd)
 * @param[in] total_size final packet size in bytes
 * @return
 *      - NULL on error, input message m is left alone
 *  - m on success
 */
fr_message_t *fr_message_and_data_commit(fr_message_set_t *ms, fr_message_t *m, size_t total_size)
{
  uint8_t *p;
  size_t align_size;

  (void) talloc_get_type_abort(ms, fr_message_set_t);

  /* m is NOT talloc'd */
  if (!m) {
    m = fr_message_and_data_reserve(ms, total_size);
    if (!m) return NULL;
  }

  fr_assert(m->status == FR_MESSAGE_USED);
  fr_assert(m->rb_size >= total_size);

  if (total_size == 0) {
    m->data = NULL;
    m->rb = NULL;
    m->data_size = m->rb_size = 0;
    message_finalise(ms, m);
    return m;
  }

  fr_assert(m->rb != NULL);
  fr_assert(m->data != NULL);

  align_size = total_size;
  CACHE_ALIGN(align_size);

  p = fr_ring_buffer_alloc(m->rb, align_size);
  fr_assert(p != NULL);
  if (!p) {
    fr_strerror_const_push("Failed allocating from ring buffer");
    return NULL;
  }

  fr_assert(p == m->data);

  m->data_size = total_size;
  m->rb_size = align_size;
  message_finalise(ms, m);

  return m;
}

/** Reserve and commit a message atomically.
 *
 *  Combines fr_message_and_data_reserve() and fr_message_and_data_commit() into a
 *  single call for callers that know the exact packet size upfront.
 *
 * @param[in] ms the message set
 * @param[in] size packet size in bytes
 * @return
 *      - NULL on error
 *  - fr_message_t* on success
 */
fr_message_t *fr_message_and_data_alloc(fr_message_set_t *ms, size_t size)
{
  fr_message_t *m;

  m = fr_message_and_data_reserve(ms, size);
  if (!m) return NULL;

  return fr_message_and_data_commit(ms, m, size);
}

/** Allocate packet data for a message, and reserve a new message
 *
 *  This function allocates a previously reserved message, and then
 *  reserves a new message.
 *
 *  The application should call fr_message_and_data_reserve() with a large
 *  buffer, and then read data into the buffer.  If the buffer
 *  contains multiple packets, the application should call
 *  fr_message_and_data_commit_with_leftover() repeatedly to allocate the full
 *  packets, while reserving room for the partial packet.
 *
 *  When the application is determines that there is only one full
 *  packet, and one partial packet in the buffer, it should call this
 *  function with actual_packet_size, and a large reserve_size.  The
 *  partial packet will be reserved.  If the ring buffer is full, the
 *  partial packet will be copied to a new ring buffer.
 *
 *  When the application determines that there are multiple full
 *  packets in the buffer, it should call this function with
 *  actual_packet_size for each buffer, and reserve_size which
 *  reserves all of the data in the buffer.  i.e. the full packets +
 *  partial packets, which should start off as the original
 *  reserve_size.
 *
 *  The application should call this function to allocate each packet,
 *  while decreasing reserve_size by each actual_packet_size that was
 *  allocated.  Once there is only one full and a partial packet in
 *  the buffer, it should use a large reserve_size, as above.
 *
 *  The application could just always ecall this function with a large
 *  reserve_size, at the cost of substantially more memcpy()s.
 *
 * @param[in] ms the message set
 * @param[in] m the message message to allocate packet data for
 * @param[in] actual_packet_size to use
 * @param[in] leftover "dirty" bytes in the buffer
 * @param[in] reserve_size to reserve for new message
 * @return
 *      - NULL on error, and input message m is left alone
 *  - fr_message_t* on success.  Will always be a new message.
 */
fr_message_t *fr_message_and_data_commit_with_leftover(fr_message_set_t *ms, fr_message_t *m,
               size_t actual_packet_size,
               size_t leftover, size_t reserve_size)
{
  bool cleaned_up;
  uint8_t *p;
  fr_message_t *m2;
  size_t m_rb_size, align_size;

  (void) talloc_get_type_abort(ms, fr_message_set_t);

  align_size = actual_packet_size;
  CACHE_ALIGN(align_size);

  /* m is NOT talloc'd */

  fr_assert(m->status == FR_MESSAGE_USED);
  fr_assert(m->rb != NULL);
  fr_assert(m->data != NULL);
  fr_assert(m->rb_size >= actual_packet_size);

  p = fr_ring_buffer_alloc(m->rb, align_size);
  fr_assert(p != NULL);
  if (!p) {
    fr_strerror_const_push("Failed allocating from ring buffer");
    return NULL;
  }

  fr_assert(p == m->data);

  m_rb_size = m->rb_size; /* for ring buffer cleanups */
  m->data_size = actual_packet_size;
  m->rb_size = align_size;
  message_finalise(ms, m);

  /*
   *  If we've allocated all of the reserved ring buffer
   *  data, then just reserve a brand new reservation.
   *
   *  This will be automatically cache aligned.
   */
  if (!leftover) return fr_message_and_data_reserve(ms, reserve_size);

  /*
   *  Reserve a new message.
   */
  m2 = message_reserve(ms, &cleaned_up);
  if (!m2) return NULL;

  /*
   *  Ensure that there's enough room to shift the next
   *  packet, so that it's cache aligned.  Moving small
   *  amounts of memory is likely faster than having two
   *  CPUs fight over the same cache lines.
   */
  reserve_size += (align_size - actual_packet_size);
  CACHE_ALIGN(reserve_size);

  /*
   *  Track how much data there is in the packet.
   */
  m2->rb = m->rb;
  m2->data_size = leftover;
  m2->rb_size = reserve_size;

  /*
   *  Try to extend the reservation.  If we can do it,
   *  return.
   */
  m2->data = fr_ring_buffer_reserve(m2->rb, reserve_size);
  if (m2->data) {
    /*
     *  The next packet pointer doesn't point to the
     *  actual data after the current packet.  Move
     *  the next packet to match up with the ring
     *  buffer allocation.
     */
    if (m2->data != (m->data + actual_packet_size)) {
      memmove(m2->data, m->data + actual_packet_size, leftover);
    }
    return m2;
  }

  /*
   *  We failed reserving more memory at the end of the
   *  current ring buffer.
   *
   *  Reserve data from a new ring buffer.  If it doesn't
   *  succeed, ensure that the old message will properly
   *  clean up the old ring buffer.
   */
  if (!message_data_reserve(ms, m2, false)) {
    m->rb_size = m_rb_size;
    return NULL;
  }

  /*
   *  If necessary, copy the remaining data from the old
   *  buffer to the new one.
   */
  if (m2->data != (m->data + actual_packet_size)) {
    memmove(m2->data, m->data + actual_packet_size, leftover);
  }

  /*
   *  The messages are in different ring buffers.  We've
   *  aligned m->rb_size above for the current packet, but
   *  there's no subsequent message to clean up this
   *  reservation.  Re-extend the current message to it's
   *  original size, so that cleaning it up will clean up the ring buffer.
   */
  if (m2->rb != m->rb) {
    m->rb_size = m_rb_size;
    return m2;
  }

  /*
   *  If we've managed to allocate the next message in the
   *  current ring buffer, then it really should have
   *  wrapped around.  In which case, re-extend the current
   *  message as above.
   */
  if (m2->data < m->data) {
    m->rb_size = m_rb_size;
    return m2;
  }

  return m2;
}

/** Count the number of used messages
 *
 * @param[in] ms the message set
 * @return
 *      - number of used messages
 */
int fr_message_set_messages_used(fr_message_set_t *ms)
{
  int i, used;

  (void) talloc_get_type_abort(ms, fr_message_set_t);

  used = 0;
  for (i = 0; i <= ms->mr_max; i++) {
    fr_ring_buffer_t *mr;

    mr = ms->mr_array[i];

    used += fr_ring_buffer_used(mr) / ms->message_size;
  }

  return used;
}

/** Garbage collect the message set.
 *
 *  This function should ONLY be called just before freeing the
 *  message set.  It is intended only for debugging, and will cause
 *  huge latency spikes if used at run time.
 *
 * @param[in] ms the message set
 */
void fr_message_set_gc(fr_message_set_t *ms)
{
  int i;

  (void) talloc_get_type_abort(ms, fr_message_set_t);

  /*
   *  Manually clean up each message ring.
   */
  for (i = 0; i <= ms->mr_max; i++) {
    (void) message_ring_gc(ms, ms->mr_array[i], INT_MAX);
  }

  /*
   *  And then do one last pass to clean up the arrays.
   */
  message_gc(ms, INT_MAX);
}

/** Print debug information about the message set.
 *
 * @param[in] ms the message set
 * @param[in] fp the FILE where the messages are printed.
 */
void fr_message_set_debug(FILE *fp, fr_message_set_t *ms)
{
  int i;

  (void) talloc_get_type_abort(ms, fr_message_set_t);

  fprintf(fp, "message arrays = %d\t(current %d)\n", ms->mr_max + 1, ms->mr_current);
  fprintf(fp, "ring buffers   = %d\t(current %d)\n", ms->rb_max + 1, ms->rb_current);

  for (i = 0; i <= ms->mr_max; i++) {
    fr_ring_buffer_t *mr = ms->mr_array[i];

    fprintf(fp, "messages[%d] =\tsize %zu, used %zu\n",
      i, fr_ring_buffer_size(mr), fr_ring_buffer_used(mr));
  }

  for (i = 0; i <= ms->rb_max; i++) {
    fprintf(fp, "ring buffer[%d] =\tsize %zu, used %zu\n",
      i, fr_ring_buffer_size(ms->rb_array[i]), fr_ring_buffer_used(ms->rb_array[i]));
  }
}

Coverage Report

Created: 2026-08-14 07:17