概要

目前维护的一套c++服务，在打印日志时会存在锁竞争。先用brpc定位锁竞争的环节，然后提出集中改进方案，以及它们的优缺点。

contention火焰图

使用brpc在压测环境下打印contention火焰图。

左边的圆圈是处理服务的接入数据，会记录接收量，os类型，业务方类型等统计参数；

右边那两个稍小圆圈是两处发送下游，会记录发出量等统计参数；

它们都是用同一把锁，当需要写入统计数据时，会去竞争这把锁；

从连接方框的边旁的数据大小，可以看到主要的锁竞争来自于处理接收数据的业务函数。原因是我们的服务接收数据是逐条接收的，但是发送数据是批量，因此数量是不对等的。

改进方案

每个统计指标对应一把锁

好处：锁的作用域得到最小力度的控制

坏处：当有多处需要写入的统计信息量较大时，这种场景不合适

自旋锁

好处：不会发生cs

坏处：当临界区较长时，对cpu的浪费不能忽视

使用atomic

好处：基础类型在x86架构上是无锁的

坏处：atomic变量虽然比加锁赋值更轻量，但比普通参数复制还是更重一些。当统计参数较多时，这部分时间开销累计起来也是很大的

对比atomic和加锁的时间开销

atomic

#include<iostream>
#include<atomic>
#include<thread>
 
using namespace std;
 
atomic<int> a(0);
 
void sum() {
    for(int i=0; i!=10000000; i++) {
        a++;
    }
}
 
int main() {
    // cout << a.is_lock_free() << endl;
    // 启动四个线程
    thread t1(sum);
    thread t2(sum);
    thread t3(sum);
    thread t4(sum);
    t1.join();
    t2.join();
    t3.join();
    t4.join();
    int ret = a.load();
    cout << ret << endl;
    return 0;
}

mutex

#include<iostream>
#include<thread>
#include<mutex>
 
using namespace std;
 
mutex m;
int a = 0;
int b = 0;
 
void sum() {
    for(int i=0; i!=10000000; i++) {
        m.lock();
        a++;
        m.unlock();
    }
}
 
int main() {
    // 启动四个线程
    thread t1(sum);
    thread t2(sum);
    thread t3(sum);
    thread t4(sum);
    t1.join();
    t2.join();
    t3.join();
    t4.join();
    cout << a << endl;
    return 0;
}

运行

didi@bogon cpptest % time ./atomic 
40000000
./atomic  3.41s user 0.01s system 395% cpu 0.865 total
didi@bogon cpptest % time ./mutex
40000000
./mutex  2.13s user 4.70s system 317% cpu 2.153 total

可见使用atomic后，系统调用的开销显著降低，性能得以提升

双版本切换

该方案的思路是使用双版本保存统计信息。当需要记录统计参数时，先获取当前版本的index，可以获得往哪个版本写。新开一个线程，每隔一个时间间隔切换index，同时打印之前的统计参数

代码

#include<iostream>
#include<atomic>
#include<vector>
#include<thread>
#include<unistd.h>
 
using namespace std;
 
struct Info {
    int a{0};
    void clear() {a=0;}
};
 
atomic<int> Index(0); // 指向当前的版本
vector<Info*> vec_infos; // 保存双版本的信息
 
void print_info(int idx) { // 打印idx版本的统计参数
    cout << vec_infos.at(idx)->a << endl;
}
 
void clear_info(int idx) { // 将idx版本的统计参数清零
    vec_infos.at(idx)->a = 0;
}
 
void printInfo() {
    int i=0; // 保证函数能退出
    int idx=0;
    while(i!=10) {
        sleep(1);
        int next_idx = (idx+1) % 2;
        clear_info(next_idx); // 将即将使用的Info清零
        Index.store(next_idx); // 切换index，之后的写入的统计数据会放到另一个Info中
        print_info(idx); // 打印之前的统计数据
        idx = next_idx; // 指向新的index
        i++;
    }
}
 
void writeInfo() { // 写入统计参数
    while(true) {
        int idx = Index.load(); // 获取当前的index
        vec_infos.at(idx)->a++;
    }
}
 
int main() {
    Info* info1 = new Info();
    Info* info2 = new Info();
    vec_infos.push_back(info1);
    vec_infos.push_back(info2); // 双版本统计参数初始化
    thread t1(printInfo); // 启动打印线程
    thread t2(writeInfo); // 启动模拟写入统计参数的线程
 
    t1.join();
    return 0;
}

这份代码其实并不严密，写入统计参数时，可能已经完成打印了，导致数据丢失。首先这种概率并不大，atomic的store操作的时间开销是大于普通变量赋值的；其次是对于统计信息，即使丢少量数据，问题不大；再则我们可以将intervel的时间分成两份，在store操作和print_info之前等待一段时间，确保之前版本的数据都被写入。