Есть файл со списком из 10 000 записей "ip+port".
Нужно мониторить состояние подключения по адресу. Реализовать чек "жив ли сервер", фактически.
И оповещать — выводить в консоль и файл запись "ts + address + up/down".
Всё это нужно делать быстро.
1ая реализация
epoll + main thread + output
Придумал, что можно в epoll получать список fd, у которых случилось событие "соединение упало/соединение установлено".
В моменте выводить информацию. Идти к следующему fd, ... Минус реализации, что проходимся в цикле по выданному пулу fd синхронно записывая в консоль, файл. А в это время уже события на fd новые могут придти.
2ая реализация
epoll + threads + background output
Создаём пулл потоков с theads_number = hardware_concurrency. Чтобы вышло сколько ядер, столько и потоков. Честное распараллеливание.
К примеру, будет 4.
class ThreadPool {
using Task = std::future<void>;
public:
explicit ThreadPool(std::size_t threads_num);
~ThreadPool();
template <typename F, typename ... Args>
void addTask(F f, Args&& ... args) {
{
std::lock_guard<std::mutex> l(cv_m_);
tasks_.emplace(std::async(std::launch::deferred, std::forward<F>(f), std::forward<Args>(args)...));
}
condition_.notify_one();
}
private:
std::mutex cv_m_;
std::condition_variable_any condition_;
std::queue<Task> tasks_;
std::vector<std::jthread> workers_;
};
ThreadPool::ThreadPool(std::size_t threads_num)
{
for (std::size_t i = 0; i < threads_num; ++i) {
workers_.emplace_back(
[this](std::stop_token stop_token){
while (true) {
std::unique_lock<std::mutex> lk(this->cv_m_);
condition_.wait(lk, stop_token, [this](){
return !this->tasks_.empty();
});
if (stop_token.stop_requested()) {
return;
}
if (!this->tasks_.empty()) {
auto f = std::move(this->tasks_.front());
this->tasks_.pop();
lk.unlock();
f.get();
}
}
}
);
}
}
Получаем от epoll список fd. Делим его на 4 — получаем 4 списка. Каждый отдаём на обработку в пул потоков. Минус в том, что сгородили целый пулл потоков ради всего-лишь вывода в файл. И файл-то один. Нужно же синхронизировать к нему доступ. Через мьютекс? Вся многопоточность убьётся об него.
Какие ещё могут быть идеи?
Подумал над memory mapped file + спин лок. Спин локом защищаем общую переменную — смещение. Поток подготовил строку для вставки в файл. Теперь ему нужно узнать по какому смещению её записать.
Он лочит спин лок, сохраняет себе смещение, обновляет его — прибавляет к нему длину строки, которую сейчас вставит. Отпускает спин лок. Вставляет по полученному смещению строку.
Т.е. критическая секция получилась маленькая.
Здравствуйте, avovana, Вы писали:
A> 1ая реализация
Конечно же ты замерил производительность, и она тебя не устроила? Не устроила чем?
Потому что реализация хоть и простая, но если одного потока хватает для обработки всех событий, то она же и будет быстрой.
A> А в это время уже события на fd новые могут придти.
Ну и что?
Они не потеряются. А если запись в output тормозит, то всё равно не сможешь в output вывести об этом сообщение, даже если сумеешь обработать событие на fd раньше.
Уметь обрабатывать события раньше имеет смысл только если почему-то нужно узнавать более точное время, когда оно наступило. Или если нужно уметь дедуплицировать события: например, если произошёл переход up->down->up, но в output ещё не успели записать строку про up->down, то не записывать в output ничего вообще — как будто соединение и не падало.
A> синхронно записывая в консоль, файл.
Можешь писать асинхронно :)
Добавить файл в epoll и точно так же слушать на нём события.
Можно и одним потоком обойтись, но ниже будет вариант с парой, который может быть проще для восприятия и реализации — потому что стандартные примитивы очередей и future/promise проще, чем автомат поверх событий epoll.
A> проходимся в цикле по выданному пулу fd синхронно записывая в консоль, файл
Скорее имеет смысл подготовить данные про группу дескрипторов и один раз записать в файл, а не на каждый дескриптор дёргать вывод в файл. То есть либо использовать, условно говоря, один writev вместо многих write, либо использовать обычную буферизацию записи (просто при выводе в консоль она может быть всего лишь построчной по умолчанию).
A> Подумал над memory mapped file + спин лок. Спин локом защищаем общую переменную — смещение. Поток подготовил строку для вставки в файл. Теперь ему нужно узнать по какому смещению её записать. A> Он лочит спин лок, сохраняет себе смещение, обновляет его — прибавляет к нему длину строки, которую сейчас вставит. Отпускает спин лок. Вставляет по полученному смещению строку. A> Т.е. критическая секция получилась маленькая.
У тебя есть замер производительности, который обнаружил проблемы в записи в файл в этом месте? Правда?
Это выглядит как злющая предварительная оптимизация.
К тому же с таким подходом не всё так просто: нужно будет ведь ещё как-то уведомлять читателя о том, откуда из файла уже можно читать данные. Читатель тогда тоже должен какой-то счётчик со смещением чтения иметь, который не совпадает со смещением для записи. И который не тривиально обновлять, так как записи блоков будет завершаться не в монотонном порядке — и счётчик обязан будет уметь как задерживаться, если за ним идёт дыра, так и прыгать на несколько записей вперёд, когда в дыру закончится запись.
Можно посмотреть как это делается правильно в lock-free-like очередях поверх циклических буферов. Но это не совсем просто и, главное, слишком избыточно.
A>Получаем от epoll список fd. Делим его на 4 — получаем 4 списка. Каждый отдаём на обработку в пул потоков. A>Минус в том, что сгородили целый пулл потоков ради всего-лишь вывода в файл. И файл-то один. Нужно же синхронизировать к нему доступ. Через мьютекс? Вся многопоточность убьётся об него.
Идея почти нормальная. Только откуда-то из непонятного места возникает синхронизация к файлу.
Заводи два типа потоков:
Первый тип слушает epoll на своём наборе дескрипторов, определяет статус соединений, формирует в локальной переменной очередь событий, которые нужно записать (не важно будет это в виде массива, очереди или словаря — смотри что удобнее).
Когда массив сформирован, то он под мьютексом докидывает этот массив в глобальную очередь — это буквально одно перемещение, обычно даже без аллокаций памяти. И уведомляет, что в очереди появились данные.
Если одного потока не хватает успевать разгребать события (например, из-за того, что для определения статуса нужно что-то ещё долго считать), то запускать несколько таких потоков параллельно.
Второй тип потоков — перекладыватель в output. Он работает в единственном экземпляре.
В бесконечном цикле ждёт событие "в очереди что-то появилось", под мьютексом делает swap со своей локальной пустой очередью, и потом начинает писать из своей локальной очереди все события в файл/консоль без каких-либо синхронизаций.
Собственно, это почти всё.
Каких-то сложных потоков данных или блокировок нет — всё собирается из стандартных примитивов.
В фантастическом случае, если например промежуточная очередь тормозит, то понятно как её можно заменить (уверен, что кто-нибудь посоветует тут поиграть со всякими lock-free очередями).
И легко сделать дедупликацию событий, чтобы промежуточная очередь не росла бесконечно и чтобы программа не упала по превышению памяти, если вдруг запись в output застопорится. Так как обрабатывающий поток будет видеть, что его же блок событий с предыдущей итерации ещё не был передан для записи, и может легко сделать агрегацию (опять же, можно это делать не блокируя соседние потоки, если они есть).
Нет. Это другая домашка Та из собеса минувших дней )
Кстати, там есть не совсем очевидные ответы. Спасибо, что стали отвечать. Получил их в том числе с Линукс форума, с cpplang.slack. Дополню сейчас в той ветки, какие мысли ещё высказывались.
Здравствуйте, reversecode, Вы писали:
R>а еще есть iouring
R>вообщем то вопрос наверное на умение и понимание устройста реакторов R>или на знание фреймворков ?
R>ну берите asio или seastar и колбасте
Справедливо. Уточню, что условие — не использовать фреймворки.
Здравствуйте, watchmaker, Вы писали:
W>Второй тип потоков — перекладыватель в output. Он работает в единственном экземпляре. W>В бесконечном цикле ждёт событие "в очереди что-то появилось", под мьютексом делает swap со своей локальной пустой очередью, и потом начинает писать из своей локальной очереди все события в файл/консоль без каких-либо синхронизаций. W> W>Собственно, это почти всё. W>Каких-то сложных потоков данных или блокировок нет — всё собирается из стандартных примитивов. W>В фантастическом случае, если например промежуточная очередь тормозит, то понятно как её можно заменить (уверен, что кто-нибудь посоветует тут поиграть со всякими lock-free очередями).
W>И легко сделать дедупликацию событий, чтобы промежуточная очередь не росла бесконечно и чтобы программа не упала по превышению памяти, если вдруг запись в output застопорится. Так как обрабатывающий поток будет видеть, что его же блок событий с предыдущей итерации ещё не был передан для записи, и может легко сделать агрегацию (опять же, можно это делать не блокируя соседние потоки, если они есть).
Спасибо за такой развёрнутый ответ! Есть над чем подумать.
То есть, возможная более простая, понятная и быстрая реализация: 1 поток слушает по epoll соединению, перекладывает в общую очередь под мьютексом, cond var нотифаит другой поток разгребатель после записи. Другой поток не читает их под мьютексом а берёт себе, чтобы уже параллельно не мешая 1му потоку писать.
Правильно понял?
А как сделать этот swap? Чтобы быстро взять данные? Т.е. не использовать какую-то очередь, а просто буфер, указатель на который меняем? Для нового же нужно выделить место? malloc с каким-то значением? Но мы же не знаем сколько epoll поток вычитает. Или фиксируем сколько читать этой константой len — malloc(len)? А в случае быстрой работы не хочется дергать системные функции.
Можно, в принципе, тогда заранее определить размер общего буфера, в который пишет 1ый поток. И этот же размер будет у локального буфера у 2ого потока. Общая len. Тогда 1ый поток получив набор дескрипторов сможет записать только сколько влезет в len. Оповестить. А с остальными что делать? Может, получил 10 событий = 10 fd, а в этом фиксированном буфере выделено байт только для 5ти.
ну тоесть они хотят что бы вы им написали в качестве тестового задания ядро tokio из раста на си или с++ с ворк стилиногом?
не ну понятно что на коленке его можно накидать в условных десяток другой строчек
но я бы предложил им для начала закинуть вам аванс хотя бы 5к баксов
и 5к баксов после выполнения
а что бы не кинули
сделать это через гаранта
Здравствуйте, avovana, Вы писали:
A>Добрый день, дорогие форумчане!
A>Есть файл со списком из 10 000 записей "ip+port". A>Нужно мониторить состояние подключения по адресу. Реализовать чек "жив ли сервер", фактически. A>И оповещать — выводить в консоль и файл запись "ts + address + up/down". A>Всё это нужно делать быстро.
A>1ая реализация A>epoll + main thread + output A>Придумал, что можно в epoll получать список fd, у которых случилось событие "соединение упало/соединение установлено". A>В моменте выводить информацию. Идти к следующему fd, ... A>Минус реализации, что проходимся в цикле по выданному пулу fd синхронно записывая в консоль, файл. А в это время уже события на fd новые могут придти.
Тогда нужно использовать io_uring, тогда и файлы можно будет писать асихнронно.
Здравствуйте, reversecode, Вы писали:
R>ну тоесть они хотят что бы вы им написали в качестве тестового задания ядро tokio из раста на си или с++ с ворк стилиногом?
Что такое ворк стилиног?
Скорее просто решение epoll + что-нибудь нативное. М.б. пул потоков. Не особо вот как Вы описали что-то мощное. На пару часов.
ну вот видите вы погулите и узнаете
как без него в нетворкинг шагать я не особо понимаю
а чего там они хотели я хз
я вообще если какие то домашки беру на собесах
то не парюсь крутыми решениями
мне за это не платят
и всегда понимаю домашки — как задачу абы как, что бы дальше было о чем позвиздеть с собеседующим при сдаче
так что делайте в лоб на epoll
максимум еще можете какуюто mutex+cv очередь на воркерах накидать
и сойдет
Здравствуйте, avovana, Вы писали:
R>>ну тоесть они хотят что бы вы им написали в качестве тестового задания ядро tokio из раста на си или с++ с ворк стилиногом? A>Что такое ворк стилиног?
Начинать надо с определения "быстро". Что значит "быстро"? Вот, предположим, на машине, где выполняется код, сеть "мигнула", и все соединения сбросились. Что нужно сделать "быстро", вывести 10.000 строк в консоль? А потом еще 10.000 что восстановилось? А, простите, зачем? Человеку это быстро не переварить, а машине из консоли читать неудобно.
Иными словами, в чем смысл очень быстро и неблокирующе обнаруживать up/down для тысяч соединений, если все равно этой информацией никак разумно не воспользоваться? Следует уточнить требования.
Ну и еще. Предположим, вы сделаете два потока, в одном пишете в консоль, в другом получаете события up/down. Где там bottleneck? Имеет ли значение то, что вы обнаружите потерю соединения на 0.1 (или 1, или даже 5 секунд) быстрее? Ведь все равно консоль занята выводом предыдущих сообщений. Или вы хотите правильно timestamp'ы вывести? Тогда да, два потока, в одном только обработка данных epoll (или io_uring, а еще лучше kqueue), и неблокирующая запись в message queue другого потока.
Здравствуйте, SkyDance, Вы писали:
SD>Ну и еще. Предположим, вы сделаете два потока, в одном пишете в консоль, в другом получаете события up/down. Где там bottleneck? Имеет ли значение то, что вы обнаружите потерю соединения на 0.1 (или 1, или даже 5 секунд) быстрее? Ведь все равно консоль занята выводом предыдущих сообщений. Или вы хотите правильно timestamp'ы вывести? Тогда да, два потока, в одном только обработка данных epoll (или io_uring, а еще лучше kqueue), и неблокирующая запись в message queue другого потока.
Вот, да. Спасибо. Тоже думал про баланс — быстрее определить или быстрее выводить.
Здравствуйте, reversecode, Вы писали:
R>так что делайте в лоб на epoll R>максимум еще можете какуюто mutex+cv очередь на воркерах накидать R>и сойдет
Да. Это самое понятное решение. Можно 1 воркер сделать, можно по числу ядер. Тред пул с очередью, mutex, cv, как Вы подметили и я привёл код.
Я сейчас задумался над предложением watchmaker. Он писал про буфер, а не соединяющую поток очередь. Что у каждого потока по буферу. Когда epoll считывающий поток получает набор фд он закидывает их в буфер.
Логгер поток, к примеру, по cv просыпается и всё что делает — swap указатели. Отдаёт указатель на буфер что у него есть(к примеру, такого же размера), забирает указатель на буффер с заполненными данными.
Это кажется быстро и кэш френдли.
Нюанс в том, что не понятно какой этот буфер делать. Если чтобы влезло 5000 fd, то может придти 7000 fd, не влезут в этот буфер. Или же логгер поток еще не записал данные в файл/консоль. Что делать epoll потоку? А ему надо быстро откинуть данные, чтобы epoll'ить снова.
malloc'ом каждый раз не хочется пользоваться. Системными функциями в целом не хочется пользоваться. Чтобы не давать повод ОС к выгрузки процесса.
Что можно с этим буфером сделать? Есть идеи? Какой-то циклический прикрутить? Но там проблема с набеганием указателя писателя.
upd. А может и не страшно, что писатель перезаписывает данные.
Т.е. да. Баланс между:
1. сейчас точно знать состояние.
Или же
2. историю хочется знать.
Наверное, для 1 — цикл буфер. Для 2 — std::queue<Task> tasks_ из сниппета выше.
лучше перечитать тз задачи
а еще лучше если можно уточнять вопросы по задаче у постановщиков
потому что это по факту то что я сказал
написать нетворк фреймворк на коленке, за спасибо и поговорить это не делается
есть разные подходы
epoll
1) в разных воркерах
2) в одном, а результаты через очередь раздаются на других воркеров
итд
есть разные режимы epoll когда можно всеми форкерами ждать в одном epoll
вообщем долго сложно не понятно
когда не ясно чего хочет постановщик задачи
и когда у вас мало знаний о том как где в каких фреймворках используют epoll
что бы завалить собеседующего контр аргументами, мол а в известном нетворк фрейморвке сделано так
а в другом извесном сделано сяк
и вот такие плюсы и минусы
Здравствуйте, reversecode, Вы писали:
R>вообщем долго сложно не понятно R>когда не ясно чего хочет постановщик задачи R>и когда у вас мало знаний о том как где в каких фреймворках используют epoll R>что бы завалить собеседующего контр аргументами, мол а в известном нетворк фрейморвке сделано так R>а в другом извесном сделано сяк R>и вот такие плюсы и минусы
Классно. Спасибо за беседу! Реализую парочку вариантов. Посмотрим на результат
народ на zig реализовывал некоторые варианты
помоему первые два
даже с воркстилингом + uring
смутно помню результаты очень приблизительно сравнимы
даже с go сравнивали
Здравствуйте, reversecode, Вы писали:
R>народ на zig реализовывал некоторые варианты R>помоему первые два R>даже с воркстилингом + uring R>смутно помню результаты очень приблизительно сравнимы R>даже с go сравнивали
R>https://gist.github.com/kprotty/5a41e9612657de00788478a7dde43d78
R>в других ревизиях надо смотреть R>они там разные варианты реализовывали R>сейчас глянул уже оставили только один
Ну вот мы тоже общались по задаче. Что, мол, можно сделать вариант с запуском в баше этой утилиты в несколько экзэмпляров.
Как я понял, входящий список отслеживаемых адресов просто бьётся. К примеру, на 4. И запускаются 4 экземпляра. Что тогда утилиту можно сделать однопоточной.
Но я вижу в этом минус. Как в system design — проблема с популярной личностью с шардами. Один шард нагружен, который обслуживает личность, другие простаивают.
Так и тут одно подмножество адресов, допустим, будет постоянно мигать/много событий. Другие нет. Тогда на этот один из 4ёх экзэмляров утилиты упадёт основная работа.
А утилита однопоточная.
Если же многопоточная:
1 поток eloll_wait и отдаёт через общую очередь таски, как в снипетте. И есть потоки разгребатели. То лучше будет.
Та из собеса минувших дней )
