aura.c

#include <aura/aura.h>
#include <aura/private.h>
#include <inttypes.h>


static void *aura_eventsys_get_autocreate(struct aura_node *node)
{
        struct aura_eventloop *loop = aura_eventloop_get_data(node);
        if (loop == NULL) {
                slog(3, SLOG_DEBUG, "aura: Auto-creating eventsystem for node");
                loop = aura_eventloop_create(node);
                if (!loop) {
                        slog(0, SLOG_ERROR, "aura: eventloop auto-creation failed");
                        aura_panic(node);
                }
                loop->autocreated = 1;
                aura_eventloop_set_data(node, loop);
        }
        return loop;
}

struct aura_node *aura_open(const char *name, const char *opts)
{
        struct aura_node *node = calloc(1, sizeof(*node));
        int ret = 0; 
        if (!node)
                return NULL;
        node->is_opening = true;

        node->poll_timeout = 250; /* 250 ms default */
        node->tr = aura_transport_lookup(name); 
        if (!node->tr) { 
                slog(0, SLOG_FATAL, "Invalid transport name: %s", name);
                goto err_free_node;
        }

        INIT_LIST_HEAD(&node->outbound_buffers);
        INIT_LIST_HEAD(&node->inbound_buffers);
        INIT_LIST_HEAD(&node->event_buffers);
        INIT_LIST_HEAD(&node->buffer_pool);
        node->gc_threshold = 10; /* This should be more than enough */

        node->status = AURA_STATUS_OFFLINE;

        /*  Eventsystem will be either lazy-initialized or created via 
         *  aura_eventloop_* functions 
         */

        if (node->tr->open)
                ret = node->tr->open(node, opts);

        if (ret != 0) 
                goto err_free_node;
        
        slog(6, SLOG_LIVE, "Created a node using transport: %s", name); 
        node->is_opening = false;
        return node;

err_free_node:
        slog(0, SLOG_FATAL, "Error opening transport: %s", name);
        free(node);
        return NULL;
}


static void cleanup_buffer_queue(struct list_head *q, bool destroy)
{
        int i = 0;

        struct list_head *pos, *tmp;
        list_for_each_safe(pos, tmp, q) {
                struct aura_buffer *b; 
                b = list_entry(pos, struct aura_buffer, qentry); 
                list_del(pos);
                if (!destroy) /* Just return it to the pool */
                        aura_buffer_release(b);
                else /* Nuke it, we're closing down */
                        aura_buffer_destroy(b);
                i++;
        }
        slog(6, SLOG_LIVE, "Cleaned up %d buffers", i);
}


void aura_close(struct aura_node *node)
{
        struct aura_eventloop *loop = aura_eventloop_get_data(node); 
        
        if (node->tr->close)
                node->tr->close(node);

        if (loop) { 
                if (loop->autocreated)
                        aura_eventloop_destroy(loop);
                else
                        aura_eventloop_del(node);
        }

        /* After transport shutdown we need to clean up 
           remaining buffers */
        cleanup_buffer_queue(&node->inbound_buffers, true);
        cleanup_buffer_queue(&node->outbound_buffers, true);
        cleanup_buffer_queue(&node->buffer_pool, true);
        aura_transport_release(node->tr);
        /* Check if we have an export table registered and nuke it */
        if (node->tbl)
                aura_etable_destroy(node->tbl);

        /* Free file descriptors */
        if (node->fds)
                free(node->fds);
        
        free(node);
        slog(6, SLOG_LIVE, "Transport closed");
}

