ЗАМЕР

bench/gopattern/semaphore.go

Скрипт, которым получены числа в статье, и запись прогона. Файл читается на сборке из репозитория — это тот самый код, который запускали, а не его копия.

Цитируется в статье

/ru/interview/golang/concurrency-patterns

Как запустить

go run bench/gopattern/internals.go
/usr/local/go1.25.1/bin/go run bench/gopattern/internals.go
go run bench/gopattern/practice.go
go run bench/gopattern/cancel.go
go run bench/gopattern/semaphore.go
(cd bench/gopattern/cases && go run ./pipeline)

Запись прогона

Замеры для урока «Паттерны конкурентности»

Файл	Что делает
`internals.go`	тринадцать наблюдений: пул воркеров на задаче для процессора (1, 2, 4, 8 воркеров при `GOMAXPROCS=2` и 1, 8 при `GOMAXPROCS=1`, со столбцом «ядер занято») и на задаче со сном (1–16 воркеров); порядок результатов и индекс; конвейер `gen → sq`; забытый `close` без таймеров и с таймером; `wg.Wait()` до чтения результатов, в том числе с буфером; fan-in с `close` после `wg.Wait()` и без него; семафор на буферизованном канале; потребитель, ушедший раньше, без `context` и с ним; остановка по первой ошибке (свой минимальный аналог `errgroup.WithContext` на `WaitGroup`, `Once` и `context.WithCancelCause`); `sync.WaitGroup.Go`
`cases/forgotclose/main.go`	конвейер, где `sq` забыла `close`; намеренно падает (раздел 5)
`cases/waitbeforeread/main.go`	пул, где `wg.Wait()` стоит до чтения результатов; намеренно падает (раздел 6)
`cases/waitbuffered/main.go`	та же ошибка с буфером результатов на 6; число задач — аргумент; на 7 падает (раздел 6)
`cases/earlyclose/main.go`	слияние, где `close(out)` стоит без `wg.Wait()`; намеренно падает (раздел 8)
`cases/wggo/main.go`	`sync.WaitGroup.Go`; на go1.24.7 не собирается (раздел 13)
`cases/go.mod`	свой модуль для подопытных программ, `go 1.24`
`cancel.go`	отмена стадии конвейера на четырёх сценариях (обычный поток, закрытый вход, отмена при открытом пустом и nil-входе, отмена при заблокированном выходе) для двух версий стадии: `range` по входу с `select` только на отправке и `select` в обеих точках ожидания; завершение определяется закрытием выхода за 500 мс, а не счётом горутин
`semaphore.go`	семафор на канале: место внутри уже запущенной горутины, место до `go`, ожидание места с `ctx.Done()`; горутины считаются своим счётчиком живых обработчиков
`cases/pipeline/main.go`	полный короткий пример урока: конвейер из двух стадий с отменой; файл без кириллицы — его показывают обе локали
`practice.go`	источник ответов двух практических задач урока: утечка двух горутин и семафор ёмкостью 2; кратность 8 воркеров к одному на задаче для процессора

go run bench/gopattern/internals.go
/usr/local/go1.25.1/bin/go run bench/gopattern/internals.go
go run bench/gopattern/practice.go
go run bench/gopattern/cancel.go
go run bench/gopattern/semaphore.go
(cd bench/gopattern/cases && go run ./pipeline)

