Gpseq

• Что такое Gpseq?

• Ссылки

Что такое Gpseq?

Gpseq - это библиотека предоставляющая параллелизм для Vala и GObject.

Она содержит следующий функционал:

• Work-stealing и managed blocking планировщик задач аналогичные планировщику Go.

• Обработка данных в стиле функциональной парадигмы с поддержкой параллельного выполнения: эквивалент потоков Java

• Fork-Join параллелизм

• Параллельная сортировка

• Фьючерсы и промисы

• 64-битные атомарные операции

• Переполнение безопасных арифметических функций для целых чисел со знаком

• ...

Ссылки

• Еще не зарегистрирован на Valadoc.org

• Локальная версия Valadoc

• GtkDoc (C API)

• (TODO) pgi-docs (Python API)

• (TODO) gjs-docs (JavaScript API)

Futures и promises

Содержание

• Future.of()

• Future.err()

• wait() and wait_until()

• value

• map()

• flat_map()

• light_map()

• map_err()

• then()

• and_then()

• zip()

• transform()

• Цепочка фьючерсов

• Зачем использовать Gpseq.Future вместо Gee.Future?

Библиотека Gpseq предоставляет два вида объектов Future и Promise.

Future - это read-only значение, которое становиться доступным в определенный момент. Значение (или исключение) обычно устанавливается с помощью Promise.

метод main() и using Gpseq;опущены

Пример 1. Futures и promises

Promise<int> promise = new Promise<int>();
Future<int> future = promise.future;
promise.set_value(123);
print("%d\n", future.value);

Вывод: 123

Promise<int> promise = new Promise<int>();
Future<int> future = promise.future;
promise.set_exception( new OptionalError.NOT_PRESENT("Oops!") );
print("%s\n", future.exception.message);

Вывод: Oops!

Операции Seq не блокируют текущий поток. Вместо этого большинство из них возвращают Future как результат. Он будет вычислен в качестве значения или исключения.

Примеры методов

Future.of ()

public static Future<G> of<G> (owned G value)

Future.of () - это вспомогательный метод, который возвращает Future, заполненный заданным значением.

Future<int> future = Future.of<int>(123);
print("%d\n", future.value);

Вывод: 123

123

Future.err ()

public static Future<G> err<G> (owned Error exception)

Future.err () - это вспомогательный метод, который возвращает Future, заполненный заданным исключением.

wait() и wait_until()

public abstract weak G wait () throws Error
public abstract bool wait_until (int64 end_time, out G value = null) throws Error

Пример 2. wait()

const ulong SEC = 1000000;

Promise<void*> promise = new Promise<void*>();
new Thread<void*>("test", () => {
    Thread.usleep(5 * SEC);
    promise.set_value(null);
    return null;
});
promise.future.wait();
assert(promise.future.ready);

Пример 3. wait_until()

const ulong SEC = 1000000;

Promise<void*> promise = new Promise<void*>();
new Thread<void*>("test", () => {
    Thread.usleep(5 * SEC);
    promise.set_value(null);
    return null;
});
bool result = promise.future.wait_until( get_monotonic_time() + 1*SEC );
assert( !promise.future.ready );
assert( !result );

value

Получает значение Future

Если значение не ещё готово, "получающая" значение сторона будет находиться в заблокированном состоянии, пока значение не будет вычислено. Если Future завершается исключением, получение значения вызовет GLib.error.

const ulong SEC = 1000000;

Promise<int> promise = new Promise<int>();
new Thread<void*>("test", () => {
    Thread.usleep(5 * SEC);
    promise.set_value(123);
    return null;
});
int val = promise.future.value;
assert(promise.future.ready);
assert(val == 123);

Promise<int> promise = new Promise<int>();
promise.set_exception( new OptionalError.NOT_PRESENT("Oops!") );
int val = promise.future.value;

Вывод: ERROR: Oops!

map()

map - применяет функцию к каждомы елементу коллекции:

Future<int> future = Future.of<string>("123").map<int>(g => int.parse(g));
assert(future.value == 123);

flat_map()

Работает как map, но с одним отличием — можно преобразовать один элемент в ноль, один или множество других.

Для того, чтобы один элемент преобразовать в ноль элементов, нужно вернуть null, либо пустой стрим. Чтобы преобразовать в один элемент, нужно вернуть Future из одного элемента, например, через Future.of(x). Для возвращения нескольких элементов, можно любыми способами создать Future с этими элементами.

