Merge pull request #12 from JuliaTrustworthyAI/11-threads-static-cann…

…ot-be-used-concurrently-or-nested

Fixes this

.github/workflows/CI.yml

@@ -19,10 +19,8 @@ jobs:
       fail-fast: false
       matrix:
         version:
-          - '1.7'
-          - '1.8'
-          - '1.9'
           - '1.10'
+          - '1.11'
         os:
           - ubuntu-latest
         arch:

CHANGELOG.md

@@ -0,0 +1,12 @@
+# Changelog
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
+
+## Version [1.0.4] - 2024-11-06
+
+### Changed
+
+- Moved from `:static` threading to dynamic task scheduling. 
+

Project.toml

@@ -1,7 +1,7 @@
 name = "TaijaParallel"
 uuid = "bf1c2c22-5e42-4e78-8b6b-92e6c673eeb0"
 authors = ["Patrick Altmeyer <[email protected]>"]
-version = "1.0.3"
+version = "1.0.4"
 
 [deps]
 CounterfactualExplanations = "2f13d31b-18db-44c1-bc43-ebaf2cff0be0"
@@ -22,16 +22,16 @@ MPIExt = "MPI"
 [compat]
 Aqua = "0.8"
 CounterfactualExplanations = "0.1, 1"
-Logging = "1.7, 1.8, 1.9, 1.10"
+Logging = "1"
 MLUtils = "0.4.4"
 MPI = "0.20"
 PackageExtensionCompat = "1"
 ProgressMeter = "1"
 Reexport = "1"
-Serialization = "1.7, 1.8, 1.9, 1.10"
+Serialization = "1"
 TaijaBase = "1"
-Test = "1.7, 1.8, 1.9, 1.10"
-julia = "1.7, 1.8, 1.9, 1.10"
+Test = "1"
+julia = "1.10"
 
 [extras]
 Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595"

src/CounterfactualExplanations.jl/threads/evaluate.jl

@@ -36,8 +36,11 @@ function TaijaBase.parallelize(
         meta_data = fill(meta_data, length(counterfactuals))
     end
 
+    # Bundle arguments:
+    args = zip(counterfactuals, meta_data)
+
     # Preallocate:
-    evaluations = [[] for _ = 1:Threads.nthreads()]
+    evaluations = Vector{Vector}(undef, length(args))
 
     # Verbosity:
     if verbose
@@ -49,16 +52,26 @@ function TaijaBase.parallelize(
         )
     end
 
-    # Bundle arguments:
-    args = zip(counterfactuals, meta_data)
+    # Training: Use @spawn to dynamically schedule tasks
+    tasks = []
 
-    Threads.@threads :static for (ce, meta) in collect(args)
-        push!(evaluations[Threads.threadid()], f(ce, meta; kwargs...))
-        if verbose
-            ProgressMeter.next!(prog)
+    # Iterate with index to preserve order
+    for (i, (ce, meta)) in enumerate(args)
+        # Each task will execute this function dynamically on an available thread
+        task = Threads.@spawn begin
+            evaluation = f(ce, meta; kwargs...)
+            evaluations[i] = evaluation         # Store result in the correct position
+            if verbose
+                ProgressMeter.next!(prog)
+            end
         end
+        push!(tasks, task)
+    end
+
+    # Wait for all tasks to complete
+    for task in tasks
+        fetch(task)
     end
-    evaluations = reduce(vcat, evaluations)
 
     return evaluations
 end

src/CounterfactualExplanations.jl/threads/generate_counterfactual.jl

@@ -28,34 +28,41 @@ function TaijaBase.parallelize(
     # Break down into chunks:
     args = zip(counterfactuals, target, data, M, generator)
 
-    # Preallocate:
-    ces = [
-        Vector{CounterfactualExplanations.AbstractCounterfactualExplanation}() for
-        _ = 1:Threads.nthreads()
-    ]
+    # Preallocate a vector for storing results in the original order
+    ces = Vector{CounterfactualExplanations.AbstractCounterfactualExplanation}(undef, length(args))
 
-    # Verbosity:
+    # Verbosity setup:
     if verbose
         prog = ProgressMeter.Progress(
             length(args);
-            desc = "Generating counterfactuals ...",
-            showspeed = true,
-            color = :green,
+            desc="Generating counterfactuals ...",
+            showspeed=true,
+            color=:green,
         )
     end
 
-    # Training:  
-    Threads.@threads :static for (x, target, data, M, generator) in collect(args)
-        ce = with_logger(NullLogger()) do
-            f(x, target, data, M, generator; kwargs...)
-        end
-        push!(ces[Threads.threadid()], ce)
-        if verbose
-            ProgressMeter.next!(prog)
+    # Training: Use @spawn to dynamically schedule tasks
+    tasks = []
+
+    # Iterate with index to preserve order
+    for (i, (x, target, data, M, generator)) in enumerate(args)
+        # Each task will execute this function dynamically on an available thread
+        task = Threads.@spawn begin
+            ce = with_logger(NullLogger()) do
+                f(x, target, data, M, generator; kwargs...)
+            end
+            ces[i] = ce  # Store result in the correct position
+            if verbose
+                ProgressMeter.next!(prog)
+            end
         end
+        push!(tasks, task)
     end
 
-    ces = reduce(vcat, ces)
+    # Wait for all tasks to complete
+    for task in tasks
+        fetch(task)
+    end
 
     return ces
 end

test/CounterfactualExplanations.jl/mpi.jl

@@ -1,6 +1,5 @@
 using CounterfactualExplanations.DataPreprocessing: CounterfactualData
 using CounterfactualExplanations.Evaluation: benchmark
-using LaplaceRedux
 using Logging
 using TaijaData
 using TaijaParallel

test/CounterfactualExplanations.jl/threads.jl

@@ -1,5 +1,7 @@
 using CounterfactualExplanations
 using CounterfactualExplanations.Convergence
+using CounterfactualExplanations.Evaluation
+using CounterfactualExplanations.Models
 using TaijaData
 using TaijaParallel: ThreadsParallelizer, @with_parallelizer
 
@@ -24,4 +26,10 @@ ces = @with_parallelizer parallelizer begin
     generate_counterfactual(xs, target, counterfactual_data, M, generator; convergence = conv)
 end
 
+@test all(CounterfactualExplanations.factual.(ces) .== (x -> x[1]).(collect(xs)))
+
+evals = @with_parallelizer parallelizer begin
+    evaluate(ces)
+end
+
 @test true