Fix scheduler token bucket waiting

pkg/network/p2p/neighbor.go

@@ -3,6 +3,7 @@ package p2p
 import (
  "context"
  "sync"
+ "sync/atomic"
  "time"
 
  "github.com/libp2p/go-libp2p/core/protocol"
@@ -15,7 +16,8 @@ import (
 )
 
 const (
- NeighborsSendQueueSize = 20_000
+ NeighborsSendQueueSize = 20_000
+ DroppedPacketDisconnectThreshold = 100
 )
 
 type queuedPacket struct {
@@ -31,7 +33,8 @@ type (
 
 // neighbor describes the established p2p connection to another peer.
 type neighbor struct {
- peer *network.Peer
+ peer *network.Peer
+ droppedPacketCounter atomic.Uint32
 
  logger log.Logger
 
@@ -84,7 +87,12 @@ func (n *neighbor) Peer() *network.Peer {
 func (n *neighbor) Enqueue(packet proto.Message, protocolID protocol.ID) {
  select {
  case n.sendQueue <- &queuedPacket{protocolID: protocolID, packet: packet}:
+ n.droppedPacketCounter.Store(0)
  default:
+ // Drop a neighbor that does not read from the full queue.
+ if n.droppedPacketCounter.Add(1) >= DroppedPacketDisconnectThreshold {
+ n.Close()
+ }
  n.logger.LogWarn("Dropped packet due to SendQueue being full")
  }
 }

pkg/protocol/engine/blockdag/events.go

@@ -3,6 +3,7 @@ package blockdag
 import (
  "github.com/iotaledger/hive.go/runtime/event"
  "github.com/iotaledger/iota-core/pkg/protocol/engine/blocks"
+ iotago "github.com/iotaledger/iota.go/v4"
 )
 
 // Events is a collection of Tangle related Events.
@@ -19,6 +20,9 @@ type Events struct {
  // MissingBlockAppended is triggered when a previously missing Block was appended.
  MissingBlockAppended *event.Event1[*blocks.Block]
 
+ // BlockNotAppended is triggered when an incoming Block could not be successfully appended.
+ BlockNotAppended *event.Event1[iotago.BlockID]
+
  // BlockInvalid is triggered when a Block is found to be invalid.
  BlockInvalid *event.Event2[*blocks.Block, error]
 
@@ -32,6 +36,7 @@ var NewEvents = event.CreateGroupConstructor(func() (newEvents *Events) {
  BlockSolid: event.New1[*blocks.Block](),
  BlockMissing: event.New1[*blocks.Block](),
  MissingBlockAppended: event.New1[*blocks.Block](),
+ BlockNotAppended: event.New1[iotago.BlockID](),
  BlockInvalid: event.New2[*blocks.Block, error](),
  }
 })

pkg/protocol/engine/blockdag/inmemoryblockdag/blockdag.go

@@ -47,6 +47,7 @@ func NewProvider(opts ...options.Option[BlockDAG]) module.Provider[*engine.Engin
 
  e.Events.PreSolidFilter.BlockPreAllowed.Hook(func(block *model.Block) {
  if _, _, err := b.Append(block); err != nil {
+ b.events.BlockNotAppended.Trigger(block.ID())
  b.LogError("failed to append block", "blockID", block.ID(), "issuer", block.ProtocolBlock().Header.IssuerID, "err", err)
  }
  }, event.WithWorkerPool(wp))

pkg/protocol/engine/congestioncontrol/scheduler/drr/basicbuffer.go

@@ -371,7 +371,7 @@ func (b *BasicBuffer) ringInsert(v interface{}) *ring.Ring {
 func (b *BasicBuffer) waitTime(rate float64, block *blocks.Block) time.Duration {
  tokensRequired := float64(block.WorkScore()) - (b.tokenBucket + rate*time.Since(b.lastScheduleTime).Seconds())
 
- return lo.Max(0, time.Duration(tokensRequired/rate))
+ return lo.Max(0, time.Duration(tokensRequired/rate)*time.Second)
 }
 
 func (b *BasicBuffer) updateTokenBucket(rate float64, tokenBucketSize float64) {

pkg/protocol/engine/congestioncontrol/scheduler/drr/scheduler.go

@@ -370,10 +370,11 @@ loop:
  case blockToSchedule = <-s.basicBuffer.blockChan:
  currentAPI := s.apiProvider.CommittedAPI()
  rate := currentAPI.ProtocolParameters().CongestionControlParameters().SchedulerRate
- if waitTime := s.basicBuffer.waitTime(float64(rate), blockToSchedule); waitTime > 0 {
+ for waitTime := s.basicBuffer.waitTime(float64(rate), blockToSchedule); waitTime > 0; {
  timer := time.NewTimer(waitTime)
  <-timer.C
  }
+
  s.basicBuffer.updateTokenBucket(float64(rate), float64(currentAPI.MaxBlockWork()))
 
  s.scheduleBasicBlock(blockToSchedule)

pkg/protocol/engine/engine.go

@@ -585,15 +585,23 @@ func (e *Engine) setupEvictionState() {
 
 func (e *Engine) setupBlockRequester() {
  e.Events.BlockRequester.LinkTo(e.BlockRequester.Events)
-
+ wp := e.Workers.CreatePool("BlockRequester", workerpool.WithWorkerCount(1)) // Using just 1 worker to avoid contention
  // We need to hook to make sure that the request is created before the block arrives to avoid a race condition
  // where we try to delete the request again before it is created. Thus, continuing to request forever.
  e.Events.BlockDAG.BlockMissing.Hook(func(block *blocks.Block) {
  e.BlockRequester.StartTicker(block.ID())
  })
+
  e.Events.BlockDAG.MissingBlockAppended.Hook(func(block *blocks.Block) {
  e.BlockRequester.StopTicker(block.ID())
- }, event.WithWorkerPool(e.Workers.CreatePool("BlockRequester", workerpool.WithWorkerCount(1)))) // Using just 1 worker to avoid contention
+ }, event.WithWorkerPool(wp))
+
+ // Remove the block from the ticker if it failed to be appended.
+ // It's executed for all blocks to avoid locking twice:
+ // once to check if the block has the ticker and then again to remove it.
+ e.Events.BlockDAG.BlockNotAppended.Hook(func(blockID iotago.BlockID) {
+ e.BlockRequester.StopTicker(blockID)
+ }, event.WithWorkerPool(wp))
 }
 
 func (e *Engine) setupPruning() {