Как подать сигнал select () для немедленного возврата?

Question

У меня есть рабочий поток, который прослушивает сокет TCP для входящего трафика и буферизирует полученные данные для доступа основного потока (назовем этот сокет A ). Однако рабочий поток также должен выполнять некоторые регулярные операции (скажем, раз в секунду), даже если данные не поступают. Поэтому я использую select() с таймаутом, поэтому мне не нужно продолжать опрос , (Обратите внимание, что вызов receive() в неблокирующем сокете и последующий спящий режим на секунду не годится: входящие данные должны быть немедленно доступны для основного потока, даже если основной поток не всегда может сразу его обработать отсюда и необходимость буферизации.)

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

При запуске программы рабочий поток создает для этой цели дополнительный сокет типа дейтаграммы (UDP) и привязывает его к некоторому случайному порту (назовем этот сокет B ). Аналогично, основной поток создает сокет дейтаграммы для отправки. При вызове select() рабочий поток теперь перечисляет как A , так и B в fd_set. Когда основной поток должен подать сигнал, он sendto() передает пару байтов соответствующему порту на localhost. Возвращаясь в рабочий поток, если B остается в fd_set после возврата select(), то вызывается recvfrom() и полученные байты просто игнорируются.

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

В идеале я хотел бы вызвать некоторую функцию, которая принимает A в качестве входных данных и ничего не делает, кроме того, что select() возвращает сразу. Однако я не знаю такой функции. (Думаю, я мог бы, например, shutdown() сокет, но побочные эффекты не совсем приемлемы:)

Если это невозможно, вторым лучшим вариантом будет создание B , который намного громче, чем настоящий сокет UDP, и на самом деле не требует выделения каких-либо ограниченных ресурсов (помимо разумного количества объем памяти). Я предполагаю, что доменные сокеты Unix сделали бы именно это, но: решение не должно быть намного менее кроссплатформенным, чем у меня в настоящее время, хотя некоторое умеренное количество #ifdef вещей хорошо. (Я ориентируюсь в основном на Windows и Linux - и, кстати, пишу на C ++.)

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

Теперь, когда я писал это, я понял, что кто-то определенно собирается ответить просто " Boost.Asio ", поэтому я только что впервые посмотрел на него ... Не смог найти очевидного решение, хотя. Обратите внимание, что я также не могу (легко) повлиять на то, как создается сокет A , но я должен иметь возможность позволить другим объектам оборачивать его, если это необходимо.

Answer 1

Вы почти у цели. Используйте трюк "self-pipe" . Откройте канал, добавьте его в select() прочитайте и напишите fd_set, запишите в него из основного потока, чтобы разблокировать рабочий поток. Это портативно через системы POSIX.

Я видел вариант аналогичной техники для Windows в одной системе (фактически используется вместе с методом выше, отделенным #ifdef WIN32). Разблокировка может быть достигнута путем добавления фиктивного (несвязанного) сокета дейтаграммы в fd_set и последующего его закрытия. Недостатком является то, что вы, конечно, должны открывать его каждый раз.

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

Answer 2

Использование конвейера вместо сокета немного чище, так как другой процесс не может его схватить и все испортить.

Использование сокета UDP определенно создает вероятность попадания паразитных пакетов и создания помех.

Анонимный канал никогда не будет доступен никакому другому процессу (если вы его не отдадите).

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

Answer 3

На Unix это будет просто с использованием трубы. Если вы работаете в Windows и хотите продолжать использовать оператор select для обеспечения совместимости вашего кода с Unix, уловка для создания несвязанного сокета UDP и его закрытия работает хорошо и легко. Но вы должны сделать его многопоточным.

Единственный способ сделать этот многопоточный безопасный способ - закрыть и воссоздать сокет в том же потоке, в котором выполняется оператор select. Конечно, это сложно, если поток блокирует выбор. И тогда в окнах приходит вызов QueueUserAPC. Когда в операторе select блокируется windows, поток может обрабатывать асинхронные вызовы процедур. Вы можете запланировать это из другого потока, используя QueueUserAPC. Windows прерывает процесс выбора, выполняет вашу функцию в том же потоке и продолжает выполнение оператора выбора. Теперь вы можете в своем методе APC закрыть сокет и воссоздать его. Гарантированный потокобезопасность, и вы никогда не потеряете сигнал.