EagerTaskSpec->DTaskSpec, EagerTaskQueue->DefaultTaskQueue

Author: jpsamaroo

docs/src/task-queues.md

@@ -14,7 +14,7 @@ queues".
 A task queue in Dagger is an object that can be configured to accept unlaunched
 tasks from `@spawn`/`spawn` and either modify them or delay their launching
 arbitrarily. By default, Dagger tasks are enqueued through the
-`EagerTaskQueue`, which submits tasks directly into the scheduler before
+`DefaultTaskQueue`, which submits tasks directly into the scheduler before
 `@spawn`/`spawn` returns. However, Dagger also has an `InOrderTaskQueue`, which
 ensures that tasks enqueued through it execute sequentially with respect to
 each other. This queue can be allocated with `Dagger.spawn_sequential`:

src/datadeps.jl

@@ -26,7 +26,7 @@ struct DataDepsTaskQueue <: AbstractTaskQueue
     # The fields following only apply when static==true
     static::Bool
     # The set of tasks that have already been seen
-    seen_tasks::Union{Vector{Pair{EagerTaskSpec,DTask}},Nothing}
+    seen_tasks::Union{Vector{Pair{DTaskSpec,DTask}},Nothing}
     # The data-dependency graph of all tasks
     g::Union{SimpleDiGraph{Int},Nothing}
     # The mapping from task to graph ID
@@ -38,7 +38,7 @@ struct DataDepsTaskQueue <: AbstractTaskQueue
                                traversal::Symbol=:inorder)
         deps = IdDict{Any, Vector{Pair{Tuple{Bool,Bool}, DTask}}}()
         if static
-            seen_tasks = Pair{EagerTaskSpec,DTask}[]
+            seen_tasks = Pair{DTaskSpec,DTask}[]
             g = SimpleDiGraph()
             task_to_id = Dict{DTask,Int}()
         else
@@ -51,7 +51,7 @@ struct DataDepsTaskQueue <: AbstractTaskQueue
     end
 end
 
-function _enqueue!(queue::DataDepsTaskQueue, fullspec::Pair{EagerTaskSpec,DTask})
+function _enqueue!(queue::DataDepsTaskQueue, fullspec::Pair{DTaskSpec,DTask})
     # If static, record this task and its edges in the graph
     if queue.static
         g = queue.g
