04 курс

Спринт 4

Пакеты hash, crypto. Безопасность информации
5/5
Многопоточность
9/9
Основы многопоточности
Многопоточность в Go
Каналы
Паттерны многопоточности: Генератор, Стоп-кран, Паттерн обработки ошибок
Паттерны многопоточности: Конвейер, Fan-In, Fan-Out, Семафор
Инкремент 15
Код-ревью инкрементов. Спринт 4
Расскажите про свой проект
Заключение
Тема 2/2: Многопоточность → Урок 3/9

Каналы

В прошлом уроке вы познакомились с горутинами и примитивами синхронизации. В этом вы узнаете о ещё одном инструменте для работы с горутинами — каналах (channels).
Каналы позволяют отправлять значения определённого типа между горутинами и могут синхронизировать их работу. Кроме каналов, в этом уроке рассмотрим оператор select и атомарные операции.

Создание каналов

Каналы создаются функциями make(chan тип_элемента) или make(chan тип_элемента, ёмкость_канала). В первом случае создастся небуферизированный канал, а во втором - буферизированный.
Операции чтения и записи в небуферизированный канал ожидают, когда другая горутина соответственно запишет или прочитает данные. Буферизированный канал накапливает данные. Когда канал пуст, горутина встаёт в режим ожидания при чтении. Когда канал полон,горутина встаёт в режим ожидания при записи.
Работа с буферизированным каналом сводится к следующему: горутина записывает данные в канал, пока он не заполнится, а принимающая горутина может читать из этого канала, пока он не станет пустым. Когда канал полон, отправляющая горутина блокируется и ждёт, пока принимающая горутина не заберёт хотя бы одно значение. То же самое происходит и на другой стороне. Принимающая горутина находится в ожидании (блокируется), когда канал пустой, но оживает и начинает читать данные, как только в канал попадает хотя бы одно значение.
GO
// создать канал для значений типа int
ch1 := make(chan int)

// создать буферизированный канал для 5 элементов для значений типа string
ch2 := make(chan string, 5)

// создать канал для структур типа MyType
ch3 := make(chan MyType)
Иногда необходимо получить текущее количество элементов в канале и его ёмкости. Например, если нужно проверить, полон ли канал, чтобы не посылать туда данные, а выполнить другие действия. Для этого используются функции len(ch) и cap(ch).
Оператор ch <- v отправляет значение v в канал ch. Для получения значения из канала используется оператор <- ch. Этот вариант применяется, когда получаемое значение не важно. Если нужно сохранить значение, то слева указывается оператор присваивания v := <-ch.
Чем отличаются операции чтения и записи в буферизированные и небуферизированные каналы?
Саша
Здорово, что ты обращаешь внимание на такие детали! Это очень важный момент. Чтобы разобраться, изучи таблицу. В ней — все важные отличия буферизированных и небуферизированных каналов.
Теперь перейдём к примеру. Предположим, в программе есть два канала. В один записываются числа, в другой — отправляется сумма двух чисел из первого канала.
GO
package main

import (
    "fmt"
    "time"
)

func main() {
    chIn := make(chan int, 50)
    chOut := make(chan int)

    go func() {
        // горутина берёт числа из chIn
        for {
            left := <-chIn
            right := <-chIn
            // получаем два числа из chIn и записываем их сумму в chOut
            chOut <- left + right
        }
    }()

    go func() {
        // горутина берёт числа из chOut
        for {
            s := <-chOut
            // пусть обработка значений из chOut занимает какое-то время
            time.Sleep(20 * time.Millisecond)
            if s%10 == 1 {
                fmt.Printf("%d ", s)
            }
        }
    }()
    // отправляем сто чисел в канал chIn
    for i := 0; i < 100; i++ {
        chIn <- i
    }
    fmt.Printf("# ")
    time.Sleep(1000 * time.Millisecond)
}
Вывод программы будет отличаться для разных значений ёмкости канала chIn.
TXT
// для chIn := make(chan int)
1 21 41 61 81 101 121 141 161 181 # 

