ghc-eventlog-loopback-0.1.0.0: cbits/writer.c

#include <string.h>
#include <Rts.h>

struct queue_entry {
    size_t entry_length;
    void* entry_contents;
    struct queue_entry* entry_next;
};

// Global state
static int loopback_cap = -1;
static HsStablePtr loopback_mvar = NULL;
static struct queue_entry * queue = NULL;
static Mutex loopback_mutex;

// Signal to the listening thread that there are chunks to read
void tryWakeup(void)
{
    if(loopback_mvar) {
      hs_try_putmvar(loopback_cap, loopback_mvar);
      loopback_mvar = NULL;
    }
}

void registerWakeup(int cap, HsStablePtr mvar)
{
    loopback_cap = cap;
    loopback_mvar = mvar;
    // immediately wake it up if the queue is not empty
    if (queue) {
        tryWakeup();
    }
}

void pushChunk(void* chunk, size_t chunk_size)
{
    ACQUIRE_LOCK(loopback_mutex);
    struct queue_entry* tail = queue;
    // This is O(N) #1 should fix this
    while(tail && tail->entry_next) {
      tail = tail->entry_next;
    }
    struct queue_entry* entry = malloc(sizeof(struct queue_entry));
    entry->entry_length = chunk_size;
    entry->entry_contents = chunk;
    entry->entry_next = NULL;
    if(tail) {
        // append to the end of the queue
        ASSERT(tail->entry_next == NULL);
        tail->entry_next = entry;
    } else {
        // queue is empty
        queue = entry;
    }
    RELEASE_LOCK(loopback_mutex);
}

void freeChunk(struct queue_entry* entry)
{
    if(entry){
        // entry->entry_contents will be freed on the Haskell side
        free(entry);
    }
}

struct queue_entry* popChunk()
{
    ACQUIRE_LOCK(loopback_mutex);
    struct queue_entry* entry = queue;
    if(entry) {
        queue = entry->entry_next;
    }
    RELEASE_LOCK(loopback_mutex);
    return entry;
}

// Initialise the loopback writer
void loopback_init(void)
{
    initMutex(&loopback_mutex);
}

bool loopback_write(void* eventlog, size_t eventlog_size)
{
    void* chunk = malloc(eventlog_size);
    memcpy(chunk, eventlog, eventlog_size);
    pushChunk(chunk, eventlog_size);
    tryWakeup();
    return true;
}

void loopback_flush(void)
{
}

void loopback_stop(void)
{
    loopback_cap = -1;
    loopback_mvar = NULL;
    // TODO: free everything on the queue?
    queue = NULL;
#if defined(THREADED_RTS)
    closeMutex(&loopback_mutex);
#endif
}

EventLogWriter loopback_writer = {
 .initEventLogWriter = loopback_init,
 .writeEventLog = loopback_write,
 .flushEventLog = loopback_flush,
 .stopEventLogWriter = loopback_stop
};


void startLoopbackWriter(void)
{
    endEventLogging();
    startEventLogging(&loopback_writer);
}

void stopLoopbackWriter(void)
{
    loopback_stop();
    endEventLogging();
}