Shulou(Shulou.com)05/31 Report--

In this issue, the editor will bring you about how to achieve distributed task RR allocation strategy based on ring queue and iterator. The article is rich in content and analyzes and describes for you from a professional point of view. I hope you can get something after reading this article.

# background

# # distributed Task assignment

In many OPS scenarios, we perform long-term tasks, such as installation, deployment environment, packaging images, and other long-time tasks. Usually, the number of task nodes is limited (excluding automatic scaling scenarios such as hpa based on K8s or knative).

So when we have a task how to allocate reasonably according to the current worker and corrdinator and tasks, the assignment is actually quite complex, and we can do it in complexity according to the load of the current system, the execution resource consumption of each task, the number of tasks in the current cluster, and so on. Here we will create the simplest RR algorithm based on task and current worker.

# # system Architecture

A coordination scheduling layer Coordinator is added between worker and task queue, which assigns tasks according to the status of current cluster tasks, perceives the status of current cluster worker and task, and coordinates the execution, termination and other operations of the whole cluster task.

# stand-alone implementation

# # overall Design

Members: represents all worker in the current cluster

Tasks: is the current task

Coordinator: is our coordinator, responsible for assigning tasks according to members and tasks

Result: is the result of allocation

# # CircularIterator

CircularIterator is our ring opposite iterator, with two methods, one is to add member to add, and one Next returns the next member based on rr.

```go

/ / CircularIterator annular iterator

Type CircularIterator struct {

List [] interface {} / / Save all member variables

Next int

}

/ / Next returns the next element

Func (c * CircularIterator) Next () interface {} {

Item: = c.list [c.next]

C.next = (c.next + 1)% len (c.list)

Return item

}

/ / Add add Task

Func (c * CircularIterator) Add (v interface {}) bool {

For _, item: = range c.list {

If v = = item {

Return false

}

}

C.list = append (c.list, v)

Return true

}

`

# # Member&Task

Member is the worker responsible for task execution. There is an AddTask method and an Execute method responsible for task execution and task addition.

Task identifies a task

```go

/ / members of the Member Task Force

Type Member struct {

Id int

Tasks [] * Task

}

/ / ID returns the current memberID

Func (m * Member) ID () int {

Return m.id

}

/ / AddTask add tasks for member

Func (m * Member) AddTask (t * Task) bool {

For _, task: = range m.tasks {

If task = = t {

Return false

}

}

M.tasks = append (m.tasks, t)

Return true

}

/ / Execute executes tasks

Func (m * Member) Execute () {

For _, task: = range m.tasks {

Fmt.Printf ("Member% d run task% s\ n", m.ID (), task.Execute ())

}

}

/ / Task task

Type Task struct {

Name string

}

/ / Execute returns the result when executing task

Func (t * Task) Execute () string {

Return "Task" + t.name + "run success"

}

`

# # Coordinator

Coordinator is the coordinator, which is responsible for the coordination and scheduling of cluster tasks according to Member and task

```go

/ / Task task

Type Task struct {

Name string

}

/ / Execute returns the result when executing task

Func (t * Task) Execute () string {

Return "Task" + t.name + "run success"

}

/ / Coordinator coordinator

Type Coordinator struct {

Members [] * Member

Tasks [] * Task

}

/ / TaskAssignments assigns tasks to member

Func (c * Coordinator) TaskAssignments () map [int] * Member {

TaskAssignments: = make (map [int] * Member)

/ / build iterator

MemberIt: = c.getMemberIterator ()

For _, task: = range c.tasks {

Member: = memberIt.Next (). (Member)

_, err: = taskAssignments [member.ID ()]

If err = = false {

TaskAssignments [member.ID ()] = member

}

Member.AddTask (task)

}

Return taskAssignments

}

Func (c * Coordinator) getMemberIterator () * CircularIterator {

/ / construct a member queue through the current member

Members: = make ([] interface {}, len (c.members))

For index, member: = range c.members {

Members [index] = member

}

Return NewCircularIterftor (members)

}

/ / AddMember add member group member

Func (c * Coordinator) AddMember (m * Member) bool {

For _, member: = range c.members {

If member = = m {

Return false

}

}

C.members = append (c.members, m)

Return true

}

/ / AddTask add Task

Func (c * Coordinator) AddTask (t * Task) bool {

For _, task: = range c.tasks {

If task = = t {

Return false

}

}

C.tasks = append (c.tasks, t)

Return true

}

`

# # testing

We first create a bunch of member and task, then call coordinator to assign the task and execute the task result

```go

Coordinator: = NewCoordinator ()

For I: = 0; I < 10; iTunes + {

M: = & Member {id: I}

Coordinator.AddMember (m)

}

For I: = 0; I < 30; iTunes + {

T: = & Task {name: fmt.Sprintf ("task% d", I)}

Coordinator.AddTask (t)

}

Result: = coordinator.TaskAssignments ()

For _, member: = range result {

Member.Execute ()

}

`

# # results

You can see that each worker is assigned tasks evenly.

