ARROW-11680: [C++] Add vendored version of folly's spsc queue

See issue for rationale / description. To consider: * I based licensing text on other vendored libraries but I'm no expert in how exactly to dot all the i's and cross all the t's. * The constructor can throw if allocation of the ring buffer fails. In general, this sort of queue shouldn't be all that large anyways so this seems unlikely. We could modify the constructor to be private with a factory method that returns `Result` but at the moment I'm not making any changes to the existing code and so I'm hesitant to start doing so for this reason. * ~~There is a test case for the above point but I could take it out if desired.~~ Had to remove test since a massive allocation did not throw on mac. * ~~There is a comparative benchmark for boost. I'm not sure if we want to leave that in?~~ Had to remove benchmark since some build environments did not have boost::lockfree::spsc_queue * I've renamed folly::ProducerConsumerQueue to arrow::util::SpscQueue. I feel the name ProducerConsumerQueue is too vague and could lead to improper usage. This queue is not safe for multiple readers or multiple writers. Closes apache#9519 from westonpace/feature/folly Lead-authored-by: Weston Pace <[email protected]> Co-authored-by: Antoine Pitrou <[email protected]> Signed-off-by: Antoine Pitrou <[email protected]>
tpn · Mar 9, 2021 · fbb0662 · fbb0662
1 parent e7c47ba
commit fbb0662
Show file tree

Hide file tree

Showing 8 changed files with 448 additions and 44 deletions.
diff --git a/LICENSE.txt b/LICENSE.txt
@@ -2119,6 +2119,22 @@ DEALINGS IN THE SOFTWARE.
 
 --------------------------------------------------------------------------------
 
+This project includes code from Folly.
+
+ * cpp/src/arrow/vendored/ProducerConsumerQueue.h
+
+is based on Folly's
+
+ * folly/Portability.h
+ * folly/lang/Align.h
+ * folly/ProducerConsumerQueue.h
+
+Copyright: Copyright (c) Facebook, Inc. and its affiliates.
+Home page: https://github.com/facebook/folly
+License: http://www.apache.org/licenses/LICENSE-2.0
+
+--------------------------------------------------------------------------------
+
 The file cpp/src/arrow/vendored/musl/strptime.c has the following license
 
 Copyright © 2005-2020 Rich Felker, et al.

diff --git a/cpp/src/arrow/testing/gtest_util.h b/cpp/src/arrow/testing/gtest_util.h
@@ -521,6 +521,50 @@ void PrintTo(const Result<T>& result, std::ostream* os) {
   }
 }
 
