Извините, ничего не найдено.

06.07.2012, 03:08

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

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

Но если вам нужна просто передача данных то welcome !

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

В общем вещь хорошая, но статья о том где вкуснее.

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

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

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

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

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

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

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

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

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

Суть контекстно-зависимого топологического решения тут в том что req сокет не может принимать пакет пока не отправит запрос, а rep не может отправлять пакет пока не получит запрос

В итоге это выливается в том что у нас есть чёткие цепочки, 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;
 }

А теперь в чём реальная магия ? Тип сокета и его подключение - это разные вещи !

Как видим тут у нас есть сокет 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

Mr_F_ · 06.07.2012, 04:12

плюсую, не так давно требовалось в срочном порядке добавить сеть в существующий нагроможденный медленно компилящийся проект, достаточно было взять 0mq и код из hello world примера на сайте, чтобы всё заработало.
все довольны, я получил свои монеты 8-)
---
ах да, до этого с сетью не работал никогда, нигде и вообще.

impersonalis · 06.07.2012, 11:14

НУ ГДЕ Ж ТЫ БЫЛ неделю назад, когда я писал (и дописал) свою библиотеку, работающую в отдельном потоке с сокетами в неблокирующем режиме. (это я типа "спасибо" сказал).
Удивительно - специально пересмотрел ряд предложений: от тяжеловесного ASIO из буста, до совсем лёгких библиотек - все имели ряд изъянов, а сабж не попался =(