diff --git a/src/internal/event_loop/event_loop.js.mbt b/src/internal/event_loop/event_loop.js.mbt index 17fe6835e..ec4382254 100644 --- a/src/internal/event_loop/event_loop.js.mbt +++ b/src/internal/event_loop/event_loop.js.mbt @@ -18,6 +18,7 @@ let _ignore_unused_import : Unit = { ignore(@fd_util.unimplemented) ignore(@os_string.unimplemented) ignore(@os_error.unimplemented) + let _ : (@job_types.Job) -> Unit = ignore let _ : (&@external_loop_integration.ExternalEventLoop) -> Unit = ignore ignore(@env_util.eprintln) } diff --git a/src/internal/event_loop/moon.pkg b/src/internal/event_loop/moon.pkg index 210082966..20c24c1b8 100644 --- a/src/internal/event_loop/moon.pkg +++ b/src/internal/event_loop/moon.pkg @@ -11,6 +11,7 @@ import { "moonbitlang/async/internal/c_buffer", "moonbitlang/async/internal/os_string", "moonbitlang/async/internal/env_util", + "moonbitlang/async/job/types" @job_types, "moonbitlang/async/external_loop_integration", } diff --git a/src/internal/event_loop/pkg.generated.mbti b/src/internal/event_loop/pkg.generated.mbti index e68419f64..1b0774d29 100644 --- a/src/internal/event_loop/pkg.generated.mbti +++ b/src/internal/event_loop/pkg.generated.mbti @@ -7,6 +7,7 @@ import { "moonbitlang/async/internal/env_util", "moonbitlang/async/internal/fd_util", "moonbitlang/async/internal/os_string", + "moonbitlang/async/job/types", "moonbitlang/core/ref", } @@ -101,6 +102,8 @@ pub fn terminate_process_by_signal(Int) -> Unit pub fn with_event_loop(async () -> Unit, max_worker_count? : Int) -> Unit raise +pub async fn[X] with_native_job(@types.Job, (Int) -> X raise, context~ : String) -> X + // Errors pub suberror KilledBySignal { KilledBySignal(Int) diff --git a/src/internal/event_loop/thread_pool.mbt b/src/internal/event_loop/thread_pool.mbt index 08fbab180..1dcc00fc3 100644 --- a/src/internal/event_loop/thread_pool.mbt +++ b/src/internal/event_loop/thread_pool.mbt @@ -61,8 +61,19 @@ extern "C" fn fetch_completion_ffi( ) -> Int = "moonbitlang_async_fetch_completion" ///| -#external -priv type Job +type Job = @job_types.Job + +///| +/// Schedule a native binding Job while keeping it alive through `finish`. +pub async fn[X] with_native_job( + job : Job, + finish : (Int) -> X raise, + context~ : String, +) -> X { + defer job.free() + let ret = perform_job_in_worker(job, context~) + finish(ret) +} ///| extern "C" fn Job::free(self : Job) -> Unit = "moonbitlang_async_free_job" diff --git a/src/internal/event_loop/thread_pool.wasm.mbt b/src/internal/event_loop/thread_pool.wasm.mbt index fc33b73ef..92a91bca7 100644 --- a/src/internal/event_loop/thread_pool.wasm.mbt +++ b/src/internal/event_loop/thread_pool.wasm.mbt @@ -78,7 +78,7 @@ fn fetch_completion_ffi( ) -> Int = "moonbitlang/async" "thread_pool/fetch_completion/unix" ///| -/// The actual job object submitted to the thread pool in runtime +/// The actual job object submitted to the thread pool in runtime. priv struct JobHandle(UInt64) ///| @@ -100,6 +100,23 @@ fn Job::Job(handle : JobHandle, copy_output? : (JobHandle) -> Unit) -> Job { { handle, copy_output } } +///| +/// Schedule a runtime-owned Job while keeping it alive through `finish`. +pub async fn[X] with_wasm_job( + runtime_job : @job_types.Job, + finish : (Int) -> X raise, + context~ : String, +) -> X { + let job = Job(JobHandle(runtime_job.into_runtime_handle())) + defer job.free() + let ret = perform_job_in_worker(job, context~) + finish(ret) +} + +///| +/// Lower the public Wasm Job to the runtime's integer Handle representation. +fn @job_types.Job::into_runtime_handle(self : @job_types.Job) -> UInt64 = "%identity" + ///| #unsafe_skip_stub_check fn JobHandle::free(self : JobHandle) -> Unit = "moonbitlang/async" "thread_pool/free_job" diff --git a/src/internal/event_loop/unimplemented.mbt b/src/internal/event_loop/unimplemented.mbt index a77e7d3fb..35058ce3e 100644 --- a/src/internal/event_loop/unimplemented.mbt +++ b/src/internal/event_loop/unimplemented.mbt @@ -21,6 +21,7 @@ let _ignore_unused_import : Unit = { ignore(@os_error.unimplemented) ignore(@coroutine.is_being_cancelled) ignore(@env_util.eprintln) + let _ : (@job_types.Job) -> Unit = ignore let _ : (&@external_loop_integration.ExternalEventLoop) -> Unit = ignore } diff --git a/src/job/internal/job_test/job_native_test.mbt b/src/job/internal/job_test/job_native_test.mbt new file mode 100644 index 000000000..3eddea24f --- /dev/null +++ b/src/job/internal/job_test/job_native_test.mbt @@ -0,0 +1,28 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +#unsafe_skip_stub_check +extern "C" fn make_external_test_job(value : Int) -> @job.Job = "moonbitlang_async_make_external_test_job" + +///| +async test "package-owned native Job uses the shared worker scheduler" { + assert_false(@async.is_being_cancelled()) + let ret = @job.with_job( + make_external_test_job(42), + (_, ret) => ret, + context="package-owned native Job", + ) + inspect(ret, content="42") +} diff --git a/src/job/internal/job_test/job_wasm_test.mbt b/src/job/internal/job_test/job_wasm_test.mbt new file mode 100644 index 000000000..c65a9d7e6 --- /dev/null +++ b/src/job/internal/job_test/job_wasm_test.mbt @@ -0,0 +1,28 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +#unsafe_skip_stub_check +fn make_sleep_job(milliseconds : Int) -> UInt64 = "moonbitlang/async" "thread_pool/make_sleep_job" + +///| +async test "runtime-owned Wasm Job uses the shared worker scheduler" { + assert_false(@async.is_being_cancelled()) + let ret = @job.with_job( + @job.Job::from_runtime_handle(make_sleep_job(1)), + (_, ret) => ret, + context="runtime-owned Wasm Job", + ) + inspect(ret, content="0") +} diff --git a/src/job/internal/job_test/moon.pkg b/src/job/internal/job_test/moon.pkg new file mode 100644 index 000000000..85f6f1986 --- /dev/null +++ b/src/job/internal/job_test/moon.pkg @@ -0,0 +1,14 @@ +import { + "moonbitlang/async", + "moonbitlang/async/job", +} for "test" + +supported_targets = "-all+native+wasm" + +options( + "native-stub": [ "test_job.c" ], + targets: { + "job_native_test.mbt": [ "native" ], + "job_wasm_test.mbt": [ "wasm" ], + }, +) diff --git a/src/job/internal/job_test/pkg.generated.mbti b/src/job/internal/job_test/pkg.generated.mbti new file mode 100644 index 000000000..eba49e0d2 --- /dev/null +++ b/src/job/internal/job_test/pkg.generated.mbti @@ -0,0 +1,12 @@ +// Generated using `moon info`, DON'T EDIT IT +package "moonbitlang/async/job/internal/job_test" + +// Values + +// Errors + +// Types and methods + +// Type aliases + +// Traits diff --git a/src/job/internal/job_test/test_job.c b/src/job/internal/job_test/test_job.c new file mode 100644 index 000000000..97215f148 --- /dev/null +++ b/src/job/internal/job_test/test_job.c @@ -0,0 +1,54 @@ +/* + * Copyright 2025 International Digital Economy Academy + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include + +void *moonbitlang_async_make_job( + int32_t size, + void (*free_job)(void *), + int32_t (*worker)(void *, int32_t *), + int32_t (*cancel_handler)(void *) +); + +struct external_test_job { + int32_t value; +}; + +static void +free_external_test_job(void *payload) { + (void)payload; +} + +static int32_t +run_external_test_job(void *payload, int32_t *error) { + struct external_test_job *job = payload; + *error = 0; + return job->value; +} + +MOONBIT_FFI_EXPORT +void * +moonbitlang_async_make_external_test_job(int32_t value) { + struct external_test_job *job = moonbitlang_async_make_job( + sizeof(struct external_test_job), + free_external_test_job, + run_external_test_job, + NULL + ); + job->value = value; + return job; +} diff --git a/src/job/job.native.mbt b/src/job/job.native.mbt new file mode 100644 index 000000000..cadba46d8 --- /dev/null +++ b/src/job/job.native.mbt @@ -0,0 +1,30 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +pub using @job_types {type Job} + +///| +/// Submit a native binding Job to the shared blocking worker scheduler. +/// +/// Host or system errors raised by the worker skip `finish`. Otherwise, +/// `finish` runs on the event-loop thread while `job` and its structured result +/// remain valid. The Job is freed after `finish` returns or raises. +pub async fn[X] with_job( + job : @job_types.Job, + finish : (@job_types.Job, Int) -> X raise, + context~ : String, +) -> X { + @event_loop.with_native_job(job, ret => finish(job, ret), context~) +} diff --git a/src/job/job.wasm.mbt b/src/job/job.wasm.mbt new file mode 100644 index 000000000..38b3a6958 --- /dev/null +++ b/src/job/job.wasm.mbt @@ -0,0 +1,30 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +pub using @job_types {type Job} + +///| +/// Submit a runtime Job to the shared blocking worker scheduler. +/// +/// Host or system errors raised by the worker skip `finish`. Otherwise, +/// `finish` runs on the event-loop thread while `job` and its structured result +/// remain valid. The Job is freed after `finish` returns or raises. +pub async fn[X] with_job( + job : @job_types.Job, + finish : (@job_types.Job, Int) -> X raise, + context~ : String, +) -> X { + @event_loop.with_wasm_job(job, ret => finish(job, ret), context~) +} diff --git a/src/job/moon.pkg b/src/job/moon.pkg new file mode 100644 index 000000000..2083bc834 --- /dev/null +++ b/src/job/moon.pkg @@ -0,0 +1,10 @@ +import { + "moonbitlang/async/internal/event_loop", + "moonbitlang/async/job/types" @job_types, +} + +supported_targets = "-all+native+wasm" + +options( + targets: { "job.native.mbt": [ "native" ], "job.wasm.mbt": [ "wasm" ] }, +) diff --git a/src/job/pkg.generated.mbti b/src/job/pkg.generated.mbti new file mode 100644 index 000000000..08479b492 --- /dev/null +++ b/src/job/pkg.generated.mbti @@ -0,0 +1,18 @@ +// Generated using `moon info`, DON'T EDIT IT +package "moonbitlang/async/job" + +import { + "moonbitlang/async/job/types", +} + +// Values +pub async fn[X] with_job(@types.Job, (@types.Job, Int) -> X raise, context~ : String) -> X + +// Errors + +// Types and methods + +// Type aliases +pub using @types {type Job} + +// Traits diff --git a/src/job/types/job.native.mbt b/src/job/types/job.native.mbt new file mode 100644 index 000000000..41c873289 --- /dev/null +++ b/src/job/types/job.native.mbt @@ -0,0 +1,22 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +/// An opaque blocking operation created by a native binding package. +/// +/// A package's C stub allocates its payload with +/// `moonbitlang_async_make_job` and returns that payload from an `extern "C"` +/// function declared with this type. +#external +pub type Job diff --git a/src/job/types/job.wasm.mbt b/src/job/types/job.wasm.mbt new file mode 100644 index 000000000..8fad40fff --- /dev/null +++ b/src/job/types/job.wasm.mbt @@ -0,0 +1,29 @@ +// Copyright 2025 International Digital Economy Academy +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +///| +/// An opaque blocking operation created by the Wasm host runtime. +#warnings("-struct_never_constructed-unused_field") +#valtype +pub struct Job { + priv handle : UInt64 +} + +///| +/// Wrap a Job Handle returned by the active Wasm host runtime. +/// +/// The Handle must identify an unsubmitted Job owned by that runtime. +pub fn Job::from_runtime_handle(handle : UInt64) -> Job { + { handle, } +} diff --git a/src/job/types/moon.pkg b/src/job/types/moon.pkg new file mode 100644 index 000000000..8b8fefa3d --- /dev/null +++ b/src/job/types/moon.pkg @@ -0,0 +1,6 @@ +options( + targets: { + "job.native.mbt": [ "native" ], + "job.wasm.mbt": [ "wasm", "wasm-gc", "js" ], + }, +) diff --git a/src/job/types/pkg.generated.mbti b/src/job/types/pkg.generated.mbti new file mode 100644 index 000000000..450b45aca --- /dev/null +++ b/src/job/types/pkg.generated.mbti @@ -0,0 +1,14 @@ +// Generated using `moon info`, DON'T EDIT IT +package "moonbitlang/async/job/types" + +// Values + +// Errors + +// Types and methods +#external +pub type Job + +// Type aliases + +// Traits