// для chIn := make(chan int, 50)
1 21 41 61 81 # 101 121 141 161 181 

// для chIn := make(chan int, 100)
# 1 21 41 61 81 101 121 141 161 181
Чем больше ёмкость канала, тем быстрее происходит отправка чисел. При ёмкости равной 100 все числа сразу помещаются в канал.
Если горутина больше не будет отправлять значения в канал и на это нужно указать принимающим горутинам, то канал следует закрыть. Для этого нужно вызвать функцию close(ch). Когда другие горутины попытаются получить значение из закрытого канала, им будут возвращаться нулевые значения.
Если горутина отправит данные в закрытый канал, это приведёт к панике. Поэтому каналы закрывать не нужно, если нет увереннности, что туда не будут записываться данные. Когда канал перестанет использоваться и на него не будет активных ссылок, сборщик мусора освободит занимаемую каналом память независимо от того, закрыт он или нет.
Проверить, закрыт канал или нет, можно так: i, ok := <-ch. Если канал закрыт, значение ok будет равно false. Получение двух значений используется, когда необходимо отличать нулевое значение, отправленное в канал, от нулевого значения, возвращаемого из закрытого канала.
GO
package main

import (
    "fmt"
)

func main() {
    ch := make(chan int, 1)
    
    // так как ёмкость канала больше 0, то можно записать
    // одно значение не ожидая, когда оно прочитается
    ch <- 10

    // попробуем закрыть канал, в котором есть значение
    close(ch)

    fmt.Println("len =", len(ch), "cap =", cap(ch))
    
    v, closed := <-ch
    fmt.Println(v, closed, "len =", len(ch))
    
    v, closed = <-ch
    fmt.Println(v, closed, "len =", len(ch))
}
Вывод программы показывает, что канал закрылся только после того, как было прочитано значение и его размер стал равен 0.
TXT
len = 1 cap = 1
10 true len = 0
0 false len = 0
Напомним, что канал создаётся функцией make(). Если вы просто определите переменную типа chan, она, как и в случае со слайсами, будет равна nil. Таким образом, есть три состояния канала:
  • равен nil;
  • инициализирован и не закрыт;
  • закрыт.
Изучите таблицу. В ней — результаты основных операций для каждого состояния:
image
Если ни одна горутина не может продолжить работу, рантайм завершает программу с ошибкой fatal error: all goroutines are asleep - deadlock!.
Рассмотрим использование каналов на практике. Допустим, вы разрабатываете сервис, который меняет размер изображения. Ресайзинг картинок — долгая операция. Возникла идея запустить N горутин, где N равно количеству ядер процессора. По задумке каждая горутина будет получать картинку из очереди и менять размер.
В примере рассмотрим три сущности:
  • Queue — очередь задач, которую реализуем с помощью каналов;
  • Resizer — модуль, который ресайзит картинки;
  • Worker — воркер, который вытаскивает задачу из очереди и обрабатывает её.
GO
package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

// Task содержит имя файла для конвертации
type Task struct {
    Filename string
}

// Queue - это очередь задач
type Queue struct {
    ch chan *Task
}

func NewQueue() *Queue {
    return &Queue{
        ch: make(chan *Task, 1),
    }
}

func (q *Queue) Push(t *Task) {
    // добавляем задачу в очередь
    q.ch <- t
}

func (q *Queue) PopWait() *Task {
    // получаем задачу
    return <-q.ch
}

type Resizer struct {
    Width  uint32
    Height uint32
}

func NewResizer(w, h uint32) *Resizer {
    return &Resizer{
        Width:  w,
        Height: h,
    }
}

func (r *Resizer) Resize(filename string) error {
    // пропустим реализацию
    time.Sleep(100 * time.Millisecond)
    return nil
}

type Worker struct {
    id      int
    queue   *Queue
    resizer *Resizer
}

func NewWorker(id int, queue *Queue, resizer *Resizer) *Worker {
    w := Worker{
        id:      id,
        queue:   queue,
        resizer: resizer,
    }
    return &w
}

