# \`break\` with a threaded loop?
**URL:** https://discourse.julialang.org/t/break-with-a-threaded-loop/78487
**Category:** General Usage
**Created:** [March 25, 2022, 6:36pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487 "2022-03-25T18:36:53Z")
**Posts on this page:** 12
**Page:** 1
### Author:  [@robsmith11](https://discourse.julialang.org/u/robsmith11)
#### Post date: [March 25, 2022, 6:36pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/1 "2022-03-25T18:36:53Z")
Is there any simple way to break from a threaded for loop (actually I want an infinite while loop, but that doesn’t work with `@threads`)? Or would I need to use something like channels for synchronization?
Trivial example:
```julia
julia> r = []
r_lock = ReentrantLock()
Threads.@threads for _ in 1:10^9
sleep(0.01)
lock(r_lock) do
push!(r, rand())
length(r) == 100 && break
end
end
ERROR: syntax: break or continue outside loop
Stacktrace:
[1] top-level scope
@ REPL[85]:3
```
---
### Author:  [@josuagrw](https://discourse.julialang.org/u/josuagrw)
#### Post date: [March 25, 2022, 6:44pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/2 "2022-03-25T18:44:33Z")
I think the best you could do, is introduce a shared flag (could be another lock) which flips when one of the threads hits your termination condition.
The other threads could look at that flag (for example with `trylock()`) and terminate when it is flipped.
This approach would probably require a `Channel`. `Threads.@threads` is really only meant for for loops with a known length.
---
### Author:  [@Mason](https://discourse.julialang.org/u/Mason)
#### Post date: [March 25, 2022, 8:50pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/3 "2022-03-25T20:50:19Z")
Yeah, `Threads.@threads` is a bad fit for this. I’d instead use Transducers.jl. It won’t guarentee that it breaks at exactly the loop iteration you ask for (in general, I think such a thing is kinda incompatible with the notion of parallelism, but here’ what your example would look like (though I’ve stored the `threadid()` in `r` instead of a random number so you can see which threads did what)
```julia
julia> using Transducers, BangBang, MicroCollections
julia> foldxt(1:10^9, init=Int[]) do r, _
sleep(0.01)
r = append!!(r, SingletonVector(Threads.threadid()))
if length(r) >= 100
reduced(r)
else
r
end
end
103-element Vector{Int64}:
2
4
1
3
2
4
1
2
3
4
4
3
2
1
4
2
1
3
4
2
3
1
2
⋮
1
2
3
1
4
2
4
3
1
1
4
3
4
3
1
3
1
2
4
2
3
1
```
* * *
**Edit:** the above code has correctness problems. @tkf has posted a fixed version here: [`break` with a threaded loop? - #10 by tkf](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/10)
---
### Author:  [@tkf](https://discourse.julialang.org/u/tkf)
#### Post date: [March 26, 2022, 5:20am UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/4 "2022-03-26T05:20:04Z")
What Mason wrote is the way to go with vanilla Transducers.jl. You can also use `break`, `return`, or `@goto` (with a label outside of the loop) inside of `@floop` to express this as well.
> [@Mason](#):
>
> I think such a thing is kinda incompatible with the notion of parallelism
Actually, that’s not the case. `break` can be expressed using a so-called _find-first_ (or _find-last_) monoid. You can search it in [Monoid - Wikipedia](https://en.wikipedia.org/wiki/Monoid). It can be computed in parallel and also the result is deterministic (unless the user opt-out this explicitly). See also: [Generic `Reduced` handling in parallel reduce by tkf · Pull Request #172 · JuliaFolds/Transducers.jl · GitHub](https://github.com/JuliaFolds/Transducers.jl/pull/172)
---
### Author:  [@Mason](https://discourse.julialang.org/u/Mason)
#### Post date: [March 26, 2022, 5:33am UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/5 "2022-03-26T05:33:59Z")
> [@tkf](#):
>
> What Mason wrote is the way to go with vanilla Transducers.jl. You can also use `break` , `return` , or `@goto` (with a label outside of the loop) inside of `@floop` to express this as well.
For what it’s worth, I had some trouble getting this to work personally. It seemed to just ignore the `Break` condition when I was trying it and I didn’t have time to track down what was going on. I can look again in the morning and try to send you an MWE
---
### Author:  [@tkf](https://discourse.julialang.org/u/tkf)
#### Post date: [March 26, 2022, 6:29am UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/6 "2022-03-26T06:29:00Z")
Yeah, let me know if there are some peculiar behavior
FWIW, here’s a demo of using `break` in parallel `@floop`
```julia
julia> function demo(xs = rand(2^20), ex = ThreadedEx(basesize = 2^10))
xmax = Threads.Atomic{Float64}(0.0)
imax = Threads.Atomic{Int}(0)
@floop ex for (i, x) in pairs(xs)
if x > 0.9999
xmax[] = x
imax[] = i
@reduce() do (j = nothing; i), (y = nothing; x)
j = i
y = x
end
break
end
end
(j => y), (imax[] => xmax[])
end;
julia> for _ in 1:100
a, b = demo()
if a != b
@show a b
break
end
end
a = 11792 => 0.99993600686011
b = 14586 => 0.9999006513025757
```
---
### Author:  [@Mason](https://discourse.julialang.org/u/Mason)
#### Post date: [March 26, 2022, 4:33pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/7 "2022-03-26T16:33:07Z")
> [@tkf](#):
>
> Yeah, let me know if there are some peculiar behavior
Here’s a MWE
```julia
julia> using FLoops, MicroCollections, BangBang
julia> @floop for i ∈ 1:100
sleep(0.001)
@reduce(r = append!!(EmptyVector(), SingletonVector(Threads.threadid())))
length(r) >= 10 && break
end;
julia> r
10-element Vector{Int64}:
1
1
1
1
1
1
1
1
1
1
```
So that didn’t multithread the loop, so explicitly providing a threaded executor I get
```julia
julia> @floop ThreadedEx(basesize=1) for i ∈ 1:100
sleep(0.001)
@reduce(r = append!!(EmptyVector(), SingletonVector(Threads.threadid())))
length(r) >= 10 && break
r
end
julia> r
100-element Vector{Int64}:
1
2
1
5
6
2
5
1
3
1
1
1
4
2
5
4
1
1
4
3
2
2
2
2
6
3
1
2
3
3
1
1
1
1
1
4
⋮
4
1
1
6
4
1
2
1
2
4
1
1
5
2
5
5
5
3
4
2
5
2
3
1
1
2
4
3
1
5
4
1
2
6
2
```
So it never actually breaks out of the loop
---
### Author:  [@tkf](https://discourse.julialang.org/u/tkf)
#### Post date: [March 27, 2022, 3:57am UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/8 "2022-03-27T03:57:10Z")
> [@Mason](#):
>
> ```julia
> julia> @floop for i ∈ 1:100
> sleep(0.001)
> @reduce(r = append!!(EmptyVector(), SingletonVector(Threads.threadid())))
> length(r) >= 10 && break
> end;
>
> ```
So `@floop` should actually be throwing an error since the accumulator `r` is accessed in the loop body outside of `@reduce`. (One way to understand parallel reduction is that the accumulator is “write only.”) But, apparently, it silently generates nonsensical code.
Using “`break`” based on `@reduce`d result is rather tricky since you have to show idempotency.
But, if you accept non-determinism, perhaps the simplest but still “lock-less” approach is something like this
```julia
julia> xs = rand(2^30)
basesize = 2^15
chunks = Iterators.partition(xs, basesize)
outputs = [Int[] for _ in chunks]
counter = Threads.Atomic{Int}(0)
@floop ThreadedEx(basesize = 1) for (output, chunk) in zip(outputs, chunks)
for x in chunk
counter[] > 10 && break
@reduce nitems += 1
if x > 0.9999
Threads.atomic_add!(counter, 1)
push!(output, Threads.threadid())
end
end
end
julia> nitems, length(xs) # processed items vs input items
(199088, 1073741824)
julia> foldl(append!, outputs; init = Int[])
11-element Vector{Int64}:
1
1
3
3
2
2
2
2
4
1
3
```
---
### Author:  [@Mason](https://discourse.julialang.org/u/Mason)
#### Post date: [March 27, 2022, 4:01am UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/9 "2022-03-27T04:01:33Z")
Ah I see. This is why I usually end up going back to Transducers.jl. Was the `foldxt` I wrote above kosher?
---
### Author:  [@tkf](https://discourse.julialang.org/u/tkf)
#### Post date: [March 28, 2022, 2:13am UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/10 "2022-03-28T02:13:40Z")
> [@Mason](#):
>
> ```julia
> julia> foldxt(1:10^9, init=Int[]) do r, _
> sleep(0.01)
> r = append!!(r, SingletonVector(Threads.threadid()))
> if length(r) >= 100
> reduced(r)
> else
> r
> end
> end
>
> ```
There are two problems in this code.
First, `init = Int[]` means that this vector is shared across multiple tasks (and threads if `Threads.nthreads() > 1`). It means that the same vector is mutated by multiple tasks that can run on different threads. So, this code is not thread-safe (in particular can have data races). You need to pass `OnInit(() -> Int[])` or `EmptyVector{Int}()` etc.
Second, the binary function passed to `foldxt` is not “good” in the sense that it probably is not doing what you want it to do. The binary function passed to `foldxt` have to combine two intermediate results. But, your function ignores the second argument. It means that it only process the first basecase. It can be observed by something like
```julia
julia> foldxt(1:100, init=EmptyVector{Int}(), basesize=10) do r, _
# sleep(0.01)
r = append!!(r, SingletonVector(Threads.threadid()))
if length(r) >= 100
reduced(r)
else
r
end
end
10-element Vector{Int64}:
1
1
1
1
1
1
1
1
1
1
```
You can see that the output has 10 elements even if the input has 100 elements.
Note that I use `init=EmptyVector{Int}()`. If you use `init=Int[]`, it may look like doing what you want it to do. However, as I said, it was not a valid Julia program.
Perhaps what you wanted to write was something like this:
```julia
1:1000 |>
Map() do _
sleep(0.01)
SingletonVector(Threads.threadid())
end |>
foldxt(init = OnInit(() -> Int[]), basesize = 1) do a, b
r = append!!(a, b)
if length(r) >= 100
reduced(r)
else
r
end
end
```
---
### Author:  [@Mason](https://discourse.julialang.org/u/Mason)
#### Post date: [March 28, 2022, 5:53pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/11 "2022-03-28T17:53:31Z")
Ah, thank you that is instructive. Multithreading is hard to do right!
---
### Author:  [@tkf](https://discourse.julialang.org/u/tkf)
#### Post date: [March 29, 2022, 3:09pm UTC](https://discourse.julialang.org/t/break-with-a-threaded-loop/78487/12 "2022-03-29T15:09:00Z")
Well, thanks for asking the question! If you didn’t get it, I’m almost sure almost no one (except me) understand how to use it 🙂 Now that I think of it, I don’t think I’ve ever explained the short-circuiting reduction clearly in the documentation. I better write a tutorial somewhere.
(This protocol was also hard to design right. I know I screwed up the current design. Fixing it for some edge cases has been my TODO for a long time…)