存储多线程函数调用输出的最佳方式

Question

我有一个函数f()它返回一个数据帧，其中的行数我事先不知道。我在多线程上下文中调用f()。我像这样存储结果：

results = [DataFrame() for _ in 1:100]
Threads.@threads for hi in 1:100
results[hi] = f(df)
end

当我运行此代码时，内存使用量激增，大概是因为results在获取数据帧的大小时必须不断调整自身大小[编辑：这不是真的]。预分配结果数组以使内存不会爆炸的最佳方法是什么？

更新 MWE ****

function func(df::DataFrame)
X = df[:time]
indices = findall(X .> 0)
end
# read in R data
rds = "blablab.rds"
objs = load(rds);
params = collect(0.5:0.005:0.7);
for i in 1:length(objs)
cols = [string(name) for name in names(objs.data[i]) if occursin("blabla",string(name))]
hypers = [(a,b) for a in cols, b in params]
results = [DataFrame() for _ in 1:length(hypers)]
# HERE IS WHERE THE MEMORY BLOWS UP
Threads.@threads for hi in 1:length(hypers)
name, val = hypers[hi]
results[hi] = func(objs.data[i])
end
end

df为 0.7GB。当我运行这段代码时，我的内存使用量高达 ~30GB!!似乎只是访问func()内部的一列df就是复制整个事情？

Answer 1

请在下面找到相同代码的两个版本 - 单线程和多线程从函数f()返回的一组DataFrame生成DataFrame，并具有随机长度。

using Random
using DataFrames
using BenchmarkTools
function f(rngs::Vector{Random.MersenneTwister}, offset)::DataFrame
t = Threads.threadid()
n = rand(rngs[t+offset], 1:20)
DataFrame(a=1:n,b=21:(20+n),t=t+offset)
end
function test_threads(rngs::Vector{Random.MersenneTwister})
res = DataFrame([Int,Int,Int],[:a,:b,:t],0)
lock = Threads.SpinLock()
Threads.@threads for i in 1:100
df = f(rngs,0)
Threads.lock(lock)
append!(res,df)
Threads.unlock(lock)
end
res
end
function test_normal(rngs::Vector{Random.MersenneTwister})    
res = DataFrame([Int,Int,Int],[:a,:b,:t],0)    
for i in 1:100
append!(res,f(rngs, i%2))
end
res
end

现在让我们进行测试：

julia> rngs = [Random.MersenneTwister(i) for i in 1:2];
julia> @btime test_normal($rngs);
891.306 μs (5983 allocations: 476.67 KiB)
rngs = [Random.MersenneTwister(i) for i in 1:Threads.nthreads()];
@btime test_threads($rngs);
674.559 μs (5549 allocations: 425.69 KiB)