跟着iLogtail学习无锁化编程

背景

跟着 iLogtail 回顾多线程基础概念

class SpinLock {    std::atomic_flag v_ = ATOMIC_FLAG_INIT;
public:    SpinLock() {}
    bool try_lock() { return !v_.test_and_set(std::memory_order_acquire); }
    void lock() {        for (unsigned k = 0; !try_lock(); ++k) {            boost::detail::yield(k);        }    }
    void unlock() { v_.clear(std::memory_order_release); }};using ScopedSpinLock = std::lock_guard<SpinLock>;

// 由中断服务例程修改的标志volatile bool data_ready = false;
// 中断服务例程void ISR() {    // 数据准备好了，设置标志    data_ready = true;}
int main() {    // 主程序循环    while (true) {        if (data_ready) {            // 处理数据            // ...            // 重置标志            data_ready = false;        }        // 执行其他任务...    }}

#include <thread>#include <atomic>#include <cassert>#include <iostream>
std::atomic<bool> x(false), y(false);int a = 0, b = 0;
void thread1() {    x.store(true, std::memory_order_relaxed); // A1    a = y.load(std::memory_order_relaxed);    // A2}
void thread2() {    y.store(true, std::memory_order_relaxed); // B1    b = x.load(std::memory_order_relaxed);    // B2}
int main() {    std::thread t1(thread1);    std::thread t2(thread2);
    t1.join();    t2.join();
    std::cout << "a: " << a << ", b: " << b << std::endl;
    return 0;}

class LogFileProfiler {private:    typedef std::unordered_map<std::string, LogStoreStatistic*> LogstoreSenderStatisticsMap;    // key : region, value :unordered_map<std::string, LogStoreStatistic*>    std::map<std::string, LogstoreSenderStatisticsMap*> mAllStatisticsMap;    std::mutex mStatisticLock;};

std::condition_variable mStopCV;
void LogtailAlarm::Stop() {    {        lock_guard<mutex> lock(mThreadRunningMux);        mIsThreadRunning = false;    }    mStopCV.notify_one();}
bool LogtailAlarm::SendAlarmLoop() {    {        unique_lock<mutex> lock(mThreadRunningMux);        while (mIsThreadRunning) {            SendAllRegionAlarm();            if (mStopCV.wait_for(lock, std::chrono::seconds(3), [this]() { return !mIsThreadRunning; })) {                break;            }        }    }    SendAllRegionAlarm();    return true;}

void ReadMetrics::ReadAsLogGroup(std::map<std::string, sls_logs::LogGroup*>& logGroupMap) const {    ReadLock lock(mReadWriteLock);    // 读链表，加读锁，允许多读    MetricsRecord* tmp = mHead;    while (tmp) {        ...        tmp = tmp->GetNext();    }}

void ReadMetrics::Clear() {    WriteLock lock(mReadWriteLock);    // 改链表，加写锁    while (mHead) {        MetricsRecord* toDelete = mHead;        mHead = mHead->GetNext();        delete toDelete;    }}

int32_t ConfigManager::FindAllMatch(vector<FileDiscoveryConfig>& allConfig,                                        const std::string& path,                                        const std::string& name /*= ""*/) {    ...    {        // 自旋锁加锁，数据查找        ScopedSpinLock cachedLock(mCacheFileAllConfigMapLock);        auto iter = mCacheFileAllConfigMap.find(cachedFileKey);        if (iter != mCacheFileAllConfigMap.end()) {            if (iter->second.second == 0                || time(NULL) - iter->second.second < INT32_FLAG(multi_config_alarm_interval)) {                allConfig = iter->second.first;                return (int32_t)allConfig.size();            }        }    }    ...    {        // 自旋锁加锁，数据插入        ScopedSpinLock cachedLock(mCacheFileAllConfigMapLock);        mCacheFileAllConfigMap[cachedFileKey] = std::make_pair(allConfig, alarmFlag ? (int32_t)time(NULL) : (int32_t)0);    }    return (int32_t)allConfig.size();}

谈谈 iLogtail 中的无锁化（Lock-Free）编程实践

class Application {private:    Application();    ~Application() = default;    // 是否接受到SigTerm信号的标识    std::atomic_bool mSigTermSignalFlag = false;};

void Application::Start() {    。。。    while (true) {        。。。        if (mSigTermSignalFlag.load()) {            LOG_INFO(sLogger, ("received SIGTERM signal", "exit process"));            Exit();        }    }}

void LogFileProfiler::AddProfilingData(const std::string& configName,                                       const std::string& region,                                       const std::string& projectName,                                       const std::string& category,                                       const std::string& convertedPath,                                       const std::string& hostLogPath,                                       const std::vector<sls_logs::LogTag>& tags,                                       uint64_t readBytes,                                       uint64_t skipBytes,                                       uint64_t splitLines,                                       uint64_t parseFailures,                                       uint64_t regexMatchFailures,                                       uint64_t parseTimeFailures,                                       uint64_t historyFailures,                                       uint64_t sendFailures,                                       const std::string& errorLine) {    string key = projectName + "_" + category + "_" + configName + "_" + hostLogPath;    // 锁    std::lock_guard<std::mutex> lock(mStatisticLock);    // 从全局map中找到每个地域的指标Map    LogstoreSenderStatisticsMap& statisticsMap = *MakesureRegionStatisticsMapUnlocked(region);    // 从地域的指标Map中找到当前采集配置的指标    std::unordered_map<string, LogStoreStatistic*>::iterator iter = statisticsMap.find(key);    // 指标计算    ...    }

class Plugin {protected:    // 指标存储数据结构    mutable MetricsRecordRef mMetricsRecordRef;};

void ProcessorInstance::Process(std::vector<PipelineEventGroup>& logGroupList) {    if (logGroupList.empty()) {        return;    }     // 计算插件输入数据指标    for (const auto& logGroup : logGroupList) {        mProcInRecordsTotal->Add(logGroup.GetEvents().size());    }
    uint64_t startTime = GetCurrentTimeInMicroSeconds();    mPlugin->Process(logGroupList);    uint64_t durationTime = GetCurrentTimeInMicroSeconds() - startTime;
    // 计算插件处理时间指标    mProcTimeMS->Add(durationTime);
    // 计算插件输出数据指标    for (const auto& logGroup : logGroupList) {        mProcOutRecordsTotal->Add(logGroup.GetEvents().size());    }    }

class Counter {private:    std::string mName;    std::atomic_long mVal;
public:    Counter(const std::string& name, uint64_t val);    uint64_t GetValue() const;    const std::string& GetName() const;    void Add(uint64_t val);    Counter* CopyAndReset();};

class WriteMetrics {private:    WriteMetrics() = default;    std::mutex mMutex;    MetricsRecord* mHead = nullptr;
    void Clear();    MetricsRecord* GetHead();
public:    ~WriteMetrics();    static WriteMetrics* GetInstance() {        static WriteMetrics* ptr = new WriteMetrics();        return ptr;    }    void PrepareMetricsRecordRef(MetricsRecordRef& ref, MetricLabels&& labels);    MetricsRecord* DoSnapshot();};
class ReadMetrics {private:    ReadMetrics() = default;    mutable ReadWriteLock mReadWriteLock;    MetricsRecord* mHead = nullptr;    void Clear();    MetricsRecord* GetHead();
public:    ~ReadMetrics();    static ReadMetrics* GetInstance() {        static ReadMetrics* ptr = new ReadMetrics();        return ptr;    }    void ReadAsLogGroup(std::map<std::string, sls_logs::LogGroup*>& logGroupMap) const;    void UpdateMetrics();};

void WriteMetrics::PrepareMetricsRecordRef(MetricsRecordRef& ref, MetricLabels&& labels) {    MetricsRecord* cur = new MetricsRecord(std::make_shared<MetricLabels>(labels));    ref.SetMetricsRecord(cur);    std::lock_guard<std::mutex> lock(mMutex);    cur->SetNext(mHead);    mHead = cur;}

void ReadMetrics::UpdateMetrics() {    MetricsRecord* snapshot = WriteMetrics::GetInstance()->DoSnapshot();    MetricsRecord* toDelete;    {        // Only lock when change head        WriteLock lock(mReadWriteLock);        toDelete = mHead;        mHead = snapshot;    }    // delete old linklist    while (toDelete) {        MetricsRecord* obj = toDelete;        toDelete = toDelete->GetNext();        delete obj;    }}

MetricsRecordRef::~MetricsRecordRef() {    if (mMetrics) {        mMetrics->MarkDeleted();    }}

MetricsRecord* WriteMetrics::DoSnapshot() {    // new read head    MetricsRecord* toDeleteHead = nullptr;
    // 遍历链表，将待删除的节点加到toDeleteHead中
    // 执行删除    while (toDeleteHead) {        MetricsRecord* toDelete = toDeleteHead;        toDeleteHead = toDeleteHead->GetNext();        delete toDelete;        writeMetricsDeleteTotal ++;    }    return snapshot;}

void ReadMetrics::UpdateMetrics() {    MetricsRecord* snapshot = WriteMetrics::GetInstance()->DoSnapshot();    MetricsRecord* toDelete;    {        // Only lock when change head        WriteLock lock(mReadWriteLock);        toDelete = mHead;        mHead = snapshot;    }    // 删除旧链表    while (toDelete) {        MetricsRecord* obj = toDelete;        toDelete = toDelete->GetNext();        delete obj;    }}