Архитектура программ на основе автоматов,
управляемых событиями

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

Собственно, $SUBJ

Как говорится, одна картинка стоит тысячи слов (которые пойдут сразу после картинки):

То, что на картинке обозначено как "Event Capture", в коде реализовано проще простого:

static void ecap__wait(struct message_buffer_api *mb)
{
        struct message msg = {0};
           int n, k;

        n = epoll_wait(epfd, events, MAX_EVENTS, -1);
        if (-1 == n) {
                if (EINTR != errno)
                        printf("OPS: %s - epoll_wait(): %s\n",
                        __func__, strerror(errno));
                return;
        }

        for (k = 0; k < n; k++) {
                msg.ptr = events[k].data.ptr;
                msg.u64 = events[k].events;
                mb->put(&msg);
        }
}

Усё. То есть спим в epoll_wait() до тех пор, пока не произойдет что-то, интересующее программу, потом преобразуем полученную информацию в сообщения и помещаем их все в буфер (обозначенный на картинке как "Message Queue"). Технически эта очередь выполнена не в виде, так сказать, классической очереди, а в виде (расширяемого) кольцевого... эээ... буфера.

Диспетчер сообщений тоже весьма незатейлив; его "центровая" функция такова:

int __edsm_dispatch_message(struct message *msg)
{
        struct edsm *src = msg->src;
        struct edsm *dst = msg->dst;
           u32 code = msg->u64;
           u32 seqn = code & 0x0000FFFF;
        struct state *state = dst->state;
        struct transition *trans = NULL;
          char etc = event_type_chars[code >> 16];

        trans = __edsm_get_transition(state, code);
        if (unlikely(NULL == trans)) {
                dbg_msg_DROPPED;
                return 0;
        } else {
                dbg_msg_ACCEPTED;
        }

        if (trans->action) {
                int ret;
                ret = trans->action(src, dst, msg->ptr);
                if (unlikely(ret))
                        state->leave(src, dst, msg->ptr);
                return ret;
        }

        state->leave(src, dst, msg->ptr);
        dst->state = state = trans->next_state;
        dbg_msg_TRANSITION;
        return state->enter(src, dst, msg->ptr);
}

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

Отличия от предыдущей версии

• Наличие очереди сообщений. Как видно из картинки, поставщиками сообщений являются "event capture" и сами автоматы. В предыдущей реинкарнации для доставки "внутренних" сообщений использовался "почтовый ящик", который был приколочен к каждому зкземпляру автомата и который был реализован в виде self-pipe, что понапрасну расходовало драгоценные файловые дескрипторы (+лишние системные вызовы). Теперь же работа с внутренними сообщениями происходит без дёрганья ОС.
• Автоматическое создание каналов типа "i/o" и "fsys"; ранее автоматически создавались только каналы типа "signal" и "timer"; ну, и ещё "почтовый ящик", но он благополучно отпал. Это в весьма значительной степени упрощает код прикладной программы.
• В файлах описания автоматов для реакций без перехода ("инструкция" '@') теперь не нужно указывать имя функции, оно формируется из названия автомата, названия состояния и кода события. Так, для примера ниже в программе должны содержаться функции с именами test_work_S0, test_work_S1, test_work_T0 и т.д. Так проще ориентироваться с собственном коде :).
• Прикладная программа напрочь "изолирована" от "event engine", доступ к его API - только через прослойку EDSM. Раньше между событийным движком и диспетчером сообщений была взаимосвязь весьма стрёмного вида, теперь оно урефакторилось.
• Чуть не забыл. Мониторинга событий в файловой системе (на основе inotify) в предыдущем варианте не было вообще.

Пример #1
(сигналы, таймеры, события файловой системы)

Описание единственного в этом примере автомата вот (test.smd):

S2
S15

T2000
T60000

$work

@work S0
@work S1
@work T0
@work T1
@work F3
@work F6
@work F7

