04 курс

Спринт 4

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

Многопоточность в Go

Горутины

Горутина — это функция, которая выполняется параллельно с другими функциями. В Go выполнить функцию — это myFunc(), a выполнить функцию в новой горутине — это go myFunc(). Да, так просто. 🙂 При вызове go myFunc() программа не дожидается завершения вызываемой функции, а выполняет идущий далее код.
GO
package main

import (
    "fmt"
    "time"
)

func say(s string) {
    for i := 0; i < 5; i++ {
        time.Sleep(time.Millisecond)        
        fmt.Print(s + ` `)
    }
}

func main() {
    // запустим две горутины с функцией say
    go say(`hello`)
    go say(`world`)
    // обычный вызов функции say
    say(`bye`)
}
Программа выведет примерно следующее:
TXT
world hello bye hello bye world bye hello world world hello bye bye
Как видите, горутины выполняются в произвольной последовательности. По умолчанию программа завершает работу, не дожидаясь окончательного выполнения горутин.
Если горутина содержит небольшое количество кода, который больше нигде не нужен, то можно определить её с помощью анонимной функции.
GO
package main

import (
    "fmt"
    "time"
)

func main() {
    for _, v := range []string{"hello", "world", "bye"} {
        go func(s string) {
            for i := 0; i < 5; i++ {
                fmt.Print(s + ` `)
                time.Sleep(50 * time.Millisecond)
            }
        }(v)
    }
    time.Sleep(time.Second)
}
Важно
Когда вы создаёте горутины в цикле с помощью анонимной функции, не используйте в этой функции переменную цикла. Иначе у горутин будет неожиданное значение переменной. Передавайте переменную цикла в горутину в виде параметра или определяйте дополнительную переменную.
Переделаем предыдущий пример, чтобы продемонстрировать эту ошибку.
GO
package main

import (
    "fmt"
    "time"
)

func main() {
    for _, v := range []string{"hello", "world", "bye"} {
        go func() {
            for i := 0; i < 5; i++ {
                fmt.Print(v + ` `)
                time.Sleep(time.Millisecond)
            }
        }()
        time.Sleep(time.Millisecond)
    }
    time.Sleep(time.Second)
}
Результат работы этой программы будет примерно таким:
TXT
hello world hello world world bye bye bye bye bye bye bye bye bye bye
Дело в том, что переменная цикла может меняться значительно быстрее, чем выполнятся горутины. Когда запрашивается значение переменной, оно может оказаться любым из возможных вариантов. Поэтому нужно передавать переменные цикла в параметрах анонимной функции или использовать ещё одну переменную в теле цикла. Во втором случае на каждой итерации создаётся новая переменная, которая захватывается.
GO
for _, v := range []string{"hello", "world", "bye"} {
    s := v
    go func() {
        for i := 0; i < 5; i++ {
            fmt.Print(s + ` `)
            time.Sleep(50 * time.Millisecond)
        }
    }()
}
Иногда можно встретить вариант, где на каждой итерации создаётся переменная с именем, как у переменной цикла. Код может выглядеть странно, но это не влияет на работу горутин. Они получают корректное значение:
GO
for _, v := range []string{"hello", "world", "bye"} {
    v := v
    go func() {
        for i := 0; i < 5; i++ {
            fmt.Print(v + ` `)
            time.Sleep(50 * time.Millisecond)
        }
    }()
}

Планировщик горутин

Как работает планировщик горутин?
Саша
Представь, что есть задача — выкопать канаву. У тебя есть команда из десяти человек, но вам дали только три лопаты. Перед началом работы, вы будете думать, как выкопать канаву быстрее и эффективнее. Примерно этим и занимается планировщик Go.
Планировщик горутин — один из элементов рантайма языка. Он отвечает за выполнение горутин в потоках операционной системы. Планировщик равномерно распределяет процессорное время между горутинами, учитывая их сихронизацию и системные вызовы.
Планировщик горутин работает с тремя типами объектов:
  • Логический процессор. Это ресурс, который объединяет поток операционной системы и очередь горутин. Для каждого логического процессора планировщик достаёт одну горутину из очереди и запускает её в потоке используя контекст логического процессора.
  • Поток операционной системы. Это сущность операционной системы, в которой происходит процесс выполнения программного кода. Логический процессор закреплён за определённым потоком. Если горутина сделала системный вызов, который переводит поток в режим ожидания результата, например, чтение файла, то планировщик создаёт дополнительный поток операционной системы. На нём будут запускаться следующие горутины из очереди до разблокировки основного потока.
  • Горутина.
