Планировщик

Планировщик выполняет процессы по принципу work-stealing. Воркеры поддерживают локальные деки и заимствуют работу друг у друга при простое.

Интерфейс Process

Планировщик работает с любым типом, реализующим интерфейс Process:

type Process interface {
    Init(ctx context.Context, method string, input payload.Payloads) error
    Step(events []Event, out *StepOutput) error
    Close()
}
Метод Назначение
Init Подготовка процесса с именем метода и входными аргументами
Step Продвижение автомата состояний входящими событиями, запись yields в output
Close Освобождение ресурсов

Параметр method в Init указывает, какую точку входа вызывать. Экземпляр процесса может предоставлять несколько точек входа, и вызывающий выбирает нужную. Это также служит проверкой корректности инициации процесса планировщиком.

Планировщик вызывает Step() многократно, передавая события (завершения yield'ов, сообщения) и собирая yield'ы (команды для диспатчинга). Процесс записывает свой статус и yield'ы в буфер StepOutput.

type Event struct {
    Type  EventType  // EventYieldComplete или EventMessage
    Tag   uint64     // Корреляционный тег для завершений yield
    Data  any        // Данные результата или payload сообщения
    Error error      // Ошибка, если yield завершился неудачей
}

Структура

Планировщик порождает GOMAXPROCS воркеров по умолчанию. У каждого воркера локальный дек для кеш-дружественного LIFO-доступа и собственная MPSC inject-очередь для асинхронных завершений, имеющих привязку к этому воркеру. Глобальная FIFO-очередь обрабатывает новые submissions и повторные постановки без привязки. Процессы отслеживаются по PID для маршрутизации сообщений.

Поиск работы

flowchart TD
    W[Воркер ищет работу] --> L{Локальный дек?}
    L -->|есть элементы| LP[Pop со дна LIFO]
    L -->|пуст| I{Inject-очередь?}
    I -->|есть элементы| IP[Pop + слить до 16 в локальный]
    I -->|пуста| G{Глобальная очередь?}
    G -->|есть элементы| GP[Pop + пакетный перенос до 16]
    G -->|пуста| S[Заимствовать у случайной жертвы]
    S --> SH[StealHalfInto дека жертвы]

Воркеры проверяют источники в порядке приоритета:

Приоритет Источник Паттерн
1 Локальный дек LIFO pop, lock-free, кеш-дружественный
2 Inject-очередь MPSC pop привязанных асинхронных завершений, слив до 16 в локальный
3 Глобальная очередь FIFO pop с пакетным переносом
4 Другие воркеры Заимствовать половину из дека жертвы

При pop из inject- или глобальной очереди воркеры берут один элемент и переносят до 16 дополнительных в свой локальный дек.

Дек Chase-Lev

У каждого воркера дек Chase-Lev для work-stealing:

type Deque struct {
    buffer atomic.Pointer[dequeBuffer]
    top    atomic.Int64  // Воры заимствуют отсюда (CAS)
    bottom atomic.Int64  // Владелец push/pop отсюда
}

Владелец push'ит и pop'ит со дна (LIFO) без синхронизации. Воры заимствуют с верха (FIFO) через CAS. Это даёт владельцу кеш-дружественный доступ к недавно добавленным элементам, распределяя более старую работу ворам.

StealHalfInto забирает половину элементов одной CAS-операцией, снижая конкуренцию.

Адаптивное вращение

Перед блокировкой на condition variable воркеры адаптивно вращаются:

Счётчик вращений Действие
< 4 Тесный цикл
4-15 Уступить поток (runtime.Gosched)
>= 16 Заблокироваться на condition variable

Состояния процесса

stateDiagram-v2
    [*] --> Ready: Submit
    Ready --> Running: CAS воркером
    Running --> Complete: done
    Running --> Blocked: yields commands
    Running --> Idle: ожидание сообщений
    Blocked --> Ready: CompleteYield
    Idle --> Ready: Send прибыл
Состояние Описание
Ready В очереди на выполнение
Running Воркер выполняет Step()
Blocked Ожидание завершения yield
Idle Ожидание сообщений
Complete Выполнение завершено

Флаг wakeup обрабатывает гонки: если обработчик вызывает CompleteYield, пока воркер ещё владеет процессом (Running), он устанавливает флаг. Воркер проверяет флаг после диспатчинга и перепланирует, если установлен.

Очередь событий

У каждого процесса MPSC (multi-producer, single-consumer) очередь событий:

  • Producers: обработчики команд (CompleteYield), отправители сообщений (Send)
  • Consumer: воркер сливает события в Step()

Очередь охраняется счётчиком поколений. Каждый producer привязывается к поколению, которое он наблюдал; Reset его увеличивает, поэтому отправитель, оставшийся от предыдущего выполнения, не может push'нуть в переиспользованную очередь.

Обычный поток событий не ограничен. Учёт включается для каждого сообщения отдельно: сообщение, несущее MaxItems или MaxBytes, допускается в рамках бюджета топика, и побеждает самый жёсткий из виденных для топика лимитов. Сообщение удерживает свою резервацию, пока потребляющий процесс её не освободит, а терминалы никогда не расходуют ёмкость бэклога.

Когда бюджет топика исчерпан, очередь добавляет одно синтетическое сообщение на место переполнившего, несущее message queue limit exceeded и следом терминальный payload. Дальнейший трафик по этому топику отбрасывается до сброса очереди, поэтому ограниченная подписка заканчивается терминалом с ошибкой, а не растёт неограниченно.

Маршрутизация сообщений

Планировщик реализует relay.Receiver для маршрутизации сообщений процессам. Send делегирует SendContext с фоновым контекстом; SendContext проверяет отмену перед поиском цели и перед допуском, потому что сам допуск неблокирующий и необратим после успеха.

Оба находят целевой PID в карте byPID и push'ат пакет в очередь процесса под текущим поколением процессора. Допуск трёхвариантный:

Результат Значение Владение пакетом
Accepted Очередь приняла пакет Очередь; освобождается планировщиком после обработки
Dropped Бюджет топика переполнен, и очередь не сохранила ничего, кроме собственного терминала переполнения Вызывающий; освобождается немедленно
Rejected Очередь закрыта либо поколение устарело Вызывающий; SendContext возвращает ErrProcessClosed

Принятый или отброшенный push затем будит процесс, если тот Idle или заблокирован. Повторная постановка идёт через injectOrGlobal, который push'ит в per-worker inject-очередь последнего воркера, когда у процесса известная привязка к воркеру, и откатывается к глобальной очереди иначе.

Завершение работы

При завершении планировщик отправляет события отмены всем выполняющимся процессам и ждёт их завершения или таймаута. Воркеры завершаются, когда работы больше нет.

См. также