Привет! Меня зовут Никита Калинский, я разработчик в Тинькофф Бизнесе. Сейчас я занимаюсь продуктом под названием «Лента операций». Физлица в желтом приложении могут отслеживать все свои операции, и мы делаем такой же инструмент для предпринимателей.

Сегодня я хочу поговорить про основы различных систем исполнения эффектов в Scala. Мы разберем, как работают системы эффектов, как они реализованы в Scala в Cats Effects и ZIO и как эволюционировали между версиями. А также обсудим неявные особенности и подводные камни исполнения сред таких библиотек.

Как работают асинхронные среды выполнения

Представим, что заказчик попросил нас написать очередной веб-сервер, который может держать несколько сотен RPS, обрабатывать запросы пользователей и ходить в базу данных.

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

В таком подходе есть проблема, которую в сообществе называют C10k. Чтобы решить ее, нужно сконфигурировать сервер так, чтобы он мог не просто обрабатывать тысячи запросов, но делать это эффективно. Создавать отдельный тред на каждый запрос — не лучшая идея. Во-первых, так можно истощить ресурсы операционной системы, во-вторых, создание треда — это дорогая операция.

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

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

Помощь приходит неожиданно — из другого языка, а именно Rust. Для него написали фреймворк асинхронного программирования Tokio. Основная идея фреймворка в том, что у нас появляется дополнительная абстракция вместо тредов — Fiber. Это легковесный тред, который можно создать просто и быстро. Из-за маленького memory footprint можно насоздавать их сколько угодно, ограничены мы только лимитом памяти своего сервиса.

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

Теперь в наш планировщик попадают не просто запросы, а запросы, обернутые в файберы. А в тред попадает уже сам файбер на выполнение.

Часто на проде нужно не только выполнять задачи асинхронно, но и уметь протаскивать какой-то контекст в запросе для его идентификации в логах или трейсинге. Например, ID пользователя или запроса. Если бы мы работали без файберов, используя только треды, на помощь пришел бы Thread Local. Он позволяет прикрепить в определенный тред некие значения — собственные для каждого треда. Если же мы хотим внутри этого треда запустить еще один, нужно пробросить и в него этот контекст.

Для этого есть расширение Inheritable Thread Local, но тут могут возникнуть уже другие проблемы: так как мы используем планировщик, у нас ограниченное количество тредов, которые будут переиспользоваться. Соответственно, каждый Thread Local после выполнения запроса нужно очищать, иначе получится каша из контекстов. Если мы используем несколько пулов, нужно еще и уметь пробрасывать контекст между двумя пулами. Это хаки, которых хотелось бы избежать.

Файбер тоже имеет контекст, в котором мы можем хранить необходимые значения. Но когда один файбер создает внутри себя другой, контекст автоматически перекладывается в контекст «ребенка». И так как файбер не переиспользуется, после выполнения задачи он удаляется коллектором мусора. То есть не нужно ничего вычищать и перемешивания контекстов не случится. 

Что такое Runtime в системе эффектов

Runtime позволяет запустить файбер, который несет внутри себя эффект, то есть описывает некое вычисление. Для этого нужны:

1. Executor, то есть Main ThreadPool, на котором будут выполняться основные вычисления. 

2. Дополнительные пулы потоков для блокирующих или отложенных вычислений (опционально).

Примерно так выглядит Main Loop любой системы эффектов. При запуске программы происходит инициализация: например, чтение всех конфигов и подключения к базе данных. Все это заворачивается в один главный файбер и отправляется в Thread Pool. Там первый тред берет этот файбер из очереди и выполняет все, что в него завернуто. 

После этого файбер ждет, когда программа завершится или придет сигнал о принудительном завершении извне. При этом на каждый запрос веб-сервер создает новый файбер, который тоже отправляется в Thread Pool, выполняется тредом и возвращает результат. Также в JVM есть Garbage Collector, который подчищает старые файберы, уже завершившие свое выполнение.

class Fiber(
  context: FiberContext,
  callStack: Stack[Any => IO[Any]]
) {

  def run(io: IO[Any]): Unit = {
    while (currIteration < maxIterations) {

      // Паттерн-матчинг по типу эффекта,

      // с заполнением стека вызовов

      // и добровольным освобождением треда при смене executor'a

      // или семантической блокировке

	}

	// Добровольное освобождение треда
  }
}

class FiberContext(
  id: FiberId,
  scope: Scope,
  localValues: Map[Any, Any]
)

Примерно так обычно выглядит абстрактный файбер:

