Как присоединиться к потоку, который висит на блокировке ввода-вывода?

Ответ 1

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

Начиная с ядра Linux 2.6.22, система предлагает новую функцию под названием signalfd(), которая может быть использована для открытия файлового дескриптора для заданного набора сигналов Unix (за исключением тех, которые прямо убивают процесс.)

// defined a set of signals
sigset_t set;
sigemptyset(&set);
sigaddset(&set, SIGUSR1);
// ... you can add more than one ...

// prevent the default signal behavior (very important)
sigprocmask(SIG_BLOCK, &set, nullptr);

// open a file descriptor using that set of Unix signal
f_socket = signalfd(-1, &set, SFD_NONBLOCK | SFD_CLOEXEC);

Теперь вы можете использовать функции poll() или select() для прослушивания сигнала вдоль более обычного дескриптора файла (сокета, файла на диске и т.д.), который вы слушали.

NONBLOCK важен, если вам нужен цикл, который может проверять сигналы и другие файловые дескрипторы снова и снова (т.е. это также важно в вашем другом дескрипторе файла).

У меня есть такая реализация, которая работает с (1) таймерами, (2) сокетами, (3) трубами, (4) сигналами Unix, (5) обычными файлами. На самом деле, действительно любой файловый дескриптор плюс таймеры.

https://github.com/m2osw/snapcpp/blob/master/snapwebsites/libsnapwebsites/src/snapwebsites/snap_communicator.cpp
https://github.com/m2osw/snapcpp/blob/master/snapwebsites/libsnapwebsites/src/snapwebsites/snap_communicator.h

Вас также могут заинтересовать такие библиотеки, как libevent

Ответ 2

Я тоже рекомендовал бы использовать выбранное или какое-то другое не-сигнальное средство для прекращения потока. Одна из причин, по которой у нас есть темы, - попытаться уйти от сигнального безумия. Тем не менее...

Обычно для передачи сигнала в поток используется pthread_kill() с SIGUSR1 или SIGUSR2. Другие предложенные сигналы - SIGTERM, SIGINT, SIGKILL - имеют семантику всей процедуры, которая вас может не заинтересовать.

Что касается поведения при отправке сигнала, я предполагаю, что он связан с тем, как вы обрабатывали сигнал. Если у вас нет обработчика, применяется действие по умолчанию этого сигнала, но в контексте потока, который получил сигнал. Таким образом, SIGALRM, например, будет обрабатываться вашим потоком, но обработка будет заключаться в завершении процесса - возможно, не в желаемом поведении.

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

Возможно, вы читаете какой-то цикл? Если чтение заканчивается с -1 возвратом, то вырваться из этого цикла и выйти из потока.

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

#include <stdlib.h>
#include <stdio.h>
#include <pthread.h>
#include <signal.h>

int global_gotsig = 0;

void *gotsig(int sig, siginfo_t *info, void *ucontext) 
{
        global_gotsig++;
        return NULL;
}

void *reader(void *arg)
{
        char buf[32];
        int i;
        int hdlsig = (int)arg;

        struct sigaction sa;
        sa.sa_handler = NULL;
        sa.sa_sigaction = gotsig;
        sa.sa_flags = SA_SIGINFO;
        sigemptyset(&sa.sa_mask);

        if (sigaction(hdlsig, &sa, NULL) < 0) {
                perror("sigaction");
                return (void *)-1;
        }
        i = read(fileno(stdin), buf, 32);
        if (i < 0) {
                perror("read");
        } else {
                printf("Read %d bytes\n", i);
        }
        return (void *)i;
}

main(int argc, char **argv)
{
        pthread_t tid1;
        void *ret;
        int i;
        int sig = SIGUSR1;

        if (argc == 2) sig = atoi(argv[1]);
        printf("Using sig %d\n", sig);

        if (pthread_create(&tid1, NULL, reader, (void *)sig)) {
                perror("pthread_create");
                exit(1);
        }
        sleep(5);
        printf("killing thread\n");
        pthread_kill(tid1, sig);
        i = pthread_join(tid1, &ret);
        if (i < 0)
                perror("pthread_join");
        else
                printf("thread returned %ld\n", (long)ret);
        printf("Got sig? %d\n", global_gotsig);

}

Ответ 3

Канонический способ сделать это с помощью pthread_cancel, где поток выполнил pthread_cleanup_push/pop, чтобы обеспечить очистку для любых ресурсов, которые он использует.

К сожалению, это НЕ МОЖЕТ использоваться в коде на С++. Любой код С++ std lib или ANY try {} catch() в вызывающем стеке во время pthread_cancel потенциально могут убить весь ваш процесс.

Единственным обходным решением является обработка SIGUSR1, установка флага остановки, pthread_kill(SIGUSR1), а затем где-либо поток блокируется при вводе-выводе, если вы получите EINTR проверить флаг остановки перед повторной попыткой ввода-вывода. На практике это не всегда удается в Linux, не знаю почему.

