问题描述
我有多个读取线程和一个写入线程.如果我在读取线程之一上锁定互斥锁并从中发送广播，是否可以保证互斥锁将被在pthread_cond_wait()上等待的写入线程锁定，或者是否有另一个在pthread_mutex_lock()上等待的读取线程将被锁定互斥体?主要问题是pthread_cond_wait()是否优先于pthread_mutex_lock()?
I have a multiple read threads and one write thread. If I lock mutex on one of the read threads and send broadcast from it, is it guaranteed that mutex will be locked by write thread waiting on pthread_cond_wait() or is there a possibility that another read thread that is wainting on pthread_mutex_lock() will lock mutex? Main question is does pthread_cond_wait() have priority over pthread_mutex_lock()?
如果没有，我如何才能实现互斥锁将始终被pthread_cond_broadcast()上的写入线程锁定?
If not, how can I achieve that the mutex will always be locked by write thread on pthread_cond_broadcast()?
示例
读取线程:
pthread_mutex_lock(mutex);
pthread_cond_broadcast(cond);
pthread_mutex_unlock(mutex);
写线程:
pthread_mutex_lock(&mutex);
pthread_cond_wait(&cond, &mutex);
推荐答案
让我们假设两个线程(读写)在同一时刻到达pthread_mutex_lock.因此，写线程在pthread_mutex_lock调用中获取互斥量，或者读线程.
Let's assume both threads, read and write, reach the pthread_mutex_lock in the same moment. So, either write thread acquire the mutex on pthread_mutex_lock call, or read thread.
如果它将是写线程，则读取的线程将在pthread_mutex_lock上等待.通过调用pthread_cond_wait进行写操作会释放mutex并在cond上进行阻止.它是原子完成的.因此，当读取线程是mutex的Grantex时，我们可以确保读取的线程在cond上等待.因此，在cond上的广播到达写入线程，不再在cond上等待，而是-仍在pthread_cond_wait的范围内-试图获得对mutex的锁定(保持为读取线程).广播cond之后，读取线程释放mutex并进入写入线程.因此，写线程最终从锁定了mutex的pthread_cond_wait退出.记得以后再解锁.
If it would be the write thread, the read one will wait on pthread_mutex_lock. The write, by calling pthread_cond_wait releases mutex and blocks on cond. It is done atomically. So, when read thread is grantex the mutex, we can be sure the the read one waits on cond. So, broadcast on cond reaches the write thread, it no more waits on cond but - still in scope of pthread_cond_wait - tries to get a lock on mutex (hold be read thread). After broadcasting cond the read thread releases the mutex and it goes to write thread. So write thread finally exits from pthread_cond_wait having the mutex locked. Remember to unlock it later.
如果它是读线程，则写操作将在pthread_mutex_lock上等待，读操作将在cond上广播信号，然后释放mutex.此后，写线程在pthread_mutex_lock上获取mutex并立即在其中释放pthread_cond_wait以等待cond(请注意，先前的cond广播对当前pthread_cond_wait没有影响).在读取线程的下一个迭代中，它获取对mutex的锁定，在cond上发送广播，并解锁mutex.这意味着写线程在cond上向前移动并在mutex上获得锁定.
If it would be the read thread, the write one will wait on pthread_mutex_lock, the read will broadcast a signal on cond then release the mutex. After then the write thread acquires the mutex on pthread_mutex_lock and immediately releases in it pthread_cond_wait waiting for cond (please note, that previous cond broadcast has no effect on current pthread_cond_wait). In the next iteration of read thread it acquires lock onmutex, send broadcast on cond and unlock mutex. It means the write thread moves forward on cond and acquires lock on mutex.
它能回答您有关优先级的问题吗?
Does it answer your question about priority?
评论后更新.
假设我们有一个线程(将其命名为A以供将来参考)，该线程持有mutex上的锁，而其他线程很少尝试获取相同的锁.一旦第一个线程释放了锁，就无法预测哪个线程将获得锁.此外，如果A线程具有循环并尝试重新获取mutex上的锁，则有可能将其授予该锁，而其他线程将继续等待.添加pthread_cond_wait不会在授予锁的范围内进行任何更改.
Let's assume we have one thread (let's name it A for future reference) holding the lock on mutex and few other trying to acquire the same lock. As soon as the lock is released by first thread, there is no predictable which thread would acquire lock. Moreover, if the A thread has a loop and tries to reacquire lock on mutex, there is a chance it would be granted this lock and other threads would keep waiting. Adding pthread_cond_wait doesn't change anything in scope of granting a lock.
让我引用POSIX规范的片段(请参阅 https://stackoverflow.com/a/9625267/2989411 参考):
Let me quote fragments of POSIX specification (see https://stackoverflow.com/a/9625267/2989411 for reference):
这仅是有关操作顺序的标准保证.将锁授予其他线程的顺序相当难以预测，并且会根据时间的一些非常细微的波动而改变.
And this is only guarantee given by standard regarding order of operations. Order of granting the lock to other threads is rather unpredictable and it changes depending on some very subtle fluctuation in timing.
仅针对互斥锁相关的代码，请使用以下代码进行一些操作:
For only mutex related code, please play a little with following code:
#define _GNU_SOURCE
#include <pthread.h>
#include <stdio.h>
#include <unistd.h>
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
void *th(void *arg) {
int i;
char *s = arg;
for (i = 0; i < 10; ++i) {
pthread_mutex_lock(&mutex);
printf("%s %d\n", s, i);
//sleep(1);
pthread_mutex_unlock(&mutex);
#if 0
pthread_yield();
#endif
}
return NULL;
}
int main() {
int i;
for (i = 0; i < 10; ++i) {
pthread_t t1, t2, t3;
printf("================================\n");
pthread_create(&t1, NULL, th, "t1");
pthread_create(&t2, NULL, th, " t2");
pthread_create(&t3, NULL, th, " t3");
pthread_join(t1, NULL);
pthread_join(t2, NULL);
pthread_join(t3, NULL);
}
return 0;
}
在一台机器(单CPU)上，总是显示从t3开始的整个循环，然后是t2，最后是t1.在另一个(2个内核)上，线程的顺序更为随机，但是几乎总是在为每个线程授予互斥锁之前，它显示每个线程的整个循环.很少有这样的情况:
On one machine (single CPU) it always shows whole loop from t3, then t2 and finally from t1. On another (2 cores) the order of threads is more random, but almost always it shows whole loop for each thread before granting the mutex to other thread. Rarely there is a situation like:
t1 8
t1 9
t3 0
t2 0
t2 1
[removed other t2 output]
t2 8
t2 9
t3 1
t3 2
通过将#if 0替换为#if 1来启用pthread_yield，并观察结果并检查输出.对我来说，它的工作方式是两个线程隔行显示其输出，然后第三个线程终于有机会工作.添加另一个或更多线程.玩耍等.它可以确认随机行为.
Enable pthread_yield by replacing #if 0 with #if 1 and watch results and check output. For me it works in a way two threads display their output interlaced, then third thread finally has a chance to work. Add another or more thread. Play with sleep, etc. It confirms the random behaviour.
如果您想尝试一下，请编译并运行以下代码.这是一个单一生产者的示例-多个消费者模型.它可以用两个参数运行:第一个是使用者线程的数量，第二个是产生的数据序列的长度.如果未提供任何参数，则将有一个使用者线程和120个项目要处理.我还建议在标记为/* play here */的位置使用sleep/usleep:更改参数的值，完全删除睡眠，在适当的时候将其移动到关键部分，或替换为pthread_yield并观察行为的变化.
If you wish to experiment a little, compile and run following piece of code. It's an example of single producer - multiple consumers model. It can be run with two parameters: first is the number of consumer threads, second is the length of produced data series. If no parameters are given there is one consumer thread and 120 items to be processed. I also recommend with sleep/usleep in places marked /* play here */: change the value of arguments, remove the sleep at all, move it - when appropriate - to critical section or replace with pthread_yield and observe changes in behaviour.
#define _GNU_SOURCE
#include <assert.h>
#include <limits.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <unistd.h>
struct data_t {
int seq;
int payload;
struct data_t *next;
};
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
struct data_t *first = NULL, *last = NULL;
int in_progress = 1;
int num_data = 120;
void push(int seq, int payload) {
struct data_t *e;
e = malloc(sizeof(struct data_t));
e->seq = seq;
e->payload = payload;
e->next = NULL;
if (last == NULL) {
assert(first == NULL);
first = last = e;
} else {
last->next = e;
last = e;
}
}
struct data_t pop() {
struct data_t res = {0};
if (first == NULL) {
res.seq = -1;
} else {
res.seq = first->seq;
res.payload = first->payload;
first = first->next;
if (first == NULL) {
last = NULL;
}
}
return res;
}
void *producer(void *arg __attribute__((unused))) {
int i;
printf("producer created\n");
for (i = 0; i < num_data; ++i) {
int val;
sleep(1); /* play here */
pthread_mutex_lock(&mutex);
val = rand() / (INT_MAX / 1000);
push(i, val);
pthread_mutex_unlock(&mutex);
pthread_cond_signal(&cond);
printf("prod %3d %3d signaled\n", i, val);
}
in_progress = 0;
printf("prod end\n");
pthread_cond_broadcast(&cond);
printf("prod end signaled\n");
return NULL;
}
void *consumer(void *arg) {
char c_id[1024];
int t_id = *(int *)arg;
sprintf(c_id, "%*s c %02d", t_id % 10, "", t_id);
printf("%s created\n", c_id);
while (1) {
struct data_t item;
pthread_mutex_lock(&mutex);
item = pop();
while (item.seq == -1 && in_progress) {
printf("%s waits for data\n", c_id);
pthread_cond_wait(&cond, &mutex);
printf("%s got signal\n", c_id);
item = pop();
}
if (!in_progress && item.seq == -1) {
printf("%s detected end of data.\n", c_id);
pthread_mutex_unlock(&mutex);
break;
}
pthread_mutex_unlock(&mutex);
printf("%s processing %3d %3d\n", c_id, item.seq, item.payload);
sleep(item.payload % 10); /* play here */
printf("%s processed %3d %3d\n", c_id, item.seq, item.payload);
}
printf("%s end\n", c_id);
return NULL;
}
int main(int argc, char *argv[]) {
int num_cons = 1;
pthread_t t_prod;
pthread_t *t_cons;
int i;
int *nums;
if (argc > 1) {
num_cons = atoi(argv[1]);
if (num_cons == 0) {
num_cons = 1;
}
if (num_cons > 99) {
num_cons = 99;
}
}
if (argc > 2) {
num_data = atoi(argv[2]);
if (num_data < 10) {
num_data = 10;
}
if (num_data > 600) {
num_data = 600;
}
}
printf("Spawning %d consumer%s for %d items.\n", num_cons, num_cons == 1 ? "" : "s", num_data);
t_cons = malloc(sizeof(pthread_t) * num_cons);
nums = malloc(sizeof(int) * num_cons);
if (!t_cons || !nums) {
printf("Out of memory!\n");
exit(1);
}
srand(time(NULL));
pthread_create(&t_prod, NULL, producer, NULL);
for (i = 0; i < num_cons; ++i) {
nums[i] = i + 1;
usleep(100000); /* play here */
pthread_create(t_cons + i, NULL, consumer, nums + i);
}
pthread_join(t_prod, NULL);
for (i = 0; i < num_cons; ++i) {
pthread_join(t_cons[i], NULL);
}
free(nums);
free(t_cons);
return 0;
}
希望我能消除您的疑虑，并给您一些代码进行实验，并对pthread行为有所信心.
I hope I have cleared your doubts and gave you some code to experiment and gain some confidence about pthread behaviour.
这篇关于pthread_cond_wait和pthread_mutex_lock优先级?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！