Skip to content

Commit 555cd23

Browse files
kpamnanyNHDaly
andauthored
Add a threadpool parameter to Channel constructor (#50858)
Without this, the task created by a `Channel` will run in the threadpool of the creating task; in the REPL, this could be the interactive threadpool. On 1.8, without threadpools: ```julia % julia +1.8 -t 8 _ _ _ _(_)_ | Documentation: https://docs.julialang.org (_) | (_) (_) | _ _ _| |_ __ _ | Type "?" for help, "]?" for Pkg help. | | | | | | |/ _` | | | | |_| | | | (_| | | Version 1.8.5 (2023-01-08) _/ |\__'_|_|_|\__'_| | Official https://julialang.org/ release |__/ | julia> for _ in 1:10 Channel{Int}(1; spawn=true) do _ Core.print("threadid=$(Threads.threadid())\n") end end threadid=2 threadid=5 threadid=2 threadid=2 threadid=1 threadid=6 threadid=7 threadid=8 threadid=3 threadid=4 ``` On 1.9, with no interactive threads: ```julia % julia +1.9 -t 8 _ _ _ _(_)_ | Documentation: https://docs.julialang.org (_) | (_) (_) | _ _ _| |_ __ _ | Type "?" for help, "]?" for Pkg help. | | | | | | |/ _` | | | | |_| | | | (_| | | Version 1.9.2 (2023-07-05) _/ |\__'_|_|_|\__'_| | Official https://julialang.org/ release |__/ | julia> for _ in 1:10 Channel{Int}(1; spawn=true) do _ Core.print("threadid=$(Threads.threadid())\n") end end threadid=4 threadid=4 threadid=4 threadid=2 threadid=3 threadid=1 threadid=7 threadid=5 threadid=8 threadid=6 ``` On 1.9, with an interactive thread: ```julia % julia +1.9 -t 7,1 _ _ _ _(_)_ | Documentation: https://docs.julialang.org (_) | (_) (_) | _ _ _| |_ __ _ | Type "?" for help, "]?" for Pkg help. | | | | | | |/ _` | | | | |_| | | | (_| | | Version 1.9.2 (2023-07-05) _/ |\__'_|_|_|\__'_| | Official https://julialang.org/ release |__/ | julia> for _ in 1:10 Channel{Int}(1; spawn=true) do _ Core.print("threadid=$(Threads.threadid())\n") end end threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 ``` With this PR, the `:default` threadpool is used instead. ```julia % julia +master -t7,1 _ _ _ _(_)_ | Documentation: https://docs.julialang.org (_) | (_) (_) | _ _ _| |_ __ _ | Type "?" for help, "]?" for Pkg help. | | | | | | |/ _` | | | | |_| | | | (_| | | Version 1.11.0-DEV.244 (2023-08-09) _/ |\__'_|_|_|\__'_| | Commit d99f249* (0 days old master) |__/ | julia> for _ in 1:10 Channel{Int}(1; spawn=true) do _ Core.print("threadid=$(Threads.threadid())\n") end end threadid=7 threadid=6 threadid=7 threadid=7 threadid=6 threadid=3 threadid=5 threadid=2 threadid=4 threadid=8 ``` And, the behavior can be overridden. ```julia % julia +master -t7,1 _ _ _ _(_)_ | Documentation: https://docs.julialang.org (_) | (_) (_) | _ _ _| |_ __ _ | Type "?" for help, "]?" for Pkg help. | | | | | | |/ _` | | | | |_| | | | (_| | | Version 1.11.0-DEV.244 (2023-08-09) _/ |\__'_|_|_|\__'_| | Commit d99f249* (0 days old master) |__/ | julia> for _ in 1:10 Channel{Int}(1; spawn=true, threadpool=:interactive) do _ Core.print("threadid=$(Threads.threadid())\n") end end threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 threadid=1 ``` --------- Co-authored-by: Nathan Daly <[email protected]>
1 parent cda570e commit 555cd23

File tree

3 files changed

+40
-7
lines changed

3 files changed

+40
-7
lines changed

base/channels.jl

Lines changed: 21 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -59,7 +59,7 @@ Channel(sz=0) = Channel{Any}(sz)
5959

6060
# special constructors
6161
"""
62-
Channel{T=Any}(func::Function, size=0; taskref=nothing, spawn=false)
62+
Channel{T=Any}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing)
6363
6464
Create a new task from `func`, bind it to a new channel of type
6565
`T` and size `size`, and schedule the task, all in a single call.
@@ -70,9 +70,14 @@ The channel is automatically closed when the task terminates.
7070
If you need a reference to the created task, pass a `Ref{Task}` object via
7171
the keyword argument `taskref`.
7272
73-
If `spawn = true`, the Task created for `func` may be scheduled on another thread
73+
If `spawn=true`, the `Task` created for `func` may be scheduled on another thread
7474
in parallel, equivalent to creating a task via [`Threads.@spawn`](@ref).
7575
76+
If `spawn=true` and the `threadpool` argument is not set, it defaults to `:default`.
77+
78+
If the `threadpool` argument is set (to `:default` or `:interactive`), this implies
79+
that `spawn=true` and the new Task is spawned to the specified threadpool.
80+
7681
Return a `Channel`.
7782
7883
# Examples
@@ -117,6 +122,9 @@ true
117122
In earlier versions of Julia, Channel used keyword arguments to set `size` and `T`, but
118123
those constructors are deprecated.
119124
125+
!!! compat "Julia 1.9"
126+
The `threadpool=` argument was added in Julia 1.9.
127+
120128
```jldoctest
121129
julia> chnl = Channel{Char}(1, spawn=true) do ch
122130
for c in "hello world"
@@ -129,12 +137,18 @@ julia> String(collect(chnl))
129137
"hello world"
130138
```
131139
"""
132-
function Channel{T}(func::Function, size=0; taskref=nothing, spawn=false) where T
140+
function Channel{T}(func::Function, size=0; taskref=nothing, spawn=false, threadpool=nothing) where T
133141
chnl = Channel{T}(size)
134142
task = Task(() -> func(chnl))
143+
if threadpool === nothing
144+
threadpool = :default
145+
else
146+
spawn = true
147+
end
135148
task.sticky = !spawn
136149
bind(chnl, task)
137150
if spawn
151+
Threads._spawn_set_thrpool(task, threadpool)
138152
schedule(task) # start it on (potentially) another thread
139153
else
140154
yield(task) # immediately start it, yielding the current thread
@@ -149,17 +163,17 @@ Channel(func::Function, args...; kwargs...) = Channel{Any}(func, args...; kwargs
149163
# of course not deprecated.)
150164
# We use `nothing` default values to check which arguments were set in order to throw the
151165
# deprecation warning if users try to use `spawn=` with `ctype=` or `csize=`.
152-
function Channel(func::Function; ctype=nothing, csize=nothing, taskref=nothing, spawn=nothing)
166+
function Channel(func::Function; ctype=nothing, csize=nothing, taskref=nothing, spawn=nothing, threadpool=nothing)
153167
# The spawn= keyword argument was added in Julia v1.3, and cannot be used with the
154168
# deprecated keyword arguments `ctype=` or `csize=`.
155-
if (ctype !== nothing || csize !== nothing) && spawn !== nothing
156-
throw(ArgumentError("Cannot set `spawn=` in the deprecated constructor `Channel(f; ctype=Any, csize=0)`. Please use `Channel{T=Any}(f, size=0; taskref=nothing, spawn=false)` instead!"))
169+
if (ctype !== nothing || csize !== nothing) && (spawn !== nothing || threadpool !== nothing)
170+
throw(ArgumentError("Cannot set `spawn=` or `threadpool=` in the deprecated constructor `Channel(f; ctype=Any, csize=0)`. Please use `Channel{T=Any}(f, size=0; taskref=nothing, spawn=false, threadpool=nothing)` instead!"))
157171
end
158172
# Set the actual default values for the arguments.
159173
ctype === nothing && (ctype = Any)
160174
csize === nothing && (csize = 0)
161175
spawn === nothing && (spawn = false)
162-
return Channel{ctype}(func, csize; taskref=taskref, spawn=spawn)
176+
return Channel{ctype}(func, csize; taskref=taskref, spawn=spawn, threadpool=threadpool)
163177
end
164178

165179
closed_exception() = InvalidStateException("Channel is closed.", :closed)

test/channel_threadpool.jl

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
# This file is a part of Julia. License is MIT: https://julialang.org/license
2+
3+
using Test
4+
using Base.Threads
5+
6+
@testset "Task threadpools" begin
7+
c = Channel{Symbol}() do c; put!(c, threadpool(current_task())); end
8+
@test take!(c) === threadpool(current_task())
9+
c = Channel{Symbol}(spawn = true) do c; put!(c, threadpool(current_task())); end
10+
@test take!(c) === :default
11+
c = Channel{Symbol}(threadpool = :interactive) do c; put!(c, threadpool(current_task())); end
12+
@test take!(c) === :interactive
13+
@test_throws ArgumentError Channel{Symbol}(threadpool = :foo) do c; put!(c, :foo); end
14+
end

test/channels.jl

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,11 @@ end
107107
@test taskref[].sticky == false
108108
@test collect(c) == [0]
109109
end
110+
let cmd = `$(Base.julia_cmd()) --depwarn=error --rr-detach --startup-file=no channel_threadpool.jl`
111+
new_env = copy(ENV)
112+
new_env["JULIA_NUM_THREADS"] = "1,1"
113+
run(pipeline(setenv(cmd, new_env), stdout = stdout, stderr = stderr))
114+
end
110115

111116
@testset "multiple concurrent put!/take! on a channel for different sizes" begin
112117
function testcpt(sz)

0 commit comments

Comments
 (0)