По умолчанию количество логических процессоров берётся из значения переменной среды GOMAXPROCS и равно количеству логических ядер компьютера. Например, если у вас восьмиядерный компьютер, планировщик будет распределять горутины по восьми потокам ОС.
Теперь разберём, как можно влиять на работу горутин. Создатели Go стремятся использовать в планировщике универсальные алгоритмы и настройки. Поэтому для управления планировщиком используют следующие функции пакета runtime:
  • GOMAXPROCS(n int) int — изменяет значение GOMAXPROCS и возвращает количество логических процессоров, которое было установлено до вызова функции. Если значение GOMAXPROCS больше, чем количество доступных ядер, скорость работы может не измениться. Чтобы узнать количество логических процессоров, укажите в параметре 0.
  • Gosched() — приостанавливает текущую горутину, чтобы планировщик переключился на другую. Работа этой горутины возобновится в порядке очерёдности. В версии Go 1.14 появился вытесняющий планировщик: он вытеснит зависшую по процессорному времени горутину. Таким образом, вызывать переключение горутин вручную не нужно.
  • LockOSThread() — привязывает горутину к текущему потоку операционной системы. Эта функция пригодится при запуске кода на языке C, который требует выполнения в одном и том же потоке.
Стоит упомянуть ещё две функции:
  • NumCPU() int возвращает количество логических ядер процессора. Можно использовать это значение для ограничения количества горутин.
  • NumGoroutine() int возвращает количество запущенных горутин. Эта функция будет полезна при наблюдении за работой программы.
Рассмотрим пример, демонстрирующий работу этих функций:
GO
package main

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

func main() {
    fmt.Println("Ядер:", runtime.NumCPU())
    fmt.Println("Логических процессоров:", runtime.GOMAXPROCS(2), 
        "Горутин:", runtime.NumGoroutine())
    go func() {
        time.Sleep(100 * time.Millisecond)
    }()
    fmt.Println("Логических процессоров:", runtime.GOMAXPROCS(0), 
        "Горутин:", runtime.NumGoroutine())
}
На 8-ядерном процессоре получим вывод:
TXT
Ядер: 8
Логических процессоров: 8 Горутин: 1
Логических процессоров: 2 Горутин: 2
Не удивляйтесь, что количество горутин отображается на одну больше, чем было запущено. Дело в том, что при запуске создаётся основная горутина, в которой выполняется код программы.

Типы sync.WaitGroup и sync.Once

Что делать, если я запустил несколько горутин и хочу в основной функции дождаться выполнения их всех?
Саша
Для этого в Go есть переменная типа sync.WaitGroup.
Переменная типа sync.WaitGroup содержит счётчик горутин, завершения которых нужно ждать. Функциональность WaitGroup очень проста, состоит всего из трёх методов:
  • (*WaitGroup) Add(delta int) — изменить значение счётчика на указанную величину;
  • (*WaitGroup) Done() — уменьшить значение счётчика на единицу;
  • (*WaitGroup) Wait() — ожидать, когда значение счётчика будет равно нулю.
Вот пример, в котором программа дожидается завершения запущенных горутин:
GO
package main

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

func main() {
    var wg sync.WaitGroup
    const n = 5

    for i := 0; i < n; i++ {
        wg.Add(1) // инкрементируем счётчик, сколько горутин нужно подождать

        go func(i int) {
            time.Sleep(100 * time.Millisecond)
            fmt.Printf("hi %d\n", i)
            // уменьшаем счётчик, когда горутина завершает работу
            wg.Done()
        }(i)
    }

    wg.Wait() // ждём все горутины
    fmt.Println("Всё готово")
}
Порядок выполнения горутин предсказать нельзя, но все они завершат свою работу. Вот один из результатов работы программы:
TXT
hi 0
hi 3
hi 4
hi 2
hi 1
Всё готово

Задание 1/4

