load.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: ed962a7e3768f414686f5b0d854eb8380b5485cf $
 *
 * @brief Load generation algorithms
 * @file io/load.c
 *
 * @copyright 2019 Network RADIUS SAS (legal@networkradius.com)
 */
RCSID("$Id: ed962a7e3768f414686f5b0d854eb8380b5485cf $")
 
#include <freeradius-devel/io/load.h>
 
/*
 *      We use *inverse* numbers to avoid numerical calculation issues.
 *
 *      i.e. The bad way is to take two small numbers divide them by
 *      alpha / beta and then add them.  That process can drop the
 *      lower digits.  Instead, we take two small numbers, add them,
 *      and then divide the result by alpha / beta.
 */
#define IBETA (4)
#define IALPHA (8)
 
#define DIFF(_rtt, _t) \
        (\
                fr_time_delta_lt(_rtt, _t) ? \
                        fr_time_delta_sub(_t, _rtt) : \
                        fr_time_delta_sub(_rtt, _t)\
        )
 
#define RTTVAR(_rtt, _rttvar, _t) \
        fr_time_delta_div(\
                fr_time_delta_add(\
                        fr_time_delta_mul(_rttvar, IBETA - 1), \
                        DIFF(_rtt, _t)\
                ), \
                fr_time_delta_wrap(IBETA)\
        )
 
#define RTT(_old, _new) fr_time_delta_wrap((fr_time_delta_unwrap(_new) + (fr_time_delta_unwrap(_old) * (IALPHA - 1))) / IALPHA)
 
typedef enum {
        FR_LOAD_STATE_INIT = 0,
        FR_LOAD_STATE_SENDING,
        FR_LOAD_STATE_GATED,
        FR_LOAD_STATE_DRAINING,
} fr_load_state_t;
 
struct fr_load_s {
        fr_load_state_t         state;
        fr_event_list_t         *el;
        fr_load_config_t const *config;
        fr_load_callback_t      callback;
        fr_load_done_callback_t done;
        void                    *uctx;
 
        fr_load_stats_t         stats;                  //!< sending statistics
        fr_time_t               step_start;             //!< when the current step started
        fr_time_t               step_end;               //!< when the current step will end
        int                     step_received;
        int                     sent_base;              //!< stats.sent when this run started, so that
                                                        ///< max_requests counts per-run, not cumulatively
 
        uint32_t                pps;
        fr_time_delta_t         delta;                  //!< between packets
 
        uint32_t                count;
        bool                    header;                 //!< for printing statistics
 
        fr_time_t               next;                   //!< The next time we're supposed to send a packet
        fr_timer_t              *ev;
};
 
fr_load_t *fr_load_generator_create(TALLOC_CTX *ctx, fr_event_list_t *el, fr_load_config_t *config,
                                    fr_load_callback_t callback, fr_load_done_callback_t done,
                                    void *uctx)
{
        fr_load_t *l;
 
        l = talloc_zero(ctx, fr_load_t);
        if (!l) return NULL;
 
        if (!config->start_pps) config->start_pps = 1;
        if (!config->milliseconds) config->milliseconds = 1000;
        if (!config->parallel) config->parallel = 1;
 
        l->el = el;
        l->config = config;
        l->callback = callback;
        l->done = done;
        l->uctx = uctx;
 
        return l;
}
 
/** Send one or more packets.
 *
 */
static void fr_load_generator_send(fr_load_t *l, fr_time_t now, int count)
{
        int i;
 
        /*
         *      Send as many packets as necessary.
         */
        l->stats.sent += count;
        l->stats.last_send = now;
 
        /*
         *      Run the callback AFTER we set the timer.  Which makes
         *      it more likely that the next timer fires on time.
         */
        for (i = 0; i < count; i++) {
                l->callback(fr_time_add(now, fr_time_delta_from_nsec(i)), l->uctx);
        }
}
 
/** Stop sending new packets, and wait for outstanding replies.
 *
 *  If every reply has already arrived then the test is complete right
 *  now, so run the "done" callback.  Otherwise completion is signalled
 *  by fr_load_generator_have_reply() when the last reply comes in.
 */
static void load_drain(fr_load_t *l, fr_time_t now)
{
        l->state = FR_LOAD_STATE_DRAINING;
 
        if (l->stats.received >= l->stats.sent) {
                l->stats.end = now;
                if (l->done) l->done(l->uctx);
        }
}
 