1. Первое, что мы видим, — контекст, внутри которого хранится Fiber ID, и Scope, к которому привязан файбер. Он может быть глобальным, если задача запускается периодически. Также это может быть Scope родителя, если файбер — это подзадача внутри другого файбера. Ну и ассоциативный массив с внутренними значениями контекста.

2. Также мы видим собственный стек вычислений, который заполняется по ходу выполнения задачи, завернутой в этот файбер. IO в этом случае — это ленивое вычисление (эффект) каждого этапа вычислений.

3. И видим главную функцию run, которая умеет запускать наши ленивые IO вычисления. В ней чаще всего есть while loop: он итерируется до заданного лимита, после которого файбер добровольно отдает тред. Внутри происходит pattern matching по типу эффекта, заполняется стек вызовов и разворачиваются вычисления.

С общей частью мы закончили, перейдем к рассмотрению библиотек.

ZIO

На картинке ниже мы видим примерный рантайм ZIO 2. Он параметризуется по типу R и типизирует ZEnvironment, который несет в себе все, что необходимо для dependency injection. Есть fiberRefs — коллекция FiberRef (изначально пустая), которая будет хранить в себе все ссылки на контексты файберов. И некий набор флагов рантайма. Также мы видим функцию run, которая позволяет принять описание эффектов в ZIO и запустить его синхронно.

Под капотом все файберы запускаются на отдельном тредпуле для вычислений. Кроме того, есть тредпул внутри ScheduledExecutorService, который нужен для запуска отложенного эффекта по расписанию.

trait Runtime[+R] {
  def environment: ZEnvironment[R]
  def fiberRefs: FiberRefs
  def runtimeFlags: RuntimeFlags

  trait UnsafeAPI {
    def run[E, A](zio: ZIO[R, E, A]): Exit[E, A]
  }
}

object Runtime {
  def apply[R](
    r: ZEnvironment[R],
    fiberRefs0: FiberRefs, 
    runtimeFlags0: RuntimeFlags
  ): Runtime[R]
}

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

В ZIO 2 создатели фреймворка реализовали собственный ThreadPool, который умеет обмениваться задачами между тредами (aka Work Stealing). Есть как глобальная очередь, в которую попадают файберы извне, так и локальные очереди у каждого из тредов. 

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

Теперь разберемся, как создавать файберы в ZIO. Представим, что у нас есть два вычисления. Они возвращают числа, и мы хотим запустить их в параллель, чтобы получить результат и сложить их. Для этого нужно вызвать fork на каждом из них. Это позволит вернуть отложенный файбер, обернутый в эффект. Чтобы запустить этот файбер, нужно скомпозировать его через flatMap. Чтобы подождать результат вычисления файбера, нужно вызвать на нем join. В этот момент произойдут семантические блокировки: сначала в ожидании результата result1, потом result2.

type ZResult = ZIO[Any, Throwable, Int]

val someComputation: ZResult = ???

val anotherComputation: ZResult = ???

for {
  fiber1  <- someComputation.fork // UIO[Fiber[Throwable, Int]]
  fiber2  <- anotherComputation.fork
  result1 <- fiber1.join // Int
  result2 <- fiber2.join // Int
} yield result1 + result2

Но не рекомендуется работать с интерфейсами файбера напрямую. Если, например, после его создания через fork в основных вычислениях произойдет ошибка, файбер продолжит выполняться до завершения собственной задачи. Со временем Garbage Collector, конечно, очистит его, но мы просто впустую потратим мощности на неактуальные вычисления.

for {
  first  <- someComputation.fork
  _      <- ZIO.fail(new RuntimeException())
  result <- first.join
} yield result

Поэтому лучше использовать безопасные функции вроде zipPar. Она позволяет «склеить» два вычисления c обработкой ошибок, то есть в случае неполадок все файберы отменятся рантаймом.

for {
  tuple <- someComputation.zipPar(anotherComputation)
  (result1, result2) = t
} yield result1 + result2

Еще одна интересная особенность реализации файберов в ZIO — принцип Fork-Join Identity: если создать новый файбер и тут же семантически заблокироваться в ожидании его результата, это должно равняться отсутствую запуска файбера в принципе.

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

type ZResult = ZIO[Any, Throwable, Int]

val someComputation: ZResult = ???

val anotherComputation: ZResult = ???

val parallelComputation = 
  for {
    fiber1  <- someComputation.fork // UIO[Fiber[Throwable, Int]]
    fiber2  <- anotherComputation.fork
    result1 <- fiber1.join // Int
    result2 <- fiber2.join // Int
  } yield result1 + result2

