mirror of
https://github.com/janet-lang/janet
synced 2025-01-24 22:26:52 +00:00
Get mailbox API working.
This commit is contained in:
Calvin Rose
parent
ee646dadf2
commit
131733549d
@ -1,8 +1,8 @@
|
||||
(defn worker-main
|
||||
"Sends 11 messages back to parent"
|
||||
[parent]
|
||||
(def name (:receive parent))
|
||||
(def interval (:receive parent))
|
||||
(def name (thread/receive))
|
||||
(def interval (thread/receive))
|
||||
(for i 0 10
|
||||
(os/sleep interval)
|
||||
(:send parent (string/format "thread %s wakeup no. %d" name i)))
|
||||
@ -19,13 +19,9 @@
|
||||
(def joe (make-worker "joe" 0.03))
|
||||
(def sam (make-worker "sam" 0.05))
|
||||
|
||||
(:close joe)
|
||||
|
||||
(try (:receive joe) ([err] (print "Got expected error: " err)))
|
||||
|
||||
# Receive out of order
|
||||
(for i 0 22
|
||||
(print (thread/receive [bob sam])))
|
||||
(for i 0 33
|
||||
(print (thread/receive)))
|
||||
|
||||
#
|
||||
# Recursive Thread Tree - should pause for a bit, and then print a cool zigzag.
|
||||
@ -38,8 +34,8 @@
|
||||
|
||||
(defn worker-tree
|
||||
[parent]
|
||||
(def name (:receive parent))
|
||||
(def depth (:receive parent))
|
||||
(def name (thread/receive))
|
||||
(def depth (thread/receive))
|
||||
(if (< depth 5)
|
||||
(do
|
||||
(defn subtree []
|
||||
@ -49,25 +45,26 @@
|
||||
(:send (inc depth))))
|
||||
(let [l (subtree)
|
||||
r (subtree)
|
||||
lrep (thread/receive l)
|
||||
rrep (thread/receive r)]
|
||||
lrep (thread/receive)
|
||||
rrep (thread/receive)]
|
||||
(:send parent [name ;lrep ;rrep])))
|
||||
(do
|
||||
(:send parent [name]))))
|
||||
|
||||
(def lines (:receive (-> (thread/new) (:send worker-tree) (:send "adam") (:send 0))))
|
||||
(-> (thread/new) (:send worker-tree) (:send "adam") (:send 0))
|
||||
(def lines (thread/receive))
|
||||
(map print lines)
|
||||
|
||||
#
|
||||
# Receive timeout
|
||||
#
|
||||
|
||||
(def slow (make-worker "slow-loras" 4))
|
||||
(def slow (make-worker "slow-loras" 0.5))
|
||||
(for i 0 50
|
||||
(try
|
||||
(let [msg (thread/receive slow 0.46)]
|
||||
(let [msg (thread/receive 0.1)]
|
||||
(print "\n" msg))
|
||||
([err] (prin ".") (:flush stdout))))
|
||||
|
||||
(print "\ndone timing, timeouts ending.")
|
||||
(try (while true (print (:receive slow))) ([err] (print "done")))
|
||||
(try (while true (print (thread/receive))) ([err] (print "done")))
|
||||
|
||||
@ -33,126 +33,72 @@
|
||||
#include <time.h>
|
||||
#include <pthread.h>
|
||||
|
||||
/* Global data */
|
||||
static pthread_rwlock_t janet_g_lock = PTHREAD_RWLOCK_INITIALIZER;
|
||||
static JanetMailbox **janet_g_mailboxes = NULL;
|
||||
static size_t janet_g_mailboxes_cap = 0;
|
||||
static size_t janet_g_mailboxes_count = 0;
|
||||
static uint64_t janet_g_next_mailbox_id = 0;
|
||||
|
||||
/* typedefed in janet.h */
|
||||
struct JanetMailbox {
|
||||
pthread_mutex_t lock;
|
||||
pthread_cond_t cond;
|
||||
uint64_t id;
|
||||
JanetMailbox *parent; /* May not have a parent */
|
||||
JanetTable *decode; /* Only allowed access by one thread */
|
||||
JanetBuffer buf;
|
||||
int refCount;
|
||||
int closed;
|
||||
};
|
||||
|
||||
static JANET_THREAD_LOCAL JanetMailbox *janet_vm_mailbox;
|
||||
static JANET_THREAD_LOCAL JanetMailbox *janet_vm_mailbox = NULL;
|
||||
|
||||
void janet_threads_init(void) {
|
||||
janet_vm_mailbox = malloc(sizeof(JanetMailbox));
|
||||
if (NULL == janet_vm_mailbox) {
|
||||
static JanetMailbox *janet_mailbox_create(JanetMailbox *parent, int refCount) {
|
||||
JanetMailbox *mailbox = malloc(sizeof(JanetMailbox));
|
||||
if (NULL == mailbox) {
|
||||
JANET_OUT_OF_MEMORY;
|
||||
}
|
||||
pthread_mutex_init(&janet_vm_mailbox->lock, NULL);
|
||||
pthread_cond_init(&janet_vm_mailbox->cond, NULL);
|
||||
janet_buffer_init(&janet_vm_mailbox->buf, 1024);
|
||||
janet_vm_mailbox->refcount = 1;
|
||||
janet_vm_mailbox->closed = 0;
|
||||
|
||||
/* Add mailbox to global table */
|
||||
pthread_rwlock_wrlock(&janet_janet_lock);
|
||||
janet_vm_mailbox->id = janet_g_next_mailbox_id++;
|
||||
size_t newcount = janet_g_mailboxes_count + 1;
|
||||
if (janet_g_mailboxes_cap < newcount) {
|
||||
size_t newcap = newcount * 2;
|
||||
JanetMailbox **mailboxes = realloc(janet_g_mailboxes, newcap * sizeof(JanetMailbox *));
|
||||
if (NULL == mailboxes) {
|
||||
pthread_rwlock_unlock(&janet_janet_lock);
|
||||
/* this maybe should be a different error, as this basically means
|
||||
* we cannot create a new thread. So janet_init should probably fail. */
|
||||
JANET_OUT_OF_MEMORY;
|
||||
return;
|
||||
}
|
||||
janet_g_mailboxes = mailboxes;
|
||||
janet_g_mailboxes_cap = newcap;
|
||||
}
|
||||
janet_g_mailboxes[janet_g_mailboxes_count] = janet_vm_mailbox;
|
||||
janet_g_mailboxes_count = newcount;
|
||||
pthread_rwlock_unlock(&janet_janet_lock);
|
||||
pthread_mutex_init(&mailbox->lock, NULL);
|
||||
pthread_cond_init(&mailbox->cond, NULL);
|
||||
janet_buffer_init(&mailbox->buf, 1024);
|
||||
mailbox->refCount = refCount;
|
||||
mailbox->closed = 0;
|
||||
mailbox->parent = parent;
|
||||
return mailbox;
|
||||
}
|
||||
|
||||
static void janet_mailbox_ref(JanetMailbox *mailbox) {
|
||||
pthread_mutex_lock(&mailbox->lock);
|
||||
mailbox->refCount++;
|
||||
pthread_mutex_unlock(&mailbox->lock);
|
||||
}
|
||||
|
||||
static JanetMailbox *janet_find_mailbox(uint64_t id, int remove) {
|
||||
JanetMailbox *ret = NULL;
|
||||
if (remove) {
|
||||
pthread_rwlock_wrlock(&janet_janet_lock);
|
||||
} else {
|
||||
pthread_rwlock_rdlock(&janet_janet_lock);
|
||||
}
|
||||
size_t i = 0;
|
||||
while (i < janet_g_mailboxes_count && janet_g_mailboxes[i]->id != mailbox->id)
|
||||
i++;
|
||||
if (i < janet_g_mailboxes_count) {
|
||||
ret = janet_g_mailboxes[i];
|
||||
if (remove) {
|
||||
janet_g_mailboxes[i] = janet_g_mailboxes[--janet_g_mailboxes_count];
|
||||
}
|
||||
}
|
||||
pthread_rwlock_unlock(&janet_janet_lock);
|
||||
return ret;
|
||||
static void janet_mailbox_destroy(JanetMailbox *mailbox) {
|
||||
pthread_mutex_destroy(&mailbox->lock);
|
||||
pthread_cond_destroy(&mailbox->cond);
|
||||
janet_buffer_deinit(&mailbox->buf);
|
||||
free(mailbox);
|
||||
}
|
||||
|
||||
/* Assumes you have the mailbox lock already */
|
||||
static void janet_mailbox_deref_with_lock(JanetMailbox *mailbox) {
|
||||
if (mailbox->refCount <= 1) {
|
||||
/* We are the last reference */
|
||||
pthread_mutex_destroy(&mailbox->lock);
|
||||
pthread_mutex_destroy(&mailbox->cond);
|
||||
janet_buffer_deinit(&mailbox->buf);
|
||||
janet_find_mailbox(mailbox->id, 1);
|
||||
free(mailbox);
|
||||
static void janet_mailbox_ref_with_lock(JanetMailbox *mailbox, int delta) {
|
||||
mailbox->refCount += delta;
|
||||
if (mailbox->refCount <= 0) {
|
||||
janet_mailbox_destroy(mailbox);
|
||||
} else {
|
||||
/* There are other references */
|
||||
if (mailbox == janet_vm_mailbox) {
|
||||
/* We own this mailbox, so mark it as closed for other references. */
|
||||
mailbox->closed = 1;
|
||||
}
|
||||
janet_vm_mailbox->refCount--;
|
||||
pthread_mutex_unlock(&mailbox->lock);
|
||||
}
|
||||
}
|
||||
|
||||
static void janet_mailbox_deref(JanetMailbox *mailbox) {
|
||||
static void janet_mailbox_ref(JanetMailbox *mailbox, int delta) {
|
||||
pthread_mutex_lock(&mailbox->lock);
|
||||
janet_mailbox_deref_with_lock(mailbox);
|
||||
janet_mailbox_ref_with_lock(mailbox, delta);
|
||||
}
|
||||
|
||||
void janet_threads_init(void) {
|
||||
if (NULL != janet_vm_mailbox) {
|
||||
return;
|
||||
}
|
||||
janet_vm_mailbox = janet_mailbox_create(NULL, 1);
|
||||
}
|
||||
|
||||
void janet_threads_deinit(void) {
|
||||
janet_mailbox_deref(janet_vm_mailbox);
|
||||
pthread_mutex_lock(&janet_vm_mailbox->lock);
|
||||
janet_vm_mailbox->closed = 1;
|
||||
janet_mailbox_ref_with_lock(janet_vm_mailbox, -1);
|
||||
janet_vm_mailbox = NULL;
|
||||
}
|
||||
|
||||
static JanetTable *janet_get_core_table(const char *name) {
|
||||
JanetTable *env = janet_core_env(NULL);
|
||||
Janet out = janet_wrap_nil();
|
||||
JanetBindingType bt = janet_resolve(env, janet_csymbol(name), &out);
|
||||
if (bt == JANET_BINDING_NONE) return NULL;
|
||||
if (!janet_checktype(out, JANET_TABLE)) return NULL;
|
||||
return janet_unwrap_table(out);
|
||||
}
|
||||
|
||||
static void janet_close_thread(JanetThread *thread) {
|
||||
if (thread->mailbox) {
|
||||
janet_mailbox_deref(thread->mailbox);
|
||||
janet_mailbox_ref(thread->mailbox, -1);
|
||||
thread->mailbox = NULL;
|
||||
}
|
||||
}
|
||||
@ -165,6 +111,7 @@ static int thread_gc(void *p, size_t size) {
|
||||
}
|
||||
|
||||
static int thread_mark(void *p, size_t size) {
|
||||
(void) size;
|
||||
JanetThread *thread = (JanetThread *)p;
|
||||
if (thread->encode) {
|
||||
janet_mark(janet_wrap_table(thread->encode));
|
||||
@ -179,10 +126,10 @@ int janet_thread_send(JanetThread *thread, Janet msg) {
|
||||
/* Ensure mailbox is not closed. */
|
||||
JanetMailbox *mailbox = thread->mailbox;
|
||||
if (NULL == mailbox) return 2;
|
||||
pthread_mutex_lock(mailbox->lock);
|
||||
pthread_mutex_lock(&mailbox->lock);
|
||||
if (mailbox->closed) {
|
||||
janet_mailbox_deref_with_lock(mailbox);
|
||||
thread->mailbox == NULL;
|
||||
janet_mailbox_ref_with_lock(mailbox, -1);
|
||||
thread->mailbox = NULL;
|
||||
return 2;
|
||||
}
|
||||
|
||||
@ -199,13 +146,16 @@ int janet_thread_send(JanetThread *thread, Janet msg) {
|
||||
mailbox->buf.count = oldcount;
|
||||
} else {
|
||||
janet_marshal(&mailbox->buf, msg, thread->encode, 0);
|
||||
if (oldcount == 0) {
|
||||
pthread_cond_signal(&mailbox->cond);
|
||||
}
|
||||
}
|
||||
|
||||
/* Cleanup */
|
||||
janet_vm_jmp_buf = old_buf;
|
||||
pthread_mutex_unlock(&mailbox->lock);
|
||||
|
||||
/* Potentially wake up a blocked thread */
|
||||
if (oldcount == 0 && ret == 0) {
|
||||
pthread_cond_signal(&mailbox->cond);
|
||||
}
|
||||
|
||||
return ret;
|
||||
}
|
||||
@ -226,12 +176,8 @@ static void janet_sec2ts(double sec, struct timespec *ts) {
|
||||
ts->tv_nsec = tvnsec;
|
||||
}
|
||||
|
||||
/* Returns 0 on successful message.
|
||||
* Returns 1 if nothing in queue or failed to get item. In this case,
|
||||
* also sets the channel's selector value.
|
||||
* Returns 2 if channel closed.
|
||||
* . */
|
||||
int janet_thread_receive(Janet *msg_out, double timeout, JanetTable *decode) {
|
||||
/* Returns 0 on successful message. Returns 1 if timedout */
|
||||
int janet_thread_receive(Janet *msg_out, double timeout) {
|
||||
pthread_mutex_lock(&janet_vm_mailbox->lock);
|
||||
|
||||
/* For timeouts */
|
||||
@ -260,7 +206,7 @@ int janet_thread_receive(Janet *msg_out, double timeout, JanetTable *decode) {
|
||||
const uint8_t *nextItem = NULL;
|
||||
Janet item = janet_unmarshal(
|
||||
janet_vm_mailbox->buf.data, janet_vm_mailbox->buf.count,
|
||||
0, rx->decode, &nextItem);
|
||||
0, janet_vm_mailbox->decode, &nextItem);
|
||||
|
||||
/* Update memory and put result into *msg_out */
|
||||
int32_t chunkCount = nextItem - janet_vm_mailbox->buf.data;
|
||||
@ -273,7 +219,10 @@ int janet_thread_receive(Janet *msg_out, double timeout, JanetTable *decode) {
|
||||
}
|
||||
}
|
||||
|
||||
if (nowait) {
|
||||
if (nowait || janet_vm_mailbox->refCount <= 1) {
|
||||
/* If there is only one ref, it is us. This means that if we
|
||||
* start waiting now, we can never possibly get a message, as
|
||||
* our reference will not propogate to other threads while we are blocked. */
|
||||
pthread_mutex_unlock(&janet_vm_mailbox->lock);
|
||||
return 1;
|
||||
}
|
||||
@ -282,14 +231,15 @@ int janet_thread_receive(Janet *msg_out, double timeout, JanetTable *decode) {
|
||||
if (timedwait) {
|
||||
if (pthread_cond_timedwait(
|
||||
&janet_vm_mailbox->cond,
|
||||
&janet_vm_mailbox->mutex,
|
||||
&janet_vm_mailbox->lock,
|
||||
&timeout_ts)) {
|
||||
return 1; /* Timeout */
|
||||
pthread_mutex_unlock(&janet_vm_mailbox->lock);
|
||||
return 1;
|
||||
}
|
||||
} else {
|
||||
pthread_cond_wait(
|
||||
&janet_vm_mailbox->cond,
|
||||
&janet_vm_mailbox->mutex);
|
||||
&janet_vm_mailbox->lock);
|
||||
}
|
||||
}
|
||||
|
||||
@ -308,9 +258,8 @@ static JanetAbstractType Thread_AT = {
|
||||
NULL
|
||||
};
|
||||
|
||||
static JanetThread *janet_make_thread(JanetMailbox *mailbox, Janet *encode) {
|
||||
static JanetThread *janet_make_thread(JanetMailbox *mailbox, JanetTable *encode) {
|
||||
JanetThread *thread = janet_abstract(&Thread_AT, sizeof(JanetThread));
|
||||
janet_mailbox_ref(mailbox)
|
||||
thread->mailbox = mailbox;
|
||||
thread->encode = encode;
|
||||
return thread;
|
||||
@ -320,24 +269,40 @@ JanetThread *janet_getthread(const Janet *argv, int32_t n) {
|
||||
return (JanetThread *) janet_getabstract(argv, n, &Thread_AT);
|
||||
}
|
||||
|
||||
static JanetTable *janet_get_core_table(const char *name) {
|
||||
JanetTable *env = janet_core_env(NULL);
|
||||
Janet out = janet_wrap_nil();
|
||||
JanetBindingType bt = janet_resolve(env, janet_csymbol(name), &out);
|
||||
if (bt == JANET_BINDING_NONE) return NULL;
|
||||
if (!janet_checktype(out, JANET_TABLE)) return NULL;
|
||||
return janet_unwrap_table(out);
|
||||
}
|
||||
|
||||
/* Runs in new thread */
|
||||
static int thread_worker(JanetMailbox *mailbox) {
|
||||
JanetFiber *fiber = NULL;
|
||||
Janet out;
|
||||
|
||||
/* Use the mailbox we were given */
|
||||
janet_vm_mailbox = mailbox;
|
||||
|
||||
/* Init VM */
|
||||
janet_init();
|
||||
|
||||
/* Get dictionaries */
|
||||
/* Get dictionaries for default encode/decode */
|
||||
JanetTable *encode = janet_get_core_table("make-image-dict");
|
||||
JanetTable *decode = janet_get_core_table("load-image-dict");
|
||||
mailbox->decode = janet_get_core_table("load-image-dict");
|
||||
|
||||
/* Create self thread */
|
||||
JanetThread *thread = janet_make_thread(mailbox, encode);
|
||||
Janet threadv = janet_wrap_abstract(thread);
|
||||
|
||||
/* Send pointer to current mailbox to parent */
|
||||
/* Create parent thread */
|
||||
JanetThread *parent = janet_make_thread(mailbox->parent, encode);
|
||||
janet_mailbox_ref(mailbox->parent, -1);
|
||||
mailbox->parent = NULL; /* only used to create the thread */
|
||||
Janet parentv = janet_wrap_abstract(parent);
|
||||
|
||||
/* Unmarshal the function */
|
||||
Janet funcv;
|
||||
int status = janet_thread_receive(&funcv, -1.0, decode);
|
||||
int status = janet_thread_receive(&funcv, -1.0);
|
||||
|
||||
if (status) goto error;
|
||||
if (!janet_checktype(funcv, JANET_FUNCTION)) goto error;
|
||||
JanetFunction *func = janet_unwrap_function(funcv);
|
||||
@ -348,34 +313,34 @@ static int thread_worker(JanetMailbox *mailbox) {
|
||||
}
|
||||
|
||||
/* Call function */
|
||||
Janet argv[1] = { threadv };
|
||||
JanetFiber *fiber = janet_fiber(func, 64, 1, argv);
|
||||
Janet out;
|
||||
janet_continue(fiber, janet_wrap_nil(), &out);
|
||||
Janet argv[1] = { parentv };
|
||||
fiber = janet_fiber(func, 64, 1, argv);
|
||||
JanetSignal sig = janet_continue(fiber, janet_wrap_nil(), &out);
|
||||
if (sig != JANET_SIGNAL_OK) {
|
||||
janet_eprintf("in thread %v: ", janet_wrap_abstract(janet_make_thread(mailbox, encode)));
|
||||
janet_stacktrace(fiber, out);
|
||||
}
|
||||
|
||||
/* Success */
|
||||
/* Normal exit */
|
||||
janet_deinit();
|
||||
return 0;
|
||||
|
||||
/* Fail */
|
||||
/* Fail to set something up */
|
||||
error:
|
||||
janet_eprintf("thread failed to start\n");
|
||||
janet_deinit();
|
||||
return 1;
|
||||
}
|
||||
|
||||
static void *janet_pthread_wrapper(void *param) {
|
||||
thread_worker((JanetChannel *)param);
|
||||
thread_worker((JanetMailbox *)param);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static int janet_thread_start_child(JanetThread *thread) {
|
||||
pthread_t handle;
|
||||
/* My rx is your tx and vice versa */
|
||||
int error = pthread_create(&handle, NULL, janet_pthread_wrapper, thread->rx);
|
||||
int error = pthread_create(&handle, NULL, janet_pthread_wrapper, thread->mailbox);
|
||||
if (error) {
|
||||
/* double close as there is no other side to close thread */
|
||||
janet_close_thread(thread);
|
||||
janet_close_thread(thread);
|
||||
return 1;
|
||||
} else {
|
||||
pthread_detach(handle);
|
||||
@ -388,68 +353,48 @@ static int janet_thread_start_child(JanetThread *thread) {
|
||||
*/
|
||||
|
||||
static Janet cfun_thread_new(int32_t argc, Janet *argv) {
|
||||
janet_arity(argc, 0, 2);
|
||||
JanetTable *encode = (argc < 1 || janet_checktype(argv[0], JANET_NIL))
|
||||
? janet_get_core_table("make-image-dict")
|
||||
: janet_gettable(argv, 0);
|
||||
JanetTable *decode = (argc < 2 || janet_checktype(argv[1], JANET_NIL))
|
||||
? janet_get_core_table("load-image-dict")
|
||||
: janet_gettable(argv, 1);
|
||||
JanetChannel *rx = malloc(2 * sizeof(JanetChannel));
|
||||
if (NULL == rx) {
|
||||
JANET_OUT_OF_MEMORY;
|
||||
}
|
||||
JanetChannel *tx = rx + 1;
|
||||
janet_channel_init(rx);
|
||||
janet_channel_init(tx);
|
||||
JanetThread *thread = janet_make_thread(rx, tx, encode, decode);
|
||||
if (janet_thread_start_child(thread))
|
||||
janet_fixarity(argc, 0);
|
||||
(void) argv;
|
||||
JanetTable *encode = janet_get_core_table("make-image-dict");
|
||||
JanetMailbox *mailbox = janet_mailbox_create(janet_vm_mailbox, 2);
|
||||
|
||||
/* one for created thread, one for ->parent reference in new mailbox */
|
||||
janet_mailbox_ref(janet_vm_mailbox, 2);
|
||||
|
||||
JanetThread *thread = janet_make_thread(mailbox, encode);
|
||||
if (janet_thread_start_child(thread)) {
|
||||
janet_mailbox_ref(mailbox, -1); /* mailbox reference */
|
||||
janet_mailbox_ref(janet_vm_mailbox, -1); /* ->parent reference */
|
||||
janet_panic("could not start thread");
|
||||
}
|
||||
return janet_wrap_abstract(thread);
|
||||
}
|
||||
|
||||
static Janet cfun_thread_send(int32_t argc, Janet *argv) {
|
||||
janet_fixarity(argc, 2);
|
||||
JanetChannel *tx = janet_getthread(argv, 0)->tx;
|
||||
if (NULL == tx) janet_panic("channel has closed");
|
||||
pthread_mutex_lock(&tx->lock);
|
||||
int status = janet_channel_send(tx, argv[1]);
|
||||
pthread_mutex_unlock(&tx->lock);
|
||||
if (status) {
|
||||
JanetThread *thread = janet_getthread(argv, 0);
|
||||
int status = janet_thread_send(thread, argv[1]);
|
||||
switch (status) {
|
||||
default:
|
||||
break;
|
||||
case 1:
|
||||
janet_panicf("failed to send message %v", argv[1]);
|
||||
case 2:
|
||||
janet_panic("thread mailbox is closed");
|
||||
}
|
||||
return argv[0];
|
||||
}
|
||||
|
||||
static Janet cfun_thread_receive(int32_t argc, Janet *argv) {
|
||||
janet_arity(argc, 1, 2);
|
||||
int status;
|
||||
Janet out = janet_wrap_nil();
|
||||
int32_t count;
|
||||
const Janet *items;
|
||||
double wait = janet_optnumber(argv, argc, 1, -1.0);
|
||||
if (janet_indexed_view(argv[0], &items, &count)) {
|
||||
/* Select on multiple threads */
|
||||
if (count == 0) janet_panic("expected at least 1 thread");
|
||||
int32_t realcount = 0;
|
||||
JanetChannel *rxs_stack[10] = {NULL};
|
||||
JanetChannel **rxs = (count > 10)
|
||||
? janet_smalloc(count * sizeof(JanetChannel *))
|
||||
: rxs_stack;
|
||||
for (int32_t i = 0; i < count; i++) {
|
||||
JanetThread *thread = janet_getthread(items, i);
|
||||
if (thread->rx != NULL) rxs[realcount++] = thread->rx;
|
||||
}
|
||||
status = janet_channel_select(realcount, rxs, &out, wait);
|
||||
if (rxs != rxs_stack) janet_sfree(rxs);
|
||||
} else {
|
||||
/* Get from one thread */
|
||||
JanetThread *thread = janet_getthread(argv, 0);
|
||||
if (NULL == thread->rx) janet_panic("channel has closed");
|
||||
status = janet_channel_select(1, &thread->rx, &out, wait);
|
||||
}
|
||||
if (status) {
|
||||
janet_panic("failed to receive message");
|
||||
janet_arity(argc, 0, 1);
|
||||
double wait = janet_optnumber(argv, argc, 0, -1.0);
|
||||
Janet out;
|
||||
int status = janet_thread_receive(&out, wait);
|
||||
switch (status) {
|
||||
default:
|
||||
break;
|
||||
case 1:
|
||||
janet_panicf("timeout after %f seconds", wait);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
@ -463,7 +408,6 @@ static Janet cfun_thread_close(int32_t argc, Janet *argv) {
|
||||
|
||||
static const JanetMethod janet_thread_methods[] = {
|
||||
{"send", cfun_thread_send},
|
||||
{"receive", cfun_thread_receive},
|
||||
{"close", cfun_thread_close},
|
||||
{NULL, NULL}
|
||||
};
|
||||
@ -477,7 +421,7 @@ static Janet janet_thread_getter(void *p, Janet key) {
|
||||
static const JanetReg threadlib_cfuns[] = {
|
||||
{
|
||||
"thread/new", cfun_thread_new,
|
||||
JDOC("(thread/new &opt encode-book decode-book)\n\n"
|
||||
JDOC("(thread/new)\n\n"
|
||||
"Start a new thread. The thread will wait for a message containing the function used to start the thread, which should be subsequently "
|
||||
"sent over after thread creation.")
|
||||
},
|
||||
@ -489,12 +433,9 @@ static const JanetReg threadlib_cfuns[] = {
|
||||
},
|
||||
{
|
||||
"thread/receive", cfun_thread_receive,
|
||||
JDOC("(thread/receive threads &opt timeout)\n\n"
|
||||
"Get a value sent to 1 or more threads. Will block if no value was sent to this thread "
|
||||
"yet. threads can also be an array or tuple of threads, in which case "
|
||||
"thread/receive will select on the first thread to return a value. Returns "
|
||||
"the message sent to the thread. If a timeout (in seconds) is provided, failure "
|
||||
"to get a message will throw an error after the timeout has elapsed.")
|
||||
JDOC("(thread/receive &opt timeout)\n\n"
|
||||
"Get a message sent to this thread. If timeout is provided, an error will be thrown after the timeout has elapsed but "
|
||||
"no messages are received.")
|
||||
},
|
||||
{
|
||||
"thread/close", cfun_thread_close,
|
||||
@ -508,6 +449,7 @@ static const JanetReg threadlib_cfuns[] = {
|
||||
/* Module entry point */
|
||||
void janet_lib_thread(JanetTable *env) {
|
||||
janet_core_cfuns(env, NULL, threadlib_cfuns);
|
||||
janet_register_abstract_type(&Thread_AT);
|
||||
}
|
||||
|
||||
#endif
|
||||
|
||||
Loading…
Reference in New Issue
Block a user