Detect and throw an error when multithreading with Boxed variables (#141)

* Detect and throw an error when multithreading with `Box`ed variables

* elaborate on error msg

* don't try to downgrade `Test`

* don't escape `lock`

* update Changelog

* bump major version, add ScopedValues dep for v1.10

* add docs on boxing

* add `@allow_boxed_captures` and `@disallow_boxed_captures`

* include `@allow_boxed_captures` and `@disallow_boxed_captures` docstrings

Author: MasonProtter

.github/workflows/downgrade_CI.yml

@@ -23,6 +23,6 @@ jobs:
           version: ${{ matrix.version }}
       - uses: cjdoris/julia-downgrade-compat-action@v1
         with:
-          skip: Pkg,TOML
+          skip: Pkg,TOML,Test
       - uses: julia-actions/julia-buildpkg@v1
       - uses: julia-actions/julia-runtest@v1

CHANGELOG.md

@@ -1,6 +1,10 @@
 OhMyThreads.jl Changelog
 =========================
 
+Version 0.8.0
+-------------
+- ![BREAKING][badge-breaking] We now detect and throw errors if an `OhMyThreads` parallel function is passed a closure containing a `Box`ed variable. This behaviour can be disabled with the new `@allow_boxed_captures` macro, and re-enabled with `@disallow_boxed_captures`. ([#141][gh-pr-141])
+
 Version 0.7.0
 -------------
 - ![BREAKING][badge-breaking] We now use ChunkSplitters version 3.0. The function `OhMyThreads.chunks` has been renamed to `OhMyThreads.index_chunks`. The new functions `index_chunks` and `chunks` (different from the old one with the same name!) are now exported. See ChunkSplitters.jl for more information.
@@ -139,3 +143,4 @@ Version 0.2.0
 
 [gh-pr-5]: https://github.com/JuliaFolds2/OhMyThreads.jl/pull/5
 [gh-pr-121]: https://github.com/JuliaFolds2/OhMyThreads.jl/pull/121
+[gh-pr-141]: https://github.com/JuliaFolds2/OhMyThreads.jl/pull/141

Project.toml

@@ -1,11 +1,12 @@
 name = "OhMyThreads"
 uuid = "67456a42-1dca-4109-a031-0a68de7e3ad5"
 authors = ["Carsten Bauer <[email protected]>", "Mason Protter <[email protected]>"]
-version = "0.7.0"
+version = "0.8.0"
 
 [deps]
 BangBang = "198e06fe-97b7-11e9-32a5-e1d131e6ad66"
 ChunkSplitters = "ae650224-84b6-46f8-82ea-d812ca08434e"
+ScopedValues = "7e506255-f358-4e82-b7e4-beb19740aa63"
 StableTasks = "91464d47-22a1-43fe-8b7f-2d57ee82463f"
 TaskLocalValues = "ed4db957-447d-4319-bfb6-7fa9ae7ecf34"
 
@@ -15,6 +16,7 @@ BangBang = "0.3.40, 0.4"
 ChunkSplitters = "3"
 StableTasks = "0.1.5"
 TaskLocalValues = "0.1"
+ScopedValues = "1.3"
 Test = "1"
 julia = "1.10"
 

docs/make.jl

@@ -27,6 +27,7 @@ makedocs(;
             "Trapezoidal Integration" => "literate/integration/integration.md"
         ],
         "Translation Guide" => "translation.md",
+        "Boxed Variables" => "literate/boxing/boxing.md",
         "Thread-Safe Storage" => "literate/tls/tls.md",
         "False Sharing" => "literate/falsesharing/falsesharing.md",
         # "Explanations" => [

docs/src/literate/boxing/Project.toml

@@ -0,0 +1,2 @@
+[deps]
+OhMyThreads = "67456a42-1dca-4109-a031-0a68de7e3ad5"

docs/src/literate/boxing/boxing.jl

@@ -0,0 +1,63 @@
+#====================================
+# Boxed Variables
+
+All multithreading in julia is built around the idea of passing around
+and executing functions, but often these functions "enclose" data from
+an outer local scope, making them what's called a "closure".
+
+Julia allows functions which capture variables to re-bind those variables
+to different values, but doing so can cause subtle race conditions in
+multithreaded code.
+
+Consider the following example:
+====================================#
+
+let out = zeros(Int, 10)
+    Threads.@threads for i in 1:10
+        A = i
+        sleep(1/100)
+        out[i] = A
+    end
+    A = 1
+    out
+end
+
+#====================================
+You may have expected that to return `[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]`,
+but the nonsense result is caused by `A` actually being a shared mutable
+container here which all the parallel tasks are accessing and mutating
+in parallel, giving unpredictable results.
+
+OhMyThreads.jl tries to protect users from this surprising behaviour:
+====================================#
+using OhMyThreads
+
+try
+    let
+        ## this throws an error!
+        out = tmap(1:10) do i
+            A = i
+            sleep(1/100)
+            A
+        end
+        A = 1
+        out
+    end
+catch e;
+    println(e.msg) # show that error
+end
+
+#====================================
+If you really desire to bypass this behaviour, you can use the
+`@allow_boxed_captures` macro
+====================================#
+
+@allow_boxed_captures let
+    out = tmap(1:10) do i
+        A = i
+        sleep(1/100)
+        A
+    end
+    A = 1
+    out
+end

docs/src/literate/boxing/boxing.md

@@ -0,0 +1,110 @@
+```@meta
+EditURL = "boxing.jl"
+```
+
+# Boxed Variables
+
+All multithreading in julia is built around the idea of passing around
+and executing functions, but often these functions "enclose" data from
+an outer local scope, making them what's called a "closure".
+
+Julia allows functions which capture variables to re-bind those variables
+to different values, but doing so can cause subtle race conditions in
+multithreaded code.
+
+Consider the following example:
+
+````julia
+let out = zeros(Int, 10)
+    Threads.@threads for i in 1:10
+        A = i
+        sleep(1/100)
+        out[i] = A
+    end
+    A = 1
+    out
+end
+````
+
+````
+10-element Vector{Int64}:
+ 6
+ 2
+ 3
+ 2
+ 3
+ 2
+ 3
+ 2
+ 3
+ 4
+````
+
+You may have expected that to return `[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]`,
+but the nonsense result is caused by `A` actually being a shared mutable
+container here which all the parallel tasks are accessing and mutating
+in parallel, giving unpredictable results.
+
+OhMyThreads.jl tries to protect users from this surprising behaviour:
+
+````julia
+using OhMyThreads
+
+try
+    let
+        # this throws an error!
+        out = tmap(1:10) do i
+            A = i
+            sleep(1/100)
+            A
+        end
+        A = 1
+        out
+    end
+catch e;
+    println(e.msg) # show that error
+end
+````
+
+````
+Attempted to capture and modify outer local variable(s) A, which would be not only slow, but could also cause a race condition. Consider marking these variables as local inside their respective closure, or redesigning your code to avoid the race condition.
+
+If these variables are inside a @one_by_one or @only_one block, consider using a mutable Ref instead of re-binding the variable.
+
+This error can be bypassed with the @allow_boxed_captures macro.
+
+````
+
+If you really desire to bypass this behaviour, you can use the
+`@allow_boxed_captures` macro
+
+````julia
+@allow_boxed_captures let
+    out = tmap(1:10) do i
+        A = i
+        sleep(1/100)
+        A
+    end
+    A = 1
+    out
+end
+````
+
+````
+10-element Vector{Int64}:
+ 7
+ 6
+ 7
+ 6
+ 7
+ 6
+ 7
+ 6
+ 7
+ 7
+````
+
+---
+
+*This page was generated using [Literate.jl](https://github.com/fredrikekre/Literate.jl).*
+

docs/src/refs/api.md

@@ -13,6 +13,8 @@ CollapsedDocStrings = true
 @local
 @only_one
 @one_by_one
+@allow_boxed_captures
+@disallow_boxed_captures
 ```
 
 ### Functions

src/OhMyThreads.jl

@@ -15,6 +15,9 @@ export chunks, index_chunks
 
 using TaskLocalValues: TaskLocalValues
 const TaskLocalValue = TaskLocalValues.TaskLocalValue
+
+using ScopedValues: ScopedValues, ScopedValue, @with
+
 include("types.jl")
 include("functions.jl")
 include("macros.jl")
@@ -26,7 +29,7 @@ using .Schedulers: Scheduler, DynamicScheduler, StaticScheduler, GreedyScheduler
 include("implementation.jl")
 include("experimental.jl")
 
-export @tasks, @set, @local, @one_by_one, @only_one
+export @tasks, @set, @local, @one_by_one, @only_one, @allow_boxed_captures, @disallow_boxed_captures
 export treduce, tmapreduce, treducemap, tmap, tmap!, tforeach, tcollect
 export Scheduler, DynamicScheduler, StaticScheduler, GreedyScheduler, SerialScheduler
 

src/implementation.jl

@@ -1,7 +1,7 @@
 module Implementation
 
 import OhMyThreads: treduce, tmapreduce, treducemap, tforeach, tmap, tmap!, tcollect
-using OhMyThreads: @spawn, @spawnat, WithTaskLocals, promise_task_local, ChannelLike
+using OhMyThreads: @spawn, @spawnat, WithTaskLocals, promise_task_local, ChannelLike, allowing_boxed_captures
 using OhMyThreads.Tools: nthtid
 using OhMyThreads: Scheduler,
                    DynamicScheduler, StaticScheduler, GreedyScheduler,
@@ -93,6 +93,7 @@ end
 
 treducemap(op, f, A...; kwargs...) = tmapreduce(f, op, A...; kwargs...)
 
+
 # DynamicScheduler: AbstractArray/Generic
 function _tmapreduce(f,
         op,
@@ -102,6 +103,7 @@ function _tmapreduce(f,
         mapreduce_kwargs)::OutputType where {OutputType}
     (; threadpool) = scheduler
     check_all_have_same_indices(Arrs)
+    throw_if_boxed_captures(f, op)
     if chunking_enabled(scheduler)
         tasks = map(_index_chunks(scheduler, first(Arrs))) do inds
             args = map(A -> view(A, inds), Arrs)
@@ -128,6 +130,7 @@ function _tmapreduce(f,
         scheduler::DynamicScheduler,
         mapreduce_kwargs)::OutputType where {OutputType, T}
     (; threadpool) = scheduler
+    throw_if_boxed_captures(f, op)
     tasks = map(only(Arrs)) do idcs
         @spawn threadpool promise_task_local(f)(idcs)
     end
@@ -143,6 +146,7 @@ function _tmapreduce(f,
         mapreduce_kwargs)::OutputType where {OutputType}
     nt = nthreads()
     check_all_have_same_indices(Arrs)
+    throw_if_boxed_captures(f, op)
     if chunking_enabled(scheduler)
         tasks = map(enumerate(_index_chunks(scheduler, first(Arrs)))) do (c, inds)
             tid = @inbounds nthtid(mod1(c, nt))
@@ -175,6 +179,7 @@ function _tmapreduce(f,
         scheduler::StaticScheduler,
         mapreduce_kwargs)::OutputType where {OutputType, T}
     check_all_have_same_indices(Arrs)
+    throw_if_boxed_captures(f, op)
     chnks = only(Arrs)
     nt = nthreads()
     tasks = map(enumerate(chnks)) do (c, idcs)
@@ -227,6 +232,7 @@ function _tmapreduce(f,
         ntasks = min(length(first(Arrs)), ntasks_desired)
         ch_len = length(first(Arrs))
     end
+    throw_if_boxed_captures(f, op)
     # TODO: Use ChannelLike for iterators that support it. Dispatch on IndexLinear?
     ch = Channel{Tuple{eltype.(Arrs)...}}(ch_len; spawn = true) do ch
         for args in zip(Arrs...)
@@ -272,6 +278,7 @@ function _tmapreduce(f,
         throw(ArgumentError("SizeUnkown iterators in combination with a greedy scheduler and chunking are currently not supported."))
     end
     check_all_have_same_indices(Arrs)
+    throw_if_boxed_captures(f, op)
     chnks = _index_chunks(scheduler, first(Arrs))
     ntasks_desired = scheduler.ntasks
     ntasks = min(length(chnks), ntasks_desired)
@@ -316,6 +323,33 @@ function check_all_have_same_indices(Arrs)
     end
 end
 
+
+function throw_if_boxed_captures(f)
+    if allowing_boxed_captures[]
+        return nothing
+    end
+    T = typeof(f)
+    if any(FT -> FT <: Core.Box, fieldtypes(T))
+        boxed_fields = join((fieldname(T, i) for i in 1:fieldcount(T) if fieldtype(T,i) <: Core.Box), ", ")
+        error("Attempted to capture and modify outer local variable(s) $boxed_fields, which would be not only slow, but could also cause a race condition. Consider marking these variables as local inside their respective closure, or redesigning your code to avoid the race condition.\n\nIf these variables are inside a @one_by_one or @only_one block, consider using a mutable Ref instead of re-binding the variable.\n\nThis error can be bypassed with the @allow_boxed_captures macro.")
+    end
+    for i ∈ 1:fieldcount(T)
+        # recurse into nested captured functions.
+        if fieldtype(T, i) <: Function
+            f_inner = getfield(f, i)
+            if f !== f_inner
+                # don't recurse into self!
+                throw_if_boxed_captures(getfield(f, i))
+            end
+        end
+    end
+end
+
+function throw_if_boxed_captures(f, fs...)
+    throw_if_boxed_captures(f)
+    throw_if_boxed_captures(fs...)
+end
+
 #-------------------------------------------------------------
 
 function treduce(op, A...; kwargs...)
@@ -401,6 +435,7 @@ function _tmap(scheduler::DynamicScheduler{NoChunking},
         _Arrs::AbstractArray...;)
     (; threadpool) = scheduler
     Arrs = (A, _Arrs...)
+    throw_if_boxed_captures(f)
     tasks = map(eachindex(A)) do i
         @spawn threadpool begin
             args = map(A -> A[i], Arrs)
@@ -417,6 +452,7 @@ function _tmap(scheduler::DynamicScheduler{NoChunking},
         A::Union{AbstractChunks, ChunkSplitters.Internals.Enumerate},
         _Arrs::AbstractArray...)
     (; threadpool) = scheduler
+    throw_if_boxed_captures(f)
     tasks = map(A) do idcs
         @spawn threadpool promise_task_local(f)(idcs)
     end
@@ -429,6 +465,7 @@ function _tmap(scheduler::StaticScheduler{NoChunking},
         A::AbstractChunks,
         _Arrs::AbstractArray...)
     nt = nthreads()
+    throw_if_boxed_captures(f)
     tasks = map(enumerate(A)) do (c, idcs)
         tid = @inbounds nthtid(mod1(c, nt))
         @spawnat tid promise_task_local(f)(idcs)
@@ -443,6 +480,7 @@ function _tmap(scheduler::StaticScheduler{NoChunking},
         _Arrs::AbstractArray...;)
     Arrs = (A, _Arrs...)
     nt = nthreads()
+    throw_if_boxed_captures(f)
     tasks = map(enumerate(A)) do (c, i)
         tid = @inbounds nthtid(mod1(c, nt))
         @spawnat tid begin
@@ -485,6 +523,7 @@ end
         map!(f, out, Arrs...)
     else
         @boundscheck check_all_have_same_indices((out, Arrs...))
+        throw_if_boxed_captures(f)
         mapping_f = maybe_rewrap(f) do f
             function mapping_function(i)
                 args = map(A -> @inbounds(A[i]), Arrs)