3 files changed, 32 insertions, 3 deletions

diff --git a/contrib/pzstd/Options.cpp b/contrib/pzstd/Options.cpp
index a0d969393c9b4..1f53f2bff78a8 100644
--- a/contrib/pzstd/Options.cpp
+++ b/contrib/pzstd/Options.cpp
@@ -91,7 +91,7 @@ void usage() {
   std::fprintf(stderr, "  -#                     : # compression level (1-%d, default:%d)\n", kMaxNonUltraCompressionLevel, kDefaultCompressionLevel);
   std::fprintf(stderr, "  -d, --decompress       : decompression\n");
   std::fprintf(stderr, "  -o                file : result stored into `file` (only if 1 input file)\n");
-  std::fprintf(stderr, "  -f, --force            : overwrite output without prompting\n");
+  std::fprintf(stderr, "  -f, --force            : overwrite output without prompting, (de)compress links\n");
   std::fprintf(stderr, "      --rm               : remove source file(s) after successful (de)compression\n");
   std::fprintf(stderr, "  -k, --keep             : preserve source file(s) (default)\n");
   std::fprintf(stderr, "  -h, --help             : display help and exit\n");
@@ -121,6 +121,7 @@ Options::Status Options::parse(int argc, const char **argv) {
   bool recursive = false;
   bool ultra = false;
   bool forceStdout = false;
+  bool followLinks = false;
   // Local copy of input files, which are pointers into argv.
   std::vector<const char *> localInputFiles;
   for (int i = 1; i < argc; ++i) {
@@ -255,6 +256,7 @@ Options::Status Options::parse(int argc, const char **argv) {
       case 'f':
         overwrite = true;
         forceStdout = true;
+        followLinks = true;
         break;
       case 't':
         test = true;
@@ -328,13 +330,29 @@ Options::Status Options::parse(int argc, const char **argv) {
     }
   }
 
+  g_utilDisplayLevel = verbosity;
+  // Remove local input files that are symbolic links
+  if (!followLinks) {
+      std::remove_if(localInputFiles.begin(), localInputFiles.end(),
+                     [&](const char *path) {
+                        bool isLink = UTIL_isLink(path);
+                        if (isLink && verbosity >= 2) {
+                            std::fprintf(
+                                    stderr,
+                                    "Warning : %s is symbolic link, ignoring\n",
+                                    path);
+                        }
+                        return isLink;
+                    });
+  }
+
   // Translate input files/directories into files to (de)compress
   if (recursive) {
     char *scratchBuffer = nullptr;
     unsigned numFiles = 0;
     const char **files =
         UTIL_createFileList(localInputFiles.data(), localInputFiles.size(),
-                            &scratchBuffer, &numFiles);
+                            &scratchBuffer, &numFiles, followLinks);
     if (files == nullptr) {
       std::fprintf(stderr, "Error traversing directories\n");
       return Status::Failure;
diff --git a/contrib/pzstd/utils/test/ThreadPoolTest.cpp b/contrib/pzstd/utils/test/ThreadPoolTest.cpp
index 1d857aae808da..89085afd434ca 100644
--- a/contrib/pzstd/utils/test/ThreadPoolTest.cpp
+++ b/contrib/pzstd/utils/test/ThreadPoolTest.cpp
@@ -10,6 +10,7 @@
 
 #include <gtest/gtest.h>
 #include <atomic>
+#include <iostream>
 #include <thread>
 #include <vector>
 
@@ -34,16 +35,19 @@ TEST(ThreadPool, AllJobsFinished) {
   std::atomic<unsigned> numFinished{0};
   std::atomic<bool> start{false};
   {
+    std::cerr << "Creating executor" << std::endl;
     ThreadPool executor(5);
     for (int i = 0; i < 10; ++i) {
       executor.add([ &numFinished, &start ] {
         while (!start.load()) {
-          // spin
+          std::this_thread::yield();
         }
         ++numFinished;
       });
     }
+    std::cerr << "Starting" << std::endl;
     start.store(true);
+    std::cerr << "Finishing" << std::endl;
   }
   EXPECT_EQ(10, numFinished.load());
 }
diff --git a/contrib/pzstd/utils/test/WorkQueueTest.cpp b/contrib/pzstd/utils/test/WorkQueueTest.cpp
index 7f58ccb3f1997..8caf170d2948d 100644
--- a/contrib/pzstd/utils/test/WorkQueueTest.cpp
+++ b/contrib/pzstd/utils/test/WorkQueueTest.cpp
@@ -10,6 +10,7 @@
 #include "utils/WorkQueue.h"
 
 #include <gtest/gtest.h>
+#include <iostream>
 #include <memory>
 #include <mutex>
 #include <thread>
@@ -201,11 +202,13 @@ TEST(WorkQueue, BoundedSizeMPMC) {
   WorkQueue<int> queue(10);
   std::vector<int> results(200, -1);
   std::mutex mutex;
+  std::cerr << "Creating popperThreads" << std::endl;
   std::vector<std::thread> popperThreads;
   for (int i = 0; i < 4; ++i) {
     popperThreads.emplace_back(Popper{&queue, results.data(), &mutex});
   }
 
+  std::cerr << "Creating pusherThreads" << std::endl;
   std::vector<std::thread> pusherThreads;
   for (int i = 0; i < 2; ++i) {
     auto min = i * 100;
@@ -218,15 +221,19 @@ TEST(WorkQueue, BoundedSizeMPMC) {
         });
   }
 
+  std::cerr << "Joining pusherThreads" << std::endl;
   for (auto& thread : pusherThreads) {
     thread.join();
   }
+  std::cerr << "Finishing queue" << std::endl;
   queue.finish();
 
+  std::cerr << "Joining popperThreads" << std::endl;
   for (auto& thread : popperThreads) {
     thread.join();
   }
 
+  std::cerr << "Inspecting results" << std::endl;
   for (int i = 0; i < 200; ++i) {
     EXPECT_EQ(i, results[i]);
   }