func (w *Worker) Loop() {
    for {
        t := w.queue.PopWait()

        err := w.resizer.Resize(t.Filename)
        if err != nil {
            fmt.Printf("error: %v\n", err)
            continue
        }

        fmt.Printf("worker #%d resized %s\n", w.id, t.Filename)
    }
}

func main() {
    queue := NewQueue()

    for i := 0; i < runtime.NumCPU(); i++ {
        w := NewWorker(i, queue, NewResizer(1024, 1024))
        go w.Loop()
    }

    for i := 0; i < 50; i++ {
        imagefile := fmt.Sprintf("gopher%d.jpg", i)
        queue.Push(&Task{Filename: imagefile})
    }

    time.Sleep(2 * time.Second)
}
Таким образом, используя каналы, удалось реализовать очередь с минимальным количеством кода.

Задание 1/7

Проверьте себя. Не запуская код, определите, каким будет результат его выполнения.
GO
func main() {
    ch := make(chan int)
    ch <- 10
    fmt.Println(<-ch)
}

Задание 2/7

При запуске следующей программы происходит ошибка (взаимная блокировка). Найдите причину и исправьте её. Программа должна выводить 7.
Подсказка: Нужно создать канал функцией make().
GO
package main

import "fmt"

func main() {
    var ch chan int

    go func() {
        ch <- 7
    }()
    v := <-ch
    fmt.Println(v)
}

Операторы для работы с каналами

Оператор range позволяет в цикле принимать значения из канала до тех пор, пока он не будет закрыт. Например, горутина должна постоянно получать значения из канала и обрабатывать их определённым образом. range используется в канале, который закрывается, иначе он будет эквивалентен бесконечному циклу.
GO
for i := range ch {
    // тело цикла
}
Этот оператор эквивалентен такому циклу:
GO
for {
    i, ok := <-ch
    if !ok {
        break
    }
    // тело цикла
}
Пример кода:
GO
package main

import "fmt"

func fibonacci(n int, ch chan int) {
    x, y := 0, 1
    for i := 0; i < n; i++ {
        ch <- x // посылаем значения в канал
        x, y = y, x+y
    }

    close(ch) // закрываем канал
}

func main() {
    ch := make(chan int, 7)
    // специально делаем буфер канала меньше,
    // чем количество чисел Фибоначчи
    go fibonacci(15, ch)

    for i := range ch {
        // считываем значения из канала, пока он не будет закрыт
        fmt.Printf("%d ", i)
    }
}
Программа выведет: 
0 1 1 2 3 5 8 13 21 34 55 89 144 233 377
Как уже было сказано, горутина может заблокироваться и ждать готовности канала. Но что делать, если надо ждать сразу несколько каналов: в одном — результата функции, в другом — сигнала, что нужно завершить программу? Можно использовать оператор select.
Изучите его синтаксис:
GO
select {
case x := <-ch1:
    // сценарий выполнится, если быстрее всего новое значение окажется в канале ch1
case y := <-ch2:
    // сценарий выполнится, если быстрее всего новое значение окажется в канале ch2
case ch3 <- z:
    // сценарий выполнится, если быстрее отправим значение в канал ch3
}
Если не выполняется ни одно из условий, select переходит в режим ожидания. Когда без блокировки нужно проверить, есть ли новые сообщения, необходимо добавить секцию default. Она выполнится вместо перехода в режим ожидания, и управление перейдёт к следующему после select коду.
GO
select {
case x := <-ch1:
    // ...
case ch3 <- z:
    // ...
default:
    // код выполнится, если не подошёл ни один из вариантов выше
    // ...
}
Важно
В операторе select выполняется только один из блоков case. Если несколько условий срабатывают одновременно, то вариант выбирается случайным образом: порядок case не имеет значения. Если нужно постоянно проверять условия оператора, поместите select в цикл.
Оператор select позволяет использовать break внутри секций, но break обеспечивает только выход из select. Чтобы прервать цикл с select, применяют следующие подходы:
  • выход из функции по return;
  • изменение условия, которое проверяется в цикле;
  • break с указанием метки цикла.
