ipc_server.hpp

Go to the documentation of this file.

#pragma once
 
#include <atomic>
#include <cstddef>
#include <cstdint>
#include <exception>
#include <functional>
#include <memory>
#include <span>
#include <string>
#include <sys/types.h>
#include <utility>
#include <vector>
 
namespace bb::ipc {
 
class ShutdownRequested : public std::exception {
    std::vector<uint8_t> response_;
 
  public:
    explicit ShutdownRequested(std::vector<uint8_t> response)
        : response_(std::move(response))
    {}
    const std::vector<uint8_t>& response() const { return response_; }
    const char* what() const noexcept override { return "Server shutdown requested"; }
};
 
class IpcServer {
  public:
    IpcServer() = default;
    virtual ~IpcServer() = default;
 
    // Abstract interface - no copy or move
    IpcServer(const IpcServer&) = delete;
    IpcServer& operator=(const IpcServer&) = delete;
    IpcServer(IpcServer&&) = delete;
    IpcServer& operator=(IpcServer&&) = delete;
 
    virtual bool listen() = 0;
 
    virtual int wait_for_data(uint64_t timeout_ns) = 0;
 
    virtual std::span<const uint8_t> receive(int client_id) = 0;
 
    virtual void release(int client_id, size_t message_size) = 0;
 
    virtual bool send(int client_id, const void* data, size_t len) = 0;
 
    virtual void close() = 0;
 
    virtual void request_shutdown()
    {
        shutdown_requested_.store(true, std::memory_order_release);
        wakeup_all();
    }
 
    using Handler = std::function<std::vector<uint8_t>(int client_id, std::span<const uint8_t> request)>;
 
    virtual int accept() { return -1; }
 
    virtual void run(const Handler& handler)
    {
        while (!shutdown_requested_.load(std::memory_order_acquire)) {
            // Try to accept new clients (non-blocking for socket servers)
            accept();
 
            int client_id = wait_for_data(100000000); // 100ms timeout
            if (client_id < 0) {
                // Timeout or error - check shutdown flag on next iteration
                continue;
            }
 
            // Receive message (blocks until complete message available, zero-copy for SHM)
            auto request = receive(client_id);
            if (request.empty()) {
                continue;
            }
 
            try {
                auto response = handler(client_id, request);
                if (!response.empty()) {
                    send(client_id, response.data(), response.size());
                }
 
                // Explicitly release/consume the message
                release(client_id, request.size());
            } catch (const ShutdownRequested& shutdown) {
                // Release message before shutting down
                release(client_id, request.size());
 
                // Send final response before shutting down
                if (!shutdown.response().empty()) {
                    send(client_id, shutdown.response().data(), shutdown.response().size());
                }
                // Graceful shutdown - exit loop and let destructors run
                return;
            }
        }
    }
 
    // Factory methods
    static std::unique_ptr<IpcServer> create_socket(const std::string& socket_path, int max_clients);
    static std::unique_ptr<IpcServer> create_shm(const std::string& base_name,
                                                 size_t request_ring_size = static_cast<size_t>(1024 * 1024),
                                                 size_t response_ring_size = static_cast<size_t>(1024 * 1024));
 
  protected:
    std::atomic<bool> shutdown_requested_{ false };
 
    virtual void wakeup_all() {};
};
 
} // namespace bb::ipc