thread.c

#define _POSIX_C_SOURCE 200809L

#include <common/common.h>
#include <lektor/thread.h>
#include <pthread.h>
#include <sys/types.h>
#include <stdlib.h>
#include <errno.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>

struct __args {
    void *(*start)(struct poller_thread_arg *);
    struct poller_thread_arg *arg;
};

void *
__start(void *args__)
{
    struct __args *args = (struct __args *) args__;
    void *(*start)(struct poller_thread_arg *) = args->start;
    struct poller_thread_arg *arg = args->arg;
    free(args__);
    return start(arg);
}

int
poller_new(struct poller_thread *th,  void *(*func)(struct poller_thread_arg *),
           void *args)
{
    pthread_mutex_t mtx1 = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t mtx2 = PTHREAD_MUTEX_INITIALIZER;
    struct poller_thread th_ = {
        .input_lock     = mtx1,
        .output_lock    = mtx2,
    };

    if (stack_new(&th_.input) || stack_new(&th_.output))
        goto out_of_memory;

    *th = th_;
    struct __args *__args = calloc(1, sizeof(struct __args));
    __args->start     = func;
    __args->arg       = args;
    __args->arg->self = th;

    if (pthread_create(&(th->th), NULL, __start, __args)) {
        LOG_ERROR("THREAD", "%s", "Failed to create a poller thread");
        return 1;
    }

    LOG_INFO("THREAD", "%s", "Create a new poller thread");
    return 0;

out_of_memory:
    LOG_ERROR("MEMORY", "%s", "Out of memory");
    stack_free(&th_.input);
    stack_free(&th_.output);
    return 1;
}

int
poller_join(struct poller_thread *th, void **ret)
{
    int sta = pthread_join(th->th, ret);

    if (sta)
        LOG_ERROR("THREAD", "%s", "Failed to join thread");

    stack_free(&th->input);
    stack_free(&th->output);
    LOG_WARN("THREAD", "%s", "Thread joined");

    memset(th, 0, sizeof(struct poller_thread));
    return sta;
}

int
th_append(struct stack *lst, pthread_mutex_t *lock, void *ptr)
{
    RETURN_IF(pthread_mutex_lock(lock), "Failed to lock", 1);
    int ret = stack_push(lst, ptr);
    RETURN_IF(pthread_mutex_unlock(lock), "Failed to lock", 1);
    return ret;
}

int
th_pop(struct stack *lst, pthread_mutex_t *lock, void **ptr)
{
    RETURN_IF(pthread_mutex_lock(lock), "Failed to lock", 1);
    int ret = stack_pop(lst, ptr);
    RETURN_IF(pthread_mutex_unlock(lock), "Failed to lock", 1);
    return ret;
}

int
th_head(struct stack *lst, pthread_mutex_t *lock, void **ptr)
{
    RETURN_IF(pthread_mutex_lock(lock), "Failed to lock", 1);
    int ret = stack_head(lst, ptr);
    RETURN_IF(pthread_mutex_unlock(lock), "Failed to lock", 1);
    return ret;
}

int
poller_append_input(struct poller_thread *th, void *ptr)
{
    return th_append(&th->input, &th->input_lock, ptr);
}

int
poller_append_output(struct poller_thread *th, void *ptr)
{
    return th_append(&th->output, &th->output_lock, ptr);
}

int
poller_pop_input(struct poller_thread *th, void **ptr)
{
    return th_pop(&th->input, &th->input_lock, ptr);
}


int
poller_pop_output(struct poller_thread *th, void **ptr)
{
    return th_pop(&th->output, &th->output_lock, ptr);
}

int
poller_head_input(struct poller_thread *th, void **ptr)
{
    return th_head(&th->input, &th->input_lock, ptr);
}

int
poller_head_output(struct poller_thread *th, void **ptr)
{
    return th_head(&th->output, &th->output_lock, ptr);
}