From 1e2f6a1f5df5fe7b24ad8df731a80769095910b8 Mon Sep 17 00:00:00 2001 From: Nick Terrell Date: Fri, 2 Sep 2016 12:23:49 -0700 Subject: [PATCH 1/4] Clean up compiler warnings + Build pzstd on travis --- .travis.yml | 2 +- contrib/pzstd/ErrorHolder.h | 5 ++--- contrib/pzstd/Makefile | 4 ++-- contrib/pzstd/SkippableFrame.cpp | 2 +- 4 files changed, 6 insertions(+), 7 deletions(-) diff --git a/.travis.yml b/.travis.yml index 3b77dd94a..d5479a974 100644 --- a/.travis.yml +++ b/.travis.yml @@ -19,7 +19,7 @@ matrix: packages: - gcc-4.8 - g++-4.8 - env: PLATFORM="Ubuntu 12.04 container" CMD="make -C tests test-zstd_nolegacy && make clean && make zlibwrapper && make clean && make cmaketest && make -C contrib/pzstd googletest && make -C contrib/pzstd test && make -C contrib/pzstd clean" + env: PLATFORM="Ubuntu 12.04 container" CMD="make -C tests test-zstd_nolegacy && make clean && make zlibwrapper && make clean && make cmaketest && make clean && make -C contrib/pzstd pzstd && make -C contrib/pzstd googletest && make -C contrib/pzstd test && make -C contrib/pzstd clean" - os: linux sudo: false env: PLATFORM="Ubuntu 12.04 container" CMD="make usan" diff --git a/contrib/pzstd/ErrorHolder.h b/contrib/pzstd/ErrorHolder.h index 4a81a068c..188badcad 100644 --- a/contrib/pzstd/ErrorHolder.h +++ b/contrib/pzstd/ErrorHolder.h @@ -9,6 +9,7 @@ #pragma once #include +#include #include #include @@ -47,9 +48,7 @@ class ErrorHolder { } ~ErrorHolder() { - if (hasError()) { - throw std::logic_error(message_); - } + assert(!hasError()); } }; } diff --git a/contrib/pzstd/Makefile b/contrib/pzstd/Makefile index 6f9231b2a..5338a5a9e 100644 --- a/contrib/pzstd/Makefile +++ b/contrib/pzstd/Makefile @@ -10,7 +10,7 @@ ZSTDDIR = ../../lib PROGDIR = ../../programs -CPPFLAGS = -I$(ZSTDDIR) -I$(ZSTDDIR)/common -I$(ZSTDDIR)/dictBuilder -I$(PROGDIR) -I. +CPPFLAGS = -I$(ZSTDDIR) -I$(ZSTDDIR)/common -I$(PROGDIR) -I. CXXFLAGS ?= -O3 CXXFLAGS += -std=c++11 CXXFLAGS += $(MOREFLAGS) @@ -55,7 +55,7 @@ main.o: main.cpp *.h utils/*.h $(CXX) $(FLAGS) -c main.cpp -o $@ pzstd: Pzstd.o SkippableFrame.o Options.o main.o libzstd.a - $(CXX) $(FLAGS) $^ -o $@$(EXT) + $(CXX) $(FLAGS) $^ -o $@$(EXT) -lpthread googletest: @git clone https://github.com/google/googletest diff --git a/contrib/pzstd/SkippableFrame.cpp b/contrib/pzstd/SkippableFrame.cpp index 20ad4cc8e..5dc95e5ab 100644 --- a/contrib/pzstd/SkippableFrame.cpp +++ b/contrib/pzstd/SkippableFrame.cpp @@ -7,7 +7,7 @@ * of patent rights can be found in the PATENTS file in the same directory. */ #include "SkippableFrame.h" -#include "common/mem.h" +#include "mem.h" #include "utils/Range.h" #include From ac14348a283b54603f60353cd3763f9434b73616 Mon Sep 17 00:00:00 2001 From: Nick Terrell Date: Fri, 2 Sep 2016 12:35:36 -0700 Subject: [PATCH 2/4] When reading from stdin, write to stdout by default --- contrib/pzstd/Options.cpp | 30 ++++++++++++++---------------- 1 file changed, 14 insertions(+), 16 deletions(-) diff --git a/contrib/pzstd/Options.cpp b/contrib/pzstd/Options.cpp index 693e9a515..616130996 100644 --- a/contrib/pzstd/Options.cpp +++ b/contrib/pzstd/Options.cpp @@ -146,23 +146,21 @@ bool Options::parse(int argc, const char** argv) { // Determine output file if not specified if (outputFile.empty()) { if (inputFile == "-") { - std::fprintf( - stderr, - "Invalid arguments: Reading from stdin, but -o not provided.\n"); - return false; - } - // Attempt to add/remove zstd extension from the input file - if (decompress) { - int stemSize = inputFile.size() - zstdExtension.size(); - if (stemSize > 0 && inputFile.substr(stemSize) == zstdExtension) { - outputFile = inputFile.substr(0, stemSize); - } else { - std::fprintf( - stderr, "Invalid argument: Unable to determine output file.\n"); - return false; - } + outputFile = "-"; } else { - outputFile = inputFile + zstdExtension; + // Attempt to add/remove zstd extension from the input file + if (decompress) { + int stemSize = inputFile.size() - zstdExtension.size(); + if (stemSize > 0 && inputFile.substr(stemSize) == zstdExtension) { + outputFile = inputFile.substr(0, stemSize); + } else { + std::fprintf( + stderr, "Invalid argument: Unable to determine output file.\n"); + return false; + } + } else { + outputFile = inputFile + zstdExtension; + } } } // Check compression level From 2fcf8a4b997ae0e6c5ca3b59685ed488bf1d11bb Mon Sep 17 00:00:00 2001 From: Nick Terrell Date: Fri, 2 Sep 2016 12:59:14 -0700 Subject: [PATCH 3/4] Update tests to reflect new default options --- contrib/pzstd/test/OptionsTest.cpp | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/contrib/pzstd/test/OptionsTest.cpp b/contrib/pzstd/test/OptionsTest.cpp index 1479d6cd2..87e79d59e 100644 --- a/contrib/pzstd/test/OptionsTest.cpp +++ b/contrib/pzstd/test/OptionsTest.cpp @@ -106,6 +106,16 @@ TEST(Options, ValidInputs) { Options expected = { 2, 23, 3, false, "silesia.tar", "silesia.tar.pzstd", false, false}; } + { + Options options; + std::array args = {{nullptr, "-n", "1"}}; + EXPECT_TRUE(options.parse(args.size(), args.data())); + } + { + Options options; + std::array args = {{nullptr, "-", "-n", "1"}}; + EXPECT_TRUE(options.parse(args.size(), args.data())); + } } TEST(Options, BadNumThreads) { @@ -153,16 +163,6 @@ TEST(Options, BadOutputFile) { std::array args = {{nullptr, "notzst", "-d", "-n", "1"}}; EXPECT_FALSE(options.parse(args.size(), args.data())); } - { - Options options; - std::array args = {{nullptr, "-n", "1"}}; - EXPECT_FALSE(options.parse(args.size(), args.data())); - } - { - Options options; - std::array args = {{nullptr, "-", "-n", "1"}}; - EXPECT_FALSE(options.parse(args.size(), args.data())); - } } TEST(Options, Extras) { From 64c1c065cc48dfc7bdcbe06f3fd374a37903e934 Mon Sep 17 00:00:00 2001 From: Nick Terrell Date: Fri, 2 Sep 2016 13:53:23 -0700 Subject: [PATCH 4/4] Add optional max size to work queue --- contrib/pzstd/Makefile | 2 +- contrib/pzstd/utils/WorkQueue.h | 59 ++++++++++++++------ contrib/pzstd/utils/test/Makefile | 2 +- contrib/pzstd/utils/test/WorkQueueTest.cpp | 65 ++++++++++++++++++++++ 4 files changed, 108 insertions(+), 20 deletions(-) diff --git a/contrib/pzstd/Makefile b/contrib/pzstd/Makefile index 5338a5a9e..c59a6d107 100644 --- a/contrib/pzstd/Makefile +++ b/contrib/pzstd/Makefile @@ -70,5 +70,5 @@ clean: $(MAKE) -C $(ZSTDDIR) clean $(MAKE) -C utils/test clean $(MAKE) -C test clean - @$(RM) -rf googletest/ libzstd.a *.o pzstd$(EXT) + @$(RM) -rf libzstd.a *.o pzstd$(EXT) @echo Cleaning completed diff --git a/contrib/pzstd/utils/WorkQueue.h b/contrib/pzstd/utils/WorkQueue.h index 3d926cc80..2fa417f41 100644 --- a/contrib/pzstd/utils/WorkQueue.h +++ b/contrib/pzstd/utils/WorkQueue.h @@ -12,6 +12,7 @@ #include #include +#include #include #include #include @@ -25,14 +26,29 @@ template class WorkQueue { // Protects all member variable access std::mutex mutex_; - std::condition_variable cv_; + std::condition_variable readerCv_; + std::condition_variable writerCv_; std::queue queue_; bool done_; + std::size_t maxSize_; + + // Must have lock to call this function + bool full() const { + if (maxSize_ == 0) { + return false; + } + return queue_.size() >= maxSize_; + } public: - /// Constructs an empty work queue. - WorkQueue() : done_(false) {} + /** + * Constructs an empty work queue with an optional max size. + * If `maxSize == 0` the queue size is unbounded. + * + * @param maxSize The maximum allowed size of the work queue. + */ + WorkQueue(std::size_t maxSize = 0) : done_(false), maxSize_(maxSize) {} /** * Push an item onto the work queue. Notify a single thread that work is @@ -44,13 +60,16 @@ class WorkQueue { */ bool push(T item) { { - std::lock_guard lock(mutex_); + std::unique_lock lock(mutex_); + while (full() && !done_) { + writerCv_.wait(lock); + } if (done_) { return false; } queue_.push(std::move(item)); } - cv_.notify_one(); + readerCv_.notify_one(); return true; } @@ -64,16 +83,19 @@ class WorkQueue { * `finish()` has been called. */ bool pop(T& item) { - std::unique_lock lock(mutex_); - while (queue_.empty() && !done_) { - cv_.wait(lock); + { + std::unique_lock lock(mutex_); + while (queue_.empty() && !done_) { + readerCv_.wait(lock); + } + if (queue_.empty()) { + assert(done_); + return false; + } + item = std::move(queue_.front()); + queue_.pop(); } - if (queue_.empty()) { - assert(done_); - return false; - } - item = std::move(queue_.front()); - queue_.pop(); + writerCv_.notify_one(); return true; } @@ -87,18 +109,19 @@ class WorkQueue { assert(!done_); done_ = true; } - cv_.notify_all(); + readerCv_.notify_all(); + writerCv_.notify_all(); } /// Blocks until `finish()` has been called (but the queue may not be empty). void waitUntilFinished() { std::unique_lock lock(mutex_); while (!done_) { - cv_.wait(lock); + readerCv_.wait(lock); // If we were woken by a push, we need to wake a thread waiting on pop(). if (!done_) { lock.unlock(); - cv_.notify_one(); + readerCv_.notify_one(); lock.lock(); } } @@ -111,7 +134,7 @@ class BufferWorkQueue { std::atomic size_; public: - BufferWorkQueue() : size_(0) {} + BufferWorkQueue(std::size_t maxSize = 0) : queue_(maxSize), size_(0) {} void push(Buffer buffer) { size_.fetch_add(buffer.size()); diff --git a/contrib/pzstd/utils/test/Makefile b/contrib/pzstd/utils/test/Makefile index 23f111e55..b9ea73e32 100644 --- a/contrib/pzstd/utils/test/Makefile +++ b/contrib/pzstd/utils/test/Makefile @@ -23,7 +23,7 @@ GTEST_LIB ?= -L $(PZSTDDIR)/googletest/build/googlemock/gtest CPPFLAGS = -I$(PZSTDDIR) $(GTEST_INC) $(GTEST_LIB) CXXFLAGS ?= -O3 CXXFLAGS += -std=c++11 -CFLAGS += $(MOREFLAGS) +CXXFLAGS += $(MOREFLAGS) FLAGS = $(CPPFLAGS) $(CXXFLAGS) $(LDFLAGS) %: %.cpp diff --git a/contrib/pzstd/utils/test/WorkQueueTest.cpp b/contrib/pzstd/utils/test/WorkQueueTest.cpp index 1b548d160..074891fda 100644 --- a/contrib/pzstd/utils/test/WorkQueueTest.cpp +++ b/contrib/pzstd/utils/test/WorkQueueTest.cpp @@ -145,6 +145,71 @@ TEST(WorkQueue, MPMC) { } } +TEST(WorkQueue, BoundedSizeWorks) { + WorkQueue queue(1); + int result; + queue.push(5); + queue.pop(result); + queue.push(5); + queue.pop(result); + queue.push(5); + queue.finish(); + queue.pop(result); + EXPECT_EQ(5, result); +} + +TEST(WorkQueue, BoundedSizePushAfterFinish) { + WorkQueue queue(1); + int result; + queue.push(5); + std::thread pusher([&queue] { + queue.push(6); + }); + // Dirtily try and make sure that pusher has run. + std::this_thread::sleep_for(std::chrono::seconds(1)); + queue.finish(); + EXPECT_TRUE(queue.pop(result)); + EXPECT_EQ(5, result); + EXPECT_FALSE(queue.pop(result)); + + pusher.join(); +} + +TEST(WorkQueue, BoundedSizeMPMC) { + WorkQueue queue(100); + std::vector results(10000, -1); + std::mutex mutex; + std::vector popperThreads; + for (int i = 0; i < 10; ++i) { + popperThreads.emplace_back(Popper{&queue, results.data(), &mutex}); + } + + std::vector pusherThreads; + for (int i = 0; i < 100; ++i) { + auto min = i * 100; + auto max = (i + 1) * 100; + pusherThreads.emplace_back( + [ &queue, min, max ] { + for (int i = min; i < max; ++i) { + queue.push(i); + } + }); + } + + for (auto& thread : pusherThreads) { + thread.join(); + } + queue.finish(); + + for (auto& thread : popperThreads) { + thread.join(); + } + + for (int i = 0; i < 10000; ++i) { + EXPECT_EQ(i, results[i]); + } +} + TEST(BufferWorkQueue, SizeCalculatedCorrectly) { { BufferWorkQueue queue;