Пробуем-с

Пропал на 3 недели, но не просто так – активно преисполнялся в Elixir. В субботу планирую постримить, но не могу выбрать направление:

– Супервизор процессов на Elixir — написать лёгкий менеджер, который будет управлять процессами: запускать, останавливать, перезапускать, мониторить, добавлять в автозапуск. systemd писать не собираюсь, но минималистичный аналог runit — почему бы и нет? Вместе разберемся, как под капотом работают системы инициализации в *nix и напишем свою маленькую, да не абы как, а в функциональном стиле, на Elixir

– Fullstack на Vue3 + Express.js — помогаю товарищу с дипломной работой. В планах: авторизация, чат, вебсокеты и медитативная вёрстка.

Какой путь выбрать — решать тебе. Буду очень рад делегировать выбор на тебя😃

Выбираем реакцией:
Супервизор – 👀
Диплом – 🌚

P.S.: нормально настроил OBS в плане баланса звука, запись со стрима будет🫡

То чувство, когда большая задача пошла в релиз без правок ни с ревью, ни с тестирования:

Начинаю ориентировочно в 12:00. Буду рад всем желающим🙏🏾
https://www.twitch.tv/lsdrfrx

Начинаем-с🏃‍♂

Через 10 минут продолжаем

Стрим длился рекордные 4 часа 40 минут🫣
За это время успел накидать пару базовых компонентов, построить скелет фронтенда и сделать регистрацию и вход в аккаунт на стороне бэкенда. Попробовал новые для себя фреймворки Hono и Drizzle. Пока всё тяп-ляп, далее буду приводить все в красивый вид – как и структуру, так и визуал
Смотри запись тут: https://www.youtube.com/watch?v=jADfBMvaW4c

Щас должно работать, забыл опубликовать👉🏾👈🏾

К слову об искусстве нейминга🙊

☠️ Starvation

Не страшно, когда приложение падает, а журнал логов превращается в километровый красный ковёр трейсбэка ошибки. Не страшно даже, когда приложение падает без ошибок. Действительно страшно то, когда приложение умерло, но все выглядит так, что оно продолжает работать. Такую ошибку я поймал сегодня, буквально 2 часа назад, прогоняя нагрузочный тест – сервис просто внезапно умолк. Супервизор живой, все процессы живые, а в логах царит гробовая тишина. Не сказать, что я спец в конкурентности – у меня есть небольшой опыт разработки на языке Go (полтора-два года, если память не изменяет) – но так вышло, что сейчас я работаю над небольшим высоконагруженным сервисом, полностью пропитанным мультипроцессингом и конкурентностью. Думаю, многие слышали о deadlock и livelock, но сегодня я буду рассказывать не об этом

Эта ошибка не была для меня сюрпризом – я знал, где она спряталась. Более того – я осознанно допустил её, чтобы быстрее добиться рабочего результата на низких нагрузках. Сейчас пришло время её исправлять, и я об этом расскажу. То самое «исправлю потом» наступило для меня сегодня😃

У меня есть пайплайн обработки событий:
- события приходят из источника;
- они группируются по stream_id;
- для каждого stream_id создаётся отдельный воркер;
- внутри одного стрима порядок обработки строго последовательный, между стримами – параллель.

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

Но есть один нюанс.
Центральный процесс (Processor) хранил состояние всех очередей, а воркеры:
- брали события на обработку;
- после каждого шага синхронно спрашивали у Processor, что у них сейчас в очереди.

То есть десятки, сотни, тысячи воркеров постоянно делали синхронный запрос к одному процессу. Под небольшой нагрузкой система работала идеально. Под средней – начинала думать. Под большой нагрузкой во время стресс-теста – гробовая тишина. Процессы живые, supervisor живой, нет эксепшенов, ошибок в логах.
Это не deadlock в классическом смысле – никто не держит ресурсы навсегда. Это не livelock – процессы не крутятся в бесполезном цикле. Это starvation.

Центральный Processor захлёбывается от входящих сообщений:
- асинхронные сообщения о завершённой обработке;
- синхронные сообщения от воркеров на получение новых событий;
- служебные сообщения.

Я нарушил одно из базовых правил конкурентного дизайна:

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

Ещё хуже – воркеры не владели своими очередями, постоянно синхронно читали чужое состояние и зависели от одного центрального процесса.

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

# Модуль Processor
# Структура состояния:
# state = %{stream_id => [item1, item2, ...]}

def handle_cast({:push, item}, state) do
  queue = Map.get(state.items, item.stream_id, [])
  new_queue = queue ++ [item]

  # если очередь новая — стартуем воркер
  start_worker_if_needed(item.stream_id)

  # Добавляем новое событие в состояние Processor
  {:noreply, push(state.items[item.stream_id], new_queue)}