/* This one is small, but tricky */
static void aura_handle_inbound(struct aura_node *node)
{
        while(1) {
                struct aura_buffer *buf;
                struct aura_object *o;  

                buf = aura_dequeue_buffer(&node->inbound_buffers); 
                if (!buf)
                        break;

                o = buf->object;
                node->current_object = o; 
                aura_buffer_rewind(buf);

                slog(4, SLOG_DEBUG, "Handling %s id %d (%s) sync_call_running=%d", 
                     object_is_method(o) ? "response" : "event", 
                     o->id, o->name, node->sync_call_running);

                if (object_is_method(o) && !o->pending) { 
                        slog(0, SLOG_WARN, "Dropping orphan call result %d (%s)", 
                             o->id, o->name);
                        aura_buffer_release(buf);
                } else if (o->calldonecb) { 
                        slog(4, SLOG_DEBUG, "Callback for method/event %d (%s)",
                             o->id, o->name);
                        o->calldonecb(node, AURA_CALL_COMPLETED, buf, o->arg);
                        aura_buffer_release(buf);
                } else if (object_is_method(o) && (node->sync_call_running)) { 
                        slog(4, SLOG_DEBUG, "Completing call for method %d (%s)",
                             o->id, o->name);
                        node->sync_call_result = AURA_CALL_COMPLETED;
                        node->sync_ret_buf = buf; 
                        o->pending--;
                        if (o->pending < 0)
                                BUG(node, "Internal BUG: pending evt count lesser than zero");
                } else {
                        /* This one is tricky. We have an event with no callback */
                        if (node->sync_event_max > 0) { /* Queue it up into event_queue if it's enabled */
                                /* If we have an overrun - drop the oldest event to free up space first*/
                                if (node->sync_event_max <= node->sync_event_count) {
                                        struct aura_buffer *todrop;
                                        const struct aura_object *dummy;
                                        int ret = aura_get_next_event(node, &dummy, &todrop);
                                        if (ret != 0)
                                                BUG(node, "Internal bug, no next event");
                                        aura_buffer_release(todrop);
                                }

                                /* Now just queue the next one */
                                aura_queue_buffer(&node->event_buffers, buf);
                                node->sync_event_count++;
                                slog(4, SLOG_DEBUG, "Queued event %d (%s) for sync readout", 
                                     o->id, o->name);
                        } else {
                                /* Last resort - try the catch-all event callback */
                                if (node->unhandled_evt_cb)
                                        node->unhandled_evt_cb(node, buf, node->unhandled_evt_arg);
                                else /* Or just drop it with a warning */
                                        slog(0, SLOG_WARN, "Dropping event %d (%s)",
                                             o->id, o->name);
                                aura_buffer_release(buf);
                        }
                }
        }

        node->current_object = NULL; 
}

const struct aura_object *aura_get_current_object(struct aura_node *node)
{
        /* Make some noise */
        if (!node->current_object) { 
                slog(0, SLOG_WARN, "Looks like you're calling aura_get_current_object() outside the callback");
                slog(0, SLOG_WARN, "Don't do that - read the docs!");
        }
        return node->current_object;
}

struct aura_eventloop *aura_eventloop_get_data(struct aura_node *node)
{
        return node->eventsys_data;
}



void aura_status_changed_cb(struct aura_node *node, 
                            void (*cb)(struct aura_node *node, int newstatus, void *arg),
                            void *arg)
{
        node->status_changed_arg = arg;
        node->status_changed_cb = cb;
}

void aura_fd_changed_cb(struct aura_node *node, 
                        void (*cb)(const struct aura_pollfds *fd, enum aura_fd_action act, void *arg),
                        void *arg)
{
        int i;
        const struct aura_pollfds *fds;
        int count = aura_get_pollfds(node, &fds);
        node->fd_changed_arg = arg;
        node->fd_changed_cb = cb;
        if (node->fd_changed_cb)
                for (i=0; i<count; i++) 
                        node->fd_changed_cb(&fds[i], AURA_FD_ADDED, node->fd_changed_arg);
}

void aura_etable_changed_cb(struct aura_node *node, 
                            void (*cb)(struct aura_node *node, 
                                       struct aura_export_table *old, 
                                       struct aura_export_table *new, 
                                       void *arg),
                            void *arg)
{
        node->etable_changed_arg = arg;
        node->etable_changed_cb = cb;
}

void aura_unhandled_evt_cb(struct aura_node *node, 
                           void (*cb)(struct aura_node *node, 
                                      struct aura_buffer *buf, 
                                      void *arg),
                           void *arg)
{
        node->unhandled_evt_cb = cb;
        node->unhandled_evt_arg = arg;
}

void aura_object_migration_failed_cb(struct aura_node *node, 
                                     void (*cb)(struct aura_node *node, 
                                                struct aura_object *failed, 
                                                void *arg),
                                     void *arg)
{
        node->object_migration_failed_cb  = cb;
        node->object_migration_failed_arg = arg;
}

int aura_core_start_call(struct aura_node *node, 
                    struct aura_object *o,
                    void (*calldonecb)(struct aura_node *dev, int status, struct aura_buffer *ret, void *arg),
                    void *arg,
                    struct aura_buffer *buf)
{
        struct aura_eventloop *loop = aura_eventsys_get_autocreate(node);
        int isfirst;

        if (!o)
                return -EBADSLT;

        if(node->status != AURA_STATUS_ONLINE) 
                return -ENOEXEC;
                
        if (o->pending) 
                return -EIO; 

        if (!loop)
                BUG(node, "Node has no assosiated event system. Fix your code!");

        isfirst = list_empty(&node->outbound_buffers);

        o->calldonecb = calldonecb; 
        o->arg = arg; 
        buf->object = o;
        o->pending++;

        aura_queue_buffer(&node->outbound_buffers, buf);
        slog(4, SLOG_DEBUG, "Queued call for id %d (%s), notifying node", o->id, o->name);

