Tasking support

.github/workflows/ci.yml

@@ -14,6 +14,7 @@ on:
   push:
     paths:
       - 'src/**'
+      - 'test/**'
       - 'scripts/**'
       - 'deps/build.jl'
       - 'deps/buildtools/**'
@@ -31,6 +32,7 @@ on:
   pull_request:
     paths:
       - 'src/**'
+      - 'test/**'
       - 'scripts/**'
       - 'deps/build.jl'
       - 'deps/buildtools/**'
@@ -102,7 +104,7 @@ jobs:
           LEGATE_SHOW_CONFIG: "1"
           LEGATE_AUTO_CONFIG: "0"
           GPUTESTS: "0" # parsed by runtests.jl
-          LEGATE_CONFIG: "--cpus 1 --utility 1 --sysmem 4000"
+          LEGATE_CONFIG: "--cpus 2 --utility 1 --sysmem 4000"
         run: |
           if julia -e 'exit(VERSION >= v"1.12" ? 0 : 1)'; then
             export JULIA_NUM_THREADS=1,0

.github/workflows/developer.yml

@@ -79,5 +79,13 @@ jobs:
           julia --color=yes -e 'using Pkg; Pkg.build("Legate")'
 
       - name: Perform Test
+        env:
+          LEGATE_CONFIG: "--cpus 2 --gpus 0 --utility 1 --sysmem 4000"
         run: |
-          julia --color=yes -e 'using Pkg; Pkg.test("Legate")'
+          julia --color=yes -e 'using Pkg; Pkg.test("Legate"; julia_args=["--threads=4,1"])'
+
+      - name: CPU tasking stress test
+        env:
+          LEGATE_CONFIG: "--cpus 4 --gpus 0 --utility 1 --sysmem 4000"
+        run: |
+          timeout 180s julia --color=yes --threads=4,1 test/cpu_tasking_stress.jl

.gitignore

@@ -13,6 +13,8 @@ build.log
 .vscode
 node_modules
 package-lock.json
