Пять лет назад мне довелось проектировать одну программу, обрабатывающую текст с управляющими командами. Количество этапов обработки было весьма существенным — 6-7 обработчиков последовательно могли пропускать через себя довольно большие объёмы данных, как бывает иногда в конвейерах Unix. Чтобы аккуратно выполнить поставленную задачу я разработал общий метод, который может пригодиться и в других местах. Идея, лежащая в основе этой статьи, действительно очень напоминает конвейеры Unix, но имеется несколько существенных отличий:

конвейер Unix работает асинхронно, в разных процессах, в то время, как здесь требуется реализовать обработку в рамках одной программы, и распараллеливание может быть нежелательно;
возможна передача любых данных, не обязательно текстовых, но которые можно охарактеризовать термином «линейные».

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

чтение файла по байтам,
декодирование в UTF-8,
препроцессор,
лексический анализ,
синтаксический анализ.

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

Преобразователи выстраиваются в виде “конвейера”, каждый следующий получает данные от предыдущего. Преобразователь также может менять тип данных, например, у лексера буквы на входе, на выходе — лексемы.

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

Что получается, если проектировать обработку линейных данных, особо не заморачиваясь

Давайте рассмотрим возможную реализацию программного интерфейса преобразователя. В самом простом варианте возможны три типа преобразователей:

class Source {
	…
public:
	T getData();
	…
};

class Sink {
	…
public:
	void putData(T);
	…
};

class Universal {
	…
public:
	void process();
};

Рис 1. Схема простой реализации конвейера.

Вот тривиальный пример такого конвейера, «сумматор» (он читает числа из std::cin, складывает их по парам, печатает в std::cout, это рабочая программа на C++):

#include <iostream>

//считывает число из std::cin
class IntSource {
public:
    int getData() {
        int next;
        std::cin>>next;
        return next;
    }

    bool good() const {
        return std::cin.good();
    }
};

//печатает число в std::cout
class IntSink {
public:
    void putData(int data) {
        std::cout<<data<<std::endl;
    }
};

//читает пару чисел из IntSource и выдаёт их сумму в IntSink
class IntUniversal {
    IntSource source;
    IntSink   sink;
public:
    void process() {
        int i1 = source.getData();
        int i2 = source.getData();

        if( good() ) sink.putData(i1+i2);
    }

    bool good() const {
        return source.good();
    }
};

int main() {
	IntUniversal belt;

	while( belt.good() ) belt.process();
}

Преобразователь Source (источник) предоставляет интерфейс, позволяющий следующему преобразователю в конвейере получить порцию данных из него. Преобразователь Sink (поглотитель) напротив, может получать сигнал от предыдущего преобразователя о том, что данные готовы. Source выдаёт данные “по требованию”, но сам заботиться об их получении, в то время, как Sink может выдать данные следующему элементу, но получает их только тогда, когда заблагорассудится предыдущему. Только универсальный преобразователь обладает полной свободой: он может получить данные из предыдущего и передать следующему когда ему вздумается. Заметим несколько простых, вынужденных правил такого упрощённого “конвейера”, основанных на ограничениях возможностей преобразователя:

первым элементом в конвейере является Source;
последним элементом служит Sink;
преобразователь Sink может непосредственно предшествовать только другому преобразователю Sink, преобразователь Source может непосредственно следовать только за другим преобразователем Source;
преобразователь Universal должен следовать за Source и предшествовать Sink.

Из этих правил следует, что единственной возможной реализаций конвейера является такая: Source₁, …, Source_n, Universal, Sink₁, …, Sink_m. Стек вызовов при работе такого конвейера может выглядеть следующим образом:

Source_k::getData
...
Source_n::getData
Universal::process,

либо

Sink_l::putData
...
Sink₁::putData
Universal::process.

Улучшаем конвейер

Использование Source и Sink накладывает ограничения на реализацию процедур, а также на их эффективность. Программисту предоставлялось бы больше удобства, если бы конвейер выглядел так: Universal, …, Universal. Воплотить это желание в жизнь можно добавив то, чего действительно не хватает конвейеру: “ленту”. Лентой у нас будет служить область памяти, хранящая данные, находящиеся “между” преобразователями. Как только преобразователь производит данные, они помещаются на ленту, откуда их может считать следующий преобразователь. Конвейер теперь усложнился и не может управляться сам, как было раньше, поэтому требуется специальный “менеджер конвейера”, следящий за лентой и за преобразователями. Преобразователи теперь будем строить в виде наследников от базовых классов с интерфейсом. Вот их упрощённый вид:


class UniversalBase {
public:
    virtual void process()=0;
};

template<class S> class UniversalSource;
template<class T> class UniversalSink;

//универсальный преобразователь для помещения в начало конвейера
template <class S>
class UniversalSource : public virtual UniversalBase {
    UniversalSink<S>* next;
protected:
    void putOnBelt(const S&);
};

//универсальный преобразователь для помещения в конец конвейера
template <class T>
class UniversalSink : public virtual UniversalBase  {
    UniversalSource<T>* prev;
protected:
    T getFromBelt();
};

//универсальный преобразователь, считывающий данные типа T и выдающий данные типа S
template <class T, class S>
class Universal : public UniversalSink<T>, public UniversalSource<S> { };

Рис 2. Схема улучшенной реализации конвейера.

