ARROW-3777: [C++] Add Slow input streams and slow filesystem

cpp/build-support/run_clang_format.py

@@ -28,12 +28,10 @@
 
 # examine the output of clang-format and if changes are
 # present assemble a (unified)patch of the difference
-def _check_one_file(completed_processes, filename):
+def _check_one_file(filename, formatted):
     with open(filename, "rb") as reader:
         original = reader.read()
 
-    returncode, stdout, stderr = completed_processes[filename]
-    formatted = stdout
     if formatted != original:
         # Run the equivalent of diff -u
         diff = list(difflib.unified_diff(
@@ -106,20 +104,21 @@ def _check_one_file(completed_processes, filename):
             [arguments.clang_format_binary, filename]
             for filename in formatted_filenames
         ], stdout=PIPE, stderr=PIPE)
-        for returncode, stdout, stderr in results:
+
+        checker_args = []
+        for filename, res in zip(formatted_filenames, results):
             # if any clang-format reported a parse error, bubble it
+            returncode, stdout, stderr = res
             if returncode != 0:
+                print(stderr)
                 sys.exit(returncode)
+            checker_args.append((filename, stdout))
 
         error = False
-        checker = partial(_check_one_file, {
-            filename: result
-            for filename, result in zip(formatted_filenames, results)
-        })
         pool = mp.Pool()
         try:
             # check the output from each invocation of clang-format in parallel
-            for filename, diff in pool.imap(checker, formatted_filenames):
+            for filename, diff in pool.starmap(_check_one_file, checker_args):
                 if not arguments.quiet:
                     print("Checking {}".format(filename))
                 if diff:

cpp/src/arrow/CMakeLists.txt

