Разбавляем асинхронное программирование функциональным на Scala

Приветствую! В этой статье будет показано, как, имея на руках обычные Future-ы, сделать в scala подобие корутин и асинхронные stream-ы. Этакий небольшой туториал по функциональному программированию.

Что это и зачем

К сожалению, простое последовательное комбинирование 2х и более асинхронных операций (загрузить конфиг, а потом картинку по адресу из конфига как пример) — это все, на что способны обычные Future-ы в качестве монад. Они не позволяют ни сохранять состояние, ни делать циклы из асинхронных операций, ни выдавать несколько (или бесконечно много) значений. Вот этими недостатками мы сейчас и займемся.

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

Применив знания из этой статьи, мы сможем этот процесс описать примерно так:

// Про 'FState' - далее, пока же просто примем, что это - такая необычная Future
def getNextConfig: FState[Config]
def getTemperature(from: String): FState[Int]

case class State(temperature: Int, sumTemp: Long, count: Int) {
  def isGood = ...
}

// Как видим, получается единый асинхронный алгоритм с состоянием, 
// которое извне этого алгоритма не видно
val handle = 
  while_ ( _.isGood)
  {  for (
        config <- getNextConfig();
        if (config.isDefined);  // пустой конфиг - прекращаем выполнение
        nextValue <- getTemperature(config().source);  // грузим значение температуры
        state <- gets[State];  // тут мы берем текущее состояние
        newState = State(nextValue, state.sumTemp + nextValue, state.count + 1);
        _ <- puts(newState);  // .. и меняем его
        _ <- runInUiThread { drawOnScreen(newState) }
  ) yield() }

Или вот так:

val configs: AsyncStream[Config] = ... // получаем откуда-то stream конфигов

def getTemperature(from: String): FState[Int]

case class State(temperature: Int, sumTemp: Long, count: Int)

// Получается то же самое, только вместо зависимости 'getNextConfig'
// мы, по сути, передаем сами данные - stream из конфигов
val handle = 
  foreach(configs) {
    config => for (
        nextValue <- getTemperature(config().source);  // грузим значение температуры
        state <- gets[State];  // тут мы берем текущее состояние
        newState = State(nextValue, state.sumTemp + nextValue, state.count + 1);
        _ <- puts(newState);  // .. и меняем его
        _ <- runInUiThread { drawOnScreen(newState) }
    ) yield()  
  }

Всех, кто заинтересовался, прошу под кат.

Stateful asynchronous computations

Это такая Future, которая позволяет сохранять и менять состояние псевдоалгоритма внутри for-конструкции, те самые gets[State] и puts[State]. И то, что придает этому самому псевдоалгоритму некую корутинообразность.

Давайте рассмотрим вот такую вот интересную сущность:

// S - тип внешнего состояния, A - тип результата
case class FState[S, +A](func: S => Future[(A, S)]) {
  def apply(s: S) = func(s)
}

Как видим, это простая обертка над функцией, принимающей текущее состояние и возвращающей Future на результат в паре с новым состоянием. Вид этой сущности получен простым комбинированием монад Future и State (а то, что мы сейчас делаем, называется монадным трансформером).

Давайте научим эту сущность быть монадой. В принципе, нам достаточно определить у этой сущности операции unit и flatMap (а еще map, которая выражается через первые две), но мы пойдем сразу тернистым путем scalaz, и получим бонусом целую алгебру операций, определенную в терминах вот этих двух.

class FStateMonad[S] extends Monad[({ type f[X] = FState[S, X]})#f] {
  type F[X] = FState[S, X]

  override def point[A](a: => A): F[A]
  override def bind[A, B](m: F[A])(f: A => F[B]): F[B]
}

class FStateMonad[S] extends Monad[({ type f[X] = FState[S, X]})#f] {
  type F[X] = FState[S, X]

  override def point[A](a: => A): F[A] = FState((s: S) => Future((a, s)))
  override def bind[A, B](m: F[A])(f: A => F[B]): F[B] =
    FState((s: S) => m(s) flatMap { pair => f(pair._1)(pair._2) })
}

