message.c

Go to the documentation of this file.

/*
 *   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: bdb8eae283d2d7e88bd6fc25fd04034fc0946734 $
 *
 * @brief Messages for inter-thread communication
 * @file io/message.c
 *
 * @copyright 2016 Alan DeKok (aland@freeradius.org)
 */
RCSID("$Id: bdb8eae283d2d7e88bd6fc25fd04034fc0946734 $")
 
#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 %zu", 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;
 
                /*
                 *      A data ring buffer only frees from its tail, and
                 *      is shared by every message ring.  The rings are
                 *      collected one at a time, so this message's data
                 *      can sit behind a live message from another ring.
                 *      Leave it until its data is the oldest in the
                 *      buffer, or we free someone else's bytes.
                 */
                if (m->rb) {
                        uint8_t *start;
                        size_t  used;
 
                        (void) fr_ring_buffer_start(m->rb, &start, &used);
                        if (start != m->data) 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
 * @param[in] compact whether to free empty rings and repack the arrays
 */
static void message_gc(fr_message_set_t *ms, int max_to_clean, bool compact)
{
        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;
 
        if (!compact) 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, true);
        *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);
 
                /*
                 *      The message slot is reserved but not finalised,
                 *      so only collect: see message_gc().
                 */
                message_gc(ms, 128, false);
 
                /*
                 *      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 %zu > max allocation %zu\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, cleaned;
 
        (void) talloc_get_type_abort(ms, fr_message_set_t);
 
        /*
         *      Manually clean up each message ring.  A ring can be
         *      blocked by a done message whose data sits behind
         *      another ring's, so keep going until a pass frees
         *      nothing.
         */
        do {
                cleaned = 0;
                for (i = 0; i <= ms->mr_max; i++) {
                        cleaned += message_ring_gc(ms, ms->mr_array[i], INT_MAX);
                }
        } while (cleaned > 0);
 
        /*
         *      And then do one last pass to clean up the arrays.
         */
        message_gc(ms, INT_MAX, true);
}
 
/** 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]));
        }
}