理解pthread锁和条件变量

Question

我在C中做了一个关于线程、锁和条件变量的练习。我需要编写一个程序来获取数据，将其转换为一个链表，启动3个线程，每个线程为列表中的每个节点计算结果，主线程在evreyone完成后打印结果。

这是主要功能：

int thread_finished_count;

// Lock and Conditional variable
pthread_mutex_t list_lock;
pthread_mutex_t thread_lock;
pthread_cond_t thread_cv;

int main(int argc, char const *argv[])
{
    node *list;
    int pairs_count, status;
    thread_finished_count = 0;

    /* get the data and start the threads */
    node *head = create_numbers(argc, argv, &pairs_count);
    list = head; // backup head for results
    pthread_t *threads = start_threads(&list);

    /* wait for threads and destroy lock */
    status = pthread_cond_wait(&thread_cv, &list_lock);
    chcek_status(status);
    status = pthread_mutex_destroy(&list_lock);
    chcek_status(status);
    status = pthread_mutex_destroy(&thread_lock);
    chcek_status(status);

    /* print result in original list */
    print_results(head);

    /* cleanup */
    wait_for_threads(threads, NUM_THREADS);
    free_list(head);
    free(threads);

    return EXIT_SUCCESS;
}

请注意，create_numbers功能正常工作，列表按预期工作。

以下是启动线程和线程功能代码：

pthread_t *start_threads(node **list)
{
    int status;
    pthread_t *threads = (pthread_t *)malloc(sizeof(pthread_t) * NUM_THREADS);
    check_malloc(threads);

    for (int i = 0; i < NUM_THREADS; i++)
    {
        status = pthread_create(&threads[i], NULL, thread_function, list);
        chcek_status(status);
    }
    return threads;
}

void *thread_function(node **list)
{
    int status, self_id = pthread_self();
    printf("im in %u\n", self_id);
    node *currentNode;

    while (1)
    {
        if (!(*list))
            break;
        status = pthread_mutex_lock(&list_lock);
        chcek_status(status);
        printf("list location %p thread %u\n", *list, self_id);
        if (!(*list))
        {
            status = pthread_mutex_unlock(&list_lock);
            chcek_status(status);
            break;
        }
        currentNode = (*list);
        (*list) = (*list)->next;
        status = pthread_mutex_unlock(&list_lock);
        chcek_status(status);
        currentNode->gcd = gcd(currentNode->num1, currentNode->num2);
        status = usleep(10);
        chcek_status(status);
    }
    status = pthread_mutex_lock(&thread_lock);
    chcek_status(status);
    thread_finished_count++;
    status = pthread_mutex_unlock(&thread_lock);
    chcek_status(status);
    if (thread_finished_count != 3)
        return NULL;
    status = pthread_cond_signal(&thread_cv);
    chcek_status(status);
    return NULL;
}
void chcek_status(int status)
{
    if (status != 0)
    {
        fputs("pthread_function() error\n", stderr);
        exit(EXIT_FAILURE);
    }
}

请注意，self_id用于调试目的。

1. 我的主要问题是分工合作。每个线程从全局链表中获取一个元素，计算gcd，然后继续获取下一个元素。只有当我在while循环中解锁互斥锁后添加usleep（10）时，我才能获得这种效果。如果我不添加usleep，第一个线程将进入并完成所有工作，而另一个线程只是等待并在所有工作完成后进入

请注意！：我考虑了可能创建第一个线程的选项，在创建第二个线程之前，第一个线程已经完成了所有作业。这就是为什么在创建evrey线程时，我在usleep（10）中添加了“我在#线程ID”检查。他们都进来了，但只有第一个在做所有的工作。这里是一个输出示例，如果我在互斥锁解锁后执行usleep（注意不同的线程ID）

与usleep合作

./v2 nums.txt
im in 1333593856
list location 0x7fffc4fb56a0 thread 1333593856
im in 1316685568
im in 1325139712
list location 0x7fffc4fb56c0 thread 1333593856
list location 0x7fffc4fb56e0 thread 1316685568
list location 0x7fffc4fb5700 thread 1325139712
list location 0x7fffc4fb5720 thread 1333593856
list location 0x7fffc4fb5740 thread 1316685568
list location 0x7fffc4fb5760 thread 1325139712
list location 0x7fffc4fb5780 thread 1333593856
list location 0x7fffc4fb57a0 thread 1316685568
list location 0x7fffc4fb57c0 thread 1325139712
list location 0x7fffc4fb57e0 thread 1333593856
list location 0x7fffc4fb5800 thread 1316685568
list location (nil) thread 1325139712
list location (nil) thread 1333593856
...
normal result output
...

如果我在互斥锁后注释掉uslip（注意相同的线程ID），而没有uslip，这就是输出

  ./v2 nums.txt
im in 2631730944
list location 0x7fffe5b946a0 thread 2631730944
list location 0x7fffe5b946c0 thread 2631730944
list location 0x7fffe5b946e0 thread 2631730944
list location 0x7fffe5b94700 thread 2631730944
list location 0x7fffe5b94720 thread 2631730944
list location 0x7fffe5b94740 thread 2631730944
list location 0x7fffe5b94760 thread 2631730944
list location 0x7fffe5b94780 thread 2631730944
list location 0x7fffe5b947a0 thread 2631730944
list location 0x7fffe5b947c0 thread 2631730944
list location 0x7fffe5b947e0 thread 2631730944
list location 0x7fffe5b94800 thread 2631730944
im in 2623276800
im in 2614822656
...
normal result output
...