Рассмотрите код и определите, каким будет результат работы программы. Выберите вариант ответа, не запуская пример.
Подсказка: суммарное значение параметров во всех Add() должно совпадать с количеством вызовов Done().
GO
func main() {
    var wg sync.WaitGroup

    n := 10
    wg.Add(n)
    for i := 0; i < n-1; i++ {
        go func(v int) {
            fmt.Println(v)
            wg.Done()
        }(i)
    }
    wg.Wait()
}
Что делать, если нужно выполнить действия только один раз? Например, вызвать функцию инициализации при первом обращении к структуре.
Саша
Для этого достаточно определить переменную типа sync.Once и вызвать для неё метод Do(), который в параметре принимает функцию инициализации. Вызов этого метода гарантирует, что функция будет выполнена только один раз. Перейдём к примеру.
Напишем код, который вычитывает конфиг (англ. config) и инициализирует структуру при первом обращении к ней:
GO
package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

type Config struct {
    once sync.Once
    vals map[string]string
}

func (c *Config) Get(k string) (string, bool) {
    c.once.Do(func() {
        // эта инициализация выполнится только один раз
        c.vals = map[string]string{
            "host": "127.0.0.1",
            "port": fmt.Sprintf("%d", rand.Intn(65535)),
        }
    })

    v, ok := c.vals[k]
    return v, ok
}

func main() {
    var cfg Config

    keys := []string{"host", "port", "port", "host", "port"}
    for _, k := range keys {
        go func(k string) {
            // в одной из горутин произойдёт инициализация cfg
            // остальные горутины будут ждать завершения инициализации
            v, ok := cfg.Get(k)
            if !ok {
                return
            }
            fmt.Printf("%s = %s\n", k, v)
        }(k)
    }

    time.Sleep(1 * time.Second)
}
Использование переменной типа sync.Once позволяет легко реализовать одноразовое выполнение определённых действий.

Примитивы синхронизации

Горутины могут использовать одни и те же переменные или ресурсы во время работы. Если операции сводятся только к получению значений, они не нарушают целостность данных. Операции, которые можно выполнять одновременно в разных горутинах, называются потокобезопасными. Однако ситуация усложняется, когда горутины могут изменять данные.
Попробуйте запустить код, который заполняет мапу в нескольких горутинах.
GO
package main

import (
    "fmt"
    "time"
)

func main() {
    m := make(map[int]int)

    for i := 0; i < 5; i++ {
        go func() {
            for j := 0; j < 1000; j++ {
                if _, ok := m[j]; !ok {
                    m[j] = j
                }
            }
        }()
    }
    time.Sleep(1 * time.Second)
    fmt.Println(len(m))
}
С большой долей вероятности вы получите ошибку concurrent map writes или concurrent map read and map write, так как эта операция не потокобезопасна. Для решения этой проблемы нужно исключить другие операции с мапой. В этом помогает синхронизация горутин. Механизмы, которые позволяют горутине на время получить эксклюзивный доступ к данным, называются примитивами синхронизации.
Для синхронизации горутин можно использовать три примитива пакета sync:
  • sync.Mutex — примитив, реализующий мьютекс в Go. Мьютекс — это механизм, который позволяет выполнить критические участки кода только одной горутиной;
  • sync.RWMutex — особый вид мьютекса, который позволяет одновременно выполняться либо произвольному количеству операций чтения, либо одной операции записи;
  • sync.Cond — переменная условия, которая останавливает горутину до получения сигнала.
Рассмотрим каждый элемент подробнее.

Тип sync.Mutex

Мьютексы применяются, если горутина читает или изменяет данные, которые могут использоваться другим потоком. Для переменной типа sync.Mutex можно вызвать два метода:
  • (m *Mutex) Lock() — блокирует мьютекс. Занять мьютекс может только одна горутина. Если другие горутины вызовут метод Lock() для занятого мьютекса, они будут ждать, пока он освободится.
  • (m *Mutex) Unlock() — разблокирует мьютекс. Горутина должна освободить мьютекс сразу после того, как она закончила работу с общим ресурсом.
Если вы применяете мьютексы для сихронизации чтения и записи определённых данных, следите за тем, чтобы не было горутин, работающих с этими данными без мьютекса. Иначе вы также будете получать ошибки конкурентного доступа.
Покажем на примере, как используется sync.Mutex. Предположим, что есть два типа горутин. Одни постоянно меняют мапу, а другие читают. Использование мьютекса синхронизирует все обращения к мапе: только одна горутина может читать или менять её.
GO
package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

