Потоко-безопасная Очередь

@JeyCi · Регистрация: 29.06.2019

Студворк — интернет-сервис помощи студентам

... вырисовалась тема отсюда

Сообщение от zayats80888

Хотя не сложно написать такую очередь(с раздельной блокировкой для "хвоста" и "головы") самостоятельно.

вот выполнила последнюю задачку отсюда - но не знаю, может лучше shared_mutex вместо просто mutex? правда на него unique_lock<shared_mutex> не ставится...

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
#include <iostream>
#include <thread>
#include <condition_variable>
#include  <mutex>
#include <chrono>
#include <queue>
 
using namespace std;
 
  bool bDone;
  mutex m;
  std::condition_variable cv;
 
int main() {
    int c = 0;
    bool done = false;
    queue<int> goods;
 
    thread producer([&]() {
        std::lock_guard<mutex> lock(m);
        for (int i = 0; i < 500; ++i) {            
            goods.push(i);
            c++;
            cv.notify_one();
        }
 
        done = true;
    });
 
    thread consumer([&]() {
        std::unique_lock<mutex> lock(m);
        while (!done) {
            while (!goods.empty()) {
                goods.pop();
                c--;
                while (!bDone) cv.wait(lock);
            }
        }
    });
 
    producer.join();
    consumer.join();
    cout << "Net: " << c << endl;
}

тут же не раздельная блокировка? - это FIFO, насколько понимаю...
а раздельная - вы, наверно, имеете ввиду для LIFO... наверно лучше на list выполнять?
sorry, что много вопросов - стою на распутье - даже в отдельную тему вынесла

@zayats80888 · 07.04.2021, 20:41

Сообщение от JeyCi

может и легче так... инача лок/анлок на каждом шаге loop - тоже больно как-то для performance

Тут вообще говорить о perfomance не приходится, пример синтетический.
Если формирование сообщения в одном потоке и его обработка в другом - трудоёмкие задачи в сравнении с передачей, то параллельная обработка выиграет в производительности, блокировки не сильно повлияют (если не удерживать на время выполнения задач).
Ну а если 500 интов "запушить" и потом "запопать", то и потоки не нужны.
Ваша изначальная задача в этом примере просто передать 500 товаров другому потоку? Очередь - это просто контейнер?

Сообщение от zayats80888

Ваш вариант с минимумом правок

Я забыл там bDone сбрасывать в конце основного цикла consumer

@JeyCi · 07.04.2021, 20:45 **[ТС]**

Сообщение от zayats80888

Я забыл там bDone сбрасывать в конце основного цикла consumer

заменила на bElementDone= 1 штука (в моём тестовом коде),
это не тот done, который в окончании всего цикла producer'a, вы, наверно, хотели сказать

ладно, завтра продолжу

@zayats80888 · 07.04.2021, 20:59

Сообщение от JeyCi

это не тот done, который в окончании всего цикла producer'a, вы, наверно, хотели сказать

Нет, просто забыл добавить строчку bDone = false; после забора товаров (ничего больше удалять/добавлять не надо)

А вообще вам уже говорили, что этот флаг не нужен, есть empty().

@JeyCi · 07.04.2021, 21:03 **[ТС]**

да и кстати, вы правы, мой декремент не отрабатывает ВООБЩЕ

под каким углом (СВОИМ) не посмотрю

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
#include <iostream>
#include <thread>
#include <condition_variable>
#include  <mutex>
#include <chrono>
#include <queue>
 
using namespace std;
 
 
int main() {
    int c = 0;
    volatile bool done = false;
    queue<int> goods;
    volatile bool bElementDone=false;
    mutex m;
    std::condition_variable cv;
 
    thread producer([&]() {               
        for (int i = 0; i < 500; ++i) { 
            std::lock_guard<mutex> lock(m);                   
            goods.push(i);
            c++;
            bElementDone=true;
            cv.notify_one();
        }
        {
        std::lock_guard<mutex> lock(m);
        done = true;
        }
    });
 
    thread consumer([&]() {
        std::unique_lock<mutex> lock(m);
        //while (!done) {
            while (!done ) {                          
                if (!goods.empty()) {
                    goods.pop();
                    c--;
                }
                while (!bElementDone ) cv.wait(lock, [&]{return bElementDone=false;}); 
                
            }
        //}
    });
 
    producer.join();
    consumer.join();
    cout << "Net: " << c << endl;
}

