diff --git a/.circleci/config.yml b/.circleci/config.yml index eb7c6d17809a..97dd0ecb955a 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -24,7 +24,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -113,10 +113,10 @@ jobs: j17_dtests: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -224,7 +224,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 1 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -410,7 +410,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -499,10 +499,10 @@ jobs: j11_dtests_latest_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -704,7 +704,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -820,10 +820,10 @@ jobs: j17_cqlsh_dtests_py38_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -930,10 +930,10 @@ jobs: j17_dtests_vnode_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1086,10 +1086,10 @@ jobs: j11_dtests_vnode_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1221,10 +1221,10 @@ jobs: j17_cqlsh_dtests_py311_latest: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1334,7 +1334,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1454,7 +1454,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1546,7 +1546,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1662,10 +1662,10 @@ jobs: j11_cqlsh_dtests_py38_latest: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1776,7 +1776,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1864,10 +1864,10 @@ jobs: j11_cqlsh_dtests_py311: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -1975,10 +1975,10 @@ jobs: j17_dtests_large_vnode_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2109,10 +2109,10 @@ jobs: j17_dtests_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2268,7 +2268,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2388,7 +2388,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2476,10 +2476,10 @@ jobs: j17_cqlsh_dtests_py311: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2586,10 +2586,10 @@ jobs: j11_cqlsh_dtests_py38: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2700,7 +2700,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2865,7 +2865,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -2957,7 +2957,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3076,7 +3076,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3272,10 +3272,10 @@ jobs: j11_dtests_large_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3362,7 +3362,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3450,10 +3450,10 @@ jobs: j11_dtests_large_vnode_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3585,10 +3585,10 @@ jobs: j11_dtests_large: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3675,7 +3675,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3767,7 +3767,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -3884,10 +3884,10 @@ jobs: j11_upgrade_dtests_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4022,7 +4022,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 1 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4141,7 +4141,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4336,7 +4336,7 @@ jobs: j17_build: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l parallelism: 1 @@ -4399,6 +4399,7 @@ jobs: - .m2 environment: - ANT_HOME: /usr/share/ant + - ANT_OPTS: "-Xmx3G" - LANG: en_US.UTF-8 - KEEP_TEST_DIR: true - DEFAULT_DIR: /home/cassandra/cassandra-dtest @@ -4447,7 +4448,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4567,7 +4568,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4658,7 +4659,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 1 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4851,7 +4852,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -4943,7 +4944,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -5214,7 +5215,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -5421,7 +5422,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 1 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -5615,7 +5616,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -5704,10 +5705,10 @@ jobs: j17_dtests_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -5812,10 +5813,10 @@ jobs: j11_dtests_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -5950,7 +5951,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6070,7 +6071,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6187,10 +6188,10 @@ jobs: j17_dtests_latest_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6324,7 +6325,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6413,10 +6414,10 @@ jobs: j11_upgrade_dtests: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6500,10 +6501,10 @@ jobs: j11_dtests_large_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6638,7 +6639,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6799,10 +6800,10 @@ jobs: j17_dtests_large_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6885,10 +6886,10 @@ jobs: j11_cqlsh_dtests_py38_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -6999,7 +7000,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7091,7 +7092,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7180,10 +7181,10 @@ jobs: j11_cqlsh_dtests_py311_latest: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7356,10 +7357,10 @@ jobs: j11_cqlsh_dtests_py311_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7470,7 +7471,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7586,10 +7587,10 @@ jobs: j17_cqlsh_dtests_py38_latest: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7696,10 +7697,10 @@ jobs: j17_cqlsh_dtests_py311_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7809,7 +7810,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 1 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -7926,10 +7927,10 @@ jobs: j17_dtests_latest: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8037,7 +8038,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8226,7 +8227,7 @@ jobs: j11_build: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l parallelism: 1 @@ -8289,6 +8290,7 @@ jobs: - .m2 environment: - ANT_HOME: /usr/share/ant + - ANT_OPTS: "-Xmx3G" - LANG: en_US.UTF-8 - KEEP_TEST_DIR: true - DEFAULT_DIR: /home/cassandra/cassandra-dtest @@ -8338,7 +8340,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8427,10 +8429,10 @@ jobs: j11_dtests_latest: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8580,10 +8582,10 @@ jobs: j11_dtests: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8667,10 +8669,10 @@ jobs: j17_cqlsh_dtests_py38: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8780,7 +8782,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -8899,7 +8901,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9064,7 +9066,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9152,10 +9154,10 @@ jobs: j17_dtests_large: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9241,7 +9243,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9406,7 +9408,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9494,10 +9496,10 @@ jobs: j11_dtests_vnode: docker: - image: apache/cassandra-testing-ubuntu2004-java11-w-dependencies:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9584,7 +9586,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9672,10 +9674,10 @@ jobs: j17_dtests_large_repeat: docker: - image: apache/cassandra-testing-ubuntu2004-java11:latest - resource_class: medium + resource_class: large working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9809,7 +9811,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra @@ -9900,7 +9902,7 @@ jobs: resource_class: medium working_directory: ~/ shell: /bin/bash -eo pipefail -l - parallelism: 4 + parallelism: 25 steps: - attach_workspace: at: /home/cassandra diff --git a/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java b/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java index a83d1f49106e..bd0b150e2363 100644 --- a/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java +++ b/src/java/org/apache/cassandra/db/compression/ZstdCompressionDictionary.java @@ -20,9 +20,13 @@ import java.time.Instant; import java.util.Objects; +import java.util.Queue; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicReference; +import com.github.luben.zstd.ZstdCompressCtx; +import com.github.luben.zstd.ZstdDecompressCtx; import com.github.luben.zstd.ZstdDictCompress; import com.github.luben.zstd.ZstdDictDecompress; import com.google.common.annotations.VisibleForTesting; @@ -46,6 +50,12 @@ public class ZstdCompressionDictionary implements CompressionDictionary, SelfRef // One ZstdDictDecompress and multiple ZstdDictCompress (per level) can be derived from the same raw dictionary content private final ConcurrentHashMap zstdDictCompressPerLevel = new ConcurrentHashMap<>(); private final AtomicReference dictDecompress = new AtomicReference<>(); + // Reusable native (de)compression contexts, pooled and borrowed per chunk by compressors sharing this + // dictionary. Compress contexts are keyed by level (the loaded CDict is level-specific); decompress contexts + // share one pool since a single DDict serves all levels. Closed by Tidy alongside the dictionary tables they + // were loaded with, so no context can outlive the native memory it references. + private final ConcurrentHashMap> compressCtxPoolPerLevel = new ConcurrentHashMap<>(); + private final Queue decompressCtxPool = new ConcurrentLinkedQueue<>(); private volatile Ref selfRef; private final Instant createdAt; @@ -189,6 +199,100 @@ public ZstdDictDecompress dictionaryForDecompression() } } + /** + * Borrow a pooled compression context with the dictionary for {@code compressionLevel} already loaded, + * creating one if the pool is empty. Reusing a context avoids allocating a native ZSTD_CCtx per call. + *
+ * IMPORTANT: Caller MUST hold a valid reference (via tryRef/ref) to this dictionary for as long as the + * borrowed context is in use, and must return it via {@link #releaseCompressCtx(int, ZstdCompressCtx)}. + * + * @param compressionLevel compression level the context should be loaded for + * @return a borrowed context; caller owns it exclusively until released + * @throws IllegalStateException if called without holding a valid reference + */ + public ZstdCompressCtx acquireCompressCtx(int compressionLevel) + { + ensureNotReleased(); + Queue pool = compressCtxPoolPerLevel.computeIfAbsent(compressionLevel, level -> new ConcurrentLinkedQueue<>()); + ZstdCompressCtx ctx = pool.poll(); + if (ctx == null) + // The compression level is carried by the precomputed CDict, so loadDict is all that is required. + ctx = new ZstdCompressCtx().loadDict(dictionaryForCompression(compressionLevel)); + return ctx; + } + + /** + * Return a context borrowed from {@link #acquireCompressCtx(int)}. If this dictionary has since been + * released, the context is closed instead of pooled, so it does not outlive the dictionary's native memory. + */ + public void releaseCompressCtx(int compressionLevel, ZstdCompressCtx ctx) + { + Queue pool = compressCtxPoolPerLevel.computeIfAbsent(compressionLevel, level -> new ConcurrentLinkedQueue<>()); + pool.offer(ctx); + if (selfRef == null || selfRef.globalCount() <= 0) // released concurrently — ensure it is not leaked + drainCompressPool(pool); + } + + /** + * Borrow a pooled decompression context with the dictionary already loaded, creating one if the pool is empty. + *
+ * IMPORTANT: Caller MUST hold a valid reference (via tryRef/ref) to this dictionary for as long as the + * borrowed context is in use, and must return it via {@link #releaseDecompressCtx(ZstdDecompressCtx)}. + * + * @throws IllegalStateException if called without holding a valid reference + */ + public ZstdDecompressCtx acquireDecompressCtx() + { + ensureNotReleased(); + ZstdDecompressCtx ctx = decompressCtxPool.poll(); + if (ctx == null) + ctx = new ZstdDecompressCtx().loadDict(dictionaryForDecompression()); + return ctx; + } + + /** + * Return a context borrowed from {@link #acquireDecompressCtx()}. If this dictionary has since been + * released, the context is closed instead of pooled. + */ + public void releaseDecompressCtx(ZstdDecompressCtx ctx) + { + decompressCtxPool.offer(ctx); + if (selfRef == null || selfRef.globalCount() <= 0) + drainDecompressPool(decompressCtxPool); + } + + private static void drainCompressPool(Queue pool) + { + ZstdCompressCtx ctx; + while ((ctx = pool.poll()) != null) + { + try + { + ctx.close(); + } + catch (Exception e) + { + logger.warn("Failed to close pooled ZstdCompressCtx", e); + } + } + } + + private static void drainDecompressPool(Queue pool) + { + ZstdDecompressCtx ctx; + while ((ctx = pool.poll()) != null) + { + try + { + ctx.close(); + } + catch (Exception e) + { + logger.warn("Failed to close pooled ZstdDecompressCtx", e); + } + } + } + @Override public Ref tryRef() { @@ -216,7 +320,8 @@ public Ref initRefLazily() { if (selfRef == null) { - selfRef = new Ref<>(this, new Tidy(zstdDictCompressPerLevel, dictDecompress)); + selfRef = new Ref<>(this, new Tidy(zstdDictCompressPerLevel, dictDecompress, + compressCtxPoolPerLevel, decompressCtxPool)); } } } @@ -244,12 +349,18 @@ private static class Tidy implements RefCounted.Tidy { private final ConcurrentHashMap zstdDictCompressPerLevel; private final AtomicReference dictDecompress; + private final ConcurrentHashMap> compressCtxPoolPerLevel; + private final Queue decompressCtxPool; Tidy(ConcurrentHashMap zstdDictCompressPerLevel, - AtomicReference dictDecompress) + AtomicReference dictDecompress, + ConcurrentHashMap> compressCtxPoolPerLevel, + Queue decompressCtxPool) { this.zstdDictCompressPerLevel = zstdDictCompressPerLevel; this.dictDecompress = dictDecompress; + this.compressCtxPoolPerLevel = compressCtxPoolPerLevel; + this.decompressCtxPool = decompressCtxPool; } /** @@ -267,6 +378,13 @@ private static class Tidy implements RefCounted.Tidy @Override public void tidy() { + // Close pooled (de)compression contexts BEFORE the dictionary tables they were loaded with, so no + // context can outlive the native memory it references. + for (Queue pool : compressCtxPoolPerLevel.values()) + drainCompressPool(pool); + compressCtxPoolPerLevel.clear(); + drainDecompressPool(decompressCtxPool); + // Close all compression dictionaries // No synchronization needed - reference counting ensures exclusive access for (ZstdDictCompress compressDict : zstdDictCompressPerLevel.values()) diff --git a/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java b/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java index 39fb09cd1882..87f292a18452 100644 --- a/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java +++ b/src/java/org/apache/cassandra/io/compress/ZstdDictionaryCompressor.java @@ -31,6 +31,8 @@ import com.github.benmanes.caffeine.cache.Caffeine; import com.github.benmanes.caffeine.cache.RemovalCause; import com.github.luben.zstd.Zstd; +import com.github.luben.zstd.ZstdCompressCtx; +import com.github.luben.zstd.ZstdDecompressCtx; import com.google.common.annotations.VisibleForTesting; import org.apache.cassandra.concurrent.ImmediateExecutor; @@ -51,11 +53,10 @@ public class ZstdDictionaryCompressor extends ZstdCompressorBase implements ICom .removalListener((ZstdCompressionDictionary dictionary, ZstdDictionaryCompressor compressor, RemovalCause cause) -> { - // Release dictionary reference when compressor is evicted from cache + // Release dictionary reference when compressor is evicted from cache. The dictionary's own Tidy + // closes any pooled native (de)compression contexts once this was the last reference. if (compressor != null && compressor.dictionaryRef != null) - { compressor.dictionaryRef.release(); - } }) .executor(ImmediateExecutor.INSTANCE) .build(); @@ -145,22 +146,39 @@ public int uncompress(byte[] input, int inputOffset, int inputLength, byte[] out return super.uncompress(input, inputOffset, inputLength, output, outputOffset); } - int dsz; + ZstdDecompressCtx ctx = null; + boolean ok = false; try { - dsz = (int) Zstd.decompressFastDict(output, outputOffset, - input, inputOffset, inputLength, - dictionary.dictionaryForDecompression()); + // Reuse a pooled context (dictionary loaded once) rather than the static one-shot, which allocates a + // fresh ZSTD_DCtx per call. + ctx = dictionary.acquireDecompressCtx(); + int dsz = ctx.decompressByteArray(output, outputOffset, output.length - outputOffset, + input, inputOffset, inputLength); + if (Zstd.isError(dsz)) + throw new IOException("Decompression failed due to " + Zstd.getErrorName(dsz)); + ok = true; + return dsz; + } + catch (IOException e) + { + throw e; } catch (Exception e) { throw new IOException("Decompression failed", e); } - - if (Zstd.isError(dsz)) - throw new IOException("Decompression failed due to " + Zstd.getErrorName(dsz)); - - return dsz; + finally + { + // success -> return to pool; failure -> close, so a context that errored is never reused + if (ctx != null) + { + if (ok) + dictionary.releaseDecompressCtx(ctx); + else + ctx.close(); + } + } } @Override @@ -172,19 +190,34 @@ public void uncompress(ByteBuffer input, ByteBuffer output) throws IOException return; } + ZstdDecompressCtx ctx = null; + boolean ok = false; try { - // Zstd compressors expect only direct bytebuffer. See ZstdCompressorBase.preferredBufferType and supports - int decompressedSize = (int) Zstd.decompressDirectByteBufferFastDict(output, output.position(), output.limit() - output.position(), - input, input.position(), input.limit() - input.position(), - dictionary.dictionaryForDecompression()); + // Zstd compressors expect only direct bytebuffer. See ZstdCompressorBase.preferredBufferType and supports. + // The context carries the dictionary (loaded once) and is reused across chunks. + ctx = dictionary.acquireDecompressCtx(); + int decompressedSize = ctx.decompressDirectByteBuffer(output, output.position(), output.limit() - output.position(), + input, input.position(), input.limit() - input.position()); output.position(output.position() + decompressedSize); input.position(input.limit()); + ok = true; } catch (Exception e) { throw new IOException("Decompression failed", e); } + finally + { + // success -> return to pool; failure -> close, so a context that errored is never reused + if (ctx != null) + { + if (ok) + dictionary.releaseDecompressCtx(ctx); + else + ctx.close(); + } + } } @Override @@ -196,19 +229,34 @@ public void compress(ByteBuffer input, ByteBuffer output) throws IOException return; } + ZstdCompressCtx ctx = null; + boolean ok = false; try { - // Zstd compressors expect only direct bytebuffer. See ZstdCompressorBase.preferredBufferType and supports - int compressedSize = (int) Zstd.compressDirectByteBufferFastDict(output, output.position(), output.limit() - output.position(), - input, input.position(), input.limit() - input.position(), - dictionary.dictionaryForCompression(compressionLevel())); + // Zstd compressors expect only direct bytebuffer. See ZstdCompressorBase.preferredBufferType and supports. + // The context carries the dictionary (loaded once) and is reused across chunks. + ctx = dictionary.acquireCompressCtx(compressionLevel()); + int compressedSize = ctx.compressDirectByteBuffer(output, output.position(), output.limit() - output.position(), + input, input.position(), input.limit() - input.position()); output.position(output.position() + compressedSize); input.position(input.limit()); + ok = true; } catch (Exception e) { throw new IOException("Compression failed", e); } + finally + { + // success -> return to pool; failure -> close, so a context that errored is never reused + if (ctx != null) + { + if (ok) + dictionary.releaseCompressCtx(compressionLevel(), ctx); + else + ctx.close(); + } + } } @VisibleForTesting