Skip to content

消除tcp消息接收路径上的不必要拷贝 #288

Open
Open
消除tcp消息接收路径上的不必要拷贝#288

Description

@dyx2025
dyx2025
opened on Aug 2, 2026

消除tcp消息接收路径上的不必要拷贝

phxpaxos的tcp消息接收路径上存在不必要的拷贝。消息先拷贝到m_oReadCacheBuffer，再把m_oReadCacheBuffer拷贝到new出来的string对象。

原代码路径：
src/communicate/tcp/message_event.cpp

int MessageEvent :: OnRead()
{
    if (m_iLeftReadLen > 0)
    {
        return ReadLeft();
    }
    
    int iReadLen = m_oSocket.receive(m_sReadHeadBuffer + m_iLastReadHeadPos, sizeof(int) - m_iLastReadHeadPos);
    if (iReadLen == 0)
    {
        BP->GetNetworkBP()->TcpOnReadMessageLenError();
           PLErr("read head fail, readlen %d, socket broken", iReadLen);
        return -1;
    }

    m_iLastReadHeadPos += iReadLen;
    if (m_iLastReadHeadPos < (int)sizeof(int))
    {
        PLImp("head read pos %d small than sizeof(int) %zu", m_iLastReadHeadPos, sizeof(int));
        return 0;
    }
    
    m_iLastReadHeadPos = 0;
    int niLen = 0;
    int iLen = 0;
    memcpy((char *)&niLen, m_sReadHeadBuffer, sizeof(int));
    iLen = ntohl(niLen) - 4;
    
    if (iLen < 0 || iLen > MAX_VALUE_SIZE)
    {
        PLErr("need to read len wrong %d", iLen);
        return -2; 
    }

    // 消息先拷贝到m_oReadCacheBuffer
    m_oReadCacheBuffer.Ready(iLen);

    m_iLeftReadLen = iLen;
    m_iLastReadPos = 0;
    
    //second read maybe no data read, so readlen == 0 is ok.
    bool bAgain = false;
    iReadLen = m_oSocket.receive(m_oReadCacheBuffer.GetPtr(), iLen, &bAgain);
    if (iReadLen == 0)
    {
        if (!bAgain)
        {
            PLErr("second read data fail, readlen %d, no again, socket broken", iReadLen);
            return -1;
        }
        else
        {
            PLErr("second read data, readlen %d need again", iReadLen);
            return 0;
        }
    }

    if (iReadLen == iLen)
    {
        ReadDone(m_oReadCacheBuffer, iLen);
        m_iLeftReadLen = 0;
        m_iLastReadPos = 0;
    }
    else if (iReadLen < iLen)
    {
        m_iLastReadPos = iReadLen;
        m_iLeftReadLen = iLen - iReadLen;

        PLImp("read buflen %d small than except len %d", iReadLen, iLen);
    }
    else
    {
        PLErr("read buflen %d large than except len %d", iReadLen, iLen);
        return -2;
    }

    return 0;
}

src/algorithm/ioloop.cpp

int IOLoop :: AddMessage(const char * pcMessage, const int iMessageLen)
{
    m_oMessageQueue.lock();

    BP->GetIOLoopBP()->EnqueueMsg();

    if ((int)m_oMessageQueue.size() > QUEUE_MAXLENGTH)
    {
        BP->GetIOLoopBP()->EnqueueMsgRejectByFullQueue();

        PLGErr("Queue full, skip msg");
        m_oMessageQueue.unlock();
        return -2;
    }

    if (m_iQueueMemSize > MAX_QUEUE_MEM_SIZE)
    {
        PLErr("queue memsize %d too large, can't enqueue", m_iQueueMemSize);
        m_oMessageQueue.unlock();
        return -2;
    }

    // 再把m_oReadCacheBuffer拷贝到new出来的string对象    
    m_oMessageQueue.add(new string(pcMessage, iMessageLen));
        
    m_iQueueMemSize += iMessageLen;

    m_oMessageQueue.unlock();

    return 0;
}

可以直接把消息拷贝到new出来的string对象，省去把消息拷贝到m_oReadCacheBuffer的操作。文末会给出模拟测试代码，评估潜在的性能收益。

修改后的代码：
src/communicate/tcp/message_event.h

