LogStore::Append应该消除不必要的锁操作

[dyx2025]

LogStore::Append应该消除不必要的锁操作

LogStore::Append将paxos log追加到valuelog文件，处于热路径上。

一个paxos group有一条IOLoop线程和一个LogStore对象。一个LogStore对象的LogStore::Append的所有调用来源于同一个paxos group的IOLoop线程。所以，LogStore::Append不存在并发问题，没必要加锁。

原代码路径：
src/logstorage/log_store.h

class LogStore
{
    // ......
    std::mutex m_oMutex;
    // ......
};

src/logstorage/log_store.cpp

int LogStore :: Append(const WriteOptions & oWriteOptions, const uint64_t llInstanceID, const std::string & sBuffer, std::string & sFileID)
{                                                                                                                                                                                       
    m_oTimeStat.Point();

    // 一个paxos group有一条IOLoop线程和一个LogStore对象。一个LogStore对象的LogStore::Append的所有调用来源于同一个paxos group的IOLoop线程。所以，LogStore::Append不存在并发问题，没必要加锁
    std::lock_guard<std::mutex> oLock(m_oMutex);

    int iFd = -1; 
    int iFileID = -1; 
    int iOffset = -1; 

    int iLen = sizeof(uint64_t) + sBuffer.size();
    int iTmpBufferLen = iLen + sizeof(int);

    int ret = GetFileFD(iTmpBufferLen, iFd, iFileID, iOffset);
    if (ret != 0)
    {   
        return ret;
    }   

    m_oTmpAppendBuffer.Ready(iTmpBufferLen);

    memcpy(m_oTmpAppendBuffer.GetPtr(), &iLen, sizeof(int));
    memcpy(m_oTmpAppendBuffer.GetPtr() + sizeof(int), &llInstanceID, sizeof(uint64_t));
    memcpy(m_oTmpAppendBuffer.GetPtr() + sizeof(int) + sizeof(uint64_t), sBuffer.c_str(), sBuffer.size());

    size_t iWriteLen = write(iFd, m_oTmpAppendBuffer.GetPtr(), iTmpBufferLen);

    if (iWriteLen != (size_t)iTmpBufferLen)
    {   
        BP->GetLogStorageBP()->AppendDataFail();
        PLG1Err("writelen %d not equal to %d, buffersize %zu errno %d", 
                iWriteLen, iTmpBufferLen, sBuffer.size(), errno);
        return -1; 
    }   

    if (oWriteOptions.bSync)
    {   
        int fdatasync_ret = fdatasync(iFd);
        if (fdatasync_ret == -1) 
        {   
            PLG1Err("fdatasync fail, writelen %zu errno %d", iWriteLen, errno);
            return -1; 
        }   
    }   

    m_iNowFileOffset += iWriteLen;

    int iUseTimeMs = m_oTimeStat.Point();
    BP->GetLogStorageBP()->AppendDataOK(iWriteLen, iUseTimeMs);
        
    uint32_t iCheckSum = crc32(0, (const uint8_t*)(m_oTmpAppendBuffer.GetPtr() + sizeof(int)), iTmpBufferLen - sizeof(int), CRC32SKIP);

    GenFileID(iFileID, iOffset, iCheckSum, sFileID);

    PLG1Imp("ok, offset %d fileid %d checksum %u instanceid %lu buffer size %zu usetime %dms sync %d",
            iOffset, iFileID, iCheckSum, llInstanceID, sBuffer.size(), iUseTimeMs, (int)oWriteOptions.bSync);

    return 0;
}

修改后的代码路径：
src/logstorage/log_store.h

class LogStore
{
    // ......
    // std::mutex m_oMutex;
    // ......
};

src/logstorage/log_store.cpp

int LogStore :: Append(const WriteOptions & oWriteOptions, const uint64_t llInstanceID, const std::string & sBuffer, std::string & sFileID)
{                                                                                                                                                                                       
    m_oTimeStat.Point();
    // std::lock_guard<std::mutex> oLock(m_oMutex);

    int iFd = -1; 
    int iFileID = -1; 
    int iOffset = -1; 

    int iLen = sizeof(uint64_t) + sBuffer.size();
    int iTmpBufferLen = iLen + sizeof(int);

    int ret = GetFileFD(iTmpBufferLen, iFd, iFileID, iOffset);
    if (ret != 0)
    {   
        return ret;
    }   

    m_oTmpAppendBuffer.Ready(iTmpBufferLen);

    memcpy(m_oTmpAppendBuffer.GetPtr(), &iLen, sizeof(int));
    memcpy(m_oTmpAppendBuffer.GetPtr() + sizeof(int), &llInstanceID, sizeof(uint64_t));
    memcpy(m_oTmpAppendBuffer.GetPtr() + sizeof(int) + sizeof(uint64_t), sBuffer.c_str(), sBuffer.size());

    size_t iWriteLen = write(iFd, m_oTmpAppendBuffer.GetPtr(), iTmpBufferLen);

    if (iWriteLen != (size_t)iTmpBufferLen)
    {   
        BP->GetLogStorageBP()->AppendDataFail();
        PLG1Err("writelen %d not equal to %d, buffersize %zu errno %d", 
                iWriteLen, iTmpBufferLen, sBuffer.size(), errno);
        return -1; 
    }   

    if (oWriteOptions.bSync)
    {   
        int fdatasync_ret = fdatasync(iFd);
        if (fdatasync_ret == -1) 
        {   
            PLG1Err("fdatasync fail, writelen %zu errno %d", iWriteLen, errno);
            return -1; 
        }   
    }   

    m_iNowFileOffset += iWriteLen;

    int iUseTimeMs = m_oTimeStat.Point();
    BP->GetLogStorageBP()->AppendDataOK(iWriteLen, iUseTimeMs);
        
    uint32_t iCheckSum = crc32(0, (const uint8_t*)(m_oTmpAppendBuffer.GetPtr() + sizeof(int)), iTmpBufferLen - sizeof(int), CRC32SKIP);

    GenFileID(iFileID, iOffset, iCheckSum, sFileID);

    PLG1Imp("ok, offset %d fileid %d checksum %u instanceid %lu buffer size %zu usetime %dms sync %d",
            iOffset, iFileID, iCheckSum, llInstanceID, sBuffer.size(), iUseTimeMs, (int)oWriteOptions.bSync);

    return 0;
}

