Feature/concurrent requests

README.md

@@ -51,7 +51,6 @@ RabbitMQ (`rabbitmq/Readme.md`):
   "host": "localhost",
   "port": 5672,
   "queue": "myqueue",
-  "replyQueue": "myqueue.reply",
   "user": "guest",
   "password": "guest",
   "vhost": "/",

kafka/GraftcodePluginsInterfaces/ITransport.h

@@ -1,12 +1,12 @@
 #pragma once
 #include "_common.h"
+#include <vector>
 
 namespace Hypertube::Native::Interfaces {
 class ITransport {
 public:
     virtual ~ITransport() = default;
     virtual int Initialize(byte callingRuntimeNumber, byte calledRuntimeNumber, byte calledRuntimeVersion) = 0;
-    virtual int SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) = 0;
-    virtual int ReadResponse(byte* responseByteArray, int32_t responseByteArrayLen) = 0;
+    virtual std::vector<byte> SendCommand(byte* messageByteArray, int32_t messageByteArrayLen) = 0;
 };
 }

kafka/GraftcodePluginsInterfaces/PendingReplies.h

@@ -0,0 +1,190 @@
+/**
+ * @file PendingReplies.h
+ * @brief Correlates concurrent requests with replies delivered by a single consumer.
+ *
+ * A caller registers its correlation id before sending, then waits on the returned ticket.
+ * The consumer completes or fails entries by id. The ticket removes its entry when it goes
+ * out of scope, so timed-out and failed requests leave nothing behind.
+ */
+
+#pragma once
+
+#include <chrono>
+#include <condition_variable>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <stdexcept>
+#include <string>
+#include <utility>
+#include <vector>
+
+namespace Graftcode::Plugins::Concurrency
+{
+    class PendingReplies
+    {
+        struct Slot
+        {
+            std::condition_variable ready;
+            bool done{ false };
+            bool failed{ false };
+            std::vector<unsigned char> payload;
+            std::string error;
+        };
+
+    public:
+        using Payload = std::vector<unsigned char>;
+
+        class Ticket
+        {
+        public:
+            Ticket(PendingReplies& owner, std::string id, std::shared_ptr<Slot> slot)
+                : owner_(&owner), id_(std::move(id)), slot_(std::move(slot))
+            {
+            }
+
+            ~Ticket()
+            {
+                if (owner_ != nullptr) {
+                    owner_->Remove(id_);
+                }
+            }
+
+            Ticket(Ticket&& other) noexcept
+                : owner_(std::exchange(other.owner_, nullptr)),
+                  id_(std::move(other.id_)),
+                  slot_(std::move(other.slot_))
+            {
+            }
+
+            Ticket(const Ticket&) = delete;
+            Ticket& operator=(const Ticket&) = delete;
+            Ticket& operator=(Ticket&&) = delete;
+
+            const std::string& Id() const { return id_; }
+
+            // Returns false on timeout; throws when the request was failed.
+            bool Wait(std::chrono::milliseconds timeout, Payload& response)
+            {
+                std::unique_lock<std::mutex> lock(owner_->mutex_);
+                if (!slot_->ready.wait_for(lock, timeout, [this] { return slot_->done; })) {
+                    return false;
+                }
+                if (slot_->failed) {
+                    throw std::runtime_error(slot_->error);
+                }
+                response = std::move(slot_->payload);
+                return true;
+            }
+
+        private:
+            PendingReplies* owner_;
+            std::string id_;
+            std::shared_ptr<Slot> slot_;
+        };
+
+        PendingReplies() = default;
+        PendingReplies(const PendingReplies&) = delete;
+        PendingReplies& operator=(const PendingReplies&) = delete;
+
+        Ticket Register(const std::string& id)
+        {
+            auto slot = std::make_shared<Slot>();
+            {
+                std::lock_guard<std::mutex> lock(mutex_);
+                if (closed_) {
+                    throw std::runtime_error(closedError_);
+                }
+                if (!slots_.emplace(id, slot).second) {
+                    throw std::runtime_error("duplicate correlation id " + id);
+                }
+            }
+            pendingChanged_.notify_all();
+            return Ticket(*this, id, std::move(slot));
+        }
+
+        // Returns false when no request is waiting for this id.
+        bool Complete(const std::string& id, Payload payload)
+        {
+            return Finish(id, false, std::move(payload), std::string());
+        }
+
+        bool Fail(const std::string& id, const std::string& error)
+        {
+            return Finish(id, true, Payload(), error);
+        }
+
+        void FailAll(const std::string& error)
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            for (auto& entry : slots_) {
+                MarkDone(*entry.second, true, Payload(), error);
+            }
+        }
+
+        // Fails every waiting request and rejects new registrations.
+        void Close(const std::string& error)
+        {
+            {
+                std::lock_guard<std::mutex> lock(mutex_);
+                closed_ = true;
+                closedError_ = error;
+                for (auto& entry : slots_) {
+                    MarkDone(*entry.second, true, Payload(), error);
+                }
+            }
+            pendingChanged_.notify_all();
+        }
+
+        bool Empty() const
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            return slots_.empty();
+        }
+
+        // Blocks until a request is registered, the set is closed, or the timeout elapses.
+        // Returns true when at least one request is waiting.
+        bool WaitForPending(std::chrono::milliseconds timeout)
+        {
+            std::unique_lock<std::mutex> lock(mutex_);
+            pendingChanged_.wait_for(lock, timeout, [this] { return closed_ || !slots_.empty(); });
+            return !closed_ && !slots_.empty();
+        }
+
+    private:
+        static void MarkDone(Slot& slot, bool failed, Payload payload, const std::string& error)
+        {
+            if (slot.done) {
+                return;
+            }
+            slot.done = true;
+            slot.failed = failed;
+            slot.payload = std::move(payload);
+            slot.error = error;
+            slot.ready.notify_all();
+        }
+
+        bool Finish(const std::string& id, bool failed, Payload payload, const std::string& error)
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            const auto it = slots_.find(id);
+            if (it == slots_.end()) {
+                return false;
+            }
+            MarkDone(*it->second, failed, std::move(payload), error);
+            return true;
+        }
+
+        void Remove(const std::string& id)
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            slots_.erase(id);
+        }
+
+        mutable std::mutex mutex_;
+        std::condition_variable pendingChanged_;
+        std::map<std::string, std::shared_ptr<Slot>> slots_;
+        bool closed_{ false };
+        std::string closedError_;
+    };
+}

