From 81ba942b8e224b164a38063e8386a47524fb0ecb Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Thu, 6 Aug 2026 23:14:13 +0800 Subject: [PATCH] feat(libmsg): extract pymsg-compatible messaging library as shared lib - Add libmsg: C++20 header-only pymsg protocol implementation - Msg.hpp: pmsg_send/pmsg_recv (blocking + deadline modes) + MsgClient - MsgServer.hpp: multi-client TCP server with select event loop + thread pool - AsyncQueue: generic async batch processing queue - RpcClient/RpcServer: JSON RPC convention layer (requires boost::json) - Msg_Basic.hpp/cpp: exception hierarchy, constants, version - Fix thread safety: MsgClient _sockfd/_running use std::atomic with exchange() - Fix deadline recv: client call() uses shared deadline for header+payload - Fix graceful shutdown: thread pool drains queued tasks before exit - Fix pmsg_send size check: enforce 64MiB limit (was 4GiB no-op check) - Align AsyncQueue batch semantics with Python: batch_size is threshold - Refactor gateway/LogC.hpp to use libmsg AsyncQueue - Update gateway Makefile to link -lus_msg, DbC.hpp to use - Add libmsg to _inst.sh build order --- _inst.sh | 1 + gateway/Makefile | 12 +- gateway/inc/DbC.hpp | 2 +- gateway/inc/LogC.hpp | 121 +++----- gateway/inc/Msg.hpp | 393 ------------------------- libmsg/.gitignore | 69 +++++ libmsg/LICENSE | 21 ++ libmsg/Makefile | 39 +++ libmsg/README.md | 545 +++++++++++++++++++++++++++++++++++ libmsg/_set.sh | 5 + libmsg/inc/us/AsyncQueue.hpp | 289 +++++++++++++++++++ libmsg/inc/us/Msg.hpp | 462 +++++++++++++++++++++++++++++ libmsg/inc/us/MsgServer.hpp | 528 +++++++++++++++++++++++++++++++++ libmsg/inc/us/Msg_Basic.hpp | 79 +++++ libmsg/inc/us/RpcClient.hpp | 181 ++++++++++++ libmsg/inc/us/RpcServer.hpp | 199 +++++++++++++ libmsg/src/Msg_Basic.cpp | 11 + 17 files changed, 2475 insertions(+), 482 deletions(-) delete mode 100644 gateway/inc/Msg.hpp create mode 100644 libmsg/.gitignore create mode 100644 libmsg/LICENSE create mode 100644 libmsg/Makefile create mode 100644 libmsg/README.md create mode 100755 libmsg/_set.sh create mode 100644 libmsg/inc/us/AsyncQueue.hpp create mode 100644 libmsg/inc/us/Msg.hpp create mode 100644 libmsg/inc/us/MsgServer.hpp create mode 100644 libmsg/inc/us/Msg_Basic.hpp create mode 100644 libmsg/inc/us/RpcClient.hpp create mode 100644 libmsg/inc/us/RpcServer.hpp create mode 100644 libmsg/src/Msg_Basic.cpp diff --git a/_inst.sh b/_inst.sh index 94cd456..7a0ac47 100755 --- a/_inst.sh +++ b/_inst.sh @@ -28,6 +28,7 @@ inst_module() { inst_module "libth" inst_module "libhttp" inst_module "libcrypt" +inst_module "libmsg" inst_module "libsqlite3" inst_module "gateway" inst_module "waf" # waf 在 gateway 之前编译，部署时由 core 先拉起 waf 再拉起 gateway diff --git a/gateway/Makefile b/gateway/Makefile index 7e9e18d..8b4415d 100644 --- a/gateway/Makefile +++ b/gateway/Makefile @@ -1,10 +1,10 @@ -VERSION := 0.0.1 +VERSION := 0.8.0 APP_NAME := gateway INSTL_DIR := /usr/local/bin CPP := g++ -INCS := -I inc/ -MY_LIBS := -L/usr/lib -lus_http -lus_httpclient -lus_crypt -LIBS := -pthread -lcurl -lssl -lsqlite3 -lz -lcrypto -lboost_json -lboost_program_options +INCS := -I inc/ -I ../libmsg/inc +MY_LIBS := -L/usr/lib -L../libmsg/build -lus_http -lus_httpclient -lus_crypt -lus_msg +LIBS := -pthread -lcurl -lssl -lsqlite3 -lz -lcrypto -lboost_json -lboost_program_options -Wl,-rpath,$(shell cd ../libmsg/build && pwd) CPPFLAGS := --std=c++20 -pipe -pthread # -Wall -Werror BUILD_DIR = build $(shell mkdir -p $(BUILD_DIR)) @@ -13,7 +13,7 @@ $(shell mkdir -p $(BUILD_DIR)) all: $(BUILD_DIR)/$(APP_NAME).out $(BUILD_DIR)/$(APP_NAME).out: $(BUILD_DIR)/main.o $(CPP) $^ $(LIBS) $(MY_LIBS) -o $@ -$(BUILD_DIR)/main.o: src/main.cpp - $(CPP) $(CPPFLAGS) -c $^ $(INCS) -o $@ +$(BUILD_DIR)/main.o: src/main.cpp inc/DbC.hpp inc/LogC.hpp inc/Conf.hpp inc/ExpBase.hpp inc/LocalTime.hpp + $(CPP) $(CPPFLAGS) -c $< $(INCS) -o $@ clear: @rm -rvf build/ diff --git a/gateway/inc/DbC.hpp b/gateway/inc/DbC.hpp index 796c468..1b37a01 100644 --- a/gateway/inc/DbC.hpp +++ b/gateway/inc/DbC.hpp @@ -1,6 +1,6 @@ #pragma once #include -#include "Msg.hpp" +#include #include namespace qing { namespace json = boost::json; diff --git a/gateway/inc/LogC.hpp b/gateway/inc/LogC.hpp index 66eaa88..aef0923 100644 --- a/gateway/inc/LogC.hpp +++ b/gateway/inc/LogC.hpp @@ -4,9 +4,10 @@ * * 设计要点（与 Python 版对齐）： * - 通过 pymsg TCP 调用 dbv2 execute 路由，逐条 INSERT 到 Logs 表 - * - 异步非阻塞：内存队列 + 后台线程批量推送，不拖慢业务 + * - 异步非阻塞：基于 pmc::net::AsyncQueue 内存队列 + 后台线程批量推送，不拖慢业务 * - fail-safe：db 服务不可用时静默丢弃（仅 stderr 提示），绝不影响业务 * - 缓冲批量：队列满 50 条或超时 2 秒时自动 flush + * - 背压保护：队列上限 10000 条，满时丢最旧日志，防止内存无限增长 * * 用法： * #include "LogC.hpp" @@ -21,20 +22,17 @@ #include #include -#include #include #include -#include -#include #include #include #include -#include #include #include -#include "Msg.hpp" +#include +#include namespace qing { @@ -62,12 +60,8 @@ public: const std::string& message, std::optional trace_id = std::nullopt, std::optional extra = std::nullopt) { - { - std::lock_guard lock(_mtx); - _queue.push_back(Entry{service, level, message, - std::move(trace_id), std::move(extra)}); - } - _cv.notify_one(); + _queue.push(Entry{service, level, message, + std::move(trace_id), std::move(extra)}); } void debug(const std::string& service, const std::string& message, @@ -96,88 +90,53 @@ public: push(service, "CRITICAL", message, std::move(trace_id), std::move(extra)); } - // 手动 flush：通知后台线程立即发送当前积攒的批次，并短暂等待其完成 + // 手动 flush：通知后台线程立即发送当前积攒的批次，并等待其完成 void flush() { - _force_flush.store(true); - _cv.notify_all(); - for (int i = 0; i < 200 && _force_flush.load(); ++i) { - std::this_thread::sleep_for(std::chrono::milliseconds(10)); - } + _queue.flush(); } // 停止后台线程并 flush 剩余日志（进程退出前调用，幂等） void shutdown() { - { - std::lock_guard lock(_mtx); - if (_stop) return; - _stop = true; - } - _cv.notify_all(); - if (_thread.joinable()) _thread.join(); + _queue.shutdown(); } // ---- 配置（构造后、首次推送前可改）------------------------------------ void set_endpoint(const std::string& addr, int port) { + std::lock_guard lock(_endpoint_mtx); _db_addr = addr; _db_port = port; } private: - LogC() : _thread(&LogC::_flush_loop, this) {} + LogC() + : _queue( + [this](std::vector&& batch) { this->_send_batch(std::move(batch)); }, + {_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5}, + [](const std::exception& e, std::vector&&) { + std::cerr << "[LogC] 发送失败: " << e.what() << std::endl; + }) + { + _queue.start(); + } + ~LogC() { shutdown(); } LogC(const LogC&) = delete; LogC& operator=(const LogC&) = delete; - // 后台线程：积攒到 batch_size 或超时 flush_interval 时发送 - void _flush_loop() { - std::vector batch; - auto last_flush = std::chrono::steady_clock::now(); - - for (;;) { - { - std::unique_lock lock(_mtx); - // 最多 0.5s 轮询一次（对齐 Python 版 chk_timeout 上限） - _cv.wait_for(lock, std::chrono::milliseconds(500), - [&] { return _stop || !_queue.empty(); }); - while (!_queue.empty() && batch.size() < _BATCH_SIZE) { - batch.push_back(std::move(_queue.front())); - _queue.pop_front(); - } - } - - const auto now = std::chrono::steady_clock::now(); - const bool due = std::chrono::duration(now - last_flush).count() - >= _FLUSH_INTERVAL; - if (!batch.empty() && (batch.size() >= _BATCH_SIZE || due - || _stop || _force_flush.load())) { - _send(batch); - batch.clear(); - last_flush = now; - _force_flush.store(false); - } - - if (_stop) { - // 退出前 flush 队列剩余 - { - std::unique_lock lock(_mtx); - while (!_queue.empty()) { - batch.push_back(std::move(_queue.front())); - _queue.pop_front(); - } - } - if (!batch.empty()) _send(batch); - _force_flush.store(false); - return; - } - } - } - // 通过 pymsg 使用 execute 路由逐条 INSERT 到 dbv2 Logs 表 - void _send(const std::vector& batch) { + void _send_batch(const std::vector& batch) { + std::string addr; + int port; + { + std::lock_guard lock(_endpoint_mtx); + addr = _db_addr; + port = _db_port; + } + const std::string now = _utc_now_iso(); // 批次共享同一时间戳（对齐 Python 版） try { - pmc::net::MsgClient client(_db_addr, _db_port, "gateway-log"); + pmc::net::MsgClient client(addr, port, "gateway-log"); for (const auto& e : batch) { json::object values; values["ts"] = now; @@ -233,17 +192,15 @@ private: // ---- 配置项 ----------------------------------------------------------- std::string _db_addr = "127.0.0.1"; int _db_port = 8005; - static constexpr size_t _BATCH_SIZE = 50; // 队列满 50 条时 flush - static constexpr double _FLUSH_INTERVAL = 2.0; // 超时 2 秒强制 flush - static constexpr double _TIMEOUT = 5.0; // socket 超时（秒） + std::mutex _endpoint_mtx; // 保护 _db_addr/_db_port 读写（set_endpoint 与后台线程并发） - // ---- 状态 ------------------------------------------------------------- - std::deque _queue; - std::mutex _mtx; - std::condition_variable _cv; - bool _stop = false; - std::atomic _force_flush{false}; - std::thread _thread; + static constexpr size_t _BATCH_SIZE = 50; // 队列满 50 条时 flush + static constexpr double _FLUSH_INTERVAL = 2.0; // 超时 2 秒强制 flush + static constexpr size_t _MAX_QUEUE_SIZE = 10000; // 队列上限，满时丢最旧 + static constexpr double _TIMEOUT = 5.0; // socket 超时（秒） + + // ---- 核心队列 --------------------------------------------------------- + pmc::net::AsyncQueue _queue; }; } // namespace qing diff --git a/gateway/inc/Msg.hpp b/gateway/inc/Msg.hpp deleted file mode 100644 index 90c1151..0000000 --- a/gateway/inc/Msg.hpp +++ /dev/null @@ -1,393 +0,0 @@ -#pragma once - -/** pymsg C++ 实现 — Length-prefixed variable-length socket message framing. - * - * Protocol: [4-byte big-endian uint32_t length][UTF-8 payload] - * - * 对应 Python 版 pymsg/pymsg/__init__.py，提供： - * - pmsg_send() / pmsg_recv() 底层协议收发函数 - * - MsgClient 短连接 send / call / 长连接 listen 客户端 - * - * Usage: - * #include "Msg.hpp" - * - * pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); - * std::string resp = client.call(R"({"sql":"SELECT 1"})", 5.0); - * client.send("hello"); - */ - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include -#include -#include -#include -#include -#include - -// macOS 没有 MSG_NOSIGNAL，用 0 替代（内核不会发 SIGPIPE） -#ifndef MSG_NOSIGNAL -#define MSG_NOSIGNAL 0 -#endif - -namespace pmc { -namespace net { - -// ============================================================================ -// Exceptions -// ============================================================================ - -class MsgException : public std::runtime_error { -public: - explicit MsgException(const std::string &msg) : std::runtime_error(msg) {} -}; - -class MsgTimeoutException : public MsgException { -public: - explicit MsgTimeoutException(const std::string &msg) : MsgException(msg) {} -}; - -class MsgConnectionException : public MsgException { -public: - explicit MsgConnectionException(const std::string &msg) : MsgException(msg) {} -}; - -/// 目标服务未启动（connect 返回 ECONNREFUSED），用于 retry_on_refused 重试机制 -class MsgConnectionRefusedException : public MsgConnectionException { -public: - explicit MsgConnectionRefusedException(const std::string &msg) - : MsgConnectionException(msg) {} -}; - -// ============================================================================ -// Low-level protocol: 4-byte big-endian length prefix + UTF-8 payload -// ============================================================================ - -namespace detail { - -constexpr size_t kHeaderSize = 4; - -// ---- 4-byte big-endian 编码 / 解码 ---------------------------------------- - -inline void writeUint32BE(char *buf, uint32_t val) noexcept { - buf[0] = static_cast((val >> 24) & 0xFFu); - buf[1] = static_cast((val >> 16) & 0xFFu); - buf[2] = static_cast((val >> 8) & 0xFFu); - buf[3] = static_cast(val & 0xFFu); -} - -inline uint32_t readUint32BE(const char *buf) noexcept { - return (static_cast(static_cast(buf[0])) << 24) | - (static_cast(static_cast(buf[1])) << 16) | - (static_cast(static_cast(buf[2])) << 8) | - (static_cast(static_cast(buf[3]))); -} - -// ---- socket 超时设置 ------------------------------------------------------ - -inline void setSockTimeout(int sockfd, double seconds) { - struct timeval tv {}; - tv.tv_sec = static_cast(seconds); - tv.tv_usec = - static_cast((seconds - static_cast(tv.tv_sec)) * 1'000'000.0); - ::setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); - ::setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); -} - -// ---- 精确读取 n 字节 ------------------------------------------------------ - -inline void recvExact(int sockfd, char *buf, size_t n) { - size_t received = 0; - while (received < n) { - ssize_t r = ::recv(sockfd, buf + received, n - received, 0); - if (r == 0) { - throw MsgConnectionException( - "connection closed: expected " + std::to_string(n) + - " bytes, got " + std::to_string(received)); - } - if (r < 0) { - if (errno == EAGAIN || errno == EWOULDBLOCK) { - throw MsgTimeoutException("recv timed out"); - } - throw MsgConnectionException( - "recv error: " + std::string(::strerror(errno))); - } - received += static_cast(r); - } -} - -// ---- 完整发送 ------------------------------------------------------------- - -inline void sendAll(int sockfd, const char *data, size_t len) { - size_t sent = 0; - while (sent < len) { - ssize_t w = - ::send(sockfd, data + sent, len - sent, MSG_NOSIGNAL); - if (w < 0) { - if (errno == EAGAIN || errno == EWOULDBLOCK) { - throw MsgTimeoutException("send timed out"); - } - throw MsgConnectionException( - "send error: " + std::string(::strerror(errno))); - } - sent += static_cast(w); - } -} - -// ---- 短连接 helper：connect → action → close ------------------------------ - -inline int openConnection(const std::string& addr, int port, double timeout) { - int fd = ::socket(AF_INET, SOCK_STREAM, 0); - if (fd < 0) { - throw MsgConnectionException( - "socket() failed: " + std::string(::strerror(errno))); - } - - struct sockaddr_in serv_addr {}; - serv_addr.sin_family = AF_INET; - serv_addr.sin_port = htons(static_cast(port)); - if (::inet_pton(AF_INET, addr.c_str(), &serv_addr.sin_addr) <= 0) { - ::close(fd); - throw MsgConnectionException("invalid address: " + addr); - } - - setSockTimeout(fd, timeout); - - if (::connect(fd, reinterpret_cast(&serv_addr), - sizeof(serv_addr)) < 0) { - int saved_errno = errno; - ::close(fd); - std::string err_msg = "could not connect to " + addr + ":" + - std::to_string(port) + " - " + ::strerror(saved_errno); - if (saved_errno == ECONNREFUSED) { - throw MsgConnectionRefusedException(err_msg); - } - throw MsgConnectionException(err_msg); - } - - return fd; -} - -} // namespace detail - -// ============================================================================ -// 底层协议函数：pmsg_send / pmsg_recv -// ============================================================================ - -/// 以 pymsg 协议发送一条字符串消息（长度前缀 + UTF-8 载荷）。 -inline void pmsg_send(int sockfd, const std::string &data) { - size_t payload_size = data.size(); - if (payload_size > 0xFFFFFFFFu) { - throw MsgException("payload too large: " + std::to_string(payload_size)); - } - char header[detail::kHeaderSize]; - detail::writeUint32BE(header, static_cast(payload_size)); - detail::sendAll(sockfd, header, detail::kHeaderSize); - detail::sendAll(sockfd, data.c_str(), payload_size); -} - -/// 以 pymsg 协议接收一条字符串消息。 -/// @throws MsgConnectionException 连接在收全消息前断开。 -/// @throws MsgTimeoutException 等待超时。 -inline std::string pmsg_recv(int sockfd) { - char header[detail::kHeaderSize]; - detail::recvExact(sockfd, header, detail::kHeaderSize); - uint32_t length = detail::readUint32BE(header); - if (length > 64 * 1024 * 1024) { // 安全上限 64 MiB - throw MsgException("message too large: " + std::to_string(length)); - } - std::string payload(static_cast(length), '\0'); - detail::recvExact(sockfd, payload.data(), length); - return payload; -} - -// ============================================================================ -// MsgClient -// ============================================================================ - -/// 基于 pymsg 协议的 TCP 客户端。 -/// -/// 支持三种操作模式： -/// - send() 短连接发送一条消息（不等待回复），消息带 name 前缀。 -/// - call() 短连接发送 + 阻塞等待回复（RPC 模式），原始消息不添加前缀。 -/// - listen() 基于 connect() 维持的长连接，持续接收消息并回调。 -class MsgClient { -public: - MsgClient(const std::string &addr, int port, const std::string &name) - : _addr(addr), _port(port), _name(name), _sockfd(-1), _running(false) {} - - ~MsgClient() { stop(); } - - // ---- 连接 --------------------------------------------------------------- - - /// 创建 TCP 套接字并连接到服务器（长连接，供 listen() 使用）。 - void connect() { - _sockfd = detail::openConnection(_addr, _port, /*timeout=*/10.0); - } - - // ---- 发送单条消息（短连接，带名字前缀）---------------------------------- - - /// 短连接发送一条消息到服务器，消息格式为 ``"name: msg"``。 - /// 不等待回复。短连接（每次创建新 socket，发完即关）。 - void send(const std::string &msg) { - int fd = detail::openConnection(_addr, _port, /*timeout=*/3.0); - try { - pmsg_send(fd, _name + ": " + msg); - ::close(fd); - } catch (...) { - ::close(fd); - throw; - } - } - - // ---- 请求-应答（短连接，RPC 模式）--------------------------------------- - - /// 短连接发送一条消息，阻塞等待并返回一条回复。 - /// - /// 不会给消息添加 ``name:`` 前缀（适用于 RPC/API 调用）。 - /// - /// @param msg 要发送的原始字符串。 - /// @param timeout 超时秒数，默认 10.0。 - /// @param retry_on_refused 目标服务未启动（ECONNREFUSED）时是否阻塞重试， - /// 直到连接成功。适用于启动顺序不确定的服务发现场景。 - /// 默认 false（连接拒绝立即抛异常）。 - /// @param retry_interval 重试间隔秒数，默认 1.0。 - /// @returns 服务器返回的字符串。 - std::string call(const std::string &msg, double timeout = 10.0, - bool retry_on_refused = false, - double retry_interval = 1.0) { - if (!retry_on_refused) { - // 快速路径：不重试，直接执行一次，保持原有行为 - int fd = detail::openConnection(_addr, _port, timeout); - try { - pmsg_send(fd, msg); - std::string resp = pmsg_recv(fd); - ::close(fd); - return resp; - } catch (...) { - ::close(fd); - throw; - } - } - - // 重试路径：仅对 ECONNREFUSED 重试，其他错误立即抛出 - constexpr double kLogEvery = 5.0; // 重试期间打印等待提示的间隔（秒） - int attempt = 0; - auto last_log = std::chrono::steady_clock::now() - - std::chrono::duration_cast( - std::chrono::duration(kLogEvery)); - - while (true) { - try { - int fd = detail::openConnection(_addr, _port, timeout); - try { - pmsg_send(fd, msg); - std::string resp = pmsg_recv(fd); - ::close(fd); - return resp; - } catch (...) { - ::close(fd); - throw; - } - } catch (const MsgConnectionRefusedException &) { - ++attempt; - auto now = std::chrono::steady_clock::now(); - double elapsed = std::chrono::duration(now - last_log).count(); - if (elapsed >= kLogEvery || attempt == 1) { - std::fprintf(stderr, - "[pymsg] 等待服务 %s:%d 就绪... (已重试 %d 次)\n", - _addr.c_str(), _port, attempt); - std::fflush(stderr); - last_log = now; - } - std::this_thread::sleep_for( - std::chrono::duration(retry_interval)); - } - // 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出 - } - } - - // ---- 持续监听（长连接，阻塞回调模式）----------------------------------- - - /// 在长连接上持续接收消息并通过回调处理。 - /// - /// 阻塞当前线程直到 stop() 被调用或连接断开。 - /// 必须先调用 connect()。 - /// - /// @param on_message 回调 ``void(const std::string &msg)``。 - void listen(std::function on_message) { - if (_sockfd < 0) { - throw MsgException("must call connect() before listen()"); - } - - _running = true; - while (_running) { - fd_set readfds; - FD_ZERO(&readfds); - FD_SET(_sockfd, &readfds); - - struct timeval tv {}; - tv.tv_sec = 0; - tv.tv_usec = 100'000; // 100 ms 轮询间隔 - - int ready = - ::select(_sockfd + 1, &readfds, nullptr, nullptr, &tv); - if (ready < 0) { - if (errno == EINTR) continue; - break; // EBADF or other fatal error - } - if (ready == 0) continue; // timeout, loop around - - try { - std::string text = pmsg_recv(_sockfd); - on_message(std::move(text)); - } catch (const MsgConnectionException &) { - break; // 连接断开，退出循环 - } catch (const MsgTimeoutException &) { - continue; // 读超时但连接仍在，继续等待 - } - } - - if (_sockfd >= 0) { - ::close(_sockfd); - _sockfd = -1; - } - } - - // ---- 停止 --------------------------------------------------------------- - - /// 设置停止标志并关闭套接字。可从任意线程调用。 - void stop() noexcept { - _running = false; - if (_sockfd >= 0) { - ::close(_sockfd); - _sockfd = -1; - } - } - - // ---- 访问器 ------------------------------------------------------------- - - const std::string &addr() const noexcept { return _addr; } - int port() const noexcept { return _port; } - const std::string &name() const noexcept { return _name; } - -private: - std::string _addr; - int _port; - std::string _name; - int _sockfd; - bool _running; -}; - -} // namespace net -} // namespace pmc diff --git a/libmsg/.gitignore b/libmsg/.gitignore new file mode 100644 index 0000000..f152158 --- /dev/null +++ b/libmsg/.gitignore @@ -0,0 +1,69 @@ +# Prerequisites +*.d + +# Compiled Object files +*.slo +*.lo +*.o +*.obj + +# Precompiled Headers +*.gch +*.pch + +# Linker files +*.ilk + +# Debugger Files +*.pdb + +# Compiled Dynamic libraries +*.so +*.dylib +*.dll +*.so.* + + +# Fortran module files +*.mod +*.smod + +# Compiled Static libraries +*.lai +*.la +*.a +*.lib + +# Executables +*.exe +*.out +*.app + +# Build directories +build/ +Build/ +build-*/ + +# CMake generated files +CMakeFiles/ +CMakeCache.txt +cmake_install.cmake +#Makefile +install_manifest.txt +compile_commands.json + +# Temporary files +*.tmp +*.log +*.bak +*.swp + +# vcpkg +vcpkg_installed/ + +# debug information files +*.dwo + +# test output & cache +Testing/ +.cache/ diff --git a/libmsg/LICENSE b/libmsg/LICENSE new file mode 100644 index 0000000..d63a519 --- /dev/null +++ b/libmsg/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 建昌风物大顺脚本 + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/libmsg/Makefile b/libmsg/Makefile new file mode 100644 index 0000000..f6edbef --- /dev/null +++ b/libmsg/Makefile @@ -0,0 +1,39 @@ +VERSION := 0.0.1 +FRAMEWORK_NAME := us +LIB_NAME := libus_msg +INSTL_TDIR := /lib +INCLU_TDIR := /usr/include +INCLU_SDIR := inc +CPP := g++ +INCS := -I inc +CPPFLAGS := --std=c++20 -O3 -pipe -Wall -Werror +LIBS := +BUILD_TDIR := build +$(shell mkdir -p $(BUILD_TDIR)) + +#-------- +all: $(BUILD_TDIR)/$(LIB_NAME).so + +$(BUILD_TDIR)/$(LIB_NAME).so: $(BUILD_TDIR)/Msg_Basic.o + $(CPP) -shared $^ $(LIBS) -o $@ + +$(BUILD_TDIR)/Msg_Basic.o: src/Msg_Basic.cpp + $(CPP) $(CPPFLAGS) -fPIC -c $^ $(INCS) -o $@ + +clear: + @rm -rvf $(BUILD_TDIR) + +install: + @echo "install UniServe_libmsg..." + @mkdir -vp $(INCLU_TDIR)/$(FRAMEWORK_NAME) + @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/Msg_Basic.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ + @cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/Msg.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/ + @cp -vf $(BUILD_TDIR)/$(LIB_NAME).so $(INSTL_TDIR)/ + @echo "done." + +uninstall: + @echo "uninstall UniServe_libmsg..." + @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/Msg_Basic.hpp + @rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/Msg.hpp + @rm -vf $(INSTL_TDIR)/$(LIB_NAME).so + @echo "done." diff --git a/libmsg/README.md b/libmsg/README.md new file mode 100644 index 0000000..12fa14a --- /dev/null +++ b/libmsg/README.md @@ -0,0 +1,545 @@ +# libmsg — UniServe 消息通信库 + +C++20 轻量级 TCP 消息帧通信库，实现 pymsg 协议（4字节大端长度前缀 + UTF-8载荷），与 Python 版 pymsg 字节级兼容。提供底层帧收发函数、MsgClient 客户端、MsgServer 多客户端服务端，以及通用异步批量队列 AsyncQueue。 + +命名空间 `pmc::net`，库文件 `libus_msg.so`，头文件安装至 `/usr/include/us/`。被 `gateway`、`db`、`dnfs`、`smtp`、`captcha`、`confs` 等所有使用 pymsg 协议的 C++ 服务依赖。 + +## 架构 + +``` +pmc::net::Msg_Basic (基类: 版本信息 / 异常层次 / MSG_CHECK 宏) +└── MsgClient TCP 客户端（短连接 send/call + 长连接 listen） + +pmc::net::MsgServer TCP 服务端（select 事件循环 + 可选线程池 + reply/broadcast） +pmc::net::AsyncQueue 通用异步批量发送队列（对应 Python AsyncRpcQueue） + +pmc::net::RpcClient JSON RPC 客户端（_route 注入 + success/error 自动解包）—— 需 boost::json +pmc::net::RpcServer JSON RPC 服务端（路由分发 + 响应包装 + 异常脱敏）—— 需 boost::json + +pmc::net 命名空间下的自由函数： +├── pmsg_send(fd, data) 底层协议发送 +├── pmsg_recv(fd) 底层协议接收（阻塞模式，依赖 SO_RCVTIMEO） +└── pmsg_recv(fd, timeout_sec) 底层协议接收（deadline 模式，总超时精确控制） + +pmc::net::msg_constants 协议常量（kHeaderSize=4, kMaxMsgSize=64MiB 等） + +异常层次： +├── MsgException 基础异常 +│ ├── MsgTimeoutException 收发超时 +│ ├── MsgConnectionException 连接异常 +│ │ └── MsgConnectionRefusedException ECONNREFUSED（用于重试） +│ └── RpcError JSON RPC 业务错误（success=false 或响应非法） +``` + +所有收发函数均为 header-only（inline 实现），`AsyncQueue` 和 `MsgServer` 也是 header-only，无第三方依赖（仅 POSIX socket + 标准库线程）。`version()` 编译进 `libus_msg.so`。 + +`RpcClient` 和 `RpcServer` 是可选的 JSON RPC 约定层，基于 `boost::json` 提供路由分发和响应自动包装，对应 Python 版 `pymsg.rpc`。使用时需额外链接 `-lboost_json`，核心库不引入此依赖。 + +## 线协议 + +``` ++--------+---------------------------+ +| 4 B BE | N B UTF-8 | +| length | payload | ++--------+---------------------------+ +``` + +- 长度字段：32 位无符号整数，大端序（network byte order） +- 载荷：UTF-8 编码文本（通常为 JSON） +- 最大消息长度：64 MiB（防 OOM 攻击） +- 传输层：TCP，无 TLS（信任本机网络边界，默认绑定 127.0.0.1） + +## 类与 API + +### Msg_Basic — 基类 + +提供版本查询和异常类型，`MsgClient` 继承此类。 + +| 成员 | 说明 | +|------|------| +| `version()` (static protected) | 返回版本字符串（编译进 .so） | +| `MsgExp` | 异常类，继承 `std::runtime_error` | +| `MSG_CHECK(cond, msg)` | 宏，条件为 false 时抛出 `std::runtime_error` | + +### 协议常量（`msg_constants` 命名空间） + +| 常量 | 值 | 说明 | +|------|----|------| +| `kHeaderSize` | 4 | 帧头长度（字节） | +| `kMaxMsgSize` | 64 MiB | 消息最大长度（防 OOM） | +| `kDefaultTimeout` | 10 秒 | call()/connect() 默认超时 | +| `kSendTimeout` | 3 秒 | send() 默认超时 | +| `kListenPollMs` | 100000 微秒 | listen() select 轮询间隔 | +| `kRetryLogEvery` | 5.0 秒 | 重试等待日志打印间隔 | + +### 底层协议函数 + +```cpp +// 发送一条消息（长度前缀 + payload） +void pmsg_send(int sockfd, const std::string& data); + +// 接收一条消息（阻塞模式，依赖 socket 已设置的 SO_RCVTIMEO） +// 抛出 MsgConnectionException（连接断开）或 MsgTimeoutException（超时） +std::string pmsg_recv(int sockfd); + +// 接收一条消息（deadline 模式，总超时不超过 timeout_sec） +// header+payload 共享同一个 deadline，使用 select() 实现精确超时控制 +// 适用于客户端 call() 和服务端，防止总等待时间翻倍 +std::string pmsg_recv(int sockfd, double timeout_sec); +``` + +### MsgClient — TCP 客户端 + +```cpp +#include + +// 构造：指定服务地址、端口、客户端名称（用于 send() 前缀） +pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); +``` + +| 方法 | 说明 | +|------|------| +| `connect()` | 创建长连接（供 listen() 使用） | +| `send(msg)` | 短连接发送 ``"name: msg"`` 格式消息，不等待回复 | +| `call(msg, timeout=10, retry_on_refused=false, retry_interval=1.0)` | 短连接 RPC：发送原始消息，阻塞等待回复 | +| `listen(on_message)` | 长连接持续接收，回调签名 `void(const std::string&)` | +| `stop()` | 设置停止标志并关闭套接字，可跨线程调用 | +| `addr()` / `port()` / `name()` | 访问器 | + +**call() 的 retry_on_refused 机制**： + +设为 `true` 时，仅对 `ECONNREFUSED`（目标服务未启动）以 1 秒间隔阻塞重试，每 5 秒打印一次等待日志。适用于微服务启动顺序不确定的场景：服务先启动的一方会自动等待依赖方就绪，无需额外服务发现组件。其他异常（超时、连接重置等）仍立即抛出。 + +### MsgServer — TCP 服务端 + +```cpp +#include + +// 构造：绑定地址和端口，可配置线程池大小 +pmc::net::MsgServer server("0.0.0.0", 8003, {.workers=4}); +``` + +对应 Python 版 `pymsg.MsgServer`，提供基于 select 的多客户端 TCP 服务端。 + +**配置结构**： + +```cpp +struct MsgServerConfig { + int workers = 0; // 线程池大小；0=同步模式（回调在主线程执行） + double recv_timeout = 10.0; // 接收单条消息的总超时（秒），防 Slowloris + int max_connections = 1000; // 最大并发连接数，超过则拒绝新连接 + int max_queue_size = 1000; // 线程池任务队列上限（workers>0 时），满时断开连接 + int listen_backlog = 10; // listen() backlog + double poll_timeout = 0.1; // select 轮询超时（秒），影响关闭响应速度 +}; +``` + +| 方法 | 说明 | +|------|------| +| `start(on_message)` | 启动服务（阻塞当前线程）；回调签名 `void(int fd, const string& addr, const string& msg)` | +| `stop()` | 停止服务（可从任意线程调用，包括回调内部） | +| `reply(fd, msg)` | 向指定客户端回复消息（线程安全，经每连接锁串行化） | +| `broadcast(msg, except_fd=-1)` | 向所有客户端广播（线程安全），可排除指定 fd | +| `client_count()` | 当前已连接客户端数量 | + +**设计特性**： + +- **两种并发模式**：`workers=0` 同步模式（简单高效），`workers>0` 线程池并发模式 +- **同连接串行保证**：同一连接的消息严格串行处理（回调执行期间暂停 select 监听该 fd） +- **线程安全发送**：`reply()`/`broadcast()` 经每连接 mutex 串行化，防止帧交错 +- **连接数限制**：`max_connections` 防止连接耗尽攻击 +- **有界任务队列**：`max_queue_size` 防止慢 handler 导致 OOM，队列满时断开连接 +- **Slowloris 防护**：使用 deadline 模式 recv（select + 剩余时间），总接收时间不超过 `recv_timeout` +- **优雅关闭**：`stop()` 可从任意线程调用，事件循环退出前清理所有连接 +- **故障隔离**：worker 线程异常仅标记 failed fd，主循环统一清理，避免 fd 并发操作 +- **TCP_NODELAY**：新连接自动禁用 Nagle 算法，降低小包延迟 + +### AsyncQueue\ — 通用异步批量发送队列 + +```cpp +#include +``` + +对应 Python 版 `pymsg.AsyncRpcQueue`，提供内存队列 + 后台线程的异步批量处理机制，适用于日志上报、指标采集、事件通知等非关键路径场景。业务线程 `push()` 非阻塞立即返回，后台线程攒批处理（批量阈值或超时），失败时通过错误回调通知，绝不阻塞主业务流程。 + +**配置结构**： + +```cpp +struct AsyncQueueConfig { + size_t batch_size = 100; // 批量发送阈值，达到此数量立即 flush + double flush_interval = 2.0; // 定时 flush 间隔（秒） + size_t max_queue_size = 10000; // 队列最大长度，0=无上限；超过时丢最旧数据 + double poll_timeout = 0.1; // 后台线程轮询超时（秒） +}; +``` + +| 方法 | 说明 | +|------|------| +| `AsyncQueue(handler, config={}, on_error={})` | 构造队列；`handler` 签名 `void(vector&& batch)`，`on_error` 签名 `void(const exception&, vector&&)` | +| `start()` | 启动后台发送线程（幂等） | +| `push(item)` | 非阻塞入队；返回 false 表示队列已关闭 | +| `flush()` | 在调用线程同步 flush 当前队列中的所有数据，阻塞直到处理完成 | +| `shutdown()` | 优雅关闭：flush 剩余数据后停止线程（幂等，析构自动调用） | +| `pending()` | 当前队列中待处理的数据量 | +| `dropped()` | 因队列满而丢弃的数据总数 | +| `is_running()` | 是否已 start 且未 shutdown | + +**设计特性**： + +- **非阻塞入队**：`push()` 立即返回，不等待网络 I/O +- **批量发送**：攒到 `batch_size` 条或超过 `flush_interval` 秒自动 flush +- **背压保护**：`max_queue_size` 限制内存队列长度，满时丢最旧数据并递增 `dropped()` 计数 +- **优雅关闭**：`shutdown()` flush 所有剩余数据后停止后台线程 +- **RAII**：析构自动调用 `shutdown()` +- **可插拔**：Handler 回调可自定义（RPC 调用、写文件、发消息等） +- **fail-safe**：Handler 抛异常时触发 `on_error` 回调，绝不传播到业务线程 + +### RpcClient — JSON RPC 客户端 + +```cpp +#include // 需链接 -lboost_json + +// 构造：指定服务地址、端口、客户端名称 +pmc::net::RpcClient rpc("127.0.0.1", 8005, "gateway"); + +// 启动等待模式（db 未启动时阻塞重试） +pmc::net::RpcClient rpc("127.0.0.1", 8005, "gateway", 10.0, true); +``` + +对应 Python 版 `pymsg.rpc.RpcClient`，在 `MsgClient` 之上封装 JSON RPC 约定： +- 自动注入 `_route` 字段 +- 自动解析响应 `{"success":true,"data":...}` / `{"success":false,"error":"..."}` +- 业务错误抛 `RpcError`，网络错误仍抛原有网络异常 + +| 方法 | 说明 | +|------|------| +| `RpcClient(addr, port, name, timeout=10, retry_on_refused=false, retry_interval=1.0)` | 构造 RPC 客户端 | +| `call(route, params={}, timeout=nullopt, retry=nullopt, retry_interval=nullopt)` | 调用路由，返回 `data` 字段值；业务错误抛 `RpcError` | +| `call_raw(route, params={}, timeout=nullopt, retry=nullopt, retry_interval=nullopt)` | 调用路由，返回完整响应 `object`（含 success/data/error），不抛 `RpcError` | +| `addr()` / `port()` / `name()` | 访问器 | + +**RPC 约定**： +- 请求格式：`{"_route": "<路由名>", ...业务字段}` +- 成功响应：`{"success": true, "data": ...}` +- 错误响应：`{"success": false, "error": "<错误信息>"}` +- 非 JSON 响应或缺 success 字段抛 `RpcError("响应不是合法 JSON: ...")` + +### RpcServer — JSON RPC 服务端 + +```cpp +#include // 需链接 -lboost_json + +// 构造：绑定地址和端口，可透传 MsgServerConfig +pmc::net::RpcServer rpc("127.0.0.1", 8005, {.workers=4}); + +// 注册路由 +rpc.add_route("ping", [](const json::object&) -> json::value { + return "pong"; +}); + +rpc.start(); // 阻塞当前线程 +``` + +对应 Python 版 `pymsg.rpc.RpcServer`，在 `MsgServer` 之上封装路由分发和响应自动包装。 + +| 方法 | 说明 | +|------|------| +| `RpcServer(addr, port, config={}, on_error={})` | 构造 RPC 服务端；`on_error` 签名 `void(const string& route, const exception& e)` | +| `add_route(name, handler)` | 注册路由；`handler` 签名 `json::value(const json::object& req)` | +| `start()` | 启动服务（阻塞当前线程） | +| `stop()` | 从另一线程停止服务 | +| `server()` | 获取底层 `MsgServer&` 引用（可用于 broadcast 等） | + +**Handler 约定**： +- 返回值自动包装为 `{"success": true, "data": <返回值>}` +- 抛出 `std::exception` 自动包装为 `{"success": false, "error": "<脱敏后错误>"}` +- 返回 `nullptr` 表示无数据（客户端收到 `data=null`） +- 业务参数中 `_route` 字段已被移除，不会透传给 handler + +**安全特性**： +- JSON 解析限制嵌套深度 32 层，防止栈溢出攻击 +- 错误信息自动脱敏：绝对路径替换为 ``，不泄露内部文件路径 +- `on_error` 钩子可记录完整异常到服务端日志（脱敏仅对客户端响应生效） +- Handler 异常不导致服务崩溃 + +### 异常层次 + +| 异常类 | 含义 | 来源 | +|--------|------|------| +| `MsgException` | 所有消息库异常的基类 | — | +| `MsgTimeoutException` | send/recv 超时 | SO_RCVTIMEO/SO_SNDTIMEO 或 deadline 到期触发 | +| `MsgConnectionException` | 连接断开/重置/地址错误 | recv 返回0/错误、connect 失败 | +| `MsgConnectionRefusedException` | connect 返回 ECONNREFUSED | 仅用于 retry_on_refused 判定 | +| `RpcError` | JSON RPC 业务错误 | 远端返回 success=false、响应非 JSON、缺少 success 字段 | + +## 构建与安装 + +**前置依赖**：g++（支持 C++20）、POSIX 系统（Linux/macOS），无第三方库依赖。 + +```bash +# 一键构建 + 安装（需 sudo） +bash _set.sh + +# 或手动 +make # 编译 libus_msg.so +make install # 安装头文件到 /usr/include/us/ + .so 到 /lib/ +make clear # 清理构建产物 +make uninstall # 卸载 +``` + +使用： + +```cpp +#include +#include +#include // 如需服务端 +#include // 如需异步队列 +#include // 如需 JSON RPC 客户端（需 boost::json） +#include // 如需 JSON RPC 服务端（需 boost::json） +``` + +链接： + +```bash +# 仅使用核心库（Msg/MsgServer/AsyncQueue） +g++ -std=c++20 ... -lus_msg + +# 使用 JSON RPC 层（RpcClient/RpcServer） +g++ -std=c++20 ... -lus_msg -lboost_json +``` + +> 与 `libus_crypt` 不同，`libus_msg.so` 不依赖 OpenSSL，无需额外链接 `-lssl -lcrypto`。所有通信函数、MsgServer 和 AsyncQueue 均为 header-only 内联实现，链接 `-lus_msg` 仅引入版本符号。 + +## 使用示例 + +### RPC 请求-应答（短连接） + +```cpp +#include +#include + +int main() { + pmc::net::MsgClient client("127.0.0.1", 8005, "demo"); + + // 简单调用（不重试） + try { + std::string resp = client.call(R"({"_route":"execute","sql":"SELECT 1"})", 5.0); + std::cout << "Response: " << resp << std::endl; + } catch (const pmc::net::MsgConnectionRefusedException& e) { + std::cerr << "服务未启动: " << e.what() << std::endl; + } + + // 启动等待模式（阻塞直到 db 就绪） + std::string resp = client.call( + R"({"_route":"execute","sql":"SELECT 1"})", + 5.0, + /*retry_on_refused=*/true + ); + std::cout << "Response after wait: " << resp << std::endl; + + return 0; +} +``` + +### 发送即忘（短连接通知） + +```cpp +pmc::net::MsgClient log_client("127.0.0.1", 8005, "log"); +log_client.send("application started"); // 发送 "log: application started" +``` + +### 长连接监听 + +```cpp +#include +#include + +int main() { + pmc::net::MsgClient listener("127.0.0.1", 9000, "subscriber"); + listener.connect(); + + listener.listen([](const std::string& msg) { + std::cout << "Received: " << msg << std::endl; + }); + + return 0; +} +``` + +### Echo 服务端（同步模式） + +```cpp +#include +#include +#include + +int main() { + pmc::net::MsgServer server("127.0.0.1", 8003); + std::thread t([&] { + server.start([&](int fd, const std::string& addr, const std::string& msg) { + server.reply(fd, "echo: " + msg); + }); + }); + + // ... 服务运行 ... + // 另一线程调用 server.stop() 可优雅关闭 + + server.stop(); + t.join(); + return 0; +} +``` + +### 广播聊天服务端（线程池模式） + +```cpp +#include +#include + +int main() { + pmc::net::MsgServer server("0.0.0.0", 8003, { + .workers = 4, + .recv_timeout = 10.0, + .max_connections = 1000, + .max_queue_size = 1000 + }); + + server.start([&](int fd, const std::string& addr, const std::string& msg) { + // 转发给所有人（排除发送者自己） + server.broadcast("[" + addr + "] " + msg, /*except_fd=*/fd); + }); + return 0; +} +``` + +### 异步批量队列（日志上报） + +```cpp +#include +#include +#include +#include +#include + +struct LogEntry { + std::string service; + std::string message; +}; + +int main() { + pmc::net::AsyncQueue queue( + [](std::vector&& batch) { + pmc::net::MsgClient client("127.0.0.1", 8005, "metrics"); + for (auto& e : batch) { + // 发送到日志服务... + } + }, + {50, 2.0, 10000, 0.5}, + [](const std::exception& e, std::vector&&) { + std::cerr << "[log] send failed: " << e.what() << std::endl; + } + ); + queue.start(); + + queue.push(LogEntry{"gateway", "request handled"}); + queue.push(LogEntry{"gateway", "db connected"}); + + // ... 应用运行 ... + + queue.shutdown(); + return 0; +} +``` + +### JSON RPC 服务端 + 客户端 + +```cpp +#include +#include +#include +#include + +int main() { + // 服务端 + pmc::net::RpcServer rpc("127.0.0.1", 8005, {.workers=4}, + [](const std::string& route, const std::exception& e) { + std::cerr << "[rpc][" << route << "] error: " << e.what() << std::endl; + }); + + rpc.add_route("ping", [](const json::object&) -> json::value { + return "pong"; + }); + + rpc.add_route("add", [](const json::object& req) -> json::value { + return req.at("a").to_number() + req.at("b").to_number(); + }); + + std::thread t([&]{ rpc.start(); }); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + + // 客户端 + pmc::net::RpcClient client("127.0.0.1", 8005, "demo"); + + json::object params; + params["a"] = 3; + params["b"] = 4; + json::value result = client.call("add", params); + std::cout << "3 + 4 = " << result.as_int64() << std::endl; // 7 + + try { + client.call("nonexistent"); + } catch (const pmc::net::RpcError& e) { + std::cout << "error: " << e.what() << std::endl; // unknown route: nonexistent + } + + rpc.stop(); + t.join(); + return 0; +} +``` + +## 项目结构 + +``` +libmsg/ +├── inc/us/ +│ ├── Msg_Basic.hpp # 基类：异常层次、常量、版本声明 +│ ├── Msg.hpp # 核心协议：pmsg_send/pmsg_recv + MsgClient（header-only） +│ ├── MsgServer.hpp # TCP 服务端：多客户端 select 事件循环 + 线程池（header-only） +│ ├── AsyncQueue.hpp # 通用异步批量队列（header-only 模板） +│ ├── RpcClient.hpp # JSON RPC 客户端（header-only，需 boost::json） +│ └── RpcServer.hpp # JSON RPC 服务端（header-only，需 boost::json） +├── src/ +│ └── Msg_Basic.cpp # version() 实现 +├── Makefile +├── _set.sh +└── README.md +``` + +## 注意事项 + +- **无外部依赖**：libmsg 核心（Msg/MsgServer/AsyncQueue）仅使用 POSIX socket API（socket/connect/recv/send/select）和 C++ 标准库（thread/mutex/condition_variable/deque/functional/unordered_map），不依赖 OpenSSL、Boost 或任何第三方库。 +- **JSON RPC 层依赖**：`RpcClient`/`RpcServer` 是基于 `boost::json` 的可选约定层，使用时需安装 Boost 并链接 `-lboost_json`。核心 `libus_msg.so` 不依赖 Boost。 +- **线程安全**： + - `MsgClient` 的 `stop()` 方法使用 `std::atomic` 保护标志位和 fd，可跨线程安全调用；`connect()` 会安全关闭旧连接。同一 `MsgClient` 实例的 `call()`/`send()`/`listen()` 不应并发调用（每个线程应创建独立实例）。`listen()` 中使用局部 fd 副本防止 stop() 关闭后 FD_SET(-1) 导致未定义行为。 + - `MsgServer` 的 `start()` 阻塞调用线程；`stop()`/`reply()`/`broadcast()`/`client_count()` 可在任意线程（包括消息回调内部）安全调用。消息回调在同步模式下运行于主线程，在线程池模式下运行于 worker 线程。reply()/broadcast() 经每连接 mutex 串行化防止帧交错。 + - `AsyncQueue` 的 `push()`/`flush()`/`shutdown()`/`pending()`/`dropped()` 可在任意线程安全调用；Handler 回调在后台线程中执行。`push()` 使用 mutex + condition_variable 唤醒后台线程，无锁竞争热点。 + - `RpcClient` 每次 `call()` 创建独立短连接，实例可在多线程中安全使用（但建议每个线程独立实例以避免连接复用问题）。 + - `RpcServer` 的 `add_route()` 应在 `start()` 之前调用；`stop()`/`server()` 可跨线程安全调用。Handler 在 MsgServer 回调线程中执行（同步模式=主线程，线程池模式=worker线程）。 +- **macOS 兼容**：自动处理 `MSG_NOSIGNAL` 缺失问题（macOS 内核不会在 send 到已关闭连接时发 SIGPIPE）。 +- **超时行为**： + - `MsgClient::call()` 使用 deadline 模式 recv（select + 剩余时间递减），header+payload 共享同一个总超时，不会出现"header 等 t 秒 + payload 再等 t 秒"的双倍超时问题。 + - `MsgClient::listen()` 使用 SO_RCVTIMEO 做 100ms 轮询超时，结合 `_running` 原子标志实现快速停止响应。 + - `MsgServer` 使用 deadline 模式（select + 剩余时间递减），单条消息总接收时间不超过 `recv_timeout`，有效防御 Slowloris 攻击。 + - `AsyncQueue::flush()` 在调用线程同步执行 `_send_batch()`，阻塞直到 handler 处理完成；`shutdown()` 永久等待后台线程结束（handler 中 socket 均有超时，不会永久阻塞）。 +- **消息大小限制**：收发两端均检查 64 MiB 上限（`kMaxMsgSize`），防止超大消息导致 OOM。 +- **retry_on_refused 仅对 ECONNREFUSED 重试**：连接超时、连接重置、路由不可达等错误不重试，立即抛出异常。 +- **MsgServer 同步 vs 线程池**： + - `workers=0`（默认）：回调在主线程执行，适合 CPU 轻量、无阻塞 I/O 的场景（如简单 echo、内存状态查询）。同步模式下回调异常会被捕获、记日志并断开该客户端，不会导致服务崩溃。 + - `workers>0`：回调提交到线程池，适合有 I/O 阻塞的场景；同一连接消息仍严格串行，不同连接可并行处理。`stop()` 会等待所有已排队任务处理完成（优雅关闭），不丢弃请求。 +- **AsyncQueue 背压策略**：队列满时丢最旧数据（drop-oldest），而非阻塞 push 或丢最新数据，保证最新数据不丢失同时防止内存无限增长。`batch_size` 是触发阈值而非批量上限，flush 时取走队列中所有数据一次处理。 +- **跨语言兼容**：C++ 实现与 Python 版 `pymsg`、Go 版（waf 内）字节级完全兼容，可无缝互操作。 + +## 许可证 + +MIT License diff --git a/libmsg/_set.sh b/libmsg/_set.sh new file mode 100755 index 0000000..312f67c --- /dev/null +++ b/libmsg/_set.sh @@ -0,0 +1,5 @@ +#!/bin/bash +cd "$(dirname "$0")" || exit +make clear +make +make install diff --git a/libmsg/inc/us/AsyncQueue.hpp b/libmsg/inc/us/AsyncQueue.hpp new file mode 100644 index 0000000..04fe109 --- /dev/null +++ b/libmsg/inc/us/AsyncQueue.hpp @@ -0,0 +1,289 @@ +#pragma once + +/** AsyncQueue — 通用异步批量发送队列（C++ 版，对应 Python 版 pymsg/pymsg/async_queue.py） + * + * 提供内存队列 + 后台线程的异步批量处理机制，适用于日志上报、指标采集、事件通知等 + * 非关键路径场景。业务线程 push() 非阻塞立即返回，后台线程攒批处理（批量阈值或超时）， + * 失败时通过错误回调通知，绝不阻塞或影响主业务流程。 + * + * 设计要点（与 Python AsyncRpcQueue 对齐）： + * - 非阻塞入队：push() 立即返回，不等待网络 I/O + * - 批量发送：攒到 batch_size 条或超过 flush_interval 秒自动 flush + * - 背压保护：max_queue_size 限制内存队列长度，满时丢最旧数据并计数 + * - 优雅关闭：shutdown() flush 剩余数据后停止后台线程 + * - RAII：析构自动 shutdown() + * - 可插拔：通过 Handler 回调自定义批量处理逻辑（RPC 调用、写文件、发消息等） + * + * Usage: + * #include + * + * struct LogEntry { std::string service; std::string msg; }; + * + * pmc::net::AsyncQueue queue( + * [](std::vector&& batch) { + * for (auto& e : batch) { send_to_server(e); } + * }, + * {50, 2.0, 10000, 0.5}, + * [](const std::exception& e, std::vector&&) { + * std::cerr << "[queue] send failed: " << e.what() << std::endl; + * } + * ); + * queue.start(); + * queue.push(LogEntry{"gateway", "started"}); + * queue.shutdown(); + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace pmc { +namespace net { + +// ============================================================================ +// AsyncQueue 配置 +// ============================================================================ + +struct AsyncQueueConfig { + size_t batch_size = 100; ///< 批量发送阈值，达到此数量立即 flush + double flush_interval = 2.0; ///< 定时 flush 间隔（秒） + size_t max_queue_size = 10000; ///< 队列最大长度，0 表示无上限；超过时丢最旧数据 + double poll_timeout = 0.1; ///< 后台线程轮询超时（秒），越小关闭响应越快但CPU占用略高 +}; + +// ============================================================================ +// AsyncQueue +// ============================================================================ + +/** 通用异步批量发送队列。 + * + * @tparam T 队列元素类型，需支持移动构造（push 时会被 std::move 到 batch 中）。 + */ +template +class AsyncQueue { +public: + /// 批量处理回调：接收一个元素批次的右值引用，处理完后 batch 被销毁。 + using Handler = std::function&& batch)>; + + /// 错误回调：接收异常和失败的批次（可用于重试或记录）。 + using ErrorHandler = std::function&& batch)>; + + // ---- 构造 / 析构 ------------------------------------------------------- + + explicit AsyncQueue(Handler handler, + AsyncQueueConfig config = {}, + ErrorHandler on_error = {}) + : _handler(std::move(handler)), + _config(config), + _on_error(std::move(on_error)), + _running(false), + _stop(false), + _wakeup(false), + _dropped(0) {} + + /// 析构函数：自动 shutdown()，保证队列中剩余数据被 flush。 + ~AsyncQueue() { shutdown(); } + + AsyncQueue(const AsyncQueue&) = delete; + AsyncQueue& operator=(const AsyncQueue&) = delete; + + // ---- 生命周期 ---------------------------------------------------------- + + /// 启动后台发送线程。幂等：已启动时无操作。 + void start() { + std::lock_guard lock(_mtx); + if (_running) return; + _stop = false; + _wakeup = false; + _running = true; + _thread = std::thread(&AsyncQueue::_run, this); + } + + /// 优雅关闭：flush 剩余数据后停止后台线程。幂等。 + void shutdown() { + { + std::lock_guard lock(_mtx); + if (!_running) return; + _stop = true; + _wakeup = true; + } + _cv.notify_all(); + if (_thread.joinable()) { + _thread.join(); + } + // 线程结束后，flush 队列中剩余的所有数据 + _flush_all(); + { + std::lock_guard lock(_mtx); + _running = false; + } + } + + // ---- 入队 -------------------------------------------------------------- + + /** 非阻塞入队一条数据。 + * + * 若队列已关闭（shutdown 后），则丢弃数据并返回 false。 + * 若队列长度达到 max_queue_size，则丢弃最旧的数据并递增 dropped 计数。 + * + * @return true 入队成功；false 队列已关闭。 + */ + bool push(T item) { + { + std::lock_guard lock(_mtx); + if (_stop) return false; + if (_config.max_queue_size > 0 && _queue.size() >= _config.max_queue_size) { + _queue.pop_front(); + ++_dropped; + } + _queue.push_back(std::move(item)); + _wakeup = true; + } + _cv.notify_one(); + return true; + } + + // ---- 手动 flush -------------------------------------------------------- + + /// 立即触发一次 flush，在调用线程同步处理当前队列中的所有数据。 + /// 对齐 Python：flush() 直接在调用线程执行 _send_batch()，阻塞直到处理完成。 + void flush() { + _send_batch(); + } + + // ---- 状态查询 ---------------------------------------------------------- + + /// 当前队列中待处理的数据量。 + size_t pending() const { + std::lock_guard lock(_mtx); + return _queue.size(); + } + + /// 因队列满而丢弃的数据总数。 + size_t dropped() const { + return _dropped.load(); + } + + /// 队列是否正在运行（已 start 且未 shutdown）。 + bool is_running() const { + std::lock_guard lock(_mtx); + return _running; + } + +private: + // ---- 后台线程主循环（对齐 Python AsyncRpcQueue._run）-------------------- + + void _run() { + auto last_flush = std::chrono::steady_clock::now(); + + while (true) { + bool should_stop = false; + { + std::unique_lock lock(_mtx); + _cv.wait_for(lock, + std::chrono::duration(_config.poll_timeout), + [this] { return _stop || _wakeup; }); + _wakeup = false; // 消费唤醒标志 + should_stop = _stop; + } + + const auto now = std::chrono::steady_clock::now(); + const bool due = std::chrono::duration(now - last_flush).count() + >= _config.flush_interval; + + size_t qsize = 0; + { + std::lock_guard lock(_mtx); + qsize = _queue.size(); + } + + // flush 条件：达到批量阈值 / 超时且有数据 / 停止时还有数据 + if (qsize >= _config.batch_size || (due && qsize > 0) || (should_stop && qsize > 0)) { + _send_batch(); + last_flush = std::chrono::steady_clock::now(); + } + + if (should_stop) break; + } + } + + // ---- 取所有待处理数据并发送（对齐 Python：batch_size 是触发阈值，不是上限）- + + void _send_batch() { + std::vector batch; + { + std::lock_guard lock(_mtx); + if (_queue.empty()) return; + batch.reserve(_queue.size()); + while (!_queue.empty()) { + batch.push_back(std::move(_queue.front())); + _queue.pop_front(); + } + } + + try { + _handler(std::move(batch)); + } catch (const std::exception& e) { + if (_on_error) { + try { + _on_error(e, std::move(batch)); + } catch (...) { + // 错误回调本身不应传播异常 + } + } + } + } + + // ---- 关闭时 flush 所有剩余数据 ---------------------------------------- + + void _flush_all() { + while (true) { + std::vector batch; + { + std::lock_guard lock(_mtx); + if (_queue.empty()) break; + batch.reserve(_queue.size()); + while (!_queue.empty()) { + batch.push_back(std::move(_queue.front())); + _queue.pop_front(); + } + } + try { + _handler(std::move(batch)); + } catch (const std::exception& e) { + if (_on_error) { + try { + _on_error(e, std::move(batch)); + } catch (...) {} + } + } + } + } + + // ---- 成员 -------------------------------------------------------------- + + Handler _handler; + AsyncQueueConfig _config; + ErrorHandler _on_error; + + mutable std::mutex _mtx; + std::condition_variable _cv; + std::deque _queue; + std::thread _thread; + bool _running; + bool _stop; + bool _wakeup; ///< 唤醒标志（push/shutdown 置 true，后台线程消费后置 false） + + std::atomic _dropped; +}; + +} // namespace net +} // namespace pmc diff --git a/libmsg/inc/us/Msg.hpp b/libmsg/inc/us/Msg.hpp new file mode 100644 index 0000000..8568f65 --- /dev/null +++ b/libmsg/inc/us/Msg.hpp @@ -0,0 +1,462 @@ +#pragma once + +/** pymsg C++ 实现 — Length-prefixed variable-length socket message framing. + * + * Protocol: [4-byte big-endian uint32_t length][UTF-8 payload] + * + * 对应 Python 版 pymsg/pymsg/__init__.py，提供： + * - pmsg_send() / pmsg_recv() 底层协议收发函数 + * - MsgClient 短连接 send / call / 长连接 listen 客户端 + * + * Usage: + * #include + * + * pmc::net::MsgClient client("127.0.0.1", 8005, "gateway"); + * std::string resp = client.call(R"({"sql":"SELECT 1"})", 5.0); + * client.send("hello"); + */ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +// macOS 没有 MSG_NOSIGNAL，用 0 替代（内核不会发 SIGPIPE） +#ifndef MSG_NOSIGNAL +#define MSG_NOSIGNAL 0 +#endif + +namespace pmc { +namespace net { + +// ============================================================================ +// Low-level protocol: 4-byte big-endian length prefix + UTF-8 payload +// ============================================================================ + +namespace detail { + +using namespace msg_constants; + +// ---- 4-byte big-endian 编码 / 解码 ---------------------------------------- + +inline void writeUint32BE(char *buf, uint32_t val) noexcept { + buf[0] = static_cast((val >> 24) & 0xFFu); + buf[1] = static_cast((val >> 16) & 0xFFu); + buf[2] = static_cast((val >> 8) & 0xFFu); + buf[3] = static_cast(val & 0xFFu); +} + +inline uint32_t readUint32BE(const char *buf) noexcept { + return (static_cast(static_cast(buf[0])) << 24) | + (static_cast(static_cast(buf[1])) << 16) | + (static_cast(static_cast(buf[2])) << 8) | + (static_cast(static_cast(buf[3]))); +} + +// ---- socket 超时设置 ------------------------------------------------------ + +inline void setSockTimeout(int sockfd, double seconds) { + struct timeval tv {}; + tv.tv_sec = static_cast(seconds); + tv.tv_usec = + static_cast((seconds - static_cast(tv.tv_sec)) * 1'000'000.0); + ::setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); + ::setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)); +} + +// ---- 精确读取 n 字节 ------------------------------------------------------ + +inline void recvExact(int sockfd, char *buf, size_t n) { + size_t received = 0; + while (received < n) { + ssize_t r = ::recv(sockfd, buf + received, n - received, 0); + if (r == 0) { + throw MsgConnectionException( + "connection closed: expected " + std::to_string(n) + + " bytes, got " + std::to_string(received)); + } + if (r < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + throw MsgTimeoutException("recv timed out"); + } + throw MsgConnectionException( + "recv error: " + std::string(::strerror(errno))); + } + received += static_cast(r); + } +} + +// ---- deadline 模式精确读取 n 字节（总超时，防 Slowloris）------------------- +// 与 Python 版 _recv_exact 对齐：每次 select 前按剩余时间收紧超时， +// header+payload 共享同一个 deadline，总接收时间不超过 timeout_sec。 + +inline void recvExactDeadline(int sockfd, char *buf, size_t n, + const std::chrono::steady_clock::time_point& deadline) { + size_t received = 0; + while (received < n) { + auto now = std::chrono::steady_clock::now(); + if (now >= deadline) { + throw MsgTimeoutException("recv timed out"); + } + auto remaining_dur = std::chrono::duration_cast(deadline - now); + double remaining = static_cast(remaining_dur.count()) / 1'000'000.0; + if (remaining < 0.001) remaining = 0.001; // 最小 1ms + + fd_set rfds; + FD_ZERO(&rfds); + FD_SET(sockfd, &rfds); + struct timeval tv; + tv.tv_sec = static_cast(remaining); + tv.tv_usec = static_cast((remaining - tv.tv_sec) * 1'000'000.0); + + int ready = ::select(sockfd + 1, &rfds, nullptr, nullptr, &tv); + if (ready < 0) { + if (errno == EINTR) continue; + throw MsgConnectionException(std::string("select error: ") + ::strerror(errno)); + } + if (ready == 0) { + throw MsgTimeoutException("recv timed out"); + } + + ssize_t r = ::recv(sockfd, buf + received, n - received, 0); + if (r == 0) { + throw MsgConnectionException( + "connection closed: expected " + std::to_string(n) + + " bytes, got " + std::to_string(received)); + } + if (r < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) continue; + throw MsgConnectionException( + "recv error: " + std::string(::strerror(errno))); + } + received += static_cast(r); + } +} + +// deadline 模式接收完整 pymsg 帧（header + payload） +inline std::string recvFrameDeadline(int sockfd, double timeout_sec) { + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::duration_cast( + std::chrono::duration(timeout_sec)); + + char header[kHeaderSize]; + recvExactDeadline(sockfd, header, kHeaderSize, deadline); + uint32_t length = readUint32BE(header); + if (length > kMaxMsgSize) { + throw MsgException("message too large: " + std::to_string(length)); + } + + std::string payload(static_cast(length), '\0'); + if (length > 0) { + recvExactDeadline(sockfd, payload.data(), length, deadline); + } + return payload; +} + +// ---- 完整发送 ------------------------------------------------------------- + +inline void sendAll(int sockfd, const char *data, size_t len) { + size_t sent = 0; + while (sent < len) { + ssize_t w = + ::send(sockfd, data + sent, len - sent, MSG_NOSIGNAL); + if (w < 0) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + throw MsgTimeoutException("send timed out"); + } + throw MsgConnectionException( + "send error: " + std::string(::strerror(errno))); + } + sent += static_cast(w); + } +} + +// ---- 短连接 helper：connect → action → close ------------------------------ + +inline int openConnection(const std::string& addr, int port, double timeout) { + int fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (fd < 0) { + throw MsgConnectionException( + "socket() failed: " + std::string(::strerror(errno))); + } + + struct sockaddr_in serv_addr {}; + serv_addr.sin_family = AF_INET; + serv_addr.sin_port = htons(static_cast(port)); + if (::inet_pton(AF_INET, addr.c_str(), &serv_addr.sin_addr) <= 0) { + ::close(fd); + throw MsgConnectionException("invalid address: " + addr); + } + + setSockTimeout(fd, timeout); + + if (::connect(fd, reinterpret_cast(&serv_addr), + sizeof(serv_addr)) < 0) { + int saved_errno = errno; + ::close(fd); + std::string err_msg = "could not connect to " + addr + ":" + + std::to_string(port) + " - " + ::strerror(saved_errno); + if (saved_errno == ECONNREFUSED) { + throw MsgConnectionRefusedException(err_msg); + } + throw MsgConnectionException(err_msg); + } + + return fd; +} + +} // namespace detail + +// ============================================================================ +// 底层协议函数：pmsg_send / pmsg_recv +// ============================================================================ + +/// 以 pymsg 协议发送一条字符串消息（长度前缀 + UTF-8 载荷）。 +inline void pmsg_send(int sockfd, const std::string &data) { + size_t payload_size = data.size(); + if (payload_size > msg_constants::kMaxMsgSize) { + throw MsgException( + "payload too large: " + std::to_string(payload_size) + + " bytes (max " + std::to_string(msg_constants::kMaxMsgSize) + ")"); + } + char header[detail::kHeaderSize]; + detail::writeUint32BE(header, static_cast(payload_size)); + detail::sendAll(sockfd, header, detail::kHeaderSize); + detail::sendAll(sockfd, data.c_str(), payload_size); +} + +/// 以 pymsg 协议接收一条字符串消息（阻塞模式，依赖 socket 已设置的超时）。 +/// @throws MsgConnectionException 连接在收全消息前断开。 +/// @throws MsgTimeoutException 等待超时。 +inline std::string pmsg_recv(int sockfd) { + char header[detail::kHeaderSize]; + detail::recvExact(sockfd, header, detail::kHeaderSize); + uint32_t length = detail::readUint32BE(header); + if (length > msg_constants::kMaxMsgSize) { // 安全上限 64 MiB + throw MsgException("message too large: " + std::to_string(length)); + } + std::string payload(static_cast(length), '\0'); + detail::recvExact(sockfd, payload.data(), length); + return payload; +} + +/// 以 pymsg 协议接收一条字符串消息（deadline 模式，总超时不超过 timeout_sec）。 +/// header 和 payload 共享同一个 deadline，防止 Slowloris 攻击。 +/// 不依赖 socket 的 SO_RCVTIMEO，使用 select() 实现精确超时控制。 +/// @param timeout_sec 总超时秒数 +/// @throws MsgConnectionException 连接在收全消息前断开。 +/// @throws MsgTimeoutException 总超时到期。 +inline std::string pmsg_recv(int sockfd, double timeout_sec) { + return detail::recvFrameDeadline(sockfd, timeout_sec); +} + +// ============================================================================ +// MsgClient +// ============================================================================ + +/// 基于 pymsg 协议的 TCP 客户端。 +/// +/// 支持三种操作模式： +/// - send() 短连接发送一条消息（不等待回复），消息带 name 前缀。 +/// - call() 短连接发送 + 阻塞等待回复（RPC 模式），原始消息不添加前缀。 +/// - listen() 基于 connect() 维持的长连接，持续接收消息并回调。 +class MsgClient : public Msg_Basic { +public: + MsgClient(const std::string &addr, int port, const std::string &name) + : _addr(addr), _port(port), _name(name), _sockfd(-1), _running(false) {} + + ~MsgClient() { stop(); } + + // ---- 连接 --------------------------------------------------------------- + + /// 创建 TCP 套接字并连接到服务器（长连接，供 listen() 使用）。 + void connect() { + int fd = detail::openConnection(_addr, _port, /*timeout=*/static_cast(msg_constants::kDefaultTimeout)); + // 如果已有旧连接，先关闭 + int old = _sockfd.exchange(fd); + if (old >= 0) { + ::shutdown(old, SHUT_RDWR); + ::close(old); + } + } + + // ---- 发送单条消息（短连接，带名字前缀）---------------------------------- + + /// 短连接发送一条消息到服务器，消息格式为 ``"name: msg"``。 + /// 不等待回复。短连接（每次创建新 socket，发完即关）。 + void send(const std::string &msg) { + int fd = detail::openConnection(_addr, _port, /*timeout=*/static_cast(msg_constants::kSendTimeout)); + try { + pmsg_send(fd, _name + ": " + msg); + ::close(fd); + } catch (...) { + ::close(fd); + throw; + } + } + + // ---- 请求-应答（短连接，RPC 模式）--------------------------------------- + + /// 短连接发送一条消息，阻塞等待并返回一条回复。 + /// + /// 不会给消息添加 ``name:`` 前缀（适用于 RPC/API 调用）。 + /// + /// @param msg 要发送的原始字符串。 + /// @param timeout 超时秒数，默认 10.0。 + /// @param retry_on_refused 目标服务未启动（ECONNREFUSED）时是否阻塞重试， + /// 直到连接成功。适用于启动顺序不确定的服务发现场景。 + /// 默认 false（连接拒绝立即抛异常）。 + /// @param retry_interval 重试间隔秒数，默认 1.0。 + /// @returns 服务器返回的字符串。 + std::string call(const std::string &msg, double timeout = static_cast(msg_constants::kDefaultTimeout), + bool retry_on_refused = false, + double retry_interval = 1.0) { + if (!retry_on_refused) { + // 快速路径：不重试，直接执行一次 + int fd = detail::openConnection(_addr, _port, timeout); + try { + pmsg_send(fd, msg); + // deadline 模式 recv：header+payload 共享总超时，防止总等待时间翻倍 + std::string resp = pmsg_recv(fd, timeout); + ::close(fd); + return resp; + } catch (...) { + ::close(fd); + throw; + } + } + + // 重试路径：仅对 ECONNREFUSED 重试，其他错误立即抛出 + int attempt = 0; + auto last_log = std::chrono::steady_clock::now() - + std::chrono::duration_cast( + std::chrono::duration(msg_constants::kRetryLogEvery)); + + while (true) { + try { + int fd = detail::openConnection(_addr, _port, timeout); + try { + pmsg_send(fd, msg); + std::string resp = pmsg_recv(fd, timeout); + ::close(fd); + return resp; + } catch (...) { + ::close(fd); + throw; + } + } catch (const MsgConnectionRefusedException &) { + ++attempt; + auto now = std::chrono::steady_clock::now(); + double elapsed = std::chrono::duration(now - last_log).count(); + if (elapsed >= msg_constants::kRetryLogEvery || attempt == 1) { + std::fprintf(stderr, + "[pymsg] 等待服务 %s:%d 就绪... (已重试 %d 次)\n", + _addr.c_str(), _port, attempt); + std::fflush(stderr); + last_log = now; + } + std::this_thread::sleep_for( + std::chrono::duration(retry_interval)); + } + // 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出 + } + } + + // ---- 持续监听（长连接，阻塞回调模式）----------------------------------- + + /// 在长连接上持续接收消息并通过回调处理。 + /// + /// 阻塞当前线程直到 stop() 被调用或连接断开。 + /// 必须先调用 connect()。 + /// + /// @param on_message 回调 ``void(const std::string &msg)``。 + void listen(std::function on_message) { + int fd = _sockfd.load(); + if (fd < 0) { + throw MsgException("must call connect() before listen()"); + } + + _running = true; + while (_running.load()) { + // 使用局部 fd，避免与 stop() 竞争导致 FD_SET(-1) UB + // 若 stop() 已将 _sockfd 置为 -1，下次循环 _running=false 退出 + int cur_fd = _sockfd.load(); + if (cur_fd < 0) break; + + fd_set readfds; + FD_ZERO(&readfds); + FD_SET(cur_fd, &readfds); + + struct timeval tv {}; + tv.tv_sec = 0; + tv.tv_usec = msg_constants::kListenPollMs; // 100 ms 轮询间隔 + + int ready = + ::select(cur_fd + 1, &readfds, nullptr, nullptr, &tv); + if (ready < 0) { + if (errno == EINTR) continue; + break; // EBADF (socket 被 stop() 关闭) 或其他致命错误 + } + if (ready == 0) continue; // timeout, loop around + + try { + std::string text = pmsg_recv(cur_fd); + on_message(std::move(text)); + } catch (const MsgConnectionException &) { + break; // 连接断开，退出循环 + } catch (const MsgTimeoutException &) { + continue; // 读超时但连接仍在，继续等待 + } + } + + // 原子地认领 fd 并关闭，防止与 stop() double-close + int close_fd = _sockfd.exchange(-1); + if (close_fd >= 0) { + ::shutdown(close_fd, SHUT_RDWR); + ::close(close_fd); + } + } + + // ---- 停止 --------------------------------------------------------------- + + /// 设置停止标志并关闭套接字。可从任意线程调用。 + void stop() noexcept { + _running = false; + // 原子地认领 fd 并关闭，防止与 listen() double-close + int fd = _sockfd.exchange(-1); + if (fd >= 0) { + ::shutdown(fd, SHUT_RDWR); // 唤醒阻塞在 recv/select 上的线程 + ::close(fd); + } + } + + // ---- 访问器 ------------------------------------------------------------- + + const std::string &addr() const noexcept { return _addr; } + int port() const noexcept { return _port; } + const std::string &name() const noexcept { return _name; } + +private: + std::string _addr; + int _port; + std::string _name; + std::atomic _sockfd; + std::atomic _running; +}; + +} // namespace net +} // namespace pmc diff --git a/libmsg/inc/us/MsgServer.hpp b/libmsg/inc/us/MsgServer.hpp new file mode 100644 index 0000000..362dc60 --- /dev/null +++ b/libmsg/inc/us/MsgServer.hpp @@ -0,0 +1,528 @@ +#pragma once + +/** MsgServer — 基于 pymsg 协议的多客户端 TCP 服务器（C++ 版，对应 Python 版 pymsg.MsgServer） + * + * 提供 select-based 事件循环 + 可选线程池并发处理，支持 reply()/broadcast() 线程安全发送。 + * + * 设计要点（与 Python 版对齐）： + * - select 事件循环，支持多客户端并发连接 + * - workers=0 同步模式（回调在主线程执行），workers>0 线程池并发模式 + * - 同一连接消息严格串行：回调执行期间暂停该连接 select 监听 + * - reply()/broadcast() 经每连接发送锁串行化，防止帧交错 + * - max_connections 限制活跃连接数，防止连接耗尽攻击 + * - max_queue_size 限制线程池任务队列深度，慢 handler 时断开连接防 OOM + * - deadline 模式 recv 超时，防 Slowloris 攻击 + * - stop() 可从任意线程（包括回调内部）安全调用，优雅关闭 + * - worker 线程出错仅标记 failed，主循环统一清理连接，避免 fd 并发操作 + * + * Usage: + * #include + * + * pmc::net::MsgServer server("0.0.0.0", 8003, {.workers=4}); + * server.start([&](int fd, const std::string& addr, const std::string& msg) { + * server.reply(fd, "echo: " + msg); // 回复发送者 + * server.broadcast(addr + ": " + msg, fd); // 广播给其他人 + * }); + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#ifndef MSG_NOSIGNAL +#define MSG_NOSIGNAL 0 +#endif + +namespace pmc { +namespace net { + +// ============================================================================ +// MsgServer 配置 +// ============================================================================ + +struct MsgServerConfig { + int workers = 0; ///< 线程池大小；0 = 同步模式（回调在主线程执行） + double recv_timeout = 10.0; ///< 接收单条消息的总超时（秒），防 Slowloris + int max_connections = 1000; ///< 最大并发连接数，超过则拒绝新连接 + int max_queue_size = 1000; ///< 线程池任务队列上限（workers>0 时），满时断开连接 + int listen_backlog = 10; ///< listen() backlog + double poll_timeout = 0.1; ///< select 轮询超时（秒），影响关闭响应速度 +}; + +// ============================================================================ +// MsgServer +// ============================================================================ + +/// 基于 pymsg 长度前缀协议的多客户端 TCP 服务器。 +class MsgServer { +public: + /// 消息回调签名：(客户端 fd, 客户端地址 "ip:port", 消息内容) + using MessageCallback = std::function; + + explicit MsgServer(const std::string& addr, int port, MsgServerConfig config = {}) + : _addr(addr), _port(port), _config(config), _server_fd(-1), _running(false) {} + + ~MsgServer() { stop(); } + + MsgServer(const MsgServer&) = delete; + MsgServer& operator=(const MsgServer&) = delete; + + // ---- 启动（阻塞当前线程）------------------------------------------------ + + /** 启动服务器，阻塞当前线程直到 stop() 被调用或发生致命错误。 + * + * @param on_message 消息回调，签名 void(int fd, const string& addr, const string& msg)。 + * - 同步模式（workers=0）：回调在主线程执行，同一连接消息天然串行。 + * - 线程池模式（workers>0）：回调提交到线程池并发执行；同一连接 + * 上的消息仍严格串行（回调完成前该连接暂停 select 监听）。 + * 回调内可安全调用 reply()/broadcast()/stop()。 + */ + void start(MessageCallback on_message) { + _on_message = std::move(on_message); + _running = true; + + // 创建监听 socket + _server_fd = ::socket(AF_INET, SOCK_STREAM, 0); + if (_server_fd < 0) { + throw MsgConnectionException(std::string("socket() failed: ") + ::strerror(errno)); + } + + int optval = 1; + ::setsockopt(_server_fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)); + + struct sockaddr_in serv_addr {}; + serv_addr.sin_family = AF_INET; + serv_addr.sin_port = htons(static_cast(_port)); + if (::inet_pton(AF_INET, _addr.c_str(), &serv_addr.sin_addr) <= 0) { + ::close(_server_fd); + _server_fd = -1; + throw MsgConnectionException("invalid bind address: " + _addr); + } + + if (::bind(_server_fd, reinterpret_cast(&serv_addr), sizeof(serv_addr)) < 0) { + int saved_errno = errno; + ::close(_server_fd); + _server_fd = -1; + throw MsgConnectionException(std::string("bind() failed: ") + ::strerror(saved_errno)); + } + + if (::listen(_server_fd, _config.listen_backlog) < 0) { + int saved_errno = errno; + ::close(_server_fd); + _server_fd = -1; + throw MsgConnectionException(std::string("listen() failed: ") + ::strerror(saved_errno)); + } + + // 启动线程池（如果配置了 workers） + std::vector pool_threads; + if (_config.workers > 0) { + _pool_running = true; + pool_threads.reserve(_config.workers); + for (int i = 0; i < _config.workers; ++i) { + pool_threads.emplace_back(&MsgServer::_worker_loop, this); + } + } + + // 主事件循环 + try { + _event_loop(); + } catch (...) { + // 确保线程池和连接被清理 + _running = false; + _shutdown_pool(pool_threads); + _close_all_clients(); + if (_server_fd >= 0) { ::close(_server_fd); _server_fd = -1; } + throw; + } + + _shutdown_pool(pool_threads); + _close_all_clients(); + if (_server_fd >= 0) { ::close(_server_fd); _server_fd = -1; } + } + + // ---- 停止 -------------------------------------------------------------- + + /// 停止服务器（可从任意线程调用，包括回调内部）。 + /// 关闭监听 socket 以解除 select 阻塞，事件循环退出前清理所有连接。 + void stop() noexcept { + bool expected = true; + if (!_running.compare_exchange_strong(expected, false)) return; + if (_server_fd >= 0) { + ::shutdown(_server_fd, SHUT_RDWR); + ::close(_server_fd); + _server_fd = -1; + } + } + + // ---- 发送（线程安全）---------------------------------------------------- + + /** 向指定客户端发送一条应答消息（线程安全）。 + * + * 经每连接发送锁串行化，workers>0 时可在回调线程安全调用。 + * 如果连接已断开则静默忽略（fail-safe，不抛异常）。 + */ + void reply(int sockfd, const std::string& msg) { + std::shared_ptr send_lock; + { + std::lock_guard lock(_clients_mtx); + auto it = _send_locks.find(sockfd); + if (it == _send_locks.end()) return; // 连接已断开 + send_lock = it->second; + } + std::lock_guard lock(*send_lock); + try { + pmsg_send(sockfd, msg); + } catch (const MsgException&) { + // 连接可能已断开，静默忽略 + } + } + + /** 向所有已连接客户端广播消息（线程安全）。 + * + * @param msg 要广播的消息。 + * @param except_fd 排除的 fd（通常是发送者自己），默认 -1 表示不排除。 + */ + void broadcast(const std::string& msg, int except_fd = -1) { + // 先拷贝一份 fd 列表，避免持锁期间发送 + std::vector>> targets; + { + std::lock_guard lock(_clients_mtx); + targets.reserve(_clients.size()); + for (auto& [fd, addr] : _clients) { + if (fd == except_fd) continue; + auto it = _send_locks.find(fd); + if (it != _send_locks.end()) { + targets.emplace_back(fd, it->second); + } + } + } + for (auto& [fd, send_lock] : targets) { + std::lock_guard slock(*send_lock); + try { + pmsg_send(fd, msg); + } catch (const MsgException&) { + // 发送失败，标记为 failed 由主循环清理 + std::lock_guard flk(_failed_mtx); + _failed.insert(fd); + } + } + } + + // ---- 状态查询 ---------------------------------------------------------- + + /// 当前已连接客户端数量。 + size_t client_count() const { + std::lock_guard lock(_clients_mtx); + return _clients.size(); + } + +private: + // ---- 内部：线程池 ------------------------------------------------------- + + struct Task { + int fd; + std::string addr; + std::string message; + }; + + void _worker_loop() { + while (true) { + Task task; + bool has_task = false; + { + std::unique_lock lock(_pool_mtx); + _pool_cv.wait(lock, [this] { return !_pool_running || !_pool_queue.empty(); }); + if (!_pool_queue.empty()) { + task = std::move(_pool_queue.front()); + _pool_queue.pop(); + has_task = true; + } else if (!_pool_running) { + return; // 队列空且已停止，退出 + } + } + if (!has_task) continue; + + // 执行回调 + bool failed = false; + try { + _on_message(task.fd, task.addr, std::move(task.message)); + } catch (const MsgConnectionException&) { + failed = true; + } catch (const std::exception& e) { + std::fprintf(stderr, "[MsgServer] callback error on fd=%d: %s\n", + task.fd, e.what()); + std::fflush(stderr); + failed = true; + } catch (...) { + failed = true; + } + if (failed) { + std::lock_guard flk(_failed_mtx); + _failed.insert(task.fd); + } + // 清除 busy 标记，恢复 select 监听 + { + std::lock_guard lock(_busy_mtx); + _busy.erase(task.fd); + } + } + } + + void _shutdown_pool(std::vector& pool_threads) { + // 停止接收新任务，唤醒所有 worker 排空队列后退出 + _pool_running = false; + _pool_cv.notify_all(); + for (auto& t : pool_threads) { + if (t.joinable()) t.join(); + } + pool_threads.clear(); + // worker 已排空队列，此时 busy 集合中的任务都已完成 + // 任何仍标记为 failed 的连接由 _close_all_clients 统一清理 + } + + // ---- 内部：主事件循环 --------------------------------------------------- + + void _event_loop() { + while (_running) { + // 清理 failed 连接 + _reap_failed(); + + // 构建 fd_set：server_fd + 非 busy 的客户端 fd + fd_set readfds, exceptfds; + FD_ZERO(&readfds); + FD_ZERO(&exceptfds); + int maxfd = -1; + + if (_server_fd >= 0) { + FD_SET(_server_fd, &readfds); + maxfd = _server_fd; + } + + std::vector watched_clients; + { + std::lock_guard clk(_clients_mtx); + std::lock_guard blk(_busy_mtx); + watched_clients.reserve(_clients.size()); + for (auto& [fd, addr] : _clients) { + if (_busy.count(fd)) continue; // 回调处理中，暂停监听 + FD_SET(fd, &readfds); + FD_SET(fd, &exceptfds); + if (fd > maxfd) maxfd = fd; + watched_clients.push_back(fd); + } + } + + if (maxfd < 0) { + // 没有任何 fd 可监听（server_fd 已关闭且无客户端），退出 + break; + } + + struct timeval tv; + tv.tv_sec = 0; + tv.tv_usec = static_cast(_config.poll_timeout * 1'000'000.0); + + int ready = ::select(maxfd + 1, &readfds, nullptr, &exceptfds, &tv); + if (ready < 0) { + if (errno == EINTR) continue; + break; // EBADF 等致命错误 + } + if (ready == 0) continue; + + // 处理异常 fd + for (int fd : watched_clients) { + if (FD_ISSET(fd, &exceptfds)) { + _remove_client(fd); + } + } + + // 处理新连接 + if (_server_fd >= 0 && FD_ISSET(_server_fd, &readfds)) { + _accept_client(); + } + + // 处理客户端消息 + for (int fd : watched_clients) { + if (FD_ISSET(fd, &readfds)) { + // 检查 fd 是否仍在客户端列表中（可能已被异常处理移除） + { + std::lock_guard lock(_clients_mtx); + if (!_clients.count(fd)) continue; + } + _handle_client(fd); + } + } + } + } + + void _accept_client() { + struct sockaddr_in client_addr {}; + socklen_t addrlen = sizeof(client_addr); + int client_fd = ::accept(_server_fd, reinterpret_cast(&client_addr), &addrlen); + if (client_fd < 0) { + // EINVAL/EBADF: server_fd 已被 stop() 关闭；其他错误忽略 + return; + } + if (client_fd >= FD_SETSIZE) { ::close(client_fd); return; } + + // 检查连接数上限 + { + std::lock_guard lock(_clients_mtx); + if (static_cast(_clients.size()) >= _config.max_connections) { + ::close(client_fd); + return; + } + char ip_str[INET_ADDRSTRLEN]; + ::inet_ntop(AF_INET, &client_addr.sin_addr, ip_str, sizeof(ip_str)); + std::string addr = std::string(ip_str) + ":" + std::to_string(ntohs(client_addr.sin_port)); + _clients[client_fd] = addr; + _send_locks[client_fd] = std::make_shared(); + } + + // 禁用 Nagle 算法，降低小包延迟 + int flag = 1; + ::setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(flag)); + } + + void _handle_client(int fd) { + std::string addr; + { + std::lock_guard lock(_clients_mtx); + auto it = _clients.find(fd); + if (it == _clients.end()) return; + addr = it->second; + } + + std::string message; + try { + // deadline 模式：header+payload 共享总超时，防 Slowloris + message = pmsg_recv(fd, _config.recv_timeout); + } catch (const MsgException&) { + _remove_client(fd); + return; + } + + if (_config.workers == 0) { + // 同步模式：在主线程执行回调 + try { + _on_message(fd, addr, std::move(message)); + } catch (const MsgConnectionException&) { + _remove_client(fd); + } catch (const std::exception& e) { + std::fprintf(stderr, "[MsgServer] callback error on fd=%d: %s\n", fd, e.what()); + std::fflush(stderr); + _remove_client(fd); + } + return; + } + + // 线程池模式：标记 busy，提交任务 + { + std::lock_guard blk(_busy_mtx); + _busy.insert(fd); + } + { + std::lock_guard plk(_pool_mtx); + if (static_cast(_pool_queue.size()) >= _config.max_queue_size) { + // 队列已满：拒绝并断开连接（防 OOM） + std::lock_guard blk(_busy_mtx); + _busy.erase(fd); + _remove_client(fd); + return; + } + _pool_queue.push(Task{fd, std::move(addr), std::move(message)}); + } + _pool_cv.notify_one(); + } + + void _reap_failed() { + std::unordered_set failed; + { + std::lock_guard flk(_failed_mtx); + failed.swap(_failed); + } + for (int fd : failed) { + { + std::lock_guard blk(_busy_mtx); + _busy.erase(fd); + } + _remove_client(fd); + } + } + + void _remove_client(int fd) { + std::lock_guard clk(_clients_mtx); + if (_clients.erase(fd) == 0) return; // 已被移除，防止 double-close + _send_locks.erase(fd); + ::close(fd); + } + + void _close_all_clients() { + std::lock_guard clk(_clients_mtx); + for (auto& [fd, addr] : _clients) { + ::close(fd); + } + _clients.clear(); + _send_locks.clear(); + std::lock_guard blk(_busy_mtx); + _busy.clear(); + std::lock_guard flk(_failed_mtx); + _failed.clear(); + } + + // ---- 成员 -------------------------------------------------------------- + + std::string _addr; + int _port; + MsgServerConfig _config; + int _server_fd; + std::atomic _running; + MessageCallback _on_message; + + // 客户端管理 + mutable std::mutex _clients_mtx; + std::unordered_map _clients; ///< fd -> "ip:port" + std::unordered_map> _send_locks; ///< fd -> 发送锁 + + // 同连接串行 + std::mutex _busy_mtx; + std::unordered_set _busy; ///< 回调执行中的 fd + + // 失败连接（worker 线程标记，主循环清理） + std::mutex _failed_mtx; + std::unordered_set _failed; + + // 线程池 + std::atomic _pool_running{false}; + std::mutex _pool_mtx; + std::condition_variable _pool_cv; + std::queue _pool_queue; +}; + +} // namespace net +} // namespace pmc diff --git a/libmsg/inc/us/Msg_Basic.hpp b/libmsg/inc/us/Msg_Basic.hpp new file mode 100644 index 0000000..1880f67 --- /dev/null +++ b/libmsg/inc/us/Msg_Basic.hpp @@ -0,0 +1,79 @@ +#pragma once + +/** libmsg — UniServe 消息通信基础库 + * + * 提供 pymsg 协议的 C++ 实现：4字节大端长度前缀 + UTF-8 载荷的 TCP 消息帧协议。 + * 对应 Python 版 pymsg/pymsg/__init__.py，字节级兼容。 + * + * 命名空间 pmc::net，库文件 libus_msg.so，头文件安装至 /usr/include/us/。 + */ + +#include +#include +#include +#include + +#define MSG_CHECK(condition, message) \ + if (!(condition)) { \ + throw std::runtime_error(message); \ + } + +namespace pmc { +namespace net { + +/// 协议常量 +namespace msg_constants { + constexpr size_t kHeaderSize = 4; ///< 帧头长度（字节） + constexpr uint32_t kMaxMsgSize = 64u * 1024u * 1024u; ///< 消息最大长度 64 MiB + constexpr int kDefaultTimeout = 10; ///< 默认超时（秒） + constexpr int kSendTimeout = 3; ///< send() 默认超时（秒） + constexpr int kListenPollMs = 100'000; ///< listen() select 轮询间隔（微秒） + constexpr double kRetryLogEvery = 5.0; ///< 重试等待日志间隔（秒） +} + +// ============================================================================ +// 异常层次 +// ============================================================================ + +/// 消息库基础异常 +class MsgException : public std::runtime_error { +public: + explicit MsgException(const std::string &msg) : std::runtime_error(msg) {} +}; + +/// 消息收发超时 +class MsgTimeoutException : public MsgException { +public: + explicit MsgTimeoutException(const std::string &msg) : MsgException(msg) {} +}; + +/// 连接异常（断开、重置、地址错误等） +class MsgConnectionException : public MsgException { +public: + explicit MsgConnectionException(const std::string &msg) : MsgException(msg) {} +}; + +/// 目标服务未启动（ECONNREFUSED），用于 retry_on_refused 重试机制 +class MsgConnectionRefusedException : public MsgConnectionException { +public: + explicit MsgConnectionRefusedException(const std::string &msg) + : MsgConnectionException(msg) {} +}; + +// ============================================================================ +// 基类（供 MsgClient 等内部工具方法继承使用） +// ============================================================================ + +class Msg_Basic { +public: + class MsgExp : public std::runtime_error { + public: + MsgExp(const std::string& msg) : std::runtime_error(msg) {} + }; +protected: + /// 版本字符串（编译进 .so） + static const char* version(); +}; + +} // namespace net +} // namespace pmc diff --git a/libmsg/inc/us/RpcClient.hpp b/libmsg/inc/us/RpcClient.hpp new file mode 100644 index 0000000..1421029 --- /dev/null +++ b/libmsg/inc/us/RpcClient.hpp @@ -0,0 +1,181 @@ +#pragma once + +/** RpcClient — pymsg 线协议之上的 JSON RPC 客户端（C++ 版，对应 Python 版 pymsg.rpc.RpcClient） + * + * 约定（与 Python/Go 客户端字节级兼容）： + * 请求: {"_route": "<路由名>", ...业务字段} + * 响应: {"success": true, "data": ...} + * {"success": false, "error": "..."} + * + * 基于 boost::json 进行 JSON 序列化/反序列化，依赖 MsgClient 进行 TCP 通信。 + * + * 注意：此头文件依赖 boost::json，核心 libmsg 库（Msg.hpp/MsgServer.hpp/AsyncQueue.hpp） + * 保持零外部依赖不变。使用 RpcClient 时需要链接 -lboost_json。 + * + * Usage: + * #include + * + * pmc::net::RpcClient rpc("127.0.0.1", 8005, "gateway"); + * json::object params; + * params["action"] = "select"; + * params["table"] = "Users"; + * json::value data = rpc.call("execute", params); // 返回 data 字段 + * + * // 启动等待模式（db 未启动时阻塞重试） + * pmc::net::RpcClient rpc("127.0.0.1", 8005, "gateway", 10.0, true); + */ + +#include + +#include +#include +#include +#include + +#include + +namespace pmc { +namespace net { + +namespace json = boost::json; + +// ============================================================================ +// RpcError — RPC 业务异常 +// ============================================================================ + +/// 远端返回 success=false 的业务错误，或响应不符合 RPC 约定（非 JSON / 缺少 success 字段）。 +/// 网络层异常（超时、连接拒绝等）仍抛 MsgTimeoutException / MsgConnectionRefusedException 等。 +class RpcError : public MsgException { +public: + explicit RpcError(const std::string& msg) : MsgException(msg) {} +}; + +// ============================================================================ +// RpcClient +// ============================================================================ + +/// JSON RPC 客户端（短连接，每次调用独立 TCP 连接）。 +class RpcClient { +public: + /** 构造 RPC 客户端。 + * + * @param addr 服务地址（通常 127.0.0.1） + * @param port 服务端口 + * @param name 客户端名称（用于调试/日志） + * @param timeout 默认超时秒数 + * @param retry_on_refused 默认是否在 ECONNREFUSED 时阻塞重试 + * @param retry_interval 重试间隔秒数 + */ + explicit RpcClient(const std::string& addr, int port, + const std::string& name = "rpc", + double timeout = 10.0, + bool retry_on_refused = false, + double retry_interval = 1.0) + : _addr(addr), _port(port), _name(name), + _timeout(timeout), + _retry_on_refused(retry_on_refused), + _retry_interval(retry_interval) {} + + // ---- RPC 调用 ---------------------------------------------------------- + + /** 调用指定路由，返回响应中的 data 字段。 + * + * @param route 路由名（自动注入为 _route 字段） + * @param params 业务参数（_route 会被覆盖） + * @param timeout 本次调用超时（nullopt = 使用构造时默认值） + * @param retry_on_refused 本次调用是否重试连接拒绝（nullopt = 使用构造时默认值） + * @return 服务端返回的 data 字段值（可能为 null） + * @throws RpcError 服务端返回 success=false，或响应非法 + * @throws MsgTimeoutException 网络超时 + * @throws MsgConnectionRefusedException 目标不可达且未重试 + * @throws MsgConnectionException 连接中断 + */ + json::value call(const std::string& route, + const json::object& params = {}, + std::optional timeout = std::nullopt, + std::optional retry_on_refused = std::nullopt, + std::optional retry_interval = std::nullopt) { + json::object req = params; + req["_route"] = route; + + double t = timeout.value_or(_timeout); + bool retry = retry_on_refused.value_or(_retry_on_refused); + double interval = retry_interval.value_or(_retry_interval); + + MsgClient client(_addr, _port, _name); + std::string resp_str = client.call(json::serialize(req), t, retry, interval); + + json::value resp; + try { + resp = json::parse(resp_str); + } catch (const std::exception& e) { + throw RpcError("响应不是合法 JSON: " + std::string(e.what())); + } + + if (!resp.is_object()) { + throw RpcError("非法 RPC 响应（不是 object）: " + resp_str.substr(0, 200)); + } + const json::object& obj = resp.as_object(); + if (!obj.contains("success")) { + throw RpcError("非法 RPC 响应（缺少 success 字段）: " + resp_str.substr(0, 200)); + } + if (!obj.at("success").as_bool()) { + std::string err = "未知错误"; + if (auto it = obj.find("error"); it != obj.end() && it->value().is_string()) { + err = it->value().as_string().c_str(); + } + throw RpcError(err); + } + if (auto it = obj.find("data"); it != obj.end()) { + return it->value(); + } + return nullptr; + } + + /** 调用指定路由，返回完整响应对象（包含 success/data/error），不抛 RpcError。 + * 适用于需要自行检查 success 字段或访问额外字段的场景。 + */ + json::object call_raw(const std::string& route, + const json::object& params = {}, + std::optional timeout = std::nullopt, + std::optional retry_on_refused = std::nullopt, + std::optional retry_interval = std::nullopt) { + json::object req = params; + req["_route"] = route; + + double t = timeout.value_or(_timeout); + bool retry = retry_on_refused.value_or(_retry_on_refused); + double interval = retry_interval.value_or(_retry_interval); + + MsgClient client(_addr, _port, _name); + std::string resp_str = client.call(json::serialize(req), t, retry, interval); + + json::value resp; + try { + resp = json::parse(resp_str); + } catch (const std::exception& e) { + throw RpcError("响应不是合法 JSON: " + std::string(e.what())); + } + if (!resp.is_object()) { + throw RpcError("非法 RPC 响应（不是 object）: " + resp_str.substr(0, 200)); + } + return resp.as_object(); + } + + // ---- 访问器 ------------------------------------------------------------ + + const std::string& addr() const noexcept { return _addr; } + int port() const noexcept { return _port; } + const std::string& name() const noexcept { return _name; } + +private: + std::string _addr; + int _port; + std::string _name; + double _timeout; + bool _retry_on_refused; + double _retry_interval; +}; + +} // namespace net +} // namespace pmc diff --git a/libmsg/inc/us/RpcServer.hpp b/libmsg/inc/us/RpcServer.hpp new file mode 100644 index 0000000..ed25584 --- /dev/null +++ b/libmsg/inc/us/RpcServer.hpp @@ -0,0 +1,199 @@ +#pragma once + +/** RpcServer — pymsg 线协议之上的 JSON RPC 服务端（C++ 版，对应 Python 版 pymsg.rpc.RpcServer） + * + * 约定（与 Python/Go 客户端字节级兼容）： + * 请求: {"_route": "<路由名>", ...业务字段} + * 响应: {"success": true, "data": ...} + * {"success": false, "error": "..."} + * + * 基于 MsgServer 提供路由分发、JSON 解析、响应自动包装、异常脱敏等功能。 + * handler 签名为 ``json::value(const json::object& req)``： + * - 返回值自动包装为 {"success": true, "data": ...} + * - 抛出的异常自动包装为 {"success": false, "error": "<脱敏后错误信息>"} + * - 服务不会因 handler 异常而崩溃 + * + * 注意：此头文件依赖 boost::json，核心 libmsg 库保持零外部依赖不变。 + * 使用 RpcServer 时需要链接 -lboost_json。 + * + * Usage: + * #include + * + * pmc::net::RpcServer rpc("127.0.0.1", 8005, {.workers=4}); + * + * rpc.add_route("ping", [](const json::object&) -> json::value { + * return "pong"; + * }); + * + * rpc.add_route("execute", [](const json::object& req) -> json::value { + * // 处理业务逻辑... + * return json::object{{"rows", json::array{}}}; + * }); + * + * rpc.start(); // 阻塞当前线程 + */ + +#include + +#include + +#include +#include +#include +#include +#include +#include + +#include + +namespace pmc { +namespace net { + +namespace json = boost::json; + +// ============================================================================ +// RpcServer +// ============================================================================ + +/// JSON RPC 服务端：基于 MsgServer，自带路由分发与响应包装。 +class RpcServer { +public: + /// 路由处理函数签名：接收请求参数 object，返回 data 值（任意 JSON 类型）。 + using Handler = std::function; + + /// 错误回调：void(route_name, exception)，handler 抛异常时调用（用于记日志）。 + using ErrorCallback = std::function; + + /** 构造 RPC 服务端。 + * + * @param addr 监听地址 + * @param port 监听端口 + * @param config MsgServer 配置（线程池/超时/连接数限制等） + * @param on_error 可选错误回调，handler 抛异常时被调用 + */ + explicit RpcServer(const std::string& addr, int port, + MsgServerConfig config = {}, + ErrorCallback on_error = {}) + : _server(addr, port, config), _on_error(std::move(on_error)) {} + + // ---- 路由注册 ---------------------------------------------------------- + + /** 注册路由处理函数。 + * + * @param name 路由名（对应请求中的 _route 字段） + * @param handler 处理函数，签名 json::value(const json::object& req) + */ + void add_route(const std::string& name, Handler handler) { + _routes[name] = std::move(handler); + } + + // ---- 启动 / 停止 ------------------------------------------------------- + + /// 启动服务（阻塞当前线程）。 + void start() { + _server.start([this](int fd, const std::string& addr, const std::string& msg) { + this->_dispatch(fd, addr, msg); + }); + } + + /// 从另一线程停止服务。 + void stop() { + _server.stop(); + } + + /// 获取底层 MsgServer 引用（可用于 broadcast 等高级操作）。 + MsgServer& server() noexcept { return _server; } + +private: + // ---- 分发与响应包装 ---------------------------------------------------- + + void _dispatch(int fd, const std::string& /*addr*/, const std::string& text) { + // 解析 JSON（限制嵌套深度防止栈溢出） + json::parse_options opts; + opts.max_depth = 32; // 与 Python 版一致 + + boost::system::error_code ec; + json::value req_val = json::parse(text, ec, {}, opts); + if (ec || !req_val.is_object()) { + _reply_error(fd, "invalid JSON request"); + return; + } + + json::object req = req_val.as_object(); + + // 提取 _route + std::string route; + if (auto it = req.find("_route"); it != req.end() && it->value().is_string()) { + route = it->value().as_string().c_str(); + req.erase(it); // 从业务参数中移除 _route + } else { + _reply_error(fd, "missing _route"); + return; + } + + // 查找 handler + auto handler_it = _routes.find(route); + if (handler_it == _routes.end()) { + _reply_error(fd, "unknown route: " + route); + return; + } + + // 执行 handler + try { + json::value result = handler_it->second(req); + _reply_success(fd, result); + } catch (const std::exception& e) { + if (_on_error) { + try { + _on_error(route, e); + } catch (...) {} + } + // 脱敏后返回：保留业务错误语义，过滤绝对路径等内部细节 + try { + _reply_error(fd, _sanitize_error(e.what())); + } catch (...) { + // _sanitize_error 或序列化失败时，使用安全的固定文案 + try { _reply_error(fd, "internal error"); } catch (...) {} + } + } + } + + void _reply_success(int fd, const json::value& data) { + json::object resp; + resp["success"] = true; + resp["data"] = data; + _server.reply(fd, json::serialize(resp)); + } + + void _reply_error(int fd, const std::string& error) { + json::object resp; + resp["success"] = false; + resp["error"] = error; + _server.reply(fd, json::serialize(resp)); + } + + // ---- 错误信息脱敏 ------------------------------------------------------ + // 匹配绝对路径（如 /opt/micro_servers/db/xxx.py 或 /usr/lib/...），替换为 + // 不返回函数堆栈、SQL、文件路径等内部细节给客户端；完整异常由 on_error 钩子记录到服务端日志。 + + static std::string _sanitize_error(const std::string& msg) { + if (msg.empty()) return msg; + try { + // 匹配以 / 开头的绝对路径段，替换为 + // 不返回函数堆栈、SQL、文件路径等内部细节给客户端 + static const std::regex path_re(R"(/[A-Za-z0-9_.\-]+(?:/[A-Za-z0-9_.\-]+)*)"); + return std::regex_replace(msg, path_re, ""); + } catch (...) { + return msg; // 正则异常时返回原文 + } + } + + // ---- 成员 -------------------------------------------------------------- + + MsgServer _server; + std::unordered_map _routes; + ErrorCallback _on_error; +}; + +} // namespace net +} // namespace pmc diff --git a/libmsg/src/Msg_Basic.cpp b/libmsg/src/Msg_Basic.cpp new file mode 100644 index 0000000..d48ac56 --- /dev/null +++ b/libmsg/src/Msg_Basic.cpp @@ -0,0 +1,11 @@ +#include "us/Msg_Basic.hpp" + +namespace pmc { +namespace net { + +const char* Msg_Basic::version() { + return "libus_msg 0.0.1"; +} + +} // namespace net +} // namespace pmc