+test_crash.sh
+*.prof
 
 _doxygen/
 _doxygen/**

Project.toml

@@ -18,7 +18,14 @@ legate_jl_wrapper_jll = "6e093c14-372f-5516-b4d7-a6ce14f1038b"
 legate_jll = "e95fb1d3-fb9e-51b5-bdb8-1a812408cac9"
 libcxxwrap_julia_jll = "3eaa8342-bff7-56a5-9981-c04077f7cee7"
 
+[weakdeps]
+CUDA = "052768ef-5323-5732-b1bb-66c8b64840ba"
+
+[extensions]
+CUDAExt = "CUDA"
+
 [compat]
+CUDA = "6.4"
 CUDACore = "6.4"
 CxxWrap = "0.17.5"
 FunctionWrappers = "1.1.3"

deps/build.jl

@@ -28,8 +28,8 @@ function build_cpp_wrapper(
 )
     @info "liblegatewrapper: Building C++ Wrapper Library"
     isdir(install_root) && (rm(install_root; recursive=true); mkdir(install_root))
-    bld_command = `$(joinpath(repo_root, "scripts/build_cpp_wrapper.sh")) $repo_root $legate_root $install_root $(Threads.nthreads())`
-    BuildTools.run_build_wrapper_script(
+    bld_command = `$(joinpath(repo_root, "scripts/build_cpp_wrapper.sh")) $repo_root $legate_root $install_root $(Sys.CPU_THREADS)`
+    return BuildTools.run_build_wrapper_script(
         repo_root, bld_command; cuda_root, cuda_enabled, log_dir=@__DIR__
     )
 end
@@ -49,7 +49,7 @@ function build_deps(pkg_root, legate_root; cuda_root=nothing, cuda_enabled=true)
         log_dir=@__DIR__, is_compatible=is_supported_version,
     )
     build_cpp_wrapper(pkg_root, legate_root, install_dir; cuda_root, cuda_enabled)
-    BuildTools.set_jll_artifact_override(:legate_jl_wrapper_jll, install_dir)
+    return BuildTools.set_jll_artifact_override(:legate_jl_wrapper_jll, install_dir)
 end
 
 function build(::LegatePreferences.JLL)
@@ -67,7 +67,7 @@ function build(::LegatePreferences.Conda)
     end
 
     is_legate_installed(legate_root; throw_errors=true)
-    build_deps(pkg_root, legate_root)
+    return build_deps(pkg_root, legate_root)
 end
 
 function build(::LegatePreferences.Developer)
@@ -83,7 +83,9 @@ function build(::LegatePreferences.Developer)
     end
 
     build_deps(pkg_root, legate_root; cuda_root, cuda_enabled)
-    set_preferences!(LegatePreferences, "LEGATE_LIBDIR" => joinpath(legate_root, "lib"); force=true)
+    return set_preferences!(
+        LegatePreferences, "LEGATE_LIBDIR" => joinpath(legate_root, "lib"); force=true
+    )
 end
 
 const mode_str = load_preference(LegatePreferences, "legate_mode", LegatePreferences.MODE_JLL)

docs/src/example.md

@@ -5,9 +5,12 @@
 Legate uses a deferred execution model. When a task is submitted in Julia, it is not executed immediately. Instead, the Legate runtime (C++) records the task, analyzes data dependencies, and schedules execution.
 
 **Interaction Model:**
-1. **Submission**: The main Julia thread submits tasks to the runtime. This is non-blocking.
-2. **Scheduling**: The Legate runtime manages resources and dependencies.
-3. **Execution**: Once ready, Legate signals Julia to execute the task. This happens on a dedicated Julia worker task (thread) that handles incoming requests from the runtime. See more information about Julia thread-safety [here](https://docs.julialang.org/en/v1/manual/calling-c-and-fortran-code/#Thread-safety).
+1. **Submission**: The main Julia thread submits tasks to the runtime via non-blocking C calls.
+2. **Scheduling**: The Legate runtime manages resources and dependencies in the background.
+3. **Execution**:
+    *   **Polling**: A low-overhead timer on the main thread polls for ready tasks from the runtime.
+    *   **Workers**: Tasks are dispatched via a thread-safe `Channel` to a pool of background worker threads (default: `nthreads-1`) to execute concurrently. This ensures that the main thread is never blocked by long-running tasks, and the runtime is not blocked by Julia's GC or other operations.
+    *   **Completion**: Workers signal completion back to the runtime upon finishing. See more information about Julia thread-safety [here](https://docs.julialang.org/en/v1/manual/calling-c-and-fortran-code/#Thread-safety).
 
 ## Arguments
 

examples/minimal_task.jl

@@ -0,0 +1,20 @@
+using Legate
+
+Legate.Experimental(true)  # tasking is experimental
+
+function task_noop(a)
+    return nothing
+end
+
+Legate.ensure_runtime!()
+rt = Legate.get_runtime()
+lib = Legate.create_library("test")
+
+my_noop_task = Legate.wrap_task(task_noop, Legate.CPUBackend)
+a_noop = Legate.create_array([10], Float32)
+
+task0 = Legate.create_julia_task(rt, lib, my_noop_task)
+Legate.add_output(task0, a_noop)
+Legate.submit_task(rt, task0)
+
+Legate.wait_ufi()

examples/tasking.jl

@@ -1,65 +1,67 @@
 using Legate
 
-# args is a vector that can be expanded
-# it is in the order that the inputs and outputs are added
-# in1 in2 ... out1 out2 ... scalar1 scalar2 ...
-function task_test(args)
-    a, b, c = args # inputs are first, then outputs
+function task_test(a, b, c)
     @inbounds @simd for i in eachindex(a)
         c[i] = a[i] + b[i]
     end
 end
 
-# init task
-function task_init(args)
-    a, b, c = args
+function task_init(a, b, c)
     @inbounds @simd for i in eachindex(a)
         a[i] = rand(Float32)
         b[i] = rand(Float32)
         c[i] = 0.0f0
     end
 end
 
-# 4 arg task
-function task_4arg(args)
-    in1, in2, out1, out2 = args
+function task_4arg(in1, in2, out1, out2)
     @inbounds @simd for i in eachindex(in1)
         out1[i] = in1[i] * 2
         out2[i] = in2[i] + 1
     end
 end
 
-# Task with Scalar argument
-function task_scalar(args)
-    a, b, scalar = args
+function task_scalar(a, b, scalar)
     @inbounds @simd for i in eachindex(a)
         b[i] = a[i] * scalar
     end
 end
 
+function task_noop(a)
+    return nothing
+end
+
 function test_driver()
+    Legate.Experimental(true)  # tasking is experimental
+    N = 1000
     rt = Legate.get_runtime()
     lib = Legate.create_library("test")
 
-    my_task = Legate.wrap_task(task_test)
-    my_init_task = Legate.wrap_task(task_init)
-    my_4arg_task = Legate.wrap_task(task_4arg)
-    my_scalar_task = Legate.wrap_task(task_scalar)
-
-    # 1. Init Task (3 args)
-    a = Legate.create_array([10, 10], Float32)
-    b = Legate.create_array([10, 10], Float32)
-    c = Legate.create_array([10, 10], Float32)
-    d = Legate.create_array([10, 10], Float32) # Extra array for 4-arg test
-
-    task = Legate.create_julia_task(rt, lib, my_init_task)
+    my_task = Legate.wrap_task(task_test, Legate.CPUBackend)
+    my_init_task = Legate.wrap_task(task_init, Legate.CPUBackend)
+    my_4arg_task = Legate.wrap_task(task_4arg, Legate.CPUBackend)
+    my_scalar_task = Legate.wrap_task(task_scalar, Legate.CPUBackend)
+    my_noop_task = Legate.wrap_task(task_noop, Legate.CPUBackend)
+
+    # 0. NOOP Task
+    a_noop = Legate.create_array([10], Float32)
+    task0 = Legate.create_julia_task(rt, lib, my_noop_task)
+    Legate.add_output(task0, a_noop)
+    Legate.submit_task(rt, task0)
+
+    # 1. Initialization Task
+    a = Legate.create_array([N], Float32)
+    b = Legate.create_array([N], Float32)
+    c = Legate.create_array([N], Float32)
+    d = Legate.create_array([N], Float32)
+
+    task1 = Legate.create_julia_task(rt, lib, my_init_task)
     init_output_vars = Vector{Legate.Variable}()
-    push!(init_output_vars, Legate.add_output(task, a))
-    push!(init_output_vars, Legate.add_output(task, b))
-    push!(init_output_vars, Legate.add_output(task, c))
-    Legate.default_alignment(task, Vector{Legate.Variable}(), init_output_vars)
-
-    Legate.submit_task(rt, task)
+    push!(init_output_vars, Legate.add_output(task1, a))
+    push!(init_output_vars, Legate.add_output(task1, b))
+    push!(init_output_vars, Legate.add_output(task1, c))
+    Legate.default_alignment(task1, Vector{Legate.Variable}(), init_output_vars)
+    Legate.submit_task(rt, task1)
 
     # 2. Compute Task (3 args)
     task2 = Legate.create_julia_task(rt, lib, my_task)
@@ -69,37 +71,50 @@ function test_driver()
     push!(input_vars, Legate.add_input(task2, b))
     push!(output_vars, Legate.add_output(task2, c))
     Legate.default_alignment(task2, input_vars, output_vars)
-
     Legate.submit_task(rt, task2)
 
     # 3. Arbitrary Arg Task (4 args: 2 in, 2 out)
     task3 = Legate.create_julia_task(rt, lib, my_4arg_task)
     in_vars_4 = Vector{Legate.Variable}()
     out_vars_4 = Vector{Legate.Variable}()
-    # Inputs: a, c
     push!(in_vars_4, Legate.add_input(task3, a))
     push!(in_vars_4, Legate.add_input(task3, c))
-    # Outputs: b (reuse), d (new)
     push!(out_vars_4, Legate.add_output(task3, b))
     push!(out_vars_4, Legate.add_output(task3, d))
     Legate.default_alignment(task3, in_vars_4, out_vars_4)
-
     Legate.submit_task(rt, task3)
 
     # 4. Scalar Arg Task (2 args + scalar)
     task4 = Legate.create_julia_task(rt, lib, my_scalar_task)
     in_vars_s = Vector{Legate.Variable}()
     out_vars_s = Vector{Legate.Variable}()
-    # Input: c (result of task2)
     push!(in_vars_s, Legate.add_input(task4, c))
-    # Output: a (reuse)
     push!(out_vars_s, Legate.add_output(task4, a))
-
-    # Add user scalar argument (Float32)
     Legate.add_scalar(task4, Legate.Scalar(2.5f0))
     Legate.default_alignment(task4, in_vars_s, out_vars_s)
-
     Legate.submit_task(rt, task4)
+
+    # --- VERIFICATION OF ASYNC EXECUTION ---
+    @info "Submitting 100 bulk tasks to verify asynchronous polling..."
+    for i in 1:100
+        t_bulk = Legate.create_julia_task(rt, lib, my_scalar_task)
+        v_in = Legate.add_input(t_bulk, c)
+        v_out = Legate.add_output(t_bulk, a)
+        Legate.add_scalar(t_bulk, Legate.Scalar(1.0f0))
+        Legate.default_alignment(t_bulk, [v_in], [v_out])
+        Legate.submit_task(rt, t_bulk)
+    end
+    @info "Bulk submission complete. Waiting for completion..."
+    Legate.wait_ufi()
+    @info "Verifying results of bulk run..."
+    val_a = Array(a)
+    val_c = Array(c)
+    if val_a ≈ val_c
+        @info "Verification successful: Target array matches source."
+    else
+        @error "Verification FAILED: Target array does not match source."
+        error("Parallel verification failed")
+    end
 end
 
 if abspath(PROGRAM_FILE) == @__FILE__

ext/CUDAExt/CUDAExt.jl

@@ -1,13 +1,8 @@
 module CUDAExt
 
-# using CUDA
-# using Legate
+using CUDA
+using Legate
 
-# using CxxWrap: CxxWrap
-# import Legate: wrap_task, create_julia_task, SUPPORTED_TYPES, JuliaGPUTask, CxxPtr, Runtime,
-#     Library, create_task, JULIA_CUSTOM_GPU_TASK, add_scalar, Scalar, register_task_function,
-#     _execute_julia_task, get_code_type, TaskArgumentGPU
-
-# include("ufi.jl")
+include("ufi.jl")
 
 end # module CUDAExt