Периодически возникает при многонитевом построении примерно следующий "паттерн": есть элемент состояния/управления (как очередь заданий), который требует контролируемого взаимодействия с помощью лока и "условной переменной". Но чтобы на время выполнения задания не блокировать тех, кто хочет установить следующее задание, лок снимается на это время. Получается следующая схема:

## Псевдокод в стиле Python
with self.lock: ## для C++ аналог -- захват лока в конструкторе
    while True:
        if not self.task: ## можно было и с предикатом, если кому удобнее читать
            self.cv.sleep()
            continue
        task = self.task
        self.task = None
        self.lock.release() ## начинается длительная операция
        try:
            self.execute(task)
        except Exception:
            ... пожаловались ...
        self.lock.acquire()

Мне не нравится в нём именно "ручное" освобождение и последующее занятие лока вокруг длительной работы. RAII или with-оператор придумали не просто так, если есть возможность взвалить контроль на них, то лучше этим воспользоваться.

Можно было бы просто перевести на очередь. Но бывает, что при этом ещё и какие-то элементы статуса отдаются обратно, или воздействие от управляющего метода более сложное. И если может быть поставлено не более одного задания, то делать это в виде очереди, ограниченной одним элементом, тоже как-то странно. Ещё бывают собственные idle операции, а ждать чтения из очереди с таймаутом не всегда возможно.

Какие есть варианты сделать это красиво без подобных ручных плясок?

Здравствуйте, netch80, Вы писали:


N>Можно было бы просто перевести на очередь. Но бывает, что при этом ещё и какие-то элементы статуса отдаются обратно, или воздействие от управляющего метода более сложное. И если может быть поставлено не более одного задания, то делать это в виде очереди, ограниченной одним элементом, тоже как-то странно. Ещё бывают собственные idle операции, а ждать чтения из очереди с таймаутом не всегда возможно.

А чем не устраивает блокируемая очередь или блокируемый дек (если надо отдавать обратно)

https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingQueue.html
https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/BlockingDeque.html

и никакие таймауты не нужны.

В Питоне не силен, но и там она вроде есть

https://codechick.io/tutorials/python/python-thread-safe-queue

Или я что-то не понял ?

Здравствуйте, Pavel Dvorkin, Вы писали:

PD>А чем не устраивает блокируемая очередь или блокируемый дек (если надо отдавать обратно)

Блокируемость в принципе не подходит, а про очередь я уже написал -- работает, но только если логика сводится к засылке чего-то через очередь.

Здравствуйте, netch80, Вы писали:

N>Блокируемость в принципе не подходит, а про очередь я уже написал -- работает, но только если логика сводится к засылке чего-то через очередь.

Ну если блокируемость не подходит, то, конечно, не годится. Хотя из исходного постинга не было ясно, что она не подходит, а было о возможности ожидании по таймауту. Это примерно то же самое, что и блокируемость на чтение — для блокируемой очереди ждем, пока в очереди что-то появится средствами самой очереди, для неблокируемой — сами ждем по таймауту.

N>Какие есть варианты сделать это красиво без подобных ручных плясок?

Забегая вперед — в пределе это вырождается в архитектуру "message broker", и соответствующие программные продукты.
Из интересных ключевых слов, например, Kafka Exactly Once Processing.

Если хочется забежать вперед другим способом — есть т.н. workflow engines, в основном занимающиеся checkpoint'ингом (в простонародье можно услышать язвительный термин retry engine).

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

Здравствуйте, netch80, Вы писали:

N>Мне не нравится в нём именно "ручное" освобождение и последующее занятие лока вокруг длительной работы. RAII или with-оператор придумали не просто так, если есть возможность взвалить контроль на них, то лучше этим воспользоваться.

Вопрос к возможностям RAII-примитивов. Например, в C++ std::unique_lock есть дополнительный флажок owns_lock который учитывается при финализации RAII, вот на нем можно сыграть — перенести ручной захват на начало итерации и поставить под условие owns_lock, тогда RAII-автоматика по правильному освоождению лока покроет тот кейс что у тебя покрыт вручную.. Что то типа такого:

    std::mutex mtx;
    std::condition_variable cv;
    bool hasTask = false;

    std::thread{
        [&]{

            std::unique_lock lock{mtx};

            for(;;)
            {
                if(!lock.owns_lock())
                    lock.lock();

                if(!hasTask)
                {
                    cv.wait(lock); // серия холостых ожиданий не приведет к дополнительным перезахватам
                    continue;
                }

                hasTask = false;
                lock.unlock();

                {
                    //делаем тяжелую задачу
                    std::this_thread::sleep_for(std::chrono::milliseconds{200});

                    if(rand() > RAND_MAX/2)
                        return;// при размотке RAII повторного unlock не случится потому что owns_lock не взведен

                    if(rand() > RAND_MAX/2)
                        break;// и так нормально тоже

                    if(rand() > RAND_MAX/2)
                        continue;// и так
                }
            }
        }}.join();

Здравствуйте, netch80, Вы писали:

N>Какие есть варианты сделать это красиво без подобных ручных плясок?

кмк если такое действительно уже стало паттерном, то оптимальней будет один раз написать хелпер а-ля InvertedLock c _enter_/_exit_ протоколом, для поддержки with и будет то же самое только raii-like

with self.lock: ## для C++ аналог -- захват лока в конструкторе
    while True:
        if not self.task: ## можно было и с предикатом, если кому удобнее читать
            self.cv.sleep()
            continue
        task = self.task
        self.task = None
        with InvertedLock(self.lock):
            try:
                self.execute(task)
            except Exception:
                ... пожаловались ...

Здравствуйте, netch80, Вы писали:

n> Периодически возникает при многонитевом построении примерно следующий "паттерн": есть элемент состояния/управления (как очередь заданий), который требует контролируемого взаимодействия с помощью лока и "условной переменной". Но чтобы на время выполнения задания не блокировать тех, кто хочет установить следующее задание, лок снимается на это время.
Вроде disruptor, там вообще можно не блокироваться. Или я что-то не понял?

Здравствуйте, netch80, Вы писали:

N>Периодически возникает при многонитевом построении примерно следующий "паттерн": есть элемент состояния/управления (как очередь заданий), который требует контролируемого взаимодействия с помощью лока и "условной переменной". Но чтобы на время выполнения задания не блокировать тех, кто хочет установить следующее задание, лок снимается на это время. Получается следующая схема:
Перечитал код внимательнее. У тебя же тут шиворот-навыворот. Ты не должен держать лок когда не надо. А должен захватывать когда надо:

while True:
    ## лочим
    with self.lock:
        while not self.task: ## не надо "if" тут, google: Spurious wakeup
            self.cv.sleep()  ## а вот это я что-то не понял. Зачем sleep??! или это await имеется в виду?
        task = self.task
        self.task = None
    ## всё, больше локи не нужны
    try:
        self.execute(task)
    except Exception:
        ... пожаловались ...