Ага, у меня код зависает при попытке вызвать suspend. Метод _runner возвращает управление в event loop, прерываясь на ближайшем await (например target = await anext(self._async_iterator)), а блокировка при этом остаётся захваченной, затем вызов suspend в каком-нибудь другом таске тоже пытается захватить ту же блокировку и зависает. Из-за этого зависания никто не возвращает управление в event loop, как следствие он не может вернуть управление в метод _runner и блокировка остаётся захваченной навсегда

@andreymal да, я пытаюсь это решить. Мне кажется, эта проблема - вариация исходной проблемы (см. самый первый код в вопросе)

Мне кажется, вы бы сильно облегчили себе жизнь, если бы сделали методы suspend/resume/cancel асинхронными. Если ваша бизнес-логика допускает, чтобы методы suspend/resume/cancel могли ненадолго зависнуть до фактического перехода операции в запрошенное состояние (судя по тому, что вы не стесняетесь использовать блокировку, я делаю вывод, что это для вас допустимо), то воткнуть какой-нибудь маленький await было бы намного проще чем возиться с блокировками и очередями (ну правда, очередь тут нафиг не нужна)

@andreymal, я попробовал, но это не работает. Управляющие функции вызываются до запуска _runner

Ну, в вашей реализации есть такая же проблема: если поставить обе функции подряд без задержек op.run(); op.suspend() то точно так же операция перейдёт в состояние suspended до начала работы _runner (по той же самой причине — здесь нет передачи управления в event loop и у него нет возможности запустить _runner до ближайшего await)

Если вам важно дождаться начала работы _runner, то метод run тоже можно сделать асинхронным ¯_(ツ)_/¯

Работает, только изначальная проблема никуда не делась: из-за того, что вы убрали блокировку внутри _runner, у операции опять появилась возможность выполнить фрагмент вычислений в состоянии suspended i.stack.imgur.com/9fDtr.png

И кстати вы замаскировали факт выполнения вычислений: из-за того, что у вас _progress_callback вызывается после await _resume_signal, он не будет вызван в состоянии suspended, хотя в это время в переменной target уже лежит готовое вычисление, выполненное в состоянии suspended

@andreymal, Я обновил код. Он выглядит рабочим. так ли это? Я освобождаю блокировку в _runner только на ожидание сигнала. Теперь все управляющие функции асинхронны.

@Aycon вроде работает (по крайней мере очевидных способов сломать я сходу не заметил)

Зато заметил, что can_step бессмысленный (после can_step = False стоит raise, который всё равно гарантированно прервёт работу _runner, ну а в сумме весь тот try-except получается бессмысленный)

Всё-таки сломал: если после suspend вызвать cancel, то будет ошибка «Lock is not acquired»