mirrors/zstd
0
0
Fork 0
mirror of https://github.com/facebook/zstd.git synced 2024-11-25 22:46:47 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #454 from terrelln/pzstd-small-files

Browse Source 
[pzstd] Fix small file frame size
This commit is contained in:
Yann Collet 2016-11-16 15:45:18 -08:00 committed by GitHub
commit 7ab956eb8b
2 changed files with 18 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
contrib/pzstd/Pzstd.cpp
View File

@ -341,15 +341,9 @@ static size_t calculateStep(
std::uintmax_t size,
size_t numThreads,
const ZSTD_parameters &params) {
size_t step = size_t{1} << (params.cParams.windowLog + 2);
// If file size is known, see if a smaller step will spread work more evenly
if (size != 0) {
const std::uintmax_t newStep = size / numThreads;
if (newStep != 0 && newStep <= std::numeric_limits<size_t>::max()) {
step = std::min(step, static_cast<size_t>(newStep));
}
}
return step;
(void)size;
(void)numThreads;
return size_t{1} << (params.cParams.windowLog + 2);
}
namespace {
@ -401,6 +395,7 @@ std::uint64_t asyncCompressChunks(
// Break the input up into chunks of size `step` and compress each chunk
// independently.
size_t step = calculateStep(size, numThreads, params);
state.log(DEBUG, "Chosen frame size: %zu\n", step);
auto status = FileStatus::Continue;
while (status == FileStatus::Continue && !state.errorHolder.hasError()) {
// Make a new input queue that we will put the chunk's input data into.
@ -415,6 +410,7 @@ std::uint64_t asyncCompressChunks(
});
// Pass the output queue to the writer thread.
chunks.push(std::move(out));
state.log(VERBOSE, "Starting a new frame\n");
// Fill the input queue for the compression job we just started
status = readData(*in, ZSTD_CStreamInSize(), step, fd, &bytesRead);
}
@ -551,11 +547,14 @@ std::uint64_t asyncDecompressFrames(
if (frameSize == 0) {
// We hit a non SkippableFrame ==> not compressed by pzstd or corrupted
// Pass the rest of the source to this decompression task
state.log(VERBOSE,
"Input not in pzstd format, falling back to serial decompression\n");
while (status == FileStatus::Continue && !state.errorHolder.hasError()) {
status = readData(*in, chunkSize, chunkSize, fd, &totalBytesRead);
}
break;
}
state.log(VERBOSE, "Decompressing a frame of size %zu", frameSize);
// Fill the input queue for the decompression job we just started
status = readData(*in, chunkSize, frameSize, fd, &totalBytesRead);
}

12
contrib/pzstd/Pzstd.h
View File

@ -40,7 +40,8 @@ class SharedState {
if (!options.decompress) {
auto parameters = options.determineParameters();
cStreamPool.reset(new ResourcePool<ZSTD_CStream>{
[parameters]() -> ZSTD_CStream* {
[this, parameters]() -> ZSTD_CStream* {
this->log(VERBOSE, "Creating new ZSTD_CStream\n");
auto zcs = ZSTD_createCStream();
if (zcs) {
auto err = ZSTD_initCStream_advanced(
@ -57,7 +58,8 @@ class SharedState {
}});
} else {
dStreamPool.reset(new ResourcePool<ZSTD_DStream>{
[]() -> ZSTD_DStream* {
[this]() -> ZSTD_DStream* {
this->log(VERBOSE, "Creating new ZSTD_DStream\n");
auto zds = ZSTD_createDStream();
if (zds) {
auto err = ZSTD_initDStream(zds);
@ -74,6 +76,12 @@ class SharedState {
}
}
~SharedState() {
// The resource pools have references to this, so destroy them first.
cStreamPool.reset();
dStreamPool.reset();
}
Logger log;
ErrorHolder errorHolder;
std::unique_ptr<ResourcePool<ZSTD_CStream>> cStreamPool;