static void load_timer(fr_timer_list_t *tl, fr_time_t now, void *uctx)
{
        fr_load_t *l = uctx;
        fr_time_delta_t delta;
        uint32_t count;
 
        /*
         *      Keep track of the overall maximum backlog for the
         *      duration of the entire test run.
         */
        l->stats.backlog = l->stats.sent - l->stats.received;
        if (l->stats.backlog > l->stats.max_backlog) l->stats.max_backlog = l->stats.backlog;
 
        /*
         *      If we're done this step, go to the next one.
         */
        if (fr_time_gteq(l->next, l->step_end)) {
                l->step_start = l->next;
                l->step_end = fr_time_add(l->next, l->config->duration);
                l->step_received = l->stats.received;
                l->pps += l->config->step;
                if (l->pps > (UINT32_MAX / l->config->milliseconds)) l->pps = UINT32_MAX / l->config->milliseconds;
 
                /*
                 *      The ramp-up has passed max_pps.  If there's no
                 *      reason to keep sending, stop and drain.
                 *      Otherwise hold the rate at max_pps, until
                 *      max_requests packets have been sent (checked
                 *      below), or forever for "unlimited".
                 */
                if (l->config->max_pps && (l->pps > l->config->max_pps)) {
                        if (!l->config->max_requests && !l->config->unlimited) {
                                load_drain(l, now);
                                return;
                        }
 
                        l->pps = l->config->max_pps;
                }
 
                l->stats.pps = l->pps;
                l->stats.skipped = 0;
                l->delta = fr_time_delta_div(fr_time_delta_from_sec(l->config->parallel), fr_time_delta_wrap(l->pps));
        }
 
        /*
         *      We don't have "pps" packets in the backlog, go send
         *      some more.  We scale the backlog by 1000 milliseconds
         *      per second.  Then, multiple the PPS by the number of
         *      milliseconds of backlog we want to keep.
         *
         *      If the backlog is smaller than packets/s *
         *      milliseconds of backlog, then keep sending.
         *      Otherwise, switch to a gated mode where we only send
         *      new packets once a reply comes in.
         */
        if (((size_t) l->stats.backlog * 1000) < ((size_t) l->pps * l->config->milliseconds)) {
                uint32_t capacity;
 
                l->state = FR_LOAD_STATE_SENDING;
                l->stats.blocked = false;
                count = l->config->parallel;
                l->stats.skipped = 0;
 
                capacity = ((l->pps * l->config->milliseconds) / 1000) - l->stats.backlog;
 
                /*
                 *      Limit "count" so that it doesn't overflow.
                 */
                if (count > capacity) count = capacity;
 
        } else {
 
                /*
                 *      We have too many packets in the backlog, we're
                 *      gated.  Don't send more packets until we have
                 *      a reply.
                 *
                 *      Note that we will send *these* packets.
                 */
                l->state = FR_LOAD_STATE_GATED;
                l->stats.blocked = true;
                count = 0;
                l->stats.skipped += l->count;
        }
 
        /*
         *      Stop after max_requests packets, if it's set.  Packets
         *      sent by a previous run of the generator don't count.
         */
        if (l->config->max_requests) {
                uint32_t sent = (uint32_t) (l->stats.sent - l->sent_base);
 
                if (sent >= l->config->max_requests) {
                        load_drain(l, now);
                        return;
                }
 
                if (count > (l->config->max_requests - sent)) count = l->config->max_requests - sent;
        }
 
        /*
         *      Skip timers if we're too busy.
         */
        l->next = fr_time_add(l->next, l->delta);
        if (fr_time_lt(l->next, now)) {
                while (fr_time_lt(fr_time_add(l->next, l->delta), now)) {
//                      l->stats.skipped += l->count;
                        l->next = fr_time_add(l->next, l->delta);
                }
        }
        delta = fr_time_sub(l->next, now);
 
        /*
         *      Set the timer for the next packet.
         */
        if (fr_timer_in(l, tl, &l->ev, delta, false, load_timer, l) < 0) {
                load_drain(l, now);
                return;
        }
 
        if (count) fr_load_generator_send(l, now, count);
}
 
 
/** Start the load generator.
 *
 */
int fr_load_generator_start(fr_load_t *l)
{
        uint32_t max;
 
        l->stats.start = fr_time();
        l->step_start = l->stats.start;
        l->step_end = fr_time_add(l->step_start, l->config->duration);
 
        l->state = FR_LOAD_STATE_SENDING;
        l->sent_base = l->stats.sent;
 
        l->pps = l->config->start_pps;
 
        /*
         *      Check for numerical overflow.  We later multiply pps*milliseconds, and we don't want overflow.
         */
        max = UINT32_MAX / l->config->milliseconds;
 
        if (l->pps > max) l->pps = max;
 
        l->stats.pps = l->pps;
        l->count = l->config->parallel;
 
        l->delta = fr_time_delta_div(fr_time_delta_from_sec(l->config->parallel), fr_time_delta_wrap(l->pps));
        l->next = fr_time_add(l->step_start, l->delta);
 
        load_timer(l->el->tl, l->step_start, l);
        return 0;
}
 
 
/** Stop the load generation through the simple expedient of deleting
 * the timer associated with it.
 *
 */
int fr_load_generator_stop(fr_load_t *l)
{
        if (!fr_timer_armed(l->ev)) return 0;
 
        FR_TIMER_DELETE_RETURN(&l->ev);
        return 0;
}
 
 
/** Tell the load generator that we have a reply to a packet we sent.
 *
 */