@@ -127,6 +127,7 @@ set(ARROW_SRCS
     io/interfaces.cc
     io/memory.cc
     io/readahead.cc
+    io/slow.cc
     testing/util.cc
     util/basic_decimal.cc
     util/bit_util.cc

cpp/src/arrow/filesystem/filesystem.cc

@@ -19,6 +19,7 @@
 
 #include "arrow/filesystem/filesystem.h"
 #include "arrow/filesystem/path_util.h"
+#include "arrow/io/slow.h"
 #include "arrow/util/logging.h"
 #include "arrow/util/macros.h"
 
@@ -75,6 +76,10 @@ std::ostream& operator<<(std::ostream& os, const FileStats& stats) {
   return os << "FileStats(" << stats.type() << ", " << stats.path() << ")";
 }
 
+std::string FileStats::extension() const {
+  return internal::GetAbstractPathExtension(path_);
+}
+
 //////////////////////////////////////////////////////////////////////////
 // FileSystem default method implementations
 
@@ -232,8 +237,91 @@ Status SubTreeFileSystem::OpenAppendStream(const std::string& path,
   return base_fs_->OpenAppendStream(s, out);
 }
 
-std::string FileStats::extension() const {
-  return internal::GetAbstractPathExtension(path_);
+//////////////////////////////////////////////////////////////////////////
+// SlowFileSystem implementation
+
+SlowFileSystem::SlowFileSystem(std::shared_ptr<FileSystem> base_fs,
+                               std::shared_ptr<io::LatencyGenerator> latencies)
+    : base_fs_(base_fs), latencies_(latencies) {}
+
+SlowFileSystem::SlowFileSystem(std::shared_ptr<FileSystem> base_fs,
+                               double average_latency)
+    : base_fs_(base_fs), latencies_(io::LatencyGenerator::Make(average_latency)) {}
+
+SlowFileSystem::SlowFileSystem(std::shared_ptr<FileSystem> base_fs,
+                               double average_latency, int32_t seed)
+    : base_fs_(base_fs), latencies_(io::LatencyGenerator::Make(average_latency, seed)) {}
+
+Status SlowFileSystem::GetTargetStats(const std::string& path, FileStats* out) {
+  latencies_->Sleep();
+  return base_fs_->GetTargetStats(path, out);
+}
+
+Status SlowFileSystem::GetTargetStats(const Selector& selector,
+                                      std::vector<FileStats>* out) {
+  latencies_->Sleep();
+  return base_fs_->GetTargetStats(selector, out);
+}
+
+Status SlowFileSystem::CreateDir(const std::string& path, bool recursive) {
+  latencies_->Sleep();
+  return base_fs_->CreateDir(path, recursive);
+}
+
+Status SlowFileSystem::DeleteDir(const std::string& path) {
+  latencies_->Sleep();
+  return base_fs_->DeleteDir(path);
+}
+
+Status SlowFileSystem::DeleteDirContents(const std::string& path) {
+  latencies_->Sleep();
+  return base_fs_->DeleteDirContents(path);
+}
+
+Status SlowFileSystem::DeleteFile(const std::string& path) {
+  latencies_->Sleep();
+  return base_fs_->DeleteFile(path);
+}
+
+Status SlowFileSystem::Move(const std::string& src, const std::string& dest) {
+  latencies_->Sleep();
+  return base_fs_->Move(src, dest);
+}
+
+Status SlowFileSystem::CopyFile(const std::string& src, const std::string& dest) {
+  latencies_->Sleep();
+  return base_fs_->CopyFile(src, dest);
+}
+
+Status SlowFileSystem::OpenInputStream(const std::string& path,
+                                       std::shared_ptr<io::InputStream>* out) {
+  latencies_->Sleep();
+  std::shared_ptr<io::InputStream> stream;
+  RETURN_NOT_OK(base_fs_->OpenInputStream(path, &stream));
+  *out = std::make_shared<io::SlowInputStream>(stream, latencies_);
+  return Status::OK();
+}
+
+Status SlowFileSystem::OpenInputFile(const std::string& path,
+                                     std::shared_ptr<io::RandomAccessFile>* out) {
+  latencies_->Sleep();
+  std::shared_ptr<io::RandomAccessFile> file;
+  RETURN_NOT_OK(base_fs_->OpenInputFile(path, &file));
+  *out = std::make_shared<io::SlowRandomAccessFile>(file, latencies_);
+  return Status::OK();
+}
+
+Status SlowFileSystem::OpenOutputStream(const std::string& path,
+                                        std::shared_ptr<io::OutputStream>* out) {
+  latencies_->Sleep();
+  // XXX Should we have a SlowOutputStream that waits on Flush() and Close()?
+  return base_fs_->OpenOutputStream(path, out);
+}
+
+Status SlowFileSystem::OpenAppendStream(const std::string& path,
+                                        std::shared_ptr<io::OutputStream>* out) {
+  latencies_->Sleep();
+  return base_fs_->OpenAppendStream(path, out);
 }
 
 }  // namespace fs

cpp/src/arrow/filesystem/filesystem.h

@@ -25,7 +25,6 @@
 #include <vector>
 
 #include "arrow/status.h"
-#include "arrow/util/compression.h"
 #include "arrow/util/visibility.h"
 
 // The Windows API defines macros from *File resolving to either
@@ -44,6 +43,7 @@ namespace arrow {
 namespace io {
 
 class InputStream;
+class LatencyGenerator;
 class OutputStream;
 class RandomAccessFile;
 
@@ -265,5 +265,47 @@ class ARROW_EXPORT SubTreeFileSystem : public FileSystem {
   Status FixStats(FileStats* st) const;
 };
 
+/// \brief EXPERIMENTAL: a FileSystem implementation that delegates to another
+/// implementation but inserts latencies at various points.
+class ARROW_EXPORT SlowFileSystem : public FileSystem {
+ public:
+  SlowFileSystem(std::shared_ptr<FileSystem> base_fs,
+                 std::shared_ptr<io::LatencyGenerator> latencies);
+  SlowFileSystem(std::shared_ptr<FileSystem> base_fs, double average_latency);
+  SlowFileSystem(std::shared_ptr<FileSystem> base_fs, double average_latency,
+                 int32_t seed);
+
+  using FileSystem::GetTargetStats;
+  Status GetTargetStats(const std::string& path, FileStats* out) override;
+  Status GetTargetStats(const Selector& select, std::vector<FileStats>* out) override;
+
+  Status CreateDir(const std::string& path, bool recursive = true) override;
+
+  Status DeleteDir(const std::string& path) override;
+  Status DeleteDirContents(const std::string& path) override;
+
+  Status DeleteFile(const std::string& path) override;
+
+  Status Move(const std::string& src, const std::string& dest) override;
+
+  Status CopyFile(const std::string& src, const std::string& dest) override;
+
+  Status OpenInputStream(const std::string& path,
+                         std::shared_ptr<io::InputStream>* out) override;
+
+  Status OpenInputFile(const std::string& path,
+                       std::shared_ptr<io::RandomAccessFile>* out) override;
+
+  Status OpenOutputStream(const std::string& path,
+                          std::shared_ptr<io::OutputStream>* out) override;
+
+  Status OpenAppendStream(const std::string& path,
+                          std::shared_ptr<io::OutputStream>* out) override;
+
+ protected:
+  std::shared_ptr<FileSystem> base_fs_;
+  std::shared_ptr<io::LatencyGenerator> latencies_;
+};
+
 }  // namespace fs
 }  // namespace arrow

cpp/src/arrow/filesystem/filesystem_test.cc

@@ -612,6 +612,27 @@ TEST_F(TestSubTreeFileSystem, GetTargetStatsSelector) {
   ASSERT_EQ(stats.size(), 0);
 }
 
+////////////////////////////////////////////////////////////////////////////
+// Generic SlowFileSystem tests
+
+class TestSlowFSGeneric : public ::testing::Test, public GenericFileSystemTest {
+ public:
+  void SetUp() override {
+    time_ = TimePoint(TimePoint::duration(42));
+    fs_ = std::make_shared<MockFileSystem>(time_);
+    slow_fs_ = std::make_shared<SlowFileSystem>(fs_, 0.001);
+  }
+
+ protected:
+  std::shared_ptr<FileSystem> GetEmptyFileSystem() override { return slow_fs_; }
+
+  TimePoint time_;
+  std::shared_ptr<MockFileSystem> fs_;
+  std::shared_ptr<SlowFileSystem> slow_fs_;
+};
+
+GENERIC_FS_TEST_FUNCTIONS(TestSlowFSGeneric);
+
 }  // namespace internal
 }  // namespace fs
 }  // namespace arrow

