pthread_join - ожидание нескольких потоков

Question

Используя потоки POSIX и C ++, у меня есть «Операция вставки», которую можно безопасно выполнять только по одному за раз.

Если у меня несколько потоков, ожидающих вставки с использованием pthread_join, тогда создается новый поток когда это закончится. Получат ли они все сигнал «Завершение потока» сразу и породят несколько вставок, или можно предположить, что поток, который сначала получает сигнал «Завершение потока», будет порождать новый поток, блокируя создание других потоков другими.

/* --- GLOBAL --- */
pthread_t insertThread;



/* --- DIFFERENT THREADS --- */
// Wait for Current insert to finish
pthread_join(insertThread, NULL); 

// Done start a new one
pthread_create(&insertThread, NULL, Insert, Data);

---

Спасибо за ответы

Программа представляет собой огромную хеш-таблицу, которая принимает запросы от клиентов через сокеты.

Каждое новое клиентское соединение порождает новый поток, из которого он может затем выполнять несколько операций, в частности поиск или вставку. поиск может проводиться параллельно. Но вставки должны быть «объединены» в один поток. Вы можете сказать, что операции поиска могут выполняться без создания нового потока для клиента, однако они могут занять некоторое время, заставляя сервер блокироваться, отбрасывая новые запросы. Дизайн старается свести к минимуму системные вызовы и создание потоков в максимально возможной степени.

Но теперь, когда я знаю, что это небезопасно, я впервые подумал, что смогу что-то сделать вместе

Спасибо

Answer 1

С opengroup.org на pthread_join :

Результаты нескольких одновременных вызовов pthread_join (), указывающих один и тот же целевой поток, не определены.

Таким образом, у вас не должно быть нескольких потоков, присоединяющихся к вашему предыдущему insertThread.

Во-первых, когда вы используете C ++, я рекомендую boost.thread . Они напоминают модель потоков POSIX, а также работают на Windows. И это помогает вам в C ++, то есть облегчает использование функциональных объектов.

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

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

Если вы действительно хотите сохранить его более или менее таким же образом, как сейчас, вы должны сделать это:

• Создайте условную переменную, например insert_finished.
• Все потоки, которые хотят выполнить вставку, ждут переменную условия.
• Как только один поток завершает свою вставку, он запускает переменную условия.
• Поскольку условная переменная требует мьютекса, вы можете просто уведомить все ожидающие потоки, все они хотят начать вставку, но поскольку только один поток может получить мьютекс одновременно, все потоки будут выполнять вставку последовательно.

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

Answer 2

Согласно спецификации Unix: «Результаты нескольких одновременных вызовов pthread_join (), указывающих один и тот же целевой поток, не определены».

«Нормальным способом» достижения единственного потока для получения задачи было бы установить переменную условия (не забудьте соответствующий мьютекс): незанятые потоки ждут в pthread_cond_wait () (или pthread_cond_timedwait ()), и когда поток, выполняющий работу, закончен, он вызывает один из бездействующих с помощью pthread_cond_signal ().

Answer 3

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

    pthread_t       insertThread = NULL;
    pthread_mutex_t insertConditionNewMutex = PTHREAD_MUTEX_INITIALIZER;
    pthread_mutex_t insertConditionDoneMutex    = PTHREAD_MUTEX_INITIALIZER;
    pthread_cond_t  insertConditionNew      = PTHREAD_COND_INITIALIZER;
    pthread_cond_t  insertConditionDone     = PTHREAD_COND_INITIALIZER;

       //Thread for new incoming connection
        void * newBatchInsert()
        {
           for(each Word)
           {
                            //Push It into the queue
                            pthread_mutex_lock(&lexicon[newPendingWord->length - 1]->insertQueueMutex);
                                lexicon[newPendingWord->length - 1]->insertQueue.push(newPendingWord);
                            pthread_mutex_unlock(&lexicon[newPendingWord->length - 1]->insertQueueMutex);

           }

                    //Send signal to worker Thread
                    pthread_mutex_lock(&insertConditionNewMutex);
                        pthread_cond_signal(&insertConditionNew);
                    pthread_mutex_unlock(&insertConditionNewMutex);

                    //Wait Until it's finished
                    pthread_cond_wait(&insertConditionDone, &insertConditionDoneMutex);

        }


            //Worker thread
            void * insertWorker(void *)
            {

                while(1)        
                {

                    pthread_cond_wait(&insertConditionNew, &insertConditionNewMutex);

                    for (int ii = 0; ii < maxWordLength; ++ii)
                    {                   
                            while (!lexicon[ii]->insertQueue.empty())
                            {

                                queueNode * newPendingWord = lexicon[ii]->insertQueue.front();


                                lexicon[ii]->insert(newPendingWord->word);

                                pthread_mutex_lock(&lexicon[ii]->insertQueueMutex);
                                lexicon[ii]->insertQueue.pop();
                                pthread_mutex_unlock(&lexicon[ii]->insertQueueMutex);

                            }

                    }

                    //Send signal that it's done
                    pthread_mutex_lock(&insertConditionDoneMutex);
                        pthread_cond_broadcast(&insertConditionDone);
                    pthread_mutex_unlock(&insertConditionDoneMutex);

                }

            }

            int main(int argc, char * const argv[]) 
            {

                pthread_create(&insertThread, NULL, &insertWorker, NULL);


                lexiconServer = new server(serverPort, (void *) newBatchInsert);

                return 0;
            }

Answer 4

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

Конечно, для пула потоков C ++, известного как ThreadFactory, идея состоит в том, чтобы поддерживать абстрактные примитивы потока, чтобы он мог обрабатывать любой из переданных ему типов функций / операций.

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

РЕЗЮМЕ: ИЗБЕГАЙТЕ PTHREAD_JOIN В любом месте, кроме основного потока.

Answer 5

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

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

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

Answer 6

Мне кажется, что вы хотите сериализовать вставки в хеш-таблицу.

Для этого вам нужен замок, а не новые темы.

Answer 7

Единственная найденная мной библиотека, которая поддерживает вставки без блокировки новых поисков - Sunrise DD (и я не уверен, поддерживает ли она одновременные вставки)

Однако переход с карты Google Sparse Hash более чем вдвое увеличивает использование памяти. Поиск должен происходить довольно редко, а не пытаться написать свою собственную библиотеку который сочетает в себе преимущества обоих, я бы предпочел просто заблокировать таблицу, приостанавливая поиск, пока изменения сделаны безопасно.

Еще раз спасибо

Answer 8

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

Поскольку точные результаты поиска недетерминированы, когда вы вставляете элементы одновременно, такая параллельная хеш-карта может быть именно тем, что вам нужно. Я не использовал параллельные хеш-таблицы в C ++, но, поскольку они доступны в Java, вы наверняка найдете библиотеку, делающую это в C ++.

Answer 9

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

Если вам нужно, чтобы код запускался в определенном потоке (например, Java AWT), тогда вам нужны условные переменные. Однако вам следует дважды подумать, окупается ли это решение. Представьте, сколько переключений контекста вам нужно, если вы вызываете свою «операцию вставки» 10000 раз в секунду.