Рекурсивный вызов в libev (EVBREAK_RECURSE) при организации многопоточного сервера

#c #многопоточность #libev

Суть алгоритма (c libevent все работало):

Используется один  луп на прослушивание сокета. 
Запускается несколько тредов, они ложатся в спячку. Каждый тред присоединяет событие
на Read 
По ассепту они все пробуждаются, и кто успел... пробудился первый, тот и захватил
мютекс, взял соединение...
Блокировку снимаем, треды засыпаю., а наш тред отрабатывает...
Берется пул и кладется в него файловый дескриптор и необходимые
    буфера (в/в) и структуры  . id
    блокируем мьютексом. 
Далее мы
    организуем самопально noblocked
    чтение, код типа buffer_event 

После того как данные будут отданы,
освобождаем пул и тред засыпает.
далее п 3.


Имею ошибку:
Assertion failed: (("libev: ev_loop recursion during release detected", loop_done
!= EVBREAK_RECURSE)), function ev_run, file ev.c, line 2392.

код github 
Читаем файл INSTALL - для большитнства unix систем должно скомпилироваться без проблем, 
должна быть установлена libev, если не компилится, то правим пути в Makefile.am и
потом все с первого шага ./autogen.sh    

Ответы

Ответ 1

Я делал классическую схему "кто успел, тот и съел": как и вы заводил пул потоков
(далее воркеров), в каждом воркере запускал ev_loop, для accept создавал столько же
ev_io вотчеров, колбек делал им один. Блокировку осуществлял при помощи механизма ev_set_loop_release_cb
(loop, void (*release)(EV_P) throw (), void (*acquire)(EV_P) throw ())

Мутекс, который лочится в acquire() и отпускается в release() для каждого цикла свой.
Лочится он перед запуском цикла, а внутри цикла, когда нужно, отпускается и обратно
забирается когда нужно, при помощи ваших колбеков.

В общем картина такая: для асинхронного accept прилетает эвент всем воркерам, для
кого первым отработал accept, тому и возвратился дескриптор клиентского сокета, остальным
вернулось -1/EAGAIN. Далее уже дескриптор полностью обслуживается циклом, на котором
он сработал. Распределение нагрузки получается достаточно равномерной.

Если что-то непонятно - дополню.

Пример на C++11: https://gitlab.com/snippets/8201