Future<int> future = Future.of<string>("123")
                           .flat_map<int>(f => {
                                int val = int.parse(f.value);
                                return Future.of<int>(val);
                           });
assert(future.value == 123);

light_map()

light_map() - это облегченная версия map(). Данная функция может быть вычислена повторно в любое время.

Пример 4. map () против light_map ()

Future<int> future = Future.of<string>("123")
                           .map<int>(g => {
                                print("map()\n");
                                return int.parse(g);
                           });
assert(future.value == 123);
assert(future.value == 123);

future = Future.of<string>("123")
               .light_map<int>(g => {
                    print("light_map()\n");
                    return int.parse(g);
               });
assert(future.value == 123);
assert(future.value == 123);

Вывод:

map()
light_map()
light_map()

map_err()

Future<int> future = Future.of<int>(123)
                           .map_err(err => {
                               return new OptionalError.NOT_PRESENT("Hello, " + err.message);
                           });
assert(future.value == 123);

Future<int> future = Future.err<int>( new OptionalError.NOT_PRESENT("Oops!") )
                           .map_err(err => {
                               return new OptionalError.NOT_PRESENT("Hello, " + err.message);
                           });
assert(future.exception != null);
assert(future.exception.message == "Hello, Oops!");

then()

Функция then() отложено запускает переданную в качестве аргумента функцию (принимающую future) - после вычисления текущего future, результатом чего может является значение или исключение.

Примечание автора перевода: Если по человечески -- С помощью функции then() можно создать цепочку ленивых вычислений, то есть поставить их в очередь, получая гарантию что следующие будут вычислены только после предыдущих.

Promise<void*> promise = new Promise<void*>();
promise.set_value(null);
promise.future.then(future => {
    if (future.exception == null) {
        print("then() with a value\n");
    } else {
        print("then() with an exception\n");
    }
});

Вывод:

then() with a value

exception .vala

Promise<void*> promise = new Promise<void*>();
promise.set_exception( new OptionalError.NOT_PRESENT("Oops!") );
promise.future.then(future => {
    if (future.exception == null) {
        print("then() with a value\n");
    } else {
        print("then() with an exception\n");
    }
});

Вывод:

then() with an exception

and_then()

Функция and_then() отложено запускает переданную в качестве аргумента функцию (принимающую future) - после вычисления текущего future, результатом чего может является значение, а не исключение .

value.vala

Promise<void*> promise = new Promise<void*>();
promise.set_value(null);
promise.future.and_then(g => {
    print("and_then()\n");
});

output:

and_then()

exception.vala

Promise<void*> promise = new Promise<void*>();
promise.set_exception( new OptionalError.NOT_PRESENT("Oops!") );
promise.future.and_then(g => {
    print("and_then()\n");
});

output: (none)

zip()

zip() объединяет значения двух фьючерсов в один.

Простой пример: zip [1,2,3] [3,2,1] - соеденяет списки [(1,3),(2,2),(3,1)]

Future<string> future = Future.of<string>("100");
Future<string> future2 = Future.of<string>("23");
Future<int> result = future.zip<string,int>((a, b) => {
                         return int.parse(a) + int.parse(b);
                     }, future2);
print("%d\n", result.value);

output: 123

transform()

transform() создает новый future применяя переданную в качестве аргумента функцию к этому future, после того как этот future будет вычислен.

Все остальные mapping функции — map(), flat_map(), then(), etc. — кроме light_map() реализованы с помощью transform().

Future<int> future = Future.of<string>("123")
                           .transform<int>(f => {
                               if (f.exception != null) {
                                   // ...
                               }
                               int val = f.value;
                               return Future.of<int>(val);
                           });

print("%d\n", future.value);

Вывод: 123

Цепочка фьючерсов

Вы можете поместить mapping методы в цепочку.

Future.of<int>(100)
      .map<int>(g => g + 23)
      .map<string>(g => @"Hello $g!")
      .and_then(g => print("%s\n", g));

вывод: Hello 123!

Future.of<int>(100)
      .and_then(g => print("Succeeded 0\n"))
      .map<int>(g => {
          throw new OptionalError.NOT_PRESENT("Oops!");
      })
      .and_then(g => print("Succeeded 1\n"))
      .then(f => {
          if (f.exception != null) print("Failed\n");
      });

Вывод:

Succeeded 0
Failed