我的目标是每个线程将采取元素，做计算，同时另一个线程将进入并采取另一个元素，新的线程将采取每个元素（或至少接近）

感谢您的阅读，感谢您的帮助。

Answer 1

我更深入地研究了glibc（Linux上的v2.30）是如何

原来\u lock（）的工作原理如下：

  if (atomic_cmp_xchg(mutex->lock, 0, 1))
    return <OK> ;             // mutex->lock was 0, is now 1

  while (1)
    {
      if (atomic_xchg(mutex->lock, 2) == 0)
        return <OK> ;        // mutex->lock was 0, is now 2

      ...do FUTEX_WAIT(2)... // suspend thread iff mutex->lock == 2...
    } ;

和_unlock（）的工作原理如下：

  if (atomic_xchg(mutex->lock, 0) == 2)  // set mutex->lock == 0
    ...do FUTEX_WAKE(1)...               // if may have waiter(s) start 1

现在：

>

mutex-

“锁定但无服务员”针对无锁争用且无需在\u unlock（）中执行FUTEX\u WAKE的情况进行优化。

\u lock（）/\u unlock（）函数在库中--它们不在内核中。

...特别是，互斥锁的所有权是库的问题，而不是内核的问题。

FUTEX\u WAIT（2）是对内核的调用，它将线程置于与互斥体相关联的挂起队列中，除非mutex-

内核检查互斥体-

FUTEX\u WAKE（1）也是对内核的调用，FUTEX手册页告诉我们：

FUTEX_WAKE（自Linux 2.6.0起）

此操作最多唤醒正在等待的服务员的“val”。。。没有提供关于哪些服务员被叫醒的保证（例如，与优先级较低的服务员相比，不保证优先级较高的服务员被叫醒）。

其中，本例中的“val”为1。

尽管文件上说“不保证哪位服务员被叫醒”，但排队的似乎至少是先进先出（FIFO）。

特别要注意：

>

_unlock（）不会将互斥锁传递给由FUTEX_WAKE启动的线程。

一旦唤醒，线程将再次尝试获取锁。。。

...但可能会被任何其他正在运行的线程打败，包括刚刚执行\u unlock（）的线程。

我相信这就是为什么你没有看到工作跨线程共享。每个线程要做的工作很少，线程可以解锁互斥锁，完成工作，然后在被解锁唤醒的线程开始工作并成功锁定互斥锁之前再次锁定互斥锁。

Answer 2

首先，您在按住锁的同时执行gcd（）工作。。。所以（a）在任何时候只有一个线程会做任何工作，尽管（b）这并不能完全解释为什么只有一个线程会做（几乎）所有的工作——正如KamilCuk所说，可能是因为要做的工作太少了，以至于（几乎）在第二个线程正常唤醒之前就已经完成了。[更奇怪的是，线程“a”解锁互斥锁和另一个线程开始运行之间可能存在一些延迟，这样线程“a”可以在另一个线程到达互斥锁之前获取互斥锁。]

POSIX说，当一个互斥锁被解锁时，如果有等待者，那么“调度策略将决定哪个线程将获取互斥锁”。默认的“调度策略”是（据我所知）定义的实现。

您可以尝试以下几种方法：（1）使用pthread\u屏障\u t将所有线程保持在thread\u function（）的开头，直到它们都在运行；（2） 在pthread\u mutex\u unlock（）之后使用sched\u yield（void）提示系统运行新运行的线程。

其次，你不应该在任何情况下把条件变量当作信号。如果main（）要知道所有线程都完成了，你需要一个计数--它可以是一个pthread_barrier_t；或者它可以是一个简单的整数，由一个互斥锁保护，带有一个“条件变量”来保持主线程当它等待时；或者它可以是一个计数（在main（）中）和一个信号量（在退出时由每个线程发布一次）。

第三，显示pthread\u cond\u wait(

通常，使用“条件变量”的模板为：

    pthread_mutex_lock(&...lock) ;

    while (!(... thing we need ...))
      pthread_cond_wait(&...cond_var, &...lock) ;

    ... do stuff now we have what we need ....

    pthread_mutex_unlock(&...lock) ;

注意：“条件变量”没有值。。。尽管名称不同，但它并不是一个标志，表示某些条件为真。“条件变量”本质上是等待重新启动的线程队列。当发出“条件变量”信号时，至少会重新启动一个等待的线程——但如果没有线程等待，则不会发生任何事情，特别是（所谓的）“条件变量”不保留信号的内存。

在新的代码中，按照上面的模板，main（）应该：

    /* wait for threads .... */

    status = pthread_mutex_lock(&thread_lock);
    chcek_status(status);

    while (thread_finished_count != 3)
      {
        pthread_cond_wait(&thread_cv, &thread_lock) ;
        chcek_status(status);
      } ;

    status = pthread_mutex_unlock(&thread_lock) ;
    chcek_status(status);

这是怎么回事？

>

main（）正在等待thread_finished_count==3

thread_finished_count是受thread_lock互斥锁保护的共享变量。

...因此，它在互斥锁下的thread\u function（）中递增。

...和main（）也必须在互斥锁下读取它。

如果main（）找到thread_finished_count！=3它必须等待。

为此，它会执行：pthread\u cond\u wait(

>

Thread锁止代码

将线程放置在等待线程的thread\u cv队列中。

它是以原子的方式进行的。

当thread\u function（）执行pthread\u cond\u信号时(

当main（）线程唤醒时，它将首先重新获取线程锁。。。

...因此，它可以继续重新读取线程\u finished\u count，查看它现在是否为3。

FWIW：我建议在加入所有线程之前不要破坏互斥体等。