parallelComputation.fork.flatMap { fiber =>
  clock.sleep(1.second).flatMap( => fiber.interrupt) // interrupt здесь отменит выполнение fiber1 и fiber2 тоже
}

Если мы хотим это обойти, можно запускать файберы на отдельном Scope. Либо создать вычисление, которое будет выполняться с определенной периодичностью при помощи delay + forever. В таком случае файбер будет привязан к глобальному Scope самой программы и завершится, только когда завершится она.

val customScope: ZScope[Exit[Any, Any]] = ??? // в ZIO2 ZScope переехал в новую структуру Scope, похожую на ZManaged

val fiberOnCustomScope: UIO[Fiber[Throwable, Unit]] =
  ZIO.effect(println("Running on custom scope"))
    .forkIn(customScope)

val fiberOnGlobalScope: UIO[Fiber[Throwable, Unit]] =
  ZIO.effect(println("Daemon running"))
    .delay(1.second)
    .forever
    .forkDaemon

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

Если же по ходу вычислений обнаруживается ошибка, весь стек вычислений файбера разворачивается в попытке найти ее обработчик. Если найти его не получается, ошибка отправляется в Main Loop и может вызвать остановку программы. Кроме того, можно явно добровольно освободить поток (Yield) или переместить вычисления на другой пул потоков (Shift), например для блокирующих операций.

def run[E](zio: ZIO[Any, R, Any]) = {
  while (curIteration < 2048) {
    val tag = zio.tag
    
    tag match {
      case ZIO.Tags.FlatMap => // Трансформация
      case ZIO.Tags.Fork => // Создание нового файбера и отправка в Executor
      case ZIO.Tags.Fail => // Разворот стека в попытке найти ZIO.Fold
      case ZIO.Tags.Yield => // Добровольно отдать тред
      case ZIO.Tags.Shift => // Отправить вычисление в другой Executor и отдать тред
      ...
    }
  }
  // Добровольно отдать тред
}

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

Соответственно, чтобы создать блокирующее вычисление, мы можем использовать effectBlocking (непрерываемый) или effectBlockingInterrupt, который под капотом будет вызывать Thread.interrupt при отмене файбера. Либо мы можем задать собственный эффект, который будет выполняться при прерывании файбера. В примере ниже при запуске вычисления открывается сокет, который гарантированно закроется при отмене вычисления.

import zio.blocking._

val simpleBlocking = effectBlocking(Thread.sleep(3000))

val blockingInterruptible =
  effectBlockingInterrupt{ Thread.sleep(3000) }

val blockingCancelable =
  effectBlockingCancelable(
    effect = socket.open()
  )(cancel = socket.close())

Кроме того, в ZIO 2 разработали специальную систему мониторинга, которая анализирует запущенные файберы на предмет вовлечения в блокирующие операции. Если она их обнаруживает, то автоматически переводит на блокирующий пул потоков, даже если пользователь явно не вызвал effectBlocking. Но, как обычно бывает с подобными системами, особенно сразу после их появления, не стоит считать, что она сработает во всех случаях. Поэтому старайтесь всегда использовать явное переключение на блокирующий пул потоков, если уверены, что описанная операция блокирующая.

Подведем промежуточные итоги по ZIO. Начнем с плюсов: 

• Файберы с семантической блокировкой.

• Их можно отменять.

• Они используют cooperative yielding.

• Fork-join identity + возможность прикрепления файбера к определенному окружению (scope).

• Определение эффектов при запуске по их тегу, а не полному типу.

• Блокирующие операции гарантированно выполняются на одном executor.

• Отмена блокирующих эффектов.

Минусы, которые были в ZIO 1, но отсутствуют в ZIO 2:

• Одна глобальная очередь задач в пуле потоков может стать источником деградации перформанса.

• Неоптимальное использование локальных кэшей процессора.

Cats Effects 2/3

В Cats Effect 2 runtime выглядел немного иначе. Там были:

• функция run, описывающая всю нашу программу как IO;

• main, который умеет запускать программу;

• contextShift — абстракция, которая позволяет переключаться между разными пулами потоков;

• timer — обертка над ScheduledExecutorService;

• executionContext — дефолтный тредпул, который хранится в нашем приложении.

trait IOApp {
  def run(args: List[String]): IO[ExitCode]

  def main(args: Array[String]): Unit =
    IOAppPlatform.main(
      args,
      Eval.later(contextShift),
      Eval.later(timer)
    )(run)