Зачем использовать Gpseq.Future, вместо Gee.Future?

Libgee уже предоставляет фьючерсы и промисы. Однако с некоторыми функциями есть проблемы, а некоторых не хватает.

• Gee версия переопределяет исключения с помощью FutureError.EXCEPTION в map функциях — map(), flat_map(), и zip(). FutureError.EXCEPTION вместо реального исключения не несет никакой информации и нежелателен. libgee#23
• Gee.Future не поддерживает цепочки вычислений фьючерсов. map() on an already completed future doesn’t return and blocks the program. libgee#31
• Exceptions can’t be thrown in the map functions, and there is no way to map exceptions. libgee#22 libgee#24

Code refactoring may be required to solve the problems, and it will break backward compatibility of libgee. Instead of waiting for the problems to be fixed, I chose to implement Gpseq’s own Future.

Compare Gpseq.Future with Gee.Future to see the difference.

Seq

Seq это последовательность элементов, поддерживающая паралельные и последовательные операции. Эквивалент Java’s streams.

Seq pipelines

Seq поддерживает различные методы обработки данных. В Seq есть 2 типа операций: промежуточных и терминальных. Seq последовательность может состоять из нуля и более промежуточных операций и терминальной операции.

Последовательность вызовов (ориг. A seq pipeline )

Промежуточные Операции

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

Промежуточные операции также делятся на те у которых есть состояния и те у которых их нет(stateless и stateful ). Операции без сохранения состояния не сохраняют состояние предыдущего элемента при обработке нового элемента. Поэтому каждый элемент может обрабатываться независимо от других элементов. В операциях с отслеживанием состояния(stateful) учитываются состояния предыдущих операций над элементами при обработке новых элементов, из-за чего обработка может быть выполнена только последовотельн(Например вычисление чисел Фибоначчи).

Следовательно, конвейеры seq, содержащие промежуточные stateful операции, могут потребовать нескольких проходов или буферизации важных промежуточных вычислений. Напротив, конвейеры seq, содержащие только stateless операции, могут обрабатываться за один проход.

Stateless операции:

filter, flat_map, map, parallel, sequential, etc.

Stateful операции:

chop, chop_ordered, distinct, limit, limit_ordered, order_by, skip, skip_ordered, etc.

Терминальные операции

Терминальные операции могут изменять источник для получения результата или побочного эффекта. После выполнения терминальной операции seq последовательность закрывается и больше не может использоваться.

Терминальные операции:

foreach, all_match, any_match, collect, collect_ordered, count, find_any, find_first, fold, group_by, iterator, max, min, none_match, partition, reduce, spliterator, etc.

Укороченные (Short-circuiting) операции

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

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

Short-circuiting операции:

all_match, any_match, chop, chop_ordered, find_any, find_first, limit, limit_ordered, none_match, etc.

Parallelism

При запуске терминальной операции конвейер seq выполняется последовательно или параллельно в зависимости от режима seq, в котором он вызывается. Последовательный режим является режимом по умолчанию. Он может быть изменен с помощью промежуточных операций sequential() или parallel().

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

Возвращаемый результат любой терминальной операции уже является вычисленным, поэтому вам не нужно ждать future для получения результата

Примеры методов

Main method, using Gpseq;, and using Gee; omitted.

foreach()

string[] array = {"dog", "cat", "pig"};
Seq.of_array<string>(array).foreach(g => print("%s\n", g));

output:

dog
cat
pig

filter()

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

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Seq.of_array<string>(array)
    .filter(g => g.length == 3)
    .foreach(g => print("%s\n", g));

output:

dog
cat
pig

map()

Применяет функцию к каждому элементу и затем возвращает стрим, в котором элементами будут результаты функции. map можно применять для изменения типа элементов.

string[] array = {"dog", "cat", "pig"};
Seq.of_array<string>(array)
    .map<string>(g => g.up())
    .foreach(g => print("%s\n", g));

output:

DOG
CAT
PIG

flat_map()

Работает как map, но с одним отличием — можно преобразовать один элемент в ноль, один или множество других.

Для того, чтобы один элемент преобразовать в ноль элементов, нужно вернуть null, либо пустой стрим. Чтобы преобразовать в один элемент, нужно вернуть стрим из одного элемента, например, через Seq.of(x). Для возвращения нескольких элементов, можно любыми способами создать стрим с этими элементами.

