counting ongoing operations

provider/provider.go

@@ -156,6 +156,7 @@ type SweepingProvider struct {
 	addLocalRecord func(mh.Multihash) error
 
 	provideCounter metric.Int64Counter
+	opStats        operationStats
 }
 
 // New creates a new SweepingProvider instance with the supplied options.
@@ -1115,7 +1116,8 @@ func (s *SweepingProvider) reprovideLateRegions() {
 }
 
 func (s *SweepingProvider) batchProvide(prefix bitstr.Key, keys []mh.Multihash) {
-	if len(keys) == 0 {
+	keyCount := len(keys)
+	if keyCount == 0 {
 		return
 	}
 	addrInfo, ok := s.selfAddrInfo()
@@ -1124,6 +1126,12 @@ func (s *SweepingProvider) batchProvide(prefix bitstr.Key, keys []mh.Multihash)
 		// provider record.
 		return
 	}
+
+	s.opStats.ongoingProvides.start(keyCount)
+	defer func() {
+		s.opStats.ongoingProvides.finish(keyCount)
+	}()
+
 	if len(keys) <= individualProvideThreshold {
 		// Don't fully explore the region, execute simple DHT provides for these
 		// keys. It isn't worth it to fully explore a region for just a few keys.
@@ -1142,6 +1150,8 @@ func (s *SweepingProvider) batchProvide(prefix bitstr.Key, keys []mh.Multihash)
 	// current provide batch.
 	extraKeys := s.provideQueue.DequeueMatching(coveredPrefix)
 	keys = append(keys, extraKeys...)
+	keyCount += len(extraKeys)
+	s.opStats.ongoingProvides.addKeys(len(extraKeys))
 	regions = keyspace.AssignKeysToRegions(regions, keys)
 
 	if !s.provideRegions(regions, addrInfo, false, false) {
@@ -1166,10 +1176,17 @@ func (s *SweepingProvider) batchReprovide(prefix bitstr.Key, periodicReprovide b
 		}
 		return
 	}
-	if len(keys) == 0 {
+	keyCount := len(keys)
+	if keyCount == 0 {
 		logger.Infof("No keys to reprovide for prefix %s", prefix)
 		return
 	}
+
+	s.opStats.ongoingProvides.start(keyCount)
+	defer func() {
+		s.opStats.ongoingProvides.finish(keyCount)
+	}()
+
 	if len(keys) <= individualProvideThreshold {
 		// Don't fully explore the region, execute simple DHT provides for these
 		// keys. It isn't worth it to fully explore a region for just a few keys.
@@ -1213,6 +1230,8 @@ func (s *SweepingProvider) batchReprovide(prefix bitstr.Key, periodicReprovide b
 				s.reschedulePrefix(prefix)
 			}
 		}
+		s.opStats.ongoingProvides.addKeys(len(keys) - keyCount)
+		keyCount = len(keys)
 	}
 	regions = keyspace.AssignKeysToRegions(regions, keys)
 

provider/stats.go

@@ -2,6 +2,7 @@ package provider
 
 import (
 	"context"
+	"sync/atomic"
 	"time"
 
 	"github.com/libp2p/go-libp2p-kad-dht/provider/stats"
@@ -89,9 +90,10 @@ func (s *SweepingProvider) Stats() stats.Stats {
 	}
 
 	ongoingOps := stats.OngoingOperations{
-		RegionProvides:   0, // TODO:
-		KeyProvides:      0, // TODO:
-		RegionReprovides: 0, // TODO:
+		RegionProvides:   int(s.opStats.ongoingProvides.opCount.Load()),
+		KeyProvides:      int(s.opStats.ongoingProvides.keyCount.Load()),
+		RegionReprovides: int(s.opStats.ongoingReprovides.opCount.Load()),
+		KeyReprovides:    int(s.opStats.ongoingReprovides.keyCount.Load()),
 	}
 
 	pastOps := stats.PastOperations{
@@ -118,3 +120,27 @@ func (s *SweepingProvider) Stats() stats.Stats {
 
 	return snapshot
 }
+
+type operationStats struct {
+	ongoingProvides   ongoingOpStats
+	ongoingReprovides ongoingOpStats
+}
+
+type ongoingOpStats struct {
+	opCount  atomic.Int32
+	keyCount atomic.Int32
+}
+
+func (s *ongoingOpStats) start(keyCount int) {
+	s.opCount.Add(1)
+	s.keyCount.Add(int32(keyCount))
+}
+
+func (s *ongoingOpStats) addKeys(keyCount int) {
+	s.keyCount.Add(int32(keyCount))
+}
+
+func (s *ongoingOpStats) finish(keyCount int) {
+	s.opCount.Add(-1)
+	s.keyCount.Add(-int32(keyCount))
+}

provider/stats/types.go

@@ -65,6 +65,7 @@ type OngoingOperations struct {
 	RegionProvides   int `json:"region_provides"`
 	KeyProvides      int `json:"key_provides"`
 	RegionReprovides int `json:"region_reprovides"`
+	KeyReprovides    int `json:"key_reprovides"`
 }
 
 type PastOperations struct {