协作队列取消

Cooperative Queue Cancellation

专题: Systems & Architecture / 系统与架构
难度: L1
来源: MyntBit

个人笔记

题目详情

在量化交易系统中，工作线程频繁轮询并发队列处理高吞吐行情或订单更新。实现协作式取消确保这些线程在系统关闭时能解除阻塞并优雅退出。

任务：实现 CooperativeQueue 类，支持 stop() 方法设置停止标志。消费线程在 pop() 时检查停止标志，队列停止后 pop() 返回特殊值指示线程应退出。使用 std::atomic<bool> 作为停止信号。

英文原题

In quantitative trading systems, worker threads frequently poll concurrent queues to process high-throughput market data or order updates. Implementing cooperative cancellation ensures that these threads can unblock and exit cleanly during system shutdown without data corruption or busy-waiting.
Task
Implement a CancellableQueue class that provides thread-safe access and supports cooperative cancellation.
Your class must implement the following methods:

void push(double val): Adds val to the

解析

问题分析

协作式取消允许消费者线程在关闭时安全停止处理队列中的请求——不是强制终止（可能导致数据损坏），而是设置取消标志并由工作线程主动检查。适合有界队列和后台工作线程。

实现

template<typename T>
class CancellableQueue {
    std::queue<T> q_; std::mutex mtx_; std::condition_variable cv_;
    std::atomic<bool> cancelled_{false};
public:
    void cancel() { cancelled_ = true; cv_.notify_all(); }
    bool push(T v) {
        if (cancelled_) return false;
        { std::lock_guard lk(mtx_); q_.push(std::move(v)); }
        cv_.notify_one(); return true;
    }
    std::optional<T> pop() {
        std::unique_lock lk(mtx_);
        cv_.wait(lk, [this]{ return !q_.empty() || cancelled_; });
        if (cancelled_ && q_.empty()) return std::nullopt;
        T v = std::move(q_.front()); q_.pop(); return v;
    }
};

复杂度与边界

时间复杂度：push/pop O(1)
空间复杂度：O(队列容量)
边界条件：(1) cancel 后 push 返回 false (2) pop 在 cancel 且队列空时返回 nullopt (3) cancel 不可逆——重启需新建队列

英文解析

Analysis

Cooperative cancellation allows consumer threads to safely stop processing queue requests during shutdown — not forced termination — this risks data corruption, but setting a cancel flag that worker threads voluntarily check. Suitable for bounded queues and background worker threads.

Solution

template<typename T>
class CancellableQueue {
    std::queue<T> q_; std::mutex mtx_; std::condition_variable cv_;
    std::atomic<bool> cancelled_{false};
public:
    void cancel() { cancelled_ = true; cv_.notify_all(); }
    bool push(T v) {
        if (cancelled_) return false;
        { std::lock_guard lk(mtx_); q_.push(std::move(v)); }
        cv_.notify_one(); return true;
    }
    std::optional<T> pop() {
        std::unique_lock lk(mtx_);
        cv_.wait(lk, [this]{ return !q_.empty() || cancelled_; });
        if (cancelled_ && q_.empty()) return std::nullopt;
        T v = std::move(q_.front()); q_.pop(); return v;
    }
};

Complexity & Edge Cases

Time complexity: push/pop O(1)
Space complexity: O(queue size)
Edge cases: (1) cancel + empty queue: pop returns nullopt immediately (2) cancel + non-empty: pop drains remaining items (3) Push after cancel returns false

Key Considerations

Cancellation visibility: Cancellation request must be visible to all threads processing queue items; atomic flag check must occur before dispatch, not after
Drain vs abort: Cooperative cancellation may either drain remaining items (graceful) or discard them (immediate); configurable per use-case requirements
Cleanup on cancel: Cancelled tasks may hold partial state (open connections, partial writes); must execute cleanup hooks, not silently abandon resources
Timeout fallback: Cooperative cancellation relies on worker code checking the flag; unresponsive tasks must have timeout fallback to force cancellation