class MessageEvent : public Event
{
    // ...
    // 修改代码
    // void ReadDone(BytesBuffer & oBytesBuffer, const int iLen);
    void ReadDone();
    // ...
    // 修改代码
    // BytesBuffer m_oReadCacheBuffer;
    std::string* read_buffer_{nullptr};
    // ...    
};

src/communicate/tcp/message_event.cpp

MessageEvent :: ~MessageEvent()
{                                                                                                                                                                                       
    while (!m_oInQueue.empty())
    {   
        QueueData tData = m_oInQueue.front();
        m_oInQueue.pop();

        delete tData.psValue;
    }

    // 新增代码
    delete  read_buffer_;
    read_buffer_ = nullptr;
}

// ...

// void MessageEvent :: ReadDone(BytesBuffer & oBytesBuffer, const int iLen)
void MessageEvent ::  ReadDone();                   
{   
    //PLHead("ok, len %d", iLen);

    // 修改代码
    /*
    m_poNetWork->OnReceiveMessage(oBytesBuffer.GetPtr(), iLen);
    
    BP->GetNetworkBP()->TcpReadOneMessageOk(iLen);
    */

    bool enable_transfer_message_ownership = true;
    bool has_message_ownership_transferred = false;
    m_poNetWork->OnReceiveMessage(read_buffer_->data(), read_buffer_->size(), read_buffer_, 
        enable_transfer_message_ownership, has_message_ownership_transferred);

    if (has_message_ownership_transferred) {
        read_buffer_ = nullptr;
    }
}

int MessageEvent :: ReadLeft()                                                              
{                                                                                           
    bool bAgain = false;
    // 修改代码                                                                    
    // int iReadLen = m_oSocket.receive(m_oReadCacheBuffer.GetPtr() + m_iLastReadPos, m_iLeftReadLen, &bAgain);
    int iReadLen = m_oSocket.receive(read_buffer_->data() + m_iLastReadPos, m_iLeftReadLen, &bAgain);
    //PLImp("readlen %d", iReadLen);                                                        
    if (iReadLen == 0)                                                                      
    {
        //socket broken                                                                     
        return -1;                                                                          
    }
    
    m_iLeftReadLen -= iReadLen;                                                             
    m_iLastReadPos += iReadLen;                                                             
    
    if (m_iLeftReadLen == 0)                                                                
    {
        修改代码
        // ReadDone(m_oReadCacheBuffer, m_iLastReadPos);
        ReadDone();

        m_iLeftReadLen = 0;
        m_iLastReadPos = 0;
    }                                                                                                                                                                                   
    
    return 0;                                                                               
}

int MessageEvent :: OnRead()
{
    if (m_iLeftReadLen > 0)
    {
        return ReadLeft();
    }
    
    int iReadLen = m_oSocket.receive(m_sReadHeadBuffer + m_iLastReadHeadPos, sizeof(int) - m_iLastReadHeadPos);
    if (iReadLen == 0)
    {
        BP->GetNetworkBP()->TcpOnReadMessageLenError();
           PLErr("read head fail, readlen %d, socket broken", iReadLen);
        return -1;
    }

    m_iLastReadHeadPos += iReadLen;
    if (m_iLastReadHeadPos < (int)sizeof(int))
    {
        PLImp("head read pos %d small than sizeof(int) %zu", m_iLastReadHeadPos, sizeof(int));
        return 0;
    }
    
    m_iLastReadHeadPos = 0;
    int niLen = 0;
    int iLen = 0;
    memcpy((char *)&niLen, m_sReadHeadBuffer, sizeof(int));
    iLen = ntohl(niLen) - 4;
    
    if (iLen < 0 || iLen > MAX_VALUE_SIZE)
    {
        PLErr("need to read len wrong %d", iLen);
        return -2; 
    }

    // 修改代码
    // m_oReadCacheBuffer.Ready(iLen);
    if (read_buffer_) {
        // 正常情况下不会频繁进入这一分支
        // 上一条接收的tcp消息没有走到ReadDone或者ReadDone里面的消息所有权没有被转移才会走到这一分支
        delete read_buffer_;
        read_buffer_ = nullptr;
    }

    read_buffer_ = new std::string;
    read_buffer_ ->resize_and_overwrite(iLen, [](const char* p, std::size_t n) {
                                    return n;
                                });

    m_iLeftReadLen = iLen;
    m_iLastReadPos = 0;
    
    //second read maybe no data read, so readlen == 0 is ok.
    bool bAgain = false;
    // 修改代码
    // iReadLen = m_oSocket.receive(m_oReadCacheBuffer.GetPtr(), iLen, &bAgain);
    iReadLen = m_oSocket.receive(read_buffer_->data(), iLen, &bAgain);
    if (iReadLen == 0)
    {
        if (!bAgain)
        {
            PLErr("second read data fail, readlen %d, no again, socket broken", iReadLen);
            return -1;
        }
        else
        {
            PLErr("second read data, readlen %d need again", iReadLen);
            return 0;
        }
    }

    if (iReadLen == iLen)
    {
        // 修改代码
        // ReadDone(m_oReadCacheBuffer, iLen);
        ReadDone();
        m_iLeftReadLen = 0;
        m_iLastReadPos = 0;
    }
    else if (iReadLen < iLen)
    {
        m_iLastReadPos = iReadLen;
        m_iLeftReadLen = iLen - iReadLen;

        PLImp("read buflen %d small than except len %d", iReadLen, iLen);
    }
    else
    {
        PLErr("read buflen %d large than except len %d", iReadLen, iLen);
        return -2;
    }

    return 0;
}

