priority

README

Priority discipline

Purpose

Used to distributes data among handlers according to priority

Also may be used to equaling distribution of data with different processing times

Principle of operation

• Prioritization:

  

• Equaling:

  

Comparison with unmanaged distribution

If different times are spent processing data of different priorities, then we will get different processing speeds in the case of using the priority discipline and without it.

For example, suppose that data from channel of priority 3 is processed in time T, data from channel of priority 2 is processed in time 5*T, and data from channel of priority 1 is processed in time 10*T, then we will get the following results:

• equaling by priority discipline:

  

• unmanaged distribution:

  

It can be seen that with unmanaged distribution, the processing speed of data with priority 3 is limited by the slowest processed data (with priority 1 and 2), but at with equaling by priority discipline the processing speed of data with priority 3 is no limited by others priorities

Usage

Example:

package main

import (
    "fmt"
    "strconv"
    "sync"

    "github.com/akramarenkov/cqos/priority"
)

func main() {
    handlersQuantity := 100
    // Preferably input channels should be buffered
    inputCapacity := 10
    itemsQuantity := 100

    inputs := map[uint]chan string{
        3: make(chan string, inputCapacity),
        2: make(chan string, inputCapacity),
        1: make(chan string, inputCapacity),
    }

    // Map key is a value of priority
    inputsOpts := map[uint]<-chan string{
        3: inputs[3],
        2: inputs[2],
        1: inputs[1],
    }

    defer func() {
        for _, input := range inputs {
            close(input)
        }
    }()

    // Data from input channels passed to handlers by output channel
    output := make(chan priority.Prioritized[string])

    // Handlers must write priority of processed data to feedback channel after it has been processed
    feedback := make(chan uint)
    defer close(feedback)

    // Used only in this example for detect that all written data are processed
    measurements := make(chan bool)
    defer close(measurements)

    // For equaling use FairDivider, for prioritization use RateDivider or custom divider
    disciplineOpts := priority.Opts[string]{
        Divider:          priority.RateDivider,
        Feedback:         feedback,
        HandlersQuantity: uint(handlersQuantity),
        Inputs:           inputsOpts,
        Output:           output,
    }

    discipline, err := priority.New(disciplineOpts)
    if err != nil {
        panic(err)
    }

    defer discipline.Stop()

    wg := &sync.WaitGroup{}
    defer wg.Wait()

    // Run handlers, that process data
    for handler := 0; handler < handlersQuantity; handler++ {
        wg.Add(1)

        go func() {
            defer wg.Done()

            for prioritized := range output {
                // Data processing
                fmt.Println(prioritized.Item)
                measurements <- true

                feedback <- prioritized.Priority
            }
        }()
    }

    // Run writers, that write data to input channels
    for priority, input := range inputs {
        wg.Add(1)

        go func(precedency uint, channel chan string) {
            defer wg.Done()

            base := strconv.Itoa(int(precedency))

            for id := 0; id < itemsQuantity; id++ {
                item := base + ":" + strconv.Itoa(id)
                
                channel <- item
            }
        }(priority, input)
    }

    defer close(output)

    received := 0

    // Wait for process all written data
    for range measurements {
        received++

        if received == itemsQuantity*len(inputs) {
            return
        }
    }
}

Documentation

Index

• Variables
• func FairDivider(priorities []uint, dividend uint, distribution map[uint]uint) map[uint]uint
• func IsNonFatalConfig(priorities []uint, divider Divider, quantity uint) bool
• func IsSuitableConfig(priorities []uint, divider Divider, quantity uint, limit float64) bool
• func PickUpMaxNonFatalQuantity(priorities []uint, divider Divider, maxQuantity uint) uint
• func PickUpMaxSuitableQuantity(priorities []uint, divider Divider, maxQuantity uint, limit float64) uint
• func PickUpMinNonFatalQuantity(priorities []uint, divider Divider, maxQuantity uint) uint
• func PickUpMinSuitableQuantity(priorities []uint, divider Divider, maxQuantity uint, limit float64) uint
• func RateDivider(priorities []uint, dividend uint, distribution map[uint]uint) map[uint]uint
• type Discipline
  • func New[Type any](opts Opts[Type]) (*Discipline[Type], error)
  • func (dsc *Discipline[Type]) AddInput(channel <-chan Type, priority uint)
  • func (dsc *Discipline[Type]) Err() <-chan error
  • func (dsc *Discipline[Type]) RemoveInput(priority uint)
  • func (dsc *Discipline[Type]) Stop()
