// process starts a blocking WDRR scheduler process, where we continuously
// evaluate if we need to attempt to enqueue an Envelope or schedule Envelopes
// to be dequeued and subsequently read and sent on the source connection.
// Internally, each p2p Channel maps to a flow, where each flow has a deficit
// and a quantum.
//
// For each Envelope requested to be enqueued, we evaluate if there is sufficient
// capacity in the shared buffer to add the Envelope. If so, it is added.
// Otherwise, we evaluate all flows of lower priority where we attempt find an
// existing Envelope in the shared buffer of sufficient size that can be dropped
// in place of the incoming Envelope. If there is no such Envelope that can be
// dropped, then the incoming Envelope is dropped.
//
// When there is nothing to be enqueued, we perform the WDRR algorithm and
// determine which Envelopes can be dequeued. For each Envelope that can be
// dequeued, it is sent on the dequeueCh. Specifically, for each flow, if it is
// non-empty, its deficit counter is incremented by its quantum value. Then, the
// value of the deficit counter is a maximal amount of bytes that can be sent at
// this round. If the deficit counter is greater than the Envelopes's message
// size at the head of the queue (HoQ), this envelope can be sent and the value
// of the counter is decremented by the message's size. Then, the size of the
// next Envelopes's message is compared to the counter value, etc. Once the flow
// is empty or the value of the counter is insufficient, the scheduler will skip
// to the next flow. If the flow is empty, the value of the deficit counter is
// reset to 0.
//
// XXX/TODO: Evaluate the single goroutine scheduler mechanism. In other words,
// evaluate the effectiveness and performance of having a single goroutine
// perform handling both enqueueing and dequeueing logic. Specifically, there
// is potentially contention between reading off of enqueueCh and trying to
// enqueue while also attempting to perform the WDRR algorithm and find the next
// set of Envelope(s) to send on the dequeueCh. Alternatively, we could consider
// separate scheduling goroutines, but then that requires the use of mutexes and
// possibly a degrading performance.
process starts a blocking WDRR scheduler process, where we continuously
evaluate if we need to attempt to enqueue an Envelope or schedule Envelopes
to be dequeued and subsequently read and sent on the source connection.
Internally, each p2p Channel maps to a flow, where each flow has a deficit
and a quantum.

For each Envelope requested to be enqueued, we evaluate if there is sufficient
capacity in the shared buffer to add the Envelope. If so, it is added.
Otherwise, we evaluate all flows of lower priority where we attempt find an
existing Envelope in the shared buffer of sufficient size that can be dropped
in place of the incoming Envelope. If there is no such Envelope that can be
dropped, then the incoming Envelope is dropped.

When there is nothing to be enqueued, we perform the WDRR algorithm and
determine which Envelopes can be dequeued. For each Envelope that can be
dequeued, it is sent on the dequeueCh. Specifically, for each flow, if it is
non-empty, its deficit counter is incremented by its quantum value. Then, the
value of the deficit counter is a maximal amount of bytes that can be sent at
this round. If the deficit counter is greater than the Envelopes's message
size at the head of the queue (HoQ), this envelope can be sent and the value
of the counter is decremented by the message's size. Then, the size of the
next Envelopes's message is compared to the counter value, etc. Once the flow
is empty or the value of the counter is insufficient, the scheduler will skip
to the next flow. If the flow is empty, the value of the deficit counter is
reset to 0.

