dwn/src/futures.c

/*
 * DWN - Desktop Window Manager
 * retoor <retoor@molodetz.nl>
 * Future/Promise Implementation - Async Result Handling
 */

#include "threading.h"
#include "util.h"

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

/* ============================================================================
 * Future Implementation
 * ============================================================================ */

typedef struct FutureCallbackNode {
    FutureCallback callback;
    void *user_data;
    struct FutureCallbackNode *next;
} FutureCallbackNode;

struct Future {
    pthread_mutex_t mutex;
    pthread_cond_t ready;
    
    FutureResult result;
    int error_code;
    _Atomic int ready_flag;
    _Atomic int has_result;
    
    FutureCallbackNode *callbacks;
    _Atomic int callback_count;
};

Future* future_create(void)
{
    Future *f = dwn_malloc(sizeof(Future));
    if (!f) return NULL;
    
    atomic_store_explicit(&f->ready_flag, 0, ATOMIC_RELEASE);
    atomic_store_explicit(&f->has_result, 0, ATOMIC_RELEASE);
    atomic_store_explicit(&f->callback_count, 0, ATOMIC_RELAXED);
    f->result = NULL;
    f->error_code = 0;
    f->callbacks = NULL;
    
    if (pthread_mutex_init(&f->mutex, NULL) != 0) {
        dwn_free(f);
        return NULL;
    }
    
    if (pthread_cond_init(&f->ready, NULL) != 0) {
        pthread_mutex_destroy(&f->mutex);
        dwn_free(f);
        return NULL;
    }
    
    return f;
}

void future_destroy(Future *f)
{
    if (!f) return;
    
    /* Free callback chain */
    FutureCallbackNode *node = f->callbacks;
    while (node) {
        FutureCallbackNode *next = node->next;
        dwn_free(node);
        node = next;
    }
    
    pthread_cond_destroy(&f->ready);
    pthread_mutex_destroy(&f->mutex);
    dwn_free(f);
}

void future_set_result(Future *f, FutureResult result)
{
    if (!f) return;
    
    pthread_mutex_lock(&f->mutex);
    
    if (atomic_load_explicit(&f->has_result, ATOMIC_ACQUIRE)) {
        pthread_mutex_unlock(&f->mutex);
        return;  /* Already set */
    }
    
    f->result = result;
    f->error_code = 0;
    atomic_store_explicit(&f->has_result, 1, ATOMIC_RELEASE);
    atomic_store_explicit(&f->ready_flag, 1, ATOMIC_RELEASE);
    
    /* Execute callbacks */
    FutureCallbackNode *node = f->callbacks;
    while (node) {
        node->callback(f, result, node->user_data);
        node = node->next;
    }
    
    pthread_cond_broadcast(&f->ready);
    pthread_mutex_unlock(&f->mutex);
}

void future_set_error(Future *f, int error_code)
{
    if (!f) return;
    
    pthread_mutex_lock(&f->mutex);
    
    if (atomic_load_explicit(&f->has_result, ATOMIC_ACQUIRE)) {
        pthread_mutex_unlock(&f->mutex);
        return;
    }
    
    f->result = NULL;
    f->error_code = error_code;
    atomic_store_explicit(&f->has_result, 1, ATOMIC_RELEASE);
    atomic_store_explicit(&f->ready_flag, 1, ATOMIC_RELEASE);
    
    /* Execute callbacks with NULL result */
    FutureCallbackNode *node = f->callbacks;
    while (node) {
        node->callback(f, NULL, node->user_data);
        node = node->next;
    }
    
    pthread_cond_broadcast(&f->ready);
    pthread_mutex_unlock(&f->mutex);
}

FutureResult future_get(Future *f, ThreadStatus *status)
{
    return future_get_timeout(f, 0, status);
}

FutureResult future_get_timeout(Future *f, uint64_t timeout_ms, ThreadStatus *status)
{
    if (!f) {
        if (status) *status = THREAD_ERROR_INVALID;
        return NULL;
    }
    
    pthread_mutex_lock(&f->mutex);
    
    if (!atomic_load_explicit(&f->ready_flag, ATOMIC_ACQUIRE)) {
        if (timeout_ms == 0) {
            pthread_cond_wait(&f->ready, &f->mutex);
        } else {
            struct timespec ts;
            clock_gettime(CLOCK_REALTIME, &ts);
            ts.tv_sec += timeout_ms / 1000;
            ts.tv_nsec += (timeout_ms % 1000) * 1000000;
            if (ts.tv_nsec >= 1000000000) {
                ts.tv_sec++;
                ts.tv_nsec -= 1000000000;
            }
            
            int ret = pthread_cond_timedwait(&f->ready, &f->mutex, &ts);
            if (ret == ETIMEDOUT) {
                pthread_mutex_unlock(&f->mutex);
                if (status) *status = THREAD_ERROR_TIMEOUT;
                return NULL;
            }
        }
    }
    
    FutureResult result = f->result;
    if (status) {
        *status = (f->error_code != 0) ? THREAD_ERROR : THREAD_OK;
    }
    
    pthread_mutex_unlock(&f->mutex);
    return result;
}

