# Reduce allocations & lock conflicts and speed up when using multithreads

**URL:** <https://discourse.julialang.org/t/reduce-allocations-lock-conflicts-and-speed-up-when-using-multithreads/125749>\
**Category:** Performance\
**Tags:** threads, yfinance\
**Created:** [February 10, 2025, 2:01pm UTC](https://discourse.julialang.org/t/reduce-allocations-lock-conflicts-and-speed-up-when-using-multithreads/125749 "2025-02-10T14:01:31Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![Stephen](https://avatars.discourse-cdn.com/v4/letter/s/96bed5/32.png) [@Stephen](https://discourse.julialang.org/u/Stephen)\
**Post date:** [February 10, 2025, 2:01pm UTC](https://discourse.julialang.org/t/reduce-allocations-lock-conflicts-and-speed-up-when-using-multithreads/125749/1 "2025-02-10T14:01:31Z")

</div>

Hello there, performance enhancement help wanted:

> **Initial function**
>
> ```julia
> using Base.Threads, DataFrames, YFinance, Statistics, Dates
> 
> function historical_low(
> code::Vector{String}, 
> startdt::String,
> l::Float64, 
> ratio::Float64
> )
> lk = ReentrantLock()
> low_σ = Vector{String}(undef, 0) # Preallocated result storage
> 
> @threads for i in code
> try
> p = get_prices(
> i, startdt = startdt, enddt = today(), interval = "1d", 
> divsplits = true, exchange_local_time = true
> ) |> DataFrame
> 
> # Compute only once to avoid redundant allocations
> today_low = p.low[end]
> min_low = minimum(p.low)
> std_low = std(p.low)
> 
> # Stock price check
> if l < today_low < min_low + ratio * std_low
> lock(lk) do
> push!(low_σ, i)  
> end
> end
> catch e
> @error "Error processing $i: $e"
> end
> end
> return unique(low_σ) # Remove duplicates before returning
> end
> 
> code = get_all_symbols("NYSE")
> @time results = historical_low(code, "2022-01-01", 5.0, 0.1)
> 
> ```

which gives

```julia
309.934138 seconds (5.00 M allocations: 1.982 GiB, 0.23% gc time, 2007 lock conflicts, 0.52% compilation time)
48-element Vector{String}:

```

After asking GPT, I got an optimized version:

> **Optimized**
>
> ```julia
> function historical_low(
> code::Vector{String}, 
> startdt::String,
> l::Float64, # l: 股价最低值
> ratio::Float64  
> )
> 	
> results = Channel{String}(Inf) # Thread-safe storage for results
> 	
> batch_size::Int = 50  
> # Split stocks into batches
> batches = Iterators.partition(code, batch_size) |> collect
> 	
> # Launch tasks asynchronously
> futures = [
> @spawn begin
> 
> local_results = Vector{String}(undef, 0)
> 
> for i in batch
> try
> p = get_prices(
> i, startdt = startdt, enddt = today(), interval = "1d",
> divsplits = true, exchange_local_time = true
> ) |> DataFrame
> 
> # Use Tables.columntable() to avoid extra allocations
> lows = Tables.columntable(p)[:low]
> 
> today_low = lows[end]
> min_low = minimum(lows)
> std_low = std(lows)
> 
> if l < today_low < min_low + ratio * std_low
> push!(local_results, i)
> end
> catch e
> @error "Error processing $i: $e"
> end
> end
> 
> # Send results to channel (avoids lock contention)
> for stock in local_results
> put!(results, stock)
> end
> end for batch in batches
> ]
> 
> # Wait for all tasks to complete
> wait.(futures)
> close(results) # Mark channel as finished
> 
> # Collect all results efficiently
> return collect(results) # Convert channel to array
> end
> @time results = historical_low(code, "2022-01-01", 5.0, 0.1)
> 
> ```

this gives:

```julia
80.924692 seconds (3.92 M allocations: 1.925 GiB, 1.28% gc time, 4950 lock conflicts)
48-element Vector{String}:

```

Can it be further optimized？  
I also tried `pmap()` for distributed computing:

> **Distributed Computing**
>
> ```julia
> using Distributed, DataFrames, YFinance, Statistics, Dates
> 
> addprocs(5)
> procs()
> @everywhere begin
> 
> using DataFrames, YFinance, Statistics, Dates # Ensure all workers have dependencies
> 
> function fetch_low_stock(i, startdt, l, ratio)
> try
> p = get_prices(
> i, startdt = startdt, enddt = today(), 
> interval = "1d", divsplits = true, exchange_local_time = true
> ) |> DataFrame
> 	
> if isempty(p) || isempty(p.low)
> return nothing
> end
> 			
> last_low, min_low, std_low = p.low[end], minimum(p.low), std(p.low)
> 			
> return (l < last_low < min_low + ratio * std_low) ? i : nothing
> 	
> catch e
> @warn "Error processing $i:$e"
> return nothing
> end
> end
> end
> 
> function historical_low_pmap(
> code::Vector{String}, 
> startdt::String,
> l::Float64, 
> ratio::Float64)
> 
> batch_size = 1000 # Adjust batch size based on available memory
> code_batches = Iterators.partition(code, batch_size) |> collect
> results = vcat(pmap(
> batch -> filter(!isnothing, fetch_low_stock.(
> batch, startdt, l, ratio)
> ), code_batches)...
> )
> return results # Remove duplicates
> end
> 
> ```

but it is even worse (more than 15 minutes), so I gave it up

---

<div class="post-metadata">

**Author:** ![jakobnissen](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/jakobnissen/32/13477_2.png) [@jakobnissen](https://discourse.julialang.org/u/jakobnissen)\
**Post date:** [February 10, 2025, 2:11pm UTC](https://discourse.julialang.org/t/reduce-allocations-lock-conflicts-and-speed-up-when-using-multithreads/125749/2 "2025-02-10T14:11:41Z")

</div>

You should profile your code to figure out what takes time. If the work inside each task is what takes the most time, then optimise that.

If your issues are contention around the channel (but I doubt that), then you can lock the channel before pushing in a loop like so:

```julia
# Send results to channel
@lock results begin
    for stock in local_results
        put!(results, stock)
    end
end

```