kafka/GraftcodePluginsInterfaces/TransportPluginAbi.h

@@ -0,0 +1,53 @@
+/**
+ * @file TransportPluginAbi.h
+ * @brief C ABI exported by transport plugin libraries.
+ *
+ * A plugin exports:
+ *  - CreateTransport
+ *  - DestroyTransport
+ *  - TransportSendCommand
+ *  - TransportFreeBuffer
+ *
+ * Plugins must be thread-safe: TransportSendCommand may be called concurrently on one channel.
+ *
+ * Every buffer returned by the plugin is allocated by the plugin and released with TransportFreeBuffer.
+ */
+
+#pragma once
+#include <stdint.h>
+#ifndef __cplusplus
+#include <stdbool.h>
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#define GRAFTCODE_TRANSPORT_OK 0
+#define GRAFTCODE_TRANSPORT_ERROR (-1)
+
+/** Opaque transport channel owned by the plugin. */
+typedef struct GraftcodeTransportChannel GraftcodeTransportChannel;
+
+/**
+ * Creates a transport channel.
+ * Returns null on failure; *errorMessage then receives a NUL-terminated message (or null).
+ */
+typedef GraftcodeTransportChannel* (*GraftcodeCreateTransportFn)(const char* ipAddress, unsigned short port, const char* configSource, char** errorMessage);
+
+/** Destroys a channel returned by CreateTransport. */
+typedef void (*GraftcodeDestroyTransportFn)(GraftcodeTransportChannel* channel);
+
+/**
+ * Sends a request and returns its response. May be called concurrently on one channel.
+ * On GRAFTCODE_TRANSPORT_OK, *response and *responseLen hold the response (*response may be null when the length is 0).
+ * Otherwise they hold a NUL-terminated error message and its length without the terminator.
+ */
+typedef int32_t (*GraftcodeTransportSendCommandFn)(GraftcodeTransportChannel* channel, const unsigned char* message, int32_t messageLen, unsigned char** response, int32_t* responseLen);
+
+/** Releases a buffer returned by the plugin. */
+typedef void (*GraftcodeTransportFreeBufferFn)(void* buffer);
+
+#ifdef __cplusplus
+}
+#endif

