了解 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 函数工作正常，列表按预期工作。

这里是start_thread和thread_function的代码：

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，FIRST线程将进入并完成所有工作，而其他线程等待并在所有工作完成后进入。

请注意！：我考虑过可能创建第一个线程的选项，直到创建第二个线程，第一个线程已经完成所有工作。这就是为什么我在创建 evrey 线程时使用 usleep(10) 添加“我在 #threadID”检查的原因。他们都进来了，但只有第一个在做所有的工作。 如果我在互斥锁解锁后睡着了，这是输出示例（注意不同的线程 ID）

睡眠时

./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
...

如果我在互斥锁后注释掉usleep，这就是输出（注意相同的线程ID） 不睡觉

  ./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
...

1. 我的第二个问题是关于线程工作的顺序。我的练习要求我不要使用 join 来同步线程（仅在最后使用“释放资源”），而是使用该条件变量。

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

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

Answer 1

我已经深入研究了 glibc（至少在 Linux 和 x86_64 上是 v2.30）如何实现 pthread_mutex_lock() 和 _unlock()。

原来_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->lock: 0 => 解锁，1 => 锁定但没有服务员，2 => 锁定服务员

  'locked-but-no-waiters' 针对没有锁争用且无需在_unlock() 中执行FUTEX_WAKE 的情况进行优化。

• _lock()/_unlock() 函数在库中——它们不在在内核中。

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

• FUTEX_WAIT(2) 是对内核的调用，它将把线程放在与互斥锁关联的挂起队列中，除非mutex->lock != 2。

  内核检查mutex->lock == 2 并将线程原子地添加到队列中。这处理在atomic_xchg(mutex->lock, 2) 之后调用_unlock() 的情况。

• FUTEX_WAKE(1) 也是对内核的调用，futex 手册页告诉我们：

  FUTEX_WAKE（自 Linux 2.6.0 起）

  此操作最多唤醒 'val' 正在等待的等待者...不能保证唤醒哪些等待者（例如，不能保证具有更高调度优先级的等待者优先被优先级较低的服务员唤醒）。

  这里的 'val' 是 1。

  虽然文档说“不能保证唤醒哪些服务员”，但队列似乎至少是 FIFO。

特别注意：

1. _unlock()不将互斥锁传递给FUTEX_WAKE启动的线程。

2. 一旦被唤醒，线程会再次尝试获取锁...

   ...但可能会被任何其他正在运行的线程击败——包括刚刚执行_unlock() 的线程。

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

Answer 2

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

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

您可以尝试以下几件事：(1) 使用pthread_barrier_t 来保持thread_function() 开头的所有线程，直到它们全部运行； (2) 在pthread_mutex_unlock() 之后使用sched_yield(void) 来提示系统运行新的可运行线程。

其次，在任何情况下都不应将“条件变量”视为信号。要让main() 知道所有线程都已完成，您需要计数——这可能是pthread_barrier_t；或者它可以是一个简单的整数，受互斥体保护，带有一个“条件变量”来保持主线程在等待时处于开启状态；或者它可以是一个计数（在main()）和一个信号量（每个线程在退出时发布一次）。

第三，您在main() 中显示pthread_cond_wait(&cv, &lock);。那时main() 必须拥有lock...，这很重要。但是：就目前而言，找到list 为空的第一个 线程将启动cv，并且main() 将继续，即使其他线程仍在运行。尽管一旦main() 重新获取lock，任何仍在运行的线程要么退出要么卡在lock 上。 （一团糟。）

---

一般来说，使用“条件变量”的模板是：

    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);

那么这里发生了什么？

1. main() 正在等待thread_finished_count == 3

2. thread_finished_count 是受thread_lock 互斥锁“保护”的共享变量。

   ...所以它在互斥锁下的thread_function() 中递增。

   ...和main() 也必须在互斥锁下读取它。

3. 如果main() 找到thread_finished_count != 3，它必须等待。

   这样做：pthread_cond_wait(&thread_cv, &thread_lock)，其中：

   • 解锁thread_lock

   • 将线程放在等待线程的thread_cv队列中。

   它会以原子方式。

4. 当thread_function() 执行pthread_cond_signal(&thread_cv) 时，它会唤醒等待线程。

5. main()线程唤醒后，会先重新获取thread_lock...

   ...所以它可以继续重新读取thread_finished_count，看看它现在是否是3。

FWIW：我建议不要在所有线程都加入之后销毁互斥锁​​等。