From ffc5b20cdb4382381bd4d7b10838fd85f089a4d5 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 17 Apr 2026 14:12:40 +0000 Subject: [PATCH] feat: P1-3 add heartbeat keep-alive goroutine - add heartbeat.go with heartbeatIntervalSecs var, SetHeartbeatInterval, runHeartbeat (no build tag, CGO-free) - add heartbeat_test.go with 5 tests covering interval validation and goroutine lifecycle - add StartHeartbeat to clib.go (CGO impl calling xcecli_heartbeat) - remove SetHeartbeatInterval stub from stub.go (now real code in heartbeat.go) - remove TestSetHeartbeatInterval_StubNoOp from stub_funcs_test.go - call SetHeartbeatInterval in OnClusterUpdated to track backend-pushed interval - call backend.StartHeartbeat in cmd/main.go after backend registration --- src/frontend/apiserver/backend/clib.go | 9 +++ src/frontend/apiserver/backend/handler.go | 1 + src/frontend/apiserver/backend/heartbeat.go | 33 ++++++++ .../apiserver/backend/heartbeat_test.go | 80 +++++++++++++++++++ src/frontend/apiserver/backend/stub.go | 3 - .../apiserver/backend/stub_funcs_test.go | 9 --- src/frontend/apiserver/cmd/main.go | 2 + 7 files changed, 125 insertions(+), 12 deletions(-) create mode 100644 src/frontend/apiserver/backend/heartbeat.go create mode 100644 src/frontend/apiserver/backend/heartbeat_test.go diff --git a/src/frontend/apiserver/backend/clib.go b/src/frontend/apiserver/backend/clib.go index 70c7a5e3..49202068 100644 --- a/src/frontend/apiserver/backend/clib.go +++ b/src/frontend/apiserver/backend/clib.go @@ -64,6 +64,7 @@ import "C" import ( "apiserver/models" logger "apiserver/utils" + "context" "fmt" "sync" "unsafe" @@ -153,6 +154,14 @@ func Cleanup() { C.idgen_ctx_fini(&gIDCtx) } +// StartHeartbeat launches a goroutine that sends periodic xcecli_heartbeat +// keep-alive messages to the controller. Stops when ctx is cancelled. +func StartHeartbeat(ctx context.Context) { + go runHeartbeat(ctx, func(secs int) { + C.xcecli_heartbeat(gCtx.ctx, C.int(secs)) + }) +} + // go_push_handler is called by the C library on every incoming request. // It must be exported so the C trampoline can call it from a C thread. // diff --git a/src/frontend/apiserver/backend/handler.go b/src/frontend/apiserver/backend/handler.go index 0a768246..7e36036a 100644 --- a/src/frontend/apiserver/backend/handler.go +++ b/src/frontend/apiserver/backend/handler.go @@ -64,6 +64,7 @@ func (h *DefaultHandler) OnClusterUpdated(incoming *models.ClusterInfo) { existing.Vcpus = incoming.Vcpus existing.InactiveHeartbeatCount = incoming.InactiveHeartbeatCount store.Clusters.Put(existing.ClusterName, existing) + SetHeartbeatInterval(incoming.HeartbeatInterval) } func (h *DefaultHandler) OnResourceUpdated(res *models.ResourceDefinitionInfo) { diff --git a/src/frontend/apiserver/backend/heartbeat.go b/src/frontend/apiserver/backend/heartbeat.go new file mode 100644 index 00000000..78975708 --- /dev/null +++ b/src/frontend/apiserver/backend/heartbeat.go @@ -0,0 +1,33 @@ +package backend + +import ( + "context" + "sync/atomic" + "time" +) + +// heartbeatIntervalSecs is the current heartbeat period in seconds. +// Updated atomically by SetHeartbeatInterval; read atomically by runHeartbeat. +var heartbeatIntervalSecs int32 = 30 + +// SetHeartbeatInterval updates the heartbeat period. +// Values <= 0 are ignored to prevent disabling the keep-alive. +func SetHeartbeatInterval(secs int) { + if secs > 0 { + atomic.StoreInt32(&heartbeatIntervalSecs, int32(secs)) + } +} + +// runHeartbeat loops until ctx is cancelled, calling fn(intervalSecs) on each tick. +// Extracted from clib.go for testability without CGO. +func runHeartbeat(ctx context.Context, fn func(int)) { + for { + interval := int(atomic.LoadInt32(&heartbeatIntervalSecs)) + select { + case <-ctx.Done(): + return + case <-time.After(time.Duration(interval) * time.Second): + fn(interval) + } + } +} diff --git a/src/frontend/apiserver/backend/heartbeat_test.go b/src/frontend/apiserver/backend/heartbeat_test.go new file mode 100644 index 00000000..134db928 --- /dev/null +++ b/src/frontend/apiserver/backend/heartbeat_test.go @@ -0,0 +1,80 @@ +package backend + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +func TestSetHeartbeatInterval_ValidValue(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 30) + SetHeartbeatInterval(60) + if got := atomic.LoadInt32(&heartbeatIntervalSecs); got != 60 { + t.Errorf("heartbeatIntervalSecs = %d, want 60", got) + } +} + +func TestSetHeartbeatInterval_ZeroIgnored(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 30) + SetHeartbeatInterval(0) + if got := atomic.LoadInt32(&heartbeatIntervalSecs); got != 30 { + t.Errorf("heartbeatIntervalSecs = %d, want 30 (zero should be ignored)", got) + } +} + +func TestSetHeartbeatInterval_NegativeIgnored(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 30) + SetHeartbeatInterval(-1) + if got := atomic.LoadInt32(&heartbeatIntervalSecs); got != 30 { + t.Errorf("heartbeatIntervalSecs = %d, want 30 (negative should be ignored)", got) + } +} + +func TestRunHeartbeat_CallsFunction(t *testing.T) { + atomic.StoreInt32(&heartbeatIntervalSecs, 1) + + var callCount int64 + fn := func(_ int) { + atomic.AddInt64(&callCount, 1) + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + go runHeartbeat(ctx, fn) + + time.Sleep(1500 * time.Millisecond) + cancel() + + if got := atomic.LoadInt64(&callCount); got < 1 { + t.Errorf("fn was called %d times, want at least 1", got) + } +} + +func TestRunHeartbeat_CancelsOnContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() // already cancelled + + var callCount int64 + fn := func(_ int) { + atomic.AddInt64(&callCount, 1) + } + + done := make(chan struct{}) + go func() { + runHeartbeat(ctx, fn) + close(done) + }() + + select { + case <-done: + // returned quickly as expected + case <-time.After(100 * time.Millisecond): + t.Error("runHeartbeat did not return promptly after context cancellation") + } + + if got := atomic.LoadInt64(&callCount); got != 0 { + t.Errorf("fn was called %d times, want 0 for already-cancelled context", got) + } +} diff --git a/src/frontend/apiserver/backend/stub.go b/src/frontend/apiserver/backend/stub.go index 1520d76f..b290eba8 100644 --- a/src/frontend/apiserver/backend/stub.go +++ b/src/frontend/apiserver/backend/stub.go @@ -36,9 +36,6 @@ func ControlHost(hostname string, action int) error { return errNoCGO } // StartHeartbeat stub — real implementation in clib.go (requires CGO). func StartHeartbeat(_ context.Context) {} -// SetHeartbeatInterval stub — real implementation in clib.go (requires CGO). -func SetHeartbeatInterval(_ int) {} - // EvalSelectForHost stub — real implementation in clib.go (requires CGO). func EvalSelectForHost(_, _ string) error { return errNoCGO } diff --git a/src/frontend/apiserver/backend/stub_funcs_test.go b/src/frontend/apiserver/backend/stub_funcs_test.go index 042107a0..4b1c3612 100644 --- a/src/frontend/apiserver/backend/stub_funcs_test.go +++ b/src/frontend/apiserver/backend/stub_funcs_test.go @@ -14,15 +14,6 @@ func TestStartHeartbeat_StubNoOp(t *testing.T) { StartHeartbeat(context.Background()) } -// TestSetHeartbeatInterval_StubNoOp verifies the no-CGO stub for -// SetHeartbeatInterval is a no-op that does not panic. -func TestSetHeartbeatInterval_StubNoOp(t *testing.T) { - // Should not panic and completes immediately - SetHeartbeatInterval(30) - SetHeartbeatInterval(0) - SetHeartbeatInterval(-1) -} - // TestEvalSelectForHost_StubErrNoCGO verifies the no-CGO stub for // EvalSelectForHost returns errNoCGO, blocking callers from using // select expressions without the real CGO backend. diff --git a/src/frontend/apiserver/cmd/main.go b/src/frontend/apiserver/cmd/main.go index c830c548..1b200b76 100644 --- a/src/frontend/apiserver/cmd/main.go +++ b/src/frontend/apiserver/cmd/main.go @@ -9,6 +9,7 @@ import ( "apiserver/store" "apiserver/utils" logger "apiserver/utils" + "context" "flag" "fmt" "io" @@ -71,6 +72,7 @@ func main() { os.Exit(1) } logger.Info("********** Backend connection registered!") + backend.StartHeartbeat(context.Background()) } else { logger.Info("********** Backend not configured — running in stub mode") }