func main() {
    var m sync.Mutex
    cache := map[int]int{}

    // горутины, которые изменяют мапу
    for i := 0; i < 10; i++ {
        go func() {
            for {
                m.Lock()
                cache[rand.Intn(5)] = rand.Intn(100)
                m.Unlock()
                time.Sleep(time.Second / 20)
            }
        }()
    }

    // горутины, которые читают мапу
    for i := 0; i < 10; i++ {
        go func() {
            for {
                m.Lock()
                fmt.Printf("%#v\n", cache)
                m.Unlock()
                time.Sleep(time.Second / 100)
            }
        }()
    }

    time.Sleep(1 * time.Second)
}
У такого решения есть недостаток: оно неоптимально с точки зрения производительности. Из мапы данные читаются чаще, чем записываются в неё. Представьте такой случай: данные меняются очень редко, но обращаться к мапе мьютекс всё равно позволяет только одной горутине. Эту проблему решает тип sync.RWMutex.

Тип sync.RWMutex

Этот тип мьютекса позволяет выполнять либо произвольное количество операций чтения, либо одну операцию записи. При этом нельзя выполнять две операции записи или одновременно запись и чтение.
Если горутине нужно изменить данные, то вызывается метод Lock(). Если горутина собирается читать данные, то она вызывает метод RLock(). Метод RLock() не даёт начать запись пока не будут завершены все операции чтения. Для разблокировки следует использовать соответствующие методы — Unlock() или RUnlock().
Переделаем наш код таким образом, чтобы читать из мапы могли несколько горутин, а записывать в неё — только одна.
GO
package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

func main() {
    // меняем тип мьютекса
    var m sync.RWMutex
    cache := map[int]int{}

    for i := 0; i < 10; i++ {
        go func() {
            for {
                // здесь остаются блокировки на запись
                m.Lock()
                cache[rand.Intn(5)] = rand.Intn(100)
                m.Unlock()
                time.Sleep(time.Second / 20)
            }
        }()
    }

    for i := 0; i < 10; i++ {
        go func() {
            for {
                // при чтении используем Rlock() и RUnlock()
                m.RLock()
                fmt.Printf("%#v\n", cache)
                m.RUnlock()
                time.Sleep(time.Second / 100)
            }
        }()
    }

    time.Sleep(1 * time.Second)
}
Обратите внимание! Для снятия блокировки на чтение и запись необходимо использовать разные методы: при RLock() используется RUnlock(), а при Lock() — Unlock().

Задание 2/4

Рассмотрите код. Каким будет результат его работы? Выберите правильный вариант.
Подсказка: похожий пример мы рассматривали в начале главы о примитивах синхронизации.
GO
func main() {
    m := make(map[int]int)

    for i := 0; i < 100; i++ {
        go func(v int) {
            m[v] = 1
        }(i)
    }
    time.Sleep(1 * time.Second)
    fmt.Println(len(m))
}

Задание 3/4

Исправьте код из предыдущего задания так, чтобы программа выполнялась корректно и выводила 100. Подсказка: используйте sync.Mutex при изменении мапы.

Тип sync.Cond

Представьте, что горутина ждёт изменения определённых условий. Чтобы она постоянно не делала проверки, можно «усыпить» её до получения сигнала. Для этого используется переменная условия (от англ. condition variable) — примитив синхронизации, который блокирует один или несколько потоков до получения сигнала от другого потока.
Мьютекс блокирует другие горутины, а переменная условия работает по обратному принципу. Одна горутина переменной условия блокирует сама себя, а другим горутинам нужно её «пробудить», то есть освободить.
Переменная типа sync.Cond содержит локер-поле L типа sync.Locker, значениями которого выступают типы *sync.Mutex или *sync.RWMutex. Значение L передаётся в функции sync.NewCond(l Locker) *Cond.
Есть три метода для работы с переменной условия:
  • (*Cond) Wait() — разблокирует локер L и вводит текущую горутину в режим ожидания до получения сигнала. При получении сигнала локер L блокируется, и выполняется следующий после Wait код. Горутина должна заблокировать L перед вызовом Wait. Чаще всего встречаются такие варианты использования этого метода:
GO
// вариант 1
c.L.Lock()
c.Wait()
// производим нужные действия
// ...
c.L.Unlock()

// вариант 2
c.L.Lock()
for !condition() {
    c.Wait()
}
// производим нужные действия
// ...
c.L.Unlock()
  • (*Cond) Signal() — разблокирует одну из ожидающих горутин, если такие есть.
  • (*Cond) Broadcast() — разблокирует все горутины в очереди.
Почему вызывается Lock() перед Wait(), а Unlock() после? И обязательны ли эти вызовы?
Саша
Дело в том, что в начале метода Wait() происходит разблокировка мьютекса L и текущая горутина встаёт в режим ожидания сигнала. Когда приходит сигнал, мьютекс L блокируется и Wait() завершает работу. Именно поэтому мы должны лочить мьютекс перед вызовом Wait() и не забывать освобождать его после.
Можешь привести пример?
Саша
sync.Cond встречается гораздо реже, чем мьютексы или каналы. sync.Cond удобно использовать, когда требуется многократное уведомление о событии неопределённого круга подпиcчиков (горутин).
Разберём пример. При изменении значения переменной нужно, чтобы её обработали несколько разных горутин (воркеров). Для оповещения всех ожидающих горутин будем использовать метод Broadcast(). Удобно, что функция, которая отправляет Broadcast-сигнал не знает, сколько запущено воркеров и не должна об этом думать:
GO
package main

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

func startWorkers(c *sync.Cond, val *int) {
    workerCount := 3

    for i := 0; i < workerCount; i++ {
        go func(workerId int) {
            c.L.Lock()
            for {
                c.Wait()
                // получили сигнал
                fmt.Printf("val %v processed by worker %v\n", *val, workerId)
            }
        }(i)
    }
}

func main() {
    var m sync.Mutex
    c := sync.NewCond(&m)
    val := 0
    startWorkers(c, &val)
    // ждём, чтобы стартанули все воркеры
    time.Sleep(100 * time.Millisecond)

    for i := 0; i < 4; i++ {
        m.Lock()
        val = i
        fmt.Printf("set val to %v\n", val)
        // отправляем сигнал всем воркерам
        c.Broadcast()
        m.Unlock()
        time.Sleep(time.Millisecond)
    }
}
Перечислим ещё раз рассмотренные примитивы синхронизации.
Если нужно, чтобы часть кода выполнялась только одной горутиной, то следует использовать примитив sync.Mutex и управлять его состоянием методами Lock() и Unlock().
Если происходят операции чтения и записи, то можно воспользоваться мьютексом типа sync.RWMutex, который позволяет горутинам одновременно читать данные и за счёт этого увеличивает скорость работы.
Примитив sync.Cond даёт горутинам возможность обмениваться сигналами, в том числе отправлять сигнал всем подписчикам. Эта возможность используется редко: в большинстве случаев для обмена сигналами удобнее использовать каналы. О них мы расскажем в следующем уроке.

Задание 4/4

В последнем задании вы сможете проверить знания по мьютексам и WaitGroup — ключевым элементам при работе с горутинами. Найдите в программе ошибки и исправьте их. Программа должна завершаться без ошибок, а мапа всегда должна содержать 100 элементов.
Подсказка: Вызов wg.Add(1) расположен внутри горутины, defer mu.Unlock() выполнится только при выходе из функции.
GO
package main

import (
    "fmt"
    "sync"
)

func main() {
    var (
        mu sync.Mutex
        wg sync.WaitGroup
    )
    m := make(map[int]int)
    for i := 0; i < 10; i++ {
        go func(v int) {
            wg.Add(1)
            for j := 0; j < 10; j++ {
                mu.Lock()
                defer mu.Unlock()

                k := 10*v + j
                m[k] = k
            }
            wg.Done()
        }(i)
    }
    wg.Wait()
    fmt.Println(len(m))
}
Это был непростой урок, но вы справились! Вы узнали, как создавать горутины и разграничивать доступ из горутин к общим ресурсам. Работа с горутинами не всегда проста, но она действительно позволяет значительно ускорить выполнение программы и улучшить её производительность. В следующем уроке расскажем о работе с каналами и атомарными операциями.

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