可以使用如下代码在测试环境和生产环境进一步检查LogStore::Append是否存在并发问题。
src/logstorage/log_store.h

class LogStore
{
    // ......
    std::mutex m_oMutex;

    // 新增代码
    std::atomic<uint64_t> append_tid_{0};
    // ......
};

src/logstorage/log_store.cpp

int LogStore :: Append(const WriteOptions & oWriteOptions, const uint64_t llInstanceID, const std::string & sBuffer, std::string & sFileID)
{
    // 新增代码
    uint64_t tid = append_tid_.load(std::memory_order_relaxed);
    uint64_t call_tid = (uint64_t)pthread_self();
    if (tid > 0) {
        if (tid != call_tid) {
             PLG1Err("different threads call Append, tid %llu call_tid %llu", tid, call_tid);
             // 可以增加其他监控代码
        }
    } else {
       bool succ = append_tid_.compare_exchange_strong(tid, call_tid, std::memory_order_release, std::memory_order_relaxed);
       if (!succ) {
             PLG1Err("different threads call Append, tid %llu call_tid %llu", tid, call_tid);
             // 可以增加其他监控代码
       }
    }
                                                                                                                                                                                       
    m_oTimeStat.Point();

    // 一个paxos group有一条IOLoop线程和一个LogStore对象。一个LogStore对象的LogStore::Append的所有调用来源于同一个paxos group的IOLoop线程。所以，LogStore::Append不存在并发问题，没必要加锁
    std::lock_guard<std::mutex> oLock(m_oMutex);

    int iFd = -1; 
    int iFileID = -1; 
    int iOffset = -1; 

    int iLen = sizeof(uint64_t) + sBuffer.size();
    int iTmpBufferLen = iLen + sizeof(int);

    int ret = GetFileFD(iTmpBufferLen, iFd, iFileID, iOffset);
    if (ret != 0)
    {   
        return ret;
    }   

    m_oTmpAppendBuffer.Ready(iTmpBufferLen);

    memcpy(m_oTmpAppendBuffer.GetPtr(), &iLen, sizeof(int));
    memcpy(m_oTmpAppendBuffer.GetPtr() + sizeof(int), &llInstanceID, sizeof(uint64_t));
    memcpy(m_oTmpAppendBuffer.GetPtr() + sizeof(int) + sizeof(uint64_t), sBuffer.c_str(), sBuffer.size());

    size_t iWriteLen = write(iFd, m_oTmpAppendBuffer.GetPtr(), iTmpBufferLen);

    if (iWriteLen != (size_t)iTmpBufferLen)
    {   
        BP->GetLogStorageBP()->AppendDataFail();
        PLG1Err("writelen %d not equal to %d, buffersize %zu errno %d", 
                iWriteLen, iTmpBufferLen, sBuffer.size(), errno);
        return -1; 
    }   

    if (oWriteOptions.bSync)
    {   
        int fdatasync_ret = fdatasync(iFd);
        if (fdatasync_ret == -1) 
        {   
            PLG1Err("fdatasync fail, writelen %zu errno %d", iWriteLen, errno);
            return -1; 
        }   
    }   

    m_iNowFileOffset += iWriteLen;

    int iUseTimeMs = m_oTimeStat.Point();
    BP->GetLogStorageBP()->AppendDataOK(iWriteLen, iUseTimeMs);
        
    uint32_t iCheckSum = crc32(0, (const uint8_t*)(m_oTmpAppendBuffer.GetPtr() + sizeof(int)), iTmpBufferLen - sizeof(int), CRC32SKIP);

    GenFileID(iFileID, iOffset, iCheckSum, sFileID);

    PLG1Imp("ok, offset %d fileid %d checksum %u instanceid %lu buffer size %zu usetime %dms sync %d",
            iOffset, iFileID, iCheckSum, llInstanceID, sBuffer.size(), iUseTimeMs, (int)oWriteOptions.bSync);

    return 0;
}