sync: Mutex, RWMutex, Once, Cond, WaitGroup, Pool

О чём глава

Сервис держит в памяти кэш: карту «ключ → значение». Её читают сотни горутин-обработчиков, и изредка какая-нибудь из них обновляет запись. Передавать карту из рук в руки через канал здесь неудобно: данные не движутся, они лежат на месте, и к ним по очереди обращаются многие. Для такого случая в пакете sync есть примитивы, которые защищают общее состояние на месте.

В главе разберём каждый из них: Mutex, RWMutex, Once, WaitGroup, Cond и Pool, а заодно sync.Map. Для каждого — что он гарантирует, когда уместен и где на нём обычно ошибаются.

Перед чтением

Нужны структуры, методы с получателем-указателем, карты, defer и горутины. Полезно помнить правило из модели памяти: если хотя бы одна горутина пишет в переменную, все обращения к ней из разных горутин должны быть синхронизированы. Для sync.Pool понадобится приведение типа x.(T) из главы «Интерфейсы».

Два термина. Общее состояние — данные, к которым обращаются несколько горутин. Критическая секция — участок кода, который работает с общим состоянием и поэтому должен выполняться под одной блокировкой.

На первом проходе изучите Mutex, WaitGroup и Once, затем RWMutex. Cond понадобится к задаче 08. Pool — оптимизация, к которой приходят после того, как код уже корректен и измерения показали, что аллокации мешают.

Мьютекс или канал

Перед выбором примитива полезно спросить себя: мне нужно передать данные из одной горутины в другую или дать многим горутинам по очереди поработать с одними и теми же данными?

• Канал — когда данные движутся: раздать задачи воркерам, собрать результаты, сообщить о событии.
• Мьютекс — когда данные лежат на месте: счётчик, кэш, карта, поля структуры, к которым обращаются многие.

Обратный выбор обычно получается неуклюжим. Защищать карту через выделенную горутину-владельца, которой по каналу присылают запросы «прочитай» и «запиши», можно, но это больше кода и медленнее, чем один sync.Mutex. А раздавать задачи воркерам через срез под мьютексом с флагами «взято» хуже, чем через канал.

sync.Mutex

Mutex (mutual exclusion, взаимное исключение) гарантирует, что в критической секции одновременно находится не больше одной горутины. Остальные, вызвавшие Lock, ждут, пока владелец не вызовет Unlock.

type Counter struct {
    mu sync.Mutex
    n  int
}
 
func (c *Counter) Inc() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.n++
}

Мьютекс защищает не только от одновременных записей. Unlock happens-before следующего Lock, поэтому всё, что одна горутина записала под мьютексом, следующая горутина, захватившая его, гарантированно увидит.

package main
 
import (
	"fmt"
	"sync"
)
 
type Counter struct {
	mu sync.Mutex
	n  int
}
 
func (c *Counter) Inc() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.n++
}
 
func main() {
	var wg sync.WaitGroup
	c := &Counter{}
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			c.Inc()
		}()
	}
	wg.Wait()
	fmt.Println("итог:", c.n)
}

итог: 1000

Без мьютекса c.n++ содержал бы гонку данных: это три шага (прочитать, прибавить, записать), и две горутины могут прочитать одно и то же старое значение. Мьютекс делает эти три шага неделимыми относительно других горутин. Как выглядит отчёт race detector для такой ошибки, показано в главе «Утечки горутин и гонки».

Три правила, которые экономят часы отладки:

• Мьютекс нельзя копировать после первого использования. Если передать структуру с мьютексом по значению, у копии будет свой независимый мьютекс, и защита перестанет работать без единой ошибки компиляции. Поэтому методы объявляют на указателе (*Counter). Команда go vet находит такие копирования (проверка copylocks).
• Критическая секция должна быть короткой. Сетевой вызов или запись на диск под Lock заставляет всех остальных ждать, пока идёт ввод-вывод. Сначала подготовьте данные, потом коротко захватите мьютекс, чтобы их сохранить.
• Unlock через defer сразу после Lock. Тогда мьютекс освободится при любом return и при панике.