end

# Модуль Worker
def process_queue(stream_id) do
  case GenServer.call(Processor, {:get_queue, stream_id}) do
    [item | _] ->
      process_item(item)
      GenServer.cast(Processor, {:item_processed, item})
      process_queue(stream_id)

    [] ->
      :ok
  end
end

Ремарка для любопытных: функция `process_queue` рекурсивная, поскольку в уже существующий стрим могут докладываться новые события.

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

А чтобы очередь не терялась при падении воркера, источник данных становится единственным source of truth. Воркер загружает очередь при старте. Если он упал – его состояние легко восстановить.

Правильный пример:

# Модуль Processor
def handle_cast({:push, item}, state) do
  unless worker_running?(item.stream_id) do
    start_worker(item.stream_id)
  end

  :ok
end

# Модуль Worker
def init(stream_id) do
  queue = Source.fetch_stream(stream_id)
  send(self(), :process_next)
  {:ok, %{stream_id: stream_id, queue: queue}}
end

def handle_info(:process_next, %{queue: [item | rest]} = state) do
  process_item(item)
  Source.mark_processed(item.id)
  send(self(), :process_next)
  {:noreply, %{state | queue: rest}}
end

def handle_info(:process_next, %{queue: []} = state) do
  notify_processor(:queue_done, state.stream_id)
  {:stop, :normal, state}
end

Более того, так и выглядит поаккуратнее.

⚡️Ports – разгон Elixir до скорости C

Elixir, как и любой другой исполняемый на виртуальных машинах язык программирования, не может похвастаться высокой производительностью в CPU-bound задачах, и даже JIT не всегда решает эту проблему. Причиной этому являются планировщик процессов, сборка мусора, hot-reloading, иммутабельность данных. Его суперсила заключается в параллелизме – если CPU-bound задачу можно распараллелить, то проблема производительности встаёт не так остро. Но что делать, если перформанса не хватает? В этом случае на подмогу приходят они – Ports и NIF. Благодаря этим решениям можно реализовать сложную логику на низкоуровневом высокопроизводительном языке вроде C, Rust или Zig и вызывать её из BEAM. Сегодня я расскажу о самом простом и безопасном способе – Ports

Port – канал связи между BEAM и внешним миром. С помощью него виртуальная машина запускает скомпилированный исполняемый файл в отдельном процессе операционной системы, общается с ним по stdio и управляет его жизненным циклом. Выглядит это следующим образом:

// adder.c
// Будет скомпилирован в adder
#include <stdio.h>
#include <stdlib.h>

int main(int argc, char *argv[]) {
    if (argc != 3) {
        printf("ERROR: Need exactly 2 arguments\n");
        return 1;
    }
    
    int a = atoi(argv[1]);
    int b = atoi(argv[2]);
    
    printf("%d\n", a + b);
    return 0;
}

Реализуем для этой программы порт на Elixir:

defmodule AdderPort do
  def add(a, b) do
    # Запускаем порт с аргументами
    port = Port.open(
      {:spawn, "./adder #{a} #{b}"},
      [:binary, :exit_status]
    )
    
    # Получаем результат из stdout
    receive do
      {^port, {:data, result}} ->
        result
        |> String.trim()
        |> String.to_integer()
        
      {^port, {:exit_status, status}} ->
        {:error, "Exit status: #{status}"}
    after
      1000 -> {:error, :timeout}
    end
  end
end



Пример прост: программа adder принимает два аргумента, складывает их и записывает в stdout. BEAM читает результат и выводит его на экран. В случае аварийного завершения процесса функция AdderPort.add вернёт ошибку со статусом завершения работы. Использование написанного порта выглядит следующим образом:

iex> AdderPort.add(5, 2)
7
iex> AdderPort.add("a", 5)
{:error, "Exit status: -1"}

Благодаря Ports можно безопасно запускать внешние процессы, передавать им сообщения и получать результат. Однако этот подход имеет и недостатки:
- Безопасность: можно запустить что угодно, даже rm -rf /
- Производительность: port запускает отдельный процесс операционной системы, вследствие чего требуется дополнительное время на сериализации и межпроцессное взаимодействие IPC. Пропускная способность также ограничена
- Утечки: каждый port – отдельный процесс ОС, а это значит, что есть риск забить сервер до отказа, если порты запускаются в цикле и не закрываются
- Эффективность: данное решение эффективно в случае, если процесс живёт долго. Для частых вызовов это не подойдёт

Это самый простой способ взаимодействия BEAM с кодом на других языках. О NIF – молниеносно быстром вызове функций без накладных ресурсов – я расскажу в следующем посте