p2p: pass start time to flowrate and cleanup constructors (#7838)

After poking around #7828, I saw the oppertunity for this cleanup,
which I think is both reasonable on its own, and quite low impact, and
removes the math around process start time.

internal/blocksync/pool.go

@@ -331,8 +331,16 @@ func (pool *BlockPool) SetPeerRange(peerID types.NodeID, base int64, height int6
 		peer.base = base
 		peer.height = height
 	} else {
-		peer = newBPPeer(pool, peerID, base, height)
+		peer = &bpPeer{
-		peer.logger = pool.logger.With("peer", peerID)
+			pool:       pool,
+			id:         peerID,
+			base:       base,
+			height:     height,
+			numPending: 0,
+			logger:     pool.logger.With("peer", peerID),
+			startAt:    time.Now(),
+		}
+
 		pool.peers[peerID] = peer
 	}
 
@@ -490,24 +498,13 @@ type bpPeer struct {
 	recvMonitor *flowrate.Monitor
 
 	timeout *time.Timer
+	startAt time.Time
 
 	logger log.Logger
 }
 
-func newBPPeer(pool *BlockPool, peerID types.NodeID, base int64, height int64) *bpPeer {
-	peer := &bpPeer{
-		pool:       pool,
-		id:         peerID,
-		base:       base,
-		height:     height,
-		numPending: 0,
-		logger:     log.NewNopLogger(),
-	}
-	return peer
-}
-
 func (peer *bpPeer) resetMonitor() {
-	peer.recvMonitor = flowrate.New(time.Second, time.Second*40)
+	peer.recvMonitor = flowrate.New(peer.startAt, time.Second, time.Second*40)
 	initialValue := float64(minRecvRate) * math.E
 	peer.recvMonitor.SetREMA(initialValue)
 }

internal/libs/flowrate/flowrate.go

@@ -14,11 +14,12 @@ import (
 
 // Monitor monitors and limits the transfer rate of a data stream.
 type Monitor struct {
-	mu      sync.Mutex    // Mutex guarding access to all internal fields
+	mu       sync.Mutex    // Mutex guarding access to all internal fields
-	active  bool          // Flag indicating an active transfer
+	active   bool          // Flag indicating an active transfer
-	start   time.Duration // Transfer start time (clock() value)
+	start    time.Duration // Transfer start time (clock() value)
-	bytes   int64         // Total number of bytes transferred
+	pStartAt time.Time     // time of process start
-	samples int64         // Total number of samples taken
+	bytes    int64         // Total number of bytes transferred
+	samples  int64         // Total number of samples taken
 
 	rSample float64 // Most recent transfer rate sample (bytes per second)
 	rEMA    float64 // Exponential moving average of rSample
@@ -45,21 +46,22 @@ type Monitor struct {
 //
 // The default values for sampleRate and windowSize (if <= 0) are 100ms and 1s,
 // respectively.
-func New(sampleRate, windowSize time.Duration) *Monitor {
+func New(startAt time.Time, sampleRate, windowSize time.Duration) *Monitor {
 	if sampleRate = clockRound(sampleRate); sampleRate <= 0 {
 		sampleRate = 5 * clockRate
 	}
 	if windowSize <= 0 {
 		windowSize = 1 * time.Second
 	}
-	now := clock()
+	now := clock(startAt)
 	return &Monitor{
-		active:  true,
+		active:   true,
-		start:   now,
+		start:    now,
-		rWindow: windowSize.Seconds(),
+		rWindow:  windowSize.Seconds(),
-		sLast:   now,
+		sLast:    now,
-		sRate:   sampleRate,
+		sRate:    sampleRate,
-		tLast:   now,
+		tLast:    now,
+		pStartAt: startAt,
 	}
 }
 
@@ -129,7 +131,7 @@ func (m *Monitor) Status() Status {
 	now := m.update(0)
 	s := Status{
 		Active:   m.active,
-		Start:    clockToTime(m.start),
+		Start:    m.pStartAt.Add(m.start),
 		Duration: m.sLast - m.start,
 		Idle:     now - m.tLast,
 		Bytes:    m.bytes,
@@ -222,7 +224,7 @@ func (m *Monitor) update(n int) (now time.Duration) {
 	if !m.active {
 		return
 	}
-	if now = clock(); n > 0 {
+	if now = clock(m.pStartAt); n > 0 {
 		m.tLast = now
 	}
 	m.sBytes += int64(n)

internal/libs/flowrate/util.go

@@ -13,18 +13,9 @@ import (
 // clockRate is the resolution and precision of clock().
 const clockRate = 20 * time.Millisecond
 
-// czero is the process start time rounded down to the nearest clockRate
-// increment.
-var czero = time.Now().Round(clockRate)
-
 // clock returns a low resolution timestamp relative to the process start time.
-func clock() time.Duration {
+func clock(startAt time.Time) time.Duration {
-	return time.Now().Round(clockRate).Sub(czero)
+	return time.Now().Round(clockRate).Sub(startAt)
-}
-
-// clockToTime converts a clock() timestamp to an absolute time.Time value.
-func clockToTime(c time.Duration) time.Time {
-	return czero.Add(c)
 }
 
 // clockRound returns d rounded to the nearest clockRate increment.

internal/p2p/conn/connection.go

@@ -134,6 +134,9 @@ type MConnConfig struct {
 
 	// Maximum wait time for pongs
 	PongTimeout time.Duration `mapstructure:"pong_timeout"`
+
+	// Process/Transport Start time
+	StartTime time.Time `mapstructure:",omitempty"`
 }
 
 // DefaultMConnConfig returns the default config.
@@ -145,33 +148,17 @@ func DefaultMConnConfig() MConnConfig {
 		FlushThrottle:           defaultFlushThrottle,
 		PingInterval:            defaultPingInterval,
 		PongTimeout:             defaultPongTimeout,
+		StartTime:               time.Now(),
 	}
 }
 
-// NewMConnection wraps net.Conn and creates multiplex connection
+// NewMConnection wraps net.Conn and creates multiplex connection with a config
 func NewMConnection(
 	logger log.Logger,
 	conn net.Conn,
 	chDescs []*ChannelDescriptor,
 	onReceive receiveCbFunc,
 	onError errorCbFunc,
-) *MConnection {
-	return NewMConnectionWithConfig(
-		logger,
-		conn,
-		chDescs,
-		onReceive,
-		onError,
-		DefaultMConnConfig())
-}
-
-// NewMConnectionWithConfig wraps net.Conn and creates multiplex connection with a config
-func NewMConnectionWithConfig(
-	logger log.Logger,
-	conn net.Conn,
-	chDescs []*ChannelDescriptor,
-	onReceive receiveCbFunc,
-	onError errorCbFunc,
 	config MConnConfig,
 ) *MConnection {
 	if config.PongTimeout >= config.PingInterval {
@@ -183,8 +170,8 @@ func NewMConnectionWithConfig(
 		conn:          conn,
 		bufConnReader: bufio.NewReaderSize(conn, minReadBufferSize),
 		bufConnWriter: bufio.NewWriterSize(conn, minWriteBufferSize),
-		sendMonitor:   flowrate.New(0, 0),
+		sendMonitor:   flowrate.New(config.StartTime, 0, 0),
-		recvMonitor:   flowrate.New(0, 0),
+		recvMonitor:   flowrate.New(config.StartTime, 0, 0),
 		send:          make(chan struct{}, 1),
 		pong:          make(chan struct{}, 1),
 		onReceive:     onReceive,

internal/p2p/conn/connection_test.go

@@ -42,7 +42,7 @@ func createMConnectionWithCallbacks(
 	cfg.PingInterval = 90 * time.Millisecond
 	cfg.PongTimeout = 45 * time.Millisecond
 	chDescs := []*ChannelDescriptor{{ID: 0x01, Priority: 1, SendQueueCapacity: 1}}
-	c := NewMConnectionWithConfig(logger, conn, chDescs, onReceive, onError, cfg)
+	c := NewMConnection(logger, conn, chDescs, onReceive, onError, cfg)
 	return c
 }
 
@@ -453,7 +453,7 @@ func newClientAndServerConnsForReadErrors(
 	}
 	logger := log.TestingLogger()
 
-	mconnClient := NewMConnection(logger.With("module", "client"), client, chDescs, onReceive, onError)
+	mconnClient := NewMConnection(logger.With("module", "client"), client, chDescs, onReceive, onError, DefaultMConnConfig())
 	err := mconnClient.Start(ctx)
 	require.NoError(t, err)
 

internal/p2p/transport_mconn.go

@@ -376,7 +376,7 @@ func (c *mConnConnection) handshake(
 		return nil, types.NodeInfo{}, nil, err
 	}
 
-	mconn := conn.NewMConnectionWithConfig(
+	mconn := conn.NewMConnection(
 		c.logger.With("peer", c.RemoteEndpoint().NodeAddress(peerInfo.NodeID)),
 		secretConn,
 		c.channelDescs,