lib/zstd
1
0
Fork 0
mirror of https://github.com/facebook/zstd.git synced 2025-07-16 07:01:58 +03:00
Code Activity

[contrib/pzstd] Prevent hangs when there are errors

Browse Source 
When two threads are using a WorkQueue and the reader thread exits due
to an error, it must call WorkQueue::finish() to wake up the writer
thread. Otherwise, if the queue is full and the writer thread is waiting
for a free slot, it could hang forever.

This can happen in pratice when decompressing a large, corrupted file
that does not contain pzstd skippable frames.
This commit is contained in:
Sean Bartell
2024-06-20 15:52:43 -05:00
committed by Nick Terrell
parent a610550e2c
commit 80af41e08a
2 changed files with 14 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
contrib/pzstd/Pzstd.cpp
View File

@ -269,7 +269,10 @@ static void compress(
std::shared_ptr<BufferWorkQueue> out, std::shared_ptr<BufferWorkQueue> out,
size_t maxInputSize) { size_t maxInputSize) {
auto& errorHolder = state.errorHolder; auto& errorHolder = state.errorHolder;
auto guard = makeScopeGuard([&] { out->finish(); }); auto guard = makeScopeGuard([&] {
in->finish();
out->finish();
});
// Initialize the CCtx // Initialize the CCtx
auto ctx = state.cStreamPool->get(); auto ctx = state.cStreamPool->get();
if (!errorHolder.check(ctx != nullptr, "Failed to allocate ZSTD_CStream")) { if (!errorHolder.check(ctx != nullptr, "Failed to allocate ZSTD_CStream")) {
@ -431,7 +434,10 @@ static void decompress(
std::shared_ptr<BufferWorkQueue> in, std::shared_ptr<BufferWorkQueue> in,
std::shared_ptr<BufferWorkQueue> out) { std::shared_ptr<BufferWorkQueue> out) {
auto& errorHolder = state.errorHolder; auto& errorHolder = state.errorHolder;
auto guard = makeScopeGuard([&] { out->finish(); }); auto guard = makeScopeGuard([&] {
in->finish();
out->finish();
});
// Initialize the DCtx // Initialize the DCtx
auto ctx = state.dStreamPool->get(); auto ctx = state.dStreamPool->get();
if (!errorHolder.check(ctx != nullptr, "Failed to allocate ZSTD_DStream")) { if (!errorHolder.check(ctx != nullptr, "Failed to allocate ZSTD_DStream")) {
@ -578,6 +584,7 @@ std::uint64_t writeFile(
FILE* outputFd, FILE* outputFd,
bool decompress) { bool decompress) {
auto& errorHolder = state.errorHolder; auto& errorHolder = state.errorHolder;
auto outsFinishGuard = makeScopeGuard([&outs] { outs.finish(); });
auto lineClearGuard = makeScopeGuard([&state] { auto lineClearGuard = makeScopeGuard([&state] {
state.log.clear(kLogInfo); state.log.clear(kLogInfo);
}); });
@ -585,6 +592,7 @@ std::uint64_t writeFile(
std::shared_ptr<BufferWorkQueue> out; std::shared_ptr<BufferWorkQueue> out;
// Grab the output queue for each decompression job (in order). // Grab the output queue for each decompression job (in order).
while (outs.pop(out)) { while (outs.pop(out)) {
auto outFinishGuard = makeScopeGuard([&out] { out->finish(); });
if (errorHolder.hasError()) { if (errorHolder.hasError()) {
continue; continue;
} }

7
contrib/pzstd/utils/WorkQueue.h
View File

@ -115,13 +115,14 @@ class WorkQueue {
} }
/** /**
* Promise that `push()` won't be called again, so once the queue is empty * Promise that either the reader side or the writer side is done.
* there will never any more work. * If the writer is done, `push()` won't be called again, so once the queue
* is empty there will never be any more work. If the reader is done, `pop()`
* won't be called again, so further items pushed will just be ignored.
*/ */
void finish() { void finish() {
{ {
std::lock_guard<std::mutex> lock(mutex_); std::lock_guard<std::mutex> lock(mutex_);
assert(!done_);
done_ = true; done_ = true;
} }
readerCv_.notify_all(); readerCv_.notify_all();