Блог

Я работаю над созданием многопоточного файлового сервера. Я создал пул потоков для обработки запросов от клиента. Вот мой код для пула потоков:

 void *worker_call(void *thread_id){
    printf("Initialized thread #%ld n", (long)thread_id);  

    //pull in initialized global configs
    extern pthread_mutex_t queue_m;
    extern pthread_cond_t worker_c;
    extern steque_t queue; 


    while(1){
        if (pthread_mutex_lock(amp;queue_m) != 0){
            fprintf(stderr, "An error occured while locking mutex in #%ld n", (long)thread_id);
        }
            while(steque_isempty(amp;queue) == 1){
                printf("thread #%ld - Going to sleep...n", (long) thread_id);
                pthread_cond_wait(amp;worker_c, amp;queue_m);
                printf("thread #%ld - I'm waking up...n", (long) thread_id);
            }
            
            int *work = steque_pop(amp;queue);
            

        if (pthread_mutex_unlock(amp;queue_m) != 0){
            fprintf(stderr, "An error occured while unlocking mutex in #%ld n", (long)thread_id);
        }

        pthread_cond_broadcast(amp;worker_c);

        sleep(1); //added to make sure that the other threads have a chance to wake up

        printf("thread #%ld - what is the value of work: %dn", (long) thread_id, *work);
        // process_request(amp;(work->ctx), work->path, work->arg, (long)thread_id);
        free(work);
    }

    return NULL;
}
  

Чтобы проверить, что мой пул потоков запускается правильно и все потоки приступают к работе, я создал следующий тест:

 int main(){
    int nthreads = 6;

    pthread_t threads[nthreads];
    long thread_ids[nthreads];

    pthread_attr_t thread_attr;
    pthread_attr_init(amp;thread_attr);
    pthread_attr_setdetachstate(amp;thread_attr, PTHREAD_CREATE_JOINABLE);
    pthread_attr_setscope(amp;thread_attr, PTHREAD_SCOPE_SYSTEM);


    for(int t=0;t<nthreads;t  ){
        thread_ids[t] = t;
        if( pthread_create(amp;threads[t],NULL, worker_call, (void *)thread_ids[t]) != 0 ) {
            printf("An error occured while creating thread: %dn", t);
        }
    }

    pthread_attr_destroy(amp;thread_attr);

    //init multi-threading configs
    extern pthread_mutex_t queue_m;

    pthread_mutexattr_t m_attr;
    pthread_mutexattr_init(amp;m_attr);
    pthread_mutexattr_settype(amp;m_attr, PTHREAD_MUTEX_ERRORCHECK);

    pthread_mutex_init(amp;queue_m, NULL);

    extern pthread_cond_t worker_c;
    pthread_cond_init(amp;worker_c, NULL);

    extern steque_t queue; 

    steque_init(amp;queue);
    //create a simple queue with each item being an int. 
    //the goal is to simply remove the items from the queue.
    for(int i = 0; i < 5; i  ){
        
        int *work = malloc(sizeof(int));
        *work = i;
        steque_enqueue(amp;queue, work);
    }
    

    printf("queue size after creating it: %dn", steque_size(amp;queue));

    pthread_cond_broadcast(amp;worker_c);

    for(int t=0;t<nthreads;t  ){
        pthread_join(threads[t], NULL);
    }

    printf("All the threads finished processingn");

    return 0;
}
  

Теперь этот код выполняется успешно, однако только один поток просыпается и выполняет всю работу, как показано в выводе ниже:

 Initialized thread #0 
thread #0 - Going to sleep...
Initialized thread #1 
thread #1 - Going to sleep...
Initialized thread #2 
thread #2 - Going to sleep...
Initialized thread #3 
thread #3 - Going to sleep...
Initialized thread #4 
thread #4 - Going to sleep...
Initialized thread #5 
thread #5 - Going to sleep...
queue size after creating it: 5
thread #5 - I'm waking up...
thread #5 - what is the value of work: 0
thread #5 - what is the value of work: 1
thread #5 - what is the value of work: 2
thread #5 - what is the value of work: 3
thread #5 - what is the value of work: 4
thread #5 - Going to sleep...

  

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

Я не могу опубликовать простой комментарий, потому что у меня нет 50 повторений. Поэтому я должен опубликовать это как ответ.

Каково ваше определение успеха, когда вы говорите «успешно запущен»? Здесь есть несколько условий гонки, которые вызывают тревогу, и я удивлен, что это вообще работает. Например, worker_call() обращается к переменным queue , mutex и cond до их фактической инициализации.

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

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

Редактировать 1:

Кроме того, в worker_call рабочий проверяет пустоту очереди перед вызовом cond_wait, но он не проверяет пустоту после пробуждения. Поскольку возможно, что элементов меньше, чем рабочих, этой функции потребуется снова проверить пустоту очереди.