diff --git a/weed/mount/peer_hrw.go b/weed/mount/peer_hrw.go new file mode 100644 index 000000000..9877bf3a8 --- /dev/null +++ b/weed/mount/peer_hrw.go @@ -0,0 +1,51 @@ +package mount + +import ( + "hash/fnv" +) + +// SeedPeer is a mount server present in the filer's mount registry. +// DataCenter and Rack ride along so downstream locality-aware decisions +// (peer re-ranking in the fetcher) don't have to re-query the filer. +type SeedPeer struct { + PeerAddr string + DataCenter string + Rack string +} + +// OwnerFor returns the peer_addr of the mount that is the HRW-assigned +// directory owner for fid on the given seed list. +// +// Rendezvous (highest-random-weight) hashing is deterministic for a given +// (fid, seed set): every mount computes the same owner without any +// consensus. When a mount joins or leaves the seed set, only 1/N of fids +// change owner — the rest remain on their existing mount. +// +// If seeds is empty, returns "". +// +// See design-weed-mount-peer-chunk-sharing.md §4.2.2. +func OwnerFor(fid string, seeds []SeedPeer) string { + if len(seeds) == 0 { + return "" + } + var bestScore uint64 + var bestAddr string + for _, s := range seeds { + score := hrwScore(s.PeerAddr, fid) + if bestAddr == "" || score > bestScore || (score == bestScore && s.PeerAddr < bestAddr) { + bestScore = score + bestAddr = s.PeerAddr + } + } + return bestAddr +} + +// hrwScore combines peer_addr and fid into a 64-bit score. Uses FNV-1a +// (cheap and fine for balancing; not a security primitive). +func hrwScore(peerAddr, fid string) uint64 { + h := fnv.New64a() + _, _ = h.Write([]byte(peerAddr)) + _, _ = h.Write([]byte{0}) // separator so "ab" + "c" != "a" + "bc" + _, _ = h.Write([]byte(fid)) + return h.Sum64() +} diff --git a/weed/mount/peer_hrw_test.go b/weed/mount/peer_hrw_test.go new file mode 100644 index 000000000..ec577388c --- /dev/null +++ b/weed/mount/peer_hrw_test.go @@ -0,0 +1,90 @@ +package mount + +import ( + "fmt" + "testing" +) + +func mkSeeds(addrs ...string) []SeedPeer { + out := make([]SeedPeer, len(addrs)) + for i, a := range addrs { + out[i] = SeedPeer{PeerAddr: a} + } + return out +} + +func TestOwnerFor_EmptySeeds(t *testing.T) { + if got := OwnerFor("3,01637037d6", nil); got != "" { + t.Errorf("expected empty owner for empty seed list, got %q", got) + } +} + +func TestOwnerFor_Deterministic(t *testing.T) { + seeds := mkSeeds("a:1", "b:1", "c:1", "d:1") + fid := "3,01637037d6" + first := OwnerFor(fid, seeds) + for i := 0; i < 100; i++ { + if got := OwnerFor(fid, seeds); got != first { + t.Fatalf("non-deterministic: iter %d got %q first %q", i, got, first) + } + } +} + +func TestOwnerFor_DistributesEvenly(t *testing.T) { + const N = 4 + seeds := mkSeeds("a:1", "b:1", "c:1", "d:1") + counts := map[string]int{} + for i := 0; i < 10000; i++ { + fid := fmt.Sprintf("%d,%x", i, i*17) + counts[OwnerFor(fid, seeds)]++ + } + // Each seed should get roughly 25% (±10% slack for a 10k sample). + for addr, c := range counts { + ratio := float64(c) / 10000.0 + if ratio < 0.225 || ratio > 0.275 { + t.Errorf("HRW distribution skewed: %s got %.3f", addr, ratio) + } + } + if len(counts) != N { + t.Errorf("expected %d distinct owners, got %d", N, len(counts)) + } +} + +func TestOwnerFor_MinimalShuffleOnSeedChange(t *testing.T) { + // Adding one seed should move ~1/(N+1) fids to the new seed and leave + // the rest on their prior owners. Tolerance generous for a 10k sample. + const trials = 10000 + before := mkSeeds("a:1", "b:1", "c:1") + after := mkSeeds("a:1", "b:1", "c:1", "d:1") + + moved := 0 + toNewSeed := 0 + for i := 0; i < trials; i++ { + fid := fmt.Sprintf("%d,%x", i, i*31) + pre := OwnerFor(fid, before) + post := OwnerFor(fid, after) + if pre != post { + moved++ + if post == "d:1" { + toNewSeed++ + } + } + } + // Expected: ~1/4 = 25% of fids move, all to the new seed. + ratio := float64(moved) / float64(trials) + if ratio < 0.20 || ratio > 0.30 { + t.Errorf("expected ~25%% fids to move on seed-add, got %.3f", ratio) + } + if moved != toNewSeed { + t.Errorf("expected every moved fid to land on the new seed; moved=%d toNewSeed=%d", moved, toNewSeed) + } +} + +func TestOwnerFor_TieBreakerDeterministic(t *testing.T) { + // Two seeds with equal hash score (engineered collision is unlikely in + // practice, but we guarantee determinism by lex-comparing addresses). + // This test just confirms OwnerFor runs without panicking on duplicates. + seeds := mkSeeds("a:1", "a:1", "b:1") + fid := "3,01637037d6" + _ = OwnerFor(fid, seeds) +} diff --git a/weed/mount/peer_registrar.go b/weed/mount/peer_registrar.go new file mode 100644 index 000000000..841f6d40a --- /dev/null +++ b/weed/mount/peer_registrar.go @@ -0,0 +1,251 @@ +package mount + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// PeerRegistrar maintains this mount's presence in every configured +// filer's mount registry and its local snapshot of the merged seed set. +// It runs two background tickers: +// +// - MountRegister heartbeats, fanned out to every filer in parallel so +// each filer keeps a fresh TTL entry for us. Mounts pointing at +// different filers therefore still see each other once all filers +// have been heartbeated. +// - MountList polling, fanned out identically, merged by peer_addr +// (newest LastSeenNs wins) so OwnerFor() has a view of the whole +// fleet regardless of which filer each peer happens to heartbeat +// through. +// +// An unreachable filer is tolerated: we log and continue as long as at +// least one filer succeeds. Entries cached on a permanently-gone filer +// fall out of the merged view on their own TTL. +type PeerRegistrar struct { + filerAddrs []pb.ServerAddress + dialFiler filerDialFn + selfPeerAddr string + selfDc string + selfRack string + registerInterval time.Duration + registerTTL time.Duration + listInterval time.Duration + + mu sync.RWMutex + seeds []SeedPeer + + // stopCtx cancels when Stop() is called. Background RPCs scope their + // deadline to this so unmount does not block on pending filer calls. + stopCtx context.Context + stopCancel context.CancelFunc + stopped atomic.Bool +} + +// filerDialFn is how the registrar reaches one configured filer. The +// production wiring is pb.WithGrpcFilerClient; tests inject a fake. +type filerDialFn func(ctx context.Context, addr pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error + +// NewPeerRegistrar constructs the registrar; Start launches the background +// loops. Callers must supply the full filer set so heartbeats and list +// polls reach every filer — otherwise mounts talking to different filers +// never observe each other. +func NewPeerRegistrar(filers []pb.ServerAddress, dial filerDialFn, selfAddr, dc, rack string) *PeerRegistrar { + ctx, cancel := context.WithCancel(context.Background()) + return &PeerRegistrar{ + filerAddrs: filers, + dialFiler: dial, + selfPeerAddr: selfAddr, + selfRack: rack, + selfDc: dc, + registerInterval: 30 * time.Second, + registerTTL: 90 * time.Second, + listInterval: 30 * time.Second, + stopCtx: ctx, + stopCancel: cancel, + } +} + +// Start does an initial register+list synchronously (so OwnerFor has a +// usable view immediately) and then kicks off the background loops. +func (r *PeerRegistrar) Start(ctx context.Context) error { + if err := r.registerOnce(ctx); err != nil { + glog.V(1).Infof("initial MountRegister: %v", err) + // Do not fail startup — the mount must still serve reads even if + // no filer yet knows about us. + } + if err := r.listOnce(ctx); err != nil { + glog.V(1).Infof("initial MountList: %v", err) + } + go r.loopRegister() + go r.loopList() + return nil +} + +// Stop halts the background loops and cancels any in-flight RPCs they +// may have launched. Safe to call multiple times. +func (r *PeerRegistrar) Stop() { + if r.stopped.Swap(true) { + return + } + r.stopCancel() +} + +// Seeds returns the currently-known seed set. Callers MUST NOT mutate the +// returned slice — it is shared with concurrent listOnce readers. Because +// listOnce atomically swaps to a brand-new slice on every refresh rather +// than mutating in place, returning the slice header directly is safe and +// avoids a per-call allocation on the read-hot OwnerFor path. +func (r *PeerRegistrar) Seeds() []SeedPeer { + r.mu.RLock() + defer r.mu.RUnlock() + return r.seeds +} + +// OwnerFor is a convenience wrapper that runs HRW against the current +// seed snapshot. +func (r *PeerRegistrar) OwnerFor(fid string) string { + return OwnerFor(fid, r.Seeds()) +} + +// filerRPCTimeout bounds a single MountRegister / MountList call so a slow +// or partitioned filer can't wedge the background loops (which run on a +// 30 s cadence). 20 s gives room for TLS handshakes on cold connections +// while leaving headroom before the next tick. +const filerRPCTimeout = 20 * time.Second + +// registerOnce fans a MountRegister out to every configured filer in +// parallel. Returns an error only if every filer failed; otherwise the +// best-effort semantics let the mount proceed when some filer is down. +func (r *PeerRegistrar) registerOnce(ctx context.Context) error { + if len(r.filerAddrs) == 0 { + return fmt.Errorf("no filers configured") + } + ctx, cancel := context.WithTimeout(ctx, filerRPCTimeout) + defer cancel() + req := &filer_pb.MountRegisterRequest{ + PeerAddr: r.selfPeerAddr, + Rack: r.selfRack, + DataCenter: r.selfDc, + TtlSeconds: int32(r.registerTTL / time.Second), + } + var wg sync.WaitGroup + var successes atomic.Int32 + for _, addr := range r.filerAddrs { + wg.Add(1) + go func(addr pb.ServerAddress) { + defer wg.Done() + err := r.dialFiler(ctx, addr, func(c filer_pb.SeaweedFilerClient) error { + _, err := c.MountRegister(ctx, req) + return err + }) + if err != nil { + glog.V(2).Infof("MountRegister %s: %v", addr, err) + return + } + successes.Add(1) + }(addr) + } + wg.Wait() + if successes.Load() == 0 { + return fmt.Errorf("MountRegister failed on all %d filer(s)", len(r.filerAddrs)) + } + return nil +} + +// listOnce polls MountList from every filer in parallel and merges the +// responses by peer_addr (newest LastSeenNs wins). This way two mounts +// heartbeating through different filers still end up in each other's +// seed view as soon as at least one filer has been listed on each side. +func (r *PeerRegistrar) listOnce(ctx context.Context) error { + if len(r.filerAddrs) == 0 { + r.mu.Lock() + r.seeds = nil + r.mu.Unlock() + return nil + } + ctx, cancel := context.WithTimeout(ctx, filerRPCTimeout) + defer cancel() + var ( + mu sync.Mutex + merged = map[string]*filer_pb.MountInfo{} + fails int + ) + var wg sync.WaitGroup + for _, addr := range r.filerAddrs { + wg.Add(1) + go func(addr pb.ServerAddress) { + defer wg.Done() + err := r.dialFiler(ctx, addr, func(c filer_pb.SeaweedFilerClient) error { + resp, err := c.MountList(ctx, &filer_pb.MountListRequest{}) + if err != nil { + return err + } + mu.Lock() + for _, m := range resp.Mounts { + if prev, ok := merged[m.PeerAddr]; !ok || m.LastSeenNs > prev.LastSeenNs { + merged[m.PeerAddr] = m + } + } + mu.Unlock() + return nil + }) + if err != nil { + mu.Lock() + fails++ + mu.Unlock() + glog.V(2).Infof("MountList %s: %v", addr, err) + } + }(addr) + } + wg.Wait() + + if fails == len(r.filerAddrs) { + return fmt.Errorf("MountList failed on all %d filer(s)", len(r.filerAddrs)) + } + + next := make([]SeedPeer, 0, len(merged)) + for _, m := range merged { + next = append(next, SeedPeer{PeerAddr: m.PeerAddr, DataCenter: m.DataCenter, Rack: m.Rack}) + } + r.mu.Lock() + r.seeds = next + r.mu.Unlock() + return nil +} + +func (r *PeerRegistrar) loopRegister() { + t := time.NewTicker(r.registerInterval) + defer t.Stop() + for { + select { + case <-r.stopCtx.Done(): + return + case <-t.C: + if err := r.registerOnce(r.stopCtx); err != nil { + glog.V(2).Infof("MountRegister heartbeat: %v", err) + } + } + } +} + +func (r *PeerRegistrar) loopList() { + t := time.NewTicker(r.listInterval) + defer t.Stop() + for { + select { + case <-r.stopCtx.Done(): + return + case <-t.C: + if err := r.listOnce(r.stopCtx); err != nil { + glog.V(2).Infof("MountList refresh: %v", err) + } + } + } +} diff --git a/weed/mount/peer_registrar_test.go b/weed/mount/peer_registrar_test.go new file mode 100644 index 000000000..7fee73d9b --- /dev/null +++ b/weed/mount/peer_registrar_test.go @@ -0,0 +1,210 @@ +package mount + +import ( + "context" + "sync" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "google.golang.org/grpc" +) + +// fakeFilerClient captures MountRegister/MountList calls and lets the test +// pre-seed MountList responses. Implements just enough of +// filer_pb.SeaweedFilerClient to drive the registrar. +type fakeFilerClient struct { + filer_pb.SeaweedFilerClient // embed for methods we don't need + + mu sync.Mutex + registerCalls []filer_pb.MountRegisterRequest + listResponse filer_pb.MountListResponse +} + +func (f *fakeFilerClient) MountRegister(ctx context.Context, req *filer_pb.MountRegisterRequest, opts ...grpc.CallOption) (*filer_pb.MountRegisterResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.registerCalls = append(f.registerCalls, *req) + return &filer_pb.MountRegisterResponse{}, nil +} + +func (f *fakeFilerClient) MountList(ctx context.Context, req *filer_pb.MountListRequest, opts ...grpc.CallOption) (*filer_pb.MountListResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + resp := f.listResponse // value copy + return &resp, nil +} + +// fakeFilerFleet maps each addr to its own fakeFilerClient so a test can +// simulate several filers with different registered/listed state. +type fakeFilerFleet struct { + clients map[pb.ServerAddress]*fakeFilerClient +} + +func (f *fakeFilerFleet) dial(ctx context.Context, addr pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error { + c, ok := f.clients[addr] + if !ok { + // Treat unknown filer as "reachable but empty" — lets tests omit + // pre-populating a client when they don't care. + c = &fakeFilerClient{} + f.clients[addr] = c + } + return fn(c) +} + +func singleFilerFleet(c *fakeFilerClient) ([]pb.ServerAddress, filerDialFn) { + addr := pb.ServerAddress("filer-1:18888") + fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{addr: c}} + return []pb.ServerAddress{addr}, fleet.dial +} + +func TestPeerRegistrar_StartPopulatesSeedsFromFiler(t *testing.T) { + fc := &fakeFilerClient{ + listResponse: filer_pb.MountListResponse{ + Mounts: []*filer_pb.MountInfo{ + {PeerAddr: "mount-a:18080", Rack: "r1"}, + {PeerAddr: "mount-b:18080", Rack: "r2"}, + }, + }, + } + filers, dial := singleFilerFleet(fc) + + r := NewPeerRegistrar(filers, dial, "self:18080", "dc1", "r1") + if err := r.registerOnce(context.Background()); err != nil { + t.Fatalf("registerOnce: %v", err) + } + if err := r.listOnce(context.Background()); err != nil { + t.Fatalf("listOnce: %v", err) + } + + fc.mu.Lock() + if len(fc.registerCalls) != 1 { + t.Errorf("expected 1 register call, got %d", len(fc.registerCalls)) + } else if fc.registerCalls[0].PeerAddr != "self:18080" { + t.Errorf("register sent wrong peer addr: %q", fc.registerCalls[0].PeerAddr) + } + fc.mu.Unlock() + + seeds := r.Seeds() + if len(seeds) != 2 { + t.Errorf("expected 2 seeds, got %d", len(seeds)) + } + + owner := r.OwnerFor("3,01637037d6") + if owner != "mount-a:18080" && owner != "mount-b:18080" { + t.Errorf("OwnerFor returned unexpected addr: %q", owner) + } +} + +func TestPeerRegistrar_HeartbeatTTLMatchesConfig(t *testing.T) { + fc := &fakeFilerClient{} + filers, dial := singleFilerFleet(fc) + r := NewPeerRegistrar(filers, dial, "self:18080", "", "") + + if err := r.registerOnce(context.Background()); err != nil { + t.Fatalf("registerOnce: %v", err) + } + + fc.mu.Lock() + defer fc.mu.Unlock() + if len(fc.registerCalls) != 1 { + t.Fatalf("expected 1 register call, got %d", len(fc.registerCalls)) + } + want := int32(r.registerTTL.Seconds()) + if got := fc.registerCalls[0].TtlSeconds; got != want { + t.Errorf("TtlSeconds: got %d want %d", got, want) + } +} + +func TestPeerRegistrar_StopIsIdempotent(t *testing.T) { + filers, dial := singleFilerFleet(&fakeFilerClient{}) + r := NewPeerRegistrar(filers, dial, "self:18080", "", "") + r.Stop() + r.Stop() // second call must be a no-op (no panic) +} + +// TestPeerRegistrar_RegisterBroadcastsToAllFilers guards the core +// multi-filer property: a single registerOnce must hit every configured +// filer so mounts pointing at different filers still converge. +func TestPeerRegistrar_RegisterBroadcastsToAllFilers(t *testing.T) { + fc1 := &fakeFilerClient{} + fc2 := &fakeFilerClient{} + fc3 := &fakeFilerClient{} + a1, a2, a3 := pb.ServerAddress("f1:18888"), pb.ServerAddress("f2:18888"), pb.ServerAddress("f3:18888") + fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{a1: fc1, a2: fc2, a3: fc3}} + + r := NewPeerRegistrar([]pb.ServerAddress{a1, a2, a3}, fleet.dial, "self:18080", "", "") + if err := r.registerOnce(context.Background()); err != nil { + t.Fatalf("registerOnce: %v", err) + } + + for addr, fc := range fleet.clients { + fc.mu.Lock() + if len(fc.registerCalls) != 1 { + t.Errorf("filer %s: got %d register calls, want 1", addr, len(fc.registerCalls)) + } + fc.mu.Unlock() + } +} + +// TestPeerRegistrar_ListMergesAcrossFilers guards cross-filer convergence: +// mount A registered on filer-1, mount B on filer-2; a registrar that +// lists both filers must see both mounts. +func TestPeerRegistrar_ListMergesAcrossFilers(t *testing.T) { + fc1 := &fakeFilerClient{ + listResponse: filer_pb.MountListResponse{ + Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-a:18080", Rack: "r1", LastSeenNs: 200}}, + }, + } + fc2 := &fakeFilerClient{ + listResponse: filer_pb.MountListResponse{ + Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-b:18080", Rack: "r2", LastSeenNs: 200}}, + }, + } + a1, a2 := pb.ServerAddress("f1:18888"), pb.ServerAddress("f2:18888") + fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{a1: fc1, a2: fc2}} + + r := NewPeerRegistrar([]pb.ServerAddress{a1, a2}, fleet.dial, "self:18080", "", "") + if err := r.listOnce(context.Background()); err != nil { + t.Fatalf("listOnce: %v", err) + } + + seeds := r.Seeds() + if len(seeds) != 2 { + t.Fatalf("want 2 merged seeds, got %d: %+v", len(seeds), seeds) + } + addrs := map[string]bool{} + for _, s := range seeds { + addrs[s.PeerAddr] = true + } + if !addrs["mount-a:18080"] || !addrs["mount-b:18080"] { + t.Errorf("merged seeds missing an entry: %+v", addrs) + } +} + +// TestPeerRegistrar_ListMergeKeepsNewestLastSeen guards the dedupe rule: +// the same mount reported by two filers collapses to one entry, keeping +// the freshest LastSeenNs for liveness-ordering decisions. +func TestPeerRegistrar_ListMergeKeepsNewestLastSeen(t *testing.T) { + fc1 := &fakeFilerClient{ + listResponse: filer_pb.MountListResponse{ + Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-a:18080", Rack: "r1", LastSeenNs: 100}}, + }, + } + fc2 := &fakeFilerClient{ + listResponse: filer_pb.MountListResponse{ + Mounts: []*filer_pb.MountInfo{{PeerAddr: "mount-a:18080", Rack: "r1", LastSeenNs: 500}}, + }, + } + a1, a2 := pb.ServerAddress("f1:18888"), pb.ServerAddress("f2:18888") + fleet := &fakeFilerFleet{clients: map[pb.ServerAddress]*fakeFilerClient{a1: fc1, a2: fc2}} + + r := NewPeerRegistrar([]pb.ServerAddress{a1, a2}, fleet.dial, "self:18080", "", "") + if err := r.listOnce(context.Background()); err != nil { + t.Fatalf("listOnce: %v", err) + } + seeds := r.Seeds() + if len(seeds) != 1 { + t.Fatalf("want 1 deduped seed, got %d", len(seeds)) + } +} diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 98845b90b..ce7cae26b 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -139,6 +139,7 @@ type WFS struct { hardLinkLockTable *util.LockTable[string] posixLocks *PosixLockTable rdmaClient *RDMAMountClient + peerRegistrar *PeerRegistrar FilerConf *filer.FilerConf filerClient *wdclient.FilerClient // Cached volume location client refreshMu sync.Mutex @@ -336,6 +337,9 @@ func NewSeaweedFileSystem(option *Option) *WFS { if wfs.rdmaClient != nil { wfs.rdmaClient.Close() } + if wfs.peerRegistrar != nil { + wfs.peerRegistrar.Stop() + } }) // Initialize RDMA client if enabled @@ -355,6 +359,36 @@ 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. + if option.PeerEnabled { + selfAddr, err := ResolvePeerAdvertiseAddr(option.PeerListen, option.PeerAdvertise) + if err != nil { + // Downstream code treats PeerEnabled as "peer infrastructure + // is ready": later PRs wire the gRPC server, fetcher hook, + // and announcer from this flag. If we can't resolve a + // reachable self-address those components would nil-deref + // or advertise garbage, so disable the feature instead of + // limping along half-initialized. + glog.Warningf("peer: cannot resolve advertise addr, disabling peer sharing: %v", err) + option.PeerEnabled = false + } else { + dial := func(ctx context.Context, addr pb.ServerAddress, fn func(client filer_pb.SeaweedFilerClient) error) error { + return pb.WithGrpcFilerClient(false, 0, addr, option.GrpcDialOption, fn) + } + wfs.peerRegistrar = NewPeerRegistrar(option.FilerAddresses, dial, selfAddr, option.PeerDataCenter, option.PeerRack) + if err := wfs.peerRegistrar.Start(context.Background()); err != nil { + glog.Warningf("peer registrar start: %v", err) + } + } + } + if wfs.option.ConcurrentWriters > 0 { wfs.concurrentWriters = util.NewLimitedConcurrentExecutor(wfs.option.ConcurrentWriters) wfs.concurrentCopiersSem = make(chan struct{}, wfs.option.ConcurrentWriters)