• type Divider
• type Handle
• type Opts
• type Prioritized
• type Simple
  • func NewSimple[Type any](opts SimpleOpts[Type]) (*Simple[Type], error)
  • func (smpl *Simple[Type]) Err() <-chan error
  • func (smpl *Simple[Type]) Stop()
• type SimpleOpts

Variables

var (
	ErrEmptyDivider     = errors.New("priorities divider was not specified")
	ErrEmptyFeedback    = errors.New("feedback channel was not specified")
	ErrEmptyOutput      = errors.New("output channel was not specified")
	ErrQuantityExceeded = errors.New("value of handlers quantity has been exceeded")
)

var (
	ErrEmptyHandle = errors.New("handle function was not specified")
)

Functions

func FairDivider

func FairDivider(priorities []uint, dividend uint, distribution map[uint]uint) map[uint]uint

Distributes quantity evenly among the priorities.

Used for equaling.

Example results:

• 6 / [3 2 1] = map[3:2, 2:2, 1:2]
• 100 / [70 20 10] = map[70:34, 20:33, 10:33]

func IsNonFatalConfig

func IsNonFatalConfig(
	priorities []uint,
	divider Divider,
	quantity uint,
) bool

Due to the imperfection of the dividing function and working with integers (since the quantity of handlers is an integer), large errors can occur when distributing handlers by priority, especially for small quantity of handlers. This function allows you to determine that with the specified combination of priorities, the dividing function and the quantity of handlers, the distribution error does not cause stop processing of one or more priorities (for none of the priorities, the quantity is not equal to zero)

func IsSuitableConfig

func IsSuitableConfig(
	priorities []uint,
	divider Divider,
	quantity uint,
	limit float64,
) bool

Due to the imperfection of the dividing function and working with integers (since the quantity of handlers is an integer), large errors can occur when distributing handlers by priority, especially for small quantity of handlers. This function allows you to determine that with the specified combination of priorities, the dividing function and the quantity of handlers, the distribution error does not exceed the limit

func PickUpMaxNonFatalQuantity

func PickUpMaxNonFatalQuantity(
	priorities []uint,
	divider Divider,
	maxQuantity uint,
) uint

Picks up the maximum quantity of handlers for which the division error does not cause stop processing of one or more priorities

func PickUpMaxSuitableQuantity

func PickUpMaxSuitableQuantity(
	priorities []uint,
	divider Divider,
	maxQuantity uint,
	limit float64,
) uint

Picks up the maximum quantity of handlers for which the division error does not exceed the limit

func PickUpMinNonFatalQuantity

func PickUpMinNonFatalQuantity(
	priorities []uint,
	divider Divider,
	maxQuantity uint,
) uint

Picks up the minimum quantity of handlers for which the division error does not cause stop processing of one or more priorities

func PickUpMinSuitableQuantity

func PickUpMinSuitableQuantity(
	priorities []uint,
	divider Divider,
	maxQuantity uint,
	limit float64,
) uint

Picks up the minimum quantity of handlers for which the division error does not exceed the limit

func RateDivider

func RateDivider(priorities []uint, dividend uint, distribution map[uint]uint) map[uint]uint

Distributes quantity between priorities in proportion to the priority value.

Used for prioritization.

Example results:

• 6 / [3 2 1] = map[3:3, 2:2, 1:1]
• 100 / [70 20 10] = map[70:70, 20:20, 10:10]

Types

type Discipline

