diff --git a/service/cluster/bootstrap.go b/cluster/bootstrap.go similarity index 100% rename from service/cluster/bootstrap.go rename to cluster/bootstrap.go diff --git a/service/cluster/p2c.go b/cluster/p2c.go similarity index 94% rename from service/cluster/p2c.go rename to cluster/p2c.go index 8aeae60..a850527 100644 --- a/service/cluster/p2c.go +++ b/cluster/p2c.go @@ -14,7 +14,6 @@ import ( // 代价函数:score = ewmaLatency × (inflight + 1)——既看历史延迟(EWMA 平滑),又看当前在途 // (+1 使空闲后端也有区分度)。冷启动 ewma=0 → score=0 → 优先被选中以探测(类慢启动)。 // -// 与有界负载一致性哈希(BoundedRing)互补:后者管局部性 + 负载上界,本器管延迟感知选优; // 二者都是无状态请求/副本 LB 原语(如把读请求在一组副本间择优),非有状态数据放置。 // // 并发安全。 @@ -23,7 +22,7 @@ type P2CBalancer struct { backends []string ewma map[string]float64 // 各后端延迟的 EWMA(纳秒) inflight map[string]int - decay float64 // EWMA 平滑系数 (0,1],越大越跟新样本 + decay float64 // EWMA 平滑系数 (0,1],越大越跟新样本 rng *rand.Rand } diff --git a/service/cluster/p2c_test.go b/cluster/p2c_test.go similarity index 100% rename from service/cluster/p2c_test.go rename to cluster/p2c_test.go diff --git a/service/cluster/peerpool.go b/cluster/peerpool.go similarity index 100% rename from service/cluster/peerpool.go rename to cluster/peerpool.go diff --git a/service/cluster/placement.go b/cluster/placement.go similarity index 67% rename from service/cluster/placement.go rename to cluster/placement.go index 940a224..6458164 100644 --- a/service/cluster/placement.go +++ b/cluster/placement.go @@ -1,15 +1,11 @@ package cluster import ( - "errors" "log/slog" ) -// errNotImplemented 标记尚未落地的控制面能力(数据迁移、跨节点转发等)。 // 这些能力依赖传输层重写(当前 Raft 走 net/rpc 静态传输),属 stretch 范围, // 见架构文档。桩实现统一返回此错误,避免调用方误以为已生效。 -var errNotImplemented = errors.New("cluster: not implemented") - // Placement 是放置控制面:组合一致性哈希环与节点注册表,回答「某 key 当前的 // 属主是谁」,并提供故障转移与再平衡的入口。 // @@ -45,13 +41,7 @@ func (p *Placement) Failover(deadNode string) { slog.Info("[cluster] failover: node removed from ring", "node", deadNode) } -// Rebalance 是再平衡桩(stretch)。 -// -// 真正的再平衡需要在成员变更后执行真实的数据迁移(把 key 的实际数据从旧属主 -// 搬到新属主),这依赖跨节点数据传输通道——当前 Raft 使用 net/rpc 静态传输, -// 无法承载分片迁移,属传输层重写范围(见架构文档)。此处仅记录 TODO 并返回 -// errNotImplemented,保留接口与调用点,待传输层就绪后填充。 -func (p *Placement) Rebalance() error { - slog.Warn("[cluster] rebalance: TODO, requires data migration over a new transport (stretch)") - return errNotImplemented +// IsLocal 判定 key 的属主是否为 self(本节点),供网关决定本地处理还是转发。 +func (p *Placement) IsLocal(key []byte, self string) bool { + return p.OwnerOf(key) == self } diff --git a/service/cluster/registry.go b/cluster/registry.go similarity index 100% rename from service/cluster/registry.go rename to cluster/registry.go diff --git a/service/cluster/registry_test.go b/cluster/registry_test.go similarity index 100% rename from service/cluster/registry_test.go rename to cluster/registry_test.go diff --git a/service/cluster/routing.go b/cluster/routing.go similarity index 85% rename from service/cluster/routing.go rename to cluster/routing.go index f5cdbcc..d89fe4c 100644 --- a/service/cluster/routing.go +++ b/cluster/routing.go @@ -1,11 +1,21 @@ -// Package cluster 实现分片分布式集群的路由与控制面骨架。 +// Package cluster 是分片集群的控制面:决定「一个 key 归属哪个分片、哪个物理节点」, +// 以及节点间的读择优与转发连接复用。它与存储、传输解耦,只依赖 bannet 做跨节点调用。 // -// 设计动机:BanDB 的定位是「数仓写入前置缓冲引擎」,向分片集群演进时需要 -// 一层与存储解耦的「放置与路由控制面」(借鉴 dubbo-go / PD 的思路)。本包 -// 只承担控制面职责——决定「一个 key 归属哪个物理节点 / 哪个分片」,以及节点 -// 存活的注册发现;真实的跨节点数据迁移与传输属于传输层重写范围,本包以桩标注。 +// 已在生产路径上运行的部分: // -// 零第三方依赖:一致性哈希仅使用标准库 hash/crc32。 +// - HashRing / ShardOf / ShardReplicas —— 一致性哈希归属与分片副本集, +// 由 service/shardkv 与 service.Router 使用。 +// - P(P2C)—— 两选一的延迟感知读择优,由 shardkv 的转发读使用。 +// - PeerPool —— 按地址复用的跨节点转发连接池,由 Router 的属主转发使用。 +// +// 仍是骨架、当前不产生行为的部分: +// +// - Registry 与 Placement 的存活视图。集群尚无心跳(Heartbeat 无调用方), +// 调用方传入远超进程寿命的 TTL,故所有节点恒被视为存活,Placement.OwnerOf +// 等价于 HashRing.NodeFor。接入心跳后它才开始起作用。 +// - Placement.Failover 能把节点摘出环,但目前没有故障检测来触发它。 +// +// 零第三方依赖:一致性哈希仅用标准库 hash/crc32。 package cluster import ( diff --git a/service/cluster/routing_test.go b/cluster/routing_test.go similarity index 100% rename from service/cluster/routing_test.go rename to cluster/routing_test.go diff --git a/docs/distributed-delivery-cluster-skeleton.md b/docs/distributed-delivery-cluster-skeleton.md index e2a3dc2..783a6ec 100644 --- a/docs/distributed-delivery-cluster-skeleton.md +++ b/docs/distributed-delivery-cluster-skeleton.md @@ -103,3 +103,11 @@ flowchart TD ## 已知(非本次引入)问题 - `Server/server.go` 无 `//go:build !pprof` 标签,与 `Server/server_pprof.go`(`//go:build pprof`)在 `-tags pprof` 下 `main` 重复声明。这是 origin/main 上的**既有问题**,本次骨架未修(遵循外科手术式修改,单独提出)。修复方式:给 `server.go` 加 `//go:build !pprof`。 + +--- + +> **后续状态(本文之后)**:本文所述骨架中,`Placement.Forward` 与 `Placement.Rebalance` +> 两个桩已移除——其理由「跨节点数据传输尚不可用」已不成立,转发能力后来由 `Router` + +> `PeerPool` 经 BanNet 落地(见 iteration-2026-08-05-shard-routing-banNet)。 +> `Registry` 与 `Placement.Failover` 保留,但集群仍无心跳,故存活视图当前不产生行为。 +> 包位置亦由 `service/cluster` 上提为顶层 `cluster`。 diff --git a/docs/iteration-2026-08-05-bounded-load-consistent-hashing.md b/docs/iteration-2026-08-05-bounded-load-consistent-hashing.md index 54d53fd..e9c2cd1 100644 --- a/docs/iteration-2026-08-05-bounded-load-consistent-hashing.md +++ b/docs/iteration-2026-08-05-bounded-load-consistent-hashing.md @@ -41,3 +41,9 @@ capacity = ⌈(1+ε) · 总负载 / 节点数⌉ - 接入副本读/请求 LB 路径(ShardKV 的读、或转发到副本集)。 - ε 可配、按节点权重的加权有界负载。 + +--- + +> **后续状态(本文之后)**:`BoundedRing` 实现已从代码库移除——它自落地起未被任何调用方 +> 引用(包括 `cluster` 包内部),实际承担归属计算的一直是 `HashRing`。本文保留为当时的 +> 设计与取舍记录;如需重新引入,代码见该次移除前的 git 历史。 diff --git a/service/cluster/bounded_ring.go b/service/cluster/bounded_ring.go deleted file mode 100644 index 46d25f4..0000000 --- a/service/cluster/bounded_ring.go +++ /dev/null @@ -1,101 +0,0 @@ -package cluster - -import ( - "math" - "sync" -) - -// BoundedRing 是「有界负载一致性哈希」(Google, Consistent Hashing with Bounded Loads): -// 在一致性哈希的局部性基础上,给每个节点设容量上限 ⌈(1+ε)·总负载/节点数⌉;分配 key 时从其 -// 哈希落点顺时针找第一个「未满」的节点。于是每个节点的负载被限制在均值的 (1+ε) 倍内、消除 -// 热点,同时仍保持一致性哈希「增删节点移动最少」的性质。 -// -// 用途定位(诚实):这是**请求/副本负载均衡**原语——带局部性偏好、但负载有界。它允许同一 key -// 的不同请求在主节点满载时溢出到邻居,故适合无状态请求分发(如把转发请求均衡到一组副本), -// **不适合有状态数据放置**(数据放置要求 key 恒定映射到同一节点,见 NodeFor/ShardOf)。 -// -// 并发安全。鸽巢原理保证:因容量之和 ≥ (1+ε)·总负载 ≥ 总负载,任一时刻必有未满节点。 -type BoundedRing struct { - mu sync.Mutex - ring *HashRing - nodes []string - epsilon float64 - load map[string]int - total int -} - -// NewBoundedRing 构造有界负载环。epsilon<=0 取默认 0.25(即负载上限为均值的 1.25 倍)。 -func NewBoundedRing(nodes []string, vnodes int, epsilon float64) *BoundedRing { - if epsilon <= 0 { - epsilon = 0.25 - } - load := make(map[string]int, len(nodes)) - for _, n := range nodes { - load[n] = 0 - } - return &BoundedRing{ - ring: NewHashRing(nodes, vnodes), - nodes: append([]string(nil), nodes...), - epsilon: epsilon, - load: load, - } -} - -// capacityLocked 返回当前每节点容量上限(调用方持锁)。 -func (b *BoundedRing) capacityLocked() int { - n := len(b.nodes) - if n == 0 { - return 0 - } - return int(math.Ceil((1 + b.epsilon) * float64(b.total) / float64(n))) -} - -// Assign 为 key 选一个节点:从其哈希落点顺时针找第一个未达容量上限的节点,并计一次负载。 -// 完成后须以同一 node 调 Release 归还。空环返回 ("", false)。 -func (b *BoundedRing) Assign(key []byte) (string, bool) { - b.mu.Lock() - defer b.mu.Unlock() - if len(b.nodes) == 0 { - return "", false - } - b.total++ - capLimit := b.capacityLocked() - for _, node := range b.ring.walkNodesFrom(key) { - if b.load[node] < capLimit { - b.load[node]++ - return node, true - } - } - // 理论到不了(鸽巢原理);防御性回退主节点。 - primary := b.ring.NodeFor(key) - b.load[primary]++ - return primary, true -} - -// Release 归还一次负载。 -func (b *BoundedRing) Release(node string) { - b.mu.Lock() - defer b.mu.Unlock() - if b.load[node] > 0 { - b.load[node]-- - b.total-- - } -} - -// Loads 返回各节点当前负载快照。 -func (b *BoundedRing) Loads() map[string]int { - b.mu.Lock() - defer b.mu.Unlock() - out := make(map[string]int, len(b.load)) - for k, v := range b.load { - out[k] = v - } - return out -} - -// Capacity 返回当前容量上限(供观测)。 -func (b *BoundedRing) Capacity() int { - b.mu.Lock() - defer b.mu.Unlock() - return b.capacityLocked() -} diff --git a/service/cluster/bounded_ring_test.go b/service/cluster/bounded_ring_test.go deleted file mode 100644 index 019ed87..0000000 --- a/service/cluster/bounded_ring_test.go +++ /dev/null @@ -1,88 +0,0 @@ -package cluster - -import ( - "fmt" - "math" - "testing" -) - -// maxLoad 返回负载分布中的峰值。 -func maxLoad(loads map[string]int) int { - m := 0 - for _, v := range loads { - if v > m { - m = v - } - } - return m -} - -// TestBoundedRing_BoundsHotspotUnderSkew:倾斜负载(80% 请求打同一热点 key)下, -// vanilla 一致性哈希会把热点 key 的全部负载压到一个节点,而有界负载环把峰值负载限制在 -// ⌈(1+ε)·M/N⌉ 内、把溢出均衡到邻居。 -func TestBoundedRing_BoundsHotspotUnderSkew(t *testing.T) { - nodes := []string{"n0", "n1", "n2", "n3"} - const M = 10000 - const eps = 0.25 - - // vanilla:每个请求按 NodeFor 落到主节点。 - ring := NewHashRing(nodes, 128) - vanilla := map[string]int{} - // bounded:满载溢出到邻居。 - bounded := NewBoundedRing(nodes, 128, eps) - - for i := 0; i < M; i++ { - var key []byte - if i%10 < 8 { // 80% 打同一热点 key - key = []byte("HOT") - } else { - key = []byte(fmt.Sprintf("k-%d", i)) - } - vanilla[ring.NodeFor(key)]++ - bounded.Assign(key) // 不 Release:度量 M 个请求的静态分布 - } - - vMax := maxLoad(vanilla) - bMax := maxLoad(bounded.Loads()) - capBound := int(math.Ceil((1 + eps) * float64(M) / float64(len(nodes)))) - - t.Logf("倾斜负载 M=%d N=%d ε=%.2f:vanilla 峰值=%d,bounded 峰值=%d,理论上界=%d", - M, len(nodes), eps, vMax, bMax, capBound) - - if bMax > capBound { - t.Fatalf("bounded 峰值应 ≤ 上界:%d > %d", bMax, capBound) - } - if bMax >= vMax { - t.Fatalf("bounded 应显著低于 vanilla 峰值(消除热点):bounded=%d vanilla=%d", bMax, vMax) - } -} - -// TestBoundedRing_ReleaseFreesCapacity:Release 归还负载后容量重新可用。 -func TestBoundedRing_ReleaseFreesCapacity(t *testing.T) { - b := NewBoundedRing([]string{"n0", "n1"}, 64, 0.0) // ε=0 → 容量=⌈总负载/2⌉ - assigned := make([]string, 0, 4) - for i := 0; i < 4; i++ { - n, ok := b.Assign([]byte("x")) // 同一 key,反复分配 → 溢出到两个节点 - if !ok { - t.Fatal("assign should succeed") - } - assigned = append(assigned, n) - } - // 两个节点应都被用到(容量上限迫使溢出)。 - distinct := map[string]bool{} - for _, n := range assigned { - distinct[n] = true - } - if len(distinct) < 2 { - t.Fatalf("bounded load should spread same key across nodes, got %v", assigned) - } - // 全部归还后总负载归零。 - for _, n := range assigned { - b.Release(n) - } - for n, l := range b.Loads() { - if l != 0 { - t.Fatalf("node %s load should be 0 after release, got %d", n, l) - } - } -} diff --git a/service/cluster/gateway.go b/service/cluster/gateway.go deleted file mode 100644 index 9e21cad..0000000 --- a/service/cluster/gateway.go +++ /dev/null @@ -1,30 +0,0 @@ -package cluster - -import ( - "context" - "log/slog" -) - -// IsLocal 判定 key 的属主是否为 self(本节点)。用于网关侧决定「本地处理」 -// 还是「转发到属主节点」。 -func (p *Placement) IsLocal(key []byte, self string) bool { - return p.OwnerOf(key) == self -} - -// ForwardFunc 抽象「把一次写入转发到指定节点」的行为。 -// -// 之所以做成函数类型:真实转发的实现(编解码、连接复用、重试)与控制面解耦, -// 由上层在传输层就绪后注入,控制面只负责决定「转发给谁」。 -type ForwardFunc func(ctx context.Context, node string, key, value []byte) error - -// Forward 是跨节点写转发桩(stretch)。 -// -// 控制面已能算出属主(OwnerOf),但真正把数据发往属主节点需要一条跨节点数据 -// 通道——当前 Raft 使用 net/rpc 静态传输,无法承载分片间的数据转发,属传输层 -// 重写范围(见架构文档)。此处保留接口与调用点、记录目标属主,当前返回 -// errNotImplemented,待传输层就绪后接入具体 ForwardFunc。 -func (p *Placement) Forward(ctx context.Context, key, value []byte) error { - owner := p.OwnerOf(key) - slog.Warn("[cluster] forward: TODO, cross-node data transport not implemented (stretch)", "owner", owner) - return errNotImplemented -} diff --git a/service/cluster_bootstrap.go b/service/cluster_bootstrap.go index f005901..335b8d1 100644 --- a/service/cluster_bootstrap.go +++ b/service/cluster_bootstrap.go @@ -4,10 +4,18 @@ import ( "log/slog" "time" + "github.com/NeverENG/BanDB/cluster" "github.com/NeverENG/BanDB/config" - "github.com/NeverENG/BanDB/service/cluster" ) +// assumeAliveTTL 让所有节点恒被视为存活。 +// +// 集群目前没有心跳——cluster.Registry.Heartbeat 无任何调用方,故存活视图不会被刷新。 +// 此时若取一个有限 TTL,所有节点会在该窗口后被判为失联,Placement.OwnerOf 返回空串, +// 路由随即整体中断。取远超进程寿命的值,是把「尚无心跳」这一事实显式固定下来, +// 而不是伪装成一个会过期的存活窗口。接入心跳后应改为真实的判活窗口。 +const assumeAliveTTL = 100 * 365 * 24 * time.Hour + // EnableShardRoutingFromConfig 按配置在 router 上开启分片路由(默认关闭时直接返回)。 // 开启时以 config.Peers 为节点地址构建一致性哈希放置,self = Peers[Me],不属本节点的 // key 经 BanNet 转发到 owner。健康探测/故障转移属后续工作,这里所有节点视为存活。 @@ -21,7 +29,7 @@ func EnableShardRoutingFromConfig(r *Router) { return } self := peers[config.G.Me] - placement := cluster.NewClusterFromPeers(peers, config.G.VNodes, 100*365*24*time.Hour) + placement := cluster.NewClusterFromPeers(peers, config.G.VNodes, assumeAliveTTL) pool := cluster.NewPeerPool(5 * time.Second) r.SetRouting(placement, self, pool) slog.Info("shard routing enabled", "self", self, "peers", peers) diff --git a/service/router.go b/service/router.go index dfce1f3..5b43ce0 100644 --- a/service/router.go +++ b/service/router.go @@ -6,11 +6,11 @@ import ( "log/slog" "github.com/NeverENG/BanDB/bannet" + "github.com/NeverENG/BanDB/cluster" "github.com/NeverENG/BanDB/pkg/admission" "github.com/NeverENG/BanDB/pkg/metrics" "github.com/NeverENG/BanDB/pkg/predicate" "github.com/NeverENG/BanDB/pkg/proto" - "github.com/NeverENG/BanDB/service/cluster" "github.com/NeverENG/BanDB/storage" ) diff --git a/service/shard_routing_integration_test.go b/service/shard_routing_integration_test.go index 9544eb3..396041f 100644 --- a/service/shard_routing_integration_test.go +++ b/service/shard_routing_integration_test.go @@ -10,10 +10,10 @@ import ( "time" "github.com/NeverENG/BanDB/bannet" + "github.com/NeverENG/BanDB/cluster" "github.com/NeverENG/BanDB/config" "github.com/NeverENG/BanDB/pkg/predicate" "github.com/NeverENG/BanDB/pkg/proto" - "github.com/NeverENG/BanDB/service/cluster" ) // memKV 是隔离的内存 KV,用作每个节点的本地 store——从而在一个进程内起多节点、 diff --git a/service/shardkv/read.go b/service/shardkv/read.go index a8297c1..cd0562a 100644 --- a/service/shardkv/read.go +++ b/service/shardkv/read.go @@ -4,8 +4,8 @@ import ( "fmt" "time" + "github.com/NeverENG/BanDB/cluster" "github.com/NeverENG/BanDB/raft" - "github.com/NeverENG/BanDB/service/cluster" ) // ShardReadArgs / ShardReadReply 是转发读 RPC 的报文:向某分片副本读取一个 key。 diff --git a/service/shardkv/shardkv.go b/service/shardkv/shardkv.go index 1243daa..bdabb6f 100644 --- a/service/shardkv/shardkv.go +++ b/service/shardkv/shardkv.go @@ -9,8 +9,8 @@ import ( "sync/atomic" "time" + "github.com/NeverENG/BanDB/cluster" "github.com/NeverENG/BanDB/raft" - "github.com/NeverENG/BanDB/service/cluster" ) // Shard 是一个分片:一个 Raft 组 + 该分片的 FSM store + 一个排空 ApplyCh 的 apply 循环。