        if (isfirst) {
                slog(4, SLOG_DEBUG, "Notifying transport of queue status change");
                node->last_checked = 0;
                aura_eventloop_interrupt(loop);
        };

        return 0;
}

int aura_core_call(
        struct aura_node *node, 
        struct aura_object *o,
        struct aura_buffer **retbuf,
        struct aura_buffer *argbuf)
{
        int ret;
        struct aura_eventloop *loop = aura_eventsys_get_autocreate(node);
        
        if (node->sync_call_running) 
                BUG(node, "Internal bug: Synchronos call within a synchronos call");

        node->sync_call_running = true;
 
        if ((ret=aura_core_start_call(node, o, NULL, NULL, argbuf))) {
                node->sync_call_result = ret;
                goto bailout;
        }

        while (o->pending) {
                aura_handle_events(loop);
        }       

        slog(4, SLOG_DEBUG, "Call completed");
        *retbuf =  node->sync_ret_buf;

bailout:
        node->sync_call_running = false; 
        return node->sync_call_result;
}

int aura_start_call_raw(
        struct aura_node *node, 
        int id,
        void (*calldonecb)(struct aura_node *dev, int status, struct aura_buffer *ret, void *arg),
        void *arg,
        ...)
{
        va_list ap;
        struct aura_buffer *buf; 
        int ret; 

        struct aura_object *o = aura_etable_find_id(node->tbl, id);
        if (!o)
                return -EBADSLT;

        va_start(ap, arg);
        buf = aura_serialize(node, o->arg_fmt, o->arglen, ap);
        va_end(ap);

        if (!buf) 
                return -EIO;
        
        ret = aura_core_start_call(node, o, calldonecb, arg, buf);

        if (ret != 0)
                aura_buffer_release(buf);

        return ret;
}

int aura_set_event_callback_raw(
        struct aura_node *node,
        int id,
        void (*calldonecb)(struct aura_node *dev, int status, struct aura_buffer *ret, void *arg),
        void *arg)
{
        struct aura_object *o = aura_etable_find_id(node->tbl, id);
        if (!o)
                return -EBADSLT;

        if (!object_is_event(o))
                return -EBADF;

        o->calldonecb = calldonecb;
        o->arg = arg;
        return 0;
}

int aura_set_event_callback(
        struct aura_node *node,
        const char *event,
        void (*calldonecb)(struct aura_node *dev, int status, struct aura_buffer *ret, void *arg),
        void *arg)
{
        struct aura_object *o = aura_etable_find(node->tbl, event);
        if (!o)
                return -EBADSLT;

        if (!object_is_event(o))
                return -EBADF;

        o->calldonecb = calldonecb;
        o->arg = arg;
        return 0;
}

int aura_start_call(
        struct aura_node *node, 
        const char *name,
        void (*calldonecb)(struct aura_node *dev, int status, struct aura_buffer *ret, void *arg),
        void *arg,
        ...)
{
        struct aura_object *o; 
        va_list ap;
        struct aura_buffer *buf; 
        int ret; 

        o = aura_etable_find(node->tbl, name);
        if (!o) 
                return -ENOENT; 

        va_start(ap, arg);
        buf = aura_serialize(node, o->arg_fmt, o->arglen, ap);
        va_end(ap);
        if (!buf) 
                return -EIO;
        
        ret = aura_core_start_call(node, o, calldonecb, arg, buf);

        if (ret != 0)
                aura_buffer_release(buf);

        return ret;
}

void aura_wait_status(struct aura_node *node, int status)
{
        struct aura_eventloop *loop = aura_eventsys_get_autocreate(node);
        while (node->status != status)
                aura_handle_events(loop);
}


int aura_call_raw(
        struct aura_node *node, 
        int id,
        struct aura_buffer **retbuf,
        ...)
{
        va_list ap;
        struct aura_buffer *buf; 

        struct aura_object *o = aura_etable_find_id(node->tbl, id);

        if (node->sync_call_running) 
                BUG(node, "Internal bug: Synchronos call within a synchronos call");

        if (!o)
                return -EBADSLT;
        
        va_start(ap, retbuf);
        buf = aura_serialize(node, o->arg_fmt, o->arglen, ap);
        va_end(ap);

        if (!buf) {
                slog(2, SLOG_WARN, "Serialization failed");
                return -EIO;
        }

        return aura_core_call(node, o, retbuf, buf);
}

int aura_call(
        struct aura_node *node, 
        const char *name,
        struct aura_buffer **retbuf,
        ...)
{
        va_list ap;
        struct aura_buffer *buf; 
        struct aura_object *o = aura_etable_find(node->tbl, name);
        
        if (!o)
                return -EBADSLT;
        
        va_start(ap, retbuf);
        buf = aura_serialize(node, o->arg_fmt, o->arglen, ap);
        va_end(ap);

        if (!buf) {
                slog(2, SLOG_WARN, "Serialization failed");
                return -EIO;
        }
        
        return aura_core_call(node, o, retbuf, buf);
}


