perf: remove unncessary goroutine pileup of skippeer workers (#4993)

ethersphere · Feb 6, 2025 · 4940ef8 · 4940ef8
1 parent 0752977
commit 4940ef8
Show file tree

Hide file tree

Showing 4 changed files with 50 additions and 11 deletions.
diff --git a/pkg/pushsync/pushsync.go b/pkg/pushsync/pushsync.go
@@ -144,7 +144,7 @@ func New(
 		metrics:                 newMetrics(),
 		tracer:                  tracer,
 		signer:                  signer,
-		errSkip:                 skippeers.NewList(),
+		errSkip:                 skippeers.NewList(time.Minute),
 		warmupPeriod:            time.Now().Add(warmupTime),
 		shallowReceiptTolerance: shallowReceiptTolerance,
 	}
@@ -387,7 +387,7 @@ func (ps *PushSync) pushToClosest(ctx context.Context, ch swarm.Chunk, origin bo
 		return nil, fmt.Errorf("pushsync: storage radius: %w", err)
 	}
 
-	skip := skippeers.NewList()
+	skip := skippeers.NewList(0)
 	defer skip.Close()
 
 	for sentErrorsLeft > 0 {

diff --git a/pkg/retrieval/retrieval.go b/pkg/retrieval/retrieval.go
@@ -104,7 +104,7 @@ func New(
 		metrics:       newMetrics(),
 		tracer:        tracer,
 		caching:       forwarderCaching,
-		errSkip:       skippeers.NewList(),
+		errSkip:       skippeers.NewList(time.Minute),
 	}
 }
 
@@ -158,7 +158,7 @@ func (s *Service) RetrieveChunk(ctx context.Context, chunkAddr, sourcePeerAddr s
 
 	v, _, err := s.singleflight.Do(ctx, flightRoute, func(ctx context.Context) (swarm.Chunk, error) {
 
-		skip := skippeers.NewList()
+		skip := skippeers.NewList(0)
 		defer skip.Close()
 
 		var preemptiveTicker <-chan time.Time

diff --git a/pkg/skippeers/skippeers.go b/pkg/skippeers/skippeers.go
@@ -25,24 +25,26 @@ type List struct {
 	wg sync.WaitGroup
 }
 
-func NewList() *List {
+func NewList(workerWakeUpDur time.Duration) *List {
 	l := &List{
 		skip: make(map[string]map[string]int64),
 		durC: make(chan time.Duration),
 		quit: make(chan struct{}),
 	}
 
-	l.wg.Add(1)
-	go l.worker()
+	if workerWakeUpDur > 0 {
+		l.wg.Add(1)
+		go l.worker(workerWakeUpDur)
+	}
 
 	return l
 }
 
-func (l *List) worker() {
+func (l *List) worker(d time.Duration) {
 
 	defer l.wg.Done()
 
-	ticker := time.NewTicker(time.Minute)
+	ticker := time.NewTicker(d)
 	defer ticker.Stop()
 
 	for {

diff --git a/pkg/skippeers/skippeers_test.go b/pkg/skippeers/skippeers_test.go
@@ -15,7 +15,7 @@ import (
 func TestPruneExpiresAfter(t *testing.T) {
 	t.Parallel()
 
-	skipList := skippeers.NewList()
+	skipList := skippeers.NewList(0)
 	t.Cleanup(func() { skipList.Close() })
 
 	chunk := swarm.RandAddress(t)
@@ -58,7 +58,7 @@ func TestPruneExpiresAfter(t *testing.T) {
 func TestPeerWait(t *testing.T) {
 	t.Parallel()
 
-	skipList := skippeers.NewList()
+	skipList := skippeers.NewList(0)
 	t.Cleanup(func() { skipList.Close() })
 
 	chunk1 := swarm.RandAddress(t)
@@ -105,3 +105,40 @@ func TestPeerWait(t *testing.T) {
 		t.Fatal("entry should be pruned")
 	}
 }
+
+func TestPruneWorker(t *testing.T) {
+	t.Parallel()
+
+	chunk1 := swarm.RandAddress(t)
+	chunk2 := swarm.RandAddress(t)
+	peer1 := swarm.RandAddress(t)
+	peer2 := swarm.RandAddress(t)
+
+	skipList := skippeers.NewList(time.Millisecond * 500)
+	t.Cleanup(func() { skipList.Close() })
+
+	skipList.Add(chunk1, peer1, time.Second)
+	if !swarm.ContainsAddress(skipList.ChunkPeers(chunk1), peer1) {
+		t.Fatal("peer should be in skiplist")
+	}
+
+	skipList.Add(chunk2, peer1, time.Second)
+	if !swarm.ContainsAddress(skipList.ChunkPeers(chunk2), peer1) {
+		t.Fatal("peer should be in skiplist")
+	}
+
+	skipList.Add(chunk2, peer2, time.Minute)
+	if !swarm.ContainsAddress(skipList.ChunkPeers(chunk2), peer1) {
+		t.Fatal("peer should be in skiplist")
+	}
+
+	time.Sleep(time.Millisecond * 1600)
+
+	if len(skipList.ChunkPeers(chunk1)) != 0 {
+		t.Fatal("entry should be pruned")
+	}
+
+	if len(skipList.ChunkPeers(chunk2)) != 1 || !swarm.ContainsAddress(skipList.ChunkPeers(chunk2), peer2) {
+		t.Fatal("peer2 should be in skiplist")
+	}
+}