Merge pull request #332 from terrelln/dev

[pzstd] linux build and CLI fixes

.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"

contrib/pzstd/ErrorHolder.h

@@ -9,6 +9,7 @@
 #pragma once
 
 #include <atomic>
+#include <cassert>
 #include <stdexcept>
 #include <string>
 
@@ -47,9 +48,7 @@ class ErrorHolder {
   }
 
   ~ErrorHolder() {
-    if (hasError()) {
-      throw std::logic_error(message_);
-    }
+    assert(!hasError());
   }
 };
 }

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
@@ -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

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

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 <cstdio>

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<const char*, 3> args = {{nullptr, "-n", "1"}};
+    EXPECT_TRUE(options.parse(args.size(), args.data()));
+  }
+  {
+    Options options;
+    std::array<const char*, 4> args = {{nullptr, "-", "-n", "1"}};
+    EXPECT_TRUE(options.parse(args.size(), args.data()));
+  }
 }
 
 TEST(Options, BadNumThreads) {
@@ -153,16 +163,6 @@ TEST(Options, BadOutputFile) {
     std::array<const char*, 5> args = {{nullptr, "notzst", "-d", "-n", "1"}};
     EXPECT_FALSE(options.parse(args.size(), args.data()));
   }
-  {
-    Options options;
-    std::array<const char*, 3> args = {{nullptr, "-n", "1"}};
-    EXPECT_FALSE(options.parse(args.size(), args.data()));
-  }
-  {
-    Options options;
-    std::array<const char*, 4> args = {{nullptr, "-", "-n", "1"}};
-    EXPECT_FALSE(options.parse(args.size(), args.data()));
-  }
 }
 
 TEST(Options, Extras) {

contrib/pzstd/utils/WorkQueue.h

@@ -12,6 +12,7 @@
 
 #include <atomic>
 #include <cassert>
+#include <cstddef>
 #include <condition_variable>
 #include <cstddef>
 #include <functional>
@@ -25,14 +26,29 @@ template <typename T>
 class WorkQueue {
   // Protects all member variable access
   std::mutex mutex_;
-  std::condition_variable cv_;
+  std::condition_variable readerCv_;
+  std::condition_variable writerCv_;
 
   std::queue<T> 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<std::mutex> lock(mutex_);
+      std::unique_lock<std::mutex> 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<std::mutex> lock(mutex_);
-    while (queue_.empty() && !done_) {
-      cv_.wait(lock);
+    {
+      std::unique_lock<std::mutex> 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<std::mutex> 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<std::size_t> size_;
 
  public:
-  BufferWorkQueue() : size_(0) {}
+  BufferWorkQueue(std::size_t maxSize = 0) : queue_(maxSize), size_(0) {}
 
   void push(Buffer buffer) {
     size_.fetch_add(buffer.size());

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

contrib/pzstd/utils/test/WorkQueueTest.cpp

@@ -145,6 +145,71 @@ TEST(WorkQueue, MPMC) {
   }
 }
 
+TEST(WorkQueue, BoundedSizeWorks) {
+  WorkQueue<int> 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<int> 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<int> queue(100);
+  std::vector<int> results(10000, -1);
+  std::mutex mutex;
+  std::vector<std::thread> popperThreads;
+  for (int i = 0; i < 10; ++i) {
+    popperThreads.emplace_back(Popper{&queue, results.data(), &mutex});
+  }
+
+  std::vector<std::thread> 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;