From 26de527a09079250bdbace4dc8d6ca9e853bee5f Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Sat, 15 Aug 2026 09:26:56 +0800 Subject: [PATCH] =?UTF-8?q?feat(memdb=C3=97libmemdb):=20memdb=20=E6=9C=8D?= =?UTF-8?q?=E5=8A=A1=E5=8C=96=E8=90=BD=E5=9C=B0=E2=80=94=E2=80=94RPC=20?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E7=AB=AF=20+=20=E6=9E=84=E5=BB=BA/=E6=B5=8B?= =?UTF-8?q?=E8=AF=95/=E9=83=A8=E7=BD=B2=E9=97=AD=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - libmemdb: 补全 flush() 惰性 TTL 驱逐；修复 explicit 误用/头文件缺失/ 时间戳碰撞误删；size()/get_db() 前 flush 消除脏读；三处双重哈希查找 合并；index_keys 去冗余字段；update(refresh_ttl) 改原地重索引， 免 remove+insert 的拷贝与重复 flush - memdb: 重写 RPC 服务端（libmsg RpcServer, :8011, 9 路由, mutex 保护）； 补全 Makefile/_set.sh/_run.sh 与 --instance/--log-level CLI； SIGINT/SIGTERM 优雅退出；注册 services.yaml - 测试: libmemdb 单元测试 22 例 + memdb RPC 集成测试 12 场景（含并发） - 清理: 删除误合入的 db2/create_sqlite3.sql 旧版（scripts/ 下为权威版） --- core/services.yaml | 19 ++ db2/create_sqlite3.sql | 39 ----- libmemdb/inc/us/memdb.hpp | 81 ++++++--- memdb/Makefile | 49 ++++++ memdb/README.md | 84 ++++++++- memdb/_run.sh | 4 + memdb/_set.sh | 4 + memdb/src/memdb.cpp | 223 ++++++++++++++++++------ memdb/test_libmemdb.cpp | 354 ++++++++++++++++++++++++++++++++++++++ memdb/test_memdb.py | 275 +++++++++++++++++++++++++++++ 10 files changed, 1016 insertions(+), 116 deletions(-) delete mode 100644 db2/create_sqlite3.sql create mode 100644 memdb/test_libmemdb.cpp create mode 100644 memdb/test_memdb.py diff --git a/core/services.yaml b/core/services.yaml index ab949d0..fa79eb3 100644 --- a/core/services.yaml +++ b/core/services.yaml @@ -36,6 +36,25 @@ services: depends_on: [] sleep_before: 0 + # --- 内存缓存 --- + - name: memdb + cmd: bash /opt/micro_servers/memdb/_run.sh + args: + - "--instance" + - "memdb" + - "--log-level" + - "info" + - "--host" + - "127.0.0.1" + - "--port" + - "8011" + - "--workers" + - "4" + - "--max-time" + - "600" + depends_on: [] + sleep_before: 0 + # --- AI 审核 --- - name: ai_review cmd: bash /opt/micro_servers/ai_review/_run.sh diff --git a/db2/create_sqlite3.sql b/db2/create_sqlite3.sql deleted file mode 100644 index e43a1d3..0000000 --- a/db2/create_sqlite3.sql +++ /dev/null @@ -1,39 +0,0 @@ --- Users 表创建脚本 --- 数据库类型: SQLite - -CREATE TABLE IF NOT EXISTS Users ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - username TEXT UNIQUE NOT NULL, - password_hash TEXT NOT NULL, - create_at DATETIME DEFAULT CURRENT_TIMESTAMP, - last_login DATETIME, - is_active BOOLEAN DEFAULT 1 -); - -CREATE TABLE IF NOT EXISTS Notes ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - pid INTEGER NOT NULL, - owner VARCHAR(128) DEFAULT 'Unknow', - title VARCHAR(128) NOT NULL, - content TEXT NOT NULL -); - --- 旧版增量 --- ALTER TABLE Notes ADD COLUMN owner VARCHAR(128) DEFAULT 'Unknow'; - --- ── 日志表（db 内置日志模块）───────────────────────────────────────── - -CREATE TABLE IF NOT EXISTS Logs ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - ts TEXT NOT NULL, - service TEXT NOT NULL, - level TEXT NOT NULL, - message TEXT NOT NULL, - trace_id TEXT, - extra TEXT -); - -CREATE INDEX IF NOT EXISTS idx_logs_service_ts ON Logs(service, ts); -CREATE INDEX IF NOT EXISTS idx_logs_level_ts ON Logs(level, ts); -CREATE INDEX IF NOT EXISTS idx_logs_ts ON Logs(ts); -CREATE INDEX IF NOT EXISTS idx_logs_trace ON Logs(trace_id); diff --git a/libmemdb/inc/us/memdb.hpp b/libmemdb/inc/us/memdb.hpp index ec3873a..9fcf62e 100644 --- a/libmemdb/inc/us/memdb.hpp +++ b/libmemdb/inc/us/memdb.hpp @@ -4,7 +4,7 @@ #include #include #include -#include +#include @@ -20,7 +20,6 @@ struct memdb_column { }; struct index_keys { - T key; std::list ::iterator item; }; @@ -39,7 +38,27 @@ private: std::multiset time_touch; size_t __size = 0; int MAX_TIME = 60 * 10; // 10分钟 - void flush(); + void flush() { + if (this -> time_touch.empty()) { + return; + } + std::chrono::system_clock::time_point now_time = std::chrono::system_clock::now(); + auto it = this -> time_touch.begin() -> item; + + while (now_time >= it -> timestamp) { + this -> __size--; + key_t to_be_del_k = it -> key; + this -> db.erase(it); + this -> keys.erase(to_be_del_k); + this -> time_touch.erase(this -> time_touch.begin()); + + if (this -> time_touch.empty()) { + return; + } + it = this -> time_touch.begin() -> item; + now_time = std::chrono::system_clock::now(); + } + } @@ -53,11 +72,12 @@ public: explicit Memdb() = default; explicit Memdb(int max_time) : MAX_TIME(std::max(0, max_time)) {} - explicit Memdb& operator=(const Memdb& to_be_cp_memdb) = default; - explicit ~Memdb() = default; + Memdb& operator=(const Memdb& to_be_cp_memdb) = default; + ~Memdb() = default; - size_t size() const { + size_t size() { + this -> flush(); return this -> __size; } @@ -65,7 +85,8 @@ public: return this -> keys; } - const std::list & get_db() const { + const std::list& get_db() { + this -> flush(); return this -> db; } @@ -89,9 +110,8 @@ public: this -> db.push_back(new_column); Memdb::index_keys new_key; - new_key.key = to_be_ins_k; new_key.item = std::prev(this -> db.end()); - this -> keys[to_be_ins_k] = new_key; + this -> keys.emplace(to_be_ins_k, new_key); Memdb::index_timestamp new_time_touch; new_time_touch.item = std::prev(this -> db.end()); @@ -104,13 +124,20 @@ public: bool remove(key_t to_be_del_k) { this -> flush(); - if (this -> keys.find(to_be_del_k) == this -> keys.end()) { + auto kit = this -> keys.find(to_be_del_k); + if (kit == this -> keys.end()) { return this -> fail; } - typename std::list ::iterator it = this -> keys[to_be_del_k].item; - this -> keys.erase(it -> key); - this -> time_touch.erase({it}); + typename std::list ::iterator it = kit -> second.item; + this -> keys.erase(kit); + auto range = this -> time_touch.equal_range({it}); + for (auto sit = range.first; sit != range.second; ++sit) { + if (sit -> item == it) { + this -> time_touch.erase(sit); + break; + } + } this -> db.erase(it); this -> __size--; @@ -119,18 +146,26 @@ public: bool update(key_t to_be_udt_k, value_t udt_v, bool need_to_udt_timetouch=false) { this -> flush(); - if (this -> keys.find(to_be_udt_k) == this -> keys.end()) { + auto kit = this -> keys.find(to_be_udt_k); + if (kit == this -> keys.end()) { return this -> fail; } - typename std::list ::iterator it = this -> keys[to_be_udt_k].item; + typename std::list ::iterator it = kit -> second.item; it -> value = udt_v; if (need_to_udt_timetouch == true) { - std::chrono::system_clock::time_point now_time = std::chrono::system_clock::now(); - T copy_k = it -> key; - U copy_v = it -> value; - this -> remove(copy_k); - this -> insert(copy_k, copy_v); + auto range = this -> time_touch.equal_range({it}); + for (auto sit = range.first; sit != range.second; ++sit) { + if (sit -> item == it) { + this -> time_touch.erase(sit); + break; + } + } + it -> timestamp = std::chrono::system_clock::now() + + std::chrono::seconds(MAX_TIME); + Memdb::index_timestamp new_time_touch; + new_time_touch.item = it; + this -> time_touch.insert(new_time_touch); } return this -> succ; @@ -138,11 +173,11 @@ public: std::optional search(key_t to_be_sch_k) { this -> flush(); - if (this -> keys.find(to_be_sch_k) == this -> keys.end()) { + auto kit = this -> keys.find(to_be_sch_k); + if (kit == this -> keys.end()) { return std::nullopt; } - typename std::list ::iterator it = this -> keys[to_be_sch_k].item; - return it -> value; + return kit -> second.item -> value; } }; \ No newline at end of file diff --git a/memdb/Makefile b/memdb/Makefile index e69de29..7d483a5 100644 --- a/memdb/Makefile +++ b/memdb/Makefile @@ -0,0 +1,49 @@ +CPP = g++ +CPPFLAGS = -std=c++20 -O3 -Wall -Werror -pipe +INCS = -I ../libmsg/inc -I ../libmemdb/inc +LIBS = -L ../libmsg/build -lus_msg -lboost_json -lboost_program_options -lpthread +RPATH = -Wl,-rpath,'$$ORIGIN/../../libmsg/build' + +BUILD = build +TARGET = $(BUILD)/memdb +SRC = src/memdb.cpp + +all: $(TARGET) + +$(BUILD): + mkdir -p $(BUILD) + +$(TARGET): $(SRC) | $(BUILD) + $(CPP) $(CPPFLAGS) $(INCS) -o $@ $< $(LIBS) $(RPATH) + +# ── libmemdb 单元测试 ──────────────────────────────────────────────────── + +TEST_LIBMEMDB = test_libmemdb +TEST_CPPFLAGS = -std=c++20 -O0 -g -Wall -Werror + +$(TEST_LIBMEMDB): test_libmemdb.cpp + $(CPP) $(TEST_CPPFLAGS) -I ../libmemdb/inc -o $@ $< + +test: $(TARGET) + @echo "=== memdb unit tests ===" + $(MAKE) $(TEST_LIBMEMDB) + ./$(TEST_LIBMEMDB) + @echo "" + @echo "=== memdb RPC integration tests (requires running server) ===" + -python3 test_memdb.py $(HOST) $(PORT) + +test-libmemdb: $(TEST_LIBMEMDB) + ./$(TEST_LIBMEMDB) + +test-memdb: $(TARGET) + @echo "Starting memdb server..." + ./$(TARGET) --host $(HOST) --port $(PORT) --workers 2 --max-time 10 & + @sleep 1 + -python3 test_memdb.py $(HOST) $(PORT); EXIT=$$? + @kill %1 2>/dev/null; wait %1 2>/dev/null + @exit $$EXIT + +clean: + rm -f $(TARGET) $(TEST_LIBMEMDB) + +.PHONY: all clean test test-libmemdb test-memdb \ No newline at end of file diff --git a/memdb/README.md b/memdb/README.md index 19b6b65..f9b0032 100644 --- a/memdb/README.md +++ b/memdb/README.md @@ -1,5 +1,83 @@ -# memdb — 内存数据库模板（实验） +# memdb — 内存缓存数据库 -基于 C++ 模板的进程内内存数据库（`include/memdb.hpp`），支持键值列、时间戳、索引与惰性删除。 +基于 `libmemdb` 模板库的 JSON RPC 缓存服务，提供键值存储、TTL 过期、并发安全的内存缓存能力。 -> **状态：实验性，未完成**。仅含模板头与空 `main`，暂无实际接入方，仅供后续 DB 层设计参考。 +## 架构 + +``` +Memdb (libmemdb 模板：LRU 链表 + 哈希索引 + TTL 过期) + └── MemdbServer (RPC 服务端：路由注册 + 互斥锁线程安全) + └── pmc::net::RpcServer (libmsg JSON RPC 服务端：多客户端 select + 线程池) +``` + +## 路由 + +| 路由 | 参数 | 返回 | 说明 | +|------|------|------|------| +| `ping` | — | `"pong"` | 健康检查 | +| `set` | `key`, `value` | `true/false` | 插入键值对（key 已存在返回 false） | +| `get` | `key` | `"value"` / `null` | 查询键值 | +| `del` | `key` | `true/false` | 删除键 | +| `update` | `key`, `value`, `refresh_ttl?` | `true/false` | 更新值，可选刷新 TTL | +| `exists` | `key` | `true/false` | 键是否存在 | +| `size` | — | `int` | 当前条目数 | +| `expire` | `key`, `value` | `true/false` | 更新值并重置 TTL（del + set） | +| `show` | — | `[{key,value},...]` | 导出所有条目 | + +## 配置 + +`ServerConfig` 结构体（`src/memdb.cpp`）： + +| 字段 | 默认值 | 说明 | +|------|--------|------| +| `listen_host` | `127.0.0.1` | 监听地址 | +| `listen_port` | `8011` | 监听端口 | +| `workers` | `4` | 线程池大小 | +| `max_time` | `600` | 默认 TTL（秒），10 分钟 | + +## 构建与运行 + +**前置依赖**：g++（C++20）、libmsg、libmemdb、boost::json + +```bash +# 构建 +bash _set.sh + +# 运行 +bash _run.sh +``` + +## 客户端示例 + +```cpp +#include + +pmc::net::RpcClient rpc("127.0.0.1", 8011, "memdb-client"); + +// 插入 +json::object params; +params["key"] = "session:abc123"; +params["value"] = R"({"user":"admin","role":"root"})"; +rpc.call("set", params); // → true + +// 查询 +rpc.call("get", {{"key", "session:abc123"}}); // → "{\"user\":\"admin\",...}" + +// 检查存在 +rpc.call("exists", {{"key", "session:abc123"}}); // → true + +// 查看大小 +rpc.call("size"); // → 1 +``` + +## 项目结构 + +``` +memdb/ +├── src/ +│ └── memdb.cpp # MemdbServer + main() +├── Makefile +├── _set.sh +├── _run.sh +└── README.md +``` \ No newline at end of file diff --git a/memdb/_run.sh b/memdb/_run.sh index e69de29..7ff9670 100644 --- a/memdb/_run.sh +++ b/memdb/_run.sh @@ -0,0 +1,4 @@ +#!/bin/bash +set -e +cd "$(dirname "$0")" +exec ./build/memdb --instance "${INSTANCE:-memdb}" --log-level "${LOG_LEVEL:-info}" "$@" \ No newline at end of file diff --git a/memdb/_set.sh b/memdb/_set.sh index e69de29..54382e1 100644 --- a/memdb/_set.sh +++ b/memdb/_set.sh @@ -0,0 +1,4 @@ +#!/bin/bash +set -e +cd "$(dirname "$0")" +make \ No newline at end of file diff --git a/memdb/src/memdb.cpp b/memdb/src/memdb.cpp index 55720a8..3641cc9 100644 --- a/memdb/src/memdb.cpp +++ b/memdb/src/memdb.cpp @@ -1,78 +1,199 @@ #include -#include - -#include - #include -#include +#include + +#include +#include +#include #include +#include +#include +#include +#include namespace json = boost::json; - - - -struct ClientConfig { - std::string connected_host = "127.0.0.1"; - int connected_port = 8005; - std::string service_name = "db2"; - double timeout = 10.0; - bool retry_on_refused = true; -}; +namespace po = boost::program_options; struct ServerConfig { - std::string listen_host = "127.0.0.1"; - int listen_port = 8011; - int workers = 4; - std::string db_url = "sqlite:///database.db"; + std::string listen_host = "127.0.0.1"; + int listen_port = 8011; + int workers = 4; + int max_time = 600; // 默认 TTL 10 分钟 }; -static std::string get_db_path(const std::string url) { - const std::string prefix = "sqlite:///"; - if (url.rfind(prefix, 0) == 0) { - return url.substr(prefix.size()); - } - return url; +static std::atomic g_running{true}; + +static void signal_handler(int) { + g_running = false; } - - -class DBConnection { +class MemdbServer { private: - ClientConfig __cfg; + ServerConfig __cfg; + Memdb __db; + pmc::net::RpcServer __rpc; + std::mutex __mtx; + static const std::string& _require_string(const json::object& req, const std::string& field) { + auto it = req.find(field); + if (it == req.end()) { + throw std::runtime_error("missing required field: " + field); + } + if (!it->value().is_string()) { + throw std::runtime_error("field '" + field + "' must be a string"); + } + return it->value().as_string(); + } public: - explicit DBConnection(const ClientConfig& cfg) : - __cfg(cfg) {} - ~DBConnection() = default; + explicit MemdbServer(const ServerConfig& cfg) + : __cfg(cfg) + , __db(cfg.max_time) + , __rpc(cfg.listen_host, cfg.listen_port, + {.workers = cfg.workers}) + { + _register_routes(); + } - std::string get_from_sql(const std::string& sql_name) { - pmc::net::RpcClient client(__cfg.connected_host, __cfg.connected_port, __cfg.service_name, __cfg.timeout, __cfg.retry_on_refused); + void start() { __rpc.start(); } + void stop() { __rpc.stop(); } - try { - auto data = client.call("execute", {{"action", "select"}, {"table", "memdb"}, {"where", {"sql_name", sql_name}}}); - - } catch (const std::exception& err) { - std::cerr << "[memdb] Error: 连接失败: " << err.what() << std::endl; - std::exit(1); - } +private: + void _register_routes() { + // ---- 健康检查 ---- + __rpc.add_route("ping", [](const json::object&) -> json::value { + return "pong"; + }); + + // ---- set: 插入键值对 ---- + __rpc.add_route("set", [this](const json::object& req) -> json::value { + const auto& key = _require_string(req, "key"); + const auto& val = _require_string(req, "value"); + std::lock_guard lock(__mtx); + return __db.insert(key, val); + }); + + // ---- get: 查询键 ---- + __rpc.add_route("get", [this](const json::object& req) -> json::value { + const auto& key = _require_string(req, "key"); + std::lock_guard lock(__mtx); + auto result = __db.search(key); + if (result.has_value()) { + return json::string(*result); + } + return nullptr; + }); + + // ---- del: 删除键 ---- + __rpc.add_route("del", [this](const json::object& req) -> json::value { + const auto& key = _require_string(req, "key"); + std::lock_guard lock(__mtx); + return __db.remove(key); + }); + + // ---- update: 更新键值（可选刷新 TTL）---- + __rpc.add_route("update", [this](const json::object& req) -> json::value { + const auto& key = _require_string(req, "key"); + const auto& val = _require_string(req, "value"); + bool refresh_ttl = false; + if (auto it = req.find("refresh_ttl"); it != req.end() && it->value().is_bool()) { + refresh_ttl = it->value().as_bool(); + } + std::lock_guard lock(__mtx); + return __db.update(key, val, refresh_ttl); + }); + + // ---- exists: 检查键是否存在 ---- + __rpc.add_route("exists", [this](const json::object& req) -> json::value { + const auto& key = _require_string(req, "key"); + std::lock_guard lock(__mtx); + return __db.search(key).has_value(); + }); + + // ---- size: 当前条目数 ---- + __rpc.add_route("size", [this](const json::object&) -> json::value { + std::lock_guard lock(__mtx); + return static_cast(__db.size()); + }); + + // ---- show: 导出所有条目 ---- + __rpc.add_route("show", [this](const json::object&) -> json::value { + std::lock_guard lock(__mtx); + json::array result; + for (const auto& col : __db.get_db()) { + json::object entry; + entry["key"] = json::string(col.key); + entry["value"] = json::string(col.value); + result.push_back(std::move(entry)); + } + return result; + }); + + // ---- expire: 更新值并重置 TTL（等价于 del + set）---- + __rpc.add_route("expire", [this](const json::object& req) -> json::value { + const auto& key = _require_string(req, "key"); + const auto& val = _require_string(req, "value"); + std::lock_guard lock(__mtx); + __db.remove(key); + return __db.insert(key, val); + }); } }; -class MemdbServer { +int main(int argc, char** argv) { + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); + ServerConfig cfg; -private: - ServerConfig __cfg; + po::options_description desc("memdb options"); + desc.add_options() + ("help,h", "display help") + ("instance", po::value()->default_value("memdb"), "instance name") + ("log-level", po::value()->default_value("info"), "log level") + ("host", po::value()->default_value("127.0.0.1"), "listen host") + ("port", po::value()->default_value(8011), "listen port") + ("workers", po::value()->default_value(4), "worker threads") + ("max-time", po::value()->default_value(600), "default TTL (seconds)"); + po::variables_map vm; + po::store(po::parse_command_line(argc, argv, desc), vm); + po::notify(vm); -public: - explicit MemdbServer(const ServerConfig& cfg) : - __cfg(cfg) {} - - void start() { - //TODO 编写 + if (vm.count("help")) { + std::cout << desc << std::endl; + return 0; } -}; \ No newline at end of file + + cfg.listen_host = vm["host"].as(); + cfg.listen_port = vm["port"].as(); + cfg.workers = vm["workers"].as(); + cfg.max_time = vm["max-time"].as(); + + auto instance = vm["instance"].as(); + auto log_level = vm["log-level"].as(); + + MemdbServer server(cfg); + + std::cout << "[memdb:" << instance << "] log-level=" << log_level + << " listening on " << cfg.listen_host + << ":" << cfg.listen_port << std::endl; + + std::thread server_thread([&server]() { + server.start(); + }); + + while (g_running) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + + std::cout << "\n[memdb:" << instance << "] shutting down..." << std::endl; + server.stop(); + if (server_thread.joinable()) { + server_thread.join(); + } + + return 0; +} \ No newline at end of file diff --git a/memdb/test_libmemdb.cpp b/memdb/test_libmemdb.cpp new file mode 100644 index 0000000..dec6270 --- /dev/null +++ b/memdb/test_libmemdb.cpp @@ -0,0 +1,354 @@ +/* + * test_libmemdb.cpp — libmemdb 模板类单元测试 + * + * 编译: g++ -std=c++20 -O0 -g -Wall -Werror -I ../libmemdb/inc -o test_libmemdb test_libmemdb.cpp + * 运行: ./test_libmemdb + */ + +#include + +#include +#include +#include +#include +#include +#include + +static int g_passed = 0; +static int g_failed = 0; + +#define TEST(name) \ + do { \ + std::cout << " " << (name) << " ... "; \ + } while (0) + +#define PASS() \ + do { \ + std::cout << "PASS" << std::endl; \ + ++g_passed; \ + } while (0) + +#define FAIL(msg) \ + do { \ + std::cout << "FAIL: " << (msg) << std::endl; \ + ++g_failed; \ + } while (0) + +#define CHECK(cond, msg) \ + do { \ + if (!(cond)) { FAIL(msg); return; } \ + } while (0) + +// ── insert ────────────────────────────────────────────────────────────── + +static void test_insert_basic() { + TEST("insert basic"); + Memdb db; + CHECK(db.insert("a", "1"), "insert new key should succeed"); + CHECK(db.size() == 1, "size should be 1"); + CHECK(db.search("a").has_value(), "search should find key"); + CHECK(db.search("a").value() == "1", "value should match"); + PASS(); +} + +static void test_insert_duplicate() { + TEST("insert duplicate"); + Memdb db; + db.insert("a", "1"); + CHECK(!db.insert("a", "2"), "insert duplicate should fail"); + CHECK(db.search("a").value() == "1", "value should be unchanged"); + CHECK(db.size() == 1, "size should still be 1"); + PASS(); +} + +static void test_insert_int_keys() { + TEST("insert int keys"); + Memdb db; + CHECK(db.insert(1, "one"), ""); + CHECK(db.insert(2, "two"), ""); + CHECK(db.search(1).value() == "one", ""); + CHECK(db.search(2).value() == "two", ""); + CHECK(db.size() == 2, ""); + PASS(); +} + +// ── remove ────────────────────────────────────────────────────────────── + +static void test_remove_existing() { + TEST("remove existing"); + Memdb db; + db.insert("a", "1"); + db.insert("b", "2"); + CHECK(db.remove("a"), "remove existing should succeed"); + CHECK(!db.search("a").has_value(), "removed key should not be found"); + CHECK(db.size() == 1, "size should be 1"); + CHECK(db.search("b").value() == "2", "other key should still exist"); + PASS(); +} + +static void test_remove_nonexistent() { + TEST("remove nonexistent"); + Memdb db; + CHECK(!db.remove("ghost"), "remove nonexistent should fail"); + CHECK(db.size() == 0, "size should still be 0"); + PASS(); +} + +static void test_remove_all() { + TEST("remove all"); + Memdb db; + db.insert("a", "1"); + db.insert("b", "2"); + db.insert("c", "3"); + db.remove("a"); + db.remove("b"); + db.remove("c"); + CHECK(db.size() == 0, "size should be 0"); + CHECK(!db.search("a").has_value(), ""); + CHECK(!db.search("b").has_value(), ""); + CHECK(!db.search("c").has_value(), ""); + PASS(); +} + +// ── update ────────────────────────────────────────────────────────────── + +static void test_update_without_refresh() { + TEST("update without refresh TTL"); + Memdb db(1); // 1s TTL + db.insert("a", "1"); + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + CHECK(db.update("a", "2", false), "update should succeed"); + CHECK(db.search("a").value() == "2", "value should be updated"); + // 500ms left — should still exist + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + CHECK(db.search("a").has_value(), "should still exist (800ms < 1s)"); + PASS(); +} + +static void test_update_with_refresh() { + TEST("update with refresh TTL"); + Memdb db(1); // 1s TTL + db.insert("a", "1"); + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + CHECK(db.update("a", "2", true), "update should succeed"); + CHECK(db.search("a").value() == "2", "value should be updated"); + // TTL was refreshed — should still exist + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + CHECK(db.search("a").has_value(), "should still exist after refresh"); + PASS(); +} + +static void test_update_nonexistent() { + TEST("update nonexistent"); + Memdb db; + CHECK(!db.update("ghost", "1"), "update nonexistent should fail"); + PASS(); +} + +// ── search ────────────────────────────────────────────────────────────── + +static void test_search_nonexistent() { + TEST("search nonexistent"); + Memdb db; + CHECK(!db.search("ghost").has_value(), "search nonexistent should return nullopt"); + PASS(); +} + +// ── size ──────────────────────────────────────────────────────────────── + +static void test_size_empty() { + TEST("size empty"); + Memdb db; + CHECK(db.size() == 0, "empty db should have size 0"); + PASS(); +} + +static void test_size_after_ops() { + TEST("size after operations"); + Memdb db; + db.insert("a", "1"); + db.insert("b", "2"); + CHECK(db.size() == 2, "size after 2 inserts"); + db.remove("a"); + CHECK(db.size() == 1, "size after remove"); + db.insert("c", "3"); + CHECK(db.size() == 2, "size after re-insert"); + PASS(); +} + +// ── TTL expiration (flush) ────────────────────────────────────────────── + +static void test_ttl_expiration() { + TEST("TTL expiration"); + Memdb db(1); // 1s TTL + db.insert("a", "1"); + db.insert("b", "2"); + CHECK(db.size() == 2, "initial size 2"); + + std::this_thread::sleep_for(std::chrono::milliseconds(1100)); + + // search triggers flush + CHECK(!db.search("a").has_value(), "a should be expired"); + CHECK(!db.search("b").has_value(), "b should be expired"); + CHECK(db.size() == 0, "size should be 0 after flush"); + PASS(); +} + +static void test_ttl_partial_expiration() { + TEST("TTL partial expiration"); + Memdb db(1); + db.insert("a", "1"); + std::this_thread::sleep_for(std::chrono::milliseconds(600)); + db.insert("b", "2"); // fresher + std::this_thread::sleep_for(std::chrono::milliseconds(500)); + // a: 1100ms old → expired, b: 500ms old → alive + CHECK(!db.search("a").has_value(), "a should be expired"); + CHECK(db.search("b").has_value(), "b should still be alive"); + CHECK(db.size() == 1, "size should be 1"); + PASS(); +} + +static void test_ttl_zero_means_immediate() { + TEST("TTL zero means immediate expiration"); + Memdb db(0); + db.insert("a", "1"); + CHECK(!db.search("a").has_value(), "zero TTL should expire immediately"); + CHECK(db.size() == 0, "size should be 0"); + PASS(); +} + +// ── get_db / show ─────────────────────────────────────────────────────── + +static void test_get_db() { + TEST("get_db returns all entries"); + Memdb db(10); + db.insert("x", "100"); + db.insert("y", "200"); + const auto& lst = db.get_db(); + CHECK(lst.size() == 2, "list should have 2 entries"); + // verify values are accessible + bool found_x = false, found_y = false; + for (const auto& col : lst) { + if (col.key == "x" && col.value == "100") found_x = true; + if (col.key == "y" && col.value == "200") found_y = true; + } + CHECK(found_x && found_y, "both entries should be in the list"); + PASS(); +} + +// ── get_keys ──────────────────────────────────────────────────────────── + +static void test_get_keys() { + TEST("get_keys returns key index"); + Memdb db; + db.insert("k1", "v1"); + const auto& keys = db.get_keys(); + CHECK(keys.size() == 1, "keys should have 1 entry"); + CHECK(keys.find("k1") != keys.end(), "k1 should be in keys"); + PASS(); +} + +// ── get_time_touch ────────────────────────────────────────────────────── + +static void test_get_time_touch() { + TEST("get_time_touch returns sorted timestamps"); + Memdb db(10); + db.insert("a", "1"); + db.insert("b", "2"); + const auto& tt = db.get_time_touch(); + CHECK(tt.size() == 2, "time_touch should have 2 entries"); + // verify sorted order (a before b since a was inserted first) + auto it = tt.begin(); + auto first = it; + ++it; + CHECK(first->item->timestamp <= it->item->timestamp, + "time_touch should be sorted by timestamp"); + PASS(); +} + +// ── copy assignment ───────────────────────────────────────────────────── + +static void test_copy_assignment() { + TEST("copy assignment"); + Memdb db1; + db1.insert("a", "1"); + Memdb db2; + db2 = db1; + CHECK(db2.search("a").value() == "1", "copied db should have same data"); + // verify independence + db2.insert("b", "2"); + CHECK(db1.size() == 1, "original should be unchanged"); + CHECK(db2.size() == 2, "copy should be independent"); + PASS(); +} + +// ── edge cases ────────────────────────────────────────────────────────── + +static void test_large_number_of_keys() { + TEST("large number of keys"); + Memdb db(60); + const int N = 1000; + for (int i = 0; i < N; ++i) { + CHECK(db.insert(i, i * 10), "all inserts should succeed"); + } + CHECK(db.size() == static_cast(N), "size should be N"); + for (int i = 0; i < N; ++i) { + CHECK(db.search(i).value() == i * 10, "all values should match"); + } + PASS(); +} + +static void test_reinsert_after_remove() { + TEST("reinsert after remove"); + Memdb db; + db.insert("a", "1"); + db.remove("a"); + CHECK(db.insert("a", "2"), "reinsert should succeed"); + CHECK(db.search("a").value() == "2", "value should be updated"); + PASS(); +} + +static void test_update_without_refresh_keeps_old_ttl() { + TEST("update without refresh keeps old TTL"); + Memdb db(1); + db.insert("a", "1"); + std::this_thread::sleep_for(std::chrono::milliseconds(800)); + db.update("a", "2", false); // value changes, TTL unchanged + // wait remaining 200ms+ — should expire + std::this_thread::sleep_for(std::chrono::milliseconds(300)); + CHECK(!db.search("a").has_value(), "should expire at original TTL"); + PASS(); +} + +// ──────────────────────────────────────────────────────────────────────── + +int main() { + std::cout << "=== libmemdb unit tests ===" << std::endl; + + test_insert_basic(); + test_insert_duplicate(); + test_insert_int_keys(); + test_remove_existing(); + test_remove_nonexistent(); + test_remove_all(); + test_update_without_refresh(); + test_update_with_refresh(); + test_update_nonexistent(); + test_search_nonexistent(); + test_size_empty(); + test_size_after_ops(); + test_ttl_expiration(); + test_ttl_partial_expiration(); + test_ttl_zero_means_immediate(); + test_get_db(); + test_get_keys(); + test_get_time_touch(); + test_copy_assignment(); + test_large_number_of_keys(); + test_reinsert_after_remove(); + test_update_without_refresh_keeps_old_ttl(); + + std::cout << "\n=== Results: " << g_passed << " passed, " + << g_failed << " failed ===" << std::endl; + return g_failed > 0 ? 1 : 0; +} \ No newline at end of file diff --git a/memdb/test_memdb.py b/memdb/test_memdb.py new file mode 100644 index 0000000..5ff127c --- /dev/null +++ b/memdb/test_memdb.py @@ -0,0 +1,275 @@ +""" +test_memdb.py — memdb RPC 服务端集成测试（pymsg 协议） + +基于 pymsg TCP 协议，直接与 memdb:8011 通信。 +测试所有 9 个 RPC 路由 + 并发安全性。 + +用法: python test_memdb.py [--host 127.0.0.1] [--port 8011] +""" + +import json +import socket +import struct +import sys +import threading +import time + + +class MemdbClient: + """memdb pymsg 协议客户端（短连接模式）。""" + + def __init__(self, host="127.0.0.1", port=8011, timeout=5.0): + self.host = host + self.port = port + self.timeout = timeout + + def call(self, route, **params): + """发送 JSON-RPC 请求，返回 (success, data)。""" + payload = {"_route": route} + payload.update(params) + sock = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(self.timeout) + sock.connect((self.host, self.port)) + body = json.dumps(payload).encode("utf-8") + sock.sendall(struct.pack(">I", len(body)) + body) + + header = self._recv_exact(sock, 4) + if header is None: + return False, "no response header" + length = struct.unpack(">I", header)[0] + payload_bytes = self._recv_exact(sock, length) + if payload_bytes is None: + return False, "incomplete payload" + resp = json.loads(payload_bytes.decode("utf-8")) + return resp.get("success", False), resp.get("data", resp.get("error")) + except Exception as e: + return False, str(e) + finally: + if sock: + sock.close() + + @staticmethod + def _recv_exact(sock, n): + data = b"" + while len(data) < n: + chunk = sock.recv(n - len(data)) + if not chunk: + return None + data += chunk + return data + + +# ── 测试框架 ──────────────────────────────────────────────────────────── + +class Counter: + def __init__(self): + self.passed = 0 + self.failed = 0 + self.failures = [] + + def check(self, condition, name, detail=""): + if condition: + print(f" [PASS] {name}") + self.passed += 1 + return True + else: + msg = f" [FAIL] {name}" + if detail: + msg += f": {detail}" + print(msg) + self.failed += 1 + self.failures.append((name, detail)) + return False + + def summary(self): + total = self.passed + self.failed + print(f"\n{'='*50}") + print(f"Results: {self.passed} passed, {self.failed} failed ({total} total)") + if self.failures: + print("Failures:") + for name, detail in self.failures: + print(f" - {name}: {detail}") + return self.failed == 0 + + +# ── 测试用例 ──────────────────────────────────────────────────────────── + +def test_ping(c, cnt): + ok, data = c.call("ping") + cnt.check(ok and data == "pong", "ping", f"got {data}") + + +def test_set_and_get(c, cnt): + ok, data = c.call("set", key="test:key1", value="hello") + cnt.check(ok and data is True, "set new key", f"got {data}") + + ok, data = c.call("get", key="test:key1") + cnt.check(ok and data == "hello", "get existing key", f"got {data}") + + ok, data = c.call("get", key="test:nonexistent") + cnt.check(ok and data is None, "get nonexistent key", f"got {data}") + + +def test_set_duplicate(c, cnt): + c.call("set", key="test:dup", value="first") + ok, data = c.call("set", key="test:dup", value="second") + cnt.check(ok and data is False, "set duplicate key", f"got {data}") + + ok, data = c.call("get", key="test:dup") + cnt.check(ok and data == "first", "value unchanged after duplicate set", f"got {data}") + + +def test_del(c, cnt): + c.call("set", key="test:del", value="tmp") + ok, data = c.call("del", key="test:del") + cnt.check(ok and data is True, "del existing key", f"got {data}") + + ok, data = c.call("get", key="test:del") + cnt.check(ok and data is None, "get after del returns null", f"got {data}") + + ok, data = c.call("del", key="test:del") + cnt.check(ok and data is False, "del nonexistent key", f"got {data}") + + +def test_update(c, cnt): + c.call("set", key="test:update", value="v1") + ok, data = c.call("update", key="test:update", value="v2") + cnt.check(ok and data is True, "update without refresh", f"got {data}") + + ok, data = c.call("get", key="test:update") + cnt.check(ok and data == "v2", "value after update", f"got {data}") + + ok, data = c.call("update", key="test:nonexistent", value="x") + cnt.check(ok and data is False, "update nonexistent", f"got {data}") + + +def test_update_with_refresh(c, cnt): + c.call("set", key="test:refresh", value="old") + ok, data = c.call("update", key="test:refresh", value="new", refresh_ttl=True) + cnt.check(ok and data is True, "update with refresh_ttl", f"got {data}") + + ok, data = c.call("get", key="test:refresh") + cnt.check(ok and data == "new", "value after refresh update", f"got {data}") + + +def test_exists(c, cnt): + c.call("set", key="test:exists", value="yes") + ok, data = c.call("exists", key="test:exists") + cnt.check(ok and data is True, "exists true", f"got {data}") + + ok, data = c.call("exists", key="test:ghost") + cnt.check(ok and data is False, "exists false", f"got {data}") + + +def test_size(c, cnt): + # 清理 + for k in ["test:s1", "test:s2", "test:s3"]: + c.call("del", key=k) + + c.call("set", key="test:s1", value="a") + c.call("set", key="test:s2", value="b") + ok, data = c.call("size") + cnt.check(ok and isinstance(data, int) and data >= 2, "size >= 2", f"got {data}") + + c.call("del", key="test:s1") + ok, data = c.call("size") + cnt.check(ok and isinstance(data, int), "size after del", f"got {data}") + + +def test_show(c, cnt): + c.call("set", key="test:show1", value="v1") + c.call("set", key="test:show2", value="v2") + ok, data = c.call("show") + cnt.check(ok and isinstance(data, list), "show returns array", f"type={type(data)}") + if isinstance(data, list): + keys = {item["key"] for item in data if "key" in item} + cnt.check("test:show1" in keys, "show contains key1") + cnt.check("test:show2" in keys, "show contains key2") + + +def test_expire(c, cnt): + c.call("set", key="test:exp", value="old") + ok, data = c.call("expire", key="test:exp", value="new") + cnt.check(ok and data is True, "expire existing key", f"got {data}") + + ok, data = c.call("get", key="test:exp") + cnt.check(ok and data == "new", "value after expire", f"got {data}") + + +def test_missing_params(c, cnt): + ok, data = c.call("set", key="only_key") + cnt.check(not ok and isinstance(data, str), "set missing 'value' returns error", f"got {data}") + + ok, data = c.call("get") + cnt.check(not ok, "get missing 'key' returns error", f"got {data}") + + +def test_concurrent(c, cnt): + """并发 set/get：验证 mutex 保护不崩溃""" + errors = [] + results = [] + + def worker(worker_id): + client = MemdbClient() + for i in range(20): + key = f"concurrent:w{worker_id}:k{i}" + ok, _ = client.call("set", key=key, value=str(worker_id)) + if not ok: + errors.append(f"w{worker_id} set failed") + ok, val = client.call("get", key=key) + if not ok or val != str(worker_id): + errors.append(f"w{worker_id} get mismatch: {val}") + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + + cnt.check(len(errors) == 0, "concurrent 4x20 set/get", f"{len(errors)} errors") + if errors: + for e in errors[:3]: + print(f" {e}") + + +# ── main ───────────────────────────────────────────────────────────────── + +def main(): + host = sys.argv[1] if len(sys.argv) > 1 else "127.0.0.1" + port = int(sys.argv[2]) if len(sys.argv) > 2 else 8011 + + print(f"=== memdb RPC integration tests ({host}:{port}) ===") + print() + + # 检查服务是否可达 + try: + s = socket.create_connection((host, port), timeout=3.0) + s.close() + except Exception as e: + print(f"[SKIP] memdb service not reachable at {host}:{port}: {e}") + print("Start the service first: cd memdb && bash _run.sh") + return 1 + + c = MemdbClient(host, port) + cnt = Counter() + + test_ping(c, cnt) + test_set_and_get(c, cnt) + test_set_duplicate(c, cnt) + test_del(c, cnt) + test_update(c, cnt) + test_update_with_refresh(c, cnt) + test_exists(c, cnt) + test_size(c, cnt) + test_show(c, cnt) + test_expire(c, cnt) + test_missing_params(c, cnt) + test_concurrent(c, cnt) + + return 0 if cnt.summary() else 1 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file