1616using namespace deepx ::heap;
1717using json = nlohmann::json;
1818
19- static const char *HEAP_QUEUE = " cmd:heap-metal:0"
20- static const char *INSTANCE_KEY = " /sys/heap-plat/heap-metal:0"
21- static const int BLOCK_TIMEOUT_SEC = 5 ;
19+ static const char *HEAP_QUEUE = " cmd:heap-metal:0"
20+ static const char *SYS_QUEUE = " sys:cmd:heap-metal:0"
21+ static const char *INSTANCE_KEY = " /sys/heap-plat/heap-metal:0"
22+ static const char *HEARTBEAT_KEY = " /sys/heartbeat/heap-metal:0"
23+ static const int BLOCK_TIMEOUT_SEC = 5 ;
24+ static const int HEARTBEAT_INTERVAL_SEC = 2 ;
2225
2326// ── Redis helpers ──
2427
@@ -50,6 +53,15 @@ static bool redis_set(redisContext *c, const std::string &key, const std::string
5053 return ok;
5154}
5255
56+ static void update_heartbeat (redisContext *c, const std::string &status) {
57+ json hb;
58+ hb[" ts"
59+ std::chrono::system_clock::now ().time_since_epoch ()).count ();
60+ hb[" status"
61+ hb[" pid" getpid ();
62+ redis_set (c, HEARTBEAT_KEY , hb.dump ());
63+ }
64+
5365static void register_instance (redisContext *c) {
5466 json reg;
5567 reg[" program" " heap-metal"
@@ -248,11 +260,17 @@ int main(int argc, char **argv) {
248260 FileRegistry reg (registry_path);
249261 LifecycleManager mgr (®);
250262
251- std::cout << " [heap] listening on " HEAP_QUEUE << " \n "
263+ std::cout << " [heap] listening on " HEAP_QUEUE << " + " SYS_QUEUE << " \n "
264+ std::cout << " [heap] heartbeat → " HEARTBEAT_KEY << " (every " HEARTBEAT_INTERVAL_SEC << " s)\n "
265+
266+ // 初始心跳
267+ update_heartbeat (redis, " running"
252268
253- // 消费循环
254- while (true ) {
255- redisReply *r = redis_cmd (redis, " BLPOP %s %d" HEAP_QUEUE , BLOCK_TIMEOUT_SEC );
269+ // 消费循环 (同时监听业务队列和系统命令队列)
270+ std::atomic<bool > running{true };
271+ auto last_heartbeat = std::chrono::steady_clock::now ();
272+ while (running) {
273+ redisReply *r = redis_cmd (redis, " BLPOP %s %s %d" HEAP_QUEUE , SYS_QUEUE , BLOCK_TIMEOUT_SEC );
256274 if (!r) {
257275 // Redis 断连 → 无限重连（不自退，heap-plat 由元程控制退出）
258276 std::cerr << " [heap] Redis disconnected, reconnecting...\n "
@@ -266,9 +284,18 @@ int main(int argc, char **argv) {
266284 }
267285 }
268286 register_instance (redis);
287+ last_heartbeat = std::chrono::steady_clock::now ();
288+ update_heartbeat (redis, " running"
269289 continue ;
270290 }
271291
292+ // ── 心跳上报 ──
293+ auto now = std::chrono::steady_clock::now ();
294+ if (std::chrono::duration_cast<std::chrono::seconds>(now - last_heartbeat).count () >= HEARTBEAT_INTERVAL_SEC ) {
295+ update_heartbeat (redis, " running"
296+ last_heartbeat = now;
297+ }
298+
272299 if (r->type == REDIS_REPLY_NIL ) {
273300 // BLPOP timeout — no tasks
274301 REDIS_FREE (r);
@@ -280,9 +307,28 @@ int main(int argc, char **argv) {
280307 continue ;
281308 }
282309
310+ std::string queue_name (r->element [0 ]->str );
283311 std::string payload (r->element [1 ]->str );
284312 REDIS_FREE (r);
285313
314+ // ── 系统命令处理 ──
315+ if (queue_name == SYS_QUEUE ) {
316+ try {
317+ json sys_cmd = json::parse (payload);
318+ std::string cmd = sys_cmd.value (" cmd" " "
319+ if (cmd == " shutdown"
320+ std::cout << " [heap] received sys shutdown command, exiting...\n "
321+ running = false ;
322+ } else {
323+ std::cerr << " [heap] unknown sys command: " " \n "
324+ }
325+ } catch (const std::exception &e) {
326+ std::cerr << " [heap] sys cmd JSON parse error: " what () << " \n "
327+ }
328+ continue ;
329+ }
330+
331+ // ── 业务命令处理 ──
286332 // 解析 JSON
287333 json task;
288334 try {
@@ -318,7 +364,10 @@ int main(int argc, char **argv) {
318364 }
319365
320366 mgr.shutdown ();
367+ // 上报 stopped 心跳，然后注销
321368 if (redis) {
369+ update_heartbeat (redis, " stopped"
370+ std::cout << " [heap] final heartbeat: stopped\n "
322371 redis_cmd (redis, " DEL %s" INSTANCE_KEY );
323372 redisFree (redis);
324373 }
0 commit comments