merge runtime/channels — MPMC buffered channels, channel_pipeline, channel_fan_out
This commit is contained in:
Will Anderson
@@ -10244,3 +10244,331 @@ el_val_t emit_event(el_val_t name_v, el_val_t duration_ms_v) {
|
||||
return trace_span_end(h);
|
||||
}
|
||||
|
||||
/* ── Threading seed primitives ───────────────────────────────────────────────
|
||||
* __thread_create(fn_name, arg) -> Int spawn El fn in a pthread, return tid
|
||||
* __thread_join(tid) -> String join thread, return result string
|
||||
* __mutex_new() -> Int allocate a mutex, return handle
|
||||
* __mutex_lock(m) lock mutex m
|
||||
* __mutex_unlock(m) unlock mutex m
|
||||
*
|
||||
* Every El fn compiles to a global C symbol. __thread_create uses dlsym to
|
||||
* look up the function by name and run it in a pthread. This means any El fn
|
||||
* with signature (String) -> String is directly threadable.
|
||||
*/
|
||||
|
||||
typedef el_val_t (*ElFn1)(el_val_t);
|
||||
|
||||
typedef struct {
|
||||
ElFn1 fn;
|
||||
el_val_t arg;
|
||||
el_val_t result;
|
||||
} ElThreadArg;
|
||||
|
||||
#define EL_THREAD_MAX 256
|
||||
|
||||
typedef struct {
|
||||
pthread_t tid;
|
||||
ElThreadArg* arg;
|
||||
int alive;
|
||||
} ElThread;
|
||||
|
||||
static ElThread _threads[EL_THREAD_MAX];
|
||||
static int _thread_count = 0;
|
||||
static pthread_mutex_t _thread_alloc_mu = PTHREAD_MUTEX_INITIALIZER;
|
||||
|
||||
static void* el_thread_runner(void* raw) {
|
||||
ElThreadArg* a = (ElThreadArg*)raw;
|
||||
a->result = a->fn(a->arg);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
el_val_t __thread_create(el_val_t fn_name_v, el_val_t arg_v) {
|
||||
const char* sym = EL_CSTR(fn_name_v);
|
||||
if (!sym || !*sym) return EL_INT(-1);
|
||||
void* p = dlsym(RTLD_DEFAULT, sym);
|
||||
if (!p) {
|
||||
fprintf(stderr, "[__thread_create] symbol not found: %s\n", sym);
|
||||
return EL_INT(-1);
|
||||
}
|
||||
ElThreadArg* a = (ElThreadArg*)malloc(sizeof(ElThreadArg));
|
||||
if (!a) return EL_INT(-1);
|
||||
a->fn = (ElFn1)p;
|
||||
a->arg = arg_v;
|
||||
a->result = EL_STR("");
|
||||
|
||||
pthread_mutex_lock(&_thread_alloc_mu);
|
||||
if (_thread_count >= EL_THREAD_MAX) {
|
||||
pthread_mutex_unlock(&_thread_alloc_mu);
|
||||
free(a);
|
||||
fprintf(stderr, "[__thread_create] thread table full\n");
|
||||
return EL_INT(-1);
|
||||
}
|
||||
int slot = _thread_count++;
|
||||
_threads[slot].arg = a;
|
||||
_threads[slot].alive = 1;
|
||||
pthread_mutex_unlock(&_thread_alloc_mu);
|
||||
|
||||
if (pthread_create(&_threads[slot].tid, NULL, el_thread_runner, a) != 0) {
|
||||
pthread_mutex_lock(&_thread_alloc_mu);
|
||||
_thread_count--;
|
||||
pthread_mutex_unlock(&_thread_alloc_mu);
|
||||
free(a);
|
||||
return EL_INT(-1);
|
||||
}
|
||||
return EL_INT(slot);
|
||||
}
|
||||
|
||||
el_val_t __thread_join(el_val_t tid_v) {
|
||||
int slot = (int)(int64_t)tid_v;
|
||||
if (slot < 0 || slot >= EL_THREAD_MAX) return EL_STR("");
|
||||
pthread_join(_threads[slot].tid, NULL);
|
||||
el_val_t result = _threads[slot].arg->result;
|
||||
free(_threads[slot].arg);
|
||||
_threads[slot].alive = 0;
|
||||
return result;
|
||||
}
|
||||
|
||||
/* Mutex table */
|
||||
|
||||
#define EL_MUTEX_MAX 64
|
||||
|
||||
typedef struct {
|
||||
pthread_mutex_t mu;
|
||||
int allocated;
|
||||
} ElMutexEntry;
|
||||
|
||||
static ElMutexEntry _mutexes[EL_MUTEX_MAX];
|
||||
static int _mutex_count = 0;
|
||||
static pthread_mutex_t _mutex_alloc_mu = PTHREAD_MUTEX_INITIALIZER;
|
||||
|
||||
el_val_t __mutex_new(void) {
|
||||
pthread_mutex_lock(&_mutex_alloc_mu);
|
||||
if (_mutex_count >= EL_MUTEX_MAX) {
|
||||
pthread_mutex_unlock(&_mutex_alloc_mu);
|
||||
fprintf(stderr, "[__mutex_new] mutex table full\n");
|
||||
return EL_INT(-1);
|
||||
}
|
||||
int slot = _mutex_count++;
|
||||
pthread_mutex_init(&_mutexes[slot].mu, NULL);
|
||||
_mutexes[slot].allocated = 1;
|
||||
pthread_mutex_unlock(&_mutex_alloc_mu);
|
||||
return EL_INT(slot);
|
||||
}
|
||||
|
||||
void __mutex_lock(el_val_t m_v) {
|
||||
int slot = (int)(int64_t)m_v;
|
||||
if (slot < 0 || slot >= EL_MUTEX_MAX || !_mutexes[slot].allocated) return;
|
||||
pthread_mutex_lock(&_mutexes[slot].mu);
|
||||
}
|
||||
|
||||
void __mutex_unlock(el_val_t m_v) {
|
||||
int slot = (int)(int64_t)m_v;
|
||||
if (slot < 0 || slot >= EL_MUTEX_MAX || !_mutexes[slot].allocated) return;
|
||||
pthread_mutex_unlock(&_mutexes[slot].mu);
|
||||
}
|
||||
|
||||
/* ── Channels ─────────────────────────────────────────────────────────────── *
|
||||
* Buffered MPMC channel backed by a mutex + condvar + circular buffer.
|
||||
* channel_new(capacity) -> Int (handle)
|
||||
* channel_send(ch, msg) — blocks if full (capacity > 0) or never (unbounded)
|
||||
* channel_recv(ch) -> String — blocks until a message is available
|
||||
* channel_try_recv(ch) -> String — non-blocking, returns "" if empty
|
||||
* channel_close(ch) — signal no more sends; recv drains remaining
|
||||
*
|
||||
* Bounded channels (cap > 0): circular buffer, sender blocks when full.
|
||||
* Unbounded channels (cap == 0): dynamic array, sender never blocks.
|
||||
*/
|
||||
#define EL_CHANNEL_MAX 64
|
||||
#define EL_CHANNEL_BUF 1024
|
||||
|
||||
typedef struct {
|
||||
char** buf;
|
||||
int cap; /* 0 = unbounded (grows dynamically) */
|
||||
int head, tail, count;
|
||||
int dyn_cap; /* allocated slots for unbounded mode */
|
||||
int closed;
|
||||
pthread_mutex_t mu;
|
||||
pthread_cond_t not_empty;
|
||||
pthread_cond_t not_full;
|
||||
} ElChannel;
|
||||
|
||||
static ElChannel _channels[EL_CHANNEL_MAX];
|
||||
static int _channel_count = 0;
|
||||
static pthread_mutex_t _channel_alloc_mu = PTHREAD_MUTEX_INITIALIZER;
|
||||
|
||||
el_val_t __channel_new(el_val_t capacity_v) {
|
||||
int cap = (int)(int64_t)capacity_v;
|
||||
if (cap < 0) cap = 0;
|
||||
|
||||
pthread_mutex_lock(&_channel_alloc_mu);
|
||||
if (_channel_count >= EL_CHANNEL_MAX) {
|
||||
pthread_mutex_unlock(&_channel_alloc_mu);
|
||||
fprintf(stderr, "[__channel_new] channel table full\n");
|
||||
return EL_INT(-1);
|
||||
}
|
||||
int slot = _channel_count++;
|
||||
pthread_mutex_unlock(&_channel_alloc_mu);
|
||||
|
||||
ElChannel* ch = &_channels[slot];
|
||||
memset(ch, 0, sizeof(*ch));
|
||||
ch->cap = cap;
|
||||
ch->closed = 0;
|
||||
ch->head = 0;
|
||||
ch->tail = 0;
|
||||
ch->count = 0;
|
||||
|
||||
if (cap > 0) {
|
||||
/* Bounded: fixed circular buffer. */
|
||||
ch->buf = (char**)malloc((size_t)cap * sizeof(char*));
|
||||
ch->dyn_cap = cap;
|
||||
} else {
|
||||
/* Unbounded: start with EL_CHANNEL_BUF slots, grow as needed. */
|
||||
ch->buf = (char**)malloc(EL_CHANNEL_BUF * sizeof(char*));
|
||||
ch->dyn_cap = EL_CHANNEL_BUF;
|
||||
}
|
||||
if (!ch->buf) {
|
||||
fprintf(stderr, "[__channel_new] out of memory\n");
|
||||
return EL_INT(-1);
|
||||
}
|
||||
|
||||
pthread_mutex_init(&ch->mu, NULL);
|
||||
pthread_cond_init(&ch->not_empty, NULL);
|
||||
pthread_cond_init(&ch->not_full, NULL);
|
||||
|
||||
return EL_INT(slot);
|
||||
}
|
||||
|
||||
void __channel_send(el_val_t ch_v, el_val_t msg_v) {
|
||||
int slot = (int)(int64_t)ch_v;
|
||||
if (slot < 0 || slot >= EL_CHANNEL_MAX) return;
|
||||
ElChannel* ch = &_channels[slot];
|
||||
|
||||
const char* msg = EL_CSTR(msg_v);
|
||||
if (!msg) msg = "";
|
||||
char* copy = strdup(msg); /* channel owns the string */
|
||||
|
||||
pthread_mutex_lock(&ch->mu);
|
||||
|
||||
if (ch->closed) {
|
||||
/* Send on closed channel is a no-op (drop the message). */
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
free(copy);
|
||||
return;
|
||||
}
|
||||
|
||||
if (ch->cap > 0) {
|
||||
/* Bounded: block while full. */
|
||||
while (ch->count >= ch->cap && !ch->closed) {
|
||||
pthread_cond_wait(&ch->not_full, &ch->mu);
|
||||
}
|
||||
if (ch->closed) {
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
free(copy);
|
||||
return;
|
||||
}
|
||||
ch->buf[ch->tail] = copy;
|
||||
ch->tail = (ch->tail + 1) % ch->cap;
|
||||
ch->count++;
|
||||
} else {
|
||||
/* Unbounded: grow the buffer if needed. */
|
||||
if (ch->count >= ch->dyn_cap) {
|
||||
int new_cap = ch->dyn_cap * 2;
|
||||
char** grown = (char**)realloc(ch->buf, (size_t)new_cap * sizeof(char*));
|
||||
if (!grown) {
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
free(copy);
|
||||
fprintf(stderr, "[__channel_send] out of memory growing channel\n");
|
||||
return;
|
||||
}
|
||||
/* The circular buffer may have wrapped. Linearise it first.
|
||||
* In unbounded mode head is always 0 (we append at tail, drain
|
||||
* from head), so a simple memmove isn't needed — but if the
|
||||
* buffer did wrap (tail < head after growth), we need to fix up.
|
||||
* Simplest safe path: if tail wrapped, move the head..old_cap
|
||||
* segment to new_cap..new_cap+(old_cap-head). */
|
||||
if (ch->tail < ch->head) {
|
||||
/* Wrapped: [head..old_cap) is the front, [0..tail) is the back. */
|
||||
int front = ch->dyn_cap - ch->head;
|
||||
memmove(grown + ch->dyn_cap, grown + ch->head, (size_t)front * sizeof(char*));
|
||||
ch->head = ch->dyn_cap;
|
||||
}
|
||||
ch->buf = grown;
|
||||
ch->dyn_cap = new_cap;
|
||||
}
|
||||
ch->buf[ch->tail] = copy;
|
||||
ch->tail = (ch->tail + 1) % ch->dyn_cap;
|
||||
ch->count++;
|
||||
}
|
||||
|
||||
pthread_cond_signal(&ch->not_empty);
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
}
|
||||
|
||||
el_val_t __channel_recv(el_val_t ch_v) {
|
||||
int slot = (int)(int64_t)ch_v;
|
||||
if (slot < 0 || slot >= EL_CHANNEL_MAX) return EL_STR("");
|
||||
ElChannel* ch = &_channels[slot];
|
||||
|
||||
pthread_mutex_lock(&ch->mu);
|
||||
|
||||
/* Block until there is a message or the channel is closed and drained. */
|
||||
while (ch->count == 0 && !ch->closed) {
|
||||
pthread_cond_wait(&ch->not_empty, &ch->mu);
|
||||
}
|
||||
|
||||
if (ch->count == 0) {
|
||||
/* Closed and empty — signal EOF. */
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
return EL_STR("");
|
||||
}
|
||||
|
||||
int buf_cap = (ch->cap > 0) ? ch->cap : ch->dyn_cap;
|
||||
char* msg = ch->buf[ch->head];
|
||||
ch->head = (ch->head + 1) % buf_cap;
|
||||
ch->count--;
|
||||
|
||||
pthread_cond_signal(&ch->not_full);
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
|
||||
/* Hand the string to the arena so it is freed after the request. */
|
||||
el_arena_track(msg);
|
||||
return EL_STR(msg);
|
||||
}
|
||||
|
||||
el_val_t __channel_try_recv(el_val_t ch_v) {
|
||||
int slot = (int)(int64_t)ch_v;
|
||||
if (slot < 0 || slot >= EL_CHANNEL_MAX) return EL_STR("");
|
||||
ElChannel* ch = &_channels[slot];
|
||||
|
||||
pthread_mutex_lock(&ch->mu);
|
||||
|
||||
if (ch->count == 0) {
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
return EL_STR("");
|
||||
}
|
||||
|
||||
int buf_cap = (ch->cap > 0) ? ch->cap : ch->dyn_cap;
|
||||
char* msg = ch->buf[ch->head];
|
||||
ch->head = (ch->head + 1) % buf_cap;
|
||||
ch->count--;
|
||||
|
||||
pthread_cond_signal(&ch->not_full);
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
|
||||
el_arena_track(msg);
|
||||
return EL_STR(msg);
|
||||
}
|
||||
|
||||
void __channel_close(el_val_t ch_v) {
|
||||
int slot = (int)(int64_t)ch_v;
|
||||
if (slot < 0 || slot >= EL_CHANNEL_MAX) return;
|
||||
ElChannel* ch = &_channels[slot];
|
||||
|
||||
pthread_mutex_lock(&ch->mu);
|
||||
ch->closed = 1;
|
||||
/* Wake all blocked recvers and senders so they can observe the close. */
|
||||
pthread_cond_broadcast(&ch->not_empty);
|
||||
pthread_cond_broadcast(&ch->not_full);
|
||||
pthread_mutex_unlock(&ch->mu);
|
||||
}
|
||||
|
||||
|
||||
@@ -751,6 +751,39 @@ el_val_t trace_span_start(el_val_t name);
|
||||
el_val_t trace_span_end(el_val_t span_handle);
|
||||
el_val_t emit_event(el_val_t name, el_val_t duration_ms);
|
||||
|
||||
/* ── Threading seed primitives ────────────────────────────────────────────────
|
||||
* These are the low-level C primitives that back thread.el and channel.el.
|
||||
* El programs call them via their El wrappers (spawn, join, __mutex_new, etc.)
|
||||
* rather than directly.
|
||||
*
|
||||
* __thread_create(fn_name, arg) — dlsym-resolves fn_name, spawns pthread,
|
||||
* returns a slot index (Int) usable with __thread_join.
|
||||
* __thread_join(tid) — joins the thread, returns its String result.
|
||||
* __mutex_new() — allocates a mutex, returns handle (Int).
|
||||
* __mutex_lock(m) — locks mutex m (blocks until available).
|
||||
* __mutex_unlock(m) — unlocks mutex m. */
|
||||
|
||||
el_val_t __thread_create(el_val_t fn_name, el_val_t arg);
|
||||
el_val_t __thread_join(el_val_t tid);
|
||||
el_val_t __mutex_new(void);
|
||||
void __mutex_lock(el_val_t m);
|
||||
void __mutex_unlock(el_val_t m);
|
||||
|
||||
/* ── Channel seed primitives ─────────────────────────────────────────────────
|
||||
* Buffered MPMC channels. All values are Strings; handles are Ints.
|
||||
*
|
||||
* __channel_new(capacity) — create channel; cap=0 means unbounded.
|
||||
* __channel_send(ch, msg) — push msg; blocks if bounded and full.
|
||||
* __channel_recv(ch) — pop msg; blocks until available; "" on closed+empty.
|
||||
* __channel_try_recv(ch) — non-blocking pop; "" if empty.
|
||||
* __channel_close(ch) — mark closed; wakes all blocked recvers/senders. */
|
||||
|
||||
el_val_t __channel_new(el_val_t capacity);
|
||||
void __channel_send(el_val_t ch, el_val_t msg);
|
||||
el_val_t __channel_recv(el_val_t ch);
|
||||
el_val_t __channel_try_recv(el_val_t ch);
|
||||
void __channel_close(el_val_t ch);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
#endif
|
||||
|
||||
Reference in New Issue
Block a user