Например, мы только что бесплатно получили вот такую совершенно великолепную операцию:

// Цикл с монадным условием и монадным телом - красота!
def whileM_[A](p: F[Boolean], body: => F[A]): F[Unit]

А как же менять состояние внутри псевдоалгоритма? Мы можем заметить, что в комбинаторе bind оно не фигурирует. Пишем:

def gets[S](): FState[S, S]
def puts[S](news: S): FState[S, S]

// Как видим, чтобы нам получить текущее состояние, его достаточно 
// просто "поднять" в значение.
def gets[S](): FState[S, S] = FState((s: S) => Future((s, s)))

// А изменить состояние еще проще
def puts[S](news: S): FState[S, S] = FState((_: S) => Future((news, news)))

Вот, в принципе, и все! Теперь мы можем написать что-то вроде:

implicit val m = new FStateMonad[Int] // состояние - обычный Int

// И эта жуть просто считает до 10...
val algo = for(

  _ <- m.whileM_(gets[Int] map (_ < 10), for(
    i <- gets[Int];
    _ <- puts(i + 1)
  ) yield(()));
  v1 <- gets[Int]

) yield (v1)

// algo(0)() should be ((10, 10))

Но мы можем написать себе столько синтаксического сахара, сколько захотим, и закончить примерно на этом:

implicit val m = new FStateMonad[Int] // состояние - обычный Int

val algo = for(

  // почти обычный for, с условием, модификатором состояния и телом
  _ <- m.forM_ ( _ < 10, _ + 1) {
    // тут любой FState
  };
  v1 <- gets[Int]

) yield (v1)

// algo(0)() should be ((10, 10))

Асинхронные stream-ы

Это как раз такая Future-а, которая способна возвращать больше одного значения асинхронно. Во втором примере в начале статьи мы просто по этому stream-у итерируемся stateful-псевдоалгоритмом. Однако со stream-ом можно делать еще несколько разных интересностей, но давайте по порядку. Начнем с построения stream-а.

Что такое stream? Упрощенно — список с хвостом, вычисляемым лениво. Вследствие этого может быть бесконечным. Наш асинхронный AsyncStream будет выглядеть примерно так:

class Pair[A, B](fp: A, sp: => B) {
  val first = fp
  lazy val second = sp
}

object Pair {
  def apply[A, B](first:  A, second: => B) = new Pair[A, B](first, second)
}

// Конец stream-а обозначим как Future(null), это обязательно должна быть
// какая-то конечная Future-а, возвращающая какое-то пустое значение.
// Мы не будем возиться тут с Option-ом, а возьмем просто null.
case class AsyncStream[A](data: Future[Pair[A, AsyncStream[A]]])

Итак, вроде бы все просто: асинхронно возвращаем значение и хвост, а ленивость хвоста должна уберечь нас от stackoverflow при пользовании бесконечными stream-ами.

Но stream-ы обладают одной очень притягательной особенностью: их можно сворачивать! Пишем:

// Почти стандартный foldLeft, только возвращает Future[B]
// - другого мы ничего вернуть не сможем
def foldLeft[B](start: B)(f: (B, A) => B): Future[B]

def foldLeft[B](start: B)(f: (B, A) => B): Future[B] = {
  // Стандартно, определим функцию с аккумулятором, которой
  // потом передадим data
  def impl(d: Future[Pair[A, AsyncStream[A]]], acc: Future[B]): Future[B] =
    d flatMap (pair => {
      if (pair eq null) acc
      else impl(pair.second.data, acc map (b => f(b, pair.first)))
    })

   impl(data, Future(start))
}

Можно еще вот так свернуть:

def flatten : Future[List[A]]

