You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

292 lines
7.9 KiB

package p2p
import (
"container/heap"
"context"
"sort"
"strconv"
"time"
"github.com/gogo/protobuf/proto"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/libs/log"
)
// pqEnvelope defines a wrapper around an Envelope with priority to be inserted
// into a priority queue used for Envelope scheduling.
type pqEnvelope struct {
envelope Envelope
priority uint
size uint
timestamp time.Time
index int
}
// priorityQueue defines a type alias for a priority queue implementation.
type priorityQueue []*pqEnvelope
func (pq priorityQueue) get(i int) *pqEnvelope { return pq[i] }
func (pq priorityQueue) Len() int { return len(pq) }
func (pq priorityQueue) Less(i, j int) bool {
// if both elements have the same priority, prioritize based on most recent
if pq[i].priority == pq[j].priority {
return pq[i].timestamp.After(pq[j].timestamp)
}
// otherwise, pick the pqEnvelope with the higher priority
return pq[i].priority > pq[j].priority
}
func (pq priorityQueue) Swap(i, j int) {
pq[i], pq[j] = pq[j], pq[i]
pq[i].index = i
pq[j].index = j
}
func (pq *priorityQueue) Push(x interface{}) {
n := len(*pq)
pqEnv := x.(*pqEnvelope)
pqEnv.index = n
*pq = append(*pq, pqEnv)
}
func (pq *priorityQueue) Pop() interface{} {
old := *pq
n := len(old)
pqEnv := old[n-1]
old[n-1] = nil
pqEnv.index = -1
*pq = old[:n-1]
return pqEnv
}
// Assert the priority queue scheduler implements the queue interface at
// compile-time.
var _ queue = (*pqScheduler)(nil)
type pqScheduler struct {
logger log.Logger
metrics *Metrics
size uint
sizes map[uint]uint // cumulative priority sizes
pq *priorityQueue
chDescs []*ChannelDescriptor
capacity uint
chPriorities map[ChannelID]uint
enqueueCh chan Envelope
dequeueCh chan Envelope
closer *tmsync.Closer
done *tmsync.Closer
}
func newPQScheduler(
logger log.Logger,
m *Metrics,
chDescs []*ChannelDescriptor,
enqueueBuf, dequeueBuf, capacity uint,
) *pqScheduler {
// copy each ChannelDescriptor and sort them by ascending channel priority
chDescsCopy := make([]*ChannelDescriptor, len(chDescs))
copy(chDescsCopy, chDescs)
sort.Slice(chDescsCopy, func(i, j int) bool { return chDescsCopy[i].Priority < chDescsCopy[j].Priority })
var (
chPriorities = make(map[ChannelID]uint)
sizes = make(map[uint]uint)
)
for _, chDesc := range chDescsCopy {
chID := chDesc.ID
chPriorities[chID] = uint(chDesc.Priority)
sizes[uint(chDesc.Priority)] = 0
}
pq := make(priorityQueue, 0)
heap.Init(&pq)
return &pqScheduler{
logger: logger.With("router", "scheduler"),
metrics: m,
chDescs: chDescsCopy,
capacity: capacity,
chPriorities: chPriorities,
pq: &pq,
sizes: sizes,
enqueueCh: make(chan Envelope, enqueueBuf),
dequeueCh: make(chan Envelope, dequeueBuf),
closer: tmsync.NewCloser(),
done: tmsync.NewCloser(),
}
}
func (s *pqScheduler) enqueue() chan<- Envelope {
return s.enqueueCh
}
func (s *pqScheduler) dequeue() <-chan Envelope {
return s.dequeueCh
}
func (s *pqScheduler) close() {
s.closer.Close()
<-s.done.Done()
}
func (s *pqScheduler) closed() <-chan struct{} {
return s.closer.Done()
}
// start starts non-blocking process that starts the priority queue scheduler.
func (s *pqScheduler) start(ctx context.Context) {
go s.process(ctx)
}
// process starts a block process where we listen for Envelopes to enqueue. If
// there is sufficient capacity, it will be enqueued into the priority queue,
// otherwise, we attempt to dequeue enough elements from the priority queue to
// make room for the incoming Envelope by dropping lower priority elements. If
// there isn't sufficient capacity at lower priorities for the incoming Envelope,
// it is dropped.
//
// After we attempt to enqueue the incoming Envelope, if the priority queue is
// non-empty, we pop the top Envelope and send it on the dequeueCh.
func (s *pqScheduler) process(ctx context.Context) {
defer s.done.Close()
for {
select {
case e := <-s.enqueueCh:
chIDStr := strconv.Itoa(int(e.channelID))
pqEnv := &pqEnvelope{
envelope: e,
size: uint(proto.Size(e.Message)),
priority: s.chPriorities[e.channelID],
timestamp: time.Now().UTC(),
}
// enqueue
// Check if we have sufficient capacity to simply enqueue the incoming
// Envelope.
if s.size+pqEnv.size <= s.capacity {
s.metrics.PeerPendingSendBytes.With("peer_id", string(pqEnv.envelope.To)).Add(float64(pqEnv.size))
// enqueue the incoming Envelope
s.push(pqEnv)
} else {
// There is not sufficient capacity to simply enqueue the incoming
// Envelope. So we have to attempt to make room for it by dropping lower
// priority Envelopes or drop the incoming Envelope otherwise.
// The cumulative size of all enqueue envelopes at the incoming envelope's
// priority or lower.
total := s.sizes[pqEnv.priority]
if total >= pqEnv.size {
// There is room for the incoming Envelope, so we drop as many lower
// priority Envelopes as we need to.
var (
canEnqueue bool
tmpSize = s.size
i = s.pq.Len() - 1
)
// Drop lower priority Envelopes until sufficient capacity exists for
// the incoming Envelope
for i >= 0 && !canEnqueue {
pqEnvTmp := s.pq.get(i)
if pqEnvTmp.priority < pqEnv.priority {
if tmpSize+pqEnv.size <= s.capacity {
canEnqueue = true
} else {
pqEnvTmpChIDStr := strconv.Itoa(int(pqEnvTmp.envelope.channelID))
s.metrics.PeerQueueDroppedMsgs.With("ch_id", pqEnvTmpChIDStr).Add(1)
s.logger.Debug(
"dropped envelope",
"ch_id", pqEnvTmpChIDStr,
"priority", pqEnvTmp.priority,
"msg_size", pqEnvTmp.size,
"capacity", s.capacity,
)
s.metrics.PeerPendingSendBytes.With("peer_id", string(pqEnvTmp.envelope.To)).Add(float64(-pqEnvTmp.size))
// dequeue/drop from the priority queue
heap.Remove(s.pq, pqEnvTmp.index)
// update the size tracker
tmpSize -= pqEnvTmp.size
// start from the end again
i = s.pq.Len() - 1
}
} else {
i--
}
}
// enqueue the incoming Envelope
s.push(pqEnv)
} else {
// There is not sufficient capacity to drop lower priority Envelopes,
// so we drop the incoming Envelope.
s.metrics.PeerQueueDroppedMsgs.With("ch_id", chIDStr).Add(1)
s.logger.Debug(
"dropped envelope",
"ch_id", chIDStr,
"priority", pqEnv.priority,
"msg_size", pqEnv.size,
"capacity", s.capacity,
)
}
}
// dequeue
for s.pq.Len() > 0 {
pqEnv = heap.Pop(s.pq).(*pqEnvelope)
s.size -= pqEnv.size
// deduct the Envelope size from all the relevant cumulative sizes
for i := 0; i < len(s.chDescs) && pqEnv.priority <= uint(s.chDescs[i].Priority); i++ {
s.sizes[uint(s.chDescs[i].Priority)] -= pqEnv.size
}
s.metrics.PeerSendBytesTotal.With(
"chID", chIDStr,
"peer_id", string(pqEnv.envelope.To),
"message_type", s.metrics.ValueToMetricLabel(pqEnv.envelope.Message)).Add(float64(pqEnv.size))
s.metrics.PeerPendingSendBytes.With(
"peer_id", string(pqEnv.envelope.To)).Add(float64(-pqEnv.size))
select {
case s.dequeueCh <- pqEnv.envelope:
case <-s.closer.Done():
return
}
}
case <-ctx.Done():
return
case <-s.closer.Done():
return
}
}
}
func (s *pqScheduler) push(pqEnv *pqEnvelope) {
chIDStr := strconv.Itoa(int(pqEnv.envelope.channelID))
// enqueue the incoming Envelope
heap.Push(s.pq, pqEnv)
s.size += pqEnv.size
s.metrics.PeerQueueMsgSize.With("ch_id", chIDStr).Add(float64(pqEnv.size))
// Update the cumulative sizes by adding the Envelope's size to every
// priority less than or equal to it.
for i := 0; i < len(s.chDescs) && pqEnv.priority <= uint(s.chDescs[i].Priority); i++ {
s.sizes[uint(s.chDescs[i].Priority)] += pqEnv.size
}
}