datadeps: Loosen task scopes

Author: jpsamaroo

Diff for: src/datadeps.jl

@@ -324,6 +324,8 @@ function distribute_tasks!(queue::DataDepsTaskQueue)
     for (spec, task) in queue.seen_tasks[task_order]
         our_proc = all_procs[proc_idx]
         our_space = only(memory_spaces(our_proc))
+        our_procs = collect(processors(our_space))
+        our_scope = UnionScope(map(ExactScope, our_procs)...)
 
         # Spawn copies before user's task, as necessary
         @dagdebug nothing :spawn_datadeps "($(spec.f)) Scheduling: $our_proc ($our_space)"
@@ -346,7 +348,7 @@ function distribute_tasks!(queue::DataDepsTaskQueue)
                 # Add copy-to operation (depends on latest owner of arg)
                 @dagdebug nothing :spawn_datadeps "($(spec.f))[$idx] Enqueueing copy-to: $data_space => $our_space"
                 arg_local = remote_args[data_space][arg]
-                copy_to_scope = ExactScope(our_proc)
+                copy_to_scope = our_scope
                 copy_to_syncdeps = Set{Any}()
                 get_write_deps!(arg, copy_to_syncdeps)
                 @dagdebug nothing :spawn_datadeps "($(spec.f))[$idx] $(length(copy_to_syncdeps)) syncdeps"
@@ -380,7 +382,7 @@ function distribute_tasks!(queue::DataDepsTaskQueue)
             end
         end
         @dagdebug nothing :spawn_datadeps "($(spec.f)) $(length(syncdeps)) syncdeps"
-        task_scope = Dagger.ExactScope(our_proc)
+        task_scope = our_scope
         spec.options = merge(spec.options, (;syncdeps, scope=task_scope))
         enqueue!(upper_queue, spec=>task)
         for (idx, (_, arg)) in enumerate(task_args)
@@ -413,7 +415,7 @@ function distribute_tasks!(queue::DataDepsTaskQueue)
             arg_remote = remote_args[data_remote_space][arg]
             @assert arg_remote !== arg_local
             data_local_proc = first(processors(data_local_space))
-            copy_from_scope = ExactScope(data_local_proc)
+            copy_from_scope = UnionScope(map(ExactScope, collect(processors(data_local_space)))...)
             copy_from_syncdeps = Set()
             get_write_deps!(arg, copy_from_syncdeps)
             @dagdebug nothing :spawn_datadeps "$(length(copy_from_syncdeps)) syncdeps"