Функция process реализуется в каждом конкретном преобразователе по-своему, и выполняет суть его назначения. Её задача — произвести некоторое количество данных и передать их менеджеру конвейера для помещения на ленту. Для этого process вызывает функцию putOnBelt, определённую в базовом классе. Важно понимать, что process каждого преобразователя может вызываться несколько раз, она должна произвести, некоторое разумное количество данных (например, одну единицу) и завершиться. Как только реализации process требуются входные данные, она обращается к менеджеру, вызывая getFromBelt.

Вот пример того же сумматора, но реализованного с использованием новой концепции конвейера. Теперь это уже не полноценная программа, ей не хватает реализации менеджера конвейера.

#include <iostream>
#inlcude <belt.h>

//считывает число из std::cin
class IntSource : public Belt::UniversalSource<int> {
public:
    void process() {
        int data;
        if( std::cin>>data ) putOnBelt(data);
    }
};

//печатает число в std::cout
class IntSink : public Belt::UniversalSink<int> {
public:
    void process() {
        if(!hasData() ) return;
        std::cout<<getFromBelt()<<std::endl;
    }
};

//читает пару чисел с ленты и кладёт их сумму на ленту
class IntUniversal : public Belt::Universal<int,int> {
public:
    void process() {
        int i1 = getFromBelt();
        int i2 = getFromBelt();

        if(!good() ) return;

        putOnBelt(i1+i2);
    }
};

int main() {
	IntSource source;
	IntUniversal universal;
	IntSink sink;

	(source >> universal >> sink).process();
}

Здесь использованы функции и классы, о которых ранее не говорилось:

bool UniversalSink::hasData(void);
bool UniversalSink::good(void); 

template<class T,class S> class UnboundedBelt : public Universal<T,S> {...};
template<class T> class RightBoundedBelt : public UniversalSink<T> {...};
template<class S> LeftBoundedBelt : public UniversalSource<S> {...};
class Belt : public UniversalBase {...};

template<class T, class R, class S> UnboundedBelt<T,S> operator >> (Universal<T,R>&,Universal<R,S>&); 
template<class R, class S> LeftBoundedBelt<S> operator >> (UniversalSource<R>&,Universal<R,S>&); 
template<class T, class R> RightBoundedBelt<T> operator >> (Universal<T,R>&,UniversalSink<R>&); 
template<class R> Belt operator >> (UniversalSource<R>&,UniversalSink<R>&);

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

Подходы к реализации менеджера конвейера

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

Функция getFromBelt считывает данные с ленты, если они имеются на ней, а в противном случае, она запускает process у предыдущего преобразователя до тех пор, пока он не выдаст какую-нибудь порцию данных на ленту. putOnBelt просто-напросто помещает данные на ленту. Она может вызывать process следующего преобразователя, чтобы он сразу же обработал их, но это не обязательно, и создаёт трудности, о которых — чуть позже.

Таким образом, стек вызовов в простом случае принимает вид:

...
менеджер конвейера
UniversalSink::getFromBelt()
Преобразователь_n₂::process()
менеджер конвейера
UniversalSink::getFromBelt()
Преобразователь_n₁::process()
менеджер конвейера

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

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

Выполнение этой аксиомы накладывает важное (и единственное) ограничение на реализацию (речь идёт только об однопоточном случае):

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

Действительно, давайте посмотрим, что получится, если это требование не выполнено, и управление передано более правому преобразователю R, в то время, как более левый преобразователь L выполняется. Если R потребуются данные, а на ленте они отсутствуют, то менеджер конвейера вынужден будет вызывать левые преобразователи, что может привести к нарушению аксиомы (А).

Вот две возможные реализации менеджера:

Рекурсивный. Вызывает process непосредственно только у последнего преобразователя, остальные запускаются рекурсивно при необходимости.
Последовательный. Вызывает process по очереди (слева направо), “нарабатывая” данные на ленту. Как только решит, что данных достаточно, переходит на один преобразователь правее. Для этого варианта желательно наличие оценок о том, сколько данных «потребляет» тот или иной преобразователь за одну итерацию process.

Возможные плюшки

В статье приведена упрощённая схема реализации конвейера. Его доработки могут реализовать:

операцию “заглядывания вперёд”: преобразователь считывает данные, но фактически они не удаляются с ленты, и возвращаются по его требованию;
“коробки конфет”: выдача данных порциями, что позволяет повысить эффективность при работе с большим количеством мелких данных, например, символами. По конвейеру передаётся сразу “коробка”, вместо передачи “по одной конфете”, что позволяет избежать вызовов функций на каждую единицу данных;
“умный” распределитель памяти (аллокатор), который позволит избегать постоянных выделений/удалений динамической памяти, при работе конвейера, и даже может избавить от лишних операций копирования;
выполнение в нескольких потоках. В случае конвейера это всегда возможно, и эффективно, если каждый из преобразователей однопоточен.

Также, по сравнению с “реализаций в лоб” большим преимуществом является: независимость отдельных частей программы друг от друга (каждый преобразователь знает только о типах его входных и выходных данных, может не знать ничего о преобразователях, к которым он подключен входом и выходом), возможность безболезненно включать новые преобразователи и менять их порядок.

Заключение

Прошу прощения у:

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

Я допускаю, что эти концепции, а может быть, гораздо более удобные, реализованы в какой-то популярной библиотеке, типа Boost (если да — будет интересно посмотреть). Но для меня содержание этой статьи представляется красивым паттерном, не только и не столько практического значения, с которым хотелось поделиться.

Ленточный конвейер для линейных данных