  implicit protected def contextShift: ContextShift[IO] =
    IOAppPlatform.defaultContextShift

  implicit protected def timer: Timer[IO] =
    IOAppPlatform.defaultTimer

  protected def executionContext: ExecutionContext =
    IOAppPlatform.defaultExecutionContext
}

Под капотом был пул потоков с фиксированной размерностью. Его максимальный размер — доступное количество ядер процессора, но не больше двух. С таким пулом были те же проблемы производительности, что и у ZIO 1.

В Cats Effect 3 runtime переписали, и он стал похож на ZIO 2: с одной глобальной очередью, локальными очередями у каждого потока и возможностью «воровать» часть задач из локальных очередей.

В итоге поменялся и сам IOApp:

• Все тредпулы — для обычных вычислений, блокирующих и по расписанию — уехали в отдельную структуру IORuntime.

final class IORuntime(
  val compute: ExecutionContext,
  val blocking: ExecutionContext,
  val scheduler: Scheduler,
  val fiberMonitor: FiberMonitor,
  val shutdown: () => Unit,
  val config: IORuntimeConfig
)

• Функциональность ContextShift и Blocker (про него чуть ниже) разнесли по другим тайпклассам (Async и Sync), полностью изолировав работу с переключением между разными тредпулами, гарантировав соблюдение local reasoning и выполнение вычислений на тех же самых тредпулах, на которых они и были запущены.

Файберы в Cats Effect запускаются примерно так же, как в ZIO, но через функцию start. Отменить их можно при помощи вызова cancel на этом файбере. При этом Fork-Join Identity в CE2 отсутствовал — его добавили в CE3. В CE тоже есть более безопасные функции для работы с файберами, например parMap. Но использовать их чуть сложнее, так как это не статичные функции IO, как в ZIO, а синтаксические расширения над тайпклассом Parallel. Их нужно явно импортировать из cats.syntax или cats.implicits объектов.

val someComputation: IO[Int] = ???

val anotherComputation: IO[Int] = ???

val unsafeJoin = for {
  fiber1 <- someComputation.start // IO[Fiber[IO, Int]]
  fiber2 <- anotherComputation.start
  res1   <- fiber1.join
  res2   <- fiber2.join
} yield res1 + res2

val safeJoin = (
  someComputation,
  anotherComputation
).parMap(_ + _)

Файбер Run Loop тоже работал иначе в CE2. Он выполнялся до тех пор, пока не выполнится все вычисление и файбер не завершится полностью. Создатели аргументировали это желанием избежать проблемы повышенных затрат на помещение файбера обратно в глобальную очередь и более оптимальным использованием локальных кэшей процессора. Паттерн-матчинг внутри происходил по типу эффекта, а не по его тегу, и это тоже было немного дольше, чем в ZIO.

def run[E](io: IO[Any], callStack: Stack[Any => IO[Any]]): Unit = {
  while (true) {
    io match {
      case Pure(v) => // Взять из стека следующее вычисление
      case ContextSwitch(...) => // Продолжить вычисление на другом Executor
      case Async(...) => // Сохранить колбэк и отдать тред
      case RaiseError(e) => // Найти обработчик ошибок или вернуть эту ошибку
      ...
    }
  }
}

В СЕ3 run loop тоже стал сильно схож с ZIO, так как появился лимит в 1024 итерации на выполнение одного файбера и паттерн-матчинг по тегу эффекта вместо проверки на тип.

Блокирующие операции точно так же выполняются на отдельном тредпуле. Для переключения на блокирующий пул потоков в CE2 была обертка Blocker. Одним из ее главных недостатков было отсутствие гарантии выполнения вычисления на пуле для блокирующих операций. Главным образом это проявлялось в наличии асинхронной границы или ручного переключения на другой пул потоков внутри этого вычисления. 

Например, в коде ниже происходит блокирующее чтение из файла, но в промежутке пользователь решил поменять тредпул для логирования прочитанной строчки, чтобы не делать этого на блокирующем тредпуле. В итоге после лога он продолжит читать файл уже не на блокирующем тредпуле. И хоть это вырожденный пример, представьте, что в программе вы в blockOn передаете IO, которое получаете в результате вызова какой-нибудь другой функции. Нельзя быть уверенным, что где-то внутри этой IO не вызывается shift. И это сильно нарушает принцип локальности рассуждений (local reasoning) о том, что делает программа.

import java.io.File
import java.util.Scanner
import cats.effect.{Blocker, ContextShift, Bracket, IO}

val blocker: Blocker = ???

val mainThreadPool: ContextShift[IO] = ???