sync.RWMutex

RWMutex различает читателей и писателей. Одновременно мьютекс могут держать либо сколько угодно читателей (RLock/RUnlock), либо один писатель (Lock/Unlock).

type Cache struct {
    mu sync.RWMutex
    m  map[string]string
}
 
func (c *Cache) Get(k string) (string, bool) {
    c.mu.RLock()
    defer c.mu.RUnlock()
    v, ok := c.m[k]
    return v, ok
}
 
func (c *Cache) Set(k, v string) {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.m[k] = v
}

Выигрыш появляется, когда чтений намного больше, чем записей, и чтение занимает заметное время: читатели не ждут друг друга. Но RWMutex сложнее устроен: он ведёт учёт читателей, и этот учёт сам становится точкой, за которую соревнуются ядра. Если критическая секция крошечная (прочитать одно поле), обычный Mutex или atomic часто оказываются быстрее. Решают это бенчмарком (go test -bench), а не на глаз.

Ещё одна деталь: если писатель ждёт Lock, новые RLock тоже ждут, пока писатель не отработает. Это защищает писателей от голодания, но означает, что рекурсивный RLock в той же горутине может привести к дедлоку.

sync.Map

Для кэша есть и готовая потокобезопасная карта sync.Map с методами Load, Store, LoadOrStore, Delete, Range. Она хорошо подходит для двух случаев, описанных в документации: ключ записывается один раз и потом много раз читается, либо разные горутины работают с непересекающимися наборами ключей. Для остальных случаев обычная map под Mutex или RWMutex проще и часто быстрее. У sync.Map нет типов: значения хранятся как any, и при чтении нужно приведение типа.

sync.Once

Once выполняет функцию ровно один раз, сколько бы горутин ни вызвали Do одновременно. Выполнение функции happens-before возврата из любого Do, поэтому все вызывающие видят полностью завершённую инициализацию.

var (
    once sync.Once
    conn *DB
)
 
func GetDB() *DB {
    once.Do(func() { conn = connect() }) // connect выполнится один раз
    return conn
}

Если сто горутин одновременно вызовут GetDB, connect выполнится один раз, а остальные 99 дождутся его окончания и получат тот же conn.

package main
 
import (
	"fmt"
	"sync"
)
 
func main() {
	var once sync.Once
	var wg sync.WaitGroup
	calls := 0
 
	for i := 0; i < 5; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			once.Do(func() {
				calls++
				fmt.Println("инициализация выполнена")
			})
		}()
	}
	wg.Wait()
	fmt.Println("тело Do отработало раз:", calls)
}

инициализация выполнена
тело Do отработало раз: 1

Подвох: если функция в Do запаниковала или вернула ошибку через замыкание, Once всё равно считает её выполненной. Повторной попытки не будет. С Go 1.21 есть удобные обёртки: sync.OnceValue(f) возвращает функцию, которая один раз вызывает f и потом отдаёт сохранённый результат, а sync.OnceValues(f) делает то же для пары «значение, ошибка»:

var loadConfig = sync.OnceValues(func() (*Config, error) {
    return readConfig("app.yaml")
})
 
cfg, err := loadConfig() // при любом числе вызовов readConfig выполнится один раз

Ошибка здесь тоже запоминается: все вызывающие получат одну и ту же ошибку. Если нужен повтор после неудачи, Once не подходит, понадобится мьютекс и флаг.

sync.WaitGroup

WaitGroup ждёт завершения группы горутин. Внутри у неё счётчик: Add(n) увеличивает его, Done() уменьшает на единицу, Wait() ждёт, пока счётчик не станет нулём.

var wg sync.WaitGroup
for _, job := range jobs {
    wg.Add(1)
    go func(j Job) {
        defer wg.Done()
        process(j)
    }(job)
}
wg.Wait() // здесь все горутины точно завершились

Два правила, которые нарушают чаще всего:

• Add вызывают до go, в запускающем коде. Если вызвать Add(1) внутри горутины, Wait может выполниться раньше, увидеть ноль и вернуться, пока горутины ещё не начались.
• Done вызывают через defer, чтобы счётчик уменьшился и при раннем return, и при панике. Пропущенный Done — это Wait, который не вернётся никогда.

С Go 1.25 у WaitGroup есть метод Go: он сам делает Add(1), запускает функцию в новой горутине и вызывает Done, когда она вернётся. Ошибиться с порядком Add при этом невозможно. Функция, переданная в wg.Go, не должна паниковать.

var wg sync.WaitGroup
for _, job := range jobs {
    wg.Go(func() { process(job) })
}
wg.Wait()

WaitGroup отвечает только за «дождаться всех». Если горутины пишут в общую переменную, её всё равно нужно защищать:

package main
 
import (
	"fmt"
	"sync"
)
 
func main() {
	var wg sync.WaitGroup
	var mu sync.Mutex
	total := 0
 
	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			mu.Lock()
			total += n * n
			mu.Unlock()
		}(i)
	}
 
	wg.Wait()
	fmt.Println("сумма квадратов 1..5:", total)
}

сумма квадратов 1..5: 55

sync.Cond

Cond (условная переменная) позволяет горутинам спать, пока не выполнится условие над общим состоянием, и просыпаться по сигналу. Её создают через sync.NewCond(&mu) и привязывают к мьютексу, который защищает это состояние.

type Queue struct {
    mu   sync.Mutex
    cond *sync.Cond // создаётся как sync.NewCond(&q.mu)
    data []int
}
 
func (q *Queue) Pop() int {
    q.mu.Lock()
    defer q.mu.Unlock()
    for len(q.data) == 0 { // цикл, а не if
        q.cond.Wait()
    }
    v := q.data[0]
    q.data = q.data[1:]
    return v
}
 
func (q *Queue) Push(v int) {
    q.mu.Lock()
    q.data = append(q.data, v)
    q.mu.Unlock()
    q.cond.Signal() // разбудить одного ждущего
}

Wait делает три вещи: отпускает мьютекс, усыпляет горутину и, проснувшись, снова захватывает мьютекс. Отпускать мьютекс нужно, иначе Push никогда бы не смог добавить данные.

Почему цикл, а не if? В Go Wait не возвращается сам по себе, без Signal или Broadcast. Но между сигналом и моментом, когда проснувшаяся горутина снова захватит мьютекс, другая горутина может успеть забрать данные. Цикл заново проверяет условие под мьютексом и не даёт работать с пустой очередью.

Signal будит одного ждущего, Broadcast — всех. Производитель и потребитель на Cond:

package main
 
import (
	"fmt"
	"sync"
)
 
func main() {
	var mu sync.Mutex
	cond := sync.NewCond(&mu)
	queue := []int{}
	done := false
 
	var wg sync.WaitGroup
	wg.Add(1)
	go func() { // потребитель
		defer wg.Done()
		for {
			mu.Lock()
			for len(queue) == 0 && !done {
				cond.Wait()
			}
			if len(queue) == 0 && done {
				mu.Unlock()
				return
			}
			v := queue[0]
			queue = queue[1:]
			mu.Unlock()
			fmt.Println("получено:", v)
		}
	}()
 
	for i := 1; i <= 3; i++ { // производитель
		mu.Lock()
		queue = append(queue, i)
		mu.Unlock()
		cond.Signal()
	}
	mu.Lock()
	done = true
	mu.Unlock()
	cond.Broadcast() // разбудить, чтобы потребитель увидел done и вышел
 
	wg.Wait()
	fmt.Println("готово")
}

получено: 1
получено: 2
получено: 3
готово

Порядок значений здесь гарантирован: очередь одна, потребитель один, и он забирает значения с начала среза. Потребитель выходит, только когда очередь пуста и выставлен done, поэтому ни одно значение не теряется.

