mirrors/janet
1
0
Fork 0
mirror of https://github.com/janet-lang/janet synced 2024-12-01 04:19:55 +00:00
Code Issues 1 Releases Wiki Activity

Merge pull request #726 from yumaikas/new-style-ev-bindings

Browse Source 
Update ev.c to use the new binding style that provides source information
This commit is contained in:
Calvin Rose 2021-07-27 20:17:19 -05:00 committed by GitHub
commit 617da24942
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 123 additions and 142 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

257
src/core/ev.c
View File

@ -721,7 +721,10 @@ static int janet_channel_pop(JanetChannel *channel, Janet *item, int is_choice)
/* Channel Methods */ /* Channel Methods */
static Janet cfun_channel_push(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_push,
"(ev/give channel value)",
"Write a value to a channel, suspending the current fiber if the channel is full."
) {
janet_fixarity(argc, 2); janet_fixarity(argc, 2);
JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT);
if (janet_channel_push(channel, argv[1], 0)) { if (janet_channel_push(channel, argv[1], 0)) {
@ -730,7 +733,10 @@ static Janet cfun_channel_push(int32_t argc, Janet *argv) {
return argv[0]; return argv[0];
} }
static Janet cfun_channel_pop(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_pop,
"(ev/take channel)",
"Read from a channel, suspending the current fiber if no value is available."
) {
janet_fixarity(argc, 1); janet_fixarity(argc, 1);
JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT);
Janet item; Janet item;
@ -740,7 +746,13 @@ static Janet cfun_channel_pop(int32_t argc, Janet *argv) {
janet_await(); janet_await();
} }
static Janet cfun_channel_choice(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_choice,
"(ev/select & clauses)",
"Block until the first of several channel operations occur. Returns a tuple of the form [:give chan] or [:take chan x], where "
"a :give tuple is the result of a write and :take tuple is the result of a write. Each clause must be either a channel (for "
"a channel take operation) or a tuple [channel x] for a channel give operation. Operations are tried in order, such that the first "
"clauses will take precedence over later clauses."
) {
janet_arity(argc, 1, -1); janet_arity(argc, 1, -1);
int32_t len; int32_t len;
const Janet *data; const Janet *data;
@ -782,19 +794,28 @@ static Janet cfun_channel_choice(int32_t argc, Janet *argv) {
janet_await(); janet_await();
} }
static Janet cfun_channel_full(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_full,
"(ev/full channel)",
"Check if a channel is full or not."
) {
janet_fixarity(argc, 1); janet_fixarity(argc, 1);
JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT);
return janet_wrap_boolean(janet_q_count(&channel->items) >= channel->limit); return janet_wrap_boolean(janet_q_count(&channel->items) >= channel->limit);
} }
static Janet cfun_channel_capacity(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_capacity,
"(ev/capacity channel)",
"Get the number of items a channel will store before blocking writers."
) {
janet_fixarity(argc, 1); janet_fixarity(argc, 1);
JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT);
return janet_wrap_integer(channel->limit); return janet_wrap_integer(channel->limit);
} }
static Janet cfun_channel_count(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_count,
"(ev/count channel)",
"Get the number of items currently waiting in a channel."
) {
janet_fixarity(argc, 1); janet_fixarity(argc, 1);
JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT); JanetChannel *channel = janet_getabstract(argv, 0, &ChannelAT);
return janet_wrap_integer(janet_q_count(&channel->items)); return janet_wrap_integer(janet_q_count(&channel->items));
@ -810,12 +831,19 @@ static void fisher_yates_args(int32_t argc, Janet *argv) {
} }
} }
static Janet cfun_channel_rchoice(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_rchoice,
"(ev/rselect & clauses)",
"Similar to ev/select, but will try clauses in a random order for fairness."
) {
fisher_yates_args(argc, argv); fisher_yates_args(argc, argv);
return cfun_channel_choice(argc, argv); return cfun_channel_choice(argc, argv);
} }
static Janet cfun_channel_new(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_channel_new,
"(ev/chan &opt capacity)",
"Create a new channel. capacity is the number of values to queue before "
"blocking writers, defaults to 0 if not provided. Returns a new channel."
) {
janet_arity(argc, 0, 1); janet_arity(argc, 0, 1);
int32_t limit = janet_optnat(argv, argc, 0, 0); int32_t limit = janet_optnat(argv, argc, 0, 0);
JanetChannel *channel = janet_abstract(&ChannelAT, sizeof(JanetChannel)); JanetChannel *channel = janet_abstract(&ChannelAT, sizeof(JanetChannel));
@ -2116,7 +2144,14 @@ error:
/* C functions */ /* C functions */
static Janet cfun_ev_go(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_ev_go,
"(ev/go fiber &opt value supervisor)",
"Put a fiber on the event loop to be resumed later. Optionally pass "
"a value to resume with, otherwise resumes with nil. Returns the fiber. "
"An optional `core/channel` can be provided as well as a supervisor. When various "
"events occur in the newly scheduled fiber, an event will be pushed to the supervisor. "
"If not provided, the new fiber will inherit the current supervisor."
) {
janet_arity(argc, 1, 3); janet_arity(argc, 1, 3);
JanetFiber *fiber = janet_getfiber(argv, 0); JanetFiber *fiber = janet_getfiber(argv, 0);
Janet value = argc >= 2 ? argv[1] : janet_wrap_nil(); Janet value = argc >= 2 ? argv[1] : janet_wrap_nil();
@ -2168,7 +2203,14 @@ static JanetEVGenericMessage janet_go_thread_subr(JanetEVGenericMessage args) {
return args; return args;
} }
static Janet cfun_ev_thread(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_ev_thread,
"(ev/thread fiber &opt value flags)",
"Resume a (copy of a) `fiber` in a new operating system thread, optionally passing `value` "
"to resume with. "
"Unlike `ev/go`, this function will suspend the current fiber until the thread is complete. "
"If you want to run the thread without waiting for a result, pass the `:n` flag to return nil immediately. "
"Otherwise, returns (a copy of) the final result from the fiber on the new thread."
) {
janet_arity(argc, 1, 3); janet_arity(argc, 1, 3);
janet_getfiber(argv, 0); janet_getfiber(argv, 0);
Janet value = argc >= 2 ? argv[1] : janet_wrap_nil(); Janet value = argc >= 2 ? argv[1] : janet_wrap_nil();
@ -2200,7 +2242,12 @@ static Janet cfun_ev_thread(int32_t argc, Janet *argv) {
} }
} }
static Janet cfun_ev_give_supervisor(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_ev_give_supervisor,
"(ev/give-supervisor tag & payload)",
"Send a message to the current supervior channel if there is one. The message will be a "
"tuple of all of the arguments combined into a single message, where the first element is tag. "
"By convention, tag should be a keyword indicating the type of message. Returns nil."
) {
janet_arity(argc, 1, -1); janet_arity(argc, 1, -1);
JanetChannel *chan = janet_vm.root_fiber->supervisor_channel; JanetChannel *chan = janet_vm.root_fiber->supervisor_channel;
if (NULL != chan) { if (NULL != chan) {
@ -2222,13 +2269,22 @@ JANET_NO_RETURN void janet_sleep_await(double sec) {
janet_await(); janet_await();
} }
static Janet cfun_ev_sleep(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_ev_sleep,
"(ev/sleep sec)",
"Suspend the current fiber for sec seconds without blocking the event loop."
) {
janet_fixarity(argc, 1); janet_fixarity(argc, 1);
double sec = janet_getnumber(argv, 0); double sec = janet_getnumber(argv, 0);
janet_sleep_await(sec); janet_sleep_await(sec);
} }
static Janet cfun_ev_deadline(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_ev_deadline,
"(ev/deadline sec &opt tocancel tocheck)",
"Set a deadline for a fiber `tocheck`. If `tocheck` is not finished after `sec` seconds, "
"`tocancel` will be canceled as with `ev/cancel`. "
"If `tocancel` and `tocheck` are not given, they default to `(fiber/root)` and "
"`(fiber/current)` respectively. Returns `tocancel`."
) {
janet_arity(argc, 1, 3); janet_arity(argc, 1, 3);
double sec = janet_getnumber(argv, 0); double sec = janet_getnumber(argv, 0);
JanetFiber *tocancel = janet_optfiber(argv, argc, 1, janet_vm.root_fiber); JanetFiber *tocancel = janet_optfiber(argv, argc, 1, janet_vm.root_fiber);
@ -2243,7 +2299,10 @@ static Janet cfun_ev_deadline(int32_t argc, Janet *argv) {
return janet_wrap_fiber(tocancel); return janet_wrap_fiber(tocancel);
} }
static Janet cfun_ev_cancel(int32_t argc, Janet *argv) { JANET_CORE_FN(cfun_ev_cancel,
"(ev/cancel fiber err)",
"Cancel a suspended fiber in the event loop. Differs from cancel in that it returns the canceled fiber immediately"
) {
janet_fixarity(argc, 2); janet_fixarity(argc, 2);
JanetFiber *fiber = janet_getfiber(argv, 0); JanetFiber *fiber = janet_getfiber(argv, 0);
Janet err = argv[1]; Janet err = argv[1];
@ -2251,14 +2310,25 @@ static Janet cfun_ev_cancel(int32_t argc, Janet *argv) {
return argv[0]; return argv[0];
} }
Janet janet_cfun_stream_close(int32_t argc, Janet *argv) { JANET_CORE_FN(janet_cfun_stream_close,
"(ev/close stream)",
"Close a stream. This should be the same as calling (:close stream) for all streams."
) {
janet_fixarity(argc, 1); janet_fixarity(argc, 1);
JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
janet_stream_close(stream); janet_stream_close(stream);
return argv[0]; return argv[0];
} }
Janet janet_cfun_stream_read(int32_t argc, Janet *argv) { JANET_CORE_FN(janet_cfun_stream_read,
"(ev/read stream n &opt buffer timeout)",
"Read up to n bytes into a buffer asynchronously from a stream. `n` can also be the keyword "
"`:all` to read into the buffer until end of stream. "
"Optionally provide a buffer to write into "
"as well as a timeout in seconds after which to cancel the operation and raise an error. "
"Returns the buffer if the read was successful or nil if end-of-stream reached. Will raise an "
"error if there are problems with the IO operation."
) {
janet_arity(argc, 2, 4); janet_arity(argc, 2, 4);
JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
janet_stream_flags(stream, JANET_STREAM_READABLE); janet_stream_flags(stream, JANET_STREAM_READABLE);
@ -2275,7 +2345,11 @@ Janet janet_cfun_stream_read(int32_t argc, Janet *argv) {
janet_await(); janet_await();
} }
Janet janet_cfun_stream_chunk(int32_t argc, Janet *argv) { JANET_CORE_FN(janet_cfun_stream_chunk,
"(ev/chunk stream n &opt buffer timeout)",
"Same as ev/read, but will not return early if less than n bytes are available. If an end of "
"stream is reached, will also return early with the collected bytes."
) {
janet_arity(argc, 2, 4); janet_arity(argc, 2, 4);
JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
janet_stream_flags(stream, JANET_STREAM_READABLE); janet_stream_flags(stream, JANET_STREAM_READABLE);
@ -2287,7 +2361,11 @@ Janet janet_cfun_stream_chunk(int32_t argc, Janet *argv) {
janet_await(); janet_await();
} }
Janet janet_cfun_stream_write(int32_t argc, Janet *argv) { JANET_CORE_FN(janet_cfun_stream_write,
"(ev/write stream data &opt timeout)",
"Write data to a stream, suspending the current fiber until the write "
"completes. Takes an optional timeout in seconds, after which will return nil. "
"Returns nil, or raises an error if the write failed.") {
janet_arity(argc, 2, 3); janet_arity(argc, 2, 3);
JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type); JanetStream *stream = janet_getabstract(argv, 0, &janet_stream_type);
janet_stream_flags(stream, JANET_STREAM_WRITABLE); janet_stream_flags(stream, JANET_STREAM_WRITABLE);
@ -2303,127 +2381,30 @@ Janet janet_cfun_stream_write(int32_t argc, Janet *argv) {
janet_await(); janet_await();
} }
static const JanetReg ev_cfuns[] = {
{
"ev/go", cfun_ev_go,
JDOC("(ev/go fiber &opt value supervisor)\n\n"
"Put a fiber on the event loop to be resumed later. Optionally pass "
"a value to resume with, otherwise resumes with nil. Returns the fiber. "
"An optional `core/channel` can be provided as well as a supervisor. When various "
"events occur in the newly scheduled fiber, an event will be pushed to the supervisor. "
"If not provided, the new fiber will inherit the current supervisor.")
},
{
"ev/thread", cfun_ev_thread,
JDOC("(ev/thread fiber &opt value flags)\n\n"
"Resume a (copy of a) `fiber` in a new operating system thread, optionally passing `value` "
"to resume with. "
"Unlike `ev/go`, this function will suspend the current fiber until the thread is complete. "
"If you want to run the thread without waiting for a result, pass the `:n` flag to return nil immediately. "
"Otherwise, returns (a copy of) the final result from the fiber on the new thread.")
},
{
"ev/give-supervisor", cfun_ev_give_supervisor,
JDOC("(ev/give-supervsior tag & payload)\n\n"
"Send a message to the current supervior channel if there is one. The message will be a "
"tuple of all of the arguments combined into a single message, where the first element is tag. "
"By convention, tag should be a keyword indicating the type of message. Returns nil.")
},
{
"ev/sleep", cfun_ev_sleep,
JDOC("(ev/sleep sec)\n\n"
"Suspend the current fiber for sec seconds without blocking the event loop.")
},
{
"ev/deadline", cfun_ev_deadline,
JDOC("(ev/deadline sec &opt tocancel tocheck)\n\n"
"Set a deadline for a fiber `tocheck`. If `tocheck` is not finished after `sec` seconds, "
"`tocancel` will be canceled as with `ev/cancel`. "
"If `tocancel` and `tocheck` are not given, they default to `(fiber/root)` and "
"`(fiber/current)` respectively. Returns `tocancel`.")
},
{
"ev/chan", cfun_channel_new,
JDOC("(ev/chan &opt capacity)\n\n"
"Create a new channel. capacity is the number of values to queue before "
"blocking writers, defaults to 0 if not provided. Returns a new channel.")
},
{
"ev/give", cfun_channel_push,
JDOC("(ev/give channel value)\n\n"
"Write a value to a channel, suspending the current fiber if the channel is full.")
},
{
"ev/take", cfun_channel_pop,
JDOC("(ev/take channel)\n\n"
"Read from a channel, suspending the current fiber if no value is available.")
},
{
"ev/full", cfun_channel_full,
JDOC("(ev/full channel)\n\n"
"Check if a channel is full or not.")
},
{
"ev/capacity", cfun_channel_capacity,
JDOC("(ev/capacity channel)\n\n"
"Get the number of items a channel will store before blocking writers.")
},
{
"ev/count", cfun_channel_count,
JDOC("(ev/count channel)\n\n"
"Get the number of items currently waiting in a channel.")
},
{
"ev/cancel", cfun_ev_cancel,
JDOC("(ev/cancel fiber err)\n\n"
"Cancel a suspended fiber in the event loop. Differs from cancel in that it returns the canceled fiber immediately")
},
{
"ev/select", cfun_channel_choice,
JDOC("(ev/select & clauses)\n\n"
"Block until the first of several channel operations occur. Returns a tuple of the form [:give chan] or [:take chan x], where "
"a :give tuple is the result of a write and :take tuple is the result of a write. Each clause must be either a channel (for "
"a channel take operation) or a tuple [channel x] for a channel give operation. Operations are tried in order, such that the first "
"clauses will take precedence over later clauses.")
},
{
"ev/rselect", cfun_channel_rchoice,
JDOC("(ev/rselect & clauses)\n\n"
"Similar to ev/select, but will try clauses in a random order for fairness.")
},
{
"ev/close", janet_cfun_stream_close,
JDOC("(ev/close stream)\n\n"
"Close a stream. This should be the same as calling (:close stream) for all streams.")
},
{
"ev/read", janet_cfun_stream_read,
JDOC("(ev/read stream n &opt buffer timeout)\n\n"
"Read up to n bytes into a buffer asynchronously from a stream. `n` can also be the keyword "
"`:all` to read into the buffer until end of stream. "
"Optionally provide a buffer to write into "
"as well as a timeout in seconds after which to cancel the operation and raise an error. "
"Returns the buffer if the read was successful or nil if end-of-stream reached. Will raise an "
"error if there are problems with the IO operation.")
},
{
"ev/chunk", janet_cfun_stream_chunk,
JDOC("(ev/chunk stream n &opt buffer timeout)\n\n"
"Same as ev/read, but will not return early if less than n bytes are available. If an end of "
"stream is reached, will also return early with the collected bytes.")
},
{
"ev/write", janet_cfun_stream_write,
JDOC("(ev/write stream data &opt timeout)\n\n"
"Write data to a stream, suspending the current fiber until the write "
"completes. Takes an optional timeout in seconds, after which will return nil. "
"Returns nil, or raises an error if the write failed.")
},
{NULL, NULL, NULL}
};
void janet_lib_ev(JanetTable *env) { void janet_lib_ev(JanetTable *env) {
janet_core_cfuns(env, NULL, ev_cfuns); JanetRegExt ev_cfuns_ext[] = {
JANET_CORE_REG("ev/give", cfun_channel_push),
JANET_CORE_REG("ev/take", cfun_channel_pop),
JANET_CORE_REG("ev/full", cfun_channel_full),
JANET_CORE_REG("ev/capacity", cfun_channel_capacity),
JANET_CORE_REG("ev/count", cfun_channel_count),
JANET_CORE_REG("ev/select", cfun_channel_choice),
JANET_CORE_REG("ev/rselect", cfun_channel_rchoice),
JANET_CORE_REG("ev/chan", cfun_channel_new),
JANET_CORE_REG("ev/go", cfun_ev_go),
JANET_CORE_REG("ev/thread", cfun_ev_thread),
JANET_CORE_REG("ev/give-supervisor", cfun_ev_give_supervisor),
JANET_CORE_REG("ev/sleep", cfun_ev_sleep),
JANET_CORE_REG("ev/deadline", cfun_ev_deadline),
JANET_CORE_REG("ev/cancel", cfun_ev_cancel),
JANET_CORE_REG("ev/close", janet_cfun_stream_close),
JANET_CORE_REG("ev/read", janet_cfun_stream_read),
JANET_CORE_REG("ev/chunk", janet_cfun_stream_chunk),
JANET_CORE_REG("ev/write", janet_cfun_stream_write),
JANET_REG_END
};
janet_core_cfuns_ext(env, NULL, ev_cfuns_ext);
janet_register_abstract_type(&janet_stream_type); janet_register_abstract_type(&janet_stream_type);
} }

8
src/include/janet.h
View File

@ -1765,26 +1765,26 @@ JANET_API Janet janet_resolve_core(const char *name);
/* no docstrings or sourcemaps */ /* no docstrings or sourcemaps */
#define JANET_REG_(JNAME, CNAME) {JNAME, CNAME, NULL, NULL, 0} #define JANET_REG_(JNAME, CNAME) {JNAME, CNAME, NULL, NULL, 0}
#define JANET_FN_(CNAME, USAGE, DOCSTRING) \ #define JANET_FN_(CNAME, USAGE, DOCSTRING) \
static Janet CNAME (int32_t argc, Janet *argv) Janet CNAME (int32_t argc, Janet *argv)
/* sourcemaps only */ /* sourcemaps only */
#define JANET_REG_S(JNAME, CNAME) {JNAME, CNAME, NULL, __FILE__, CNAME##_sourceline_} #define JANET_REG_S(JNAME, CNAME) {JNAME, CNAME, NULL, __FILE__, CNAME##_sourceline_}
#define JANET_FN_S(CNAME, USAGE, DOCSTRING) \ #define JANET_FN_S(CNAME, USAGE, DOCSTRING) \
static int32_t CNAME##_sourceline_ = __LINE__; \ static int32_t CNAME##_sourceline_ = __LINE__; \
static Janet CNAME (int32_t argc, Janet *argv) Janet CNAME (int32_t argc, Janet *argv)
/* docstring only */ /* docstring only */
#define JANET_REG_D(JNAME, CNAME) {JNAME, CNAME, CNAME##_docstring_, NULL, 0} #define JANET_REG_D(JNAME, CNAME) {JNAME, CNAME, CNAME##_docstring_, NULL, 0}
#define JANET_FN_D(CNAME, USAGE, DOCSTRING) \ #define JANET_FN_D(CNAME, USAGE, DOCSTRING) \
static const char CNAME##_docstring_[] = USAGE "\n\n" DOCSTRING; \ static const char CNAME##_docstring_[] = USAGE "\n\n" DOCSTRING; \
static Janet CNAME (int32_t argc, Janet *argv) Janet CNAME (int32_t argc, Janet *argv)
/* sourcemaps and docstrings */ /* sourcemaps and docstrings */
#define JANET_REG_SD(JNAME, CNAME) {JNAME, CNAME, CNAME##_docstring_, __FILE__, CNAME##_sourceline_} #define JANET_REG_SD(JNAME, CNAME) {JNAME, CNAME, CNAME##_docstring_, __FILE__, CNAME##_sourceline_}
#define JANET_FN_SD(CNAME, USAGE, DOCSTRING) \ #define JANET_FN_SD(CNAME, USAGE, DOCSTRING) \
static int32_t CNAME##_sourceline_ = __LINE__; \ static int32_t CNAME##_sourceline_ = __LINE__; \
static const char CNAME##_docstring_[] = USAGE "\n\n" DOCSTRING; \ static const char CNAME##_docstring_[] = USAGE "\n\n" DOCSTRING; \
static Janet CNAME (int32_t argc, Janet *argv) Janet CNAME (int32_t argc, Janet *argv)
/* Choose defaults for source mapping and docstring based on config defs */ /* Choose defaults for source mapping and docstring based on config defs */
#if defined(JANET_NO_SOURCEMAPS) && defined(JANET_NO_DOCSTRINGS) #if defined(JANET_NO_SOURCEMAPS) && defined(JANET_NO_DOCSTRINGS)