def flatten : Future[List[A]] =
    foldLeft[List[A]](Nil)((list, el) => el :: list) map (_.reverse)

Еще парочка полезных функций, в основном для работы с бесконечными stream-ами:

def takeWhile(p: A => Boolean): AsyncStream[A]
def take(n: Int): AsyncStream[A]

def takeWhile(p: A => Boolean): AsyncStream[A] =
  new AsyncStream[A](data map (pair => {
     if (pair eq null) null
     else if (!p(pair.first)) null
     else Pair(pair.first, pair.second.takeWhile(p))
   }))

def take(n: Int): AsyncStream[A] =
    if (n <= 0) nil
    else new AsyncStream[A](data map (pair => {
      if (pair eq null) null
      else Pair(pair.first, pair.second.take(n - 1))
    }))

Сворачивать-то мы научились, а вот строить такой stream пока что неудобно. Исправим это и напишем универсальный генератор stream-а:

// 'gen' и является функцией-генератором, она должна возвратить 
// Future(null), если больше значений сгенерировать не получается
def genAsyncStream[S,A](start: S)(gen: S => Future[(A, S)]): AsyncStream[A]

def genAsyncStream[S,A](start: S)(gen: S => Future[(A, S)]): AsyncStream[A] =
  new AsyncStream[A](
    gen(start) match {
      case _: NoFuture => Future(null)
      case future => future map (pair => { // Future[Pair[A, AsyncStream]]
        if (pair eq null) null
        else Pair(pair._1, genAsyncStream(pair._2)(gen))
      })})

Кстати, с помощью foldLeft + genAsyncStream stream-ы можно копировать.

Stream-ы можно соединять:

def concat[A](s1: AsyncStream[A], s2: AsyncStream[A]): AsyncStream[A]

def concat[A](s1: AsyncStream[A], s2: AsyncStream[A]): AsyncStream[A] =
  new AsyncStream[A](s1.data flatMap (pair => {
    if (pair eq null) s2.data
    else Future(Pair(pair.first, concat(pair.second, s2)))
  }))

А еще асинхронный stream — это тоже монада:

class AsyncStreamMonad extends Monad[AsyncStream] {
  override def point[A](a: => A): AsyncStream[A]
  override def bind[A, B](
    ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B]
}

class AsyncStreamMonad extends Monad[AsyncStream] {
  override def point[A](a: => A): AsyncStream[A] = unit(a)
  override def bind[A, B](
    ma: AsyncStream[A])(f: A => AsyncStream[B]): AsyncStream[B] =
    new AsyncStream[B](ma.data flatMap (pair => {
      if (pair eq null) Future(null)
      else f(pair.first).data map (
        pair2 => Pair(pair2.first, concat(pair2.second, bind(pair.second)(f))))
    }))
}

В основном, построение асинхронного stream-а на этом можно закончить.

FState + AsyncStream = ?

На самом деле, они прекрасно уживаются. Давайте посмотрим на функцию-генератор в genAsyncStream, ничего не напоминает? Да это же FState!

А теперь давайте научимся итерироваться по stream-у:

  def foreach[A, S]
    (stream: AsyncStream[A])(f: A => FState[S, Any]): FState[S, Unit]

// Сделаем это foldLeft-ом
def foreach[A, S]
  (stream: AsyncStream[A])(f: A => FState[S, Any]): FState[S, Unit] =
  FState(s => {
    stream.foldLeft(Future(s))(
    (futureS, a) => futureS.flatMap(s2 => f(a)(s2).map(_._2))).flatten.map( ((), _) )
  })

То есть мы совершенно спокойно можем написать генератор, который yield-ит значения в асинхронный stream, потом этот stream куда-то передать, и по нему уже итерироваться другим алгоритмом — довольно удобно, имхо, когда надо передать данные из одного модуля программы в другой и не хочется внедрять зависимости.

Итог

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

Код здесь.

Автор: iboltaev

Источник