```bash

Member 6 run task Task task 6 run success

Member 6 run task Task task 16 run success

Member 6 run task Task task 26 run success

Member 8 run task Task task 8 run success

Member 8 run task Task task 18 run success

Member 8 run task Task task 28 run success

Member 0 run task Task task 0 run success

Member 0 run task Task task 10 run success

Member 0 run task Task task 20 run success

Member 3 run task Task task 3 run success

Member 3 run task Task task 13 run success

Member 3 run task Task task 23 run success

Member 4 run task Task task 4 run success

Member 4 run task Task task 14 run success

Member 4 run task Task task 24 run success

Member 7 run task Task task 7 run success

Member 7 run task Task task 17 run success

Member 7 run task Task task 27 run success

Member 9 run task Task task 9 run success

Member 9 run task Task task 19 run success

Member 9 run task Task task 29 run success

Member 1 run task Task task 1 run success

Member 1 run task Task task 11 run success

Member 1 run task Task task 21 run success

Member 2 run task Task task 2 run success

Member 2 run task Task task 12 run success

Member 2 run task Task task 22 run success

Member 5 run task Task task 5 run success

Member 5 run task Task task 15 run success

Member 5 run task Task task 25 run success

`

# # complete Code

```go

Package main

Import "fmt"

/ / CircularIterator annular iterator

Type CircularIterator struct {

List [] interface {}

Next int

}

/ / Next returns the next element

Func (c * CircularIterator) Next () interface {} {

Item: = c.list [c.next]

C.next = (c.next + 1)% len (c.list)

Return item

}

/ / Add add Task

Func (c * CircularIterator) Add (v interface {}) bool {

For _, item: = range c.list {

If v = = item {

Return false

}

}

C.list = append (c.list, v)

Return true

}

/ / members of the Member Task Force

Type Member struct {

Id int

Tasks [] * Task

}

/ / ID returns the current memberID

Func (m * Member) ID () int {

Return m.id

}

/ / AddTask add tasks for member

Func (m * Member) AddTask (t * Task) bool {

For _, task: = range m.tasks {

If task = = t {

Return false

}

}

M.tasks = append (m.tasks, t)

Return true

}

/ / Execute executes tasks

Func (m * Member) Execute () {

For _, task: = range m.tasks {

Fmt.Printf ("Member% d run task% s\ n", m.ID (), task.Execute ())

}

}

/ / Task task

Type Task struct {

Name string

}

/ / Execute returns the result when executing task

Func (t * Task) Execute () string {

Return "Task" + t.name + "run success"

}

/ / Coordinator coordinator

Type Coordinator struct {

Members [] * Member

Tasks [] * Task

}

/ / TaskAssignments assigns tasks to member

Func (c * Coordinator) TaskAssignments () map [int] * Member {

TaskAssignments: = make (map [int] * Member)

/ / build iterator

MemberIt: = c.getMemberIterator ()

For _, task: = range c.tasks {

Member: = memberIt.Next (). (Member)

_, err: = taskAssignments [member.ID ()]

If err = = false {

TaskAssignments [member.ID ()] = member

}

Member.AddTask (task)

}

Return taskAssignments

}

Func (c * Coordinator) getMemberIterator () * CircularIterator {

/ / construct a member queue through the current member

Members: = make ([] interface {}, len (c.members))

For index, member: = range c.members {

Members [index] = member

}

Return NewCircularIterftor (members)

}

/ / AddMember add member group member

Func (c * Coordinator) AddMember (m * Member) bool {

For _, member: = range c.members {

If member = = m {

Return false

}

}

C.members = append (c.members, m)

Return true

}

/ / AddTask add Task

Func (c * Coordinator) AddTask (t * Task) bool {

For _, task: = range c.tasks {

If task = = t {

Return false

}

}

C.tasks = append (c.tasks, t)

Return true

}

/ / NewCircularIterftor returns iterator

Func NewCircularIterftor (list [] interface {}) * CircularIterator {

Iterator: = CircularIterator {}

For _, item: = range list {

Iterator.Add (item)

}

Return & iterator

}

/ / NewCoordinator returns the coordinator

Func NewCoordinator () * Coordinator {

C: = Coordinator {}

Return & c

}

Func main () {

Coordinator: = NewCoordinator ()

For I: = 0; I < 10; iTunes + {

M: = & Member {id: I}

Coordinator.AddMember (m)

}

For I: = 0; I < 30; iTunes + {

T: = & Task {name: fmt.Sprintf ("task% d", I)}

Coordinator.AddTask (t)

}

Result: = coordinator.TaskAssignments ()

For _, member: = range result {

Member.Execute ()

}

}

`

Task coordination is a very complex matter. Although the internal task platform realizes task-based combination and appplication, task scheduling allocates a piece, which is still not done. Just simply according to the tree task to simply do some branch task execution, there is time to do it in the future, to continue to study the next module.

The above is how to implement distributed task RR allocation strategy based on ring queue and iterator. If you happen to have similar doubts, please refer to the above analysis to understand. If you want to know more about it, you are welcome to follow the industry information channel.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.