Move rrdbot code into right project.

1 files changed, 424 insertions, 0 deletions

diff --git a/daemon/common/server-mainloop.c b/daemon/common/server-mainloop.c
new file mode 100644
index 0000000..07f02b2
--- /dev/null
+++ b/daemon/common/server-mainloop.c
@@ -0,0 +1,424 @@
+
+#include "usuals.h"
+#include <errno.h>
+#include <sys/time.h>
+
+#include "server-mainloop.h"
+
+typedef struct _socket_callback
+{
+    int fd;
+    server_socket_callback callback;
+    void* arg;
+
+    struct _socket_callback* next;
+}
+socket_callback;
+
+typedef struct _timer_callback
+{
+    struct timeval at;
+    struct timeval interval;
+    server_timer_callback callback;
+    void* arg;
+
+    struct _timer_callback* next;
+}
+timer_callback;
+
+typedef struct _server_context
+{
+    int stopped;
+    fd_set read_fds;
+    fd_set write_fds;
+    int max_fd;
+    socket_callback* callbacks;
+    timer_callback* timers;
+}
+server_context;
+
+/* Global context */
+static server_context ctx;
+
+static void
+timeval_add(struct timeval* t1, struct timeval* t2)
+{
+    ASSERT(t1->tv_usec < 1000000);
+    ASSERT(t2->tv_usec < 1000000);
+
+    t1->tv_sec += t2->tv_sec;
+    t1->tv_usec += t2->tv_usec;
+    if(t1->tv_usec >= 1000000)
+    {
+        t1->tv_usec -= 1000000;
+        t1->tv_sec += 1;
+    }
+}
+
+static void
+timeval_subtract(struct timeval* t1, struct timeval* t2)
+{
+    ASSERT(t1->tv_usec < 1000000);
+    ASSERT(t2->tv_usec < 1000000);
+
+    t1->tv_sec -= t2->tv_sec;
+    if(t1->tv_usec < t2->tv_usec)
+    {
+        t1->tv_usec += 1000000;
+        t1->tv_sec -= 1;
+    }
+    t1->tv_usec -= t2->tv_usec;
+}
+
+static int
+timeval_compare(struct timeval* t1, struct timeval* t2)
+{
+    ASSERT(t1->tv_usec < 1000000);
+    ASSERT(t2->tv_usec < 1000000);
+
+    if(t1->tv_sec > t2->tv_sec)
+        return 1;
+    else if(t1->tv_sec < t2->tv_sec)
+        return -1;
+    else
+    {
+        if(t1->tv_usec > t2->tv_usec)
+            return 1;
+        else if(t1->tv_usec < t2->tv_usec)
+            return -1;
+        else
+            return 0;
+    }
+}
+
+#define timeval_empty(tv) \
+    ((tv)->tv_sec == 0 && (tv)->tv_usec == 0)
+
+#define timeval_to_ms(tv) \
+    ((((uint64_t)(tv).tv_sec) * 1000L) + (((uint64_t)(tv).tv_usec) / 1000L))
+
+static int
+timeval_dump(struct timeval* tv)
+{
+    fprintf(stderr, "{ %d:%d }", tv->tv_sec, tv->tv_usec / 1000);
+}
+
+static int
+add_timer(int ms, int oneshot, server_timer_callback callback, void* arg)
+{
+    struct timeval interval;
+    timer_callback* cb;
+    int i;
+
+    ASSERT(ms > 0);
+    ASSERT(callback != NULL);
+
+    interval.tv_sec = ms / 1000;
+    interval.tv_usec = (ms % 1000) * 1000; /* into micro seconds */
+
+    cb = (timer_callback*)calloc(1, sizeof(*cb));
+    if(!cb)
+    {
+        errno = ENOMEM;
+        return -1;
+    }
+
+    if(gettimeofday(&(cb->at), NULL) == -1)
+    {
+        free(cb);
+        return -1;
+    }
+
+    timeval_add(&(cb->at), &interval);
+
+    if (oneshot)
+        memset(&(cb->interval), 0, sizeof(cb->interval));
+    else
+        memcpy(&(cb->interval), &interval, sizeof(cb->interval));
+
+    cb->callback = callback;
+    cb->arg = arg;
+
+    cb->next = ctx.timers;
+    ctx.timers = cb;
+
+    return 0;
+}
+
+static timer_callback*
+remove_timer(timer_callback* timcb)
+{
+    timer_callback* cb;
+    timer_callback* next;
+    int i;
+
+    if(!ctx.timers)
+        return;
+
+    /* First in list */;
+    if(ctx.timers == timcb)
+    {
+        cb = ctx.timers;
+        ctx.timers = ctx.timers->next;
+        free(cb);
+        return ctx.timers;
+    }
+
+    /* One ahead processing of rest */
+    for(cb = ctx.timers; cb->next; cb = cb->next)
+    {
+        if(cb->next == timcb)
+        {
+            next = cb->next->next;
+            free(cb->next);
+            cb->next = next;
+            return cb->next;
+        }
+    }
+}
+
+void
+server_init()
+{
+    memset(&ctx, 0, sizeof (ctx));
+    FD_ZERO(&ctx.read_fds);
+    FD_ZERO(&ctx.write_fds);
+
+    ctx.max_fd = -1;
+    ctx.stopped = 1;
+    ctx.callbacks = NULL;
+    ctx.timers = NULL;
+}
+
+void
+server_uninit()
+{
+    timer_callback* timcb;
+    timer_callback* timn;
+    socket_callback* sockcb;
+    socket_callback* sockn;
+
+    for(timcb = ctx.timers; timcb; timcb = timn)
+    {
+        timn = timcb->next;
+        free(timcb);
+    }
+
+    ctx.timers = NULL;
+
+    for(sockcb = ctx.callbacks; sockcb; sockcb = sockn)
+    {
+        sockn = sockcb->next;
+        free(sockcb);
+    }
+
+    ctx.timers = NULL;
+}
+
+uint64_t
+server_get_time()
+{
+    struct timeval tv;
+    if(gettimeofday(&tv, NULL) == -1)
+        return 0L;
+    return timeval_to_ms(tv);
+}
+
+int
+server_run()
+{
+    struct timeval* timeout;
+    struct timeval tv, current;
+    timer_callback* timcb;
+    socket_callback* sockcb;
+    fd_set rfds, wfds;
+    int r, i;
+
+    /* No watches have been set */
+    ASSERT(ctx.max_fd > -1);
+
+    ctx.stopped = 0;
+
+    while(!ctx.stopped)
+    {
+        /* Watch for the various fds */
+        memcpy(&rfds, &ctx.read_fds, sizeof(rfds));
+        memcpy(&wfds, &ctx.write_fds, sizeof(wfds));
+
+        /* Prepare for timers */
+        timeout = NULL;
+        if(gettimeofday(&current, NULL) == -1)
+            return -1;
+
+        /* Cycle through timers */
+        for(timcb = ctx.timers; timcb; )
+        {
+            ASSERT(timcb->callback);
+
+            /* Call any timers that have already passed */
+            if(timeval_compare(&current, &timcb->at) >= 0)
+            {
+                /* Convert to milliseconds, and make the call */
+                r = (timcb->callback)(timeval_to_ms(current), timcb->arg);
+
+                /* Reset timer if so desired */
+                if (r == 1 && !timeval_empty(&timcb->interval))
+                {
+                    timeval_add(&timcb->at, &timcb->interval);
+
+                    /* If the time has already passed, just use current time */
+                    if(timeval_compare(&(timcb->at), &current) <= 0)
+                        memcpy(&(timcb->at), &current, sizeof(timcb->at));
+                }
+
+                /* Otherwise remove it. Either one shot, or returned 0 */
+                else
+                {
+                    timcb = remove_timer(timcb);
+                    continue;
+                }
+            }
+
+            /* Get soonest timer */
+            if (!timeout || timeval_compare(timeout, &timcb->at) < 0)
+                timeout = &timcb->at;
+
+            timcb = timcb->next;
+        }
+
+        /* Convert to an offset */
+        if(timeout)
+        {
+            memcpy(&tv, timeout, sizeof(tv));
+            timeout = &tv;
+            timeval_subtract(timeout, &current);
+        }
+
+        /* fprintf(stderr, "selecting with timeout: ");
+           timeval_dump(timeout);
+           fprintf(stderr, "\n"); */
+
+        r = select(ctx.max_fd, &rfds, &wfds, NULL, timeout);
+        if (r < 0)
+        {
+            /* Interrupted so try again, and possibly exit */
+            if (errno == EINTR)
+                continue;
+
+            /* Programmer errors */
+            ASSERT (errno != EBADF);
+            ASSERT (errno != EINVAL);
+            return r;
+        }
+
+        /* Timeout, just jump to timeout processing */
+        if(r == 0)
+            continue;
+
+        for(sockcb = ctx.callbacks; sockcb; sockcb = sockcb->next)
+        {
+            ASSERT(sockcb->fd != -1);
+
+            /* Call any that are set */
+            if (FD_ISSET(sockcb->fd, &rfds))
+                (sockcb->callback)(sockcb->fd, SERVER_READ, sockcb->arg);
+            if (FD_ISSET(sockcb->fd, &wfds))
+                (sockcb->callback)(sockcb->fd, SERVER_WRITE, sockcb->arg);
+        }
+    }
+
+    return 0;
+}
+
+void
+server_stop()
+{
+    ctx.stopped = 1;
+}
+
+int
+server_stopped()
+{
+    return ctx.stopped;
+}
+
+int
+server_watch(int fd, int type, server_socket_callback callback, void* arg)
+{
+    socket_callback* cb;
+    int i;
+    ASSERT(type != 0);
+    ASSERT(fd != -1);
+    ASSERT(callback != NULL);
+
+    cb = (socket_callback*)calloc(sizeof(*cb), 1);
+    if(!cb)
+    {
+        errno = ENOMEM;
+        return -1;
+    }
+
+    cb->fd = fd;
+    cb->callback = callback;
+    cb->arg = arg;
+
+    cb->next = ctx.callbacks;
+    ctx.callbacks = cb;
+
+    if (type & SERVER_READ)
+        FD_SET(fd, &ctx.read_fds);
+    if (type & SERVER_WRITE)
+        FD_SET(fd, &ctx.write_fds);
+
+    if(fd >= ctx.max_fd)
+        ctx.max_fd = fd + 1;
+
+    return 0;
+}
+
+void
+server_unwatch(int fd)
+{
+    socket_callback* cb;
+    int i;
+
+    ASSERT(fd != -1);
+
+    FD_CLR(fd, &ctx.read_fds);
+    FD_CLR(fd, &ctx.write_fds);
+
+    if(!ctx.callbacks)
+        return;
+
+    /* First in list */;
+    if(ctx.callbacks->fd == fd)
+    {
+        cb = ctx.callbacks;
+        ctx.callbacks = cb->next;
+        free(cb);
+        return;
+    }
+
+    /* One ahead processing of rest */
+    for(cb = ctx.callbacks; cb->next; cb = cb->next)
+    {
+        if(cb->next->fd == fd)
+        {
+            cb->next = cb->next->next;
+            free(cb->next);
+            return;
+        }
+    }
+}
+
+int
+server_timer(int ms, server_timer_callback callback, void* arg)
+{
+    return add_timer(ms, 0, callback, arg);
+}
+
+int
+server_oneshot(int ms, server_timer_callback callback, void* arg)
+{
+    return add_timer(ms, 1, callback, arg);
+}