Fix @spawn_or_run_task with interactive threads

.github/workflows/ci.yml

@@ -36,7 +36,7 @@ jobs:
       - uses: julia-actions/julia-buildpkg@v1
       - uses: julia-actions/julia-runtest@v1
         env:
-          JULIA_NUM_THREADS: 4
+          JULIA_NUM_THREADS: 4,1
       - uses: julia-actions/julia-processcoverage@v1
       - uses: codecov/codecov-action@v1
         with:

NEWS.md

@@ -10,6 +10,13 @@
   keyword argument
   ([#3380](https://github.com/JuliaData/DataFrames.jl/pull/3380))
 
+## Bug fixes
+
+* Always use the default thread pool for multithreaded operations,
+  instead of using the interactive thread pool when Julia was started
+  with `-tM,N` with N > 0
+  ([#3385](https://github.com/JuliaData/DataFrames.jl/pull/3385))
+
 # DataFrames.jl v1.6.1 Release Notes
 
 ## Bug fixes

docs/src/lib/functions.md

@@ -7,7 +7,8 @@ CurrentModule = DataFrames
 ## Multithreading support
 
 By default, selected operations in DataFrames.jl automatically use multiple threads
-when available. It is task-based and implemented using the `@spawn` macro from Julia Base.
+when available. Multi-threading is task-based and implemented using the `@spawn`
+macro from Julia Base. Tasks are therefore scheduled on the `:default` threadpool.
 Functions that take user-defined functions and may run it in parallel
 accept a `threads` keyword argument which allows disabling multithreading
 when the provided function requires serial execution or is not thread-safe.

src/other/utils.jl

@@ -221,16 +221,23 @@
 Equivalent to `Threads.@spawn` if `threads === true`,
 otherwise run `expr` and return a `Task` that returns its value.
 """
-macro spawn_or_run_task(threads, expr)
-    letargs = Base._lift_one_interp!(expr)
+macro spawn_or_run_task(threads, ex)
+    letargs = Base._lift_one_interp!(ex)
 
-    thunk = esc(:(()->($expr)))
+    thunk = :(()->($(esc(ex))))
+    @static if VERSION >= v"1.10.0-DEV"
+        Base.replace_linenums!(thunk, __source__)
+    end
     var = esc(Base.sync_varname)
+    spawn_set_thrpool = VERSION >= v"1.9.0" ?
+        :(Base.Threads._spawn_set_thrpool(task, :default)) :
+        :()
     quote
         let $(letargs...)
             if $(esc(threads))
                 local task = Task($thunk)
                 task.sticky = false
+                $(spawn_set_thrpool)
             else
                 # Run expr immediately
                 res = $thunk()
@@ -253,16 +260,23 @@
 Equivalent to `Threads.@spawn` if `threads === true`,
 otherwise run `expr`.
 """
-macro spawn_or_run(threads, expr)
-    letargs = Base._lift_one_interp!(expr)
+macro spawn_or_run(threads, ex)
+    letargs = Base._lift_one_interp!(ex)
 
-    thunk = esc(:(()->($expr)))
+    thunk = :(()->($(esc(ex))))
+    if VERSION >= v"1.10.0-DEV"
+        Base.replace_linenums!(thunk, __source__)
+    end
     var = esc(Base.sync_varname)
+    spawn_set_thrpool = VERSION >= v"1.9.0" ?
+        :(Base.Threads._spawn_set_thrpool(task, :default)) :
+        :()
     quote
         let $(letargs...)
             if $(esc(threads))
                 local task = Task($thunk)
                 task.sticky = false
+                $(spawn_set_thrpool)
                 if $(Expr(:islocal, var))
                     put!($var, task)
                 end