include/phxpaxos/network.h

class NetWork
{
    // ...
    //When receive a message, call this funtion.
    //This funtion is async, just enqueue an return.
    // 修改代码
    // int OnReceiveMessage(const char * pcMessage, const int iMessageLen);
    // 如果pcMessage和iMessageLen代表的消息来自new出来的string对象，应该把string对象的指针出给message_container
    // enable_transfer_message_ownership表示是否允许把pcMessage和iMessageLen代表的消息的所有权转移出去
    // has_message_ownership_transferred表示pcMessage和iMessageLen代表的消息的所有权转移出去是否已经转移出去
    int OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
        bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred);
    // ...
};

src/communicate/network.cpp

// 修改代码
/*
int NetWork :: OnReceiveMessage(const char * pcMessage, const int iMessageLen)                                                                                                          
{
    if (m_poNode != nullptr)
    {   
        m_poNode->OnReceiveMessage(pcMessage, iMessageLen);
    }   
    else
    {   
        PLHead("receive msglen %d", iMessageLen);
    }   

    return 0;
}
*/

int NetWork :: OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
    bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred)                                                                                                       
{
    if (has_message_ownership_transferred) {
        PLErr("message ownership has been transferred");
        return -2;
    }

    if (m_poNode != nullptr)
    {   
        m_poNode->OnReceiveMessage(pcMessage, iMessageLen, message_container,
            enable_transfer_message_ownership, has_message_ownership_transferred);
    }   
    else
    {   
        PLHead("receive msglen %d", iMessageLen);
    }   

    return 0;
}

include/phxpaxos/node.h

class Node
{ 
    // ...
    // 修改代码
    // virtual int OnReceiveMessage(const char * pcMessage, const int iMessageLen) = 0;
    // 参数定义参考NetWork::OnReceiveMessage
    virtual int OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
         bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred) = 0;
    // ...
};

src/node/pnode.h

class PNode : public Node
{
    // ...
    // 修改代码
    // int OnReceiveMessage(const char * pcMessage, const int iMessageLen);
    // 参数定义参考NetWork::OnReceiveMessage
    int OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
         bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred);
    // ...
};

src/node/pnode.cpp

// 修改代码
/*
int PNode :: OnReceiveMessage(const char * pcMessage, const int iMessageLen)
{                                                                                                                                                                                       
    if (pcMessage == nullptr || iMessageLen <= 0)
    {
        PLErr("Message size %d to small, not valid.", iMessageLen);
        return -2;
    }
            
    int iGroupIdx = -1;
    
    memcpy(&iGroupIdx, pcMessage, GROUPIDXLEN);
    
    if (!CheckGroupID(iGroupIdx))
    {
        PLErr("Message groupid %d wrong, groupsize %zu", iGroupIdx, m_vecGroupList.size());
        return Paxos_GroupIdxWrong;
    }
            
    return m_vecGroupList[iGroupIdx]->GetInstance()->OnReceiveMessage(pcMessage, iMessageLen);
}
*/

