如何用 Folly ThreadCachedInt 实现高竞争多线程计数器

如何用 Folly ThreadCachedInt 实现高竞争多线程计数器

【免费下载链接】follyAn open-source C++ library developed and used at Facebook.项目地址: https://gitcode.com/GitHub_Trending/fol/folly

当多个线程对同一个原子计数器高频自增时，std::atomic_fetch_add会因为锁争用而成为瓶颈。Folly 的 ThreadCachedInt 用“线程本地缓存 + 全局原子计数器”的组合解决这个问题：每次自增只写本线程的本地计数，累计到一定次数后才批量 flush 到全局计数器。官方文档给出的结论是：在高竞争环境下，自增性能最高可比std::atomic_fetch_add快 10 倍。它适合“高写入、低读取”的访问模式；读取侧需要你在“可能过期的快速读”和“精确但昂贵的全量读”之间做选择。

引入 ThreadCachedInt

ThreadCachedInt是模板类folly::ThreadCachedInt<IntT, Tag>，头文件即folly/ThreadCachedInt.h，依赖folly/ThreadLocal.h与folly/Likely.h。

使用 CMake 构建的 folly 时直接包含头文件即可：

#include <folly/ThreadCachedInt.h>

使用 Buck/Buck2 构建时，对应的库目标是//folly:thread_cached_int（见 folly/BUCK），它导出的依赖是:likely和:thread_local。仓库自带的多进程测试//folly/test:thread_cached_int_test的完整依赖见 folly/test/BUCK。

构造函数签名为：

explicit ThreadCachedInt(IntT initialVal = 0, uint32_t cacheSize = 1000)

initialVal是初始值，cacheSize是本地缓存的 flush 阈值：线程本地计数累计超过该值后，才通过一次原子fetch_add汇入全局计数器。注意ThreadCachedInt禁止拷贝构造和拷贝赋值，只能以引用方式共享同一实例。

区分两种读模式：readFast 与 readFull

写入门槛低，但读模式选错会直接影响计数语义，这是使用ThreadCachedInt最关键的决策点。

• readFast()：单次 relaxed load 读取全局计数器，速度等同一次普通加载，但不包含尚未 flush 的本地缓存增量，返回值可能过期。
• readFull()：需要获取全局互斥锁并遍历所有线程本地计数逐一累加，返回精确值，但显著慢于readFast。文档明确指出readFull比readFast慢得多；另外，一个新线程第一次使用该对象时也要获取同一把锁来插入线程本地引用。

文档给出的最小说明示例（来自 folly/docs/ThreadCachedInt.md）：

ThreadCachedInt<int64_t> val; EXPECT_EQ(0, val.readFast()); ++val; // increment in thread local counter only EXPECT_EQ(0, val.readFast()); // increment has not been flushed EXPECT_EQ(1, val.readFull()); // accumulates all thread local counters val.set(2); EXPECT_EQ(2, val.readFast()); EXPECT_EQ(2, val.readFull());

这个例子的要点：++val之后readFast()仍读到 0，因为增量还留在线程本地缓存中；readFull()才把本地缓存汇总进去。如果你的业务只需要趋势值（如 QPS 采样），用readFast；如果需要精确总量（如周期结算、对账），用readFull。

多写者下自增并核对总数

下面的代码改编自仓库测试 folly/test/ThreadCachedIntTest.cpp 中的MultiThreadedCached用例（该用例使用FOR_EACH_RANGE宏与 gflags，此处展开为普通循环并固定 8 个线程，逻辑与断言语义保持一致），可以直接作为高竞争场景下的正确性验证模板：