Bracket[IO, Throwable].bracket(
  IO.delay(new Scanner(new File(“someFile”)))
) { scanner => blocker.blockOn{
  for {
    line <- IO.delay(scanner.nextLine)
    _ <- IO.shift(mainThreadPool)
    _ <- IO.delay(log(line))
    … // тут продолжите чтение файла
  } yield ()
}}(scanner => IO.delay(scanner.close))

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

class Blocker(ec: ExecutionContext) {
  def blockOn[A](io: IO[A])(implicit cs: ContextShift[IO]): IO[A] =
	cs.blockOn(this)(io)
}

implicit val cs: ContextShift[IO] = ???

val blocker: Blocker[IO] = ???

val jdbcRequest: IO[Int] = ???

blocker.blockOn(jdbcRequest)

В CE3 решили упростить иерархию тайпклассов относительно блокирующих операций. К тому же из-за перечисленных выше проблем функциональность Blocker унесли в Sync, а файберы блокирующих операций теперь можно попытаться отменить. При этом можно задать различное поведение для отмены:

• Sync[IO].blocking — выполнить непрерываемое вычисление на блокирующем пуле потоков.

• Sync[IO].interruptible — попробовать отменить файбер один раз, послав в него Thread.interrupt(). Если не отменился, просто продолжаем выполнение программы.

• Sync[IO].interruptibleMany — пробовать отменять несколько раз, пока не получится.

val nonInterruptible =
  Sync[IO].blocking(println("blocking non interruptible"))

val interruptibleOnce =
  Sync[IO].interruptible(
    println("blocking interruptible one try")
  )

val interruptibleMany =
  Sync[IO].interruptibleMany(
    println("blocking interruptible many tries")
  )

Функцию ContextShift в CE3 теперь выполняет Async. Он несет в себе ссылку на тредпул, к которому привязаны основные вычисления, и позволяет выполнить вычисление F[A] на явно заданном другом тредпуле. При этом после выполнения F[A] гарантируется возврат на основной.

trait Async[F[_]] {
  def evalOn[A](fa: F[A], ec: ExecutionContext): F[A]
  def executionContext: F[ExecutionContext]
}

Изменения рантаймов в ближайших релизах

Разработчики ZIO хотят в скорых релизах реализовать поддержку Project Loom — реализации нативного рантайма с файберами в JVM. Одна из главных целей поддержки Loom в ZIO, кроме, конечно, нативной поддержки файберов на уровне JVM, — это облегчение перехода на Loom для пользователей ZIO. То есть создатели ZIO хотят обеспечить максимально незаметный и плавный переход для всех желающих.

В Cats Effect же в релизе 3.5.0 должны избавиться от отдельного пула потоков ScheduledExecutorService для выполнения операций по расписанию. Они хотят перенести эту работу полностью на главный пул вычислений: каждый поток периодически или при отсутствии задач в его очереди будет проверять, не истекли ли таймеры у подобных операций, инициализированных на нем. Если истекли, то продолжит их выполнение, иначе будет «припаркован» до истечения ближайшего таймера. Это даст прирост производительности в 15—20%, так как теперь не придется после истечения таймера переносить вычисления обратно на главный пул потоков. Это особенно актуально, например, для http-серверов, обрабатывающих большое количество параллельных запросов, так как при открытии нового соединения на него вешается тайм-аут для последующего закрытия.

Заключение

На этом все. Если по определенным причинам перед вами стоит выбор  — CE2 или ZIO 1, то лучше выбрать второй вариант. С ним у вас будет выигрыш в производительности, к тому же вы избежите проблем из-за неоптимальности выполнения файберов и различных неявных переключений между пулами потоков.

Если же выбирать нужно между CE3 и ZIO2, стоит опираться исключительно на то, что вам кажется удобнее в использовании. Каких-то кардинальных различий в асинхронной модели выполнения эффектов и производительности у этих библиотек практически нет.

Если интересно узнать о разнице по производительности между библиотеками, советую более-менее честный бенчмарк от создателя Cats Effect.

Полезные ссылочки:

• Про Cats Effect 3 и его эволюцию из Cats Effect 2 от создателей библиотеки.

• Про модель потоков от них же и почему файберы быстры в исполнении. (некоторые картинки про планировщики были взяты отсюда)

• Про файберы в Rust Tokio и корутины в Go.

• Еще одно объяснение подкапотной работы файберов. 

• Про файберы в ZIO.

• Новинки ZIO2 и его дальнейшие планы.

• Видео моего доклада.