Потоки бегут порциями

Question

У меня есть список с 1000-10000 рабочих мест. Он хранится в простом массиве символов. Мне нужно закончить работу над списком как можно быстрее.

Я хочу запустить 10 потоков в любое время. Как мне запустить новый поток сразу (или почти сразу) после завершения предыдущего потока?

#define THREADS_LIMIT 10
const char * const jobs[]= { "data1", .... }
...
for (i = 0; i < THREADS_LIMIT; ++i)
{
// run first 10 threads
}

Как мне начать обсуждение после того, как предыдущий будет сделан? Я могу запустить 10 потоков, затем подождать, пока ВСЕ не будет сделано, и затем запустить следующие 10. Но есть ли более эффективный способ?

Answer 1

Не останавливайте никакие темы.Просто попросите каждый поток перейти к следующему заданию, а не разрушать его и создавать новое;они заканчиваются, когда больше не остается работы.Теперь ваша основная процедура просто запускает потоки THREAD_LIMIT, а затем pthread_join s их всех.

Answer 2

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

#define THREADS_LIMIT 10

static pthread_mutex_t job_mutex = PTHREAD_MUTEX_INITIALIZER;
static const char * const jobs[]= { "data1", .... };

char const* get_job(void) 
{
    pthread_mutex_lock( &job_mutex);

    static char const* next_job = jobs;
    static char const* jobs_end = jobs + (sizeof(jobs)/sizeof(jobs[0]));

    char const* result = NULL;

    if (next_job != jobs_end) {
        result = next_job++;
    }

    pthread_mutex_unlock( &job_mutex);

    return result;
}



...
for (i = 0; i < THREADS_LIMIT; ++i)
{
// run first 10 threads
}


void* thread_fun( void* arg)
{
    for (char const* my_job = get_job(); my_job != NULL; my_job = get_job()) {
        // work on the job *my_job
    }

    return 0;
}

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

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

Answer 3

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

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

Затем, в конце этих 10000 заданий, вы помещаете 10 заданий завершения, которые приведут к выходу рабочего потока, а затем ожидаете их.

Вся схема выглядит так:

main thread:
    create a queue
    do ten times:
        start a worker thread
    do ten thousand times:
        add normal job to queue
    do ten times:
        add shutdown job to queue
    wait for all workers to finish
    exit

worker thread:
    while true:
        get job from queue
        if job is shutdown job:
            exit
        process job

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

---

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

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

Answer 4

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

Основная проблема проектирования заключается в том, как вы распределяете задачи по пулу, такие вопросы могут бытьупрощается благодаря пакету промежуточного программного обеспечения, например 0mq , для выполнения IPC без блокировки между планировщиком и рабочими.

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