не получится же всё время на подсказках

Добавлено через 50 секунд

Сообщение от zayats80888

этот флаг не нужен

но не понятно, как он нарушает логику... если сам её придерживается

@zayats80888 · 07.04.2021, 21:17

Сообщение от JeyCi

да и кстати, вы правы, мой декремент не отрабатывает ВООБЩЕ

Сообщение от JeyCi

while (!done )

Что если producer уже звершил работу? Работа consumer закончится даже не начавшись(что у вас и происходит, иначе бы consumer завис, см. далее).

Сообщение от JeyCi

while (!bElementDone ) cv.wait(lock, [&]{return bElementDone=false;});

cv.wait(lock, [&]{return bElementDone=false;} это очень примерно следующее:

C++
1
2
3
4
5
while(true) {
  bElementDone=false;
  if (bElementDone == true) break;
  cv.wait(lock);
}

Это ожидание покинуть не возможно.

Подставляем в ваше выражение:

C++
1
2
3
4
5
6
while (!bElementDone ) {
  while(true) {
    bElementDone=false;
    if (bElementDone == true) break;
    cv.wait(lock);
}

Просто у вас до этого даже дело не доходит.

@oleg-m1973 · 07.04.2021, 21:20

JeyCi,

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
int main()
{
    volatile bool stop = false;
    int c = 0;
    queue<int> goods;
 
    mutex m;
    std::condition_variable cv;
 
    thread producer([&]() 
    {
        for (int i = 0; i < 500; ++i) 
        {
            std::lock_guard<mutex> lock(m);
            goods.push(i);
            c++;
            cv.notify_one();
        }
 
    m.lock();
    stop = true;
    m.unlock();
    cv.notify_all();
 
    });
 
    thread consumer([&]() 
    {
        auto wait = [&]() -> std::optional<int>
        {
            std::unique_lock<mutex> lock(m);
            while (goods.empty())
            {
                if (stop)
                    return std::nullopt;
 
                cv.wait(lock);
            }
 
            auto n = goods.front();
            goods.pop();
            c--;
            return n;       
        };
 
        for (;;)
        {
            auto val = wait();
            if (!val)
                break;
 
            std::cout << (*val + 1) << std::endl;
        }
    });
 
    producer.join();
    
    consumer.join();
    cout << "Net: " << c << endl;
}

@JeyCi · 07.04.2021, 21:33 **[ТС]**

zayats80888, убедили... ставлю

C++
1
2
                while (!bElementDone ) cv.wait(lock, [&]{return bElementDone;}); 
                bElementDone=false;

декремент начинает очень редко срабатывать, перестаю виснуть, предчуствую, что у меня не везде соблюдается синхронизация флагов и int c...- хотя я же под локом std::unique_lock<mutex> lock(m); в consumer'e...
- продолжаю править дальше...
спасибо за вскрытие грубой ошибки...
oleg-m1973, и за ваш вариант спасибо

@zayats80888 · 07.04.2021, 21:43

Сообщение от JeyCi

ставлю

Либо cv.wait(lock, [&]{return bElementDone;}); либо while (!bElementDone ) cv.wait(lock);, а то масло маслянное получается.

Сообщение от JeyCi

декремент начинает очень редко срабатывать

И об этом я тоже написал, while (!done ) вклинится, например, после первых двух элементов, вы их вытащите, пойдёте спать, за это время producer накидает остальных, выставит флаг done, вы проснётесь и на следующей итерации цикла проверка done закончит работу, оставшееся вы так и не заберёте.

@JeyCi · 07.04.2021, 21:55 **[ТС]**

Сообщение от zayats80888

а то масло маслянное получается.

от ложных пробуждений... имхо

Добавлено через 50 секунд

Сообщение от zayats80888

вы проснётесь и на следующей итерации цикла проверка done закончит работу, оставшееся вы так и не заберёте.

вероятно, вы правы...

@oleg-m1973 · 07.04.2021, 21:58

Сообщение от JeyCi

от ложных пробуждений... имхо

Забудь уже об этих "ложных пробуждениях". Проектируй свой код в расчёте на то, что каждое пробуждение ложное.

@zayats80888 · 07.04.2021, 21:59

Сообщение от JeyCi

от ложных пробуждений... имхо

Они и так учитываются предикатом или условием цикла ожидания!

@JeyCi · 07.04.2021, 22:06 **[ТС]**

мы добили его - моим ходом логической мысли... хоть как-то... ok! декремент выводит в 0

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
#include <iostream>
#include <thread>
#include <condition_variable>
#include  <mutex>
#include <chrono>
#include <queue>
 #include <atomic> 
 
using namespace std;
 
 
int main() {
    int c = 0;
    volatile std::atomic_bool  done = false;
    queue<int> goods;
    volatile bool bElementDone=false;
    mutex m;
    std::condition_variable cv;
 
    thread producer([&]() {               
        for (int i = 0; i < 500; ++i) { 
            std::lock_guard<mutex> lock(m);                   
            goods.push(i);
            c++;
            bElementDone=true;
            cv.notify_one();
        }
        {
        std::lock_guard<mutex> lock(m);
        done = true;
        }
    });
 
    thread consumer([&]() {
        std::unique_lock<mutex> lock(m);
            while (!goods.empty()) {                          
                goods.pop();
                c--;
                if (!done) {
                    // NOT while (!bElementDone ) cv.wait(lock, [&]{return bElementDone;});  
                    // while (!bElementDone ) cv.wait(lock);
                    cv.wait(lock, [&]{return bElementDone;});               
                    bElementDone=false; 
                }
            }
    });
 
    producer.join();
    consumer.join();
    cout << "Net: " << c << endl;
}

всем спасибо
... только в правильности объявлений не совсем уверена

@zayats80888 · 07.04.2021, 22:12

Сообщение от JeyCi

while (!goods.empty())

Подумайте, а что если consumer запустится раньше producer.

@oleg-m1973 · 07.04.2021, 22:13

Сообщение от JeyCi

мы добили его - моим ходом логической мысли... хоть как-то ok!

Нет, не добили

Сообщение от JeyCi

volatile bool bElementDone=false;

Это вообще не нужно

Сообщение от JeyCi

{
std::lock_guard<mutex> lock(m);
done = true;
}

После этого нужно сделать notify_one/all
И, собственно, получится тот код, который я тебе показывал, только хуже.

@JeyCi · 07.04.2021, 22:16 **[ТС]**

Сообщение от oleg-m1973

только хуже.

чем?
помимо того, что блокировку можно убрать в producer перед done=true (если его объявить atomic_bool)

@zayats80888 · 07.04.2021, 22:19

Сообщение от oleg-m1973

После этого нужно сделать notify_one/all

И прверять флаг в предикате, иначе можно зависнуть, если вытащили последний элемент перед установкой флага в продюсере.
Это тот же косяк, о котором я говорил в 33 посте про свою портянку.

@oleg-m1973 · 07.04.2021, 22:23

Сообщение от JeyCi

чем?

Тем, что это примитивная задача. У неё правильных решений - по пальцам пересчитать (а вот неправильных - тьма). Одно из них я тебе показал.

@JeyCi · 08.04.2021, 08:11 **[ТС]**

Сообщение от oleg-m1973

Это вообще не нужно

- да, хотели понять логику, чтобы убрать всё лишнее... чтобы логика стала стройной

ХОТЯ

просто думала, что если notify'ить после загрузки каждого элемента bElementDone - то consumer УЖЕ будет обрабатывать имеющиеся в очереди элементы, хоть она ещё и не full... т.е. всё-таки параллельно... по FIFO

Сообщение от oleg-m1973

После этого нужно сделать notify_one/all

-- да, забыла уведомить consumer'а

Кликните здесь для просмотра всего текста

C++
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
#include <iostream>
#include <thread>
#include <condition_variable>
#include  <mutex>
#include <queue>
#include <atomic> 
 
using namespace std;
 
int main() {
    int c = 0;
    volatile std::atomic_bool  done = false;
    queue<int> goods;
    mutex m;
    std::condition_variable cv;
 
    thread producer([&]() {               
        for (int i = 0; i < 500; ++i) { 
            std::lock_guard<mutex> lock(m);                   
            goods.push(i);
            c++;
        }
        done = true;
        cv.notify_one();
    });
 
    thread consumer([&]() {
        std::unique_lock<mutex> lock(m);
            while (!goods.empty()) {                          
               goods.pop();
               c--;
               if (!done) cv.wait(lock, [&]{return done;});              
            }
    });
 
    producer.join();
    consumer.join();
    cout << "Net: " << c << endl;
}

НО т.к.

Сообщение от DrOffset

нельзя освобождать блокировку перед уведомлением.

- то блокировка, устанавливаеиая в for-loop'е, смотрится как-то не очень... НО т.к. флаг переделала на atomic_bool - то очень даже терпимо...
p.s.
но выходить из producer'a в consumer будет только после загрузки всей очереди - т.е. опять же producer блокирован на всё время своей работы - только блокировка рывками лочится/разлочивается... (на каждом loop)...
единственное преимущество такого подхода, вероятно, в том, что на время загрузки очереди ничего не будет виснуть?.. - в смысле Event-loop самого приложения...
p.p.s
- остальные нюансы - касаются вопросов скорости - чтобы определиться с предпочтениями по тому или иному коду этой ветки

@oleg-m1973 · 08.04.2021, 10:00

Сообщение от JeyCi

-- да, забыла уведомить consumer'а
Кликните здесь для просмотра всего текста

Ужасно. Раньше и то было лучше.

@JeyCi · 08.04.2021, 10:48 **[ТС]**

конечно, ужасно... ведь все советы применила с этой ветки...
а по делу начинают появляться только ваши эмоции...

мне совет от DrOffset кажется самым валидным

Сообщение от DrOffset

Потому что нельзя освобождать блокировку перед уведомлением.

что в принципе большими буквами в комментах было ещё здесь ... спасибо DrOffset'у за напоминание...
ВЫВОД: не старайтесь казаться умнее самой логики - опУститесь до негативных эмоций, которые даже объяснениям не поддаются... потеряете дар речи

Новые блоги и статьи Все статьи Все блоги /
сукцессия 33. открытые вопросы от клауде anaschu 19.07.2026 "Что накопилось за эту часть А — тринадцать правок, из которых шесть пришли из ваших вопросов и каждая оказалась реальной ошибкой, а не калибровкой: односторонний симбиоз, отсутствующий листопад,. . .	32 сукцессия anaschu 19.07.2026 сукцессия 28‑мерное ядро стабилизировано Коллеги, фиксирую разбор инженерных правок и их изоморфную проекцию на экономику, меметику и половой отбор. Модель теперь не «подкручивает» сходимость —. . .	сукцессия 31: модель микоризы - это модель ещё нескольких явлений, социальных и экономических anaschu 18.07.2026 Теория «Всего»: апдейт v1. 1. 2 — 28‑мерное ядро стабилизировано Коллеги, фиксирую разбор инженерных правок и их изоморфную проекцию на экономику, меметику и половой отбор. Модель теперь не. . .	сукцессия 30. Массив проверяющих друг друга моделей anaschu 18.07.2026 Архитектура сети взаимопроверяющих моделей микоризной сукцессии (v2. 0) Развитие тензорного ОДУ-ядра и создание кросс-платформенного калибровочного полигона Уважаемые коллеги! В продолжение. . .
Грибы - это женщины, деревья - это мужчины. Анти инь янь для союза мужчины и женщины. anaschu 18.07.2026 ГЛАВНЫЙ НАУЧНО-ФИЛОСОФСКИЙ ВЫВОД: Сексуально-Репродуктивный Капитализм против Государства Моногамии Коллеги, мы вышли на финишную прямую 20-мерного ОДУ-моделирования вековой сукцессии (ветка. . .	Текстовый слайдер с нумерацией страниц и кроссбраузерность. russiannick 18.07.2026 Один мой друг написал текстовый слайдер с нумерацией страниц. Сегодня я расскажу как он умудрился подружить его с кроссбраузерностью. Очевидно сложно угодить с шириной каждому браузеру. Один крадет. . .	Грибы - это мемы, дерево - это человек. Применение микоризной модели к теории меметики Докинза. anaschu 18.07.2026 Меметический изоморфизм теории «всего»: Грибная сукцессия как модель эволюции идей Докинза Базовый код системы ОДУ микоризной сукцессии описывает не просто биологию, а универсальную. . .	Грибы - это корпорации, деревья - это государства. Механизм массонского захвата власти anaschu 18.07.2026 Макроэкономический изоморфизм теории «всего»: Грибная сукцессия как модель транснационального капитализма Базовый код системы ОДУ микоризной сукцессии описывает не просто биологию, а. . .

Потоко-безопасная Очередь

Решение