feat: add epoll-waitable queues

udf-runner-cpp/v2/BUILD.bazel

@@ -212,3 +212,68 @@ cc_test(
     target_compatible_with = ["@platforms//os:linux"],
     deps = [":json_schema"],
 )
+
+cc_library(
+    name = "moodycamel_queues",
+    hdrs = [
+        "include/exasol/udf/v2/mpmc_queue.hpp",
+        "include/exasol/udf/v2/spsc_queue.hpp",
+    ],
+    includes = ["include"],
+    deps = [
+        "@v2_concurrentqueue//:concurrentqueue",
+        "@v2_readerwriterqueue//:readerwriterqueue",
+    ],
+)
+
+cc_test(
+    name = "moodycamel_queues_test",
+    srcs = ["moodycamel_queues_test.cc"],
+    copts = ["-std=c++20"],
+    deps = [":moodycamel_queues"],
+)
+
+cc_binary(
+    name = "moodycamel_queues_shared",
+    srcs = ["moodycamel_queues_shared.cc"],
+    copts = ["-std=c++20"],
+    linkshared = 1,
+    deps = [":moodycamel_queues"],
+)
+
+cc_test(
+    name = "moodycamel_symbol_leak_test",
+    srcs = ["moodycamel_symbol_leak_test.cc"],
+    copts = ["-std=c++20"],
+    data = [":moodycamel_queues_shared"],
+    args = ["$(location :moodycamel_queues_shared)"],
+    target_compatible_with = ["@platforms//os:linux"],
+    deps = [":moodycamel_queues"],
+)
+
+cc_library(
+    name = "waitable_queue",
+    hdrs = ["include/exasol/udf/v2/waitable_queue.hpp"],
+    includes = ["include"],
+    deps = [":moodycamel_queues"],
+    target_compatible_with = ["@platforms//os:linux"],
+)
+
+cc_test(
+    name = "waitable_queue_test",
+    srcs = ["waitable_queue_test.cc"],
+    copts = ["-std=c++20"],
+    deps = [":waitable_queue"],
+    target_compatible_with = ["@platforms//os:linux"],
+)
+
+cc_binary(
+    name = "waitable_queue_benchmark",
+    srcs = ["waitable_queue_benchmark.cc"],
+    copts = ["-std=c++20"],
+    deps = [
+        ":waitable_queue",
+        "@google_benchmark//:benchmark_main",
+    ],
+    target_compatible_with = ["@platforms//os:linux"],
+)

udf-runner-cpp/v2/MODULE.bazel

