c-relay-pg/src/thread_pool.c

#define _GNU_SOURCE

#include "thread_pool.h"
#include "debug.h"

#include <pthread.h>
#include <stdlib.h>
#include <string.h>

typedef struct thread_pool_job_node {
    uint64_t job_id;
    thread_pool_job_t job;
    struct thread_pool_job_node* next;
} thread_pool_job_node_t;

typedef struct {
    thread_pool_job_node_t* head;
    thread_pool_job_node_t* tail;
    int size;
    int max_size;
    pthread_mutex_t mutex;
    pthread_cond_t cond;
} thread_pool_queue_t;

typedef struct {
    int running;
    uint64_t next_job_id;

    int reader_count;
    pthread_t* readers;
    pthread_t writer;

    thread_pool_queue_t read_q;
    thread_pool_queue_t write_q;

    thread_pool_wake_loop_cb wake_loop_cb;
    void* wake_loop_ctx;

    pthread_mutex_t state_mutex;
} thread_pool_state_t;

static thread_pool_state_t g_pool = {0};

static void queue_init(thread_pool_queue_t* q, int max_size) {
    memset(q, 0, sizeof(*q));
    q->max_size = (max_size > 0) ? max_size : 4096;
    pthread_mutex_init(&q->mutex, NULL);
    pthread_cond_init(&q->cond, NULL);
}

static void queue_destroy(thread_pool_queue_t* q) {
    pthread_mutex_lock(&q->mutex);
    thread_pool_job_node_t* cur = q->head;
    while (cur) {
        thread_pool_job_node_t* next = cur->next;
        if (cur->job.payload_free && cur->job.payload) {
            cur->job.payload_free(cur->job.payload);
        }
        free(cur);
        cur = next;
    }
    q->head = q->tail = NULL;
    q->size = 0;
    pthread_mutex_unlock(&q->mutex);

    pthread_mutex_destroy(&q->mutex);
    pthread_cond_destroy(&q->cond);
}

static int queue_push(thread_pool_queue_t* q, thread_pool_job_node_t* node) {
    pthread_mutex_lock(&q->mutex);
    if (q->size >= q->max_size) {
        pthread_mutex_unlock(&q->mutex);
        return 0;
    }

    node->next = NULL;
    if (!q->tail) {
        q->head = q->tail = node;
    } else {
        q->tail->next = node;
        q->tail = node;
    }
    q->size++;
    pthread_cond_signal(&q->cond);
    pthread_mutex_unlock(&q->mutex);
    return 1;
}

static thread_pool_job_node_t* queue_pop(thread_pool_queue_t* q) {
    pthread_mutex_lock(&q->mutex);
    while (g_pool.running && q->size == 0) {
        pthread_cond_wait(&q->cond, &q->mutex);
    }

    if (!g_pool.running && q->size == 0) {
        pthread_mutex_unlock(&q->mutex);
        return NULL;
    }

    thread_pool_job_node_t* node = q->head;
    q->head = node->next;
    if (!q->head) q->tail = NULL;
    q->size--;
    pthread_mutex_unlock(&q->mutex);
    return node;
}

static void wake_event_loop(void) {
    if (g_pool.wake_loop_cb) {
        g_pool.wake_loop_cb(g_pool.wake_loop_ctx);
    }
}

static void complete_job_with_status(thread_pool_job_node_t* node, thread_pool_status_t status, const char* message) {
    thread_pool_result_t result;
    memset(&result, 0, sizeof(result));
    result.job_id = node->job_id;
    result.type = node->job.type;
    result.status = status;
    result.session = node->job.session;
    result.message = message;

    if (node->job.result_cb) {
        node->job.result_cb(&result, node->job.result_cb_ctx);
    }
    wake_event_loop();

    if (node->job.payload_free && node->job.payload) {
        node->job.payload_free(node->job.payload);
    }
    free(node);
}

static void* reader_worker_main(void* arg) {
    (void)arg;
    while (g_pool.running) {
        thread_pool_job_node_t* node = queue_pop(&g_pool.read_q);
        if (!node) break;

        // Scaffold only: execution wiring is intentionally deferred.
        complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED,
                                 "Read worker scaffold active; execution not wired yet");
    }
    return NULL;
}

static void* writer_worker_main(void* arg) {
    (void)arg;
    while (g_pool.running) {
        thread_pool_job_node_t* node = queue_pop(&g_pool.write_q);
        if (!node) break;

        // Scaffold only: execution wiring is intentionally deferred.
        complete_job_with_status(node, THREAD_POOL_STATUS_NOT_IMPLEMENTED,
                                 "Write worker scaffold active; execution not wired yet");
    }
    return NULL;
}