string[] array = {"dog", "cat", "pig"};
Seq.of_array<string>(array)
    .flat_map<string>(g => {
        var list = new ArrayList<string>();
        list.add(g);
        list.add(g + "2");
        return list.iterator();
    })
    .foreach(g => print("%s\n", g));

output:

dog
dog2
cat
cat2
pig
pig2

skip(), limit(), and chop()

Example 1. skip()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Seq.of_array<string>(array)
    .skip(3) // It is equivalent to chop(3)
    .foreach(g => print("%s\n", g));

output:

boar
bear

Example 2. limit()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Seq.of_array<string>(array)
    .limit(3) // It is equivalent to chop(0, 3)
    .foreach(g => print("%s\n", g));

output:

dog
cat
pig

Example 3. chop()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Seq.of_array<string>(array)
    .chop(2, 1)
    .foreach(g => print("%s\n", g));

output:

pig

skip_ordered(), limit_ordered(), and chop_ordered()

They are the ordered version of skip/limit/chop. They always respect encounter order regardless of sequential/parallel execution. They are, however, quite expensive on parallel execution.

order_by()

string[] array = {"dog", "cat", "pig"};
Seq.of_array<string>(array)
    .order_by()
    .foreach(g => print("%s\n", g));

output:

cat
dog
pig

distinct()

string[] array = {"dog", "dog", "cat", "cat", "pig"};
Seq.of_array<string>(array)
    .distinct()
    .foreach(g => print("%s\n", g));

output:

dog
cat
pig

all_match() and any_match()

You should read the documentation of Future first.Example 4. all_match()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
bool result = (!) Seq.of_array<string>(array)
                     .all_match(g => g.length == 3).value;
print("%s\n", result.to_string());

output:

false

Example 5. any_match()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Future<bool?> future = Seq.of_array<string>(array)
                          .any_match(g => g.length == 3);
bool result = (!) future.value;
print("%s\n", result.to_string());

output:

true

	(!) is the explicit non-null cast operator in Vala. (Currently not necessary, except when using --enable-experimental-non-null option)

count()

string[] array = {"dog", "dog", "cat", "cat", "pig"};
int64 count = (!) Seq.of_array<string>(array).distinct().count().value;
print("%" + int64.FORMAT + "\n", count);

output:

3

find_any() and find_first()

Example 6. find_any()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Optional<string> result = Seq.of_array<string>(array)
                             .find_any(g => g.length == 4).value;
print("%s\n", result.value);

output:

boar

Example 7. find_first()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Optional<string> result = Seq.of_array<string>(array)
                             .find_first(g => g.length == 4).value;
print("%s\n", result.value);

output:

boar

Example 8. find_any() vs find_first() on parallel execution

string[] array = {"dog", "cat", "pig", "boar", "bear"};

Optional<string> any = Seq.of_array<string>(array)
                          .parallel()
                          .find_any(g => g.length == 3).value;
assert(any.value == "dog" || any.value == "cat" || any.value == "pig");

Optional<string> first = Seq.of_array<string>(array)
                            .parallel()
                            .find_any(g => g.length == 3).value;
assert(first.value == "dog");

max() and min()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Optional<int> max = Seq.of_array<string>(array)
                          .map<int>(g => g.length)
                          .max();
Optional<int> min = Seq.of_array<string>(array)
                          .map<int>(g => g.length)
                          .min();
print("max: %d\n", max.value);
print("min: %d\n", min.value);

output:

max: 4
min: 3

Seq.iterate()

public static Seq<G> iterate<G> (owned G seed,
                                 owned Gpseq.Predicate<G> pred,
                                 owned Gpseq.MapFunc<G,G> next,
                                 TaskEnv? env = null)

Each element is generated by the given next function applied to the previous element, and the initial element is the seed.Example 9. For loop

Seq.iterate<int>(0, i => i < 5, i => ++i)
    .foreach(g => print("%s ", g));

output:

0 1 2 3 4

reduce() and fold()

Example 10. reduce()

Optional<int> result = Seq.iterate<int>(0, i => i < 5, i => ++i)
                          .reduce((a, b) => a + b);
print("%d\n", result.value);

output:

10

Example 11. fold()

Optional<int> result = Seq.iterate<int>(0, i => i < 5, i => ++i)
                          .fold<int>((a, b) => a + b, (a, b) => a + b, 0);
print("%d\n", result.value);

output:

10