XXX/TODO: Evaluate the single goroutine scheduler mechanism. In other words,
evaluate the effectiveness and performance of having a single goroutine
perform handling both enqueueing and dequeueing logic. Specifically, there
is potentially contention between reading off of enqueueCh and trying to
enqueue while also attempting to perform the WDRR algorithm and find the next
set of Envelope(s) to send on the dequeueCh. Alternatively, we could consider
separate scheduling goroutines, but then that requires the use of mutexes and
possibly a degrading performance.
// process starts a blocking WDRR scheduler process where we continuously // evaluate if we need to attempt to enqueue an Envelope or schedule Envelopes // to be dequeued and subsequently read and sent on the source connection . // Internally each p2p Channel maps to a flow where each flow has a deficit // and a quantum . // // For each Envelope requested to be enqueued we evaluate if there is sufficient // capacity in the shared buffer to add the Envelope . If so it is added . // Otherwise we evaluate all flows of lower priority where we attempt find an // existing Envelope in the shared buffer of sufficient size that can be dropped // in place of the incoming Envelope . If there is no such Envelope that can be // dropped then the incoming Envelope is dropped . // // When there is nothing to be enqueued we perform the WDRR algorithm and // determine which Envelopes can be dequeued . For each Envelope that can be // dequeued it is sent on the dequeueCh . Specifically for each flow if it is // non - empty its deficit counter is incremented by its quantum value . Then the // value of the deficit counter is a maximal amount of bytes that can be sent at // this round . If the deficit counter is greater than the Envelopes ' s message // size at the head of the queue ( HoQ ) this envelope can be sent and the value // of the counter is decremented by the message ' s size . Then the size of the // next Envelopes ' s message is compared to the counter value etc . Once the flow // is empty or the value of the counter is insufficient the scheduler will skip // to the next flow . If the flow is empty the value of the deficit counter is // reset to 0 . // // XXX / TODO : Evaluate the single goroutine scheduler mechanism . In other words // evaluate the effectiveness and performance of having a single goroutine // perform handling both enqueueing and dequeueing logic . Specifically there // is potentially contention between reading off of enqueueCh and trying to // enqueue while also attempting to perform the WDRR algorithm and find the next // set of Envelope ( s ) to send on the dequeueCh . Alternatively we could consider // separate scheduling goroutines but then that requires the use of mutexes and // possibly a degrading performance .
func (s *wdrrScheduler) process() {
	defer s.doneCh.Close()
	for {
		select {
		case <-s.closer.Done():
			return
		case e := <-s.enqueueCh:
			chIDStr := strconv.Itoa(int(e.channelID))
			wEnv := wrappedEnvelope{envelope: e, size: uint(proto.Size(e.Message))}
			msgSize := wEnv.size
			s.metrics.PeerPendingSendBytes.With("peer_id", string(e.To)).Add(float64(msgSize))
			if s.size+wEnv.size > s.capacity {
				chPriority := s.chPriorities[e.channelID]
				var (
					canDrop  bool
					dropIdx  int
					dropChID ChannelID
				)
				for i := len(s.chDescs) - 1; i >= 0 && uint(s.chDescs[i].Priority) < chPriority && !canDrop; i-- {
					currChID := ChannelID(s.chDescs[i].ID)
					flow := s.buffer[currChID]
					for j := 0; j < len(flow) && !canDrop; j++ {
						if flow[j].size >= wEnv.size {
							canDrop = true
							dropIdx = j
							dropChID = currChID
							break
						}
					}
				}
				if canDrop {
					chIDStr = strconv.Itoa(int(dropChID))
					chPriority = s.chPriorities[dropChID]
					msgSize = s.buffer[dropChID][dropIdx].size
					s.size -= msgSize
					s.buffer[dropChID] = append(s.buffer[dropChID][:dropIdx], s.buffer[dropChID][dropIdx+1:]...)
					s.size += wEnv.size
					s.buffer[e.channelID] = append(s.buffer[e.channelID], wEnv)
					s.metrics.PeerQueueMsgSize.With("ch_id", chIDStr).Set(float64(wEnv.size))
				}
				s.metrics.PeerQueueDroppedMsgs.With("ch_id", chIDStr).Add(1)
				s.logger.Debug(
					"dropped envelope",
					"ch_id", chIDStr,
					"priority", chPriority,
					"capacity", s.capacity,
					"msg_size", msgSize,
				)
			} else {
				s.metrics.PeerQueueMsgSize.With("ch_id", chIDStr).Set(float64(wEnv.size))
				s.buffer[e.channelID] = append(s.buffer[e.channelID], wEnv)
				s.size += wEnv.size
			}
		default:
			for _, chDesc := range s.chDescs {
				chID := ChannelID(chDesc.ID)
				if len(s.buffer[chID]) > 0 {
					s.deficits[chID] += s.quanta[chID]
					d := s.deficits[chID]
					we := s.buffer[chID][0]
					 1. send the Envelope
					 2. update the scheduler's shared buffer's size
					 3. update the flow's deficit
					 4. remove from the flow's queue
					 5. grab the next HoQ Envelope and flow's deficit
					for len(s.buffer[chID]) > 0 && d >= we.size {
						s.metrics.PeerSendBytesTotal.With(
							"chID", fmt.Sprint(chID),
							"peer_id", string(we.envelope.To)).Add(float64(we.size))
						s.dequeueCh <- we.envelope
						s.size -= we.size
						s.deficits[chID] -= we.size
						s.buffer[chID] = s.buffer[chID][1:]
						if len(s.buffer[chID]) > 0 {
							d = s.deficits[chID]
							we = s.buffer[chID][0]
						}
					}
				}
				 reset the flow's deficit to zero if it is empty
				if len(s.buffer[chID]) == 0 {
					s.deficits[chID] = 0
				}
			}
		}
	}
}
func ( s * wdrrScheduler ) process ( ) { defer s . doneCh . Close ( ) for { select { case <- s . closer . Done ( ) : return case e := <- s . enqueueCh : chIDStr := strconv . Itoa ( int ( e . channelID ) ) wEnv := wrappedEnvelope { envelope : e , size : uint ( proto . Size ( e . Message ) ) } msgSize := wEnv . size s . metrics . PeerPendingSendBytes . With ( " " , string ( e . To ) ) . Add ( float64 ( msgSize ) ) if s . size + wEnv . size > s . capacity { chPriority := s . chPriorities [ e . channelID ] var ( canDrop bool dropIdx int dropChID ChannelID ) for i := len ( s . chDescs ) - 1 ; i >= 0 && uint ( s . chDescs [ i ] . Priority ) < chPriority && ! canDrop ; i -- { currChID := ChannelID ( s . chDescs [ i ] . ID ) flow := s . buffer [ currChID ] for j := 0 ; j < len ( flow ) && ! canDrop ; j ++ { if flow [ j ] . size >= wEnv . size { canDrop = true dropIdx = j dropChID = currChID break } } } if canDrop { chIDStr = strconv . Itoa ( int ( dropChID ) ) chPriority = s . chPriorities [ dropChID ] msgSize = s . buffer [ dropChID ] [ dropIdx ] . size s . size -= msgSize s . buffer [ dropChID ] = append ( s . buffer [ dropChID ] [ : dropIdx ] , s . buffer [ dropChID ] [ dropIdx + 1 : ] ... ) s . size += wEnv . size s . buffer [ e . channelID ] = append ( s . buffer [ e . channelID ] , wEnv ) s . metrics . PeerQueueMsgSize . With ( " " , chIDStr ) . Set ( float64 ( wEnv . size ) ) } s . metrics . PeerQueueDroppedMsgs . With ( " " , chIDStr ) . Add ( 1 ) s . logger . Debug ( " " , " " , chIDStr , " " , chPriority , " " , s . capacity , " " , msgSize , ) } else { s . metrics . PeerQueueMsgSize . With ( " " , chIDStr ) . Set ( float64 ( wEnv . size ) ) s . buffer [ e . channelID ] = append ( s . buffer [ e . channelID ] , wEnv ) s . size += wEnv . size } default : for _ , chDesc := range s . chDescs { chID := ChannelID ( chDesc . ID ) if len ( s . buffer [ chID ] ) > 0 { s . deficits [ chID ] += s . quanta [ chID ] d := s . deficits [ chID ] we := s . buffer [ chID ] [ 0 ] for len ( s . buffer [ chID ] ) > 0 && d >= we . size { s . metrics . PeerSendBytesTotal . With ( " " , fmt . Sprint ( chID ) , " " , string ( we . envelope . To ) ) . Add ( float64 ( we . size ) ) s . dequeueCh <- we . envelope s . size -= we . size s . deficits [ chID ] -= we . size s . buffer [ chID ] = s . buffer [ chID ] [ 1 : ] if len ( s . buffer [ chID ] ) > 0 { d = s . deficits [ chID ] we = s . buffer [ chID ] [ 0 ] } } } if len ( s . buffer [ chID ] ) == 0 { s . deficits [ chID ] = 0 } } } } }
process starts a blocking WDRR scheduler process, where we continuously
evaluate if we need to attempt to enqueue an Envelope or schedule Envelopes
to be dequeued and subsequently read and sent on the source connection.
process starts a blocking WDRR scheduler process where we continuously evaluate if we need to attempt to enqueue an Envelope or schedule Envelopes to be dequeued and subsequently read and sent on the source connection .
// attempt to enqueue the incoming Envelope // If we're at capacity, we need to either drop the incoming Envelope or // an Envelope from a lower priority flow. Otherwise, we add the (wrapped) // envelope to the flow's queue. // Evaluate all lower priority flows and determine if there exists an // Envelope that is of equal or greater size that we can drop in favor // of the incoming Envelope. // If we can drop an existing Envelope, drop it and enqueue the incoming // Envelope. // Drop Envelope for the lower priority flow and update the queue's // buffer size // add the incoming Envelope and update queue's buffer size // We either dropped the incoming Enevelope or one from an existing // lower priority flow. // we have sufficient capacity to enqueue the incoming Envelope // perform the WDRR algorithm // only consider non-empty flows // bump flow's quantum // grab the flow's current deficit counter and HoQ (wrapped) Envelope // While the flow is non-empty and we can send the current Envelope // on the dequeueCh: // // 1. send the Envelope // 2. update the scheduler's shared buffer's size // 3. update the flow's deficit // 4. remove from the flow's queue // 5. grab the next HoQ Envelope and flow's deficit // reset the flow's deficit to zero if it is empty