@@ -139,15 +139,15 @@ function _enqueue!(queue::DataDepsTaskQueue, fullspec::Pair{EagerTaskSpec,DTask}
         spec.options = merge(opts, (;syncdeps, scope))
     end
 end
-function enqueue!(queue::DataDepsTaskQueue, spec::Pair{EagerTaskSpec,DTask})
+function enqueue!(queue::DataDepsTaskQueue, spec::Pair{DTaskSpec,DTask})
     _enqueue!(queue, spec)
     if queue.static
         push!(queue.seen_tasks, spec)
     else
         enqueue!(queue.upper_queue, spec)
     end
 end
-function enqueue!(queue::DataDepsTaskQueue, specs::Vector{Pair{EagerTaskSpec,DTask}})
+function enqueue!(queue::DataDepsTaskQueue, specs::Vector{Pair{DTaskSpec,DTask}})
     for spec in specs
         _enqueue!(queue, spec)
     end
@@ -467,7 +467,7 @@ is experimental and subject to change.
 function spawn_datadeps(f::Base.Callable; static::Bool=true,
                         traversal::Symbol=:inorder)
     wait_all(; check_errors=true) do
-        queue = DataDepsTaskQueue(get_options(:task_queue, EagerTaskQueue());
+        queue = DataDepsTaskQueue(get_options(:task_queue, DefaultTaskQueue());
                                   static, traversal)
         result = with_options(f; task_queue=queue)
         if queue.static

src/queue.jl

@@ -1,4 +1,4 @@
-mutable struct EagerTaskSpec
+mutable struct DTaskSpec
     f
     args::Vector{Pair{Union{Symbol,Nothing},Any}}
     options::NamedTuple
@@ -8,25 +8,25 @@ abstract type AbstractTaskQueue end
 
 function enqueue! end
 
-struct EagerTaskQueue <: AbstractTaskQueue end
-enqueue!(::EagerTaskQueue, spec::Pair{EagerTaskSpec,DTask}) =
+struct DefaultTaskQueue <: AbstractTaskQueue end
+enqueue!(::DefaultTaskQueue, spec::Pair{DTaskSpec,DTask}) =
     eager_launch!(spec)
-enqueue!(::EagerTaskQueue, specs::Vector{Pair{EagerTaskSpec,DTask}}) =
+enqueue!(::DefaultTaskQueue, specs::Vector{Pair{DTaskSpec,DTask}}) =
     eager_launch!(specs)
 
-enqueue!(spec::Pair{EagerTaskSpec,DTask}) =
-    enqueue!(get_options(:task_queue, EagerTaskQueue()), spec)
-enqueue!(specs::Vector{Pair{EagerTaskSpec,DTask}}) =
-    enqueue!(get_options(:task_queue, EagerTaskQueue()), specs)
+enqueue!(spec::Pair{DTaskSpec,DTask}) =
+    enqueue!(get_options(:task_queue, DefaultTaskQueue()), spec)
+enqueue!(specs::Vector{Pair{DTaskSpec,DTask}}) =
+    enqueue!(get_options(:task_queue, DefaultTaskQueue()), specs)
 
 struct LazyTaskQueue <: AbstractTaskQueue
-    tasks::Vector{Pair{EagerTaskSpec,DTask}}
-    LazyTaskQueue() = new(Pair{EagerTaskSpec,DTask}[])
+    tasks::Vector{Pair{DTaskSpec,DTask}}
+    LazyTaskQueue() = new(Pair{DTaskSpec,DTask}[])
 end
-function enqueue!(queue::LazyTaskQueue, spec::Pair{EagerTaskSpec,DTask})
+function enqueue!(queue::LazyTaskQueue, spec::Pair{DTaskSpec,DTask})
     push!(queue.tasks, spec)
 end
-function enqueue!(queue::LazyTaskQueue, specs::Vector{Pair{EagerTaskSpec,DTask}})
+function enqueue!(queue::LazyTaskQueue, specs::Vector{Pair{DTaskSpec,DTask}})
     append!(queue.tasks, specs)
 end
 function spawn_bulk(f::Base.Callable)
@@ -44,7 +44,7 @@ struct InOrderTaskQueue <: AbstractTaskQueue
     InOrderTaskQueue(upper_queue) = new(upper_queue,
                                         Set{DTask}())
 end
-function _add_prev_deps!(queue::InOrderTaskQueue, spec::EagerTaskSpec)
+function _add_prev_deps!(queue::InOrderTaskQueue, spec::DTaskSpec)
     # Add previously-enqueued task(s) to this task's syncdeps
     opts = spec.options
     syncdeps = get(Set{Any}, opts, :syncdeps)
@@ -53,15 +53,15 @@ function _add_prev_deps!(queue::InOrderTaskQueue, spec::EagerTaskSpec)
     end
     spec.options = merge(opts, (;syncdeps,))
 end
-function enqueue!(queue::InOrderTaskQueue, spec::Pair{EagerTaskSpec,DTask})
+function enqueue!(queue::InOrderTaskQueue, spec::Pair{DTaskSpec,DTask})
     if length(queue.prev_tasks) > 0
         _add_prev_deps!(queue, first(spec))
         empty!(queue.prev_tasks)
     end
     push!(queue.prev_tasks, last(spec))
     enqueue!(queue.upper_queue, spec)
 end
-function enqueue!(queue::InOrderTaskQueue, specs::Vector{Pair{EagerTaskSpec,DTask}})
+function enqueue!(queue::InOrderTaskQueue, specs::Vector{Pair{DTaskSpec,DTask}})
     if length(queue.prev_tasks) > 0
         for (spec, task) in specs
             _add_prev_deps!(queue, spec)
@@ -74,26 +74,26 @@ function enqueue!(queue::InOrderTaskQueue, specs::Vector{Pair{EagerTaskSpec,DTas
     enqueue!(queue.upper_queue, specs)
 end
 function spawn_sequential(f::Base.Callable)
-    queue = InOrderTaskQueue(get_options(:task_queue, EagerTaskQueue()))
+    queue = InOrderTaskQueue(get_options(:task_queue, DefaultTaskQueue()))
     return with_options(f; task_queue=queue)
 end
 
 struct WaitAllQueue <: AbstractTaskQueue
     upper_queue::AbstractTaskQueue
     tasks::Vector{DTask}
 end
-function enqueue!(queue::WaitAllQueue, spec::Pair{EagerTaskSpec,DTask})
+function enqueue!(queue::WaitAllQueue, spec::Pair{DTaskSpec,DTask})
     push!(queue.tasks, spec[2])
     enqueue!(queue.upper_queue, spec)
 end
-function enqueue!(queue::WaitAllQueue, specs::Vector{Pair{EagerTaskSpec,DTask}})
+function enqueue!(queue::WaitAllQueue, specs::Vector{Pair{DTaskSpec,DTask}})
     for (_, task) in specs
         push!(queue.tasks, task)
     end
     enqueue!(queue.upper_queue, specs)
 end
 function wait_all(f; check_errors::Bool=false)
-    queue = WaitAllQueue(get_options(:task_queue, EagerTaskQueue()), DTask[])
+    queue = WaitAllQueue(get_options(:task_queue, DefaultTaskQueue()), DTask[])
     result = with_options(f; task_queue=queue)
     for task in queue.tasks
         if check_errors

src/submission.jl

@@ -193,13 +193,13 @@ function eager_process_elem_submission_to_local(id_map, x)
     end
 end
 # TODO: This can probably operate in-place
-function eager_process_args_submission_to_local(id_map, spec::Pair{EagerTaskSpec,DTask})
+function eager_process_args_submission_to_local(id_map, spec::Pair{DTaskSpec,DTask})
     return Base.mapany(first(spec).args) do pos_x
         pos, x = pos_x
         return pos => eager_process_elem_submission_to_local(id_map, x)
     end
 end
-function eager_process_args_submission_to_local(id_map, specs::Vector{Pair{EagerTaskSpec,DTask}})
+function eager_process_args_submission_to_local(id_map, specs::Vector{Pair{DTaskSpec,DTask}})
     return Base.mapany(specs) do spec
         eager_process_args_submission_to_local(id_map, spec)
     end
@@ -217,7 +217,7 @@ function eager_process_options_submission_to_local(id_map, options::NamedTuple)
         return options
     end
 end
-function eager_spawn(spec::EagerTaskSpec)
+function eager_spawn(spec::DTaskSpec)
     # Generate new DTask
     uid = eager_next_id()
     future = ThunkFuture()
@@ -226,7 +226,7 @@ function eager_spawn(spec::EagerTaskSpec)
     # Return unlaunched DTask
     return DTask(uid, future, finalizer_ref)
 end
-function eager_launch!((spec, task)::Pair{EagerTaskSpec,DTask})
+function eager_launch!((spec, task)::Pair{DTaskSpec,DTask})
     # Lookup DTask -> ThunkID
     local args, options
     lock(Sch.EAGER_ID_MAP) do id_map
@@ -240,7 +240,7 @@ function eager_launch!((spec, task)::Pair{EagerTaskSpec,DTask})
                              spec.f, args, options)
     task.thunk_ref = thunk_id.ref
 end
-function eager_launch!(specs::Vector{Pair{EagerTaskSpec,DTask}})
+function eager_launch!(specs::Vector{Pair{DTaskSpec,DTask}})
     ntasks = length(specs)
 
     uids = [task.uid for (_, task) in specs]

src/thunk.jl

@@ -430,13 +430,13 @@ function spawn(f, args...; kwargs...)
     args_kwargs = args_kwargs_to_pairs(args, kwargs)
 
     # Get task queue, and don't let it propagate
-    task_queue = get_options(:task_queue, EagerTaskQueue())
+    task_queue = get_options(:task_queue, DefaultTaskQueue())
     options = NamedTuple(filter(opt->opt[1] != :task_queue, Base.pairs(options)))
     propagates = filter(prop->prop != :task_queue, propagates)
     options = merge(options, (;propagates))
 
     # Construct task spec and handle
-    spec = EagerTaskSpec(f, args_kwargs, options)
+    spec = DTaskSpec(f, args_kwargs, options)
     task = eager_spawn(spec)
 
     # Enqueue the task into the task queue

test/task-queues.jl

@@ -8,7 +8,7 @@ function task_queue_wait_update(w, r, op, x, y)
     return op(x, y)
 end
 
-@testset "EagerTaskQueue" begin
+@testset "DefaultTaskQueue" begin
     r = Ref(0)
     R = Dagger.@mutable r
     d = begin