Из этого описания мы (и соответствующий компонент движка) видим, что автомат "ловит" два сигнала (SIGINT и SIGTERM), имеет 2 таймера на 2 и 60 секунд (первый будет использоваться в периодическом режиме, второй - в однократном); автомат может находиться только в одном состоянии, которое называется 'work'. Помимо сигналов и срабатываний таймеров, автомат будет реагировать на 3 вида событий в файловой системе (в заданном каталоге) - "файл перемещен из X" (F6), "файл перемещён в X" (F7) и "закрыт файл, открытый для записи" (F3).

Примечание: если это описание делать в коде (как это делается тут), то получается гораздо более громоздко.

Основной файл программы выглядит весьма лаконично, что есть good:

#include <stdio.h>
#include "edsm.h"

struct edsm_template mt[] = {
        [0] = {
                .source = "test.smd",
                .desc = NULL,
                .state_set = NULL,
                .expected_instances = 1,
                .instance_seqn = 1,
        },
};

struct edsm_api *engine;
struct edsm *m;

int main(void)
{
        int stop;

        printf("Test: signals, timers,
        	file system monitoring\n");

        engine = edsm_init(".", mt, 1);
        m = engine->new(&mt[0], NULL);
        engine->run(m, CALL_ENTER_FUNCTION);

        for (;;) {
                stop = engine->exec();
                if (stop)
                        break;
                engine->wait();
        }

        return 0;
}

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

Пример #2
(echo server)

Рассмотрим теперь более содержательный пример, эхо-сервер. В этом примере будет продемонстрировано, как можно строить иерархии автоматов, а также как использовать пулы автоматов. Сей сервер будет строиться на основе 5-ти типов автоматов:

• RX - автомат чтения данных, экземпляры хранятся в пуле;
• TX - автомат записи данных, экземпляры хранятся в пуле;
• WORKER - автомат, в который заложена схема взаимодействия с клиентом, в данном случае это схема "запрос-ответ"; экземпляры также хранятся в пуле;
• LISTENER - автомат, принимающий соединения от клиентов; существует в единственном экземпляре;
• MANAGER - посредник/прослойка между автоматом LISTENER и автоматами WORKER; понадобилась для того, чтобы отвязать LISTENER от особенностей построения программы, например, он ничего не знает про упомянутые пулы, для него существует только MANAGER.

Автоматы RX и TX являются универсальными - я их (или подобные им) использую для работы практически со "всем" - сокеты, COM-порты и т.п. Они учитывают всякие особенности в виде partial read/write, отслеживают любые обломы вроде POLLHUP, автомат RX всегда принимает данные с некоторым таймаутом, на основе автомата TX можно делать асинхронный connect() и всякое такое.

В общем виде работу сервера можно описать следующим образом. Автомат LISTENER ожидает клиента, когда он появляется, принимает соединение и передаёт нужную информацию автомату MANAGER. Этот автомат берёт из пула автомат WORKER и передаёт ему клиента на обслуживание. Поскольку эхо-сервер (впрочем, как и куча других) работает по схеме "запрос-ответ", автомату WORKER нужно сначала принять запрос, потом отправить ответ. Для приёма запроса WORKER берёт из пула автомат RX и просит его, мол давай, принимай данные-то. После каждой принятой порции данных автомат RX сообщает "заказчику", что данные есть, заказчик проверяет, пришел ли запрос целиком и, если нет, просит принимать дальше. Как только запрос пришёл целиком (в нашем случае это значит, что приехал символ '\n'), WORKER "отпускает" автомат чтения данных и берёт из соответствующего пула автомат записи данных, с помощью которого и отправляет ответ (в нашем случае это просто "эхо"). По окончании работы с клиентом (клиент сам отсоединился, не прислал вовремя запрос, произошёл какой-то сбой) WORKER сообщает об этом менеджеру, а тот, в свою очередь - автомату LISTENER, который в итоге закрывает соединение со стороны сервера.

Замечание по поводу того, зачем отделять чтение данных (RX) от записи данных (TX), ведь можно эти дела реализовать в едином автомате; более того, эту функциональность можно реализовать вообще в автомате WORKER. Так вот - декомпозиция автоматов уменьшает количество состояний в каждом, упрощая таким образом проектирование. В том, что вместо единого RXTX имеется раздельные RX и TX, есть ещё одна "выгода". Если автоматы хранятся в пулах, то при некоторых условиях для (одновременного) обслуживания N клиентов понадобится N атоматов WORKER, но при этом значительно меньшее количество автоматов RX и TX.

Ну а теперь - жуткие подробности. Для начала приведу схемы всех автоматов (в виде текста, ибо во-первых, влом, а во-вторых, картинки легко рисуются по текстовому описанию). Итак...

Автомат RX

T180000

$init
$idle
$work

+init M0 idle

@idle M1
+idle M0 work

@work D0
@work M1
@work D2
@work T0
+work M0 idle

У этого автомата есть таймер на 3 минуты и находиться он может в трёх состояниях - INIT, IDLE и WORK. Состояние INIT нужно для того, чтобы... собственно, для этого оно и нужно, после чего автомат отправляется в состояние IDLE, на входе в которое помещает себя в свой пул. Откуда его берёт WORKER и пинает сообщением M1, после чего автомат чтения переводит себя в состояние WORK. В этом состоянии он обрабатывает несколько видов сообщений: D0 - это фактически POLLIN, M1 - это сообщение от рабочего автомата "давай ещё считывай", D2 - это POLLERR/POLLHUP, T0 - это сработал таймер, то есть время ожидания данных истекло, ну и по M0 автомат переходит в IDLE, где, как мы помним, он самопомещается в свой пул. После получения данных автомат отсылает сообщение M1 автомату WORKER.

Автомат TX

$init
$idle
$work

+init M0 idle

@idle M1
+idle M0 work

@work D1
@work D2
+work M0 idle

Отличается от своего собрата отсутствием таймера (он ему не нужен) и тем, какие сообщения он обрабатывает в состоянии WORK: D1 - это POLLOUT, D2 - опять же POLLERR/POLLHUP.

Автомат WORKER

$init
$idle
$getp
$ackp

+init M0 idle

@idle M1
+idle M0 getp

@getp M1
@getp M2
+getp M0 ackp
+getp M3 idle

@ackp M1
@ackp M2
+ackp M0 getp
+ackp M3 idle

После инициализации своих внутренних дел автомат переходит в состояние IDLE, где пребывает в пуле до тех, пока его оттуда не попросит MANAGER. В состоянии GETP автомат занимается приёмом запроса, эксплуатируя для этого автомат RX, в состоянии ACKP - отправкой эхо-ответа, эксплуатируя, соответственно, автомат TX.

Автомат MANAGER

S2
S15

$init
$work

+init M0 work

@work M0
@work M1
@work S0
@work S1

Имеет 2 состояния, одно для инициализации, второе - рабочее. В этом втором ловит два сигнала (SIGINT и SIGTERM) и два более содержательных сообщения: M0 - это от автомата приёма соединений по поводу нового клиента, M1 - от рабочего автомата по поводу того, что клиент отпал.

Автомат LISTENER

$init
$work

+init M0 work

@work D0
@work M0
@work M1

Тоже имеет 2 состояния, INIT и WORK. В рабочем состоянии обрабатывает сообщения D0 (от ОС, "можно принимать соединение"), M0 (от менеджера, "проводи клиента") и M1 (опять от менеджера, "мы закругляемся", но, кажись, это не будет работать).

main.c

Опять же вполне лаконичен:

#include <stdio.h>
#include <stdlib.h>

#include "edsm.h"
#include "macro-magic.h"
#include "echo-server.h"
#include "state-machines/state-machines.h"

void prepare_pools(struct echo_server *s)
{
	struct pool *p;
	   int n = MAX_CLIENTS;

	p = new_pool("rx", n, n);
	if (NULL == p) {
		exit(1);
	}
	s->rx_pool = p;

	p = new_pool("tx", n, n);
	if (NULL == p) {
		exit(1);
	}
	s->tx_pool = p;

	p = new_pool("workers", n, n);
	if (NULL == p) {
		exit(1);
	}
	s->worker_pool = p;
}

void prepare_machines(struct edsm_api *e,
		      struct edsm_template *t,
		      struct echo_server *s)
{
	struct pool *p;
	struct edsm *m;
	   int k, n = MAX_CLIENTS, err;

static	struct listener_info li = {0};

	p = s->rx_pool;
	for (k = 0; k < n; k++) {
		m = e->new(&t[TN_RX]);
		err = pool_new_obj(p, m);
		if (err) {
			exit(1);
		}
		m->restroom = p;
		e->run(m, NULL);
	}

	p = s->tx_pool;
	for (k = 0; k < n; k++) {
		m = e->new(&t[TN_TX]);
		err = pool_new_obj(p, m);
		if (err) {
			exit(1);
		}
		m->restroom = p;
		e->run(m, NULL);
	}

	m = s->manager = e->new(&t[TN_MANAGER]);
	e->run(m, s);

	p = s->worker_pool;
	for (k = 0; k < n; k++) {
		m = e->new(&t[TN_WORKER]);
		err = pool_new_obj(p, m);
		if (err) {
			exit(1);
		}
		m->restroom = p;
		e->run(m, s);
	}

	li.port = TCP_PORT;
	li.manager = s->manager;
	m = s->listener = e->new(&t[TN_LISTENER]);
	e->run(m, &li);
}

struct edsm_template mt[] = {

	[TN_RX] = {
		.source = "rx.smd",
		.desc = NULL,
		.state_set = NULL,
		.expected_instances = MAX_CLIENTS,
		.instance_seqn = 0,
		.create_io_channel = 0,
	},

	[TN_TX] = {
		.source = "tx.smd",
		.desc = NULL,
		.state_set = NULL,
		.expected_instances = MAX_CLIENTS,
		.instance_seqn = 0,
		.create_io_channel = 0,
	},

	[TN_WORKER] = {
		.source = "worker.smd",
		.desc = NULL,
		.state_set = NULL,
		.expected_instances = MAX_CLIENTS,
		.instance_seqn = 0,
		.create_io_channel = 1,
	},

	[TN_MANAGER] = {
		.source = "manager.smd",
		.desc = NULL,
		.state_set = NULL,
		.expected_instances = 1,
		.instance_seqn = 0,
		.create_io_channel = 0,
	},

	[TN_LISTENER] = {
		.source = "listener.smd",
		.desc = NULL,
		.state_set = NULL,
		.expected_instances = 1,
		.instance_seqn = 0,
		.create_io_channel = 1,
	},
};

struct edsm_api *engine = NULL;
struct echo_server server = {0};

int main(void)
{
	printf("EDSM-G2 based echo-server\n");
	engine = edsm_init("smd", mt, ARRAY_SIZE(mt));

	prepare_pools(&server);
	prepare_machines(engine, mt, &server);

	main_loop;

	return 0;
}

P.S.
В кои-то веки сподобился заменить линейный поиск перехода/действия по коду события/сообщения табличкой с адресами соответствующих структур.

P.P.S.
Окончательная редакция. Устаканился регламент взаимодействия автоматов RX/TX с их пользователями, эхо-сервер оформлен в виде сервиса systemd, усовершенствована система логирования плюс всякие мелкие улучшения. В качестве бонуса прилагаются динамический массив и префиксное дерево.

Скачать

Дата последней модификации: 2021-02-27