forum.boolean.name - 0mq, готовим сеть вкуснее (иначе)

Как часто вы сталкивались что вам нужны сокеты, а все либы - какой-то тяжеловесный ужас ? Вам нужна быстрая передача пакетов, а UDP шлёт дейтаграммы без досылок. Вам нужно чтобы оно работало бл*, а в итоге вы компилите уже второй час какой-то ужас. При этом лень самому написать "эти 20 килобайт" с двумя сокетами и потоком :crazy:

Теперь уточним - если вам нужен HTTP (или REST api скажем), то "эти 20 килобайт" всё же придется писать, таков уж путь веба :) Но если вам нужна просто передача данных то welcome !

Долгое время для геймдева не было ничего лучше RakNet, но капитализм взял своё, и теперь она стоит как обычный серьезный middleware, отдавать пару тыс, а то и все 15k$ за сетевую либу это не очень просто, особенно если бюджет игры у нас меньше чем 15k$ :crazy: В общем вещь хорошая, но статья о том где вкуснее.

Забегая наперёд скажу что мы опять поговорим о "готовим вкуснее", те представленное решение использует tcp сокеты (и не только их).

В общем сегодня мы копнём в messaging, unix'ы базируются на подходе что у нас будет куча программ которые делают одно действие, но делают его хорошо (набор инструментаря) и какой-то фронт-енд для него, собственно если процессов много, то им надо как-то взаимодействовать с друг дружкой, вспомним еще то что "все программы в одном компьютере" это собственно наследие ibm pc, в реальности же дело было так что часть программы была на одном компьютере, часть в другом, и они взаимодействовали через разные сети\протоколы :crazy: (собственно x11 и щас по сети на ура пробрасывается), в общем взаимодействие разных инстансов программ в рамках одной системы и в рамках нескольких компьютеров - это реалии даже вне ip стека.

Если у нас куча программ, и каждой надо слать друг дружке сообщения, то мы как раз попали куда надо !

0mq (zeromq) была создана чтобы решить весь геморой в отправке сообщений, библиотека написана на C++, использует свой формат ZMTP и передаёт его поверх tcp, icp и интерконект (внутри процесса).

Как же оно решает все наши проблемы ? А просто, 0mq :

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

Казалось бы описание обычной сетевой либы, которую компилить 2 часа с матюками, но прорыв заключается в последних двух пунктах ! Особенно в последнем. Ибо не зря её еще называют The Intelligent Transport Layer.

Простой вариант, у нас есть программа которая делает запрос (request - req), и программа которая отвечает на запрос (replier - rep) :

Суть контекстно-зависимого топологического решения тут в том что req сокет не может принимать пакет пока не отправит запрос, а rep не может отправлять пакет пока не получит запрос :crazy: В итоге это выливается в том что у нас есть чёткие цепочки, req : send, recv, send, recv, send, recv, ..., а у rep : recv, send, recv, send, ...

Пока просто ? давайте глянем на код :

Код:

//

//  Hello World сервер

//  Биндит REP сокет к tcp://*:5555

//  Ожидает "Hello" от клиента, отвечает с "World"

//

 #include <zmq.h>

 #include <stdio.h>

 #include <unistd.h>

 #include <string.h>



int main (void)

 {

    void *context = zmq_init (1);



    //  Открытие сокета

    void *responder = zmq_socket (context, ZMQ_REP);

    zmq_bind (responder, "tcp://*:5555");



    while (1) {

        //  Ожидаем запрос

        zmq_msg_t request;

        zmq_msg_init (&request);

        zmq_recv (responder, &request, 0);

        printf ("Received Hello\n");

        zmq_msg_close (&request);



        //  Делаем "работу"

        sleep (1);



        //  Отправляем ответ

        zmq_msg_t reply;

        zmq_msg_init_size (&reply, 5);

        memcpy (zmq_msg_data (&reply), "World", 5);

        zmq_send (responder, &reply, 0);

        zmq_msg_close (&reply);

    }



    zmq_close (responder);

    zmq_term (context);

    return 0;

 }

39 строчек, да ? А теперь в чём магия ? Магия в том что это - всё, то есть, да, это БЛ*ТЬ полноценная программа, это само обрабатывает все возможные разрывы, это само создает себе потоки, это само работает !

Чтобы не стоять на месте, давайте рассмотрим далее :

У нас есть pub сокет - в него можно только писать, и sub сокет - он только принимает данные, сам.

* тут в примерах используется s_send и s_recv, это функции хелперы для туториала, они просто создают сообщение\получают и записывают\читают строку, этот код вынесли чтобы лучше показать суть примера

Код:

//

//  Weather update server

//  Binds PUB socket to tcp://*:5556

//  Publishes random weather updates

//

 #include "zhelpers.h"



int main (void)

 {

    //  Prepare our context and publisher

    void *context = zmq_init (1);

    void *publisher = zmq_socket (context, ZMQ_PUB);

    zmq_bind (publisher, "tcp://*:5556");

    zmq_bind (publisher, "ipc://weather.ipc");



    //  Initialize random number generator

    srandom ((unsigned) time (NULL));

    while (1) {

        //  Get values that will fool the boss

        int zipcode, temperature, relhumidity;

        zipcode     = randof (100000);

        temperature = randof (215) - 80;

        relhumidity = randof (50) + 10;



        //  Send message to all subscribers

        char update [20];

        sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);

        s_send (publisher, update);

    }

    zmq_close (publisher);

    zmq_term (context);

    return 0;

 }

Код:

//

//  Weather update client

//  Connects SUB socket to tcp://localhost:5556

//  Collects weather updates and finds avg temp in zipcode

//

 #include "zhelpers.h"



int main (int argc, char *argv [])

 {

    void *context = zmq_init (1);



    //  Socket to talk to server

    printf ("Collecting updates from weather server…\n");

    void *subscriber = zmq_socket (context, ZMQ_SUB);

    zmq_connect (subscriber, "tcp://localhost:5556");



    //  Subscribe to zipcode, default is NYC, 10001

    char *filter = (argc > 1)? argv [1]: "10001 ";

    zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE, filter, strlen (filter));



    //  Process 100 updates

    int update_nbr;

    long total_temp = 0;

    for (update_nbr = 0; update_nbr < 100; update_nbr++) {

        char *string = s_recv (subscriber);

        int zipcode, temperature, relhumidity;

        sscanf (string, "%d %d %d",

            &zipcode, &temperature, &relhumidity);

        total_temp += temperature;

        free (string);

    }

    printf ("Average temperature for zipcode '%s' was %dF\n",

        filter, (int) (total_temp / update_nbr));



    zmq_close (subscriber);

    zmq_term (context);

    return 0;

 }

Обычно их используют для передачи нотификейшенов и прочего, уточню что zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE, filter, strlen (filter)); устанавливает фильтр, 0mq сама фильтрует пакеты по первым N байт (где N это размер нашего фильтра).

А теперь map-reduce ! Схоже с pub\sub, push сокеты используются для передачи, а pull для получения, основное отличие в том что они используются для распределения сообщений (те каждый worker не получает все сообщения, а только часть).

Код:

//

//  Task ventilator

//  Binds PUSH socket to tcp://localhost:5557

//  Sends batch of tasks to workers via that socket

//

 #include "zhelpers.h"



int main (void)

 {

    void *context = zmq_init (1);



    //  Socket to send messages on

    void *sender = zmq_socket (context, ZMQ_PUSH);

    zmq_bind (sender, "tcp://*:5557");



    //  Socket to send start of batch message on

    void *sink = zmq_socket (context, ZMQ_PUSH);

    zmq_connect (sink, "tcp://localhost:5558");



    printf ("Press Enter when the workers are ready: ");

    getchar ();

    printf ("Sending tasks to workers…\n");



    //  The first message is "0" and signals start of batch

    s_send (sink, "0");



    //  Initialize random number generator

    srandom ((unsigned) time (NULL));



    //  Send 100 tasks

    int task_nbr;

    int total_msec = 0;     //  Total expected cost in msecs

    for (task_nbr = 0; task_nbr < 100; task_nbr++) {

        int workload;

        //  Random workload from 1 to 100msecs

        workload = randof (100) + 1;

        total_msec += workload;

        char string [10];

        sprintf (string, "%d", workload);

        s_send (sender, string);

    }

    printf ("Total expected cost: %d msec\n", total_msec);

    sleep (1);              //  Give 0MQ time to deliver



    zmq_close (sink);

    zmq_close (sender);

    zmq_term (context);

    return 0;

 }

Код:

//

//  Task worker

//  Connects PULL socket to tcp://localhost:5557

//  Collects workloads from ventilator via that socket

//  Connects PUSH socket to tcp://localhost:5558

//  Sends results to sink via that socket

//

 #include "zhelpers.h"



int main (void)

 {

    void *context = zmq_init (1);



    //  Socket to receive messages on

    void *receiver = zmq_socket (context, ZMQ_PULL);

    zmq_connect (receiver, "tcp://localhost:5557");



    //  Socket to send messages to

    void *sender = zmq_socket (context, ZMQ_PUSH);

    zmq_connect (sender, "tcp://localhost:5558");



    //  Process tasks forever

    while (1) {

        char *string = s_recv (receiver);

        //  Simple progress indicator for the viewer

        fflush (stdout);

        printf ("%s.", string);



        //  Do the work

        s_sleep (atoi (string));

        free (string);



        //  Send results to sink

        s_send (sender, "");

    }

    zmq_close (receiver);

    zmq_close (sender);

    zmq_term (context);

    return 0;

 }

Код:

//

//  Task sink

//  Binds PULL socket to tcp://localhost:5558

//  Collects results from workers via that socket

//

 #include "zhelpers.h"



int main (void)

 {

    //  Prepare our context and socket

    void *context = zmq_init (1);

    void *receiver = zmq_socket (context, ZMQ_PULL);

    zmq_bind (receiver, "tcp://*:5558");



    //  Wait for start of batch

    char *string = s_recv (receiver);

    free (string);



    //  Start our clock now

    int64_t start_time = s_clock ();



    //  Process 100 confirmations

    int task_nbr;

    for (task_nbr = 0; task_nbr < 100; task_nbr++) {

        char *string = s_recv (receiver);

        free (string);

        if ((task_nbr / 10) * 10 == task_nbr)

            printf (":");

        else

            printf (".");

        fflush (stdout);

    }

    //  Calculate and report duration of batch

    printf ("Total elapsed time: %d msec\n",

        (int) (s_clock () - start_time));



    zmq_close (receiver);

    zmq_term (context);

    return 0;

 }

А теперь в чём реальная магия ? Тип сокета и его подключение - это разные вещи ! :crazy: Как видим тут у нас есть сокет push и pull, но в каких-то случаях мы их просто биндим на порт, а в каких-то коннектимся к адресу.

Почему ?

Давайте посмотрим на банальное сетевое начало :

Прокрутите в голове сколько кода нужно и как он будет выглядеть чтобы сделать такую топологию надёжной и работающей в отдельном потоке, нужно же отслеживать ошибки передачи, делать очереди и тд и тп. Если у нас только так и остаётся - всё ок.

Но на самом деле так почти никогда не остаётся, под релиз у вас будет вот так :

Думаю любой программист будет явно недоволен писать такое на сокетах во время кранча в 4 утра.

0mq тут выступает красной и синей пилюлей одновременно, с одной стороны мы начинаем смотреть на сеть по другому, нам начинает нравится писать сетевой код, становится легко его дебажить даже в голове, с другой стороны это сильный удар по устоявшемся устоям сетевого программирования.

Дальнейшее погружение тут : http://www.zeromq.org/ и http://zguide.zeromq.org/page:all

А теперь набор фактов :

1) добавить 0mq в свой проект - дело 5-10 минут, простые исходники, простая компиляция
2) вы навсегда забываете про головную боль, слова "ошибка передачи", "разрыв соединения" и тд
3) вы не думаете о том posix или win потоки используются, вы не думаете о потоках вообще
4) вы можете нагрузить канал передачи полностью, шлите хоть миллион пакетов в секунду - всё придет с максимальной скоростью канала связи
5) вы не думаете о порядке приема\передачи сообщений, он всегда такой который вы и ожидали
6) вы спите более спокойно и кранчи проходят менее болезненно
7) доступно на c, c++, c#, clojure, cl, erlang, f#, felix, go, haskell, haxe, java, lua, node.js, objective-c, perl, php, python, ruby, scala, tcl, ada, basic

ps. статья направлена заинтриговать читателя и предоставить читателю решения его сетевых проблем, если интересно могу написать где, кто и как использует 0mq :)