C++并发学习

基础学习内存序默认的memory_order_seq_cst多线程之间全局保持一致性。memory_order_acq_rel是仅次于默认的内存序是acquire和release的结合不保证所有线程看到的结果都是一致的acquire是读操作release是写操作这个不能搞混acquire的代码屏障是发生在改代码之后也就是防止后面的代码重排序到前面release的代码屏障则是发生在改代码之前前面的代码不能重排序到后面来那为什么release写入后acquire就能获取到呢CAScompare_exchange_weak和compare_exchange_strongcompare_exchange_weak可能会伪失败当存在时还可能会失败用于循环中使用compare_exchange_strong强保证单次使用为什么weak会伪失败缓存一致性协议冲突如MESI协议内存访问重排序多核间的内存屏障同步高并发编程方法总结1、精确计数案例一多个任务执行申请资源 -- 执行 -- 释放资源 申请的是同一个资源资源有两个优先级且如果资源已经申请则可以复用这里边就需要统计申请资源的次数当申请全部释放后才会去释放资源当然也可以用单例类生成类时申请资源销毁类时释放缺点就是会一直占着资源如果优先级高的话无法让别的进程申请资源使用count记录申请次数当count 0时表示资源已经申请不需要重复申请当count 0时表示释放资源enum class Priority : uint32_t { Background static_castuint32_t(OHOS::HDI::Himediacomm::CopybitPt::V1_0::MDC_USER_RS_PIC), // MDC_USER_RS_PIC 1 2, Foreground static_castuint32_t(OHOS::HDI::Himediacomm::CopybitPt::V1_0::MDC_USER_RS_VIDEO), // MDC_USER_RS_VIDEO 1 1, }; struct PriorityState { const Priority prio; int32_t channel -1; std::atomicint count{0}; std::atomicint64_t lastFailUs{0}; std::mutex mtx; explicit PriorityState(Priority p) : prio(p), channel(-1), count(0), lastFailUs(0) {} PriorityState(const PriorityState) delete; PriorityState operator(const PriorityState) delete; }; int AAECopybitManager::RequestChannel(Priority prio) { PriorityState prioStat GetState(prio); if (GetTime() - prioStat.lastFailUs.load(std::memory_order_acquire) COOL_DOWN_US) { return ERR_TIMED_OUT; } int expected prioStat.count.load(std::memory_order_relaxed); while (expected 0) { if (prioStat.count.compare_exchange_weak(expected, expected 1, std::memory_order_release, std::memory_order_relaxed)) { return OK; } } std::lock_guardstd::mutex lock(prioStat.mtx); if (prioStat.count.load(std::memory_order_relaxed) 0) { prioStat.count.fetch_add(1, std::memory_order_relaxed); return OK; } prioStat.channel aaeResource_.RequestChannelByCap(CAP, static_castuint32_t(prio)); // 申请资源 if (prioStat.channel 0) { prioStat.count.store(1, std::memory_order_release); return OK; } int64_t now GetTime(); prioStat.lastFailUs.store(now, std::memory_order_release); MHC_LOGE([ffrt_media] Request channel failed (Prio:%d, Now:%lld), static_castint(prio), now); return ERR_DEVICE_BUSY; } void AAECopybitManager::ReleaseChannel(Priority qos) { PriorityState prioStat GetState(qos); int expected prioStat.count.load(std::memory_order_relaxed); while (expected 1) { if (prioStat.count.compare_exchange_weak(expected, expected - 1, std::memory_order_release, std::memory_order_relaxed)) { return; } } std::lock_guardstd::mutex lock(prioStat.mtx); int oldCount prioStat.count.fetch_sub(1, std::memory_order_acq_rel); if (oldCount 1) { aaeResource_.ReleaseChannel(prioStat.channel); // 释放资源 return; } if (oldCount 0) { prioStat.count.store(0, std::memory_order_release); MHC_LOGE([ffrt_media] Critical Error: Unmatched ReleaseChannel call for priority %d, prioStat.prio); } }2、近似计数案例一统计流量包这里采用多个线程维护一个count只统计自己的数量如果有需要则会累加所有线程的总数。3、分区分区的意思就是对资源进行分别处理例如下面的哲学家进餐问题可以将内存均等的隔离到每个CPU上这样就不需要加锁影响其他线程的运行。就是分别进行上锁有个例子就是哲学家进餐的问题有四个哲学家可以将两个人分为一组这样的话锁的力度就会减少并发度会增加案例一复合队列实现#include iostream #include list #include mutex #include optional // 模拟侵入式链表中的数据节点 struct Node { int value; Node(int v) : value(v) {} }; class ConcurrentDeque { private: // 分段结构 struct Segment { std::listNode* data; std::mutex mtx; // 为了优化缓存行填充可以在这里添加 padding // char padding[64]; }; Segment left; Segment right; public: ConcurrentDeque() default; // 左侧入队 void push_left(Node* node) { std::lock_guardstd::mutex lock(left.mtx); left.data.push_back(node); } // 右侧入队 void push_right(Node* node) { std::lock_guardstd::mutex lock(right.mtx); right.data.push_front(node); } // 左侧出队 Node* pop_left() { std::unique_lockstd::mutex lock_l(left.mtx); if (!left.data.empty()) { return get_and_pop(left.data, true); } // 左侧为空尝试从右侧重平衡 std::unique_lockstd::mutex lock_r(right.mtx); if (right.data.empty()) return nullptr; // 重平衡将右侧数据全部搬移到左侧 left.data.splice(left.data.end(), right.data); // 搬移后从左侧取 return get_and_pop(left.data, true); } // 右侧出队 Node* pop_right() { std::unique_lockstd::mutex lock_r(right.mtx); if (!right.data.empty()) { return get_and_pop(right.data, false); } // 右侧为空遵循先左后右顺序规避死锁 lock_r.unlock(); std::unique_lockstd::mutex lock_l(left.mtx); lock_r.lock(); if (left.data.empty() right.data.empty()) return nullptr; // 如果重试发现右侧还是空且左侧有货则重平衡 if (right.data.empty() !left.data.empty()) { right.data.splice(right.data.end(), left.data); } if (right.data.empty()) return nullptr; return get_and_pop(right.data, false); } private: Node* get_and_pop(std::listNode* l, bool is_left) { Node* res is_left ? l.back() : l.front(); if (is_left) l.pop_back(); else l.pop_front(); return res; } }; int main() { ConcurrentDeque dq; dq.push_left(new Node(1)); dq.push_right(new Node(2)); if (auto n dq.pop_left()) std::cout Pop Left: n-value std::endl; if (auto n dq.pop_left()) std::cout Pop Left (rebalanced): n-value std::endl; return 0; }