一个线程对另一个线程(消费者/生产者)表现出兴趣 [英] One thread showing interest in another thread (consumer / producer)
问题描述
我想必须让线程(消费者)对另一个线程(生产者)制造东西表示兴趣.但并非一直如此.
I would like to have to possibility to make thread (consumer) express interest in when another thread (producer) makes something. But not all the time.
基本上，我想一次性消费.理想情况下，生产者会很高兴地从事其业务，直到一个(或多个)消费者发出他们想要的东西的信号，在这种情况下，生产者会将一些数据推送到变量中并表示已经这样做了.使用者将等待直到变量变满为止.
Basically I want to make a one-shot consumer. Ideally the producer through would go merrily about its business until one (or many) consumers signal that they want something, in which case the producer would push some data into a variable and signal that it has done so. The consumer will wait until the variable has become filled.
还必须如此，以便一次性使用的消费者可以确定它已经等待了太长时间而放弃了等待(la pthread_cond_timedwait)
It must also be so that the one-shot consumer can decide that it has waited too long and abandon the wait (a la pthread_cond_timedwait)
我已经阅读了许多文章以及关于如何以不同方式同步线程的问题.目前，我倾向于使用条件变量方法.
I've been reading many articles and SO questions about different ways to synchronize threads. Currently I'm leaning towards a condition variable approach.
我想知道这是否是一个很好的解决方法(作为线程编程的新手，我可能在那里有很多错误)，或者(使用)使用信号量可能会更好这个情况?还是完全其他的东西?只是原子分配给指针变量(如果可用)?我目前看不到它们如何安全地工作，可能是因为我试图保持安全，该应用程序应该运行数月而不会锁定.没有生产者中的互斥体，我可以做吗?即:仅发出条件变量的信号?
I would like to know if this is a good way to go about it (being a novice at thread programming I probably have quite a few bugs in there), or if it perhaps would be better to (ab)use semaphores for this situation? Or something else entirely? Just an atomic assign to a pointer variable if available? I currently don't see how these would work safely, probably because I'm trying to stay on the safe side, this application is supposed to run for months, without locking up. Can I do without the mutexes in the producer? i.e.: just signal a condition variable?
我当前的代码如下:
consumer {
pthread_mutex_lock(m);
pred = true; /* signal interest */
while (pred) {
/* wait a bit and hopefully get an answer before timing out */
pthread_cond_timedwait(c, m, t);
/* it is possible that the producer never produces anything, in which
case the pred will stay true, we must "designal" interest here,
unfortunately the also means that a spurious wake could make us miss
a good answer, no? How to combat this? */
pred = false;
}
/* if we got here that means either an answer is available or we timed out */
//... (do things with answer if not timed out, otherwise assign default answer)
pthread_mutex_unlock(m);
}
/* this thread is always producing, but it doesn't always have listeners */
producer {
pthread_mutex_lock(m);
/* if we have a listener */
if (pred) {
buffer = "work!";
pred = false;
pthread_cond_signal(c);
}
pthread_mutex_unlock(m);
}
注意:我使用的是现代Linux，可以在必要时使用特定于平台的功能 注意2:我使用了看似全局变量m，c和t.但是对于每个消费者而言，这些都会有所不同.
NOTE: I'm on a modern linux and can make use of platform-specific functionality if necessary NOTE2: I used the seemingly global variables m, c, and t. But these would be different for every consumer.
我希望一个线程能够注册一个事件，等待一个指定的时间，然后继续执行.理想情况下，应该有多个线程可以同时注册时间，并且所有线程都应该获得相同的事件(时间跨度中出现的所有事件).
I want a thread to be able to register for an event, wait for it for a specified time and then carry on. Ideally it should be possible for more than one thread to register at the same time and all threads should get the same events (all events that came in the timespan).
推荐答案
您想要的东西类似于c ++中的std::future(文档).消费者要求生产者使用特定功能执行一项任务.该函数创建一个名为 future (或 promise )的结构，该结构包含互斥锁，与任务关联的条件变量以及结果的空指针，然后将其返回给来电者.还将该结构，任务ID和参数(如果有)放在生产者处理的工作队列中.
What you want is something similar to a std::future in c++ (doc). A consumer requests a task to be performed by a producer using a specific function. That function creates a struct called future (or promise), holding a mutex, a condition variable associated with the task as well as a void pointer for the result, and returns it to the caller. It also put that struct, the task id, and the parameters (if any) in a work queue handled by the producer.
struct future_s {
pthread_mutex_t m;
pthread_cond_t c;
int flag;
void *result;
};
// basic task outline
struct task_s {
struct future_s result;
int taskid;
};
// specific "mytask" task
struct mytask_s {
struct future_s result;
int taskid;
int p1;
float p2;
};
future_s *do_mytask(int p1, float p2){
// allocate task data
struct mytask_s * t = alloc_task(sizeof(struct mytask_s));
t->p1 = p1;
t->p2 = p2;
t->taskid = MYTASK_ID;
task_queue_add(t);
return (struct future_s *)t;
}
然后，生产者将任务从队列中拉出，进行处理，然后终止，将结果放到将来并触发变量.
Then the producer pull the task out of the queue, process it, and once terminated, put the result in the future and trigger the variable.
消费者可能会等待未来或做其他事情.
The consumer may wait for the future or do something else.
对于可取消的期货，请在结构中包含一个标志以指示该任务已取消.那么未来将是:
For a cancellable futures, include a flag in the struct to indicate that the task is cancelled. The future is then either:
- 已交付，消费者是所有者，必须对其进行分配
- 取消后，生产者仍然是所有者并对其进行处置.
因此，生产者必须在触发条件变量之前检查未来是否被取消.
The producer must therefore check that the future has not been cancelled before triggering the condition variable.
对于共享"的未来，该标志将变成许多订户.如果数字大于零，则必须交付订单.拥有结果的消费者将由所有消费者决定(先到先得?结果是否传递给所有消费者?).
For a "shared" future, the flag turns into a number of subscribers. If the number is above zero, the order must be delivered. The consumer owning the result is left to be decided between all consumers (First come first served? Is the result passed along to all consumers?).
任何对未来结构的访问都必须互斥(与条件变量配合使用).
Any access to the future struct must be mutexed (which works well with the condition variable).
关于队列，可以使用链接列表或数组来实现(对于容量有限的版本).由于创建 futures 的函数可以同时调用，因此必须使用锁来保护它们，而锁通常是通过互斥锁实现的.
Regarding the queues, they may be implemented using a linked list or an array (for versions with limited capacity). Since the functions creating the futures may be called concurrently, they have to be protected with a lock, which is usually implemented with a mutex.
这篇关于一个线程对另一个线程(消费者/生产者)表现出兴趣的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