group_by()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Map<int,Gee.List<string>> result = Seq.of_array<string>(array)
                                      .group_by<int>(g => g.length)
                                      .value;
print("3-length: %d\n", result[3].size);
print("4-length: %d\n", result[4].size);

output:

3-length: 3
4-length: 2

partition()

string[] array = {"dog", "cat", "pig", "boar", "bear"};
Map<bool,Gee.List<string>> result = Seq.of_array<string>(array)
                                      .partition(g => g.length == 3)
                                      .value;
print("true: %d\n", result[true].size);
print("false: %d\n", result[false].size);

output:

true: 3
false: 2

iterator()

string[] array = {"dog", "dog", "cat", "cat", "pig"};
Iterator<string> iter = Seq.of_array<string>(array).distinct().iterator();
iter.foreach(g => {
    print("%s\n", g);
    return true;
});

output:

dog
cat
pig

spliterator()

Spliterator is an object for traversing and partitioning elements of a data source, used in Seq.

You should read the documentation of Spliterator first.

string[] array = {"dog", "dog", "cat", "cat", "pig"};
Spliterator<string> spliter = Seq.of_array<string>(array)
                                 .distinct()
                                 .spliterator();
spliter.each(g => print("%s\n", g));

output:

dog
cat
pig

collect() and collect_ordered()

See the section Collectors.

Both perform a mutable reduction operation on the elements of a seq.collect()

collect() performs a concurrent reduction if the seq is in parallel mode and the collector is CONCURRENT.collect_ordered()

collect_ordered() preserves encounter order even though the seq is in parallel mode, if the collector is not CONCURRENT or not UNORDERED. If, however, the seq is in parallel mode and the collector is CONCURRENT and UNORDERED, it performs an unordered concurrent reduction.

Error handling

Most seq operations take delegates which can throw errors. If an error occurs in the delegate, the returned future is completed with the error.Example 12. Error handling

try {
    Seq.iterate<int>(0, i => i < 100, i => ++i)
        .parallel()
        .foreach(i => {
            if (i == 42) {
                throw new OptionalError.NOT_PRESENT("%d? Oops!", i);
            }
        }).wait(); // Gpseq.Future.wait() throws Error
} catch (Error err) {
    error(err.message);
}

output:

ERROR: 42? Oops!

Collectors

Collector is an object for mutable reduction operation that accumulates input elements into a mutable accumulator, and transforms it into a final result. It is used with collect() or collect_ordered().

Gpseq.Collectors namespace provides various useful collector implementations. Here are some examples:

Main method, using Gpseq;, using Gpseq.Collectors;, and using Gee; omitted.Example 13. Collectors.to_list()

string[] array = {"dog", "dog", "cat", "cat", "pig"};
Gee.List<string> list = Seq.of_array<string>(array)
                                 .distinct()
                                 .collect( to_list<int>() )
                                 .value;
Seq.of_list<string>(list).foreach(g => print("%s\n", g));

output:

dog
cat
pig

Example 14. Collectors.to_generic_array()

string[] array = {"dog", "dog", "cat", "cat", "pig"};
GenericArray<string> array = Seq.of_array<string>(array)
                                 .distinct()
                                 .collect( to_generic_array<int>() )
                                 .value;
Seq.of_generic_array<string>(array).foreach(g => print("%s\n", g));

output:

dog
cat
pig

	GenericArray (Vala) == GPtrArray (C-lang)

Example 15. Collectors.sum_int()

string[] array = {"dog", "cat", "pig"};
int sum = Seq.of_array<string>(array)
             .collect( sum_int<string>(g => g.length) )
             .value;
print("%d\n", sum);

output:

9

	The arithmetic wraps around on overflow; e.g. the sum of int.MAX and 1 ⇒ int.MIN

Example 16. Collectors.count()

string[] array = {"dog", "cat", "pig"};
int64 result = (!) Seq.of_array<string>(array)
                     .collect( count<string>() )
                     .value;
print("%d\n", result);

output:

3

Example 17. Collectors.join()with_default_delimiter.vala

string[] array = {"dog", "cat", "pig"};
string result = Seq.of_array<string>(array)
                   .collect( join() )
                   .value;
print("%s\n", result);

output:

dogcatpig

delimiter_specified.vala

string[] array = {"dog", "cat", "pig"};
string result = Seq.of_array<string>(array)
                   .collect( join(",") )
                   .value;