int PNode :: OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
     bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred);
{
    if (has_message_ownership_transferred) {
        PLErr("message ownership has been transferred");
        return -2;
    }
                                                                                                                                                                                       
    if (pcMessage == nullptr || iMessageLen <= 0)
    {
        PLErr("Message size %d to small, not valid.", iMessageLen);
        return -2;
    }
            
    int iGroupIdx = -1;
    
    memcpy(&iGroupIdx, pcMessage, GROUPIDXLEN);
    
    if (!CheckGroupID(iGroupIdx))
    {
        PLErr("Message groupid %d wrong, groupsize %zu", iGroupIdx, m_vecGroupList.size());
        return Paxos_GroupIdxWrong;
    }
            
    return m_vecGroupList[iGroupIdx]->GetInstance()->OnReceiveMessage(pcMessage, iMessageLen, message_container
        enable_transfer_message_ownership, has_message_ownership_transferred);
}

src/algorithm/instance.h

class Instance
{
    // ...
    //this funciton only enqueue, do nothing.
    // 修改代码
    // int OnReceiveMessage(const char * pcMessage, const int iMessageLen);
    // 参数定义参考NetWork::OnReceiveMessage
    int OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
         bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred);
    // ... 
};

src/algorithm/instance.cpp

// 修改代码
/*
int Instance :: OnReceiveMessage(const char * pcMessage, const int iMessageLen)                                                                                                         
{   
    m_oIOLoop.AddMessage(pcMessage, iMessageLen);

    return 0; 
}
*/

int Instance :: OnReceiveMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
     bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred)
{
    if (has_message_ownership_transferred) {
        PLErr("message ownership has been transferred");
        return -2;
    }

    return m_oIOLoop.AddMessage(pcMessage, iMessageLen, message_container
        enable_transfer_message_ownership, has_message_ownership_transferred);
}

src/algorithm/ioloop.h

class IOLoop : public Thread
{
    // ...
    // 修改代码
    // int AddMessage(const char * pcMessage, const int iMessageLen);
    // 参数定义参考NetWork::OnReceiveMessage
    int AddMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
         bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred);
    // ...
};

src/algorithm/ioloop.cpp

// 修改代码
// int IOLoop :: AddMessage(const char * pcMessage, const int iMessageLen)
int AddMessage(const char * pcMessage, const int iMessageLen, std::string* const message_container,
     bool enable_transfer_message_ownership, bool&  has_message_ownership_transferred)
{
    // 新增代码
    if (has_message_ownership_transferred) {
        PLErr("message ownership has been transferred");
        return -2;
    }

    m_oMessageQueue.lock();

    BP->GetIOLoopBP()->EnqueueMsg();

    if ((int)m_oMessageQueue.size() > QUEUE_MAXLENGTH)
    {
        BP->GetIOLoopBP()->EnqueueMsgRejectByFullQueue();

        PLGErr("Queue full, skip msg");
        m_oMessageQueue.unlock();
        return -2;
    }

    if (m_iQueueMemSize > MAX_QUEUE_MEM_SIZE)
    {
        PLErr("queue memsize %d too large, can't enqueue", m_iQueueMemSize);
        m_oMessageQueue.unlock();
        return -2;
    }

    // 修改代码
    // m_oMessageQueue.add(new string(pcMessage, iMessageLen));
    if (enable_transfer_message_ownership && message_container && pcMessage == message_container->data() && iMessageLen == message_container->size()) {
        m_oMessageQueue.add(message_container);
        has_message_ownership_transferred = true;
    } else { 
        m_oMessageQueue.add(new string(pcMessage, iMessageLen));
    }
        
    m_iQueueMemSize += iMessageLen;

    m_oMessageQueue.unlock();

    return 0;
}

src/communicate/udp.cpp