internals.go собирает и запускает cases/* тем же go, которым запущен сам (runtime.GOROOT()/bin/go), с GOTOOLCHAIN=local. Путь к cases в выводе заменён на <cases>, номера горутин — на <N>, смещения +0x… в стеке убраны.

Записи прогонов — в runs/, go1.24.7 linux/amd64, 2 ядра. runs/internals-go1.25.txt — тот же internals.go под go1.25.1: отличается раздел 13 (WaitGroup.Go есть, go vet предупреждает о версии в go.mod) и миллисекунды в разделах 1 и 2. cancel.go, semaphore.go и cases/pipeline на go1.25.1 дают тот же вывод, кроме строки версии (проверено diff), поэтому отдельных записей для них нет; cancel.go и semaphore.go дали одинаковый вывод в трёх запусках подряд.

Итоговая фраза раздела 9 internals.go — «Семафор ограничивает работу, а не число горутин: все 12 созданы» — верна для показанной там расстановки (место занимается внутри уже запущенной горутины). При месте, занятом до go, живых обработчиков не больше ёмкости — semaphore.go, раздел 2.

Функция sqCtx в internals.go читает вход через range и слушает отмену только на отправке; в связке раздела 11 этого хватает, а на открытом пустом входе — нет (cancel.go, раздел 1). Оставлена как есть, чтобы runs/internals.txt соответствовал коду.

Машину делят другие процессы. Времена — лучший из N кругов, конфигурации в разделе 1 и в practice.go прогоняются вперемешку. Столбец «ядер занято» — процессорное время процесса (getrusage), делённое на настенное: если соседи займут ядро, это видно в нём. Кратности — наблюдение этой конфигурации (go1.24.7/go1.25.1, linux/amd64, 2 ядра), а не константа: на другой машине, задаче и версии их нужно мерить заново.

Что получилось

На задаче для процессора больше GOMAXPROCS воркеров не ускоряет. При GOMAXPROCS=2 два воркера — 2,0 к одному, четыре и восемь — 1,9. Что потолок ставит именно GOMAXPROCS, видно по GOMAXPROCS=1: восемь воркеров — 1,0 при «ядер занято» 1,0, хотя свободных ядер два (выше 2,0 этот столбец на двух ядрах не поднимется при любом GOMAXPROCS).

На задаче со сном ускорение растёт с числом воркеров. 16 задач по 20 мс: 324 мс у одного воркера, 41 мс у восьми (8,0), 20 мс у шестнадцати (16,0).

Результаты приходят в порядке готовности. Задачи a b c d по 80, 20, 60, 40 мс пришли как [b d c a] во всех 20 повторах этого прогона; по индексу — [A B C D]. Порядок прихода не гарантирован: на загруженной машине он может поменяться, поэтому строка «совпал … из 20» от запуска к запуску может отличаться.

Забытый close роняет программу, только если спят все. cases/forgotclose: fatal error: all goroutines are asleep - deadlock!, код выхода 2. Та же стадия в программе с живым time.After: через 500 мс потребитель (горутина, запущенная из main.main) всё ещё в range, падения нет, +1 горутина. Причина — ранние выходы checkdead в runtime/proc.go (строки 6055–6058 и 6119–6124 на go1.24.7).

wg.Wait() до чтения результатов — взаимная блокировка. main в [sync.WaitGroup.Wait], четыре горутины в [chan send]. С буфером на 6 программа проходит на 6 задачах и падает на 7.

Ранний close в слиянии — паника. cases/earlyclose: получено 0 значений, затем panic: send on closed channel в строке out <- v.

Семафор ограничивает работу, а не горутины. Ёмкость 3: одновременно работали 3, горутин было 12, 81 мс против 20 без семафора.

Потребитель, ушедший без context, оставляет горутины. +2, обе в [chan send]; с select на ctx.Done() и defer cancel() — +0. Минимальная группа с первой ошибкой прерывает долгую работу на 10 с и возвращает первую ошибку, context.Cause отдаёт её же.

sync.WaitGroup.Go — с go1.25. На go1.24.7: wg.Go undefined. На go1.25.1 программа печатает сумму, а go vet при go 1.24 в go.mod пишет sync.Go requires go1.25 or later (module is go1.24).

Отмену стадия слышит только там, где её слушает. Стадия с range по входу и select только на отправке на открытом пустом входе и на nil-входе после cancel() выход за 500 мс не закрыла; стадия с select в обеих точках ожидания закрыла его во всех четырёх сценариях (cancel.go).

Семафор ограничивает и горутины, если место занимать до go. Место внутри горутины: обработчиков жило одновременно 12, работало 3. Место до go: жило 3, работало 3. С ctx.Done() в ожидании места раздача остановилась после трёх запусков с context canceled (semaphore.go).

Скрипт

159 строк

//go:build ignore

// Семафор на буферизованном канале: где занимать место — внутри горутины
// или до go — и что от этого зависит.
//
// ЗАЧЕМ ЭТА ПРОГРАММА. В internals.go (раздел 9) место в семафоре занималось
// ВНУТРИ уже запущенной горутины: работали одновременно трое, а горутин было
// двенадцать. Это свойство расстановки, а не семафора вообще. Если занять
// место ДО `go`, цикл раздачи сам встаёт на отправке в полный канал, и новых
// горутин не появляется, пока не освободится место. Третий вариант — то же
// с отменой: ожидание места тоже должно слушать ctx.Done().
//
// Горутины считаются своим счётчиком (живые обработчики), а не
// runtime.NumGoroutine(): в программе есть и служебные горутины, и
// отличить их по одному числу нельзя.
//
// ЗАПУСК:
//
//	go run bench/gopattern/semaphore.go
package main

import (
	"context"
	"errors"
	"fmt"
	"runtime"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

func show(title string) {
	fmt.Println()
	fmt.Println(title)
	fmt.Println(strings.Repeat("─", len([]rune(title))))
}

// counters — сколько обработчиков живо и сколько работает, и максимумы обоих.
type counters struct {
	alive, busy, maxAlive, maxBusy atomic.Int64
}

func raise(max *atomic.Int64, n int64) {
	for {
		m := max.Load()
		if n <= m || max.CompareAndSwap(m, n) {
			return
		}
	}
}

func (c *counters) enter() { raise(&c.maxAlive, c.alive.Add(1)) }
func (c *counters) leave() { c.alive.Add(-1) }
func (c *counters) work() {
	raise(&c.maxBusy, c.busy.Add(1))
	time.Sleep(20 * time.Millisecond)
	c.busy.Add(-1)
}

const tasks, limit = 12, 3

// acquireInside — горутина на задачу, место занимается внутри.
func acquireInside() *counters {
	var c counters
	sem := make(chan struct{}, limit)
	var wg sync.WaitGroup
	for i := 0; i < tasks; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			c.enter()
			defer c.leave()
			sem <- struct{}{}
			defer func() { <-sem }()
			c.work()
		}()
	}
	wg.Wait()
	return &c
}

// acquireBeforeGo — место занимается до go, освобождается в defer горутины.
func acquireBeforeGo() *counters {
	var c counters
	sem := make(chan struct{}, limit)
	var wg sync.WaitGroup
	for i := 0; i < tasks; i++ {
		sem <- struct{}{}
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			c.enter()
			defer c.leave()
			c.work()
		}()
	}
	wg.Wait()
	return &c
}

// acquireWithCancel — то же, что acquireBeforeGo, но ожидание места слушает
// ctx.Done(). Обработчики держат места, пока не закрыт release; когда трое
// запустились, контекст отменяется — четвёртое ожидание места должно
// закончиться отменой, а не вечным стоянием.
func acquireWithCancel() (started int, err error) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	sem := make(chan struct{}, limit)
	release := make(chan struct{})
	var running atomic.Int64
	var wg sync.WaitGroup
	for i := 0; i < tasks; i++ {
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			err = ctx.Err()
		}
		if err != nil {
			break
		}
		started++
		wg.Add(1)
		go func() {
			defer wg.Done()
			defer func() { <-sem }()
			if running.Add(1) == limit {
				cancel() // все места заняты — отменяем раздачу
			}
			<-release
		}()
	}
	close(release)
	wg.Wait()
	return started, err
}

func main() {
	fmt.Printf("%s %s/%s | семафор: место занимать внутри горутины или до go\n",
		runtime.Version(), runtime.GOOS, runtime.GOARCH)

	show("1) ГОРУТИНА НА ЗАДАЧУ, МЕСТО ЗАНИМАЕТСЯ ВНУТРИ")
	c := acquireInside()
	fmt.Printf("  задач %d, ёмкость %d: обработчиков жило одновременно %d, работало %d\n",
		tasks, limit, c.maxAlive.Load(), c.maxBusy.Load())

	show("2) МЕСТО ЗАНИМАЕТСЯ ДО GO")
	c = acquireBeforeGo()
	fmt.Printf("  задач %d, ёмкость %d: обработчиков жило одновременно %d, работало %d\n",
		tasks, limit, c.maxAlive.Load(), c.maxBusy.Load())

	show("3) МЕСТО ДО GO, ОЖИДАНИЕ МЕСТА СЛУШАЕТ CTX.DONE()")
	started, err := acquireWithCancel()
	fmt.Printf("  запущено обработчиков: %d из %d\n", started, tasks)
	fmt.Printf("  раздача остановилась с ошибкой: %v (errors.Is context.Canceled: %v)\n",
		err, errors.Is(err, context.Canceled))
}