Демоны

Демон — это несколько воркеров одной логики под присмотром. Упавшего перезапустят с нарастающей задержкой, зависшего убьют и заменят, а число рабочих рук можно менять на ходу — вручную или по нагрузке.

Что такое демон и зачем

Демон — компонент из двух частей: мастера-супервизора и флота одинаковых воркеров, которыми он управляет.

Проблема. Процесс решает задачу «работать непрерывно», но остаётся один. Как только работы становится больше, чем успевает один воркер, всплывают вопросы, которых у одиночки не было:

• запустить пять копий — как? Пять классов-близнецов или пять юнитов systemd?
• один упал — кто заметит и поднимет? И что, если он падает в цикле — как не положить систему бесконечными перезапусками?
• один завис на сетевом вызове навсегда — он ведь формально жив, ps его видит;
• ночью работы вдвое меньше — как убрать лишних, не оборвав тех, кто занят?

Каждый ответ по отдельности несложен, но вместе это уже супервизор, который вам не хочется писать.

Решение. Winter даёт его готовым. Вы описываете тело одного воркера и говорите, сколько их нужно; мастер следит за флотом: форкает, считает падения, выдерживает паузы перед перезапуском, убивает зависших, добавляет и убирает воркеров по вашему сигналу.

Быстрый старт

Наследуйте Daemon, опишите работу в workerRun() и укажите число реплик:

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Stereotype\Daemon;

class Emails extends Daemon
{
  #[Autowired] private OutboxRepository $outbox;
  #[Autowired] private Mailer $mailer;

  protected int $replicas = 4;        // четыре воркера

  protected function workerRun(): void
  {
      while ($this->isRunning()) {
          $letter = $this->outbox->takeNext();

          if ($letter === null) {
              $this->sleep(1);
              continue;
          }

          $this->markBusy();
          try {
              $this->mailer->send($letter);
              $this->outbox->markSent($letter->id);
          } finally {
              $this->markIdle();
          }
      }
  }
}

Объявляем в манифесте:

#[EnableDaemon(\Main\Process\Emails::class)]
final class Application extends WinterApplication { /* ... */ }

Или запускаем отдельно:

php call daemon main.process.Emails start -d
php call daemon main.process.Emails status

Daemon ● RUNNING
PID        51204
State      RUNNING
Activity   busy
Uptime     6h 12m
Workers    4
Restarts   1

SLOT  PID      STATE       ACT    UPTIME    RESTARTS
#0    51210    running     busy   6h 12m    0
#1    51211    running     idle   6h 12m    0
#2    51212    running     busy   6h 12m    0
#3    52890    running     idle   41m       1

Воркер отдельным классом

Если тело воркера уже описано самостоятельным процессом, workerRun() можно не писать — укажите класс:

class Emails extends Daemon
{
  protected int $replicas = 4;
  protected ?string $workerClass = \Main\Process\SendWorker::class;
}

Так удобно, когда один и тот же воркер должен уметь запускаться и в одиночку, и флотом. Тело обязано быть задано одним из двух способов: без workerRun() и без $workerClass демон уйдёт в бесконечный цикл падений.

Настройки

Минимальное число реплик — единица: $replicas = 0 будет поднято до 1. Свести флот к нулю можно только динамически, через desiredReplicas().

Перезапуск упавших

Что делать с умершим воркером, решает политика перезапуска.

use Flytachi\Winter\Kernel\Process\Daemon\{RestartPolicy, RestartMode};

protected function restart(): RestartPolicy
{
  return new RestartPolicy(
      mode:        RestartMode::ON_FAILURE,
      maxRestarts: 0,
      backoff:     1.0,
  );
}

Три режима, названные по соглашению Kubernetes и systemd — если вы описывали там рестарт-политику, семантика та же:

Нарастающая пауза

Перезапуск не мгновенный: пауза удваивается с каждым падением этого воркера и упирается в потолок 30 секунд. При базовой паузе в секунду это даёт

1 → 2 → 4 → 8 → 16 → 30 → 30 → …

Смысл в том, чтобы воркер, падающий из-за недоступной базы, не превратился в бесконечный цикл форков и не мешал системе восстановиться.

Сторож зависаний

Упавшего воркера видно, зависшего — нет: процесс жив, ps его показывает, работа стоит. Для этого есть отдельный механизм:

protected float $livenessTimeout = 120.0;

Каждый воркер регулярно подаёт признак жизни. Если признака нет дольше указанного времени, мастер убивает воркера принудительно, и дальше он идёт обычным путём падения — с паузой и перезапуском.

