mq.c

Go to the documentation of this file.

#include <mq.h>
#include <string.h>
#include <tpa.h>
#include <flexarray.h>
#include <panic.h>
#include <errno.h>
#include <scheduler.h>
 
static const size_t MessageQueuePageItemsTarget = 16;
 
static tpa_t*   mqs;
static uint64_t next_mq = 1;
 
struct MessageQueuePage {
    size_t allocated;
 
    size_t bytes;
 
    size_t items;
 
    size_t pop_position;
 
    size_t push_position;
 
    struct MessageQueuePage* next;
};
 
struct MessageQueue {
    size_t max_items;
 
    size_t max_bytes;
 
    size_t bytes;
 
    size_t items;
 
    size_t average_message_size;
 
    allocator_t* alloc;
 
    struct MessageQueuePage* first_page;
 
    struct MessageQueuePage* last_page;
 
    flexarray_t notify_teardown;
};
 
void init_mq(allocator_t* alloc) {
    mqs = tpa_new(alloc, sizeof(struct MessageQueue), 4080, 0);
}
 
uint64_t mq_create(allocator_t* alloc) {
    if(!next_mq) {
        panic_message("Mutex namespace overflow!");
    }
 
    struct MessageQueue mq = {
        .max_items            = 0,
        .max_bytes            = 0,
        .bytes                = 0,
        .items                = 0,
        .average_message_size = 0,
 
        .alloc = alloc,
 
        .first_page = 0,
        .last_page  = 0,
        .notify_teardown = new_flexarray(sizeof(mq_notifier), 0, alloc),
    };
 
    tpa_set(mqs, next_mq, &mq);
 
    return next_mq++;
}
 
void mq_destroy(uint64_t mq) {
    struct MessageQueue* data = (struct MessageQueue*)tpa_get(mqs, mq);
 
    if(!data) {
        return;
    }
 
    size_t teardown_notifiers = flexarray_length(data->notify_teardown);
    for(size_t i = 0; i < teardown_notifiers; ++i) {
        mq_notifier notif = 0;
        flexarray_get(data->notify_teardown, &notif, i);
        if(notif) {
            notif(mq);
        }
    }
 
    struct MessageQueuePage* current = data->first_page;
    while(current) {
        struct MessageQueuePage* next = current->next;
        data->alloc->dealloc(data->alloc, current);
        current = next;
    }
 
    delete_flexarray(data->notify_teardown);
 
    tpa_set(mqs, mq, 0);
}
 
static void mq_alloc_page(struct MessageQueue* mq, size_t min_size) {
    size_t alloc_size = mq->average_message_size * MessageQueuePageItemsTarget;
 
    if(alloc_size < min_size) {
        alloc_size = min_size; // let's hope this size is uncommon and next page has average messages again
    }
 
    struct MessageQueuePage* page = (struct MessageQueuePage*)mq->alloc->alloc(mq->alloc, alloc_size + sizeof(struct MessageQueuePage));
    memset(page, 0, alloc_size + sizeof(struct MessageQueuePage));
 
    page->allocated = alloc_size;
 
    if(!mq->first_page) {
        mq->first_page = page;
    }
 
    if(mq->last_page) {
        mq->last_page->next = page;
    }
 
    mq->last_page = page;
}
 
uint64_t mq_push(uint64_t mq, struct Message* message) {
    struct MessageQueue* data = (struct MessageQueue*)tpa_get(mqs, mq);
 
    if(!data) {
        return ENOENT;
    }
 
    if((
         data->max_bytes &&
         data->bytes + message->size >= data->max_bytes
       ) ||
       (
         data->max_items &&
         data->items >= data->max_items
       )
    ) {
        return ENOMEM;
    }
 
    data->average_message_size += message->size;
    if(data->average_message_size != message->size) {
        data->average_message_size /= 2;
    }
 
    if(!data->last_page ||
        data->last_page->allocated - data->last_page->push_position < message->size) {
        mq_alloc_page(data, message->size);
    }
 
    void* message_pos = (void*)((uint64_t)data->last_page + data->last_page->push_position + sizeof(struct MessageQueuePage));
    memcpy(message_pos, message, message->size); // TODO: either validate size or make sure only the process crashes
 
    data->last_page->push_position += message->size;
    data->last_page->bytes         += message->size;
    ++data->last_page->items;
 
    data->bytes += message->size;
    ++data->items;
 
    union wait_data wd;
    wd.message_queue = mq;
    scheduler_waitable_done(wait_reason_message, wd, 1);
 
    return 0;
}
 
uint64_t mq_pop(uint64_t mq, struct Message* msg) {
    uint64_t error;
    if((error = mq_peek(mq, msg))) {
        return error;
    }
 
    struct MessageQueue* data = (struct MessageQueue*)tpa_get(mqs, mq);
 
    data->first_page->pop_position += msg->size;
    data->first_page->bytes        -= msg->size;
    --data->first_page->items;
 
    data->bytes -= msg->size;
    data->items--;
 
    if(!data->first_page->items) {
        if(data->first_page == data->last_page) {
            data->last_page = 0;
        }
 
        struct MessageQueuePage* next = data->first_page->next;
        data->alloc->dealloc(data->alloc, data->first_page);
        data->first_page = next;
    }
 
    return 0;
}
 
uint64_t mq_peek(uint64_t mq, struct Message* msg) {
    struct MessageQueue* data = (struct MessageQueue*)tpa_get(mqs, mq);
 
    if(!data) {
        return ENOENT;
    }
 
    if(!data->items) {
        return ENOMSG;
    }
 
    struct Message* msg_in_page = (struct Message*)((uint64_t)data->first_page + data->first_page->pop_position + sizeof(struct MessageQueuePage));
 
    if(msg_in_page->size <= msg->size) {
        memcpy(msg, msg_in_page, msg_in_page->size);
        return 0;
    }
    else {
        msg->size = msg_in_page->size;
        msg->type = MT_Invalid;
        return EMSGSIZE;
    }
}
 
uint64_t mq_notify_teardown(mq_id_t mq, mq_notifier notifier) {
    struct MessageQueue* data = (struct MessageQueue*)tpa_get(mqs, mq);
 
    if(!data) {
        return ENOENT;
    }
 
    size_t teardown_notifiers = flexarray_length(data->notify_teardown);
    for(size_t i = 0; i < teardown_notifiers; ++i) {
        mq_notifier existing = 0;
        flexarray_get(data->notify_teardown, &existing, i);
        if(existing == notifier) {
            return EEXIST;
        }
    }
 
    flexarray_append(data->notify_teardown, &notifier);
    return 0;
}