epoll IO с рабочими потоками в C

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

Сами данные, вероятно, лучше всего хранить в кольцевом буфере, однако нам понадобится отдельный буфер для каждого источника ввода (если мы получим частичный пакет, мы можем продолжить получение из других источников, не разделяя данные. Остается вопрос, как информировать когда новый пакет готов, и дать им указатель на данные в указанном пакете. Поскольку данных здесь мало, всего несколько указателей, наиболее элегантным способом сделать это будет с помощью очередей сообщений posix. Они обеспечивают возможность для нескольких отправителей и нескольких получателей писать и читать сообщения, всегда гарантируя получение каждого сообщения ровно 1 потоком.

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

struct DataSource
{
    int SourceFD;
    char DataBuffer[MAX_PACKET_SIZE * (THREAD_COUNT + 1)];
    char *LatestPacket;
    char *CurrentLocation
    int SizeLeft;
};

SourceFD, очевидно, является файловым дескриптором рассматриваемого потока данных, DataBuffer - это место, где содержимое пакетов хранится во время обработки, это кольцевой буфер. Указатель LatestPacket используется для временного хранения указателя на наиболее часто отправленный пакет на случай, если мы получим частичный пакет и перейдем к другому источнику, прежде чем передать пакет. CurrentLocation хранит, где заканчивается последний пакет, чтобы мы знали, где разместить следующий или куда продолжить в случае частичного приема. Оставшийся размер - это место, оставшееся в буфере, он будет использоваться, чтобы определить, можем ли мы уместить пакет или нужно вернуться к началу.

Таким образом, функция приема будет эффективно

• Скопируйте содержимое пакета в буфер
• Переместите CurrentLocation, чтобы указать на конец пакета
• Обновите SizeLeft, чтобы учесть теперь уменьшенный буфер
• Если мы не можем поместить пакет в конец буфера, мы выполняем цикл
• Если и там нет места, мы пробуем еще раз чуть позже, тем временем переходя к другому источнику.
• Если бы у нас было частичное получение, сохраните указатель LatestPacket, чтобы указать на начало пакета и перейти к другому потоку, пока мы не получим остальные.
• Отправьте сообщение с помощью очереди сообщений posix в рабочий поток, чтобы он мог обработать данные, сообщение будет содержать указатель на структуру DataSource, чтобы он мог работать с ней, ему также нужен указатель на пакет, над которым он работает, и его размер, их можно вычислить, когда мы получим пакет

Рабочий поток будет выполнять свою обработку, используя полученные указатели, а затем увеличивать SizeLeft, чтобы поток-получатель знал, что он может продолжить заполнение буфера. Атомарные функции потребуются для работы со значением размера в структуре, поэтому мы не получаем условий гонки с помощью свойства size (поскольку возможно, что он записывается одновременно рабочим и потоком ввода-вывода, что приводит к потерям записи, см. комментарий ниже), они перечислены здесь и просты и чрезвычайно полезны.

Теперь я дал некоторые общие сведения, но остановлюсь на конкретных моментах:

1. Использование EventFD в качестве механизма синхронизации - в значительной степени плохая идея, вы обнаружите, что используете изрядное количество ненужного процессорного времени, и очень сложно выполнить какую-либо синхронизацию. В частности, если у вас есть несколько потоков, которые используют один и тот же дескриптор файла, у вас могут возникнуть серьезные проблемы. По сути, это неприятный прием, который иногда работает, но не может заменить правильную синхронизацию.
2. Также плохая идея - попытаться разгрузить прием, как описано выше, вы можете обойти проблему со сложным IPC, но, честно говоря, маловероятно, что получение ввода-вывода займет достаточно времени, чтобы остановить ваше приложение, ваш ввод-вывод также, вероятно, намного медленнее, чем процессор поэтому получение с несколькими потоками мало что даст. (это предполагает, что вы не говорите, что имеете несколько 10-гигабитных сетевых карт).
3. Использование мьютексов или блокировок здесь - глупая идея, она намного лучше подходит для кодирования без блокировки, учитывая небольшой объем (одновременно) общих данных, вы действительно просто передаете работу и данные. Это также повысит производительность принимающего потока и сделает ваше приложение более масштабируемым. Использование упомянутых здесь функций http://gcc.gnu.org/onlinedocs/gcc-4.1.2/gcc/Atomic-Builtins.html, вы можете сделать это легко и просто. Если вы сделали это таким образом, вам понадобится семафор, его можно разблокировать каждый раз, когда пакет будет получен и заблокирован каждым потоком, который запускает задание, чтобы динамически разрешить больше потоков, если будет готово больше пакетов, что будет иметь гораздо меньше накладных расходов, чем решение homebrew с мьютексами.
4. Здесь нет особой разницы с любым пулом потоков, вы создаете много потоков, а затем все они блокируются в mq_receive в очереди сообщений данных для ожидания сообщений. По завершении они отправляют свой результат обратно в основной поток, который добавляет очередь сообщений с результатами в свой список epoll. Затем он может получать результаты таким образом, это просто и очень эффективно для небольших полезных данных, таких как указатели. Это также будет использовать мало ЦП и не заставит основной поток тратить время на управление рабочими.

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

Надеюсь, это поможет, но уже поздно, поэтому, если я что-то пропустил или у вас есть вопросы, не стесняйтесь комментировать, чтобы уточнить или получить дополнительные объяснения.

В моих тестах один экземпляр epoll на поток намного превосходил сложные модели потоковой передачи. Если сокеты слушателя добавлены ко всем экземплярам epoll, рабочие будут просто accept(2), а победитель получит соединение и обработает его в течение всего срока его службы.

Ваши рабочие могут выглядеть примерно так:

for (;;) {
    nfds = epoll_wait(worker->efd, &evs, 1024, -1);

    for (i = 0; i < nfds; i++)
        ((struct socket_context*)evs[i].data.ptr)->handler(
            evs[i].data.ptr,
            evs[i].events);
}

И каждый дескриптор файла, добавленный к экземпляру epoll, может иметь struct socket_context связанный с ним:

void listener_handler(struct socket_context* ctx, int ev)
{
    struct socket_context* conn;

    conn->fd = accept(ctx->fd, NULL, NULL);
    conn->handler = conn_handler;

    /* add to calling worker's epoll instance or implement some form
     * of load balancing */
}

void conn_handler(struct socket_context* ctx, int ev)
{
    /* read all available data and process. if incomplete, stash
     * data in ctx and continue next time handler is called */
}

void dummy_handler(struct socket_context* ctx, int ev)
{
    /* handle exit condition async by adding a pipe with its
     * own handler */
}

Мне нравится эта стратегия, потому что:

• очень простой дизайн;
• все нити идентичны;
• рабочие и связи изолированы - нельзя наступать друг другу на пятки или звонить read(2) не тому работнику;
• никаких блокировок не требуется (ядро заботится о синхронизации на accept(2));
• в некоторой степени естественная балансировка нагрузки, поскольку ни один занятый работник не будет активно бороться за accept(2).

И несколько замечаний по epoll:

• использовать режим запуска по фронту, неблокирующие сокеты и всегда читать до EAGAIN;
• избегайте dup(2) семейства вызовов, чтобы избавить себя от некоторых сюрпризов (epoll регистрирует дескрипторы файла, но фактически отслеживает описания файлов);
• вы можете безопасно epoll_ctl(2) экземпляры epoll других потоков;
• используйте большой struct epoll_event буфер для epoll_wait(2), чтобы избежать голода.

Некоторые другие примечания:

• используйте accept4(2), чтобы сохранить системный вызов;
• используйте один поток на ядро ​​(1 для каждого физического, если привязан к ЦП, или 1 для каждого логического, если привязан к вводу-выводу);
• _14 _ / _ 15_ будет быстрее, если количество подключений мало.

Надеюсь, это поможет.

Я отправляю тот же ответ в другом сообщении: Я хочу дождаться и дескриптора файла, и мьютекса. Каков рекомендуемый способ сделать это?

==========================================================

Это очень распространенная проблема, особенно когда вы разрабатываете сетевую серверную программу. Основной вид большинства серверных программ Linux будет выглядеть следующим образом:

epoll_add(serv_sock);
while(1){
    ret = epoll_wait();
    foreach(ret as fd){
        req = fd.read();
        resp = proc(req);
        fd.send(resp);
    }
}

Это однопоточный (основной поток) серверный фреймворк на основе epoll. Проблема в том, что он однопоточный, а не многопоточный. Это требует, чтобы proc () никогда не блокировался или не запускался в течение значительного времени (скажем, 10 мс для общих случаев).

Если proc () когда-либо будет работать в течение длительного времени, НАМ НУЖНЫ НЕСКОЛЬКО ПОТОКОВ и выполнение proc () в отдельном потоке (рабочем потоке).

Мы можем отправить задачу рабочему потоку, не блокируя основной поток, используя очередь сообщений на основе мьютексов, это достаточно быстро.

Затем нам нужен способ получить результат задачи из рабочего потока. Как? Однако если мы просто проверим очередь сообщений напрямую, до или после epoll_wait (), действие проверки будет выполнено после завершения epoll_wait (), а epoll_wait () обычно блокируется на 10 микросекунд (общие случаи), если все файловые дескрипторы ожидают не активны.

Для сервера 10 мс - довольно много времени! Можем ли мы сигнализировать epoll_wait () о немедленном завершении работы, когда сгенерирован результат задачи?

Да! Я опишу, как это делается, в одном из моих проектов с открытым исходным кодом.

Создайте канал для всех рабочих потоков, и epoll также будет ожидать на этом конвейере. Как только результат задачи сгенерирован, рабочий поток записывает один байт в конвейер, после чего epoll_wait () завершится примерно за то же время! - Linux pipe имеет задержку от 5 до 20 мкс.

В моем проекте SSDB (дисковая база данных NoSQL, совместимая с протоколом Redis), я создаю SelectableQueue для передачи сообщения между основным потоком и рабочими потоками. Как и его название, SelectableQueue имеет файловый дескриптор, который может ждать epoll.

SelectableQueue: https://github.com/ideawu/ssdb/blob/master/src/util/thread.h#L94

Использование в основном потоке:

epoll_add(serv_sock);
epoll_add(queue->fd());
while(1){
    ret = epoll_wait();
    foreach(ret as fd){
        if(fd is worker_thread){
            sock, resp = worker->pop_result();
            sock.send(resp);
        }
        if(fd is client_socket){
            req = fd.read();
            worker->add_task(fd, req);
        }
    }
}

Использование в рабочем потоке:

fd, req = queue->pop_task();
resp = proc(req);
queue->add_result(fd, resp);