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

У меня есть рабочий поток, который слушает сокет 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, который намного dummier, чем настоящий UDP-сокет, и на самом деле не требует выделения каких-либо ограниченных ресурсов (за пределами разумного объема памяти). Я полагаю, что сокеты домена Unix будут делать именно это, но: решение не должно быть гораздо менее кросс-платформенным, чем то, что у меня есть сейчас, хотя некоторые умеренные количество вещей #ifdef в порядке. (Я ориентируюсь главным образом на Windows и Linux - и, кстати, написание С++.)

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

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

Ответ 1

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

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

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

Ответ 2

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

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

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

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

Ответ 3

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

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