queue

Documentation

Index

• Variables
• func NewFFQueue(ctx context.Context, pollInterval time.Duration, spawnDeadline time.Duration) *ffQueue
• func NewThreadPacker(ctx context.Context, maxPackSize int, timeout time.Duration) *threadPacker
• type CallQueue
• type PeerCall
• type ThreadPack
• type ThreadPacker

Variables

var (
	// Size of incoming requests buffer
	InBufSize = 1

	// Size of packed threads buffer
	OutBufSize = 1
)

Functions

func NewFFQueue

func NewFFQueue(
	ctx context.Context,
	pollInterval time.Duration,
	spawnDeadline time.Duration,
) *ffQueue

Fair FIFO-queue with isolated per-peer processing and adaptive invocation rate. Queue is polled with specified frequency and every scheduled call expected to be spawned until its deadline. At every moment only one call for the peer/thread pair exists in the queue. Scheduled operations could be replaced with a new ones based on the priority value (new higher-priority call replaces waiting one).

func NewThreadPacker

func NewThreadPacker(ctx context.Context, maxPackSize int, timeout time.Duration) *threadPacker

Packer accumulates peer-related thread requests and packs it into the limited-size containers during time-window constrained by provided timeout.

Types

type CallQueue

type CallQueue interface {
	// Make call immediately and synchronously return its result.
	Call(p peer.ID, t thread.ID, c PeerCall) error

	// Schedule call to be invoked later.
	Schedule(p peer.ID, t thread.ID, priority int, c PeerCall) bool
}

type PeerCall

type PeerCall func(context.Context, peer.ID, thread.ID) error

type ThreadPack

type ThreadPack struct {
	Peer    peer.ID
	Threads []thread.ID
}

type ThreadPacker

type ThreadPacker interface {
	// Add thread to peer's queue
	Add(pid peer.ID, tid thread.ID)

	// Start packing incoming thread requests
	Run() <-chan ThreadPack
}