void UDPRecv :: run()
{
    m_bIsStarted = true;

    char sBuffer[65536] = {0};

    struct sockaddr_in addr;
    socklen_t addr_len = sizeof(struct sockaddr_in);
    memset(&addr, 0, sizeof(addr));

    while(true)
    {
        if (m_bIsEnd)
        {
            PLHead("UDPRecv [END]");
            return;
        }

        struct pollfd fd;
        int ret;

        fd.fd = m_iSockFD;
        fd.events = POLLIN;
        ret = poll(&fd, 1, 500);

        if (ret == 0 || ret == -1)
        {
            continue;
        }
        
        int iRecvLen = recvfrom(m_iSockFD, sBuffer, sizeof(sBuffer), 0,
                (struct sockaddr *)&addr, &addr_len);

        //printf("recvlen %d, buffer %s client %s\n",
                //iRecvLen, sBuffer, inet_ntoa(addr.sin_addr));
        
        BP->GetNetworkBP()->UDPReceive(iRecvLen);

        if (iRecvLen > 0)
        {
            // 修改代码   
            // m_poDFNetWork->OnReceiveMessage(sBuffer, iRecvLen);
            m_poDFNetWork->OnReceiveMessage(sBuffer, iRecvLen, nullptr, false, false);
        }   
    }   
}

模拟测试代码test_copy_receive_msg.cpp为了简化代码，省去了跨线程的队列操作部分，也省去了以上众多OnReceiveMessage函数调用。OriginalMethod模拟原方案（消息先拷贝到m_oReadCacheBuffer，再把m_oReadCacheBuffer拷贝到new出来的string对象），ImprovedMethod模拟改进后的方案（消息直接拷贝到new出来的string对象）。测试的消息长度分别为1/2/5/10/20/50/100/200/500/1000/2000/5000/10000/20000/50000/100000/200000/500000/1000000/2000000/5000000/10000000/20000000/50000000/100000000字节。

测试结果：

  • 无编译优化选项：消息长度小于等于1000字节时，OriginalMethod耗时更少。消息长度大于等于5000字节时，ImprovedMethod耗时更少。
  • O1编译优化选项：ImprovedMethod的耗时是OriginalMethod的0.2041 - 0.9981。
  • O2编译优化选项：ImprovedMethod的耗时是OriginalMethod的0.1740 - 0.9896。
  • O3编译优化选项：ImprovedMethod的耗时是OriginalMethod的0.1745 - 0.9650。

由于phxpaxos开启O2编译优化选项，应该使用改进后的方案。

test_copy_receive_msg.cpp代码：

#include <time.h>
#include <string.h>

#include <unordered_map>
#include <string>

#include <benchmark/benchmark.h>

char tmp_buf[104857600]; // 104857600 = 100 * 1024 * 1024

std::unordered_map<size_t, std::string> g_size2string;

static void Init(size_t n) {
    auto it = g_size2string.find(n);
    if (it != g_size2string.end()) {
        return;
    }

    srand(static_cast<unsigned int>(time(NULL)));
    std::string str;
    str.resize_and_overwrite(n, [](char* buf, std::size_t buf_len) {
                return buf_len;
            });

    char b = static_cast<char>(rand());
    for (char& c: str) {
        c = b;
        ++b;
    }

    g_size2string[n] = std::move(str);
}

static void OriginalMethod(benchmark::State& state) {
    int n = state.range(0);
    Init(n);

    volatile size_t total_size = 0;
    std::string& str = g_size2string[n];
    for (auto _: state) {
        memcpy(tmp_buf, str.data(), n);
        std::string* s = new std::string(tmp_buf, n);
        total_size += s->size();
        delete s;
        benchmark::ClobberMemory();
    }
}

BENCHMARK(OriginalMethod)->ArgsProduct({{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 50000, 100000, 200000, 500000, 1000000, 2000000, 5000000, 10000000, 20000000, 50000000, 100000000}});


static void ImprovedMethod(benchmark::State& state) {
    int n = state.range(0);
    Init(n);

    volatile size_t total_size = 0;
    std::string& str = g_size2string[n];
    for (auto _: state) {
        std::string* s = new std::string();
        s->resize_and_overwrite(n, [](char* buf, std::size_t buf_len) {
                    return buf_len;
                });
        memcpy(s->data(), str.data(), n);
        total_size += s->size();
        delete s;
        benchmark::ClobberMemory();
    }
}

BENCHMARK(ImprovedMethod)->ArgsProduct({{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 20000, 50000, 100000, 200000, 500000, 1000000, 2000000, 5000000, 10000000, 20000000, 50000000, 100000000}});

BENCHMARK_MAIN();

test_copy_receive_msg.cpp编译命令：

g++ -std=c++23 test_copy_receive_msg.cpp -lbenchmark

Activity

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    No labels
    No labels

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions