Пример построения неблокирующего веб-приложения

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

Недавно захотел создать свой сервис проверки номеров ICQ на невидимость. Алгоритм проверки старый и известный, но до сих пор работающий — отправка специально сформированного служебного сообщения и анализ ответа сервера. Необходимо было держать несколько постоянных подключений к серверу ICQ, а также иметь веб-интерфейс для запросов на проверку. Очевидное решение — создание демона, который создает несколько потоков для ICQ-соединений, и как-либо получает команды от веб-приложения, использующего несколько процессов-воркеров (или на preforked архитектуре) — для возможности обрабатывать http-запросы от нескольких клиентов. Но я решил освоить новую для себя технологию и сделать приложение, поддерживающее несколько соединений и отвечающее клиентам, используя всего лишь один поток.

Event loops

В Perl есть множество реализаций event loops — фреймворков для создания событийно-ориентированных приложений. Это модули, которые предоставляют интерфейс для регистрации обработчиков различных событий (срабатывание таймера, получение сигнала, появление в сокете данных, которые можно считать), а также функцию, после вызова которой происходит блокировка выполнения программы и начинается обработка событий. При наступлении события происходит вызов колбэка, указанного при регистрации. Многие event loops имеют возможность указывать бэкенд для оповещения о событиях, например, высокопроизводительные kqueue и epoll.
На CPAN есть множество модулей, использующих те или иные event loops. Порой бывает, что есть все нужные модули для решения задачи, но они используют разные event loops с несовместимыми интерфейсами. Что же делать в таких случаях?

AnyEvent

AnyEvent — DBI для событийно-ориентированного программирования. AnyEvent предоставляет интерфейс, дающий возможность в любой момент сменить используемый event loop на другой. Также AnyEvent позволяет использовать вместе модули, использующие разные event loops. Именно поэтому я написал реализацию протокола ICQ с использованием AnyEvent.

Протокол ICQ

Первоочередная задача — подключиться к серверу сообщений ICQ и реализовать алгоритм проверки. Три имеющихся на CPAN модуля были довольно старыми и были выполнены на блокирующих сокетах, поэтому не подходили мне. Ведь любая блокировка внутри колбэка вызовет блокировку обработки всех остальных событий! Но все же для простоты я сначала сделал реализацию на блокирующих сокетах, используя многочисленные готовые решения на других языках и эти модули, и потом начал переделывать все на AnyEvent. Ниже приведен код получения одного пакета ICQ — FLAP:

# блокирующий вариант
sub recv {
    my ($self) = @_;
 
    # прочитать 6 байт - заголовок FLAP
    sysread $self->socket, my $data, 6;
    # в последних двух байтах содержится длина пакета
    my $length = unpack('n', substr($data, -2));
    # во втором байте - номер канала
    my $channel = unpack('C', substr($data, 1, 1));
    # читаем данные, размер которых мы извлекли из заголовка
    sysread $self->socket, $data, $length;
    # теперь у нас есть все для создания OC::ICQ::FLAP
    return new OC::ICQ::FLAP($channel, $data);
}
 
# неблокирующий вариант
sub connect {
    my ($self, $host, $port) = @_;
 
    # используем AnyEvent::Handle в качестве абстракции над сокетом для более удобной работы
    $self->{io} = new AnyEvent::Handle (
        connect => [$host, $port],
        # регистрируем колбэки на случай разрыва соединения и ошибки
        on_error => sub {$self->on_error(@_)},
        on_disconnect => sub {$self->on_disconnect(@_)},
    );
 
    # регистрируем колбэк, который будет вызван после того, как из сокета можно будет прочитать 6 байт - заголовок FLAP
    $self->{io}->push_read(chunk => 6, sub {$self->on_read_header(@_)});
}
 
# колбэк, вызываемый при получении заголовка FLAP
sub on_read_header {
    my ($self, $io, $header) = @_;
 
    # читаем и запоминаем номер канала - он будет нужен для создания OC::ICQ::FLAP
    $self->{flap_channel} = unpack('C', substr($header, 1, 1));
    # регистрируем другой колбэк, который будет вызван при получении данных указанного в заголовке размера
    $io->push_read(chunk => unpack('n', substr($header, -2)), sub {$self->on_read_data(@_)});
}
 
# колбэк, вызываемый при получении данных
sub on_read_data {
    my ($self, $io, $data) = @_;
 
    # мы получили данные, теперь можно создавать OC::ICQ::FLAP, используя заранее сохраненный номер канала
    $self->_process_flap(new OC::ICQ::FLAP($self->{flap_channel}, $data));
    # вновь ждем заголовок FLAP
    $io->push_read(chunk => 6, sub {$self->on_read_header(@_)});
}

Всю остальную логику, лежащую в _process_flap, переделывать почти не пришлось. Для поддержания соединения нужно посылать пустые FLAP по 5 каналу раз в 2 минуты. Для этого можно использовать предоставляемую AnyEvent функцию timer:

# устанавливаем таймер, вызывающий колбэк каждые 2 минуты
$self->{keepalive_timer} = AnyEvent->timer(after => 120, interval => 120, cb => sub {$self->send_keepalive});
 
sub send_keepalive {
    my ($self) = @_;
 
    if ($self->{state} == ONLINE) {
        $self->send(new OC::ICQ::FLAP(5, ''));
    }
}
 
# для удаления таймера нужно удалить все сильные ссылки на него. Единственная сохранена в $self->{keepalive_timer}
delete $self->{keepalive_timer};

Веб-интерфейс

Отлично, реализация протокола ICQ и алгоритма проверки готова, теперь нужен веб-интерфейс. Для подключения веб-приложений к веб-серверу существует замечательный протокол FastCGI, и на CPAN я нашел две его асинхронные реализации для EV и IO::Async. Выбрал EV из-за его быстродействия. Далее был сделан несложный url-диспатчер на атрибутах и прикручен простой шаблонизатор Text::MicroMason — всё, мини-фреймворк для создания асинхронных веб-приложений готов.
Text::MicroMason хранит шаблоны в скомпилированном виде в памяти, что замечательно сказывается на производительности, но что делать если нужно изменить шаблон? Не останавливать же демона, обрывая соединения всех ICQ-клиентов? AnyEvent и EV предоставляют возможность устанавливать обработчики на сигналы, этим можно воспользоваться.

    my $sigusr1_watcher = EV::signal('USR1', \&restart) unless $^O =~ /MSWin32/;
    my $sigusr2_watcher = EV::signal('USR2', \&load_templates) unless $^O =~ /MSWin32/;
 

Теперь при получении SIGUSR1 будет заново подгружен конфиг, удалены все старые ICQ-клиенты и созданы новые, а при получении SIGUSR2 будут перезагружены шаблоны. Как и в случае с таймером, обязательно нужно сохранять возвращаемое EV::signal/AnyEvent->signal значение.