chore(broadcast): cleanup of unused code

Author: aleksander-vedvik

broadcast.go

@@ -36,10 +36,6 @@ type broadcastServer struct {
 	logger            *slog.Logger
 }
 
-func (srv *Server) GetStats() broadcast.Metrics {
-	return srv.broadcastSrv.manager.GetStats()
-}
-
 func newBroadcastServer(serverOpts *serverOptions) *broadcastServer {
 	h := fnv.New32a()
 	_, _ = h.Write([]byte(serverOpts.listenAddr))

broadcast/manager.go

@@ -20,7 +20,6 @@ type Manager interface {
 	AddHandler(method string, handler any)
 	Close() error
 	ResetState()
-	GetStats() Metrics
 }
 
 type manager struct {
@@ -163,20 +162,3 @@ func (mgr *manager) Close() error {
 func (mgr *manager) ResetState() {
 	mgr.state.reset()
 }
-
-func (mgr *manager) GetStats() Metrics {
-	m := mgr.state.getStats()
-	return Metrics{
-		TotalNum: uint64(m.totalMsgs),
-		Dropped:  m.droppedMsgs,
-		FinishedReqs: struct {
-			Total     uint64
-			Succesful uint64
-			Failed    uint64
-		}{
-			Total:     m.numReqs,
-			Succesful: m.finishedReqs,
-			Failed:    m.numReqs - m.finishedReqs,
-		},
-	}
-}

broadcast/metrics.go

@@ -11,25 +11,10 @@ import (
 
 const NumShards = 16
 
-type shardMetrics struct {
-	totalMsgs            uint64
-	numMsgs              uint64
-	droppedMsgs          uint64
-	numBroadcastMsgs     uint64
-	droppedBroadcastMsgs uint64
-	numReqs              uint64
-	finishedReqs         uint64
-	lifetimes            [][]time.Time
-	avgLifetime          time.Duration
-	minLifetime          time.Duration
-	maxLifetime          time.Duration
-}
-
 type shard struct {
 	mut         sync.RWMutex
 	id          int
 	parentCtx   context.Context
-	metrics     shardMetrics
 	procs       map[uint64]*BroadcastProcessor
 	reqTTL      time.Duration
 	router      Router
@@ -62,7 +47,6 @@ func createShards(ctx context.Context, shardBuffer, sendBuffer int, router Route
 }
 
 func (s *shard) handleMsg(msg *Content) shardResponse {
-	//s.metrics.numMsgs++
 	// Optimization: first check with a read lock if the processor already exists
 	if p, ok := s.getProcessor(msg.BroadcastID); ok {
 		return s.process(p, msg)
@@ -120,7 +104,6 @@ func (s *shard) process(p *BroadcastProcessor, msg *Content) shardResponse {
 	// must check if the req is done to prevent deadlock
 	select {
 	case <-p.ctx.Done():
-		//s.metrics.droppedMsgs++
 		return shardResponse{
 			err: AlreadyProcessedErr{},
 		}
@@ -138,11 +121,9 @@ func (s *shard) process(p *BroadcastProcessor, msg *Content) shardResponse {
 }
 
 func (s *shard) handleBMsg(msg *Msg) {
-	//s.metrics.numBroadcastMsgs++
 	if req, ok := s.getProcessor(msg.BroadcastID); ok {
 		select {
 		case <-req.ctx.Done():
-			//s.metrics.droppedBroadcastMsgs++
 		case req.broadcastChan <- msg:
 		}
 	}
@@ -178,7 +159,6 @@ func (s *shard) addProcessor(msg *Content) (*BroadcastProcessor, bool) {
 	// check size of s.reqs. If too big, then perform necessary cleanup.
 	// should only affect the current shard and not the others.
 	ctx, cancel := context.WithTimeout(s.parentCtx, s.reqTTL)
-	//req := &BroadcastRequest{
 	var logger *slog.Logger
 	if s.logger != nil {
 		logger = s.logger.With(logging.BroadcastID(msg.BroadcastID))
@@ -229,38 +209,3 @@ func (s *shard) gc(nextGC time.Duration) {
 	}
 	s.procs = newReqs
 }
-
-func (s *shard) getStats() shardMetrics {
-	s.mut.RLock()
-	defer s.mut.RUnlock()
-	s.metrics.numReqs = uint64(len(s.procs))
-	s.metrics.lifetimes = make([][]time.Time, len(s.procs))
-	s.metrics.totalMsgs = s.metrics.numBroadcastMsgs + s.metrics.numMsgs
-	minLifetime := 100 * time.Hour
-	maxLifetime := time.Duration(0)
-	totalLifetime := time.Duration(0)
-	i := 0
-	for _, req := range s.procs {
-		select {
-		case <-req.ctx.Done():
-			s.metrics.finishedReqs++
-		default:
-		}
-		s.metrics.lifetimes[i] = []time.Time{req.started, req.ended}
-		lifetime := req.ended.Sub(req.started)
-		if lifetime > 0 {
-			if lifetime < minLifetime {
-				minLifetime = lifetime
-			}
-			if lifetime > maxLifetime {
-				maxLifetime = lifetime
-			}
-			totalLifetime += lifetime
-		}
-		i++
-	}
-	s.metrics.minLifetime = minLifetime
-	s.metrics.maxLifetime = maxLifetime
-	s.metrics.avgLifetime = totalLifetime
-	return s.metrics
-}

broadcast/shard.go

@@ -96,30 +96,6 @@ func (s *BroadcastState) getShard(i uint16) *shard {
 	return s.shards[i]
 }
 
-func (state *BroadcastState) getStats() shardMetrics {
-	m := shardMetrics{
-		lifetimes: make([][]time.Time, 0),
-	}
-	for _, shard := range state.shards {
-		metric := shard.getStats()
-		m.totalMsgs += metric.totalMsgs
-		m.numMsgs += metric.numMsgs
-		m.droppedMsgs += metric.droppedMsgs
-		m.numBroadcastMsgs += metric.numBroadcastMsgs
-		m.droppedBroadcastMsgs += metric.droppedBroadcastMsgs
-		m.numReqs += metric.numReqs
-		m.finishedReqs += metric.finishedReqs
-		m.lifetimes = append(m.lifetimes, metric.lifetimes...)
-		m.avgLifetime += metric.avgLifetime
-		m.maxLifetime += metric.maxLifetime
-		m.minLifetime += metric.minLifetime
-	}
-	if m.numReqs > 0 {
-		m.avgLifetime /= time.Duration(m.numReqs)
-	}
-	return m
-}
-
 type shardResponse struct {
 	err              error
 	reqCtx           context.Context

broadcast/state.go

@@ -147,7 +147,7 @@ func TestSimpleBroadcastTo(t *testing.T) {
 
 func TestSimpleBroadcastCancel(t *testing.T) {
 	numSrvs := 3
-	numReqs := 1000
+	numReqs := 10
 	_, srvAddrs, srvCleanup, err := createSrvs(numSrvs)
 	if err != nil {
 		t.Error(err)