- Write pid file if filename is set

[l2tpns.git] / tbf.c

// L2TPNS: token bucket filters

char const *cvs_id_tbf = "$Id: tbf.c,v 1.4 2004/07/08 16:54:35 bodea Exp $";

#define _GNU_SOURCE

#include <string.h>
#include <unistd.h>
#include <sys/mman.h>

#include "l2tpns.h"
#include "util.h"
#include "tbf.h"

// Need a time interval.

#define TBF_MAX_QUEUE   2       // Maximum of 2 queued packet per
#define TBF_MAX_SIZE    3000    // Maxiumum queued packet size is 2048.

#define TBF_MAX_CREDIT  6000    // Maximum 6000 bytes of credit.
#define TBF_RATE        360     // 360 bytes per 1/10th of a second.

typedef struct {
        int             credit;
        int             lasttime;
        int             queued;
        int             oldest; // Position of packet in the ring buffer.
        sessionidt      sid;    // associated session ID.
        int             max_credit; // Maximum amount of credit available (burst size).
        int             rate;   // How many bytes of credit per second we get? (sustained rate)
        void            (*send)(sessionidt s, u8 *, int);       // Routine to actually send out the data.
        int             prev;   // Timer chain position.
        int             next;   // Timer chain position.

        u32     b_queued;       // Total bytes sent through this TBF
        u32     b_sent;         // Total bytes sucessfully made it to the network.
        u32     p_queued;       // ditto packets.
        u32     p_sent;         // ditto packets.
        u32     b_dropped;      // Total bytes dropped.
        u32     p_dropped;      // Total packets dropped.
        u32     p_delayed;      // Total packets not sent immediately.

        int             sizes[TBF_MAX_QUEUE];
        char            packets[TBF_MAX_QUEUE][TBF_MAX_SIZE];
} tbft;


static tbft *filter_list = NULL;
static int filter_list_size = 0;

static int timer_chain = -1;    // Head of timer chain.

static void tbf_run_queue(int tbf_id);

void init_tbf(void)
{
        filter_list = mmap(NULL, sizeof(*filter_list) * MAXTBFS, PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, 0, 0);
        if (!filter_list)
                return;

        filter_list_size = MAXTBFS;
        filter_list[0].sid = -1;        // Reserved.
}
//
// Put a TBF on the timer list.
// This is a doubly linked list..
// We put ourselves on the tail of the list.
//
static void add_to_timer(int id)
{
        if (!filter_list)
                return;

        if (timer_chain == -1) {
                filter_list[id].next = filter_list[id].prev = id;
                timer_chain = id;
                return;
        }

        filter_list[id].next = timer_chain;
        filter_list[id].prev = filter_list[timer_chain].prev;
        filter_list[filter_list[timer_chain].prev].next = id;
        filter_list[timer_chain].prev = id;
}

//
// Remove a TBF from the timer list.
// This is a doubly linked list.
static void del_from_timer(int id)
{
        if (!filter_list)
                return;

        if (filter_list[id].next == id) {       // Last element in chain?
                if (timer_chain != id) { // WTF?
                        log(0,0,0,0, "Removed a singleton element from TBF, but tc didn't point to it!\n");
                } else
                        timer_chain = -1;
                filter_list[id].next = filter_list[id].prev = 0;
                return;
        }

        filter_list[filter_list[id].next].prev = filter_list[id].prev;
        filter_list[filter_list[id].prev].next = filter_list[id].next;
        if (timer_chain == id)
                timer_chain = filter_list[id].next;

        filter_list[id].next = filter_list[id].prev = 0;        // Mark as off the timer chain.
}

//
// Free a token bucket filter structure for re-use.
//

int free_tbf(int tid)
{
        if (tid < 1)    // Make sure we don't free id # 0
                return -1;

        if (!filter_list)       // WTF?
                return -1;

        if (filter_list[tid].next)
                del_from_timer(tid);
        filter_list[tid].sid = 0;

        return 0;       // Done!
}