cpp/src/arrow/io/memory_test.cc

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <chrono>
 #include <cstdint>
 #include <cstdlib>
 #include <cstring>
@@ -26,6 +27,7 @@
 #include "arrow/buffer.h"
 #include "arrow/io/interfaces.h"
 #include "arrow/io/memory.h"
+#include "arrow/io/slow.h"
 #include "arrow/status.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/testing/util.h"
@@ -313,5 +315,39 @@ TEST(TestMemcopy, ParallelMemcopy) {
   }
 }
 
+template <typename SlowStreamType>
+void TestSlowInputStream() {
+  using clock = std::chrono::high_resolution_clock;
+
+  auto stream = std::make_shared<BufferReader>(util::string_view("abcdefghijkl"));
+  const double latency = 0.6;
+  auto slow = std::make_shared<SlowStreamType>(stream, latency);
+
+  ASSERT_FALSE(slow->closed());
+  std::shared_ptr<Buffer> buf;
+  auto t1 = clock::now();
+  ASSERT_OK(slow->Read(6, &buf));
+  auto t2 = clock::now();
+  AssertBufferEqual(*buf, "abcdef");
+  auto dt = std::chrono::duration_cast<std::chrono::duration<double>>(t2 - t1).count();
+  ASSERT_LT(dt, latency * 3);  // likely
+  ASSERT_GT(dt, latency / 3);  // likely
+
+  util::string_view view;
+  ASSERT_OK(slow->Peek(4, &view));
+  ASSERT_EQ(view, util::string_view("ghij"));
+
+  ASSERT_OK(slow->Close());
+  ASSERT_TRUE(slow->closed());
+  ASSERT_TRUE(stream->closed());
+  ASSERT_OK(slow->Close());
+  ASSERT_TRUE(slow->closed());
+  ASSERT_TRUE(stream->closed());
+}
+
+TEST(TestSlowInputStream, Basics) { TestSlowInputStream<SlowInputStream>(); }
+
+TEST(TestSlowRandomAccessFile, Basics) { TestSlowInputStream<SlowRandomAccessFile>(); }
+
 }  // namespace io
 }  // namespace arrow