Как я могу выйти из очереди блокировки C ++?

Question

После прочтения некоторых других статей я узнал, что могу реализовать очередь блокирования c ++ следующим образом:

template<typename T>
class BlockingQueue {
public:
    std::mutex mtx;
    std::condition_variable not_full;
    std::condition_variable not_empty;
    std::queue<T> queue;
    size_t capacity{5};

    BlockingQueue()=default;
    BlockingQueue(int cap):capacity(cap) {}
    BlockingQueue(const BlockingQueue&)=delete;
    BlockingQueue& operator=(const BlockingQueue&)=delete;

    void push(const T& data) {
        std::unique_lock<std::mutex> lock(mtx);
        while (queue.size() >= capacity) {
            not_full.wait(lock, [&]{return queue.size() < capacity;});
        }
        queue.push(data);
        not_empty.notify_all();
    }

    T pop() {
        std::unique_lock<std::mutex> lock(mtx);
        while (queue.empty()) {
            not_empty.wait(lock, [&]{return !queue.empty();});
        }
        T res = queue.front();
        queue.pop();
        not_full.notify_all();
        return res;
    }

    bool empty() {
        std::unique_lock<std::mutex> lock(mtx);
        return queue.empty();
    }

    size_t size() {
        std::unique_lock<std::mutex> lock(mtx);
        return queue.size();
    }

    void set_capacity(const size_t capacity) {
        this->capacity = (capacity > 0 ? capacity : 10);
    }
};

Это работает для меня, но я не знаю, как я могу закрыть его, если Я запускаю его в фоновом потоке:

void main() {
    BlockingQueue<float> q;
    bool stop{false};
    auto fun = [&] {
        std::cout << "before entering loop\n";
        while (!stop) {
            q.push(1);
        }
        std::cout << "after entering loop\n";
    };
    std::thread t_bg(fun);
    t_bg.detach();
    // Some other tasks here
    stop = true;
    // How could I shut it down before quit here, or could I simply let the operation system do that when the whole program is over?
}

Проблема в том, что, когда я хочу закрыть фоновый поток, фоновый поток мог спать, потому что очередь заполнена и операция pu sh заблокирован. Как я могу остановить это, когда я хочу, чтобы фоновый поток остановился?

Answer 1

Одним из простых способов было бы добавить флаг, который вы установили извне, когда хотите прервать операцию pop(), которая уже заблокирована. И тогда вам нужно будет решить, что прерванный pop() вернет. Один из способов - создать исключение, другой - вернуть std::optional<T>. Вот первый метод (я напишу только измененные части.)

Добавьте этот тип там, где вы считаете нужным:

struct AbortedPopException {};

Добавьте это в поля своего класса:

mutable std::atomic<bool> abort_flag = false;

Также добавьте этот метод:

void abort () const {
    abort_flag = true;
}

Измените while l oop в методе pop() следующим образом: (вам вообще не нужен while, так как я считаю, что условная переменная wait() метод, который принимает лямбду, делает не пробуждение / возврат внезапно; т. е. l oop уже находится в ожидании.)

    not_empty.wait(lock, [this]{return !queue.empty() || abort_flag;});

    if (abort_flag)
        throw AbortedPopException{};

Вот и все (я верю).

В вашем main(), когда вы хотите отключить «потребителя», вы можете позвонить abort() в свою очередь. Но вам придется справиться и с брошенным исключением. В основном это ваш сигнал «выхода».

Некоторые примечания:

1. Не отсоединяйте от потоков! Особенно здесь, где AFAICT нет никаких причин для этого (и некоторой реальной опасности тоже.) Просто подайте им сигнал о выходе (любым подходящим способом) и join() их.

2. Ваш stop флаг должен быть атомом c. Вы читаете из него в фоновом потоке и пишете в него из основного потока, и они могут (и фактически могут) перекрываться во времени, поэтому ... гонка данных!

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

ОБНОВЛЕНИЕ 1 : В ответ на комментарий ОП об отключении ... Вот что происходит в вашей основной ветке:

1. Вы порождаете поток «продюсера» (то есть тот, который помещал вещи в очередь)
2. Затем вы делаете всю работу, которую хотите сделать (например, потребляете вещи в очереди)
3. Иногда, возможно, в конце main(), вы даете сигнал об остановке потока (например, устанавливая флаг stop в true)
4. тогда, и только тогда вы join() с поток.

Это правда, что ваш основной поток будет блокироваться, пока он ожидает, пока поток поймает сигнал "стоп", выйдет из своего l oop и вернется из функции потока , но это очень очень короткое ожидание. И тебе больше нечего делать. Что еще более важно, вы будете знать, что ваш поток завершился чисто и предсказуемо, и с этого момента вы точно знаете, что этот поток не будет работать (это не важно для вас, но может иметь решающее значение для какой-либо другой задачи с потоками).

Это шаблон, которому вы обычно хотите следовать в порождающем рабочем потоке, который l oop при выполнении короткого задания.

Обновление 2 : о "полном" и «емкость» очереди. Хорошо. Это, безусловно, ваше решение. Никаких проблем с этим.

Обновление 3 : о «выбрасывании» и возврате «пустого» объекта для сигнализации об отмене «блокировки» pop() ». Я не думаю, что есть что-то плохое в метании; особенно, поскольку это очень и очень редко (это происходит один раз в конце операции производителя / потребителя). Однако, если все типы T, которые вы хотите сохранить в вашем Queue, имеют «недопустимый» или «пустой» "Государство, то вы, конечно, можете использовать это. Но бросание является более общим, если более "неприглядным" для некоторых людей.