add channels to El — buffered MPMC channel with send/recv/close
Introduces Go-style channels as El's mid-flight communication primitive, completing the threading model: threads can now not only spawn/join but also communicate while running. Part 1 — seed layer (el_runtime.c / el_runtime.h): - Add __thread_create/__thread_join/__mutex_new/__mutex_lock/__mutex_unlock as C seed primitives (dlsym-based thread dispatch, pthread mutex table) - Add __channel_new/__channel_send/__channel_recv/__channel_try_recv/__channel_close as MPMC channel seed primitives backed by mutex + condvar + circular buffer - Bounded channels (cap > 0): circular buffer, sender blocks when full - Unbounded channels (cap == 0): dynamic array, grows on demand, never blocks - channel_close wakes all blocked recvers/senders; recv drains then returns "" Part 2 — El API (runtime/channel.el): - channel_new/send/recv/try_recv/close — thin wrappers over seed layer - channel_pipeline — spawn N worker threads reading from in_ch, applying fn_name, writing to out_ch; workers exit on "" sentinel from close - channel_drain — collect all messages from a closed channel into [String] - channel_fan_out — send a [String] list into a channel then close it Part 3 — codegen.el: - Register all 10 seed builtins (__thread_* + __channel_*) in builtin_arity so the arity checker validates call sites at compile time
This commit is contained in:
Will Anderson
@@ -10188,3 +10188,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);
|
||||
}
|
||||
|
||||
|
||||
@@ -749,6 +749,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
|
||||
|
||||
@@ -2111,6 +2111,18 @@ fn builtin_arity(name: String) -> Int {
|
||||
if str_eq(name, "get") { return 2 }
|
||||
if str_eq(name, "map_get") { return 2 }
|
||||
if str_eq(name, "map_set") { return 3 }
|
||||
// Threading seed primitives
|
||||
if str_eq(name, "__thread_create") { return 2 }
|
||||
if str_eq(name, "__thread_join") { return 1 }
|
||||
if str_eq(name, "__mutex_new") { return 0 }
|
||||
if str_eq(name, "__mutex_lock") { return 1 }
|
||||
if str_eq(name, "__mutex_unlock") { return 1 }
|
||||
// Channel seed primitives
|
||||
if str_eq(name, "__channel_new") { return 1 }
|
||||
if str_eq(name, "__channel_send") { return 2 }
|
||||
if str_eq(name, "__channel_recv") { return 1 }
|
||||
if str_eq(name, "__channel_try_recv") { return 1 }
|
||||
if str_eq(name, "__channel_close") { return 1 }
|
||||
// -1 sentinel: variadic / unknown / user-defined -> no check.
|
||||
return -1
|
||||
}
|
||||
|
||||
@@ -0,0 +1,164 @@
|
||||
// channel.el — Go-style channels for El
|
||||
//
|
||||
// Channels are the communication primitive for concurrent El programs.
|
||||
// Threads send values into a channel; other threads receive them.
|
||||
// Channels are typed by convention — all values are Strings.
|
||||
//
|
||||
// Backed by four seed primitives in el_runtime.c:
|
||||
// __channel_new(capacity) -> Int create channel; cap=0 = unbounded
|
||||
// __channel_send(ch, msg) push msg; blocks if bounded and full
|
||||
// __channel_recv(ch) -> String pop msg; blocks until available; "" on close
|
||||
// __channel_try_recv(ch) -> String non-blocking pop; "" if empty
|
||||
// __channel_close(ch) mark closed; wake all blocked recvers
|
||||
//
|
||||
// Usage:
|
||||
// let ch: Int = channel_new(10) // buffered channel, capacity 10
|
||||
// spawn("producer", int_to_str(ch))
|
||||
// let msg: String = channel_recv(ch)
|
||||
|
||||
// ── Core channel API ─────────────────────────────────────────────────────────
|
||||
|
||||
// channel_new — create a channel with the given buffer capacity.
|
||||
//
|
||||
// capacity: 0 = unbounded (never blocks sender)
|
||||
// N = bounded buffer of N messages (sender blocks when full)
|
||||
//
|
||||
// Returns a channel handle (Int) to pass to send/recv/close.
|
||||
fn channel_new(capacity: Int) -> Int {
|
||||
return __channel_new(capacity)
|
||||
}
|
||||
|
||||
// channel_send — send a message into the channel.
|
||||
//
|
||||
// Blocks if the channel is bounded and full.
|
||||
// No-op if the channel is already closed.
|
||||
fn channel_send(ch: Int, msg: String) {
|
||||
__channel_send(ch, msg)
|
||||
}
|
||||
|
||||
// channel_recv — receive the next message from the channel.
|
||||
//
|
||||
// Blocks until a message is available.
|
||||
// Returns "" when the channel is closed and all buffered messages are drained.
|
||||
// The "" sentinel signals end-of-stream to consumers in a loop.
|
||||
fn channel_recv(ch: Int) -> String {
|
||||
return __channel_recv(ch)
|
||||
}
|
||||
|
||||
// channel_try_recv — non-blocking receive.
|
||||
//
|
||||
// Returns the next message if one is available, or "" if the channel is empty.
|
||||
// Does not block. Callers must distinguish "" (empty) from a legitimate ""
|
||||
// message by convention — use a non-empty sentinel in the message protocol.
|
||||
fn channel_try_recv(ch: Int) -> String {
|
||||
return __channel_try_recv(ch)
|
||||
}
|
||||
|
||||
// channel_close — signal that no more messages will be sent.
|
||||
//
|
||||
// After close, channel_recv continues to drain buffered messages then
|
||||
// returns "" on every subsequent call. channel_send on a closed channel
|
||||
// is a no-op (the message is dropped).
|
||||
fn channel_close(ch: Int) {
|
||||
__channel_close(ch)
|
||||
}
|
||||
|
||||
// ── channel_pipeline ─────────────────────────────────────────────────────────
|
||||
|
||||
// channel_pipeline — producer/consumer pipeline with parallel workers.
|
||||
//
|
||||
// Reads messages from in_ch, applies fn_name to each, writes results to out_ch.
|
||||
// Spawns `workers` concurrent worker threads — each drains in_ch independently,
|
||||
// so messages are processed in arrival order within each worker but not globally.
|
||||
//
|
||||
// fn_name must be an El fn with signature (String) -> String.
|
||||
//
|
||||
// Call channel_close(in_ch) to signal EOF. Workers exit when they receive "".
|
||||
// The caller must also close out_ch after all workers finish (via join).
|
||||
//
|
||||
// let in_ch: Int = channel_new(0)
|
||||
// let out_ch: Int = channel_new(0)
|
||||
// channel_pipeline(in_ch, out_ch, "process_item", 4)
|
||||
// channel_send(in_ch, "work-1")
|
||||
// channel_close(in_ch)
|
||||
// let result: String = channel_recv(out_ch)
|
||||
fn channel_pipeline(in_ch: Int, out_ch: Int, fn_name: String, workers: Int) {
|
||||
let i: Int = 0
|
||||
while i < workers {
|
||||
let arg: String = "{\"in_ch\":" + int_to_str(in_ch) +
|
||||
",\"out_ch\":" + int_to_str(out_ch) +
|
||||
",\"fn\":\"" + fn_name + "\"}"
|
||||
let _tid: Int = spawn("_channel_worker", arg)
|
||||
let i = i + 1
|
||||
}
|
||||
}
|
||||
|
||||
// _channel_worker — internal worker for channel_pipeline.
|
||||
//
|
||||
// Reads messages from in_ch until it receives "" (closed+empty), applies
|
||||
// fn_name to each, and writes results to out_ch. Runs in its own thread
|
||||
// (spawned by channel_pipeline).
|
||||
fn _channel_worker(arg: String) -> String {
|
||||
let in_ch: Int = str_to_int(json_get(arg, "in_ch"))
|
||||
let out_ch: Int = str_to_int(json_get(arg, "out_ch"))
|
||||
let fn_name: String = json_get(arg, "fn")
|
||||
let running: Bool = true
|
||||
while running {
|
||||
let msg: String = channel_recv(in_ch)
|
||||
if str_eq(msg, "") {
|
||||
let running = false
|
||||
} else {
|
||||
// Spawn fn_name in a child thread so it cannot block the worker loop.
|
||||
let tid: Int = spawn(fn_name, msg)
|
||||
let result: String = join(tid)
|
||||
channel_send(out_ch, result)
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// ── channel_drain ────────────────────────────────────────────────────────────
|
||||
|
||||
// channel_drain — collect all messages from ch into a list.
|
||||
//
|
||||
// Reads until the channel is closed and empty (recv returns "").
|
||||
// Returns a [String] of all messages received.
|
||||
//
|
||||
// Typical usage: close the channel from the producer side, then call
|
||||
// channel_drain from the consumer to collect results.
|
||||
fn channel_drain(ch: Int) -> [String] {
|
||||
let results: [String] = el_list_empty()
|
||||
let running: Bool = true
|
||||
while running {
|
||||
let msg: String = channel_recv(ch)
|
||||
if str_eq(msg, "") {
|
||||
let running = false
|
||||
} else {
|
||||
let results = el_list_append(results, msg)
|
||||
}
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
// ── channel_fan_out ───────────────────────────────────────────────────────────
|
||||
|
||||
// channel_fan_out — send every item in a list into a channel.
|
||||
//
|
||||
// items: [String] — items to send
|
||||
// ch: Int — destination channel
|
||||
//
|
||||
// Sends all items then closes the channel to signal end-of-stream.
|
||||
// Intended for the producer side of a pipeline:
|
||||
//
|
||||
// channel_fan_out(items, in_ch)
|
||||
// let results: [String] = channel_drain(out_ch)
|
||||
fn channel_fan_out(items: [String], ch: Int) {
|
||||
let n: Int = el_list_len(items)
|
||||
let i: Int = 0
|
||||
while i < n {
|
||||
let item: String = el_list_get(items, i)
|
||||
channel_send(ch, item)
|
||||
let i = i + 1
|
||||
}
|
||||
channel_close(ch)
|
||||
}
|
||||
Reference in New Issue
Block a user