From fe9ca35bbdc853f1d531c89d189be897183c2e5b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 18 Apr 2026 20:18:38 -0700 Subject: [PATCH] peer chunk sharing 5/8: mount chunk-directory shard (#9134) mount: tier-2 chunk directory + FetchChunk streaming on one gRPC port Collapses the old two-port design (HTTP peer-serve + separate gRPC directory) into a single gRPC service that handles every mount-to- mount exchange: ChunkAnnounce, ChunkLookup, and the new FetchChunk byte stream. * peer_directory.go: fid -> holders shard, HRW-gated; returns holders in LRU order; capacity-bounded; Sweep handles eviction under write-lock while Lookup runs under RLock (hot path is concurrent). * peer_grpc.go: single MountPeer gRPC server implementing all three RPCs. FetchChunk frames bytes at 1 MiB per Send so the default 4 MiB message cap does not constrain chunk size; cache miss returns gRPC NOT_FOUND so clients distinguish miss from transport error. Reuses pb.NewGrpcServer for consistent keepalive + msg-size tuning. * peer_bytepool.go: sync.Pool wrapper around *[]byte that the server uses to avoid a fresh 8 MiB allocation per FetchChunk call. * WFS wiring starts the gRPC server on option.PeerListen (the single peer port) using the advertise address resolved in PR #3 as the HRW identity. A background sweeper evicts expired directory entries every 60 s. --- weed/mount/peer_directory.go | 283 +++++++++++++++++++++++++++++ weed/mount/peer_directory_test.go | 217 ++++++++++++++++++++++ weed/mount/peer_fake_cache_test.go | 41 +++++ weed/mount/peer_grpc.go | 254 ++++++++++++++++++++++++++ weed/mount/peer_grpc_test.go | 150 +++++++++++++++ weed/mount/weedfs.go | 56 +++++- 6 files changed, 994 insertions(+), 7 deletions(-) create mode 100644 weed/mount/peer_directory.go create mode 100644 weed/mount/peer_directory_test.go create mode 100644 weed/mount/peer_fake_cache_test.go create mode 100644 weed/mount/peer_grpc.go create mode 100644 weed/mount/peer_grpc_test.go diff --git a/weed/mount/peer_directory.go b/weed/mount/peer_directory.go new file mode 100644 index 000000000..c65b609b3 --- /dev/null +++ b/weed/mount/peer_directory.go @@ -0,0 +1,283 @@ +package mount + +import ( + "sort" + "sync" + "sync/atomic" + "time" +) + +// maxPeerDirectoryEntries caps the total (fid, holder) edges a single mount +// will hold in its shard of the chunk directory. Each edge is a +// (uint32, time.Time) map entry (~40 B with Go's map overhead), plus the +// per-peer metadata is interned — see peerMetadata below — so 500k edges +// come out to roughly 20 MB regardless of fleet size. A full directory +// rejects new announces until TTL frees room. +const maxPeerDirectoryEntries = 500000 + +// peerID is the interned identifier for a peer. Peer addresses and locality +// labels live once in the peers slice; the (fid -> holder) index references +// peers by this fixed-size id so we avoid paying for the ~48 B worth of +// string headers (addr, dc, rack) on every edge. +type peerID uint32 + +// peerMetadata is the per-peer data that used to be inlined on every +// holderEntry. Fields are updated in place on each announce so a peer +// reconfiguring its dc/rack is picked up without churn. +type peerMetadata struct { + addr string + dataCenter string + rack string +} + +// PeerDirectory is the tier-2 chunk directory: an in-memory +// `fid -> set` map, holding entries only for fids for which this +// mount is the HRW-assigned owner. Populated by inbound ChunkAnnounce RPCs, +// queried by inbound ChunkLookup RPCs. Entries expire on TTL. +// +// Peer addresses are interned into peerID. That way the hot index stores +// `fid -> (peerID -> expiry)`, a map of two fixed-size values, instead of +// a full (addr, dc, rack, expiry) struct per edge. The peer slice itself +// is bounded by fleet size (typically a few hundred), so the interning +// table's memory cost is negligible next to the per-edge savings. +// +// Lookup takes only a read lock; TTL eviction is the sole responsibility of +// the Sweep goroutine, which runs on a periodic ticker. This keeps the hot +// read path concurrent even under heavy announce traffic. +// +// See design-weed-mount-peer-chunk-sharing.md §4.2.2. +type PeerDirectory struct { + mu sync.RWMutex + + // Interning table. peers[id] gives the peer's metadata; peerIDByAddr + // maps an address back to its id when a new announce comes in. + peers []peerMetadata + peerIDByAddr map[string]peerID + + // fid -> (peerID -> expiry). The inner map's value is a bare timestamp + // — metadata is looked up via peers[id] on demand. + index map[string]map[peerID]time.Time + + // entryCount is maintained in lock-step with map mutations (under + // mu), so Stats() can report O(1) instead of scanning the index. + entryCount int + + announces atomic.Int64 + lookups atomic.Int64 + rejected atomic.Int64 + evictions atomic.Int64 + + clock func() time.Time +} + +// NewPeerDirectory constructs a directory using the real wall clock. +func NewPeerDirectory() *PeerDirectory { + return newPeerDirectoryWithClock(time.Now) +} + +func newPeerDirectoryWithClock(clock func() time.Time) *PeerDirectory { + return &PeerDirectory{ + index: map[string]map[peerID]time.Time{}, + peerIDByAddr: map[string]peerID{}, + clock: clock, + } +} + +// AnnounceResult reports how an inbound announce was handled per-fid: fids +// this receiver owns were recorded; others were rejected so the caller can +// retry the correct owner after refreshing its seed view. +type AnnounceResult struct { + Rejected []string +} + +// OwnerCheck is a predicate provided by the caller (backed by HRW on the +// current seed list) that says whether self is the owner for a given fid. +type OwnerCheck func(fid string) bool + +// internPeer returns the interned id for peerAddr, allocating a new slot +// if this is the first announce from that peer. Must be called under mu +// (write). dc/rack are refreshed on each call so reconfig is picked up. +func (d *PeerDirectory) internPeer(addr, dc, rack string) peerID { + if id, ok := d.peerIDByAddr[addr]; ok { + p := &d.peers[id] + p.dataCenter = dc + p.rack = rack + return id + } + id := peerID(len(d.peers)) + d.peers = append(d.peers, peerMetadata{addr: addr, dataCenter: dc, rack: rack}) + d.peerIDByAddr[addr] = id + return id +} + +// Announce records the caller as a holder for every accepted fid. Any fid +// for which ownerCheck returns false is rejected and surfaced back. A +// directory at maxPeerDirectoryEntries silently drops *new* edges +// (renewals of existing edges continue to succeed). +func (d *PeerDirectory) Announce(peerAddr, dataCenter, rack string, fids []string, ttl time.Duration, ownerCheck OwnerCheck) AnnounceResult { + if ttl <= 0 { + ttl = 60 * time.Second + } + expiry := d.clock().Add(ttl) + + d.mu.Lock() + defer d.mu.Unlock() + + pid := d.internPeer(peerAddr, dataCenter, rack) + + result := AnnounceResult{} + for _, fid := range fids { + if ownerCheck != nil && !ownerCheck(fid) { + result.Rejected = append(result.Rejected, fid) + d.rejected.Add(1) + continue + } + holders, ok := d.index[fid] + if !ok { + if d.entryCount >= maxPeerDirectoryEntries { + continue + } + holders = map[peerID]time.Time{} + d.index[fid] = holders + } + if _, exists := holders[pid]; !exists { + if d.entryCount >= maxPeerDirectoryEntries { + continue + } + d.entryCount++ + } + holders[pid] = expiry + d.announces.Add(1) + } + return result +} + +// LookupHolder is the public form of a holder entry. DataCenter and Rack +// are carried through so the fetcher can re-sort by its own locality +// before picking a peer to dial. +type LookupHolder struct { + PeerAddr string + DataCenter string + Rack string +} + +// maxLookupHolders caps how many holders a single ChunkLookup response +// returns per fid. The directory stores a holder per mount caching the +// fid, which on a large hot file can reach fleet size (hundreds). The +// fetcher typically only tries the first 1-3 before succeeding, so +// returning every entry is pure wire-overhead. LRU ordering ensures the +// freshest holders are the ones kept. +const maxLookupHolders = 16 + +// LookupResult holds the merged lookup response. Fids for which this +// receiver is not the owner are split out so the caller can retry. +type LookupResult struct { + PeersByFid map[string][]LookupHolder + NotOwnerFids []string +} + +// Lookup returns known holders for the requested fids. Expired edges are +// filtered out of the response but NOT deleted here — Sweep handles that +// under a write lock on its own schedule, so Lookup can stay read-only and +// concurrent. A fid owned by this receiver with no known holders returns +// an empty peer list. +// +// Holders are returned in LRU order (most recently announced first). Since +// each Announce stamps a fresh `now + ttl` expiry, sorting by expiry +// descending is equivalent to ordering by last-seen descending. Fetchers +// should try these in the returned order — the most recent announcer is +// the holder most likely to still have the chunk cached, given that each +// mount's chunk_cache is itself LRU. +func (d *PeerDirectory) Lookup(fids []string, ownerCheck OwnerCheck) LookupResult { + now := d.clock() + result := LookupResult{ + PeersByFid: map[string][]LookupHolder{}, + } + + // Collect candidate holders under the read lock, including the peer + // metadata copies — Announce mutates peers[id] in place when an + // existing peer re-announces, so reading peers[pid] outside the lock + // would race with that write. Sorting happens after the unlock. + type candidate struct { + peerMetadata + expiry time.Time + } + perFid := make(map[string][]candidate, len(fids)) + + d.mu.RLock() + for _, fid := range fids { + d.lookups.Add(1) + if ownerCheck != nil && !ownerCheck(fid) { + result.NotOwnerFids = append(result.NotOwnerFids, fid) + continue + } + holders, ok := d.index[fid] + if !ok { + result.PeersByFid[fid] = nil + continue + } + cands := make([]candidate, 0, len(holders)) + for pid, expiry := range holders { + if !expiry.After(now) { + continue // Sweep will clean this up + } + cands = append(cands, candidate{peerMetadata: d.peers[pid], expiry: expiry}) + } + perFid[fid] = cands + } + d.mu.RUnlock() + + for fid, cands := range perFid { + sort.Slice(cands, func(i, j int) bool { + if !cands[i].expiry.Equal(cands[j].expiry) { + return cands[i].expiry.After(cands[j].expiry) // newest first + } + return cands[i].addr < cands[j].addr // deterministic tiebreak + }) + // Cap response at maxLookupHolders. The fetcher typically tries 1-3 + // holders before succeeding; shipping every holder for hot fids is + // wire-overhead without hit-rate gain. + if len(cands) > maxLookupHolders { + cands = cands[:maxLookupHolders] + } + out := make([]LookupHolder, len(cands)) + for i, c := range cands { + out[i] = LookupHolder{PeerAddr: c.addr, DataCenter: c.dataCenter, Rack: c.rack} + } + result.PeersByFid[fid] = out + } + return result +} + +// Sweep purges expired edges. Safe to call periodically. Peer slots are +// left in place — their count is bounded by fleet size (a few hundred at +// most) and never grows faster than the fleet itself, so reclaiming them +// isn't worth the bookkeeping. +func (d *PeerDirectory) Sweep() int { + now := d.clock() + d.mu.Lock() + defer d.mu.Unlock() + n := 0 + for fid, holders := range d.index { + for pid, expiry := range holders { + if !expiry.After(now) { + delete(holders, pid) + d.entryCount-- + n++ + } + } + if len(holders) == 0 { + delete(d.index, fid) + } + } + d.evictions.Add(int64(n)) + return n +} + +// Stats exposes counters for observability. O(1) — entryCount is kept in +// lock-step with map mutations. +func (d *PeerDirectory) Stats() (announces, lookups, rejected, evictions int64, entries int) { + d.mu.RLock() + defer d.mu.RUnlock() + return d.announces.Load(), d.lookups.Load(), d.rejected.Load(), d.evictions.Load(), d.entryCount +} diff --git a/weed/mount/peer_directory_test.go b/weed/mount/peer_directory_test.go new file mode 100644 index 000000000..a4b463f1b --- /dev/null +++ b/weed/mount/peer_directory_test.go @@ -0,0 +1,217 @@ +package mount + +import ( + "fmt" + "sort" + "testing" + "time" +) + +// alwaysOwner is a trivial OwnerCheck used by tests that aren't exercising +// ownership rejection. +func alwaysOwner(string) bool { return true } + +// ownerSet returns an OwnerCheck that accepts only the listed fids. +func ownerSet(fids ...string) OwnerCheck { + m := map[string]struct{}{} + for _, f := range fids { + m[f] = struct{}{} + } + return func(fid string) bool { + _, ok := m[fid] + return ok + } +} + +func TestPeerDirectory_AnnounceAndLookup(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + res := d.Announce("mount-a:18080", "", "r1", []string{"3,a", "3,b"}, 60*time.Second, alwaysOwner) + if len(res.Rejected) != 0 { + t.Errorf("expected 0 rejected, got %v", res.Rejected) + } + + lookup := d.Lookup([]string{"3,a", "3,b", "3,c"}, alwaysOwner) + if got := len(lookup.PeersByFid["3,a"]); got != 1 { + t.Errorf("3,a: expected 1 holder, got %d", got) + } + if got := len(lookup.PeersByFid["3,b"]); got != 1 { + t.Errorf("3,b: expected 1 holder, got %d", got) + } + if got := len(lookup.PeersByFid["3,c"]); got != 0 { + t.Errorf("3,c (unknown): expected 0 holders, got %d", got) + } +} + +func TestPeerDirectory_OwnerRejection(t *testing.T) { + d := NewPeerDirectory() + + res := d.Announce("mount-a:18080", "", "", []string{"3,mine", "3,yours"}, time.Minute, ownerSet("3,mine")) + if len(res.Rejected) != 1 || res.Rejected[0] != "3,yours" { + t.Errorf("expected 3,yours rejected, got %v", res.Rejected) + } + + lookup := d.Lookup([]string{"3,mine", "3,yours"}, ownerSet("3,mine")) + if got := len(lookup.PeersByFid["3,mine"]); got != 1 { + t.Errorf("3,mine should have 1 holder, got %d", got) + } + if len(lookup.NotOwnerFids) != 1 || lookup.NotOwnerFids[0] != "3,yours" { + t.Errorf("expected 3,yours in not-owner list, got %v", lookup.NotOwnerFids) + } +} + +func TestPeerDirectory_MultipleHolders(t *testing.T) { + d := NewPeerDirectory() + d.Announce("mount-a:18080", "", "r1", []string{"3,x"}, time.Minute, alwaysOwner) + d.Announce("mount-b:18080", "", "r2", []string{"3,x"}, time.Minute, alwaysOwner) + + holders := d.Lookup([]string{"3,x"}, alwaysOwner).PeersByFid["3,x"] + if len(holders) != 2 { + t.Fatalf("expected 2 holders, got %d", len(holders)) + } + sort.Slice(holders, func(i, j int) bool { return holders[i].PeerAddr < holders[j].PeerAddr }) + if holders[0].PeerAddr != "mount-a:18080" || holders[1].PeerAddr != "mount-b:18080" { + t.Errorf("unexpected holder addrs: %+v", holders) + } +} + +func TestPeerDirectory_RenewExtendsExpiry(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + d.Announce("mount-a:18080", "", "", []string{"3,a"}, 30*time.Second, alwaysOwner) + + now = now.Add(25 * time.Second) + d.Announce("mount-a:18080", "", "", []string{"3,a"}, 30*time.Second, alwaysOwner) + + now = now.Add(10 * time.Second) + lookup := d.Lookup([]string{"3,a"}, alwaysOwner) + if len(lookup.PeersByFid["3,a"]) != 1 { + t.Errorf("renew should keep entry alive, got %d holders", len(lookup.PeersByFid["3,a"])) + } +} + +func TestPeerDirectory_TTLExpiry(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + d.Announce("mount-a:18080", "", "", []string{"3,a"}, 10*time.Second, alwaysOwner) + now = now.Add(15 * time.Second) + holders := d.Lookup([]string{"3,a"}, alwaysOwner).PeersByFid["3,a"] + if len(holders) != 0 { + t.Errorf("expected TTL-expired holder to be gone, got %+v", holders) + } +} + +func TestPeerDirectory_Sweep(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + d.Announce("mount-a:18080", "", "", []string{"3,a"}, 10*time.Second, alwaysOwner) + d.Announce("mount-b:18080", "", "", []string{"3,b"}, 60*time.Second, alwaysOwner) + + now = now.Add(30 * time.Second) + got := d.Sweep() + if got != 1 { + t.Errorf("expected 1 swept entry, got %d", got) + } + _, _, _, _, entries := d.Stats() + if entries != 1 { + t.Errorf("expected 1 surviving entry, got %d", entries) + } +} + +// TestPeerDirectory_LookupOrdersByRecency guards the LRU contract: the +// most recently-announced holder comes first in the lookup response, so +// fetchers that try holders in order hit the holder most likely to still +// have the chunk cached locally. +func TestPeerDirectory_LookupOrdersByRecency(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + d.Announce("mount-a:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner) + now = now.Add(10 * time.Second) + d.Announce("mount-b:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner) + now = now.Add(10 * time.Second) + d.Announce("mount-c:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner) + + got := d.Lookup([]string{"3,x"}, alwaysOwner).PeersByFid["3,x"] + if len(got) != 3 { + t.Fatalf("expected 3 holders, got %d", len(got)) + } + want := []string{"mount-c:18080", "mount-b:18080", "mount-a:18080"} + for i, w := range want { + if got[i].PeerAddr != w { + t.Errorf("holder[%d] = %q, want %q (LRU order)", i, got[i].PeerAddr, w) + } + } +} + +// TestPeerDirectory_LookupOrderAfterRenewal ensures renewing an older +// holder promotes it to the head — matches LRU touch-on-access semantics +// and keeps the freshest liveness signal in front. +func TestPeerDirectory_LookupOrderAfterRenewal(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + d.Announce("mount-a:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner) + now = now.Add(10 * time.Second) + d.Announce("mount-b:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner) + now = now.Add(10 * time.Second) + d.Announce("mount-a:18080", "", "", []string{"3,x"}, 60*time.Second, alwaysOwner) + + got := d.Lookup([]string{"3,x"}, alwaysOwner).PeersByFid["3,x"] + if len(got) != 2 { + t.Fatalf("expected 2 holders, got %d", len(got)) + } + if got[0].PeerAddr != "mount-a:18080" { + t.Errorf("after renewal, head should be mount-a; got %q", got[0].PeerAddr) + } +} + +func TestPeerDirectory_StatsCountersIncrement(t *testing.T) { + d := NewPeerDirectory() + d.Announce("mount-a:18080", "", "", []string{"3,a", "3,b"}, time.Minute, alwaysOwner) + d.Lookup([]string{"3,a", "3,c"}, alwaysOwner) + d.Announce("mount-a:18080", "", "", []string{"3,nope"}, time.Minute, ownerSet("nothing")) + + announces, lookups, rejected, _, _ := d.Stats() + if announces != 2 { + t.Errorf("announces counter: %d", announces) + } + if lookups != 2 { + t.Errorf("lookups counter: %d", lookups) + } + if rejected != 1 { + t.Errorf("rejected counter: %d", rejected) + } +} + +// TestPeerDirectory_LookupCapsHolderList verifies that a hot fid with +// more than maxLookupHolders holders returns only the top-N LRU entries. +// The fetcher typically tries only the first 1-3 anyway, so caching +// every holder on the wire is overhead. +func TestPeerDirectory_LookupCapsHolderList(t *testing.T) { + now := time.Unix(1000, 0) + d := newPeerDirectoryWithClock(func() time.Time { return now }) + + // Announce 32 holders at strictly increasing timestamps so LRU order + // is well-defined and deterministic. + total := 2 * maxLookupHolders + for i := 0; i < total; i++ { + addr := fmt.Sprintf("mount-%02d:18080", i) + d.Announce(addr, "", "", []string{"3,hot"}, 60*time.Second, alwaysOwner) + now = now.Add(time.Millisecond) + } + + got := d.Lookup([]string{"3,hot"}, alwaysOwner).PeersByFid["3,hot"] + if len(got) != maxLookupHolders { + t.Fatalf("expected %d holders in response, got %d", maxLookupHolders, len(got)) + } + // Head should be the most recent (last announced). + want := fmt.Sprintf("mount-%02d:18080", total-1) + if got[0].PeerAddr != want { + t.Errorf("head of list: got %q want %q (LRU order)", got[0].PeerAddr, want) + } +} diff --git a/weed/mount/peer_fake_cache_test.go b/weed/mount/peer_fake_cache_test.go new file mode 100644 index 000000000..01068057f --- /dev/null +++ b/weed/mount/peer_fake_cache_test.go @@ -0,0 +1,41 @@ +package mount + +// fakeChunkCache satisfies chunk_cache.ChunkCache with a simple in-memory +// map keyed by fid. Just enough to exercise the peer-serve handler paths. +type fakeChunkCache struct { + chunks map[string][]byte +} + +func newFakeChunkCache() *fakeChunkCache { + return &fakeChunkCache{chunks: map[string][]byte{}} +} + +func (f *fakeChunkCache) Put(fid string, data []byte) { + buf := make([]byte, len(data)) + copy(buf, data) + f.chunks[fid] = buf +} + +func (f *fakeChunkCache) ReadChunkAt(data []byte, fileId string, offset uint64) (int, error) { + b, ok := f.chunks[fileId] + if !ok { + return 0, nil + } + if int(offset) >= len(b) { + return 0, nil + } + return copy(data, b[offset:]), nil +} + +func (f *fakeChunkCache) SetChunk(fileId string, data []byte) { + f.Put(fileId, data) +} + +func (f *fakeChunkCache) IsInCache(fileId string, lockNeeded bool) bool { + _, ok := f.chunks[fileId] + return ok +} + +func (f *fakeChunkCache) GetMaxFilePartSizeInCache() uint64 { + return 8 * 1024 * 1024 +} diff --git a/weed/mount/peer_grpc.go b/weed/mount/peer_grpc.go new file mode 100644 index 000000000..7f8280568 --- /dev/null +++ b/weed/mount/peer_grpc.go @@ -0,0 +1,254 @@ +package mount + +import ( + "context" + "fmt" + "net" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/mount_peer_pb" + "github.com/seaweedfs/seaweedfs/weed/util/chunk_cache" + "github.com/seaweedfs/seaweedfs/weed/util/mem" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// fetchChunkStreamSize is the frame size used when server-streaming a chunk's +// bytes back to a peer. 1 MiB is well above gRPC's 32 KiB preferred frame but +// comfortably under the default 4 MiB message cap, so each Recv on the client +// returns quickly and the chunk is assembled with ~16 Recv calls for typical +// 16 MiB chunks. +const fetchChunkStreamSize = 1 * 1024 * 1024 + +// maxFetchChunkBytes caps the size of a single FetchChunk read buffer. +// The caller's expected_size is untrusted — a misbehaving peer could +// claim 1 TiB and OOM the server. 64 MiB is well above any realistic +// chunk (SeaweedFS defaults to 16 MiB, filer manifests cap at 32 MiB); +// anything larger is treated as invalid input. +const maxFetchChunkBytes = 64 * 1024 * 1024 + +// PeerGrpcServer is the single mount-to-mount gRPC endpoint. It serves: +// - ChunkAnnounce / ChunkLookup — the tier-2 directory RPCs, populated +// by inbound announces and queried by inbound lookups. Each handler +// is HRW-gated on the caller-side seed view. +// - FetchChunk (server stream) — serves bytes from the local +// chunk_cache to peers. Replaces the earlier HTTP-only peer-serve +// endpoint: one port, one authentication path, one connection pool. +// +// Transport credentials (TLS/mTLS) are injected via serverOpts. When the +// caller passes nothing the server runs plaintext — only appropriate on +// a trusted single-host test cluster. Production wiring passes +// security.LoadServerTLS from security.toml so cross-host traffic is +// authenticated + encrypted. +type PeerGrpcServer struct { + mount_peer_pb.UnimplementedMountPeerServer + + dir *PeerDirectory + cache chunk_cache.ChunkCache + ownerFor func(fid string) string // HRW owner predicate on current seeds + selfAddr string + serverOpts []grpc.ServerOption + grpcS *grpc.Server + listener net.Listener + stopped bool +} + +// NewPeerGrpcServer constructs the server. cache is the local chunk_cache +// (used to serve FetchChunk); dir is the local directory shard (used to +// answer ChunkAnnounce / ChunkLookup); ownerFor returns the HRW owner of a +// fid on the current seed view; serverOpts are forwarded to the underlying +// gRPC server (use grpc.Creds(...) for TLS/mTLS). +func NewPeerGrpcServer(cache chunk_cache.ChunkCache, dir *PeerDirectory, ownerFor func(fid string) string, selfAddr string, serverOpts ...grpc.ServerOption) *PeerGrpcServer { + return &PeerGrpcServer{ + cache: cache, + dir: dir, + ownerFor: ownerFor, + selfAddr: selfAddr, + serverOpts: serverOpts, + } +} + +// Start binds a TCP listener at addr and registers the MountPeer service. +func (s *PeerGrpcServer) Start(addr string) error { + ln, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("peer grpc listen %s: %w", addr, err) + } + s.listener = ln + s.grpcS = pb.NewGrpcServer(s.serverOpts...) + mount_peer_pb.RegisterMountPeerServer(s.grpcS, s) + go func() { + if err := s.grpcS.Serve(ln); err != nil && err != grpc.ErrServerStopped { + glog.Warningf("peer-grpc terminated: %v", err) + } + }() + glog.V(0).Infof("peer-grpc listening on %s", ln.Addr()) + return nil +} + +// Stop halts the gRPC server without waiting for in-flight streams. +func (s *PeerGrpcServer) Stop() { + if s.stopped { + return + } + s.stopped = true + if s.grpcS != nil { + s.grpcS.Stop() + } +} + +// Addr returns the bound address (useful when the caller used ":0"). +func (s *PeerGrpcServer) Addr() string { + if s.listener == nil { + return "" + } + return s.listener.Addr().String() +} + +// ChunkAnnounce accepts holder entries for fids this mount owns; rejects +// others so the caller can retry against the correct owner. +func (s *PeerGrpcServer) ChunkAnnounce(ctx context.Context, req *mount_peer_pb.ChunkAnnounceRequest) (*mount_peer_pb.ChunkAnnounceResponse, error) { + ttl := time.Duration(req.TtlSeconds) * time.Second + res := s.dir.Announce(req.PeerAddr, req.DataCenter, req.Rack, req.FileIds, ttl, s.ownerPredicate) + return &mount_peer_pb.ChunkAnnounceResponse{ + RejectedFileIds: res.Rejected, + }, nil +} + +// ChunkLookup returns known holders for each requested fid in LRU order. +func (s *PeerGrpcServer) ChunkLookup(ctx context.Context, req *mount_peer_pb.ChunkLookupRequest) (*mount_peer_pb.ChunkLookupResponse, error) { + res := s.dir.Lookup(req.FileIds, s.ownerPredicate) + resp := &mount_peer_pb.ChunkLookupResponse{ + PeersByFid: make(map[string]*mount_peer_pb.PeerSet, len(res.PeersByFid)), + NotOwnerFileIds: res.NotOwnerFids, + } + for fid, holders := range res.PeersByFid { + peers := &mount_peer_pb.PeerSet{} + for _, h := range holders { + peers.Peers = append(peers.Peers, &mount_peer_pb.PeerInfo{ + PeerAddr: h.PeerAddr, + DataCenter: h.DataCenter, + Rack: h.Rack, + }) + } + resp.PeersByFid[fid] = peers + } + return resp, nil +} + +// FetchChunk streams bytes of a cached chunk back to the caller. Missing +// fid → gRPC NOT_FOUND. Bytes are framed at fetchChunkStreamSize so gRPC's +// default 4 MiB message cap does not constrain chunk size. +func (s *PeerGrpcServer) FetchChunk(req *mount_peer_pb.FetchChunkRequest, stream mount_peer_pb.MountPeer_FetchChunkServer) error { + if s.cache == nil { + return status.Error(codes.Unavailable, "chunk cache not configured") + } + fid := req.FileId + if fid == "" { + return status.Error(codes.InvalidArgument, "missing file_id") + } + + // Size the read buffer to the caller-reported chunk length. The + // TieredChunkCache.ReadChunkAt wrapper only returns success when n + // equals len(data), so a buffer larger than the actual stored chunk + // (e.g. an 8 MiB buffer for a 2 MiB chunk) makes every read look + // like a miss. Client sends expected_size from FileChunk.Size so we + // allocate exactly the right amount. Fall back to the max-part-size + // when the caller left it zero (older clients) and hope the chunk + // is that big. + // + // expected_size is untrusted — cap at maxFetchChunkBytes to prevent + // a misbehaving peer from requesting an OOM-sized allocation. + if req.ExpectedSize > maxFetchChunkBytes { + return status.Errorf(codes.InvalidArgument, + "expected_size %d exceeds max %d", req.ExpectedSize, maxFetchChunkBytes) + } + if req.Length > maxFetchChunkBytes { + return status.Errorf(codes.InvalidArgument, + "length %d exceeds max %d", req.Length, maxFetchChunkBytes) + } + readSize := int(req.ExpectedSize) + if readSize <= 0 { + max := s.cache.GetMaxFilePartSizeInCache() + if max == 0 { + max = 8 * 1024 * 1024 + } + readSize = int(max) + } + // mem.Allocate rounds up to the nearest power-of-2 slot backed by a + // shared sync.Pool; avoids an allocation per FetchChunk call. + buf := mem.Allocate(readSize) + defer mem.Free(buf) + + n, err := s.cache.ReadChunkAt(buf, fid, 0) + if err != nil || n <= 0 { + return status.Errorf(codes.NotFound, "fid %s not cached", fid) + } + + // Apply optional offset / length range to the whole-chunk buffer. + // Default (both 0) is a full-chunk transfer starting at byte 0, + // which is what the fetcher uses today. Non-default is reserved + // for future range-read callers. + lo := int(req.Offset) + if lo < 0 || lo > n { + return status.Errorf(codes.OutOfRange, "offset %d outside chunk length %d", lo, n) + } + hi := n + if req.Length > 0 { + hi = lo + int(req.Length) + if hi > n { + hi = n + } + } + + for off := lo; off < hi; off += fetchChunkStreamSize { + end := off + fetchChunkStreamSize + if end > hi { + end = hi + } + if sendErr := stream.Send(&mount_peer_pb.FetchChunkResponse{ + Data: buf[off:end], + }); sendErr != nil { + return sendErr + } + } + return nil +} + +func (s *PeerGrpcServer) ownerPredicate(fid string) bool { + if s.ownerFor == nil { + return true // no HRW configured → accept all (single-mount mode) + } + return s.ownerFor(fid) == s.selfAddr +} + +// peerDirectorySweepInterval is how often the mount evicts expired +// directory entries. Lookup no longer deletes inline (it takes only an +// RLock), so this sweeper is the sole memory reclamation path. +const peerDirectorySweepInterval = 60 * time.Second + +// runPeerDirectorySweeper runs until stopCh closes. Stopping it on +// unmount prevents the goroutine from outliving the WFS, which matters +// for tests and for any embedded use of NewSeaweedFileSystem where the +// filesystem object is recreated without a process exit. +func (wfs *WFS) runPeerDirectorySweeper(stopCh <-chan struct{}) { + ticker := time.NewTicker(peerDirectorySweepInterval) + defer ticker.Stop() + for { + select { + case <-stopCh: + return + case <-ticker.C: + dir := wfs.peerDirectory + if dir == nil { + return + } + if evicted := dir.Sweep(); evicted > 0 { + glog.V(2).Infof("peer directory: evicted %d expired entries", evicted) + } + } + } +} diff --git a/weed/mount/peer_grpc_test.go b/weed/mount/peer_grpc_test.go new file mode 100644 index 000000000..1a51d8bb0 --- /dev/null +++ b/weed/mount/peer_grpc_test.go @@ -0,0 +1,150 @@ +package mount + +import ( + "context" + "io" + "net" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/mount_peer_pb" + "github.com/seaweedfs/seaweedfs/weed/util/chunk_cache" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +func newTestPeerGrpc(t *testing.T, cache chunk_cache.ChunkCache, ownerFor func(fid string) string, selfAddr string) (mount_peer_pb.MountPeerClient, func()) { + t.Helper() + dir := NewPeerDirectory() + srv := NewPeerGrpcServer(cache, dir, ownerFor, selfAddr) + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + srv.listener = ln + srv.grpcS = grpc.NewServer() + mount_peer_pb.RegisterMountPeerServer(srv.grpcS, srv) + go srv.grpcS.Serve(ln) + + conn, err := grpc.NewClient(ln.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + t.Fatalf("dial: %v", err) + } + client := mount_peer_pb.NewMountPeerClient(conn) + cleanup := func() { + conn.Close() + srv.Stop() + } + return client, cleanup +} + +func TestPeerGrpcServer_AnnounceAndLookup(t *testing.T) { + self := "self:18080" + client, cleanup := newTestPeerGrpc(t, nil, func(fid string) string { + return self + }, self) + defer cleanup() + + _, err := client.ChunkAnnounce(context.Background(), &mount_peer_pb.ChunkAnnounceRequest{ + FileIds: []string{"3,a", "3,b"}, + PeerAddr: "holder:18080", + TtlSeconds: 60, + }) + if err != nil { + t.Fatalf("announce: %v", err) + } + + resp, err := client.ChunkLookup(context.Background(), &mount_peer_pb.ChunkLookupRequest{ + FileIds: []string{"3,a", "3,missing"}, + }) + if err != nil { + t.Fatalf("lookup: %v", err) + } + if got := len(resp.PeersByFid["3,a"].Peers); got != 1 { + t.Errorf("3,a: expected 1 holder, got %d", got) + } + if _, ok := resp.PeersByFid["3,missing"]; !ok { + t.Errorf("3,missing should have a (nil/empty) peer set entry") + } +} + +func TestPeerGrpcServer_OwnerMismatch(t *testing.T) { + client, cleanup := newTestPeerGrpc(t, nil, func(fid string) string { + return "some-other:18080" + }, "self:18080") + defer cleanup() + + ann, err := client.ChunkAnnounce(context.Background(), &mount_peer_pb.ChunkAnnounceRequest{ + FileIds: []string{"3,x"}, + PeerAddr: "holder:18080", + TtlSeconds: 60, + }) + if err != nil { + t.Fatalf("announce: %v", err) + } + if len(ann.RejectedFileIds) != 1 || ann.RejectedFileIds[0] != "3,x" { + t.Errorf("expected 3,x rejected, got %v", ann.RejectedFileIds) + } +} + +// TestPeerGrpcServer_FetchChunk_StreamsHit exercises the new byte-transfer +// path: a cached chunk returned as multiple gRPC frames, concatenated by +// the caller into the original payload. +func TestPeerGrpcServer_FetchChunk_StreamsHit(t *testing.T) { + cache := newFakeChunkCache() + // A payload deliberately larger than fetchChunkStreamSize to force + // multiple Send() calls. + payload := make([]byte, fetchChunkStreamSize*2+123) + for i := range payload { + payload[i] = byte(i & 0xff) + } + cache.Put("3,stream", payload) + + client, cleanup := newTestPeerGrpc(t, cache, nil, "self:18080") + defer cleanup() + + stream, err := client.FetchChunk(context.Background(), &mount_peer_pb.FetchChunkRequest{FileId: "3,stream"}) + if err != nil { + t.Fatalf("FetchChunk: %v", err) + } + var got []byte + for { + resp, err := stream.Recv() + if err == io.EOF { + break + } + if err != nil { + t.Fatalf("Recv: %v", err) + } + got = append(got, resp.Data...) + } + if len(got) != len(payload) { + t.Fatalf("got %d bytes, want %d", len(got), len(payload)) + } + for i, b := range got { + if b != payload[i] { + t.Fatalf("byte mismatch at offset %d: got %02x want %02x", i, b, payload[i]) + } + } +} + +// TestPeerGrpcServer_FetchChunk_NotFound verifies that a miss returns +// gRPC NOT_FOUND so the client can distinguish miss from transport error. +func TestPeerGrpcServer_FetchChunk_NotFound(t *testing.T) { + cache := newFakeChunkCache() + client, cleanup := newTestPeerGrpc(t, cache, nil, "self:18080") + defer cleanup() + + stream, err := client.FetchChunk(context.Background(), &mount_peer_pb.FetchChunkRequest{FileId: "3,missing"}) + if err != nil { + t.Fatalf("FetchChunk dial: %v", err) + } + _, err = stream.Recv() + if err == nil { + t.Fatalf("expected error on missing fid, got nil") + } + if status.Code(err) != codes.NotFound { + t.Errorf("expected NotFound code, got %v", status.Code(err)) + } +} diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index ce7cae26b..1f365beb8 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -140,6 +140,9 @@ type WFS struct { posixLocks *PosixLockTable rdmaClient *RDMAMountClient peerRegistrar *PeerRegistrar + peerDirectory *PeerDirectory + peerGrpcServer *PeerGrpcServer + peerDirectoryStop chan struct{} // closed on unmount to stop the sweeper goroutine FilerConf *filer.FilerConf filerClient *wdclient.FilerClient // Cached volume location client refreshMu sync.Mutex @@ -337,6 +340,17 @@ func NewSeaweedFileSystem(option *Option) *WFS { if wfs.rdmaClient != nil { wfs.rdmaClient.Close() } + if wfs.peerGrpcServer != nil { + wfs.peerGrpcServer.Stop() + } + if wfs.peerDirectoryStop != nil { + select { + case <-wfs.peerDirectoryStop: + // already closed + default: + close(wfs.peerDirectoryStop) + } + } if wfs.peerRegistrar != nil { wfs.peerRegistrar.Stop() } @@ -360,13 +374,13 @@ func NewSeaweedFileSystem(option *Option) *WFS { } // Peer chunk sharing: register with every configured filer's mount - // registry. PR 3 resolved the advertise address; pass it through so - // the registrar heartbeats a reachable identity rather than a wildcard - // bind. Broadcasting to the full filer set is what lets mounts pointing - // at different filers see each other — each filer's registry is - // in-memory and there is no filer-to-filer sync. The gRPC server that - // serves ChunkAnnounce/Lookup/FetchChunk is started later (PR 5); until - // then the registrar is a no-op beyond heartbeats. + // registry + start the single gRPC server that handles ChunkAnnounce / + // ChunkLookup / FetchChunk. Broadcasting registration to the full + // filer set is what lets mounts pointing at different filers see + // each other — each filer's registry is in-memory with no + // filer-to-filer sync, so the registrar reconstructs the union + // client-side. One port, one identity — the advertise address + // resolved in PR #3 is used for everything. if option.PeerEnabled { selfAddr, err := ResolvePeerAdvertiseAddr(option.PeerListen, option.PeerAdvertise) if err != nil { @@ -386,6 +400,34 @@ func NewSeaweedFileSystem(option *Option) *WFS { if err := wfs.peerRegistrar.Start(context.Background()); err != nil { glog.Warningf("peer registrar start: %v", err) } + + wfs.peerDirectory = NewPeerDirectory() + // Wire TLS/mTLS from security.toml's grpc.mount section so + // cross-host peer RPCs are authenticated + encrypted. When + // the section is empty both options come back nil and the + // server runs plaintext — intentional for dev/test. + peerTLSCreds, peerTLSVerify := security.LoadServerTLS(util.GetViper(), "grpc.mount") + var peerServerOpts []grpc.ServerOption + if peerTLSCreds != nil { + peerServerOpts = append(peerServerOpts, peerTLSCreds) + } + if peerTLSVerify != nil { + peerServerOpts = append(peerServerOpts, peerTLSVerify) + } + wfs.peerGrpcServer = NewPeerGrpcServer( + wfs.chunkCache, + wfs.peerDirectory, + wfs.peerRegistrar.OwnerFor, + selfAddr, + peerServerOpts..., + ) + if err := wfs.peerGrpcServer.Start(option.PeerListen); err != nil { + glog.Warningf("peer grpc start: %v", err) + wfs.peerGrpcServer = nil + } else { + wfs.peerDirectoryStop = make(chan struct{}) + go wfs.runPeerDirectorySweeper(wfs.peerDirectoryStop) + } } }