Но в любом случае бесполезно говорить о том, нужно ли вам звонить какой-либо третьей стороне lib, потому что они, скорее всего, будут иметь жесткий цикл, который просто перезапускает ввод-вывод на EINTR. Обратное проектирование их файлового дескриптора, чтобы закрыть его, также не сократит его - они могут ждать на семафоре или другом ресурсе. В этом случае просто невозможно написать рабочий код, период. Да, это полностью повреждено мозгом. Поговорите с парнями, которые разработали исключения С++ и pthread_cancel. Предположительно это может быть исправлено в некоторой будущей версии С++. Удачи вам в этом.

Ответ 4

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

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

Ответ 5

Зависит от ожидания ожидания ввода-вывода.

Если поток находится в состоянии "Бесперебойное IO" (отображается как "D" сверху), то на самом деле вы абсолютно ничего не можете с этим поделать. Обычно потоки только вводят это состояние, делая что-то вроде ожидания перехода страницы (или требуемой загрузки, например, из файла mmap'd или общей библиотеки и т.д.), Однако сбой (особенно сервер NFS) может вызвать он должен оставаться в этом состоянии дольше.

Поистине нет способа избежать этого состояния "D". Нить не будет реагировать на сигналы (их можно отправить, но они будут поставлены в очередь).

Если это нормальная функция ввода-вывода, такая как read(), write() или функция ожидания, например select() или poll(), сигналы будут доставлены нормально.

Ответ 6

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

Идея заключалась бы в создании файла из основного цикла (или по 1 на поток, как предполагает тайм-аут), это даст вам более тонкий контроль над тем, какие потоки разбужены). Все потоки, которые блокируют ввод/вывод файлов, будут делать select(), используя файлы, которые они пытаются использовать, а также файл, созданный основным контуром (в качестве члена прочитанного набор дескрипторов файлов). Это должно привести к возврату всех вызовов select().

Код для обработки этого "события" из основного цикла должен быть добавлен к каждому из потоков.

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

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

Ответ 7

Думаю, как вы сказали, единственным способом было бы отправить сигнал, затем поймать и обработать его соответствующим образом. Альтернативами могут быть SIGTERM, SIGUSR1, SIGQUIT, SIGHUP, SIGINT и т.д.

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

Ответ 8

Я всегда добавляю функцию "kill", связанную с функцией потока, которую я запускаю до объединения, которая гарантирует, что поток будет соединяться в разумные сроки. Когда поток использует блокировку IO, я пытаюсь использовать систему для разрыва блокировки. Например, при использовании сокета я бы отключил выключение вызова (2) или закрыть (2) на нем, что приведет к тому, что сетевой стек завершит его чисто.

Реализация сокетов Linux является потокобезопасной.

Ответ 9

Я удивлен, что никто не предложил pthread_cancel. Недавно я написал многопоточную программу ввода-вывода и вызвал функцию cancel(), после чего команда join() работала просто отлично.

Я изначально попробовал pthread_kill(), но закончил просто завершение всей программы сигналами, которые я тестировал.

Ответ 10

Если вы блокируете стороннюю библиотеку, которая работает в EINTR, вам может потребоваться комбинация использования pthread_kill с сигналом (USR1 и т.д.), вызывающая пустую функцию (не SIG_IGN) с фактическим закрытием/заменой файловый дескриптор. Используя dup2, чтобы заменить fd на /dev/null или аналогичный, вы заставите стороннюю библиотеку получить результат конца файла при повторном чтении.

Обратите внимание, что с помощью dup() вначале первого сокета вы можете избежать фактического закрытия сокета.

Ответ 11

Сигналы и поток - это тонкая проблема в Linux по разным страницам руководства. Используете ли вы LinuxThreads или NPTL (если вы работаете в Linux)?

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

Вы должны использовать timed select или poll и установить глобальный флаг для завершения потока.

Ответ 12

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

Когда событие i/o запущено, условное обозначение должно быть сообщено.

Основной поток может просто сигнализировать об этом условии, перебирая предикат цикла на false.

что-то вроде:

while (!_finished)
{
    pthread_cond_wait(&cond);
    handleio();
}
cleanup();

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

Ответ 13

struct pollfd pfd;
pfd.fd = socket;
pfd.events = POLLIN | POLLHUP | POLLERR;
pthread_lock(&lock);
while(thread_alive)
{
    int ret = poll(&pfd, 1, 100);
    if(ret == 1)
    {
        //handle IO
    }
    else
    {
         pthread_cond_timedwait(&lock, &cond, 100);
     }
}
pthread_unlock(&lock);

thread_alive - это специфичная для потока переменная, которая может использоваться в сочетании с сигналом для уничтожения потока.

как для раздела ввода-вывода дескриптора, вам нужно убедиться, что вы использовали open с опцией O_NOBLOCK, или если у его сокета есть аналогичный флаг, вы можете установить MSG_NOWAIT??. для других fds im not sure