Курс Elixir: Стримы

Идея ленивых вычислений состоит в том, что у нас есть некая функция, результат которой мы хотим вычислить не сразу, а чуть позже, когда будет нужно. Как это может быть полезно? Рассмотрим на примерах.

В большинстве языков программирования вычисления происходят в момент их описания, а к примеру, в Haskell, все вычисления по умолчанию ленивы.

Наивная реализация ленивых вычислений - анонимная функция, сохраненная в переменную, вызов которой сделаем дальше по коду, основная сложность возникает, когда мы хотим сделать вызов комбинации из таких "ленивых функций". Для этого есть модуль Stream. В нем есть такие же функции работы с коллекциями как map, filter, reduce, но с некоторыми отличиями, которые мы рассмотрим дальше:

numbers = [1,2,3]

numbers
  |> Enum.map(&(&1 * &1))
  |> Enum.zip([:a, :b, :c])
  |> Enum.filter(fn({n, s}) -> n > 2 and s != :c end)
# => [{4, :b}]

В этом случае мы прошлись три раза по всему списку, так как промежуточные результаты передаются дальше по цепочке вызовов функций. Если бы мы обходили список в каком-нибудь императивном языке, то использовали бы примерно в 3 раза меньше памяти и времени CPU. Теперь используем модуль Stream:

numbers = [1,2,3]

numbers
  |> Stream.map(&(&1 * &1))
  |> Stream.zip([:a, :b, :c])
  |> Stream.filter(fn({n, s}) -> n > 2 and s != :c end)
  |> Enum.to_list
# => [{4, :b}]

Результат остался тем же, только в конце добавили функцию Enum.to_list, которая запускает цепочку ленивых вычислений. Без него результат был бы таким:

numbers = [1, 2, 3]

numbers
  |> Stream.map(&(&1 * &1))
  |> Stream.zip([:a, :b, :c])
  |> Stream.filter(fn({n, s}) -> n > 2 and s != :c end)
# => #Stream<[
# =>   enum: #Function<73.29975488/2 in Stream.zip_with/2>,
# =>   funs: [#Function<40.29975488/1 in Stream.filter/2>]
# => ]>

Мы получили некую структуру данных, в ней описаны вычисления, которые будут применены к этой структуре при вычислении.

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

Enum.map(1..10_000_000, &(&1 + 1)) |> Enum.take(5)
# => [2, 3, 4, 5, 6]

Stream.map(1..10_000_000, &(&1 + 1)) |> Enum.take(5)
# => [2, 3, 4, 5, 6]

Второй случай вычислится быстрее. Почему? В первом случае вычислится весь список, во втором только нужная часть списка, а точнее первые пять элементов, именно так Stream и работает.

Возьмем еще один пример. Допустим у нас есть большой файл, который содержит всю текстовую часть статей из Википедии, который весит примерно 20 Гб и мы хотим в нем найти самое длинное слово. Реализуем через Enum:

File.read!("data/wiki.txt")
  |> String.split("\n")
  |> Enum.map(fn(line) -> String.split(line, " ") end)
  |> Enum.map(fn(words) -> ...
  |> Enum.max_by

В этом случае мы загрузим в оперативную память весь файл, который весит 20 Гб, так еще и пройдемся по нему 4 раза!

Теперь воспользуемся Stream:

File.stream!("data/wiki.txt")
  |> Stream.map(fn(line) -> String.split(line, " ") end)
  |> Stream.map(fn(words) -> ...
  |> Enum.max_by

Функция File.stream! возвращает ленивую коллекцию, которая построчно читает файл. В этом случае мы загружаем в оперативную память данные небольшими порциями и выполняем только один проход по ним. Последний вызов Enum.max_by запускает вычисления. Не факт, что этот код будет быстрее, так как мы много раз обращаемся к диску, но он точно расходует намного меньше оперативной памяти.

Иногда коллекции бывает бесконечными. На такой случай Stream имеет функции cycle, repeatedly, iterate, unfold, resource. Рассмотрим cycle и unfold.

Stream.cycle принимает коллекцию и генерирует бесконечную коллекцию из повторений элементов переданной:

Stream.cycle([1, 2, 3, 4, 5]) |> Enum.take(20)
# => [1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 1, 2, 3, 4, 5]

Это можно использовать, например для генерации HTML элементов:

texts = ["foo", "bar", "hello", "hexlet"]

Stream.cycle(["red", "black", "green"])
  |> Stream.zip(texts)
  |> Enum.map(fn {color, text} -> "<p class=text-#{color}>#{text}</p>" end)

# => ["<p class=text-red>foo</p>", "<p class=text-black>bar</p>", "<p class=text-green>hello</p>", "<p class=text-red>hexlet</p>"]

Stream.unfold можно представить, как функцию, обратную свертке (fold или reduce). Если fold сворачивает список в одиночное значение, то unfold наоборот, из одиночного значения разворачивает список.

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

Рассмотрим пример с числами Фибоначчи:

initial_state = {0, 1}

unfolder = fn({val1, val2}) ->
  curr_val = val1
  new_state = {val2, val1 + val2}
  {curr_val, new_state}
end

Stream.unfold(initial_state, unfolder) |> Enum.take(20)
# => [0, 1, 1, 2, 3, 5, 8, 13, 21, 34, 55, 89, 144, 233, 377, 610, 987, 1597, 2584, 4181]

Задание

Создайте функцию generate, которая случайно генерирует бесконечную последовательность чисел от 1 до 20 и берет n элементов от этой коллекции:

Solution.generate(5)
# => [2, 10, 20, 12, 11]
Solution.generate(2)
# => [7, 14]
Solution.generate(20)
# => [2, 9, 18, 1, 3, 16, 1, 1, 20, 20, 9, 11, 10, 14, 10, 15, 3, 3, 10, 8]