feat(scheduler): use sharded keymutex in peer manager

- Replace single `sync.Mutex` with `keymutex.KeyMutex` (256 shards)
- Lock/unlock per peer ID instead of globally, reducing contention
- Add `k8s.io/utils` dependency for `keymutex` package

Signed-off-by: Gaius <gaius.qi@gmail.com>

Author: gaius-qi

go.mod

@@ -79,6 +79,7 @@ require (
 	gorm.io/driver/postgres v1.4.8
 	gorm.io/gorm v1.25.0
 	gorm.io/plugin/soft_delete v1.2.1
+	k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3
 	moul.io/zapgorm2 v1.3.0
 	stathat.com/c/consistent v1.0.0
 )

go.sum

@@ -1379,6 +1379,8 @@ honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
 honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k=
+k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3 h1:jVkFFVfXdXP74B/zbO3hM3hpSFD0xvhQ5U686DPurkE=
+k8s.io/utils v0.0.0-20260707023825-cf1189d6abe3/go.mod h1:M2s5JB1lIYP3jzZdorPLHXIPJzt9vv2muW5a6L9DtNM=
 moul.io/zapgorm2 v1.3.0 h1:+CzUTMIcnafd0d/BvBce8T4uPn6DQnpIrz64cyixlkk=
 moul.io/zapgorm2 v1.3.0/go.mod h1:nPVy6U9goFKHR4s+zfSo1xVFaoU7Qgd5DoCdOfzoCqs=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=

scheduler/resource/standard/peer_manager.go

@@ -23,6 +23,8 @@ import (
 	"sync"
 	"time"
 
+	"k8s.io/utils/keymutex"
+
 	pkggc "d7y.io/dragonfly/v2/pkg/gc"
 	"d7y.io/dragonfly/v2/scheduler/config"
 )
@@ -32,6 +34,9 @@ const (
 	GCPeerID = "peer"
 )
 
+// peerMutexShardCount is the number of mutex shards for peer mutations.
+const peerMutexShardCount = 256
+
 // PeerManager is the interface used for peer manager.
 type PeerManager interface {
 	// Load returns peer for a key.
@@ -73,8 +78,8 @@ type peerManager struct {
 	// pieceDownloadTimeout is timeout of downloading piece.
 	pieceDownloadTimeout time.Duration
 
-	// mu is peer mutex.
-	mu *sync.Mutex
+	// mu is a mutex for peer mutations.
+	mu keymutex.KeyMutex
 }
 
 // New peer manager interface.
@@ -84,7 +89,7 @@ func newPeerManager(cfg *config.GCConfig, gc pkggc.GC) (PeerManager, error) {
 		peerTTL:              cfg.PeerTTL,
 		hostTTL:              cfg.HostTTL,
 		pieceDownloadTimeout: cfg.PieceDownloadTimeout,
-		mu:                   &sync.Mutex{},
+		mu:                   keymutex.NewHashed(peerMutexShardCount),
 	}
 
 	if err := gc.Add(pkggc.Task{
@@ -111,8 +116,8 @@ func (p *peerManager) Load(key string) (*Peer, bool) {
 
 // Store sets peer.
 func (p *peerManager) Store(peer *Peer) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	p.mu.LockKey(peer.ID)
+	defer p.mu.UnlockKey(peer.ID) // nolint: errcheck
 
 	p.Map.Store(peer.ID, peer)
 	peer.Task.StorePeer(peer)
@@ -123,8 +128,8 @@ func (p *peerManager) Store(peer *Peer) {
 // Otherwise, it stores and returns the given peer.
 // The loaded result is true if the peer was loaded, false if stored.
 func (p *peerManager) LoadOrStore(peer *Peer) (*Peer, bool) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	p.mu.LockKey(peer.ID)
+	defer p.mu.UnlockKey(peer.ID) // nolint: errcheck
 
 	rawPeer, loaded := p.Map.LoadOrStore(peer.ID, peer)
 	if !loaded {
@@ -137,8 +142,8 @@ func (p *peerManager) LoadOrStore(peer *Peer) (*Peer, bool) {
 
 // Delete deletes peer for a key.
 func (p *peerManager) Delete(key string) {
-	p.mu.Lock()
-	defer p.mu.Unlock()
+	p.mu.LockKey(key)
+	defer p.mu.UnlockKey(key) // nolint: errcheck
 
 	if peer, loaded := p.Load(key); loaded {
 		p.Map.Delete(key)