Пример:
GO
package main

import "fmt"

func fibonacci(ch chan int, quit chan bool) {
    x, y := 0, 1
loop: // метка цикла
    for {
        select {
        case ch <- x: // ждём, когда заберут значение из канала,
                      // чтобы сгенерировать следующее
            x, y = y, x+y
        case <-quit:  // параллельно ждём сигнала об окончании работы
            break loop
        }
    }
    fmt.Println("Выход")
}

func main() {
    ch := make(chan int)
    quit := make(chan bool)

    go func() {
        for i := 0; i < 15; i++ {
            fmt.Println(<-ch)
        }
        // подаём сигнал об окончании работы
        quit <- true
    }()

    fibonacci(ch, quit)
}
select может принимать данные из нескольких каналов, но некоторые из этих каналов могут закрыться в процессе работы. Так как из закрытого канала возвращаются нулевые значения, select по-прежнему будет выполнять соответствующий ему блок case. Чтобы этого избежать, можно присвоить закрытому каналу nil — тогда оператор select будет пропускать варианты с нулевыми каналами.
GO
for {
    select {
    // этот вариант будет пропускаться, если ch == nil
    case v, ok := <-ch:
        if !ok {
            ch = nil
            continue
        }
        // ...
    }
}
Разберём, в каких случаях можно использовать закрытые каналы. Предположим, надо завершить работу программы и оповестить об этом горутины, в которых есть оператор select. Для этого создадим канал done := make(chan struct{}). Его будут читать все горутины.
Каналы с пустой структурой часто используются для оповещений о событиях. Если отправить пустую структуру в канал done, то его получит только одна горутина. Но нужно отправить сигнал всем горутинам. Так как закрытый канал отдаёт нулевые значения в неблокирующем режиме, достаточно закрыть канал, чтобы оповестить о событии все потоки.
GO
var done = make(chan struct{})

func worker(wg *sync.WaitGroup, i int) {
    for {
        select {
        case <-done:
            fmt.Println("Завершаем", i)
            wg.Done()
            return
        default:
            fmt.Println(i)
        }
        time.Sleep(50 * time.Millisecond)
    }
}

func main() {
    var wg sync.WaitGroup

    // создаём горутины
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go worker(&wg, i)
    }
    time.Sleep(1 * time.Second)
    // сообщаем горутинам о завершении работы
    close(done)
    // ждём завершения всех горутин
    wg.Wait()
}

Задание 3/7

Ниже приведена программа, которая получает значения из одного канала и отправляет удвоенные значения в другой канал. Но есть одна проблема:иногда последние значения не печатаются. Пример вывода: 0 2 4 6 8 10 12 14 16 18 20 22 24. Исправьте код, чтобы он работал правильно. Изменения нужно внести только в горутины.
Подсказка: Можно обойтись без select и использовать операторы range.
GO
package main

import "fmt"

func main() {
    chIn := make(chan int)
    chOut := make(chan int)
    quit := make(chan struct{})

    go func() {
        for i := 0; i < 15; i++ {
            chIn <- i
        }
        close(quit)
    }()
    go func(){
           var x int
        for {
            select {
            case x = <-chIn:
                chOut <- x * 2
            case <-quit:
                return
            }
        }
    }()
    go func() {
        for {
            fmt.Printf("%d ", <-chOut)
        }
    }()
    <-quit
}

Однонаправленные каналы

Часто каналы передаются в аргументах функций. Внутри функций каналы используются только для отправки или получения данных. Если передаваемый в аргументе канал будет использоваться только для чтения или записи, то для параметра функции можно указать однонаправленный тип канала:
  • <-chan — только для получения данных. Например, par <-chan int;
  • chan<- — только для отправки данных. Например, par chan<- string.
Пример использования однонаправленных каналов:
GO
// Generate отправляет в канал out односимвольные строки.
func Generate(out chan<- string) {
    for ch := 'a'; ch <= 'z'; ch++ {
        out <- string([]rune{ch})
    }
    close(out)
}