У Cond есть серьёзное ограничение: ожидание Wait нельзя поставить веткой в select, значит, его нельзя совместить с таймаутом или отменой через контекст. Поэтому многие задачи, где напрашивается Cond, чище решаются каналами. Cond уместен, когда ждущих много, у них общее состояние под мьютексом и нужно будить то одного (Signal), то всех (Broadcast). Задача 08 (своя WaitGroup) как раз такая.

sync.Pool

Pool — набор временных объектов для повторного использования. Он снижает число выделений памяти и нагрузку на сборщик мусора там, где одни и те же короткоживущие объекты создаются тысячи раз в секунду, например буферы в HTTP-обработчиках.

var bufPool = sync.Pool{
    New: func() any { return new(bytes.Buffer) },
}
 
func handle() {
    b := bufPool.Get().(*bytes.Buffer)
    b.Reset()          // объект мог вернуться с данными прошлого использования
    defer bufPool.Put(b)
    // ... используем b как временный буфер
}

Get достаёт объект из пула или вызывает New, если пул пуст. Put возвращает объект в пул. Рантайм может выбросить объекты из пула в любой момент, обычно при сборке мусора, поэтому Pool — не пул соединений и не кэш: класть туда то, что обязано сохраниться (соединения, открытые файлы), нельзя.

Два правила: сбрасывайте состояние объекта после Get (как b.Reset()) и не добавляйте Pool без измерений. На редких выделениях он ничего не даст, а код усложнит. Запускаемого примера здесь нет: эффект Pool виден только в бенчмарках под нагрузкой.

Частые ошибки

• Копирование Mutex или WaitGroup по значению. Две независимые копии — нет защиты. Находится через go vet.
• Add внутри горутины. Wait может вернуться раньше времени.
• cond.Wait() под if. Условие нужно перепроверять в цикле.
• Долгий ввод-вывод под Lock. Все остальные ждут, пока идёт сетевой вызов.
• RWMutex на крошечной секции. Учёт читателей стоит дороже, чем экономия.
• Чтение без мьютекса «потому что только читаю». Если кто-то пишет, читатель тоже должен брать блокировку. Иначе это гонка данных.

Мысленная модель

Каждый примитив отвечает на свой вопрос:

• Mutex — «по одному, пожалуйста»;
• RWMutex — «читать могут многие, писать — один»;
• WaitGroup — «дождаться, пока все закончат»;
• Once — «сделать один раз и запомнить»;
• Cond — «спать, пока не изменится состояние»;
• Pool — «не выделять заново то, что можно переиспользовать».

Сначала сформулируйте, какая синхронизация нужна, и только потом выбирайте инструмент.

Попробуйте сами

Добавьте к Counter метод Value() int, который читает n под тем же мьютексом, что и Inc. Напишите тест: 10 горутин делают по 100 вызовов Inc, тест ждёт всех и проверяет Value. Запустите go test -race .. Какие обращения нужно защищать, если Value вызывают, пока горутины ещё работают?

Проверка

func (c *Counter) Value() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.n
}

Тест в файле counter_test.go того же пакета:

func TestCounter(t *testing.T) {
    c := &Counter{}
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := 0; j < 100; j++ {
                c.Inc()
            }
        }()
    }
    wg.Wait()
    if got := c.Value(); got != 1000 {
        t.Fatalf("Value() = %d; хотим 1000", got)
    }
}

Тесту нужны импорты testing и sync. Защищать нужно и запись в Inc, и чтение в Value, одним и тем же мьютексом. Wait в тесте определяет момент итоговой проверки, но не заменяет защиту отдельных обращений.

Что дальше

Для одиночного счётчика или флага мьютекс избыточен: следующая глава, atomic, показывает, как обойтись одной атомарной операцией и где эта экономия заканчивается.

В задачах курса этой главе соответствует блок 2 (задачи 05–08: счётчик, кэш с RWMutex, однократная инициализация, своя WaitGroup) и блок 5 (шардированная карта, singleflight). В блоке 7 мьютексы и WaitGroup работают вместе с каналами и context внутри пулов воркеров.