Масштабирование

Число воркеров может меняться на ходу. Переопределите desiredReplicas() — он опрашивается регулярно, и его ответ становится целью:

protected function desiredReplicas(): int
{
  $depth = $this->outbox->pendingCount();

  return match (true) {
      $depth > 10_000 => 20,
      $depth > 1_000  => 10,
      $depth > 0      => 4,
      default         => 1,
  };
}

Возвращать можно и ноль — тогда флот свернётся полностью и развернётся, когда работа появится.

Почему флот меняется не сразу

Ответ desiredReplicas() — это сигнал, а не команда. Супервизор сглаживает его, чтобы случайный всплеск не дёргал флот туда-сюда:

use Flytachi\Winter\Kernel\Process\Daemon\ScalingPolicy;

protected function scaling(): ScalingPolicy
{
  return new ScalingPolicy(
      scaleInterval:          1.0,
      scaleUpDelay:           0.0,
      scaleDownStabilization: 60.0,
      cooldown:               3.0,
      scaleStep:              0,
  );
}

Модель намеренно несимметрична: вверх быстро, вниз осторожно. Всплеск нагрузки нужно обслужить немедленно, а вот убирать воркеров из-за минутного затишья — плохая идея: через минуту их придётся поднимать заново.

Сокращение уважает занятость: при прочих равных первыми уходят простаивающие воркеры, а занятому дают доработать в пределах $grace.

Общая работа на несколько воркеров

Главное отличие демона от процесса в повседневной работе — воркеров несколько, и они лезут за работой одновременно. Фреймворк их запускает и стережёт, но не координирует: как поделить работу, решаете вы.

Задачу нужно захватывать атомарно: пометить своей и получить обратно в одном действии, чтобы между «выбрал» и «занял» не вклинился сосед.

Самый переносимый приём — пометить строку условным обновлением и убедиться, что обновилась именно она:

-- 1. пытаемся занять: помечаем ровно одну свободную строку своим маркером
UPDATE outbox
 SET claimed_by = :worker, claimed_at = NOW()
WHERE id = (SELECT id FROM outbox WHERE claimed_by IS NULL ORDER BY id LIMIT 1);

-- 2. забираем то, что заняли именно мы
SELECT * FROM outbox WHERE claimed_by = :worker AND claimed_at IS NOT NULL LIMIT 1;

В PostgreSQL и MySQL 8 то же самое короче — выборкой с пропуском занятых строк:

SELECT * FROM outbox
WHERE claimed_by IS NULL
ORDER BY id LIMIT 1
 FOR UPDATE SKIP LOCKED;

Маркером удобно брать номер слота или PID воркера: тогда по зависшим задачам видно, кто их занял.

Воркер не знает своего номера

Напрашивается разделить работу заранее — «воркер №0 берёт чётные записи, №1 нечётные». Так не получится: номер слота знает только супервизор, воркеру он не передаётся, и получить его изнутри workerRun() нельзя.

Это сделано намеренно. Номера живут ровно до первого падения: упавшего воркера заменят, флот может вырасти или сжаться, а разрезание, привязанное к числу реплик, после каждого такого события станет неверным — часть работы достанется двоим, часть никому.

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

-- воркер занимает не задачу, а целый раздел работы
UPDATE shards
 SET owner = :pid, taken_at = NOW()
WHERE id = (SELECT id FROM shards
            WHERE owner IS NULL OR taken_at < NOW() - INTERVAL '5 minutes'
            ORDER BY id LIMIT 1);

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

Хуки жизненного цикла

Все четыре необязательны и вызываются в мастере.

protected function onWorkerExit(int $slot, int $pid, bool $crashed): void
{
  if ($crashed) {
      $this->metrics->increment('emails.worker_crashed');
  }
}

protected function tick(): void
{
  $this->metrics->gauge('emails.queue_depth', $this->outbox->pendingCount());
}

Исключение внутри хука перехватывается и логируется — сломанный хук не уронит супервизор.

Состояния воркеров

В выводе status у каждого слота своё состояние. Оно помогает понять, что происходит с флотом прямо сейчас.

Номер слота стабилен: перезапуск возвращает воркера в тот же слот, поэтому #2 в логах всегда означает одного и того же участника флота.

Управление из кода

Демон наследует у процесса ту же статическую поверхность управления, поэтому запускать, останавливать и опрашивать флот можно прямо из приложения.

use Main\Process\Emails;