#include <folly/ThreadCachedInt.h> #include <atomic> #include <thread> #include <unistd.h> #include <vector> int main() { const int numThreads = 8; const int kNumInserts = 100000; const int numPerThread = kNumInserts / numThreads; // 缓存阈值略小于单线程自增次数，保证线程结束前本地缓存基本未 flush folly::ThreadCachedInt<int64_t> counter(0, numPerThread - 2); std::atomic<bool> run(true); std::atomic<int> threadsDone(0); std::vector<std::thread> threads; for (int i = 0; i < numThreads; ++i) { threads.emplace_back([&] { for (int k = 0; k < numPerThread; ++k) { ++counter; // 注意：只支持前置自增，counter++ 不可用 } std::atomic_fetch_add(&threadsDone, 1); while (run.load()) { usleep(100); // 保持线程存活，使本地缓存保持未 flush 状态 } }); } while (threadsDone.load() < numThreads) { usleep(100); } // 此刻所有线程已完成自增，但本地缓存尚未 flush： // readFast() 不等于 kNumInserts（原测试用 EXPECT_NE 断言）， // readFull() 等于 kNumInserts（原测试用 EXPECT_EQ 断言）。 int64_t stale = counter.readFast(); int64_t exact = counter.readFull(); run.store(false); for (auto& t : threads) { t.join(); } // 线程退出时本地缓存自动 flush，此后 readFast() 也等于 kNumInserts int64_t afterJoin = counter.readFast(); (void)stale; (void)exact; (void)afterJoin; return 0; }

验证方式与仓库测试的断言一一对应：

1. 线程完成自增但未退出时，readFast()小于总数、readFull()恰好等于总数（原测试为EXPECT_NE(kNumInserts, TCInt64.readFast())与EXPECT_EQ(kNumInserts, TCInt64.readFull())）。
2. 所有线程 join 之后，线程本地计数在析构时 flush 回父计数器（实现中~IntCache()会调用flush()），此时readFast()也等于总数。这正是文档强调的“临时线程退出后计数不丢失”的保证。

用 cacheSize 控制 flush 粒度

cacheSize越大，本地缓存驻留越久，自增时触碰全局原子的频率越低，高竞争下越快——但readFast()的过期窗口也越大。构造后还可以动态调整：

counter.setCacheSize(100); // 运行时修改阈值 uint32_t sz = counter.getCacheSize();

仓库测试文件末尾附有一段“文档示例”基准结果（20 线程、dual 12-core Xeon X5650 @ 2.67GHz，来自 ThreadCachedIntTest.cpp 的注释），展示不同 cacheSize 下的自增吞吐，仅供量级参考，不是固定预期：

可以看出 cacheSize 从 0 提到 1000，64 位自增吞吐从约 21 M/s 提升到约 626 M/s，接近纯thread_local自增的速度；而同一份示例中ThrCache_readFull在高竞争下约 99.69 ns/次，印证了readFull代价远高于readFast。

限制与边界

• set是尽力而为的：头文件注释写明set在部分边缘情况下可能丢失计数（missing counts）。它只重置全局计数器并标记所有本地缓存待重置；已被标记重置的本地计数在readFull遍历时被跳过，在下次自增时被清零。
• readFastAndReset不重置缓存增量，只交换全局计数器；需要“取出并清零”语义时用readFullAndReset，它比“readFull()+set(0)”更准确，但也只是近似（文档说它会比顺序调用“好很多”而非完全精确）。
• 不支持后置自增：只有operator++()/operator--()，counter++无法编译。
• 全局锁按类型共享：默认情况下每种整数类型共享一把全局互斥锁，且Tag缺省为IntT本身。头文件注释和文档都建议：应用中有大量ThreadCachedInt实例时，通过引入不同的Tag模板参数拆分这把锁，例如ThreadCachedInt<int64_t, struct CounterA> a;。
• 首次使用会加锁：新线程第一次增量该对象时要获取全局锁插入本地引用，极端延迟敏感路径上需注意这一点。

低竞争场景的替代方案

文档在 Alternate Implementations 一节说明了取舍：如果竞争不激烈，测试文件里的ShardedAtomicInt（实现见 ThreadCachedIntTest.cpp）可能更合适——它用 2048 个std::atomic<int64_t>按线程 ID 哈希分散自增，readFull无锁求和。文档给出的示例数据是：要达到与 24 线程下ThreadCachedInt相近的自增性能需要约 2000 个槽位，内存约为 20 倍，低竞争下readFull慢约 50 倍；而低竞争环境下用更小的数组就能兼顾内存与读取性能。两者自增性能相近，按你的读写比例和竞争程度选择即可，不必默认选ThreadCachedInt。

【免费下载链接】follyAn open-source C++ library developed and used at Facebook.项目地址: https://gitcode.com/GitHub_Trending/fol/folly