type Discipline[Type any] struct {
	// contains filtered or unexported fields
}

Main prioritization discipline.

Preferably input channels should be buffered for performance reasons.

Data from input channels passed to handlers by output channel.

Handlers must write priority of processed data to feedback channel after it has been processed.

For equaling use FairDivider, for prioritization use RateDivider or custom divider

func New

func New[Type any](opts Opts[Type]) (*Discipline[Type], error)

Creates and runs main prioritization discipline

func (*Discipline[Type]) AddInput

func (dsc *Discipline[Type]) AddInput(channel <-chan Type, priority uint)

Adds or updates (if it added previously) input channel for specified priority

func (*Discipline[Type]) Err

func (dsc *Discipline[Type]) Err() <-chan error

Returns a channel with errors. If an error occurs (the value from the channel is not equal to nil) the discipline terminates its work. The most likely cause of the error is an incorrectly working dividing function in which the sum of the distributed quantities is not equal to the original quantity.

The single nil value means that the discipline has terminated in normal mode

func (*Discipline[Type]) RemoveInput

func (dsc *Discipline[Type]) RemoveInput(priority uint)

Removes input channel for specified priority

func (*Discipline[Type]) Stop

func (dsc *Discipline[Type]) Stop()

Terminates work of the discipline.

Use for wait completion at terminates via context

type Divider

type Divider func(priorities []uint, dividend uint, distribution map[uint]uint) map[uint]uint

Distributes quantity of something by priorities. Determines how handlers are distributed among priorities.

Slice of priorities is passed to this function sorted from highest to lowest.

Sum of the distributed quantities must equal the original quantity.

If distribution is nil then it must be created and returned, otherwise it must be updated and returned.

type Handle

type Handle[Type any] func(ctx context.Context, item Type)

Callback function called in handlers of simplified prioritization discipline when an item is received.

Function should be interrupted when context is canceled

type Opts

type Opts[Type any] struct {
	// Terminates (cancels) work of the discipline
	Ctx context.Context
	// Determines how handlers are distributed among priorities
	Divider Divider
	// Handlers must write priority of processed data to feedback channel after it has been processed
	Feedback <-chan uint
	// Between how many handlers you need to distribute data
	HandlersQuantity uint
	// Channels with input data, should be buffered for performance reasons. Map key is a value of priority
	Inputs map[uint]<-chan Type
	// Handlers should read distributed data from this channel
	Output chan<- Prioritized[Type]
}

Options of the created main prioritization discipline

type Prioritized

type Prioritized[Type any] struct {
	Item     Type
	Priority uint
}

Describes the data distributed by the prioritization discipline

type Simple

type Simple[Type any] struct {
	// contains filtered or unexported fields
}

Simplified version of the discipline that runs handlers on its own and hides the output and feedback channels

func NewSimple

func NewSimple[Type any](opts SimpleOpts[Type]) (*Simple[Type], error)

Creates and runs simplified prioritization discipline

func (*Simple[Type]) Err

func (smpl *Simple[Type]) Err() <-chan error

Returns a channel with errors. If an error occurs (the value from the channel is not equal to nil) the discipline terminates its work. The most likely cause of the error is an incorrectly working dividing function in which the sum of the distributed quantities is not equal to the original quantity.

The single nil value means that the discipline has terminated in normal mode

func (*Simple[Type]) Stop

func (smpl *Simple[Type]) Stop()

Terminates work of the discipline.

Use for wait completion at terminates via context

type SimpleOpts

type SimpleOpts[Type any] struct {
	// Terminates (cancels) work of the discipline
	Ctx context.Context
	// Determines how handlers are distributed among priorities
	Divider Divider
	// Callback function called in handlers when an item is received
	Handle Handle[Type]
	// Between how many handlers you need to distribute data
	HandlersQuantity uint
	// Channels with input data, should be buffered for performance reasons. Map key is a value of priority
	Inputs map[uint]<-chan Type
}

Options of the created simplified prioritization discipline