feat(libmsg): add pymsg C++ library with AsyncQueue, MsgServer, and JSON RPC layer
- Add libmsg module: pymsg protocol implementation (4-byte BE length prefix + UTF-8) - Add AsyncQueue<T>: generic async batch queue with backpressure (aligned with Python AsyncRpcQueue) - Add MsgServer: select-based multi-client TCP server with optional thread pool - Add RpcClient/RpcServer: JSON RPC convention layer with automatic route injection - Refactor gateway/LogC.hpp to use AsyncQueue for async batch log delivery - Update gateway/DbC.hpp to add call_msg() using pymsg TCP protocol - Remove duplicated gateway/inc/Msg.hpp (now provided by libmsg) - Update gateway/Makefile and _inst.sh build order Fixes included: - pmsg_send size check uses kMaxMsgSize (64MiB) instead of 4GiB - MsgClient._running uses std::atomic<bool> for thread safety - Deadline-based recv timeout to prevent Slowloris (header+payload shared deadline)
This commit is contained in:
qing committed
1 parent
7ff73b7d07
commit
1d44e0dcb8
17 files changed
+2483
-482
No files matched your search
@@ -29,6 +29,7 @@ inst_module "libth"
|
||||
inst_module "libhttp"
|
||||
inst_module "libcrypt"
|
||||
inst_module "libsqlite3"
|
||||
inst_module "libmsg"
|
||||
inst_module "gateway"
|
||||
inst_module "waf" # waf 在 gateway 之前编译，部署时由 core 先拉起 waf 再拉起 gateway
|
||||
inst_module "static"
|
||||
|
||||
+6
-6
@@ -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/
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
#pragma once
|
||||
#include <us/HttpClient.hpp>
|
||||
#include "Msg.hpp"
|
||||
#include <us/Msg.hpp>
|
||||
#include <boost/json.hpp>
|
||||
namespace qing {
|
||||
namespace json = boost::json;
|
||||
|
||||
+39
-82
@@ -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 <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cstdio>
|
||||
#include <ctime>
|
||||
#include <deque>
|
||||
#include <iostream>
|
||||
#include <mutex>
|
||||
#include <optional>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include <boost/json.hpp>
|
||||
|
||||
#include "Msg.hpp"
|
||||
#include <us/AsyncQueue.hpp>
|
||||
#include <us/Msg.hpp>
|
||||
|
||||
namespace qing {
|
||||
|
||||
@@ -62,12 +60,8 @@ public:
|
||||
const std::string& message,
|
||||
std::optional<std::string> trace_id = std::nullopt,
|
||||
std::optional<std::string> extra = std::nullopt) {
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> lock(_endpoint_mtx);
|
||||
_db_addr = addr;
|
||||
_db_port = port;
|
||||
}
|
||||
|
||||
private:
|
||||
LogC() : _thread(&LogC::_flush_loop, this) {}
|
||||
LogC()
|
||||
: _queue(
|
||||
[this](std::vector<Entry>&& batch) { this->_send_batch(std::move(batch)); },
|
||||
{_BATCH_SIZE, _FLUSH_INTERVAL, _MAX_QUEUE_SIZE, 0.5},
|
||||
[](const std::exception& e, std::vector<Entry>&&) {
|
||||
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<Entry> batch;
|
||||
auto last_flush = std::chrono::steady_clock::now();
|
||||
|
||||
for (;;) {
|
||||
{
|
||||
std::unique_lock<std::mutex> 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<double>(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<std::mutex> 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<Entry>& batch) {
|
||||
void _send_batch(const std::vector<Entry>& batch) {
|
||||
std::string addr;
|
||||
int port;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<Entry> _queue;
|
||||
std::mutex _mtx;
|
||||
std::condition_variable _cv;
|
||||
bool _stop = false;
|
||||
std::atomic<bool> _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<Entry> _queue;
|
||||
};
|
||||
|
||||
} // namespace qing
|
||||
@@ -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 <cerrno>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <fcntl.h>
|
||||
#include <netinet/in.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/select.h>
|
||||
#include <sys/types.h>
|
||||
#include <unistd.h>
|
||||
|
||||
// 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<char>((val >> 24) & 0xFFu);
|
||||
buf[1] = static_cast<char>((val >> 16) & 0xFFu);
|
||||
buf[2] = static_cast<char>((val >> 8) & 0xFFu);
|
||||
buf[3] = static_cast<char>(val & 0xFFu);
|
||||
}
|
||||
|
||||
inline uint32_t readUint32BE(const char *buf) noexcept {
|
||||
return (static_cast<uint32_t>(static_cast<unsigned char>(buf[0])) << 24) |
|
||||
(static_cast<uint32_t>(static_cast<unsigned char>(buf[1])) << 16) |
|
||||
(static_cast<uint32_t>(static_cast<unsigned char>(buf[2])) << 8) |
|
||||
(static_cast<uint32_t>(static_cast<unsigned char>(buf[3])));
|
||||
}
|
||||
|
||||
// ---- socket 超时设置 ------------------------------------------------------
|
||||
|
||||
inline void setSockTimeout(int sockfd, double seconds) {
|
||||
struct timeval tv {};
|
||||
tv.tv_sec = static_cast<time_t>(seconds);
|
||||
tv.tv_usec =
|
||||
static_cast<long>((seconds - static_cast<double>(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<size_t>(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<size_t>(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<uint16_t>(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<struct sockaddr *>(&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<uint32_t>(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<size_t>(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::steady_clock::duration>(
|
||||
std::chrono::duration<double>(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<double>(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<double>(retry_interval));
|
||||
}
|
||||
// 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 持续监听（长连接，阻塞回调模式）-----------------------------------
|
||||
|
||||
/// 在长连接上持续接收消息并通过回调处理。
|
||||
///
|
||||
/// 阻塞当前线程直到 stop() 被调用或连接断开。
|
||||
/// 必须先调用 connect()。
|
||||
///
|
||||
/// @param on_message 回调 ``void(const std::string &msg)``。
|
||||
void listen(std::function<void(const std::string &)> 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
|
||||
@@ -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/
|
||||
@@ -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.
|
||||
@@ -0,0 +1,47 @@
|
||||
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 $(INCLU_SDIR)/$(FRAMEWORK_NAME)/MsgServer.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/AsyncQueue.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/RpcClient.hpp $(INCLU_TDIR)/$(FRAMEWORK_NAME)/
|
||||
@cp -vf $(INCLU_SDIR)/$(FRAMEWORK_NAME)/RpcServer.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 $(INCLU_TDIR)/$(FRAMEWORK_NAME)/MsgServer.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/AsyncQueue.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/RpcClient.hpp
|
||||
@rm -vf $(INCLU_TDIR)/$(FRAMEWORK_NAME)/RpcServer.hpp
|
||||
@rm -vf $(INSTL_TDIR)/$(LIB_NAME).so
|
||||
@echo "done."
|
||||
@@ -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<T> 通用异步批量发送队列（对应 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<T>` 和 `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 <us/Msg.hpp>
|
||||
|
||||
// 构造：指定服务地址、端口、客户端名称（用于 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 <us/MsgServer.hpp>
|
||||
|
||||
// 构造：绑定地址和端口，可配置线程池大小
|
||||
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\<T\> — 通用异步批量发送队列
|
||||
|
||||
```cpp
|
||||
#include <us/AsyncQueue.hpp>
|
||||
```
|
||||
|
||||
对应 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<T>&& batch)`，`on_error` 签名 `void(const exception&, vector<T>&&)` |
|
||||
| `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 <us/RpcClient.hpp> // 需链接 -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 <us/RpcServer.hpp> // 需链接 -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 层，防止栈溢出攻击
|
||||
- 错误信息自动脱敏：绝对路径替换为 `<redacted>`，不泄露内部文件路径
|
||||
- `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 <us/Msg_Basic.hpp>
|
||||
#include <us/Msg.hpp>
|
||||
#include <us/MsgServer.hpp> // 如需服务端
|
||||
#include <us/AsyncQueue.hpp> // 如需异步队列
|
||||
#include <us/RpcClient.hpp> // 如需 JSON RPC 客户端（需 boost::json）
|
||||
#include <us/RpcServer.hpp> // 如需 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 <us/Msg.hpp>
|
||||
#include <iostream>
|
||||
|
||||
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 <us/Msg.hpp>
|
||||
#include <iostream>
|
||||
|
||||
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 <us/MsgServer.hpp>
|
||||
#include <iostream>
|
||||
#include <thread>
|
||||
|
||||
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 <us/MsgServer.hpp>
|
||||
#include <thread>
|
||||
|
||||
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 <us/AsyncQueue.hpp>
|
||||
#include <us/Msg.hpp>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
#include <iostream>
|
||||
|
||||
struct LogEntry {
|
||||
std::string service;
|
||||
std::string message;
|
||||
};
|
||||
|
||||
int main() {
|
||||
pmc::net::AsyncQueue<LogEntry> queue(
|
||||
[](std::vector<LogEntry>&& 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<LogEntry>&&) {
|
||||
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 <us/RpcServer.hpp>
|
||||
#include <us/RpcClient.hpp>
|
||||
#include <thread>
|
||||
#include <iostream>
|
||||
|
||||
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<int>() + req.at("b").to_number<int>();
|
||||
});
|
||||
|
||||
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<T>` 的 `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
|
||||
Executable
+5
@@ -0,0 +1,5 @@
|
||||
#!/bin/bash
|
||||
cd "$(dirname "$0")" || exit
|
||||
make clear
|
||||
make
|
||||
make install
|
||||
@@ -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 <us/AsyncQueue.hpp>
|
||||
*
|
||||
* struct LogEntry { std::string service; std::string msg; };
|
||||
*
|
||||
* pmc::net::AsyncQueue<LogEntry> queue(
|
||||
* [](std::vector<LogEntry>&& batch) {
|
||||
* for (auto& e : batch) { send_to_server(e); }
|
||||
* },
|
||||
* {50, 2.0, 10000, 0.5},
|
||||
* [](const std::exception& e, std::vector<LogEntry>&&) {
|
||||
* std::cerr << "[queue] send failed: " << e.what() << std::endl;
|
||||
* }
|
||||
* );
|
||||
* queue.start();
|
||||
* queue.push(LogEntry{"gateway", "started"});
|
||||
* queue.shutdown();
|
||||
*/
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cstddef>
|
||||
#include <deque>
|
||||
#include <exception>
|
||||
#include <functional>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
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<T>
|
||||
// ============================================================================
|
||||
|
||||
/** 通用异步批量发送队列。
|
||||
*
|
||||
* @tparam T 队列元素类型，需支持移动构造（push 时会被 std::move 到 batch 中）。
|
||||
*/
|
||||
template <typename T>
|
||||
class AsyncQueue {
|
||||
public:
|
||||
/// 批量处理回调：接收一个元素批次的右值引用，处理完后 batch 被销毁。
|
||||
using Handler = std::function<void(std::vector<T>&& batch)>;
|
||||
|
||||
/// 错误回调：接收异常和失败的批次（可用于重试或记录）。
|
||||
using ErrorHandler = std::function<void(const std::exception& e, std::vector<T>&& 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<std::mutex> lock(_mtx);
|
||||
if (_running) return;
|
||||
_stop = false;
|
||||
_wakeup = false;
|
||||
_running = true;
|
||||
_thread = std::thread(&AsyncQueue::_run, this);
|
||||
}
|
||||
|
||||
/// 优雅关闭：flush 剩余数据后停止后台线程。幂等。
|
||||
void shutdown() {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_mtx);
|
||||
if (!_running) return;
|
||||
_stop = true;
|
||||
_wakeup = true;
|
||||
}
|
||||
_cv.notify_all();
|
||||
if (_thread.joinable()) {
|
||||
_thread.join();
|
||||
}
|
||||
// 线程结束后，flush 队列中剩余的所有数据
|
||||
_flush_all();
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_mtx);
|
||||
_running = false;
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 入队 --------------------------------------------------------------
|
||||
|
||||
/** 非阻塞入队一条数据。
|
||||
*
|
||||
* 若队列已关闭（shutdown 后），则丢弃数据并返回 false。
|
||||
* 若队列长度达到 max_queue_size，则丢弃最旧的数据并递增 dropped 计数。
|
||||
*
|
||||
* @return true 入队成功；false 队列已关闭。
|
||||
*/
|
||||
bool push(T item) {
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> lock(_mtx);
|
||||
return _queue.size();
|
||||
}
|
||||
|
||||
/// 因队列满而丢弃的数据总数。
|
||||
size_t dropped() const {
|
||||
return _dropped.load();
|
||||
}
|
||||
|
||||
/// 队列是否正在运行（已 start 且未 shutdown）。
|
||||
bool is_running() const {
|
||||
std::lock_guard<std::mutex> 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<std::mutex> lock(_mtx);
|
||||
_cv.wait_for(lock,
|
||||
std::chrono::duration<double>(_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<double>(now - last_flush).count()
|
||||
>= _config.flush_interval;
|
||||
|
||||
size_t qsize = 0;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<T> batch;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<T> batch;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<T> _queue;
|
||||
std::thread _thread;
|
||||
bool _running;
|
||||
bool _stop;
|
||||
bool _wakeup; ///< 唤醒标志（push/shutdown 置 true，后台线程消费后置 false）
|
||||
|
||||
std::atomic<size_t> _dropped;
|
||||
};
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -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 <us/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 <us/Msg_Basic.hpp>
|
||||
|
||||
#include <atomic>
|
||||
#include <cerrno>
|
||||
#include <chrono>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <fcntl.h>
|
||||
#include <netinet/in.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/select.h>
|
||||
#include <sys/types.h>
|
||||
#include <unistd.h>
|
||||
|
||||
// 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<char>((val >> 24) & 0xFFu);
|
||||
buf[1] = static_cast<char>((val >> 16) & 0xFFu);
|
||||
buf[2] = static_cast<char>((val >> 8) & 0xFFu);
|
||||
buf[3] = static_cast<char>(val & 0xFFu);
|
||||
}
|
||||
|
||||
inline uint32_t readUint32BE(const char *buf) noexcept {
|
||||
return (static_cast<uint32_t>(static_cast<unsigned char>(buf[0])) << 24) |
|
||||
(static_cast<uint32_t>(static_cast<unsigned char>(buf[1])) << 16) |
|
||||
(static_cast<uint32_t>(static_cast<unsigned char>(buf[2])) << 8) |
|
||||
(static_cast<uint32_t>(static_cast<unsigned char>(buf[3])));
|
||||
}
|
||||
|
||||
// ---- socket 超时设置 ------------------------------------------------------
|
||||
|
||||
inline void setSockTimeout(int sockfd, double seconds) {
|
||||
struct timeval tv {};
|
||||
tv.tv_sec = static_cast<time_t>(seconds);
|
||||
tv.tv_usec =
|
||||
static_cast<long>((seconds - static_cast<double>(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<size_t>(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<std::chrono::microseconds>(deadline - now);
|
||||
double remaining = static_cast<double>(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<time_t>(remaining);
|
||||
tv.tv_usec = static_cast<long>((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<size_t>(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::steady_clock::duration>(
|
||||
std::chrono::duration<double>(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<size_t>(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<size_t>(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<uint16_t>(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<struct sockaddr *>(&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<uint32_t>(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<size_t>(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<double>(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<double>(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<double>(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::steady_clock::duration>(
|
||||
std::chrono::duration<double>(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<double>(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<double>(retry_interval));
|
||||
}
|
||||
// 其他异常（MsgTimeoutException、普通MsgConnectionException等）不捕获，直接向上抛出
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 持续监听（长连接，阻塞回调模式）-----------------------------------
|
||||
|
||||
/// 在长连接上持续接收消息并通过回调处理。
|
||||
///
|
||||
/// 阻塞当前线程直到 stop() 被调用或连接断开。
|
||||
/// 必须先调用 connect()。
|
||||
///
|
||||
/// @param on_message 回调 ``void(const std::string &msg)``。
|
||||
void listen(std::function<void(const std::string &)> 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<int> _sockfd;
|
||||
std::atomic<bool> _running;
|
||||
};
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -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 <us/MsgServer.hpp>
|
||||
*
|
||||
* 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 <us/Msg_Basic.hpp>
|
||||
#include <us/Msg.hpp>
|
||||
|
||||
#include <algorithm>
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <condition_variable>
|
||||
#include <cstdio>
|
||||
#include <cstring>
|
||||
#include <deque>
|
||||
#include <functional>
|
||||
#include <memory>
|
||||
#include <mutex>
|
||||
#include <queue>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <unordered_map>
|
||||
#include <unordered_set>
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
#include <arpa/inet.h>
|
||||
#include <fcntl.h>
|
||||
#include <netinet/in.h>
|
||||
#include <netinet/tcp.h>
|
||||
#include <sys/socket.h>
|
||||
#include <sys/select.h>
|
||||
#include <sys/types.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#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<void(int sockfd, const std::string& addr, const std::string& message)>;
|
||||
|
||||
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<uint16_t>(_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<struct sockaddr*>(&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<std::thread> 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<std::mutex> send_lock;
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_clients_mtx);
|
||||
auto it = _send_locks.find(sockfd);
|
||||
if (it == _send_locks.end()) return; // 连接已断开
|
||||
send_lock = it->second;
|
||||
}
|
||||
std::lock_guard<std::mutex> 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<std::pair<int, std::shared_ptr<std::mutex>>> targets;
|
||||
{
|
||||
std::lock_guard<std::mutex> 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<std::mutex> slock(*send_lock);
|
||||
try {
|
||||
pmsg_send(fd, msg);
|
||||
} catch (const MsgException&) {
|
||||
// 发送失败，标记为 failed 由主循环清理
|
||||
std::lock_guard<std::mutex> flk(_failed_mtx);
|
||||
_failed.insert(fd);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 状态查询 ----------------------------------------------------------
|
||||
|
||||
/// 当前已连接客户端数量。
|
||||
size_t client_count() const {
|
||||
std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> flk(_failed_mtx);
|
||||
_failed.insert(task.fd);
|
||||
}
|
||||
// 清除 busy 标记，恢复 select 监听
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(_busy_mtx);
|
||||
_busy.erase(task.fd);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void _shutdown_pool(std::vector<std::thread>& 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<int> watched_clients;
|
||||
{
|
||||
std::lock_guard<std::mutex> clk(_clients_mtx);
|
||||
std::lock_guard<std::mutex> 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<long>(_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<std::mutex> 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<struct sockaddr*>(&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<std::mutex> lock(_clients_mtx);
|
||||
if (static_cast<int>(_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<std::mutex>();
|
||||
}
|
||||
|
||||
// 禁用 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<std::mutex> 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<std::mutex> blk(_busy_mtx);
|
||||
_busy.insert(fd);
|
||||
}
|
||||
{
|
||||
std::lock_guard<std::mutex> plk(_pool_mtx);
|
||||
if (static_cast<int>(_pool_queue.size()) >= _config.max_queue_size) {
|
||||
// 队列已满：拒绝并断开连接（防 OOM）
|
||||
std::lock_guard<std::mutex> 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<int> failed;
|
||||
{
|
||||
std::lock_guard<std::mutex> flk(_failed_mtx);
|
||||
failed.swap(_failed);
|
||||
}
|
||||
for (int fd : failed) {
|
||||
{
|
||||
std::lock_guard<std::mutex> blk(_busy_mtx);
|
||||
_busy.erase(fd);
|
||||
}
|
||||
_remove_client(fd);
|
||||
}
|
||||
}
|
||||
|
||||
void _remove_client(int fd) {
|
||||
std::lock_guard<std::mutex> clk(_clients_mtx);
|
||||
if (_clients.erase(fd) == 0) return; // 已被移除，防止 double-close
|
||||
_send_locks.erase(fd);
|
||||
::close(fd);
|
||||
}
|
||||
|
||||
void _close_all_clients() {
|
||||
std::lock_guard<std::mutex> clk(_clients_mtx);
|
||||
for (auto& [fd, addr] : _clients) {
|
||||
::close(fd);
|
||||
}
|
||||
_clients.clear();
|
||||
_send_locks.clear();
|
||||
std::lock_guard<std::mutex> blk(_busy_mtx);
|
||||
_busy.clear();
|
||||
std::lock_guard<std::mutex> flk(_failed_mtx);
|
||||
_failed.clear();
|
||||
}
|
||||
|
||||
// ---- 成员 --------------------------------------------------------------
|
||||
|
||||
std::string _addr;
|
||||
int _port;
|
||||
MsgServerConfig _config;
|
||||
int _server_fd;
|
||||
std::atomic<bool> _running;
|
||||
MessageCallback _on_message;
|
||||
|
||||
// 客户端管理
|
||||
mutable std::mutex _clients_mtx;
|
||||
std::unordered_map<int, std::string> _clients; ///< fd -> "ip:port"
|
||||
std::unordered_map<int, std::shared_ptr<std::mutex>> _send_locks; ///< fd -> 发送锁
|
||||
|
||||
// 同连接串行
|
||||
std::mutex _busy_mtx;
|
||||
std::unordered_set<int> _busy; ///< 回调执行中的 fd
|
||||
|
||||
// 失败连接（worker 线程标记，主循环清理）
|
||||
std::mutex _failed_mtx;
|
||||
std::unordered_set<int> _failed;
|
||||
|
||||
// 线程池
|
||||
std::atomic<bool> _pool_running{false};
|
||||
std::mutex _pool_mtx;
|
||||
std::condition_variable _pool_cv;
|
||||
std::queue<Task> _pool_queue;
|
||||
};
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -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 <cstddef>
|
||||
#include <cstdint>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
|
||||
#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
|
||||
@@ -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 <us/RpcClient.hpp>
|
||||
*
|
||||
* 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 <us/Msg.hpp>
|
||||
|
||||
#include <cstddef>
|
||||
#include <optional>
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
|
||||
#include <boost/json.hpp>
|
||||
|
||||
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<double> timeout = std::nullopt,
|
||||
std::optional<bool> retry_on_refused = std::nullopt,
|
||||
std::optional<double> 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<double> timeout = std::nullopt,
|
||||
std::optional<bool> retry_on_refused = std::nullopt,
|
||||
std::optional<double> 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
|
||||
@@ -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 <us/RpcServer.hpp>
|
||||
*
|
||||
* 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 <us/MsgServer.hpp>
|
||||
|
||||
#include <boost/system/error_code.hpp>
|
||||
|
||||
#include <cctype>
|
||||
#include <cstddef>
|
||||
#include <functional>
|
||||
#include <regex>
|
||||
#include <string>
|
||||
#include <unordered_map>
|
||||
|
||||
#include <boost/json.hpp>
|
||||
|
||||
namespace pmc {
|
||||
namespace net {
|
||||
|
||||
namespace json = boost::json;
|
||||
|
||||
// ============================================================================
|
||||
// RpcServer
|
||||
// ============================================================================
|
||||
|
||||
/// JSON RPC 服务端：基于 MsgServer，自带路由分发与响应包装。
|
||||
class RpcServer {
|
||||
public:
|
||||
/// 路由处理函数签名：接收请求参数 object，返回 data 值（任意 JSON 类型）。
|
||||
using Handler = std::function<json::value(const json::object& req)>;
|
||||
|
||||
/// 错误回调：void(route_name, exception)，handler 抛异常时调用（用于记日志）。
|
||||
using ErrorCallback = std::function<void(const std::string& route, const std::exception& e)>;
|
||||
|
||||
/** 构造 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/...），替换为 <redacted>
|
||||
// 不返回函数堆栈、SQL、文件路径等内部细节给客户端；完整异常由 on_error 钩子记录到服务端日志。
|
||||
|
||||
static std::string _sanitize_error(const std::string& msg) {
|
||||
if (msg.empty()) return msg;
|
||||
try {
|
||||
// 匹配以 / 开头的绝对路径段，替换为 <redacted>
|
||||
// 不返回函数堆栈、SQL、文件路径等内部细节给客户端
|
||||
static const std::regex path_re(R"(/[A-Za-z0-9_.\-]+(?:/[A-Za-z0-9_.\-]+)*)");
|
||||
return std::regex_replace(msg, path_re, "<redacted>");
|
||||
} catch (...) {
|
||||
return msg; // 正则异常时返回原文
|
||||
}
|
||||
}
|
||||
|
||||
// ---- 成员 --------------------------------------------------------------
|
||||
|
||||
MsgServer _server;
|
||||
std::unordered_map<std::string, Handler> _routes;
|
||||
ErrorCallback _on_error;
|
||||
};
|
||||
|
||||
} // namespace net
|
||||
} // namespace pmc
|
||||
@@ -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
|
||||
Reference in new issue
Block a user