$pid = Emails::dispatch();     // супервизор ушёл в фон
Emails::stop();                // остановить весь флот

Как и у процесса, start() блокирует вызывающего — из веба используйте dispatch().

Состояние флота

status() возвращает не просто состояние мастера, а снимок всего флота: к обычным полям процесса добавляются два.

У каждого воркера доступно:

Это позволяет строить собственные проверки и панели, не разбирая вывод консоли:

use Flytachi\Winter\Kernel\Process\Activity;

$status = Emails::status();

if ($status === null) {
  Emails::dispatch();                         // флот не поднят — поднимем
  return;
}

$busy = array_filter(
  $status->workers,
  fn($w) => $w->activity === Activity::BUSY,
);

// все воркеры заняты дольше минуты — повод присмотреться
if (count($busy) === count($status->workers)) {
  $this->alerts->fire('Emails: весь флот занят');
}

Управление из консоли

php call daemon list                        # все демоны и размер их флотов
php call daemon main.process.Emails         # супервизор в текущем терминале
php call daemon main.process.Emails start -d    # в фоне
php call daemon main.process.Emails stop        # мягкая остановка всего флота
php call daemon main.process.Emails status      # состояние + таблица воркеров
php call daemon main.process.Emails status -v   # плюс расход ресурсов мастера

Псевдоним команды — call dmn.

Примеры

Флот фиксированного размера

Самый частый случай: работы стабильно много, автомасштабирование не нужно.

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Stereotype\Daemon;

class Webhooks extends Daemon
{
  #[Autowired] private WebhookRepository $hooks;
  #[Autowired] private HttpClient $http;

  protected int $replicas = 8;
  protected float $grace = 20.0;
  protected float $livenessTimeout = 90.0;

  protected function workerRun(): void
  {
      while ($this->isRunning()) {
          $hook = $this->hooks->takeNext();

          if ($hook === null) {
              $this->sleep(1);
              continue;
          }

          $this->markBusy();

          try {
              $this->http->post($hook->url, $hook->payload);
              $this->hooks->markDelivered($hook->id);
          } catch (\Throwable $e) {
              $this->hooks->retryLater($hook->id);
              $this->logger->warning('webhook failed', ['id' => $hook->id]);
          } finally {
              $this->markIdle();
          }
      }
  }
}

Три момента, без которых флот работает хуже: markBusy() не даёт остановить воркера посреди доставки и сбрасывает состояние между задачами; retryLater() в catch возвращает задачу в очередь; $livenessTimeout ловит зависшие HTTP-вызовы.

Автомасштабирование по глубине очереди

<?php

namespace Main\Process;

use Flytachi\Winter\DI\Attribute\Autowired;
use Flytachi\Winter\Kernel\Process\Daemon\ScalingPolicy;
use Flytachi\Winter\Kernel\Process\Stereotype\Daemon;

class Imports extends Daemon
{
  #[Autowired] private ImportQueue $queue;

  protected int $replicas = 2;          // базовый размер
  protected float $grace = 120.0;       // импорт нельзя рвать на середине

  protected function desiredReplicas(): int
  {
      return min(16, max(2, (int) ceil($this->queue->pendingCount() / 50)));
  }

  protected function scaling(): ScalingPolicy
  {
      return new ScalingPolicy(
          scaleUpDelay:           5.0,    // всплеск должен продержаться 5 с
          scaleDownStabilization: 300.0,  // сокращаем только после 5 минут затишья
          scaleStep:              4,      // не больше 4 воркеров за раз
      );
  }
}

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

Обратите внимание, что workerRun() здесь нет — значит, тело берётся из $workerClass. Если его тоже не задать, демон будет падать при каждом форке.

Один воркер, но с присмотром

Демон осмысленен и с единственной репликой: вы получаете перезапуск после падений и сторож зависаний, которых у обычного процесса нет.

class SnmpPoller extends Daemon
{
  protected int $replicas = 1;
  protected float $livenessTimeout = 60.0;

  protected function restart(): RestartPolicy
  {
      return new RestartPolicy(mode: RestartMode::ALWAYS, backoff: 5.0);
  }

  protected function workerRun(): void { /* ... */ }
}

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

Дальше

• Процессы — примитивы тела воркера: цикл, паузы, единицы работы
• Состав приложения — где объявляется #[EnableDaemon]
• Планировщик — когда работа привязана ко времени
• Логирование — куда пишет супервизор
• Actuator / Health — наблюдение за приложением снаружи