// Process читает строки из канала in, переводит их в верхний регистр
// и отправляет в канал out.
func Process(in <-chan string, out chan<- string) {
    for v := range in {
        out <- strings.ToUpper(v)
    }
    close(out)
}

func main() {
    lower := make(chan string)
    upper := make(chan string)
    go Generate(lower)
    go Process(lower, upper)

    // выводим строки из канала upper по мере получения
    for s := range upper {
        fmt.Print(s)
    }
}
Программа выведет: 
ABCDEFGHIJKLMNOPQRSTUVWXYZ

Задание 4/7

Представьте, что вы написали простенький код с каналами. Пока вы пили чай, ваша кошка прошлась по клавиатуре и удалила часть кода: надеемся, она сделала это неспециально. Теперь вам нужно восстановить текст программы.
Подсказка: Были удалены следующие фрагменты: loop, <-in2, out, ch1, ch2, chout.
GO
package main

import "fmt"

func process(in1, in2 <-chan int, out chan<- int) {
loop:
    for {
        select {
        case x, ok := <-in1:
            if !ok {
                break 
            }
            out <- x * 2
        case x, ok := :
            if !ok {
                break loop
            }
            out <- x * 3
        }
    }
    close()
}

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)
    chout := make(chan int)

    go func() {
        for i := 0; i <= 20; i++ {
            select {
            case <- i:
            case ch2 <- i:
            }
        }
        close(ch1)
        close()
    }()
    go process(ch1, ch2, )
    for i := range chout {
        fmt.Printf("%d ", i)
    }
}

Атомарные операции

Теперь разберём, что такое атомарные операции. Но для начала рассмотрим пример, в котором будем увеличивать значение общего счётчика одновременно из разных горутин:
GO
package main

import (
    "fmt"
    "sync"
)