print("%s\n", result);

output:

dog,cat,pig

Example 18. Collectors.partition_with()

public Collector<Map<bool,V>,Object,G> partition_with<V,G> (
        owned Predicate<G> pred,
        Collector<V,Object,G> downstream)

partition_with.vala

string[] array = {"dog", "cat", "pig", "boar", "bear"};
// Map<bool,Gee.List<string>>
var map = Seq.of_array<string>(array)
              .collect( partition_with<Gee.List<string>,string>(
                    g => g.length,
                    to_list<string>()) )
              .value;
print("true: %d\n", map[true].size);
print("false: %d\n", map[false].size);

output:

true: 3
false: 2

	Seq.partition(pred) is equivalent to collect( partition<G>(pred) ). And partition<G>(pred) is equivalent to partition_with<Gee.List<G>,G>(pred, to_list<G>()).

Example 19. Collectors.wrap()

wrap() takes a collector and returns a collector will produce a Wrapper object containing the result of the given collector.

string[] array = {"dog", "cat", "pig"};
Wrapper<int64?> result = Seq.of_array<string>(array)
                     .collect( wrap<int64?,string>(count<string>()) )
                     .value;
print("%d\n", result.value);

output:

3

Example 20. Collectors.tee()

public Collector<A,Object,G> tee<A,G> (
        owned Collector<Object,Object,G>[] downstreams,
        owned TeeMergeFunc<A> merger)

public delegate A TeeMergeFunc<A> (Object[] results) throws Error

tee.vala

var seq = Seq.iterate<int>(0, i => i < 5, i => ++i);
var collector = tee<Gee.List<int64?>,string>(
    {
        wrap<int64?,G>( sum_int64<string>(g => g) ),
        wrap<int64?,G>( count<string>() )
    },
    (results) => {
        var list = new ArrayList<int64?>();
        list.add( ((Wrapper<int64?>)results[0]).value );
        list.add( ((Wrapper<int64?>)results[1]).value );
        return list;
    }
);
Gee.List<int64?> result = seq.collect(collector).value;
print("sum: %d\n", result[0]);
print("count: %d\n", result[1]);

output:

sum: 10
count: 5

Custom collectors

A collector implements four methods: create_accumulator, accumulate, combine, and finish.

1. create_accumulator() - Creates a new accumulator, such as Gee.Collection.

2. accumulate() - Incorporates a new element into a accumulator.

3. combine() - Combines two accumulators into one.

4. finish() - Transforms the accumulator into a final result.

The methods must satisfy an identity and an associativity constraints. The identity constraint means that combining any accumulator with an empty accumulator must produce an equivalent result. i.e. an accumulator a must be equivalent to collector.accumulate(a, collector.create_accumulator()).

The associativity constraint means that splitting the computation must produce an equivalent result. i.e.:Associativity constraint

// the two computations below must be equivalent.
// collector: a collector
// g0, g1: elements

A a0 = collector.create_accumulator();
collector.accumulate(g0, a0);
collector.accumulate(g1, a0);
A r0 = collector.finish(a0);

A a1 = collector.create_accumulator();
collector.accumulate(g0, a1);
A a2 = collector.create_accumulator();
collector.accumulate(g1, a2);
A r1 = collector.finish( collector.combine(a1, a2) );

Collectors also have a property, features. it provides hints that can be used to optimize the operation.

Here is the collector implementation of Collectors.to_collection<G>():CollectionCollector.vala

using Gee;

private class Gpseq.Collectors.CollectionCollector<G> : Object, Collector<Collection<G>,Collection<G>,G> {
    private Supplier<Collection<G>> _factory;
    private CollectorFeatures _features;

    public CollectionCollector (Supplier<Collection<G>> factory, CollectorFeatures features) {
        _factory = factory;
        _features = features;
    }

    public CollectorFeatures features {
        get {
            return _features;
        }
    }

    public Collection<G> create_accumulator () throws Error {
        return _factory.supply();
    }

    public void accumulate (G g, Collection<G> a) throws Error {
        a.add(g);
    }

    public Collection<G> combine (Collection<G> a, Collection<G> b) throws Error {
        a.add_all(b);
        return a;
    }

    public Collection<G> finish (Collection<G> a) throws Error {
        return a;
    }
}

Worker pools

TODO

References

• More detailed description about seq pipelines can be found on the Seq documentations: Valadoc, GtkDoc