+// A data type with only move constructors.
+struct MoveOnlyDataType {
+  explicit MoveOnlyDataType(int x) : data(new int(x)) {}
+
+  MoveOnlyDataType(const MoveOnlyDataType& other) = delete;
+  MoveOnlyDataType& operator=(const MoveOnlyDataType& other) = delete;
+
+  MoveOnlyDataType(MoveOnlyDataType&& other) { MoveFrom(&other); }
+  MoveOnlyDataType& operator=(MoveOnlyDataType&& other) {
+    MoveFrom(&other);
+    return *this;
+  }
+
+  ~MoveOnlyDataType() { Destroy(); }
+
+  void Destroy() {
+    if (data != nullptr) {
+      delete data;
+      data = nullptr;
+      moves = -1;
+    }
+  }
+
+  void MoveFrom(MoveOnlyDataType* other) {
+    Destroy();
+    data = other->data;
+    other->data = nullptr;
+    moves = other->moves + 1;
+  }
+
+  int ToInt() const { return data == nullptr ? -42 : *data; }
+
+  bool operator==(int other) const { return data != nullptr && *data == other; }
+  bool operator==(const MoveOnlyDataType& other) const {
+    return data != nullptr && other.data != nullptr && *data == *other.data;
+  }
+  friend bool operator==(int left, const MoveOnlyDataType& right) {
+    return right == left;
+  }
+
+  int* data = nullptr;
+  int moves = 0;
+};
+
 }  // namespace arrow
 
 namespace nonstd {

diff --git a/cpp/src/arrow/util/CMakeLists.txt b/cpp/src/arrow/util/CMakeLists.txt
@@ -54,6 +54,7 @@ add_arrow_test(utility-test
                ${IO_UTIL_TEST_SOURCES}
                iterator_test.cc
                logging_test.cc
+               queue_test.cc
                range_test.cc
                rle_encoding_test.cc
                stl_util_test.cc
@@ -80,6 +81,7 @@ add_arrow_benchmark(decimal_benchmark)
 add_arrow_benchmark(hashing_benchmark)
 add_arrow_benchmark(int_util_benchmark)
 add_arrow_benchmark(machine_benchmark)
+add_arrow_benchmark(queue_benchmark)
 add_arrow_benchmark(range_benchmark)
 add_arrow_benchmark(tdigest_benchmark)
 add_arrow_benchmark(thread_pool_benchmark)

diff --git a/cpp/src/arrow/util/future_test.cc b/cpp/src/arrow/util/future_test.cc
@@ -65,50 +65,6 @@ struct IterationTraits<Foo> {
   static Foo End() { return Foo(-1); }
 };
 
-// A data type with only move constructors.
-struct MoveOnlyDataType {
-  explicit MoveOnlyDataType(int x) : data(new int(x)) {}
-
-  MoveOnlyDataType(const MoveOnlyDataType& other) = delete;
-  MoveOnlyDataType& operator=(const MoveOnlyDataType& other) = delete;
-
-  MoveOnlyDataType(MoveOnlyDataType&& other) { MoveFrom(&other); }
-  MoveOnlyDataType& operator=(MoveOnlyDataType&& other) {
-    MoveFrom(&other);
-    return *this;
-  }
-
-  ~MoveOnlyDataType() { Destroy(); }
-
-  void Destroy() {
-    if (data != nullptr) {
-      delete data;
-      data = nullptr;
-      moves = -1;
-    }
-  }
-
-  void MoveFrom(MoveOnlyDataType* other) {
-    Destroy();
-    data = other->data;
-    other->data = nullptr;
-    moves = other->moves + 1;
-  }
-
-  int ToInt() const { return data == nullptr ? -42 : *data; }
-
-  bool operator==(int other) const { return data != nullptr && *data == other; }
-  bool operator==(const MoveOnlyDataType& other) const {
-    return data != nullptr && other.data != nullptr && *data == *other.data;
-  }
-  friend bool operator==(int left, const MoveOnlyDataType& right) {
-    return right == left;
-  }
-
-  int* data = nullptr;
-  int moves = 0;
-};
-
 template <>
 struct IterationTraits<MoveOnlyDataType> {
   static MoveOnlyDataType End() { return MoveOnlyDataType(-1); }

diff --git a/cpp/src/arrow/util/queue.h b/cpp/src/arrow/util/queue.h
@@ -0,0 +1,29 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include "arrow/vendored/ProducerConsumerQueue.h"
+
+namespace arrow {
+namespace util {
+
+template <typename T>
+using SpscQueue = arrow_vendored::folly::ProducerConsumerQueue<T>;
+
+}
+}  // namespace arrow
diff --git a/cpp/src/arrow/util/queue_benchmark.cc b/cpp/src/arrow/util/queue_benchmark.cc
@@ -0,0 +1,85 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <algorithm>
+#include <iterator>
+#include <thread>
+#include <vector>
+
+#include <benchmark/benchmark.h>
+
+#include "arrow/buffer.h"
+#include "arrow/util/logging.h"
+#include "arrow/util/queue.h"
+
+namespace arrow {
+
+namespace util {
+
+static constexpr int64_t kSize = 100000;
+
+void Throughput(benchmark::State& state) {
+  SpscQueue<std::shared_ptr<Buffer>> queue(16);
+
+  std::vector<std::shared_ptr<Buffer>> source;
+  std::vector<std::shared_ptr<Buffer>> sink;
+  source.reserve(kSize);
+  sink.resize(kSize);
+  const uint8_t data[1] = {0};
+  for (int64_t i = 0; i < kSize; i++) {
+    source.push_back(std::make_shared<Buffer>(data, 1));
+  }
+
+  for (auto _ : state) {
+    std::thread producer([&] {
+      auto itr = std::make_move_iterator(source.begin());
+      auto end = std::make_move_iterator(source.end());
+      while (itr != end) {
+        while (!queue.Write(*itr)) {
+        }
+        itr++;
+      }
+    });
+
+    std::thread consumer([&] {
+      auto itr = sink.begin();
+      auto end = sink.end();
+      while (itr != end) {
+        auto next = queue.FrontPtr();
+        if (next != nullptr) {
+          (*itr).swap(*next);
+          queue.PopFront();
+          itr++;
+        }
+      }
+    });
+
+    producer.join();
+    consumer.join();
+    std::swap(source, sink);
+  }
+
+  for (const auto& buf : source) {
+    ARROW_CHECK(buf && buf->size() == 1);
+  }
+  state.SetItemsProcessed(state.iterations() * kSize);
+}
+
+BENCHMARK(Throughput)->UseRealTime();
+
+}  // namespace util
+}  // namespace arrow
diff --git a/cpp/src/arrow/util/queue_test.cc b/cpp/src/arrow/util/queue_test.cc
@@ -0,0 +1,55 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include "arrow/testing/gtest_util.h"
+#include "arrow/util/queue.h"
+
+namespace arrow {
+namespace util {
+
+TEST(TestSpscQueue, TestMoveOnly) {
+  SpscQueue<MoveOnlyDataType> queue(3);
+  ASSERT_TRUE(queue.IsEmpty());
+  ASSERT_FALSE(queue.IsFull());
+  ASSERT_EQ(queue.SizeGuess(), 0);
+
+  MoveOnlyDataType in(42);
+  queue.Write(std::move(in));
+  ASSERT_FALSE(queue.IsEmpty());
+  ASSERT_FALSE(queue.IsFull());
+  ASSERT_EQ(queue.SizeGuess(), 1);
+
+  queue.Write(43);
+  ASSERT_FALSE(queue.IsEmpty());
+  ASSERT_TRUE(queue.IsFull());
+  ASSERT_EQ(queue.SizeGuess(), 2);
+
+  MoveOnlyDataType out = std::move(*queue.FrontPtr());
+  ASSERT_EQ(42, *out.data);
+  queue.PopFront();
+  ASSERT_TRUE(queue.Read(out));
+  ASSERT_EQ(43, *out.data);
+
+  ASSERT_TRUE(queue.IsEmpty());
+  ASSERT_FALSE(queue.IsFull());
+  ASSERT_EQ(queue.SizeGuess(), 0);
+}
+
+}  // namespace util
+}  // namespace arrow