lipstick 💄🐷

// Based on https://doc.rust-lang.org/book/ch20-03-graceful-shutdown-and-cleanup.html
include![<stdio>];
include![<glib>];

/* MESSAGE */

struct Message {
    type_: u8,
    job: fn(gpointer),
    data: gpointer,
}

fn message_new(job: fn(gpointer), data: gpointer) -> &Message {
    let message: &Message = malloc(sizeof(Message));

    message.type_ = 0;
    message.job = job;
    message.data = data;

    message
}

fn message_new_terminate() -> &Message {
    let message: &Message = malloc(sizeof(Message));

    message.type_ = 1;

    message
}

fn message_destroy(message: &Message) {
    free(message);
}

/* WORKER */

fn worker_run(data: gpointer) -> gpointer {
    let queue: &GAsyncQueue = data as &GAsyncQueue;

    loop {
        let message: &Message = g_async_queue_pop(queue) as &Message;

        let type_: u8 = message.type_;
        let data: gpointer = message.data;
        let job: fn(gpointer) = message.job;

        message_destroy(message);

        match type_ {
            0 => {
                job(data);
            }
            _ => {
                break;
            }
        }
    }

    NULL
}

/* POOL */

struct WorkerPool {
    queue: &GAsyncQueue,
    workers: &GList,
}

fn worker_pool_new(size: usize) -> &WorkerPool {
    let pool: &WorkerPool = malloc(sizeof(WorkerPool));

    pool.queue = g_async_queue_new();

    for i in 0..size {
        let thread: &GThread = g_thread_new("worker", worker_run, pool.queue);

        pool.workers = g_list_append(pool.workers, thread as gpointer);
    }

    pool
}

fn worker_pool_execute(pool: &WorkerPool, job: fn(gpointer), data: gpointer) {
    let message: &Message = message_new(job, data);

    g_async_queue_push(pool.queue, message as gpointer);
}

fn worker_pool_destroy(pool: &WorkerPool) {
    let size: int = g_list_length(pool.workers);

    for i in 0..size {
        let message: &Message = message_new_terminate();

        g_async_queue_push(pool.queue, message);
    }

    let node: &GList = pool.workers;

    while node != NULL {
        g_thread_join((*node).data as &GThread);
        node = (*node).next;
    }

    g_list_free(pool.workers);
    g_async_queue_unref(pool.queue);
    free(pool);
}