//
// Allocate a new token bucket filter.
//
int new_tbf(int sid, int max_credit, int rate, void (*f)(sessionidt, u8 *, int))
{
        int i;
        static int p = 0;

        log(3,0,0,0, "Allocating new TBF (sess %d, rate %d, helper %p)\n", sid, rate, f);

        if (!filter_list)
                return 0;       // Couldn't alloc memory!

//    again:

        for (i = 0 ; i < filter_list_size ; ++i, p = (p+1)%filter_list_size ) {
                if (filter_list[p].sid)
                        continue;

                memset((void*) &filter_list[p], 0, sizeof(filter_list[p]) ); // Clear counters and data.
                filter_list[p].sid = sid;
                filter_list[p].credit = max_credit;
                filter_list[p].queued = 0;
                filter_list[p].max_credit = max_credit;
                filter_list[p].rate = rate;
                filter_list[p].oldest = 0;
                filter_list[p].send = f;
                return p;
        }

#if 0
        // All allocated filters are used! Increase the size of the allocated
        // filters.

        {
                int new_size = filter_list_size * 2;
                tbft *new = mremap(filter_list, filter_list_size * sizeof(*new), new_size * sizeof(*new), MREMAP_MAYMOVE);

                if (new == MAP_FAILED)
                {
                        log(0,0,0,0, "Ran out of token bucket filters and mremap failed!  Sess %d will be un-throttled\n", sid);
                        return 0;
                }

                i = filter_list_size;
                filter_list_size = new_size;
                filter_list = new;
        }

        for (; i < filter_list_size; ++i)
                filter_list[i].sid = 0;

        goto again;
#else
        log(0,0,0,0, "Ran out of token bucket filters!  Sess %d will be un-throttled\n", sid);
        return 0;
#endif
}

//
// Sanity check all the TBF records. This is
// typically done when we become a master..
//
void fsck_tbfs(void)
{
        int i , sid;

        if (!filter_list)
                return;

        for (i = 1; i < filter_list_size; ++i) {
                if (!filter_list[i].sid)        // Is it used??
                        continue;

                sid = filter_list[i].sid;
                if (i != session[sid].tbf_in &&
                        i != session[sid].tbf_out) { // Ooops.

                        free_tbf(i);            // Mark it as free...
                }
        }

        for (i = 0; i < config->cluster_highest_sessionid ; ++i) {
                if (session[i].tbf_in && filter_list[session[i].tbf_in].sid != i) {
                        filter_list[session[i].tbf_in].sid = i; // Ouch!? FIXME. What to do here?
                }
                if (session[i].tbf_out && filter_list[session[i].tbf_out].sid != i) {
                        filter_list[session[i].tbf_out].sid = i; // Ouch!? FIXME. What to do here?
                }
        }
}


//
// Run a packet through a token bucket filter.
// If we can send it right away, we do. Else we
// try and queue it to send later. Else we drop it.
//
int tbf_queue_packet(int tbf_id, char * data, int size)
{
        int i;
        tbft * f;

        if (!filter_list)
                return -1;

        if (tbf_id > filter_list_size || tbf_id < 1) {  // Out of range ID??
                // Very bad. Just drop it.
                return -1;
        }

        f = &filter_list[tbf_id];

        if (!f->sid)            // Is this a real structure??
                return -1;

        tbf_run_queue(tbf_id);  // Caculate credit and send any queued packets if possible..

        f->b_queued += size;
        f->p_queued ++;

        if (!f->queued && f->credit > size) { // If the queue is empty, and we have
                                // enough credit, just send it now.
                f->credit -= size;
                if (f->send) {
                        f->send(f->sid, data, size);
                        f->b_sent += size;
                        f->p_sent ++;
                } else {
                        f->b_dropped += size;
                        f->p_dropped ++;
                }
                return size;
        }

                // Not enough credit. Can we have room in the queue?
        if (f->queued >= TBF_MAX_QUEUE) {
                f->p_dropped ++;
                f->b_dropped += size;
                return -1;      // No, just drop it.
        }

                // Is it too big to fit into a queue slot?
        if (size >= TBF_MAX_SIZE) {
                f->p_dropped ++;
                f->b_dropped += size;
                return -1;      // Yes, just drop it.
        }

                // Ok. We have a slot, and it's big enough to
                // contain the packet, so queue the packet!
        i = ( f->oldest + f->queued ) % TBF_MAX_QUEUE;
        memcpy(f->packets[i], data, size);

        f->sizes[i] = size;
        f->queued ++;
        f->p_delayed ++;

        if (!f->next)   // Are we off the timer chain?
                add_to_timer(tbf_id);   // Put ourselves on the timer chain.

        return 0;       // All done.
}