@@ -6,6 +6,27 @@ module(
 bazel_dep(name = "rules_cc", version = "0.2.17")
 bazel_dep(name = "platforms", version = "1.0.0")
 bazel_dep(name = "flatbuffers", version = "25.2.10")
+bazel_dep(name = "google_benchmark", version = "1.9.5")
+
+# FlatBuffers currently selects versions of these transitive build tools that
+# still use the removed incompatible_use_toolchain_transition rule attribute.
+# Override them so the v2 module can be analyzed by Bazel 9.
+single_version_override(
+    module_name = "aspect_bazel_lib",
+    version = "2.19.4",
+)
+single_version_override(
+    module_name = "rules_foreign_cc",
+    version = "0.15.1",
+)
+single_version_override(
+    module_name = "rules_go",
+    version = "0.63.0",
+)
+single_version_override(
+    module_name = "aspect_rules_esbuild",
+    version = "0.24.0",
+)
 
 http_archive = use_repo_rule(
     "@bazel_tools//tools/build_defs/repo:http.bzl",
@@ -80,3 +101,39 @@ cc_library(
 )
 """,
 )
+
+http_archive(
+    name = "v2_readerwriterqueue",
+    urls = ["https://github.com/cameron314/readerwriterqueue/archive/refs/tags/v1.0.7.tar.gz"],
+    sha256 = "532224ed052bcd5f4c6be0ed9bb2b8c88dfe7e26e3eb4dd9335303b059df6691",
+    strip_prefix = "readerwriterqueue-1.0.7",
+    build_file_content = """
+load("@rules_cc//cc:cc_library.bzl", "cc_library")
+
+package(default_visibility = ["//visibility:public"])
+
+cc_library(
+    name = "readerwriterqueue",
+    hdrs = glob(["*.h"]),
+    includes = ["."],
+)
+""",
+)
+
+http_archive(
+    name = "v2_concurrentqueue",
+    urls = ["https://github.com/cameron314/concurrentqueue/archive/refs/tags/v1.0.5.tar.gz"],
+    sha256 = "4d6368a27492d86011fde5ca0cf386dce7c49cd425aa3d9b063ca6ec373a6ef3",
+    strip_prefix = "concurrentqueue-1.0.5",
+    build_file_content = """
+load("@rules_cc//cc:cc_library.bzl", "cc_library")
+
+package(default_visibility = ["//visibility:public"])
+
+cc_library(
+    name = "concurrentqueue",
+    hdrs = glob(["*.h"]),
+    includes = ["."],
+)
+""",
+)

udf-runner-cpp/v2/include/exasol/udf/v2/mpmc_queue.hpp

@@ -0,0 +1,19 @@
+#pragma once
+
+// Keep all upstream moodycamel symbols below the project-owned namespace. Do
+// not include these upstream headers directly in project or consumer code.
+#define moodycamel exasol::udf::v2::third_party::moodycamel
+#include <blockingconcurrentqueue.h>
+#include <concurrentqueue.h>
+#undef moodycamel
+
+namespace exasol::udf::v2 {
+
+template <typename T>
+using MpmcQueue = third_party::moodycamel::ConcurrentQueue<T>;
+
+template <typename T>
+using BlockingMpmcQueue =
+    third_party::moodycamel::BlockingConcurrentQueue<T>;
+
+}  // namespace exasol::udf::v2

udf-runner-cpp/v2/include/exasol/udf/v2/spsc_queue.hpp

@@ -0,0 +1,20 @@
+#pragma once
+
+// Moodycamel is header-only. Rewrite its namespace while parsing the upstream
+// headers so the resulting symbols cannot collide with an unrelated copy that
+// may be loaded into the same process.
+#define moodycamel exasol::udf::v2::third_party::moodycamel
+#include <readerwritercircularbuffer.h>
+#include <readerwriterqueue.h>
+#undef moodycamel
+
+namespace exasol::udf::v2 {
+
+template <typename T>
+using SpscQueue = third_party::moodycamel::ReaderWriterQueue<T>;
+
+template <typename T>
+using SpscCircularBuffer =
+    third_party::moodycamel::BlockingReaderWriterCircularBuffer<T>;
+
+}  // namespace exasol::udf::v2

udf-runner-cpp/v2/include/exasol/udf/v2/waitable_queue.hpp

@@ -0,0 +1,170 @@
+#pragma once
+
+#if !defined(__linux__)
+#error "exasol::udf::v2::WaitableQueue requires Linux eventfd"
+#endif
+
+#include <sys/eventfd.h>
+#include <unistd.h>
+
+#include <cerrno>
+#include <cstdint>
+#include <iterator>
+#include <system_error>
+#include <utility>
+
+#include <exasol/udf/v2/mpmc_queue.hpp>
+#include <exasol/udf/v2/spsc_queue.hpp>
+
+namespace exasol::udf::v2 {
+
+// Adds an epoll-compatible readiness descriptor to a queue. The descriptor
+// signals that one or more queue elements may be available; it is not a
+// one-to-one mapping between eventfd counter values and queue elements.
+template <typename Queue>
+class WaitableQueue {
+   public:
+    using queue_type = Queue;
+
+    WaitableQueue()
+        : notification_fd_(::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) {
+        if (notification_fd_ == -1) {
+            throw std::system_error(errno, std::generic_category(),
+                                    "eventfd");
+        }
+    }
+
+    explicit WaitableQueue(Queue queue)
+        : queue_(std::move(queue)),
+          notification_fd_(::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC)) {
+        if (notification_fd_ == -1) {
+            throw std::system_error(errno, std::generic_category(),
+                                    "eventfd");
+        }
+    }
+
+    ~WaitableQueue() {
+        if (notification_fd_ != -1) {
+            ::close(notification_fd_);
+        }
+    }
+
+    WaitableQueue(const WaitableQueue&) = delete;
+    WaitableQueue& operator=(const WaitableQueue&) = delete;
+
+    WaitableQueue(WaitableQueue&& other) noexcept
+        : queue_(std::move(other.queue_)),
+          notification_fd_(std::exchange(other.notification_fd_, -1)) {}
+
+    WaitableQueue& operator=(WaitableQueue&& other) noexcept {
+        if (this != &other) {
+            if (notification_fd_ != -1) {
+                ::close(notification_fd_);
+            }
+            queue_ = std::move(other.queue_);
+            notification_fd_ = std::exchange(other.notification_fd_, -1);
+        }
+        return *this;
+    }
+
+    [[nodiscard]] int native_handle() const noexcept {
+        return notification_fd_;
+    }
+
+    template <typename T>
+    [[nodiscard]] bool enqueue(T&& value) {
+        if (!queue_.enqueue(std::forward<T>(value))) {
+            return false;
+        }
+        notify();
+        return true;
+    }
+
+    template <typename InputIt>
+    std::size_t enqueue_batch(InputIt first, InputIt last) {
+        std::size_t enqueued = 0;
+        for (; first != last; ++first) {
+            if (!queue_.enqueue(*first)) {
+                break;
+            }
+            ++enqueued;
+        }
+        if (enqueued != 0) {
+            notify();
+        }
+        return enqueued;
+    }
+
+    template <typename Output>
+    [[nodiscard]] bool try_dequeue(Output& value) {
+        return queue_.try_dequeue(value);
+    }
+
+    // Drains all eventfd notifications and returns their accumulated count.
+    // Callers should then dequeue until the queue is empty and recheck it
+    // before going back to epoll_wait().
+    std::uint64_t drain_notifications() {
+        std::uint64_t total = 0;
+        for (;;) {
+            std::uint64_t value = 0;
+            const ssize_t result = ::read(notification_fd_, &value,
+                                          sizeof(value));
+            if (result == sizeof(value)) {
+                total += value;
+                continue;
+            }
+            if (result == -1 && errno == EINTR) {
+                continue;
+            }
+            if (result == -1 && errno == EAGAIN) {
+                return total;
+            }
+            if (result == -1) {
+                throw std::system_error(errno, std::generic_category(),
+                                        "read eventfd");
+            }
+            throw std::system_error(EIO, std::generic_category(),
+                                    "short read from eventfd");
+        }
+    }
+
+    Queue& queue() noexcept { return queue_; }
+    const Queue& queue() const noexcept { return queue_; }
+
+   private:
+    void notify() {
+        constexpr std::uint64_t signal = 1;
+        for (;;) {
+            const ssize_t result =
+                ::write(notification_fd_, &signal, sizeof(signal));
+            if (result == sizeof(signal)) {
+                return;
+            }
+            if (result == -1 && errno == EINTR) {
+                continue;
+            }
+            // A saturated eventfd is already readable. The queue item remains
+            // available, so no additional notification is needed.
+            if (result == -1 && errno == EAGAIN) {
+                return;
+            }
+            if (result == -1) {
+                throw std::system_error(errno, std::generic_category(),
+                                        "write eventfd");
+            }
+            throw std::system_error(EIO, std::generic_category(),
+                                    "short write to eventfd");
+        }
+    }
+
+    Queue queue_;
+    int notification_fd_;
+};
+
+template <typename T>
+using WaitableSpscQueue = WaitableQueue<SpscQueue<T>>;
+
+template <typename T>
+using WaitableMpmcQueue = WaitableQueue<MpmcQueue<T>>;
+
+}  // namespace exasol::udf::v2

udf-runner-cpp/v2/moodycamel_queues_shared.cc

@@ -0,0 +1,10 @@
+#include <exasol/udf/v2/mpmc_queue.hpp>
+#include <exasol/udf/v2/spsc_queue.hpp>
+
+extern "C" void exasol_udf_v2_moodycamel_queue_anchor() {
+    exasol::udf::v2::SpscQueue<int> spsc;
+    spsc.enqueue(1);
+
+    exasol::udf::v2::MpmcQueue<int> mpmc;
+    mpmc.enqueue(2);
+}

udf-runner-cpp/v2/moodycamel_queues_test.cc

@@ -0,0 +1,27 @@
+#include <cassert>
+
+#include <exasol/udf/v2/mpmc_queue.hpp>
+#include <exasol/udf/v2/spsc_queue.hpp>
+
+int main() {
+    exasol::udf::v2::SpscQueue<int> spsc;
+    assert(spsc.enqueue(7));
+    int value = 0;
+    assert(spsc.try_dequeue(value));
+    assert(value == 7);
+
+    exasol::udf::v2::SpscCircularBuffer<int> circular(2);
+    assert(circular.try_enqueue(8));
+    assert(circular.try_dequeue(value));
+    assert(value == 8);
+
+    exasol::udf::v2::MpmcQueue<int> mpmc;
+    assert(mpmc.enqueue(9));
+    assert(mpmc.try_dequeue(value));
+    assert(value == 9);
+
+    exasol::udf::v2::BlockingMpmcQueue<int> blocking;
+    assert(blocking.enqueue(10));
+    assert(blocking.try_dequeue(value));
+    assert(value == 10);
+}