bool future_is_ready(Future *f)
{
    if (!f) return false;
    return atomic_load_explicit(&f->ready_flag, ATOMIC_ACQUIRE) != 0;
}

void future_then(Future *f, FutureCallback callback, void *user_data)
{
    if (!f || !callback) return;
    
    pthread_mutex_lock(&f->mutex);
    
    /* If already ready, execute immediately */
    if (atomic_load_explicit(&f->ready_flag, ATOMIC_ACQUIRE)) {
        pthread_mutex_unlock(&f->mutex);
        callback(f, f->result, user_data);
        return;
    }
    
    /* Add to callback chain */
    FutureCallbackNode *node = dwn_malloc(sizeof(FutureCallbackNode));
    if (!node) {
        pthread_mutex_unlock(&f->mutex);
        return;
    }
    
    node->callback = callback;
    node->user_data = user_data;
    node->next = f->callbacks;
    f->callbacks = node;
    
    atomic_fetch_add_explicit(&f->callback_count, 1, ATOMIC_RELAXED);
    
    pthread_mutex_unlock(&f->mutex);
}

/* ============================================================================
 * Composite Futures
 * ============================================================================ */

typedef struct {
    Future *composite;
    _Atomic int remaining;
    uint32_t total;
    pthread_mutex_t mutex;
} CompositeFutureData;

static void all_callback(Future *f, FutureResult result, void *user_data)
{
    (void)f;
    (void)result;
    
    CompositeFutureData *data = user_data;
    
    int rem = atomic_fetch_sub_explicit(&data->remaining, 1, ATOMIC_ACQ_REL) - 1;
    
    if (rem == 0) {
        future_set_result(data->composite, (void*)(uintptr_t)data->total);
    }
}

Future* future_all(Future **futures, uint32_t count)
{
    if (!futures || count == 0) return NULL;
    
    Future *composite = future_create();
    if (!composite) return NULL;
    
    if (count == 1) {
        /* Optimize single future */
        FutureResult r = future_get(futures[0], NULL);
        future_set_result(composite, r);
        return composite;
    }
    
    CompositeFutureData *data = dwn_malloc(sizeof(CompositeFutureData));
    if (!data) {
        future_destroy(composite);
        return NULL;
    }
    
    data->composite = composite;
    atomic_store_explicit(&data->remaining, count, ATOMIC_RELEASE);
    data->total = count;
    pthread_mutex_init(&data->mutex, NULL);
    
    /* Attach callback to each future */
    for (uint32_t i = 0; i < count; i++) {
        if (future_is_ready(futures[i])) {
            /* Already done */
            if (atomic_fetch_sub_explicit(&data->remaining, 1, ATOMIC_ACQ_REL) - 1 == 0) {
                future_set_result(composite, (void*)(uintptr_t)count);
                pthread_mutex_destroy(&data->mutex);
                dwn_free(data);
                return composite;
            }
        } else {
            future_then(futures[i], all_callback, data);
        }
    }
    
    return composite;
}

static void any_callback(Future *f, FutureResult result, void *user_data)
{
    (void)f;
    
    CompositeFutureData *data = user_data;
    
    pthread_mutex_lock(&data->mutex);
    
    if (!future_is_ready(data->composite)) {
        future_set_result(data->composite, result);
    }
    
    pthread_mutex_unlock(&data->mutex);
}

Future* future_any(Future **futures, uint32_t count)
{
    if (!futures || count == 0) return NULL;
    
    Future *composite = future_create();
    if (!composite) return NULL;
    
    if (count == 1) {
        FutureResult r = future_get(futures[0], NULL);
        future_set_result(composite, r);
        return composite;
    }
    
    CompositeFutureData *data = dwn_malloc(sizeof(CompositeFutureData));
    if (!data) {
        future_destroy(composite);
        return NULL;
    }
    
    data->composite = composite;
    pthread_mutex_init(&data->mutex, NULL);
    
    /* Attach callback to each future */
    for (uint32_t i = 0; i < count; i++) {
        if (future_is_ready(futures[i])) {
            future_set_result(composite, futures[i]->result);
            pthread_mutex_destroy(&data->mutex);
            dwn_free(data);
            return composite;
        } else {
            future_then(futures[i], any_callback, data);
        }
    }
    
    return composite;
}