void aura_enable_sync_events(struct aura_node *node, int count)
{
        while(node->sync_event_max >= count) {
                const struct aura_object *o;
                struct aura_buffer *buf;
                int ret = aura_get_next_event(node, &o, &buf);
                if (ret!=0)
                        BUG(node, "Internal bug while resizing event queue (failed to drop some events)");
                aura_buffer_release(buf);
        }
        node->sync_event_max = count;
}

int aura_get_pending_events(struct aura_node *node)
{
        return node->sync_event_count;
}

int aura_get_next_event(struct aura_node *node, const struct aura_object ** obj, struct aura_buffer **retbuf)
{
        struct aura_eventloop *loop = aura_eventsys_get_autocreate(node);

        while (!node->sync_event_count) {
                aura_handle_events(loop);
        }

        *retbuf = aura_dequeue_buffer(&node->event_buffers);
        if (!(*retbuf))
                aura_panic(node);

        *obj = (const struct aura_object *)(*retbuf)->object;
        node->sync_event_count--;
        return 0;
}

void aura_call_fail(struct aura_node *node, struct aura_object *o)
{
        if (o->pending && o->calldonecb)
                o->calldonecb(node, AURA_CALL_TRANSPORT_FAIL, NULL, o->arg);
        if (o->pending)
                o->pending--;

        node->sync_call_result = AURA_CALL_TRANSPORT_FAIL;
        node->sync_ret_buf = NULL;
}

void aura_set_status(struct aura_node *node, int status)
{
        int oldstatus = node->status;
        node->status = status;

        if (oldstatus == status)
                return;

        if (node->is_opening)
                BUG(node, "Transport BUG: Do not call aura_set_status in open()");              
                
        if ((oldstatus == AURA_STATUS_OFFLINE) && (status == AURA_STATUS_ONLINE)) {
                /* Dump etable */
                int i;
                slog(2, SLOG_INFO, "Node %s is now going online", node->tr->name);
                slog(2, SLOG_INFO, "--- Dumping export table ---");
                for (i=0; i< node->tbl->next; i++) {
                        slog(2, SLOG_INFO, "%d. %s %s %s(%s )  [out %d bytes] | [in %d bytes] ",
                             node->tbl->objects[i].id,
                             object_is_method((&node->tbl->objects[i])) ? "METHOD" : "EVENT ",
                             node->tbl->objects[i].ret_pprinted,
                             node->tbl->objects[i].name,
                             node->tbl->objects[i].arg_pprinted,
                             node->tbl->objects[i].arglen,
                             node->tbl->objects[i].retlen);
                }
                slog(1, SLOG_INFO, "-------------8<-------------");
        }
        if ((oldstatus == AURA_STATUS_ONLINE) && (status == AURA_STATUS_OFFLINE)) {
                int i;

                slog(2, SLOG_INFO, "Node %s going offline, clearing outbound queue",
                     node->tr->name);
                cleanup_buffer_queue(&node->outbound_buffers, false);
                /* Handle any remaining inbound messages */
                aura_handle_inbound(node);
                /* Cancel any pending calls */
                for (i=0; i < node->tbl->next; i++) {
                        struct aura_object *o;
                        o=&node->tbl->objects[i];
                        if (o->pending && o->calldonecb)
                                o->calldonecb(node, AURA_CALL_TRANSPORT_FAIL, NULL, o->arg);
                        if (o->pending)
                                o->pending--;
                }
                /* If any of the synchronos calls are running - inform them */
                node->sync_call_result = AURA_CALL_TRANSPORT_FAIL;
                node->sync_ret_buf = NULL;
        }

        if (node->status_changed_cb)
                node->status_changed_cb(node, status, node->status_changed_arg);
}

void aura_set_node_endian(struct aura_node *node, enum aura_endianness en)
{
        if (aura_get_host_endianness() != en)
                node->need_endian_swap = true;
}
void aura_process_node_event(struct aura_node *node, const struct aura_pollfds *fd)
{
        if (fd && node->tr->loop)
                node->tr->loop(node, fd);

        uint64_t curtime = aura_platform_timestamp();

        if ((curtime - node->last_checked > node->poll_timeout) && node->tr->loop) {
                node->tr->loop(node, NULL);
                node->last_checked = curtime;
        }

        /* Now grab all we got from the inbound queue and fire the callbacks */
        aura_handle_inbound(node);
}

void aura_eventloop_set_data(struct aura_node *node, struct aura_eventloop *data)
{
        node->eventsys_data = data;
}