The highest tagged major version is
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 ¶
Constants ¶
This section is empty.
Variables ¶
Functions ¶
func FairDivider ¶ added in v1.0.0
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 RateDivider ¶ added in v1.0.0
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]) 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
type Divider ¶ added in v1.0.0
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 Opts ¶
type Opts[Type any] struct { // 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 ¶ added in v1.0.0
Describes the data distributed by the prioritization discipline
Source Files