//
// Send queued packets from the filter if possible.
// (We're normally only called if this is possible.. )
static void tbf_run_queue(int tbf_id)
{
        tbft * f;

        if (!filter_list)
                return;

        f = &filter_list[tbf_id];

                // Calculate available credit...
        f->credit += (TIME - f->lasttime) * f->rate / 10; // current time is 1/10th of a second.
        if (f->credit > f->max_credit)
                f->credit = f->max_credit;
        f->lasttime = TIME;

        while (f->queued > 0 && f->credit >= f->sizes[f->oldest]) { // While we have enough credit..

                if (f->send) {
                        f->send(f->sid, f->packets[f->oldest], f->sizes[f->oldest]);
                        f->b_sent += f->sizes[f->oldest];
                        f->p_sent ++;
                } else {
                        f->b_dropped += f->sizes[f->oldest];
                        f->p_dropped ++;
                }

                f->credit -= f->sizes[f->oldest];

                f->oldest = (f->oldest + 1 ) % TBF_MAX_QUEUE;
                f->queued--;    // One less queued packet..
        }

        if (f->queued)  // Still more to do. Hang around on the timer list.
                return;

        if (f->next)    // Are we on the timer list??
                del_from_timer(tbf_id); // Nothing more to do. Get off the timer list.
}

//
// Periodically walk the timer list..
//
int tbf_run_timer(void)
{
        int i = timer_chain;
        int count = filter_list_size + 1;       // Safety check.
        int last = -1;
        int tbf_id; // structure being processed.

        if (timer_chain < 0)
                return 0;       // Nothing to do...

        if (!filter_list)       // No structures built yet.
                return 0;

        last = filter_list[i].prev; // last element to process.

        do {
                tbf_id = i;
                i = filter_list[i].next;        // Get the next in the queue.

                tbf_run_queue(tbf_id);  // Run the timer queue..
        } while ( timer_chain > 0 && i && tbf_id != last && --count > 0);


#if 0   // Debugging.
        for (i = 0; i < filter_list_size; ++i) {
                if (!filter_list[i].next)
                        continue;
                if (filter_list[i].lasttime == TIME)    // Did we just run it?
                        continue;

                log(1,0,0,0, "Missed tbf %d! Not on the timer chain?(n %d, p %d, tc %d)\n", i,
                        filter_list[i].next, filter_list[i].prev, timer_chain);
                tbf_run_queue(i);
        }
#endif

        return 1;
}

int cmd_show_tbf(struct cli_def *cli, char *command, char **argv, int argc)
{
        int i;
        int count = 0;

        if (CLI_HELP_REQUESTED)
                return CLI_HELP_NO_ARGS;

        if (!config->cluster_iam_master) {
                cli_print(cli, "Can't do this on a slave.  Do it on %s", inet_toa(config->cluster_master_address));
                return CLI_OK;
        }

        if (!filter_list)
                return CLI_OK;

        cli_print(cli,"%6s %5s %5s %6s %6s | %7s %7s %8s %8s %8s %8s", "TBF#", "Sid", "Rate", "Credit", "Queued",
                "ByteIn","PackIn","ByteSent","PackSent", "PackDrop", "PackDelay");

        for (i = 1; i < filter_list_size; ++i) {
                if (!filter_list[i].sid) // Is it used?
                        continue;       // No.

                cli_print(cli, "%5d%1s %5d %5d %6d %6d | %7d %7d %8d %8d %8d %8d",
                        i, (filter_list[i].next ? "*" : " "),
                        filter_list[i].sid,
                        filter_list[i].rate * 8,
                        filter_list[i].credit,
                        filter_list[i].queued,

                        filter_list[i].b_queued,
                        filter_list[i].p_queued,
                        filter_list[i].b_sent,
                        filter_list[i].p_sent,
                        filter_list[i].p_dropped,
                        filter_list[i].p_delayed);
                ++count;
        }
        cli_print(cli, "%d tbf entries used, %d total", count, filter_list_size);
        return CLI_OK;
}