Deep Engineering

ЗАМЕР

bench/gopattern/cancel.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).

Скрипт

199 строк
//go:build ignore

// Отмена стадии конвейера: где именно стадия ждёт и какая из этих точек
// слушает ctx.Done().
//
// ЗАЧЕМ ЭТА ПРОГРАММА. В internals.go (раздел 11) стадия sqCtx читала вход
// через `for n := range in` и слушала ctx.Done() только на отправке. В той
// связке это сработало: источник countFromCtx сам закрыл свой выход по
// отмене, range в sq вышел. Но у стадии ДВЕ точки ожидания — получение из
// входа и отправка в выход, — и range отмену не слушает. Здесь обе версии
// стадии проверяются на четырёх сценариях, и завершение определяется не
// счётом горутин, а сигналом: закрылся ли выход стадии за отведённое время.
//
// СЦЕНАРИИ:
//
//  1. Обычный поток: вход 1, 2, 3 и закрыт.
//  2. Вход закрыт и пуст.
//  3. Отмена, пока вход открыт и пуст (никто не отправляет и не закрывает);
//     и то же для nil-входа.
//  4. Отмена, пока выход заблокирован (никто не читает); вход бесконечный,
//     так что выход может закрыться только по отмене.
//
// ЗАПУСК:
//
//	go run bench/gopattern/cancel.go
package main

import (
	"context"
	"fmt"
	"runtime"
	"strings"
	"time"
)

// sqRange — стадия из раздела 11 internals.go: вход через range, отмена
// только на отправке.
func sqRange(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			select {
			case out <- n * n:
			case <-ctx.Done():
				return
			}
		}
	}()
	return out
}

// sqCtx — обе точки ожидания в select с ctx.Done().
func sqCtx(ctx context.Context, in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for {
			var n int
			var ok bool
			select {
			case <-ctx.Done():
				return
			case n, ok = <-in:
				if !ok {
					return
				}
			}
			select {
			case <-ctx.Done():
				return
			case out <- n * n:
			}
		}
	}()
	return out
}

// closedWithin дочитывает out до закрытия, но не дольше d. Это и есть
// проверка завершения: стадия закрывает выход в defer, значит закрытый
// выход — сигнал, что её горутина вернулась. Таймаут здесь — не способ
// «дождаться», а граница, после которой проверка честно говорит «нет».
func closedWithin(out <-chan int, d time.Duration) ([]int, bool) {
	var got []int
	timeout := time.After(d)
	for {
		select {
		case v, ok := <-out:
			if !ok {
				return got, true
			}
			got = append(got, v)
		case <-timeout:
			return got, false
		}
	}
}

// countFrom — бесконечный вход без отмены: закрывается никогда. В сценарии 4
// это важно: выход стадии может закрыться только по ctx.Done(), а не потому,
// что кончился вход.
func countFrom() <-chan int {
	out := make(chan int)
	go func() {
		for n := 1; ; n++ {
			out <- n
		}
	}()
	return out
}

func gen(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}

type stage func(context.Context, <-chan int) <-chan int

func yesno(b bool) string {
	if b {
		return "да"
	}
	return "НЕТ"
}

func main() {
	fmt.Printf("%s %s/%s | отмена стадии конвейера: две точки ожидания\n",
		runtime.Version(), runtime.GOOS, runtime.GOARCH)
	const wait = 500 * time.Millisecond
	stages := []struct {
		name string
		f    stage
	}{{"СТАДИЯ sqRange: RANGE ПО ВХОДУ, SELECT ТОЛЬКО НА ОТПРАВКЕ", sqRange}, {"СТАДИЯ sqCtx: SELECT И НА ПОЛУЧЕНИИ, И НА ОТПРАВКЕ", sqCtx}}

	for i, st := range stages {
		title := fmt.Sprintf("%d) %s", i+1, st.name)
		fmt.Println()
		fmt.Println(title)
		fmt.Println(strings.Repeat("─", len([]rune(title))))

		// 1. Обычный поток.
		{
			ctx, cancel := context.WithCancel(context.Background())
			got, ok := closedWithin(st.f(ctx, gen(1, 2, 3)), wait)
			cancel()
			fmt.Printf("  обычный поток 1, 2, 3:            получено %v, выход закрыт: %s\n", got, yesno(ok))
		}
		// 2. Вход закрыт и пуст.
		{
			ctx, cancel := context.WithCancel(context.Background())
			in := make(chan int)
			close(in)
			_, ok := closedWithin(st.f(ctx, in), wait)
			cancel()
			fmt.Printf("  вход закрыт и пуст:               выход закрыт: %s\n", yesno(ok))
		}
		// 3. Отмена при открытом пустом входе.
		{
			ctx, cancel := context.WithCancel(context.Background())
			in := make(chan int) // никто не отправляет и не закрывает
			out := st.f(ctx, in)
			cancel()
			_, ok := closedWithin(out, wait)
			fmt.Printf("  отмена, вход открыт и пуст:       выход закрыт за 500 мс: %s\n", yesno(ok))
		}
		{
			ctx, cancel := context.WithCancel(context.Background())
			var in chan int // nil: получение из него ждёт вечно
			out := st.f(ctx, in)
			cancel()
			_, ok := closedWithin(out, wait)
			fmt.Printf("  отмена, вход nil:                 выход закрыт за 500 мс: %s\n", yesno(ok))
		}
		// 4. Отмена при заблокированном выходе.
		{
			ctx, cancel := context.WithCancel(context.Background())
			out := st.f(ctx, countFrom())
			// Даём стадии получить 1 и встать на отправке 1: читать выход
			// пока некому. Это не гарантия завершения — только постановка
			// сценария; проверка ниже от неё не зависит: вход бесконечный,
			// так что закрыться выход может только по отмене.
			time.Sleep(50 * time.Millisecond)
			cancel()
			_, ok := closedWithin(out, wait)
			fmt.Printf("  отмена, выход никто не читает:    выход закрыт за 500 мс: %s\n", yesno(ok))
		}
	}

	fmt.Println()
	fmt.Println("  cancel() — сигнал, а не остановка: стадия выходит, только если в")
	fmt.Println("  момент ожидания слушает ctx.Done(). range по входу его не слушает.")
}