fix: measuring total sent and received bytes

sync/handler_blocks_response_test.go

@@ -133,8 +133,8 @@ func TestSyncing(t *testing.T) {
 	shouldPublishMessageWithThisType(t, networkBob, message.MessageTypeHello)
 
 	// Ensure peers are connected and block heights are correct
-	assert.Len(t, syncAlice.Peers(), 1)
-	assert.Len(t, syncBob.Peers(), 1)
+	assert.Equal(t, syncAlice.PeerSet().Len(), 1)
+	assert.Equal(t, syncBob.PeerSet().Len(), 1)
 	assert.Equal(t, syncAlice.state.LastBlockHeight(), uint32(0))
 	assert.Equal(t, syncBob.state.LastBlockHeight(), uint32(100))
 

sync/interface.go

@@ -10,6 +10,6 @@ type Synchronizer interface {
 	Stop()
 	Moniker() string
 	SelfID() peer.ID
-	Peers() []peerset.Peer
+	PeerSet() *peerset.PeerSet
 	Fingerprint() string
 }

sync/mock.go

@@ -14,8 +14,8 @@ import (
 var _ Synchronizer = &MockSync{}
 
 type MockSync struct {
-	ID      peer.ID
-	PeerSet *peerset.PeerSet
+	TestID      peer.ID
+	TestPeerSet *peerset.PeerSet
 }
 
 func MockingSync() *MockSync {
@@ -43,8 +43,8 @@ func MockingSync() *MockSync {
 	ps.UpdateHeight(pid1, util.RandUint32(100000))
 
 	return &MockSync{
-		ID:      network.TestRandomPeerID(),
-		PeerSet: ps,
+		TestID:      network.TestRandomPeerID(),
+		TestPeerSet: ps,
 	}
 }
 
@@ -58,12 +58,12 @@ func (m *MockSync) Fingerprint() string {
 }
 
 func (m *MockSync) SelfID() peer.ID {
-	return m.ID
+	return m.TestID
 }
 
 func (m *MockSync) Moniker() string {
 	return "test-moniker"
 }
-func (m *MockSync) Peers() []peerset.Peer {
-	return m.PeerSet.GetPeerList()
+func (m *MockSync) PeerSet() *peerset.PeerSet {
+	return m.TestPeerSet
 }

sync/peerset/peer_set.go

@@ -16,18 +16,22 @@
 type PeerSet struct {
 	lk sync.RWMutex
 
-	peers            map[peer.ID]*Peer
-	sessions         map[int]*Session
-	nextSessionID    int
-	maxClaimedHeight uint32
-	sessionTimeout   time.Duration
+	peers              map[peer.ID]*Peer
+	sessions           map[int]*Session
+	nextSessionID      int
+	maxClaimedHeight   uint32
+	sessionTimeout     time.Duration
+	totalSentBytes     int
+	totalReceivedBytes int
+	startedAt          time.Time
 }
 
 func NewPeerSet(sessionTimeout time.Duration) *PeerSet {
 	return &PeerSet{
 		peers:          make(map[peer.ID]*Peer),
 		sessions:       make(map[int]*Session),
 		sessionTimeout: sessionTimeout,
+		startedAt:      time.Now(),
 	}
 }
 
@@ -298,6 +302,14 @@
 
 	p := ps.mustGetPeer(pid)
 	p.ReceivedBytes += c
+	ps.totalReceivedBytes += c
+}
+
+func (ps *PeerSet) IncreaseTotalSentBytesCounter(c int) {
+	ps.lk.Lock()
+	defer ps.lk.Unlock()
+
+	ps.totalSentBytes += c
 }
 
 func (ps *PeerSet) IncreaseSendSuccessCounter(pid peer.ID) {
@@ -315,3 +327,24 @@
 	p := ps.mustGetPeer(pid)
 	p.SendFailed++
 }
+
+func (ps *PeerSet) TotalSentBytes() int {
+	ps.lk.RLock()
+	defer ps.lk.RUnlock()
+
+	return ps.totalSentBytes
+}
+
+func (ps *PeerSet) TotalReceivedBytes() int {
+	ps.lk.RLock()
+	defer ps.lk.RUnlock()
+
+	return ps.totalReceivedBytes
+}
+
+func (ps *PeerSet) StartedAt() time.Time {
+	ps.lk.RLock()
+	defer ps.lk.RUnlock()
+
+	return ps.startedAt
+}

sync/peerset/peer_set_test.go

@@ -70,6 +70,7 @@ func TestPeerSet(t *testing.T) {
 		peerSet.IncreaseInvalidBundlesCounter(pid1)
 		peerSet.IncreaseReceivedBundlesCounter(pid1)
 		peerSet.IncreaseReceivedBytesCounter(pid1, 100)
+		peerSet.IncreaseTotalSentBytesCounter(200)
 		peerSet.IncreaseSendFailedCounter(pid1)
 		peerSet.IncreaseSendSuccessCounter(pid1)
 
@@ -79,6 +80,8 @@ func TestPeerSet(t *testing.T) {
 		assert.Equal(t, peer1.ReceivedBytes, 100)
 		assert.Equal(t, peer1.SendFailed, 1)
 		assert.Equal(t, peer1.SendSuccess, 1)
+		assert.Equal(t, peerSet.TotalReceivedBytes(), 100)
+		assert.Equal(t, peerSet.TotalSentBytes(), 200)
 	})
 
 	t.Run("Testing UpdateStatus", func(t *testing.T) {

sync/sync.go

@@ -317,6 +317,7 @@ func (sync *synchronizer) sendTo(msg message.Message, to peer.ID, sessionID int)
 			sync.logger.Info("sending bundle to a peer", "bundle", bdl, "to", to)
 			sync.peerSet.IncreaseSendSuccessCounter(to)
 		}
+		sync.peerSet.IncreaseTotalSentBytesCounter(len(data))
 	}
 }
 
@@ -332,6 +333,7 @@ func (sync *synchronizer) broadcast(msg message.Message) {
 		} else {
 			sync.logger.Info("broadcasting new bundle", "bundle", bdl)
 		}
+		sync.peerSet.IncreaseTotalSentBytesCounter(len(data))
 	}
 }
 
@@ -343,8 +345,8 @@ func (sync *synchronizer) Moniker() string {
 	return sync.config.Moniker
 }
 
-func (sync *synchronizer) Peers() []peerset.Peer {
-	return sync.peerSet.GetPeerList()
+func (sync *synchronizer) PeerSet() *peerset.PeerSet {
+	return sync.peerSet
 }
 
 // downloadBlocks starts downloading blocks from the network.

www/grpc/blockchain.go

@@ -191,8 +191,6 @@ func (s *blockchainServer) GetValidatorByNumber(_ context.Context,
 		return nil, status.Errorf(codes.NotFound, "validator not found")
 	}
 
-	// TODO: make a function
-	// proto validator from native validator
 	return &pactus.GetValidatorResponse{
 		Validator: validatorToProto(val),
 	}, nil
@@ -209,8 +207,6 @@ func (s *blockchainServer) GetValidator(_ context.Context,
 		return nil, status.Errorf(codes.NotFound, "validator not found")
 	}
 
-	// TODO: make a function
-	// proto validator from native validator
 	return &pactus.GetValidatorResponse{
 		Validator: validatorToProto(val),
 	}, nil