feat(bitswap.decision.Engine) use PriorityQueue for Engine.Outbox

[btc]

This PR makes bitswap's decision engine more intelligent. Now, incoming block requests are pushed into a priority queue and sorted using a pluggable comparator.

The default comparator (provisionally-titled V1) is defined as follows:

// V1 respects the target peer's wantlist priority. For tasks involving
// different peers, the oldest task is prioritized.
var V1 = func(a, b *peerRequestTask) bool {
    if a.Target == b.Target {
        return a.Entry.Priority > b.Entry.Priority
    }
    return FIFO(a, b)
}

// FIFO is a basic task comparator that returns tasks in the order created.
var FIFO = func(a, b *peerRequestTask) bool {
    return a.created.Before(b.created)
}

The engine interface remains the same:

func NewEngine(ctx context.Context, bs bstore.Blockstore) *Engine

// MessageReceived performs book-keeping. Returns error if passed invalid
// arguments.
func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error

func (e *Engine) MessageSent(p peer.ID, m bsmsg.BitSwapMessage) error

func (e *Engine) Outbox() <-chan Envelope

func (e *Engine) Peers() []peer.ID
    Returns a slice of Peers with whom the local node has active sessions

type Envelope struct {
    // Peer is the intended recipient
    Peer peer.ID
    // Message is the payload
    Message bsmsg.BitSwapMessage
}
    Envelope contains a message for a Peer

It's built ontop of a generic pq that could probably be used elsewhere:

package pq
    import "."


TYPES

type Elem interface {
    // SetIndex stores the int index.
    SetIndex(int)
    // Index returns the last given by SetIndex(int).
    Index() int
}
    Elem describes elements that can be added to the PQ. Clients must
    implement this interface.

type ElemComparator func(a, b Elem) bool
    ElemComparator returns true if pri(a) > pri(b)

type PQ interface {
    // Push adds the ele
    Push(Elem)
    // Pop returns the highest priority Elem in PQ.
    Pop() Elem
    // Len returns the number of elements in the PQ.
    Len() int
    // Update `fixes` the PQ.
    Update(index int)
}
    PQ is a basic priority queue.

func New(cmp ElemComparator) PQ
    New creates a PQ with a client-supplied comparator.

[whyrusleeping]

I was planning a refactor in bitswap that would hold a queue for each partner, and then hold a priority queue of the partners to decide who we service next. The idea behind it is that we need to try and avoid 'slow' peers holding up the entire system.

Thats not to say that this interferes with that in any way, just letting you know my plans.

[btc]

RFCR @jbenet @whyrusleeping

[jbenet]

is there a strong reason to order these? if so maybe comment it? at first glance i wouldn't know and maybe screw up

[btc]

There isn't. I switched the order because I mistakenly believed that assigning to the variable would magically signal to the other goroutine immediately. In the moment, I had forgotten how to write software.

[jbenet]

❤️ 👍

[jbenet]

@briantigerchow RFM (though see comment above)

[btc]

Since RFM, exchange/bitswap/decision/pq package has been extracted to thirdparty/pq

Merging.

cc @jbenet @whyrusleeping