kafka/GraftcodePluginsInterfaces/TransportPluginExports.h

@@ -0,0 +1,112 @@
+/**
+ * @file TransportPluginExports.h
+ * @brief Implements the transport plugin C ABI (TransportPluginAbi.h) on top of ITransport.
+ */
+
+#pragma once
+#include "ITransport.h"
+#include "TransportPluginAbi.h"
+
+#include <cstdlib>
+#include <cstring>
+#include <exception>
+#include <new>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+namespace Graftcode::Plugins::TransportAbi
+{
+    using Hypertube::Native::Interfaces::ITransport;
+
+    // Allocates a NUL-terminated copy released with TransportFreeBuffer.
+    inline unsigned char* CopyBuffer(const void* data, std::size_t length)
+    {
+        auto* buffer = static_cast<unsigned char*>(std::malloc(length + 1));
+        if (buffer == nullptr) {
+            return nullptr;
+        }
+        if (length > 0) {
+            std::memcpy(buffer, data, length);
+        }
+        buffer[length] = 0;
+        return buffer;
+    }
+
+    inline char* CopyMessage(const std::string& message)
+    {
+        return reinterpret_cast<char*>(CopyBuffer(message.data(), message.size()));
+    }
+
+    template <typename Factory>
+    GraftcodeTransportChannel* CreateTransport(Factory&& factory, char** errorMessage)
+    {
+        if (errorMessage != nullptr) {
+            *errorMessage = nullptr;
+        }
+        std::string error;
+        try {
+            ITransport* transport = factory();
+            return reinterpret_cast<GraftcodeTransportChannel*>(transport);
+        }
+        catch (const std::exception& e) {
+            error = e.what();
+        }
+        catch (...) {
+            error = "unknown transport plugin error";
+        }
+        if (errorMessage != nullptr) {
+            *errorMessage = CopyMessage(error);
+        }
+        return nullptr;
+    }
+
+    inline void DestroyTransport(GraftcodeTransportChannel* channel)
+    {
+        delete reinterpret_cast<ITransport*>(channel);
+    }
+
+    inline int32_t SendCommand(
+        GraftcodeTransportChannel* channel,
+        const unsigned char* message,
+        int32_t messageLen,
+        unsigned char** response,
+        int32_t* responseLen)
+    {
+        if (response == nullptr || responseLen == nullptr) {
+            return GRAFTCODE_TRANSPORT_ERROR;
+        }
+        *response = nullptr;
+        *responseLen = 0;
+
+        std::string error;
+        try {
+            if (channel == nullptr) {
+                throw std::invalid_argument("transport channel is null");
+            }
+            const std::vector<byte> result = reinterpret_cast<ITransport*>(channel)->SendCommand(
+                const_cast<byte*>(message), messageLen);
+            unsigned char* buffer = CopyBuffer(result.data(), result.size());
+            if (buffer == nullptr) {
+                throw std::bad_alloc();
+            }
+            *response = buffer;
+            *responseLen = static_cast<int32_t>(result.size());
+            return GRAFTCODE_TRANSPORT_OK;
+        }
+        catch (const std::exception& e) {
+            error = e.what();
+        }
+        catch (...) {
+            error = "unknown transport plugin error";
+        }
+        *response = CopyBuffer(error.data(), error.size());
+        *responseLen = *response != nullptr ? static_cast<int32_t>(error.size()) : 0;
+        return GRAFTCODE_TRANSPORT_ERROR;
+    }
+
+    inline void FreeBuffer(void* buffer)
+    {
+        std::free(buffer);
+    }
+}