func count() {
    var counter int64

    var wg sync.WaitGroup

    // горутины увеличивают значение счётчика
    for i := 0; i < 25; i++ {
        wg.Add(1)
        go func() {
            for i := 0; i < 2000; i++ {
                counter++
            }
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Printf("%d ", counter)
}

func main() {
    // делаем несколько попыток
    for i := 0; i < 5; i++ {
        count()
    }
}
Если запустить эту программу, то она выведет примерно следующее:
TXT
41131 47906 42555 50000 47864
Это происходит из-за поэтапного увеличения счётчика:
  1. Берём текущее значение переменной.
  2. Увеличиваем его на 1.
  3. Записываем новое значение в переменную.
Эти операции очень быстрые. Но если имеется много горутин, которые постоянно меняют значение переменной, то может возникнут такая ситуация:
  • горутина 1: получает значение (1);
  • горутина 1: увеличивает его на 1 (2);
  • горутина 2: получает значение (1);
  • горутина 1: записывает значение (2);
  • горутина 2: увеличивает значение на 1 (2);
  • горутина 2: записывает значение (2).
Обе горутины увеличивали счётчик, но он изменился только на 1. В таких случаях нужно использовать мьютексы. Однако при очень большом количестве операций, мьютексы будут замедлять работу программы. Поэтому для изменения целочисленных значений были добавлены более быстрые атомарные операции.
Атомарные операции — это такие операции, которые выполняются за один шаг относительно других горутин. Для корректной работы рассмотренного примера необходимо, чтобы изменение счётчика было атомарной операцией. Только в таком случае можно избежать одновременного увеличения его двумя горутинами.
В пакете sync/atomic стандартной библиотеки реализованы атомарные операции для типов: int32, int64, uint32, uint64, uintptr, unsafe.Pointer.
Ниже перечислены функции для типа int64. Пакет sync/atomic содержит аналогичные функции и для других типов.
  • AddInt64(addr *int64, delta int64) — добавить delta к значению addr. Функция возвращает новое значение.
GO
    var addr int64
    atomic.AddInt64(&addr, 1)
  • CompareAndSwapInt64(addr *int64, old int64, new int64) — сравнить значение addr с old и заменить на new, если они равны. Функция возвращает true, если значение addr было заменено.
GO
    var addr int64
    atomic.CompareAndSwapInt64(&addr, 0, 77)
  • SwapInt64(addr *int64, new int64) — изменить значение addr на new. Функция возвращает предыдущее значение.
GO
    var addr int64
    old := atomic.SwapInt64(&addr, 3)
  • LoadInt64(addr *int64) — разыменовать указатель и возвратить значение.
GO
    var addr int64
    val := atomic.LoadInt64(&addr)
  • StoreInt64(addr *int64, val int64) — записать значение по указателю.
GO
    var addr int64
    atomic.StoreInt64(&addr, 5)
Переделаем пример с счётчиком, используя атомарные операции.
GO
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

func count() {
    var counter int64

    var wg sync.WaitGroup

    // горутины увеличивают значение счётчика
    for i := 0; i < 25; i++ {
        wg.Add(1)
        go func() {
            for i := 0; i < 2000; i++ {
                atomic.AddInt64(&counter, 1)
            }
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Printf("%d ", atomic.LoadInt64(&counter))
}

func main() {
    // делаем несколько попыток
    for i := 0; i < 5; i++ {
        count()
    }
}
Сейчас программа выводит ожидаемый результат:
TXT
50000 50000 50000 50000 50000
Также в пакете определены типы-обёртки: atomic.Bool, atomic.Int32, atomic.Int64, atomic.Pointerи так далее — с соответствующими методами. Перечислим методы для типа atomic.Bool:
GO
// CompareAndSwap сравнивает значение с `old` и в случае равенства заменяет
// на `new`. Метод возвращает `true`, если значение было заменено.
func (x *Bool) CompareAndSwap(old, new bool) (swapped bool)

// Load возвращает логическое значение.
func (x *Bool) Load() bool

// Store записывает значение в переменную.
func (x *Bool) Store(val bool)

// Swap изменяет значение на указанное и возвращает предыдущее значение.
func (x *Bool) Swap(new bool) (old bool)
Перепишем пример со счётчиком используя тип atomic.Int64.
GO
package main

import (
    "fmt"
    "sync"
    "sync/atomic"
)

func count() {
    var counter atomic.Int64

    var wg sync.WaitGroup

    // горутины увеличивают значение счётчика
    for i := 0; i < 25; i++ {
        wg.Add(1)
        go func() {
            for i := 0; i < 2000; i++ {
                counter.Add(1)
            }
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Printf("%d ", counter.Load())
}

func main() {
    // делаем несколько попыток
    for i := 0; i < 5; i++ {
        count()
    }
}
Согласитесь, что код стал выглядеть приятнее.

Задание 5/7

Перед вами код, в котором горутины увеличивают значение счётчика. Чаще всего программа выводит неверное значение. Исправьте код так, чтобы значение счётчика было правильным.
Подсказка: Используйте атомарную операцию atomic.AddInt64() или тип atomic.Int64.
GO
package main

import (
    "fmt"
    "sync"
)

var counter int64

func worker(wg *sync.WaitGroup) {
    for i := 0; i < 10000; i++ {
        counter++
    }
    wg.Done()
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 20; i++ {
        wg.Add(1)
        go worker(&wg)
    }
    wg.Wait()
    // программа должна выводить 200000
    fmt.Println(counter)
}

Задание 6/7

Какая функция пакета atomic атомарно реализует то же, что функция Foo?
GO
func Foo(i *int64, v int64) int64 {
    ret = *i
    *i = v
    return ret
}

Задание 7/7

Не запуская код, определите, что выведет программа.
GO
func main() {
    ch := make(chan string, 1)
    ch <- "OK"
    fmt.Println(<-ch)
}
В этом уроке вы изучили основные принципы работы с каналами в Go. Рассмотрели, как создавать каналы, отправлять и получать данные через них, а также использовать операторы select и range для управления потоком данных. Вы изучили атомарные операции и узнали, как их применять в горутинах. Все эти знания пригодятся при создании эффективных и надежные мнопоточных программ.

Дополнительные материалы