C ++ 11 Thread: несколько потоков ожидают условную переменную
В настоящее время я работаю над проблемой, которая имитирует расширенную модель Producer-Worker. В этой задаче доступно 3 рабочих и 3 инструмента, а для работы им нужны 2 инструмента (и материалы, но они не имеют значения). Если в хранилище имеется> = 2 инструмента, рабочий получит 2. В противном случае они будут ожидать переменную условия, которая будет сигнализироваться при наличии> = 2.
Это нормально для двух рабочих: один будет работать, затем вернет инструменты в хранилище, а другой ожидающий работник проснется и возьмет 2 инструмента. Проблема в том, что с 3 работниками всегда будет один голодный, чтобы получить инструменты.
После некоторого тестирования я заметил, что потоки, ожидающие условную переменную, структурированы в виде стека. Есть ли возможность сделать его в форме?
(1 ждет, 2 ждет, и 3 ждет. Когда 1 проснулся и хочет сделать другое, он должен ждать позади 2 и 3.)
Вот один пример вывода. Код слишком длинный, поэтому я опубликую его, если это действительно необходимо. Есть 3 рабочих потока и 1 мьютекс инструмента. Тот, кто голодает, отличается от каждого другого бега.
1 Tools taken. Remaining: 1
2 Waiting on tools...
3 Waiting on tools...
1 Operator Product made. Tools returned. Tools now:3
3 Tools taken. Remaining: 1
1 Waiting on tools...
3 Materials returned for switch.
3 Operator Product made. Tools returned. Tools now:3
1 Tools taken. Remaining: 1
3 Waiting on tools...
1 Materials returned for switch.
1 Operator Product made. Tools returned. Tools now:3
3 Tools taken. Remaining: 1
1 Waiting on tools...
3 Materials returned for switch.
3 Operator Product made. Tools returned. Tools now:3
1 Tools taken. Remaining: 1
3 Waiting on tools...
1 Materials returned for switch.
1 Operator Product made. Tools returned. Tools now:3
3 Tools taken. Remaining: 1
1 Waiting on tools...
3 Materials returned for switch.
3 Operator Product made. Tools returned. Tools now:3
1 Tools taken. Remaining: 1
3 Waiting on tools...
1 Materials returned for switch.
...
(Как вы можете видеть 2 никогда не получает инструменты …)
Обновление: 2013/07/05
Я добавил немного кода.
int tools = 3; //global
string last; //current last product on output buffer
mutex toolsMutex;
mutex matSearchMutex;
int main(){
//Initializing Producers
Producer prod1(1);
Producer prod2(2);
Producer prod3(3);thread p1(processor,1);
thread p2(processor,2);
thread p3(processor,3);
p1.detach();
p2.detach();
p3.detach();
while(true){//forever running
}
return 0;
}
Процессор:
//Processor method
void processor(int i){
srand(time(NULL));
while (true){ //forever runningbool hasTools = false;
bool productMade = false;
while (productMade == false){ //while product has yet to be made.
//choose what to make...if (hasTools == false){
thread matT(getMaterials,whatToMake);
thread toolT(getTools,i);
toolT.join();
matT.join();
hasTools = true;
}
else{ //tools acquired but no materials
thread matT(getMaterials,whatToMake);
matT.join();
}
if (recordedLast.compare(last) != 0){
//return materials and acquire new ones the next run
continue;
}
else {
makeProduct(whatToMake);
unique_lock<mutex> locker(toolMutex);
tools = tools + 2;
cout << i << " Operator Product made. Tools returned. Tools now:" << tools << endl;
productMade = true;
if (tools >=2) toolsCV.notify_one();
}
//done processing
}}
}
makeProducts:
void makeProduct(int i){
unique_lock<mutex> mainMatLock(matSearchMutex);
// make product according to i
this_thread::sleep_for(chrono::milliseconds(rand() % 1000 + 10));
}
getTools:
void getTools(int i){
unique_lock<mutex> locker(toolMutex);
if (tools <2){
cout << i << " Waiting on tools..." << endl;
toolsCV.wait(locker);}
tools = tools - 2;//tools acquired
cout << i <<" Tools taken. Remaining: " << tools << endl;
}
Спасибо тем, кто ответил. Сегодня вечером я попытаюсь реализовать очередь ожидания, используя несколько переменных условия.
(P.S. Есть ли какой-нибудь лучший способ форматировать код здесь, в переполнении стека? Кроме четырех пробелов …
Решение
std::condition_variable не указывает, какая ожидающая нить просыпается при вызове notify_one, Поэтому вы должны написать код, который не заботится о том, какой поток проснулся. Стандартный шаблон заключается в том, что какой бы поток ни проснулся, этот поток должен выполнить ту работу, которая должна быть выполнена.
Если вам требуется, чтобы потоки просыпались в определенном порядке, используйте другой механизм. Вы могли бы, например, иметь отдельный std::condition_variable для каждого потока, а затем помещать потоки в очередь, когда им нужны инструменты. Когда поток получает доступ к инструментам, он может затем сигнализировать переменную условия, соответствующую потоку в начале очереди. Затем эта нить проснется, а остальные останутся спящими (по модулю ложных пробуждений).
Другие решения
Когда есть несколько потоков, ожидающих условия, порядок
в котором они проснулись (notify_all) или какой
проснулся (notify_one) не уточняется. Если вам нужен какой-то
заказа, вам нужно использовать notify_allи реализовать это
сам. Вы можете сохранить очередь ожидающих потоков: before
в ожидании (но после получения мьютекса), нажмите идентификатор потока на
конец очереди. В цикле, включите эту тему в
перед очереди и необходимые инструменты доступны «. Когда вы получаете
инструменты, удалить идентификатор из передней части очереди и вызвать
notify_all снова.
Реальная проблема здесь в том, что если у вас есть рабочие потоки и ограниченное количество необходимых ресурсов, вам не нужно заботиться о том, какие потоки действительно активированы, вам нужно только заботиться о том, чтобы работа была выполнена. Единственная разница здесь в регистрации. Количество потоков, которые вы определили, — это количество потоков, которые могут работать параллельно, которое ограничено ресурсами до одного.
Если это не подходит для вас, тогда вам нужно действовать самостоятельно, как объяснено в других ответах.
Один из подходов может заключаться в использовании полноценного семафора, который совместно используется потоками вместо условной переменной. Таким образом, вы можете ждать определенного подсчета.
#include <mutex>
#include <thread>
#include <condition_variable>
using std::mutex;
using std::condition_variable;
class Semaphore
{
public:
/**
* Construct a counting semaphore with an initial value
* @param cnt The value of the initial semaphore count
*/
Semaphore(unsigned int cnt);
/**
* acquire a semaphore count
* @param numRes The number of count ressources to acquire
*/
void acquire(unsigned int numRes = 1);
/**
* try to acquire a semaphore count.
* @param numRes The number of count ressources that the method tries to acquire
* @return true, if numRes could be aquired
* false, otherwise
*/
bool tryAcquire(unsigned int numRes = 1);
/**
* release one semaphore cnt
* @param numRes The number of count ressources to release
*/
void release(unsigned int numRes = 1);
private:
unsigned int cnt;
mutex mut;
condition_variable cond;
};
Реализация выглядит так:
void Semaphore::acquire(unsigned int numRes)
{
unique_lock<mutex> lock(mut);
while (cnt < numRes)
{
cond.wait(lock);
}
cnt-=numRes;
}
bool Semaphore::tryAcquire(unsigned int numRes)
{
unique_lock<mutex> lock(mut);
if (cnt>=numRes)
{
cnt -= numRes;
return true;
}
return false;
}
void Semaphore::release(unsigned int numRes)
{
{
unique_lock<mutex> lock(mut);
cnt += numRes;
}
// notify <numRes> waiting entities
for (unsigned int i = 0; i<numRes; ++i)
{
cond.notify_one();
}
}
detector