int thread_pool_init(const thread_pool_config_t* config) {
    if (!config) return -1;

    pthread_mutex_lock(&g_pool.state_mutex);
    if (g_pool.running) {
        pthread_mutex_unlock(&g_pool.state_mutex);
        return 0;
    }

    g_pool.reader_count = (config->reader_threads > 0) ? config->reader_threads : 4;
    g_pool.readers = calloc((size_t)g_pool.reader_count, sizeof(pthread_t));
    if (!g_pool.readers) {
        pthread_mutex_unlock(&g_pool.state_mutex);
        return -1;
    }

    queue_init(&g_pool.read_q, config->max_queue_depth);
    queue_init(&g_pool.write_q, config->max_queue_depth);

    g_pool.next_job_id = 1;
    g_pool.wake_loop_cb = config->wake_loop_cb;
    g_pool.wake_loop_ctx = config->wake_loop_ctx;
    g_pool.running = 1;

    for (int i = 0; i < g_pool.reader_count; i++) {
        if (pthread_create(&g_pool.readers[i], NULL, reader_worker_main, NULL) != 0) {
            g_pool.running = 0;
            pthread_cond_broadcast(&g_pool.read_q.cond);
            pthread_cond_broadcast(&g_pool.write_q.cond);
            pthread_mutex_unlock(&g_pool.state_mutex);
            return -1;
        }
    }

    if (pthread_create(&g_pool.writer, NULL, writer_worker_main, NULL) != 0) {
        g_pool.running = 0;
        pthread_cond_broadcast(&g_pool.read_q.cond);
        pthread_cond_broadcast(&g_pool.write_q.cond);
        pthread_mutex_unlock(&g_pool.state_mutex);
        return -1;
    }

    pthread_mutex_unlock(&g_pool.state_mutex);
    DEBUG_LOG("Thread pool initialized: %d readers + 1 writer", g_pool.reader_count);
    return 0;
}

void thread_pool_shutdown(void) {
    pthread_mutex_lock(&g_pool.state_mutex);
    if (!g_pool.running) {
        pthread_mutex_unlock(&g_pool.state_mutex);
        return;
    }

    g_pool.running = 0;
    pthread_cond_broadcast(&g_pool.read_q.cond);
    pthread_cond_broadcast(&g_pool.write_q.cond);
    pthread_mutex_unlock(&g_pool.state_mutex);

    for (int i = 0; i < g_pool.reader_count; i++) {
        pthread_join(g_pool.readers[i], NULL);
    }
    pthread_join(g_pool.writer, NULL);

    free(g_pool.readers);
    g_pool.readers = NULL;
    g_pool.reader_count = 0;

    queue_destroy(&g_pool.read_q);
    queue_destroy(&g_pool.write_q);

    DEBUG_LOG("Thread pool shutdown complete");
}

int thread_pool_is_running(void) {
    return g_pool.running;
}

static thread_pool_status_t submit_to_queue(thread_pool_queue_t* q, const thread_pool_job_t* job, uint64_t* out_job_id) {
    if (!g_pool.running) return THREAD_POOL_STATUS_SHUTTING_DOWN;
    if (!job) return THREAD_POOL_STATUS_INTERNAL_ERROR;

    thread_pool_job_node_t* node = calloc(1, sizeof(*node));
    if (!node) return THREAD_POOL_STATUS_INTERNAL_ERROR;

    node->job = *job;

    pthread_mutex_lock(&g_pool.state_mutex);
    node->job_id = g_pool.next_job_id++;
    pthread_mutex_unlock(&g_pool.state_mutex);

    if (out_job_id) *out_job_id = node->job_id;

    if (!queue_push(q, node)) {
        if (job->payload_free && job->payload) {
            job->payload_free(job->payload);
        }
        free(node);
        return THREAD_POOL_STATUS_QUEUE_FULL;
    }

    return THREAD_POOL_STATUS_OK;
}

thread_pool_status_t thread_pool_submit_read(const thread_pool_job_t* job, uint64_t* out_job_id) {
    return submit_to_queue(&g_pool.read_q, job, out_job_id);
}

thread_pool_status_t thread_pool_submit_write(const thread_pool_job_t* job, uint64_t* out_job_id) {
    return submit_to_queue(&g_pool.write_q, job, out_job_id);
}

__attribute__((constructor))
static void thread_pool_state_init_once(void) {
    pthread_mutex_init(&g_pool.state_mutex, NULL);
}