worker.c

Go to the documentation of this file.

#include "worker.h"
 
#include <libplatform/atomic.h>
#include <ascii-chat/common.h>
#include <libplatform/thread.h>
#include <ascii-chat/util/time.h>

struct quic_worker {
  quic_udp_endpoint_t *endpoint;
  mutex_t *lane;
  quic_worker_failure_fn failure;
  void *context;
  platform_thread_t thread;
  atomic_t stopping;
  atomic_t wake_pending;
  bool failure_on_lane;
  asciichat_error_t result;
};
 
static void *quic_worker_main(void *arg) {
  quic_worker_t *worker = arg;
  while (!atomic_load_u64(&worker->stopping)) {
    uint64_t wait_ns = NS_PER_SEC_INT;
    bool want_write = false;
    quic_connection_t *failed = NULL;
    mutex_lock(worker->lane);
    uint64_t now = time_get_ns();
    asciichat_error_t result = quic_udp_pump(worker->endpoint, now, &failed);
    bool handled = false;
    if (result != ASCIICHAT_OK && worker->failure && worker->failure_on_lane) {
      bool recovered = worker->failure(failed, result, worker->context);
      if (failed && recovered)
        handled = !quic_udp_contains(worker->endpoint, failed);
      if (failed && !handled)
        (void)worker->failure(NULL, result, worker->context);
    }
    atomic_store_u64(&worker->wake_pending, 0);
    if (result == ASCIICHAT_OK)
      result = quic_udp_wait_parameters(worker->endpoint, now, NS_PER_SEC_INT, &wait_ns, &want_write);
    mutex_unlock(worker->lane);
    if (result != ASCIICHAT_OK) {
      if (handled)
        continue;
      if (failed && worker->failure && !worker->failure_on_lane && worker->failure(failed, result, worker->context)) {
        mutex_lock(worker->lane);
        bool still_present = quic_udp_contains(worker->endpoint, failed);
        mutex_unlock(worker->lane);
        if (!still_present)
          continue;
      }
      asciichat_error_context_t error_context;
      if (HAS_ERRNO(&error_context))
        log_error("QUIC worker stopped: %s", error_context.context_message);
      worker->result = result;
      break;
    }
    if (atomic_load_u64(&worker->stopping))
      break;
    bool ready = false;
    result = quic_udp_wait(worker->endpoint, wait_ns, want_write, &ready);
    if (result != ASCIICHAT_OK) {
      if (worker->failure_on_lane && worker->failure) {
        mutex_lock(worker->lane);
        (void)worker->failure(NULL, result, worker->context);
        mutex_unlock(worker->lane);
      }
      worker->result = result;
      break;
    }
  }
  return NULL;
}
 
static asciichat_error_t worker_start(quic_udp_endpoint_t *endpoint, mutex_t *lane, quic_worker_failure_fn failure,
                                      void *context, bool failure_on_lane, quic_worker_t **out) {
  if (!out)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Missing QUIC worker output");
  *out = NULL;
  if (!endpoint || !lane)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC worker endpoint or lane");
  quic_worker_t *worker = SAFE_CALLOC(1, sizeof(*worker), quic_worker_t *);
  if (!worker)
    return SET_ERRNO(ERROR_MEMORY, "Cannot allocate QUIC worker");
  worker->endpoint = endpoint;
  worker->lane = lane;
  worker->failure = failure;
  worker->context = context;
  worker->failure_on_lane = failure_on_lane;
  if (platform_thread_create(&worker->thread, "quic_udp_worker", quic_worker_main, worker) != 0) {
    SAFE_FREE(worker);
    return SET_ERRNO(ERROR_THREAD, "Cannot start QUIC worker");
  }
  *out = worker;
  return ASCIICHAT_OK;
}
 
asciichat_error_t quic_worker_start(quic_udp_endpoint_t *endpoint, mutex_t *lane, quic_worker_failure_fn failure,
                                    void *context, quic_worker_t **out) {
  return worker_start(endpoint, lane, failure, context, false, out);
}
 
asciichat_error_t quic_worker_start_locked_failure(quic_udp_endpoint_t *endpoint, mutex_t *lane,
                                                   quic_worker_failure_fn failure, void *context, quic_worker_t **out) {
  return worker_start(endpoint, lane, failure, context, true, out);
}
 
void quic_worker_wake(void *arg) {
  quic_worker_t *worker = arg;
  if (worker && atomic_exchange_u64(&worker->wake_pending, 1) == 0 && quic_udp_wake(worker->endpoint) != ASCIICHAT_OK)
    atomic_store_u64(&worker->wake_pending, 0);
}
 
asciichat_error_t quic_worker_stop(quic_worker_t **worker) {
  if (!worker || !*worker)
    return SET_ERRNO(ERROR_INVALID_PARAM, "Invalid QUIC worker stop");
  atomic_store_u64(&(*worker)->stopping, 1);
  quic_worker_wake(*worker);
  if (thread_join(&(*worker)->thread, NULL) != 0)
    return SET_ERRNO(ERROR_THREAD, "Cannot join QUIC worker");
  asciichat_error_t result = (*worker)->result;
  SAFE_FREE(*worker);
  *worker = NULL;
  return result;
}