diff --git a/go.mod b/go.mod index d7bb6b188..a9dabeb03 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,7 @@ module github.com/libp2p/go-libp2p-kad-dht go 1.25.7 require ( + github.com/gammazero/cascadeq v0.2.0 github.com/gammazero/deque v1.2.1 github.com/google/gopacket v1.1.19 github.com/google/uuid v1.6.0 @@ -13,7 +14,6 @@ require ( github.com/ipfs/go-datastore v0.9.2 github.com/ipfs/go-detect-race v0.0.1 github.com/ipfs/go-ds-pebble v0.5.12 - github.com/ipfs/go-dsqueue v0.2.0 github.com/ipfs/go-libdht v0.5.0 github.com/ipfs/go-log/v2 v2.9.2 github.com/ipfs/go-test v0.4.1 @@ -63,13 +63,13 @@ require ( github.com/dunglas/httpsfv v1.1.0 // indirect github.com/filecoin-project/go-clock v0.1.0 // indirect github.com/flynn/noise v1.1.0 // indirect + github.com/gammazero/fsutil v0.2.0 // indirect github.com/getsentry/sentry-go v0.27.0 // indirect github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect github.com/gogo/protobuf v1.3.2 // indirect github.com/golang/snappy v0.0.5-0.20231225225746-43d5d4cd4e0e // indirect github.com/gorilla/websocket v1.5.3 // indirect - github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect github.com/huin/goupnp v1.3.0 // indirect github.com/ipfs/go-block-format v0.2.4 // indirect github.com/ipld/go-ipld-prime v0.24.0 // indirect diff --git a/go.sum b/go.sum index c95d4ca70..b1fc06cf9 100644 --- a/go.sum +++ b/go.sum @@ -81,8 +81,12 @@ github.com/flynn/noise v1.1.0/go.mod h1:xbMo+0i6+IGbYdJhF31t2eR1BIU0CYc12+BNAKwU github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8= github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= +github.com/gammazero/cascadeq v0.2.0 h1:LHq6hWLQvQCHVtCl5hDWYSavqXRFCKZ8upCdRtsPpfo= +github.com/gammazero/cascadeq v0.2.0/go.mod h1:aEkjsO3wVhT09lheCjiKbEmT7GtGa03WXGmclOZp5wY= github.com/gammazero/deque v1.2.1 h1:9fnQVFCCZ9/NOc7ccTNqzoKd1tCWOqeI05/lPqFPMGQ= github.com/gammazero/deque v1.2.1/go.mod h1:5nSFkzVm+afG9+gy0VIowlqVAW4N8zNcMne+CMQVD2g= +github.com/gammazero/fsutil v0.2.0 h1:/MqQHCBoT07KGY75avKqG/SI0zS693zr1Ljx8cwKWhs= +github.com/gammazero/fsutil v0.2.0/go.mod h1:UhNgS1Hr75DBX6zqBEOB4AAZQiNytnr3Mc0ZpiLKPz4= github.com/getsentry/sentry-go v0.27.0 h1:Pv98CIbtB3LkMWmXi4Joa5OOcwbmnX88sF5qbK3r3Ps= github.com/getsentry/sentry-go v0.27.0/go.mod h1:lc76E2QywIyW8WuBnwl8Lc4bkmQH4+w1gwTf25trprY= github.com/ghemawat/stream v0.0.0-20171120220530-696b145b53b9 h1:r5GgOLGbza2wVHRzK7aAj6lWZjfbAwiu/RDCVOKjRyM= @@ -127,8 +131,6 @@ github.com/gxed/hashland/murmur3 v0.0.1/go.mod h1:KjXop02n4/ckmZSnY2+HKcLud/tcmv github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8= github.com/hashicorp/golang-lru v1.0.2 h1:dV3g9Z/unq5DpblPpw+Oqcv4dU/1omnb4Ok8iPY6p1c= github.com/hashicorp/golang-lru v1.0.2/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4= -github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k= -github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM= github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/huin/goupnp v1.3.0 h1:UvLUlWDNpoUdYzb2TCn+MuTWtcjXKSza2n6CBdQ0xXc= @@ -149,12 +151,8 @@ github.com/ipfs/go-detect-race v0.0.1 h1:qX/xay2W3E4Q1U7d9lNs1sU9nvguX0a7319XbyQ github.com/ipfs/go-detect-race v0.0.1/go.mod h1:8BNT7shDZPo99Q74BpGMK+4D8Mn4j46UU0LZ723meps= github.com/ipfs/go-ds-badger v0.0.7/go.mod h1:qt0/fWzZDoPW6jpQeqUjR5kBfhDNB65jd9YlmAvpQBk= github.com/ipfs/go-ds-leveldb v0.1.0/go.mod h1:hqAW8y4bwX5LWcCtku2rFNX3vjDZCy5LZCg+cSZvYb8= -github.com/ipfs/go-ds-leveldb v0.5.2 h1:6nmxlQ2zbp4LCNdJVsmHfs9GP0eylfBNxpmY1csp0x0= -github.com/ipfs/go-ds-leveldb v0.5.2/go.mod h1:2fAwmcvD3WoRT72PzEekHBkQmBDhc39DJGoREiuGmYo= github.com/ipfs/go-ds-pebble v0.5.12 h1:idO/w4i3IBA6vZtVWsyG5IlPIgwd62iUaQZBl/Kv+yI= github.com/ipfs/go-ds-pebble v0.5.12/go.mod h1:H2zy28KMQSiAflUxpKzKHqbpSHRWPZS5/bi4ymAJOjY= -github.com/ipfs/go-dsqueue v0.2.0 h1:MBi9w3oSiX98Xc+Y7NuJ9G8MI6mAT4IGdO9dHEMCZzU= -github.com/ipfs/go-dsqueue v0.2.0/go.mod h1:8FfNQC4DMF/KkzBXRNB9Rb3MKDW0Sh98HMtXYl1mLQE= github.com/ipfs/go-ipfs-delay v0.0.0-20181109222059-70721b86a9a8/go.mod h1:8SP1YXK1M1kXuc4KJZINY3TQQ03J2rwBG9QfXmbRPrw= github.com/ipfs/go-ipfs-util v0.0.1/go.mod h1:spsl5z8KUnrve+73pOhSVZND1SIxPW5RyBCNzQxlJBc= github.com/ipfs/go-libdht v0.5.0 h1:ZN+eCqwahZvUeT0e4DsIxRtm78Mc9UR5tmZUiMsrGjQ= @@ -398,7 +396,6 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= -github.com/syndtr/goleveldb v1.0.0 h1:fBdIW9lB4Iz0n9khmH8w27SJ3QEJ7+IgjPEwGSZiFdE= github.com/syndtr/goleveldb v1.0.0/go.mod h1:ZVVdQEZoIme9iO1Ch2Jdy24qqXrMMOU6lpPAyBWyWuQ= github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0= github.com/wangjia184/sortedset v0.0.0-20160527075905-f5d03557ba30/go.mod h1:YkocrP2K2tcw938x9gCOmT5G5eCD6jsTz0SZuyAqwIE= diff --git a/provider/buffered/provider.go b/provider/buffered/provider.go index f7e7a454b..f83367069 100644 --- a/provider/buffered/provider.go +++ b/provider/buffered/provider.go @@ -4,8 +4,8 @@ import ( "errors" "sync" + "github.com/gammazero/cascadeq" "github.com/ipfs/go-datastore" - "github.com/ipfs/go-dsqueue" "github.com/ipfs/go-log/v2" "github.com/libp2p/go-libp2p-kad-dht/provider/internal" mh "github.com/multiformats/go-multihash" @@ -36,7 +36,7 @@ type SweepingProvider struct { newItems chan struct{} Provider internal.Provider - queue *dsqueue.DSQueue + queue *cascadeq.Queue batchSize int logger *log.ZapEventLogger } @@ -44,23 +44,26 @@ type SweepingProvider struct { // New creates a new SweepingProvider that wraps the given provider with // buffering capabilities. Operations are queued and processed asynchronously // in batches for improved performance. -func New(prov internal.Provider, ds datastore.Batching, opts ...Option) *SweepingProvider { +func New(prov internal.Provider, ds datastore.Batching, queuePath string, opts ...Option) (*SweepingProvider, error) { cfg := getOpts(opts) + + queue, err := cascadeq.New(queuePath, + cascadeq.WithSnapshotInterval(cfg.idleWriteTime)) + if err != nil { + return nil, err + } s := &SweepingProvider{ done: make(chan struct{}), closed: make(chan struct{}), - newItems: make(chan struct{}, 1), - Provider: prov, - queue: dsqueue.New(ds, cfg.dsName, - dsqueue.WithDedupCacheSize(0), // disable deduplication - dsqueue.WithIdleWriteTime(cfg.idleWriteTime), - ), + newItems: make(chan struct{}, 1), + Provider: prov, + queue: queue, batchSize: cfg.batchSize, logger: log.Logger(cfg.loggerName), } go s.worker() - return s + return s, nil } // Close stops the provider and releases all resources. @@ -136,6 +139,8 @@ func (s *SweepingProvider) executeOperation(f func(...mh.Multihash) error, keys // It runs in a separate goroutine and continues until the provider is closed. func (s *SweepingProvider) worker() { defer close(s.done) + res := make([][]byte, s.batchSize) + var emptyQueue bool for { if emptyQueue { @@ -154,16 +159,15 @@ func (s *SweepingProvider) worker() { } } - res, err := s.queue.GetN(s.batchSize) - if err != nil { - s.logger.Warnf("BufferedSweepingProvider unable to dequeue: %v", err) - continue - } - if len(res) < s.batchSize { + n := s.queue.Drain(res) + if n < s.batchSize { // Queue was fully drained. emptyQueue = true + if n == 0 { + continue + } } - ops, err := getOperations(res) + ops, err := getOperations(res[:n]) if err != nil { s.logger.Warnf("BufferedSweepingProvider unable to parse dequeued item: %v", err) continue @@ -187,11 +191,16 @@ func (s *SweepingProvider) worker() { // enqueue adds operations to the queue for asynchronous processing. func (s *SweepingProvider) enqueue(op byte, keys ...mh.Multihash) error { + data := make([][]byte, 0, len(keys)) for _, h := range keys { - if err := s.queue.Put(toBytes(op, h)); err != nil { - return err - } + data = append(data, toBytes(op, h)) } + + err := s.queue.PutBatch(data) + if err != nil { + return err + } + select { case s.newItems <- struct{}{}: default: @@ -242,12 +251,11 @@ func (s *SweepingProvider) StopProviding(keys ...mh.Multihash) error { return s.enqueue(stopProvidingOp, keys...) } -// Clear clears the all the keys from the provide queue and returns the number -// of keys that were cleared. +// Clear clears the all the keys from the provide queue. // // The keys are not deleted from the keystore, so they will continue to be // reprovided as scheduled. -func (s *SweepingProvider) Clear() int { +func (s *SweepingProvider) Clear() error { return s.Provider.Clear() } diff --git a/provider/buffered/provider_test.go b/provider/buffered/provider_test.go index 4d225deec..33107a7b6 100644 --- a/provider/buffered/provider_test.go +++ b/provider/buffered/provider_test.go @@ -78,9 +78,9 @@ func (f *fakeProvider) StopProviding(keys ...mh.Multihash) error { return nil } -func (f *fakeProvider) Clear() int { +func (f *fakeProvider) Clear() error { // Unused - return 0 + return nil } func (f *fakeProvider) RefreshSchedule() error { @@ -103,10 +103,13 @@ func TestQueueingMechanism(t *testing.T) { synctest.Test(t, func(t *testing.T) { fake := newFakeProvider() ds := datastore.NewMapDatastore() - provider := New(fake, ds, + provider, err := New(fake, ds, t.TempDir(), WithDsName("test1"), WithIdleWriteTime(time.Millisecond), WithBatchSize(10)) + if err != nil { + t.Fatal(err) + } defer provider.Close() keys := random.Multihashes(3) @@ -240,19 +243,23 @@ func TestBatchProcessing(t *testing.T) { synctest.Test(t, func(t *testing.T) { fake := newFakeProvider() ds := datastore.NewMapDatastore() - provider := New(fake, ds, + provider, err := New(fake, ds, t.TempDir(), WithDsName("test4"), - WithBatchSize(3), // Process 3 operations at once + WithBatchSize(1000), // Process 1000 operations at once WithIdleWriteTime(time.Second)) + if err != nil { + t.Fatal(err) + } defer provider.Close() - // Queue multiple keys - total of 3 operations (2 from ProvideOnce + 1 from StartProviding) - keys := random.Multihashes(3) + // Queue multiple keys - total of 30000 operations (2 from ProvideOnce + 29998 + // from StartProviding). Should be enough to overflow to disk. + keys := random.Multihashes(30000) if err := provider.ProvideOnce(keys[0], keys[1]); err != nil { t.Fatalf("ProvideOnce failed: %v", err) } - if err := provider.StartProviding(false, keys[2]); err != nil { + if err := provider.StartProviding(false, keys[2:]...); err != nil { t.Fatalf("StartProviding failed: %v", err) } synctest.Wait() @@ -273,8 +280,8 @@ func TestBatchProcessing(t *testing.T) { for _, call := range fake.startProvidingCalls { totalStartProvidingCalls += len(call.keys) } - if totalStartProvidingCalls != 1 { - t.Errorf("Expected 1 total key in StartProviding calls, got %d", totalStartProvidingCalls) + if totalStartProvidingCalls != len(keys)-2 { + t.Errorf("Expected %d total key in StartProviding calls, got %d", len(keys)-2, totalStartProvidingCalls) } }) } diff --git a/provider/dual/provider.go b/provider/dual/provider.go index a257b0feb..6e8b8ea37 100644 --- a/provider/dual/provider.go +++ b/provider/dual/provider.go @@ -208,13 +208,12 @@ func (s *SweepingProvider) StopProviding(keys ...mh.Multihash) error { return nil } -// Clear clears the all the keys from the provide queues of both DHTs and -// returns the number of keys that were cleared (sum of both queues). +// Clear clears the all the keys from the provide queues of both DHTs. // // The keys are not deleted from the keystore, so they will continue to be // reprovided as scheduled. -func (s *SweepingProvider) Clear() int { - return s.LAN.Clear() + s.WAN.Clear() +func (s *SweepingProvider) Clear() error { + return errors.Join(s.LAN.Clear(), s.WAN.Clear()) } // RefreshSchedule scans the Keystore for any keys that are not currently diff --git a/provider/internal/interface.go b/provider/internal/interface.go index cda531c27..7688c70b1 100644 --- a/provider/internal/interface.go +++ b/provider/internal/interface.go @@ -12,7 +12,7 @@ type Provider interface { StartProviding(force bool, keys ...mh.Multihash) error StopProviding(keys ...mh.Multihash) error ProvideOnce(keys ...mh.Multihash) error - Clear() int + Clear() error RefreshSchedule() error Close() error } diff --git a/provider/internal/queue/prefix.go b/provider/internal/queue/prefix.go index e197b1c63..1cf856c1c 100644 --- a/provider/internal/queue/prefix.go +++ b/provider/internal/queue/prefix.go @@ -68,15 +68,12 @@ func (q *prefixQueue) Size() int { return q.queue.Len() } -// Clear removes all keys from the queue and returns the number of keys that -// were removed. -func (q *prefixQueue) Clear() int { - size := q.Size() - +// Clear removes all keys from the queue. +func (q *prefixQueue) Clear() error { q.queue.Clear() *q.prefixes = trie.Trie[bitstr.Key, struct{}]{} - return size + return nil } // removeSuperstrings finds all superstrings of `prefix` in the trie, removes diff --git a/provider/internal/queue/provide.go b/provider/internal/queue/provide.go index fcbe49dc9..38e39e03e 100644 --- a/provider/internal/queue/provide.go +++ b/provider/internal/queue/provide.go @@ -191,17 +191,15 @@ func (q *ProvideQueue) NumRegions() int { return q.queue.Size() } -// Clear removes all keys from the queue and returns the number of keys that -// were removed. -func (q *ProvideQueue) Clear() int { +// Clear removes all keys from the queue. +func (q *ProvideQueue) Clear() error { q.mu.Lock() defer q.mu.Unlock() - size := q.keys.Size() q.queue.Clear() *q.keys = trie.Trie[bit256.Key, mh.Multihash]{} - return size + return nil } // Persist saves the current state of the queue to the provided datastore. diff --git a/provider/internal/queue/provide_test.go b/provider/internal/queue/provide_test.go index afad7dbba..869a514ab 100644 --- a/provider/internal/queue/provide_test.go +++ b/provider/internal/queue/provide_test.go @@ -282,8 +282,7 @@ func TestProvideClearQueue(t *testing.T) { require.Equal(t, q.queue.queue.Len(), len(prefixes)) require.Equal(t, q.Size(), len(prefixes)*nMultihashesPerPrefix) - cleared := q.Clear() - require.Equal(t, len(prefixes)*nMultihashesPerPrefix, cleared) + require.NoError(t, q.Clear()) require.True(t, q.IsEmpty()) require.True(t, q.keys.IsEmptyLeaf()) diff --git a/provider/internal/queue/reprovide.go b/provider/internal/queue/reprovide.go index a67dff0e7..7d9ad8cac 100644 --- a/provider/internal/queue/reprovide.go +++ b/provider/internal/queue/reprovide.go @@ -61,9 +61,8 @@ func (q *ReprovideQueue) Size() int { return q.queue.Size() } -// Clear removes all prefixes from the queue and returns the number of removed -// prefixes. -func (q *ReprovideQueue) Clear() int { +// Clear removes all prefixes from the queue. +func (q *ReprovideQueue) Clear() error { q.mu.Lock() defer q.mu.Unlock() return q.queue.Clear() diff --git a/provider/internal/queue/reprovide_test.go b/provider/internal/queue/reprovide_test.go index 9327dc406..a60e1e0bd 100644 --- a/provider/internal/queue/reprovide_test.go +++ b/provider/internal/queue/reprovide_test.go @@ -104,8 +104,7 @@ func TestReprovideClearQueue(t *testing.T) { q.Enqueue(k) } - cleared := q.Clear() - require.Equal(t, len(keys), cleared) + require.NoError(t, q.Clear()) require.True(t, q.IsEmpty()) require.Equal(t, 0, q.queue.prefixes.Size()) require.Equal(t, 0, q.queue.queue.Len()) diff --git a/provider/provider.go b/provider/provider.go index 2de56af1c..397f060a2 100644 --- a/provider/provider.go +++ b/provider/provider.go @@ -2163,14 +2163,13 @@ func (s *SweepingProvider) StopProviding(keys ...mh.Multihash) error { return err } -// Clear clears the all the keys from the provide queue and returns the number -// of keys that were cleared. +// Clear clears all the keys from the provide queue. // // The keys are not deleted from the keystore, so they will continue to be // reprovided as scheduled. -func (s *SweepingProvider) Clear() int { +func (s *SweepingProvider) Clear() error { if s.closed() { - return 0 + return nil } return s.provideQueue.Clear() } diff --git a/provider/provider_test.go b/provider/provider_test.go index d386b1f4e..87a9ddad6 100644 --- a/provider/provider_test.go +++ b/provider/provider_test.go @@ -1426,7 +1426,7 @@ func TestClose(t *testing.T) { require.ErrorIs(t, err, ErrClosed) err = prov.ProvideOnce(newMh) require.ErrorIs(t, err, ErrClosed) - require.Equal(t, 0, prov.Clear()) + require.NoError(t, prov.Clear()) err = prov.workerPool.Acquire(burstWorker) require.ErrorIs(t, err, reservedpool.ErrClosed)