fr_load_reply_t fr_load_generator_have_reply(fr_load_t *l, fr_time_t request_time)
{
        fr_time_t now;
        fr_time_delta_t t;
 
        /*
         *      Note that the replies may come out of order with
         *      respect to the request.  So we can't use this reply
         *      for any kind of timing.
         */
        now = fr_time();
        t = fr_time_sub(now, request_time);
 
        l->stats.rttvar = RTTVAR(l->stats.rtt, l->stats.rttvar, t);
        l->stats.rtt = RTT(l->stats.rtt, t);
 
        l->stats.received++;
 
        /*
         *      t is in nanoseconds.
         */
        if (fr_time_delta_lt(t, fr_time_delta_wrap(1000))) {
               l->stats.times[0]++; /* < microseconds */
        } else if (fr_time_delta_lt(t, fr_time_delta_wrap(10000))) {
               l->stats.times[1]++; /* microseconds */
        } else if (fr_time_delta_lt(t, fr_time_delta_wrap(100000))) {
               l->stats.times[2]++; /* 10s of microseconds */
        } else if (fr_time_delta_lt(t, fr_time_delta_wrap(1000000))) {
               l->stats.times[3]++; /* 100s of microseconds */
        } else if (fr_time_delta_lt(t, fr_time_delta_wrap(10000000))) {
               l->stats.times[4]++; /* milliseconds */
        } else if (fr_time_delta_lt(t, fr_time_delta_wrap(100000000))) {
               l->stats.times[5]++; /* 10s of milliseconds */
        } else if (fr_time_delta_lt(t, fr_time_delta_wrap(NSEC))) {
               l->stats.times[6]++; /* 100s of milliseconds */
        } else {
               l->stats.times[7]++; /* seconds */
        }
 
        /*
         *      Still sending packets.  Rely on the timer to send more
         *      packets.
         */
        if (l->state == FR_LOAD_STATE_SENDING) return FR_LOAD_CONTINUE;
 
        /*
         *      The send code has decided that the backlog is too
         *      high.  New requests are blocked until replies come in.
         *      Since we have a reply, send another request.  Unless
         *      we've already sent max_requests packets, in which case
         *      the timer will notice and start draining.
         */
        if (l->state == FR_LOAD_STATE_GATED) {
                if ((l->stats.skipped > 0) &&
                    (!l->config->max_requests ||
                     ((uint32_t) (l->stats.sent - l->sent_base) < l->config->max_requests))) {
                        l->stats.skipped--;
                        fr_load_generator_send(l, now, 1);
                }
                return FR_LOAD_CONTINUE;
        }
 
        /*
         *      We're still sending or gated, tell the caller to
         *      continue.
         */
        if (l->state != FR_LOAD_STATE_DRAINING) {
                return FR_LOAD_CONTINUE;
        }
        /*
         *      Not yet received all replies.  Wait until we have all
         *      replies.
         */
        if (l->stats.received < l->stats.sent) return FR_LOAD_CONTINUE;
 
        l->stats.end = now;
        return FR_LOAD_DONE;
}
 
/** Print load generator statistics in CVS format.
 *
 */
size_t fr_load_generator_stats_sprint(fr_load_t *l, fr_time_t now, char *buffer, size_t buflen)
{
        double now_f, last_send_f;
 
        if (!l->header) {
                l->header = true;
                return snprintf(buffer, buflen, "\"time\",\"last_packet\",\"rtt\",\"rttvar\",\"pps\",\"pps_accepted\",\"sent\",\"received\",\"backlog\",\"max_backlog\",\"<usec\",\"us\",\"10us\",\"100us\",\"ms\",\"10ms\",\"100ms\",\"s\",\"blocked\"\n");
        }
 
 
        now_f = fr_time_delta_unwrap(fr_time_sub(now, l->stats.start)) / (double)NSEC;
 
        last_send_f = fr_time_delta_unwrap(fr_time_sub(l->stats.last_send, l->stats.start)) / (double)NSEC;
 
        /*
         *      Track packets/s.  Since times are in nanoseconds, we
         *      have to scale the counters up by NSEC.  And since NSEC
         *      is 1B, the calculations have to be done via 64-bit
         *      numbers, and then converted to a final 32-bit counter.
         */
        if (fr_time_gt(now, l->step_start)) {
                l->stats.pps_accepted =
                        fr_time_delta_unwrap(
                                fr_time_delta_div(fr_time_delta_from_sec(l->stats.received - l->step_received),
                                                  fr_time_sub(now, l->step_start))
                        );
        }
 
        return snprintf(buffer, buflen,
                        "%f,%f,"
                        "%" PRIu64 ",%" PRIu64 ","
                        "%d,%d,"
                        "%d,%d,"
                        "%d,%d,"
                        "%d,%d,%d,%d,%d,%d,%d,%d,"
                        "%d\n",
                        now_f, last_send_f,
                        fr_time_delta_unwrap(l->stats.rtt), fr_time_delta_unwrap(l->stats.rttvar),
                        l->stats.pps, l->stats.pps_accepted,
                        l->stats.sent, l->stats.received,
                        l->stats.backlog, l->stats.max_backlog,
                        l->stats.times[0], l->stats.times[1], l->stats.times[2], l->stats.times[3],
                        l->stats.times[4], l->stats.times[5], l->stats.times[6], l->stats.times[7],
                        l->stats.blocked);
}
 
fr_load_stats_t const * fr_load_generator_stats(fr_load_t const *l)
{
        return &l->stats;
}