You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@arrow.apache.org by "svilupp (via GitHub)" <gi...@apache.org> on 2023/03/12 13:39:23 UTC
[GitHub] [arrow-julia] svilupp opened a new pull request, #397: Add locks for de-/compressor + initialize decompressor
svilupp opened a new pull request, #397:
URL: https://github.com/apache/arrow-julia/pull/397
Fixes https://github.com/apache/arrow-julia/issues/396
This PR proposes to:
- fix the issue with indexing by `Threads.threadid()` by locking the compressor/decompressor (see https://github.com/apache/arrow-julia/issues/396)
- initialize an array of decompressors, similar to what has been previously done for compressors (eg, https://github.com/apache/arrow-julia/blob/9b36c8b1ec9efbdc63009d1b8cd72ee705fc1711/src/Arrow.jl#L80)
Acquiring locks introduces some additional overhead (I was not confident we could use SpinLocks given we traverse several packages and into C-code, so I opted for a slightly slower ReentrantLock).
I've created a minimalistic workload (100 partitions with 1 row) to see mostly the lock overhead.
I couldn't detect any slowdown in writing speed.
Since we're initializing decompressors upfront, we get c. 30% speed up when reading (no need to initialize for each record batch)
```
using Arrow, Tables, Random
# Speed test
N = 100
len = 100
t = Tables.rowtable((; x1 = map(x -> randstring(len), 1:N)));
fn="test_partitioned.arrow"
## Before
# Write:
@btime Arrow.write($fn, Iterators.partition($t, 1); compress = :lz4)
# 2.067 ms (12912 allocations: 13.24 MiB)
# Read:
@btime t=Arrow.Table($fn);
# 718.458 μs (9305 allocations: 533.62 KiB)
## After
# Write:
@btime Arrow.write($fn, Iterators.partition($t, 1); compress = :lz4)
# 2.021 ms (12987 allocations: 13.24 MiB)
# Read:
@btime t=Arrow.Table($fn);
# 461.083 μs (9712 allocations: 543.52 KiB)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-julia] baumgold commented on a diff in pull request #397: Add locks for de-/compressor + initialize decompressor
Posted by "baumgold (via GitHub)" <gi...@apache.org>.
baumgold commented on code in PR #397:
URL: https://github.com/apache/arrow-julia/pull/397#discussion_r1134720104
##########
src/arraytypes/arraytypes.jl:
##########
@@ -35,14 +35,19 @@ function toarrowvector(x, i=1, de=Dict{Int64, Any}(), ded=DictEncoding[], meta=g
@debugv 2 "converting top-level column to arrow format: col = $(typeof(x)), compression = $compression, kw = $(kw.data)"
Review Comment:
This function should probably be broken down into a few smaller functions for clarity:
```julia
toarrowvector(a::ArrowVector, compression::Nothing, kw...) = a
function toarrowvector(a::ArrowVector, compression::AbstractVector{LZ4FrameCompressor}, kw...)
lock(LZ4_FRAME_COMPRESSOR_LOCK[Threads.threadid()]) do
compress(Meta.CompressionTypes.LZ4_FRAME, compression[tid], a)
end
end
function toarrowvector(a::ArrowVector, compression::LZ4FrameCompressor, Vector{ZstdCompressor}, ZstdCompressor}=nothing, kw...)
compress(Meta.CompressionTypes.LZ4_FRAME, compression, a)
end
function toarrowvector(a::ArrowVector, compression::AbstractVector{ZstdCompressor}, kw...)
compress(Meta.CompressionTypes.ZSTD, compression, a)
end
function toarrowvector(a::ArrowVector, compression::ZstdCompressor, kw...)
lock(ZSTD_COMPRESSOR_LOCK[Threads.threadid()]) do
compress(Meta.CompressionTypes.ZSTD, compression[tid], a)
end
end
function toarrowvector(x, i=1, de=Dict{Int64, Any}(), ded=DictEncoding[], meta=getmetadata(x); compression::Union{Nothing, Vector{LZ4FrameCompressor}, LZ4FrameCompressor, Vector{ZstdCompressor}, ZstdCompressor}=nothing, kw...)
@debugv 2 "converting top-level column to arrow format: col = $(typeof(x)), compression = $compression, kw = $(kw.data)"
@debugv 3 x
A = arrowvector(x, i, 0, 0, de, ded, meta; compression=compression, kw...)
A = toarrowvector(A, compression)
@debugv 2 "converted top-level column to arrow format: $(typeof(A))"
@debugv 3 A
return A
end
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-julia] baumgold commented on a diff in pull request #397: Add locks for de-/compressor + initialize decompressor
Posted by "baumgold (via GitHub)" <gi...@apache.org>.
baumgold commented on code in PR #397:
URL: https://github.com/apache/arrow-julia/pull/397#discussion_r1134720104
##########
src/arraytypes/arraytypes.jl:
##########
@@ -35,14 +35,19 @@ function toarrowvector(x, i=1, de=Dict{Int64, Any}(), ded=DictEncoding[], meta=g
@debugv 2 "converting top-level column to arrow format: col = $(typeof(x)), compression = $compression, kw = $(kw.data)"
Review Comment:
This function should probably be broken down into a few smaller functions for clarity:
```julia
toarrowvector(a::ArrowVector, compression::Nothing, kw...) = a
function toarrowvector(a::ArrowVector, compression::AbstractVector{LZ4FrameCompressor}, kw...)
tid=Threads.threadid()
lock(LZ4_FRAME_COMPRESSOR_LOCK[tid]) do
compress(Meta.CompressionTypes.LZ4_FRAME, compression[tid], a)
end
end
function toarrowvector(a::ArrowVector, compression::LZ4FrameCompressor, Vector{ZstdCompressor}, ZstdCompressor}=nothing, kw...)
compress(Meta.CompressionTypes.LZ4_FRAME, compression, a)
end
function toarrowvector(a::ArrowVector, compression::AbstractVector{ZstdCompressor}, kw...)
compress(Meta.CompressionTypes.ZSTD, compression, a)
end
function toarrowvector(a::ArrowVector, compression::ZstdCompressor, kw...)
tid=Threads.threadid()
lock(ZSTD_COMPRESSOR_LOCK[tid]) do
compress(Meta.CompressionTypes.ZSTD, compression[tid], a)
end
end
function toarrowvector(x, i=1, de=Dict{Int64, Any}(), ded=DictEncoding[], meta=getmetadata(x); compression::Union{Nothing, Vector{LZ4FrameCompressor}, LZ4FrameCompressor, Vector{ZstdCompressor}, ZstdCompressor}=nothing, kw...)
@debugv 2 "converting top-level column to arrow format: col = $(typeof(x)), compression = $compression, kw = $(kw.data)"
@debugv 3 x
A = arrowvector(x, i, 0, 0, de, ded, meta; compression=compression, kw...)
A = toarrowvector(A, compression)
@debugv 2 "converted top-level column to arrow format: $(typeof(A))"
@debugv 3 A
return A
end
```
##########
src/Arrow.jl:
##########
@@ -73,6 +73,13 @@ include("show.jl")
const LZ4_FRAME_COMPRESSOR = LZ4FrameCompressor[]
const ZSTD_COMPRESSOR = ZstdCompressor[]
+const LZ4_FRAME_DECOMPRESSOR = LZ4FrameDecompressor[]
+const ZSTD_DECOMPRESSOR = ZstdDecompressor[]
+# add locks for multithreaded (de/)compression (because we index by threadid which might not be safe under Julia >1.8)
+const LZ4_FRAME_COMPRESSOR_LOCK = ReentrantLock[]
Review Comment:
A single thread can only do one thing at a time so maybe we can have a single lock per thread to be shared amongst all compressors/decompressors?
##########
src/Arrow.jl:
##########
@@ -82,6 +89,19 @@ function __init__()
lz4 = LZ4FrameCompressor(; compressionlevel=4)
CodecLz4.TranscodingStreams.initialize(lz4)
push!(LZ4_FRAME_COMPRESSOR, lz4)
+ # Locks
+ push!(LZ4_FRAME_COMPRESSOR_LOCK, ReentrantLock())
+ push!(ZSTD_COMPRESSOR_LOCK, ReentrantLock())
+ # Decompressors
+ zstdd = ZstdDecompressor()
+ CodecZstd.TranscodingStreams.initialize(zstdd)
Review Comment:
Maybe we should be a bit lazier about initializing these as I believe they're quite expensive. Shall we only initialize them only as needed? For example, if a user is only decompressing (reading) data then there's no reason to pay the penalty of initializing compressors (and vice versa). Same idea for LZ4 vs ZSTD - if a user only uses one then no need to initialize both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-julia] quinnj closed pull request #397: Add locks for de-/compressor + initialize decompressor
Posted by "quinnj (via GitHub)" <gi...@apache.org>.
quinnj closed pull request #397: Add locks for de-/compressor + initialize decompressor
URL: https://github.com/apache/arrow-julia/pull/397
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-julia] quinnj commented on pull request #397: Add locks for de-/compressor + initialize decompressor
Posted by "quinnj (via GitHub)" <gi...@apache.org>.
quinnj commented on PR #397:
URL: https://github.com/apache/arrow-julia/pull/397#issuecomment-1563568459
Not sure if @svilupp wasn't able to see this through, but I've cleaned up their code here: https://github.com/apache/arrow-julia/pull/445. I can confirm that I was able to reproduce the original segfault on Apple M1 and that the code in that PR resolves the issue. I believe that PR should also address @baumgold's comments from this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [arrow-julia] baumgold commented on a diff in pull request #397: Add locks for de-/compressor + initialize decompressor
Posted by "baumgold (via GitHub)" <gi...@apache.org>.
baumgold commented on code in PR #397:
URL: https://github.com/apache/arrow-julia/pull/397#discussion_r1134725578
##########
src/Arrow.jl:
##########
@@ -82,6 +89,19 @@ function __init__()
lz4 = LZ4FrameCompressor(; compressionlevel=4)
CodecLz4.TranscodingStreams.initialize(lz4)
push!(LZ4_FRAME_COMPRESSOR, lz4)
+ # Locks
+ push!(LZ4_FRAME_COMPRESSOR_LOCK, ReentrantLock())
+ push!(ZSTD_COMPRESSOR_LOCK, ReentrantLock())
+ # Decompressors
+ zstdd = ZstdDecompressor()
+ CodecZstd.TranscodingStreams.initialize(zstdd)
Review Comment:
Maybe we should be a bit lazier about initializing these as I believe they're quite expensive. Shall we initialize them only as needed? For example, if a user is only decompressing (reading) data then there's no reason to pay the penalty of initializing compressors (and vice versa). Same idea for LZ4 vs ZSTD - if a user only uses one then no need to initialize both.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@arrow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org