Skip to content

优化目标节点地址信息在发送消息路径上的传递方式 #281

Open
Open
优化目标节点地址信息在发送消息路径上的传递方式#281

Description

@dyx2025
dyx2025
opened on Jun 28, 2026

优化目标节点地址信息在发送消息路径上的传递方式

目前的传递方式是，先把64位的nodeid转为NodeInfo，再从NodeInfo获取ip和port，再把ip和port转换为sockaddr_in。其实64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略。传递ip和port过程中，带来了一些额外操作(详情请看下文的源码注释)。发送消息路径是热路径，有必要优化传递方式。

原代码路径：

src/communicate/communicate.h

class Communicate : public MsgTransport                                                     
{                                                                                           
    // ......
    // 64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略                                                                              
    int Send(const int iGroupIdx, const nodeid_t iNodeID,                                                                                                                               
            const NodeInfo & tNodeInfo, const std::string & sMessage, const int iSendType);
    // ......                                                                 
};

src/communicate/communicate.cpp

//  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略
int Communicate :: Send(const int iGroupIdx, const nodeid_t iNodeID, 
        const NodeInfo & oNodeInfo, const std::string & sMessage, const int iSendType)
{
    if ((int)sMessage.size() > MAX_VALUE_SIZE)
    {   
        BP->GetNetworkBP()->SendRejectByTooLargeSize();
        PLGErr("Message size too large %zu, max size %u, skip message", 
                sMessage.size(), MAX_VALUE_SIZE);
        return 0;
    }   

    BP->GetNetworkBP()->Send(sMessage);
        
    if (sMessage.size() > m_iUDPMaxSize || iSendType == Message_SendType_TCP)
    {   
        BP->GetNetworkBP()->SendTcp(sMessage);
        return m_poNetwork->SendMessageTCP(iGroupIdx, oNodeInfo.GetIP(), oNodeInfo.GetPort(), sMessage);
    }   
    else
    {   
        BP->GetNetworkBP()->SendUdp(sMessage);
        return m_poNetwork->SendMessageUDP(iGroupIdx, oNodeInfo.GetIP(), oNodeInfo.GetPort(), sMessage);
    }   
}

int Communicate :: SendMessage(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage, const int iSendType)
{
    return Send(iGroupIdx, iSendtoNodeID, NodeInfo(iSendtoNodeID), sMessage, iSendType);
}

int Communicate :: BroadcastMessage(const int iGroupIdx, const std::string & sMessage, const int iSendType)
{
    const std::set<nodeid_t> & setNodeInfo = m_poConfig->GetSystemVSM()->GetMembershipMap();
        
    for (auto & it : setNodeInfo)
    {   
        if (it != m_iMyNodeID)
        {   
            Send(iGroupIdx, it, NodeInfo(it), sMessage, iSendType);
        }   
    }   

    return 0;
}

int Communicate :: BroadcastMessageFollower(const int iGroupIdx, const std::string & sMessage, const int iSendType)
{
    const std::map<nodeid_t, uint64_t> & mapFollowerNodeInfo = m_poConfig->GetMyFollowerMap(); 
        
    for (auto & it : mapFollowerNodeInfo)
    {
        if (it.first != m_iMyNodeID)
        {
            Send(iGroupIdx, it.first, NodeInfo(it.first), sMessage, iSendType);
        }
    }
    
    PLGDebug("%zu node", mapFollowerNodeInfo.size());

    return 0;
}

int Communicate :: BroadcastMessageTempNode(const int iGroupIdx, const std::string & sMessage, const int iSendType)
{
    const std::map<nodeid_t, uint64_t> & mapTempNode = m_poConfig->GetTmpNodeMap(); 
    
    for (auto & it : mapTempNode)
    {
        if (it.first != m_iMyNodeID)
        {
            Send(iGroupIdx, it.first, NodeInfo(it.first), sMessage, iSendType);
        }
    }
    
    PLGDebug("%zu node", mapTempNode.size());

    return 0;
}

include/phxpaxos/network.h

class NetWork
{
    // ......
    //  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略
    virtual int SendMessageTCP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage) = 0;

    virtual int SendMessageUDP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage) = 0;
    // ......
};

src/communicate/dfnetwork.h

class DFNetWork : public NetWork
{
    // ......
    //  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略
    int SendMessageTCP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage);
    
    int SendMessageUDP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage);
    // ......
};

src/communicate/dfnetwork.cpp

//  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略
int DFNetWork :: SendMessageTCP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage)
{
    return m_oTcpIOThread.AddMessage(iGroupIdx, sIp, iPort, sMessage);
}

int DFNetWork :: SendMessageUDP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage)
{
    return m_oUDPSend.AddMessage(sIp, iPort, sMessage);
}

src/communicate/tcp/tcp.h

class TcpWrite : public Thread                                                              
{                                                                                                                                                                     
    // ......
    //  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略                                                                                            
    int AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage);
    // ......
}; 

class TcpIOThread 
{
    // ......
    //  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略 
    int AddMessage(const int iGroupIdx, const std::string & sIP, const int iPort, const std::string & sMessage);
    // ...... 
};

src/communicate/tcp/tcp.cpp

//  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略
int TcpWrite :: AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
{
    return m_oTcpClient.AddMessage(sIP, iPort, sMessage);                                                                                                                               
}
// ......
//  64位的nodeid可以直接转化为sockaddr_in，NodeInfo的相关转换可以省略
int TcpIOThread :: AddMessage(const int iGroupIdx, const std::string & sIP, const int iPort, const std::string & sMessage)                                                              
{
    int iIndex = iGroupIdx % (int)m_vecTcpWrite.size();
    return m_vecTcpWrite[iIndex]->AddMessage(sIP, iPort, sMessage);
}

src/communicate/tcp/tcp_client.h

class TcpClient
{
    // ......
    // 传递ip和port过程中，带来了一些额外操作
    int AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage);
    // .......                                                                                           
    // 传递ip和port过程中，带来了一些额外操作
    MessageEvent * GetEvent(const std::string & sIP, const int iPort);
    
    MessageEvent * CreateEvent(const uint64_t llNodeID, const std::string & sIP, const int iPort); 
    // ......    
};

src/communicate/tcp/tcp_client.cpp

// 传递ip和port过程中，带来了一些额外操作
int TcpClient :: AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
{
    //PLImp("ok");
    // 在GetEvent，每次都要从ip和port转换回64位的nodeid
    MessageEvent * poEvent = GetEvent(sIP, iPort);
    if (poEvent == nullptr)
    {   
        PLErr("no event created for this ip %s port %d", sIP.c_str(), iPort);
        return -1; 
    }   

    return poEvent->AddMessage(sMessage);
}

// 传递ip和port过程中，带来了一些额外操作
MessageEvent * TcpClient :: GetEvent(const std::string & sIP, const int iPort)
{
    // 每次都要从ip和port转换回64位的nodeid
    uint32_t iIP = (uint32_t)inet_addr(sIP.c_str());
    uint64_t llNodeID = (((uint64_t)iIP) << 32) | iPort;

    std::lock_guard<std::mutex> oLockGuard(m_oMutex);

    auto it = m_mapEvent.find(llNodeID);
    if (it != end(m_mapEvent))
    {   
        return it->second;
    }   

    return CreateEvent(llNodeID, sIP, iPort);
}

MessageEvent * TcpClient :: CreateEvent(const uint64_t llNodeID, const std::string & sIP, const int iPort)
{
    PLImp("start, ip %s port %d", sIP.c_str(), iPort);

    Socket oSocket;
    oSocket.setNonBlocking(true);
    oSocket.setNoDelay(true);
    SocketAddress oAddr(sIP, iPort);
    oSocket.connect(oAddr);

    MessageEvent * poEvent = new MessageEvent(MessageEventType_SEND, oSocket.detachSocketHandle(), 
            oAddr, m_poEventLoop, m_poNetWork);
    assert(poEvent != nullptr);

    m_mapEvent[llNodeID] = poEvent;
    m_vecEvent.push_back(poEvent);

    PLImp("ok, ip %s port %d", sIP.c_str(), iPort);

    return poEvent;
}

src/communicate/udp.h

class UDPSend : public Thread
{
    // ......
    // 传递ip和port过程中，带来了一些额外操作
    int AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage);

    struct QueueData
    {
        std::string m_sIP;
        int m_iPort;
        std::string m_sMessage;
    };

private:
    void SendMessage(const std::string & sIP, const int iPort, const std::string & sMessage);
    // ......
};

src/communicate/udp.cpp

void UDPSend :: SendMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
{
    // 每次都要从ip转换回nodeid的高32位
    struct sockaddr_in addr;
    int addr_len = sizeof(struct sockaddr_in);
    memset(&addr, 0, sizeof(addr));
    
    addr.sin_family = AF_INET;
    addr.sin_port = htons(iPort);
    addr.sin_addr.s_addr = inet_addr(sIP.c_str());
    
    int ret = sendto(m_iSockFD, sMessage.data(), (int)sMessage.size(), 0, (struct sockaddr *)&addr, addr_len);
    if (ret > 0)
    {   
        BP->GetNetworkBP()->UDPRealSend(sMessage);
    }
}

void UDPSend :: run()
{
    m_bIsStarted = true;
    
    while(true)
    {   
        QueueData * poData = nullptr;
        
        m_oSendQueue.lock();
        
        bool bSucc = m_oSendQueue.peek(poData, 1000);
        if (bSucc)
        {   
            m_oSendQueue.pop();
        }
        
        m_oSendQueue.unlock();
        
        if (poData != nullptr)
        {
            // 在SendMessage中， 每次都要从ip转换回nodeid的高32位 
            SendMessage(poData->m_sIP, poData->m_iPort, poData->m_sMessage);
            delete poData;
        }
        
        if (m_bIsEnd)
        {   
            PLHead("UDPSend [END]");
            return;
        }
    }
}

int UDPSend :: AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
{
    m_oSendQueue.lock();

    if ((int)m_oSendQueue.size() > UDP_QUEUE_MAXLEN)
    {
        BP->GetNetworkBP()->UDPQueueFull();
        //PLErr("queue length %d too long, can't enqueue", m_oSendQueue.size());

        m_oSendQueue.unlock();

        return -2;
    }

    // 拷贝ip和port，最少需要拷贝9个字节（合法的ip字符串最少需要7个字符，类型为int的port最少需要2个字节），ip字符串还有一些构造函数和析构函数的开销。拷贝64为nodeid，只需要8字节
    QueueData * poData = new QueueData;
    poData->m_sIP = sIP;
    poData->m_iPort = iPort;
    poData->m_sMessage = sMessage;

    m_oSendQueue.add(poData);
    m_oSendQueue.unlock();

    return 0;
}

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

class Communicate : public MsgTransport                                                     
{                                                                                           
    // ......
    /*                                                                              
    int Send(const int iGroupIdx, const nodeid_t iNodeID,                                                                                                                               
            const NodeInfo & tNodeInfo, const std::string & sMessage, const int iSendType);
    */

    // 修改代码
    int Send(const int iGroupIdx, const nodeid_t iNodeID,                                                                                                                               
            const std::string & sMessage, const int iSendType);

    // ......                                                                 
};

src/communicate/communicate.cpp

/*
int Communicate :: Send(const int iGroupIdx, const nodeid_t iNodeID, 
        const NodeInfo & oNodeInfo, const std::string & sMessage, const int iSendType)
*/
// 修改代码
int Communicate :: Send(const int iGroupIdx, const nodeid_t iNodeID, 
        const std::string & sMessage, const int iSendType)
{
    if ((int)sMessage.size() > MAX_VALUE_SIZE)
    {   
        BP->GetNetworkBP()->SendRejectByTooLargeSize();
        PLGErr("Message size too large %zu, max size %u, skip message", 
                sMessage.size(), MAX_VALUE_SIZE);
        return 0;
    }   

    BP->GetNetworkBP()->Send(sMessage);
        
    if (sMessage.size() > m_iUDPMaxSize || iSendType == Message_SendType_TCP)
    {   
        BP->GetNetworkBP()->SendTcp(sMessage);
        // return m_poNetwork->SendMessageTCP(iGroupIdx, oNodeInfo.GetIP(), oNodeInfo.GetPort(), sMessage);
        // 修改代码
        return m_poNetwork->SendMessageTCP(iGroupIdx, iNodeID, sMessage);
    }   
    else
    {   
        BP->GetNetworkBP()->SendUdp(sMessage);
        // return m_poNetwork->SendMessageUDP(iGroupIdx, oNodeInfo.GetIP(), oNodeInfo.GetPort(), sMessage);
        // 修改代码
        return m_poNetwork->SendMessageUDP(iGroupIdx, iNodeID, sMessage);
    }   
}

int Communicate :: SendMessage(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage, const int iSendType)
{
    // return Send(iGroupIdx, iSendtoNodeID, NodeInfo(iSendtoNodeID), sMessage, iSendType);
    // 修改代码
    return Send(iGroupIdx, iSendtoNodeID, sMessage, iSendType);
}

int Communicate :: BroadcastMessage(const int iGroupIdx, const std::string & sMessage, const int iSendType)
{
    const std::set<nodeid_t> & setNodeInfo = m_poConfig->GetSystemVSM()->GetMembershipMap();
        
    for (auto & it : setNodeInfo)
    {   
        if (it != m_iMyNodeID)
        {   
            // Send(iGroupIdx, it, NodeInfo(it), sMessage, iSendType);
            // 修改代码
            Send(iGroupIdx, it, sMessage, iSendType);
        }   
    }   

    return 0;
}

int Communicate :: BroadcastMessageFollower(const int iGroupIdx, const std::string & sMessage, const int iSendType)
{
    const std::map<nodeid_t, uint64_t> & mapFollowerNodeInfo = m_poConfig->GetMyFollowerMap(); 
        
    for (auto & it : mapFollowerNodeInfo)
    {
        if (it.first != m_iMyNodeID)
        {
            // Send(iGroupIdx, it.first, NodeInfo(it.first), sMessage, iSendType);
            // 修改代码
            Send(iGroupIdx, it.first, sMessage, iSendType);
        }
    }
    
    PLGDebug("%zu node", mapFollowerNodeInfo.size());

    return 0;
}

int Communicate :: BroadcastMessageTempNode(const int iGroupIdx, const std::string & sMessage, const int iSendType)
{
    const std::map<nodeid_t, uint64_t> & mapTempNode = m_poConfig->GetTmpNodeMap(); 
    
    for (auto & it : mapTempNode)
    {
        if (it.first != m_iMyNodeID)
        {
            Send(iGroupIdx, it.first, NodeInfo(it.first), sMessage, iSendType);
            // 修改代码
            Send(iGroupIdx, it.first, sMessage, iSendType);
        }
    }
    
    PLGDebug("%zu node", mapTempNode.size());

    return 0;
}

include/phxpaxos/network.h

class NetWork
{
    // ......
    /*
    virtual int SendMessageTCP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage) = 0;

    virtual int SendMessageUDP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage) = 0;
    */

    // 修改代码
    virtual int SendMessageTCP(const int iGroupIdx, const nodeid_t iSendtoNodeID const std::string & sMessage) = 0;

    virtual int SendMessageUDP(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage) = 0;
    // ......
};

src/communicate/dfnetwork.h

class DFNetWork : public NetWork
{
    // ......
    /*
    int SendMessageTCP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage);
    
    int SendMessageUDP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage);
    */

    // 修改代码
    int SendMessageTCP(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage);
    
    int SendMessageUDP(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage);
    // ......
};

src/communicate/dfnetwork.cpp

/*
int DFNetWork :: SendMessageTCP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage)
{
    return m_oTcpIOThread.AddMessage(iGroupIdx, sIp, iPort, sMessage);
}

int DFNetWork :: SendMessageUDP(const int iGroupIdx, const std::string & sIp, const int iPort, const std::string & sMessage)
{
    return m_oUDPSend.AddMessage(sIp, iPort, sMessage);
}
*/

// 修改代码
int DFNetWork :: SendMessageTCP(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage)
{
    return m_oTcpIOThread.AddMessage(iGroupIdx, iSendtoNodeID, sMessage);
}

int DFNetWork :: SendMessageUDP(const int iGroupIdx, const nodeid_t iSendtoNodeID, const std::string & sMessage)
{
    return m_oUDPSend.AddMessage(iSendtoNodeID, sMessage);

src/communicate/tcp/tcp.h

class TcpWrite : public Thread                                                              
{                                                                                                                                                                     
    // ......                                                                                          
    // int AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage);
    // 修改代码
    int AddMessage(const nodeid_t iSendtoNodeID, const std::string & sMessage);
    // ......
}; 

class TcpIOThread 
{
    // ......
    // int AddMessage(const int iGroupIdx, const std::string & sIP, const int iPort, const std::string & sMessage);
    // 修改代码
    int AddMessage(const int iGroupIdx, const nodeid_t iSendtoNodeID,  const std::string & sMessage);
    // ...... 
};

src/communicate/tcp/tcp.cpp

/*
int TcpWrite :: AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
{
    return m_oTcpClient.AddMessage(sIP, iPort, sMessage);                                                                                                                               
}
*/

// 修改代码
int TcpWrite :: AddMessage(const nodeid_t iSendtoNodeID, const std::string & sMessage)
{
    return m_oTcpClient.AddMessage(iSendtoNodeID, sMessage);                                                                                                                               
}

// ......
/*
int TcpIOThread :: AddMessage(const int iGroupIdx, const std::string & sIP, const int iPort, const std::string & sMessage)                                                              
{
    int iIndex = iGroupIdx % (int)m_vecTcpWrite.size();
    return m_vecTcpWrite[iIndex]->AddMessage(sIP, iPort, sMessage);
} 
*/

// 修改代码
int TcpIOThread :: AddMessage(const nodeid_t iSendtoNodeID, const std::string & sMessage)                                                              
{
    int iIndex = iGroupIdx % (int)m_vecTcpWrite.size();
    return m_vecTcpWrite[iIndex]->AddMessage(iSendtoNodeID, sMessage);
}

src/communicate/tcp/tcp_client.h

class TcpClient
{
    // ......
    // int AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage);
    // 修改代码
    int AddMessage(const nodeid_t iSendtoNodeID, const std::string & sMessage);
    // .......                                                                                           
    // MessageEvent * GetEvent(const std::string & sIP, const int iPort);
    
    // MessageEvent * CreateEvent(const uint64_t llNodeID, const std::string & sIP, const int iPort); 
    // 修改代码
    MessageEvent * GetEvent(const nodeid_t iSendtoNodeID);

    MessageEvent * CreateEvent(const uint64_t llNodeID);                   
    // ......    
};

src/communicate/tcp/tcp_client.cpp

// int TcpClient :: AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
// 修改代码
int TcpClient :: AddMessage(const nodeid_t iSendtoNodeID, const std::string & sMessage)
{
    //PLImp("ok");
    // MessageEvent * poEvent = GetEvent(sIP, iPort);
    // 修改代码
    MessageEvent * poEvent = GetEvent(iSendtoNodeID);
    if (poEvent == nullptr)
    {   
        // PLErr("no event created for this ip %s port %d", sIP.c_str(), iPort);
        // 修改代码
        PLErr("no event created for this nodeid %" PRIu64 "", iSendtoNodeID);
        return -1; 
    }   

    return poEvent->AddMessage(sMessage);
}

// MessageEvent * TcpClient :: GetEvent(const std::string & sIP, const int iPort)
// 修改代码
MessageEvent * TcpClient :: GetEvent(const nodeid_t llNodeID)
{
    /*
    uint32_t iIP = (uint32_t)inet_addr(sIP.c_str());
    uint64_t llNodeID = (((uint64_t)iIP) << 32) | iPort;
    */

    std::lock_guard<std::mutex> oLockGuard(m_oMutex);

    auto it = m_mapEvent.find(llNodeID);
    if (it != end(m_mapEvent))
    {   
        return it->second;
    }

    // 修改代码
    return CreateEvent(llNodeID);
}

// MessageEvent * TcpClient :: CreateEvent(const uint64_t llNodeID, const std::string & sIP, const int iPort)
// 修改代码
MessageEvent * TcpClient :: CreateEvent(const uint64_t llNodeID)
{
    // PLImp("start, ip %s port %d", sIP.c_str(), iPort);
    // 修改代码
    PLImp("start, nodeid %" PRIu64 "", llNodeID);

    Socket oSocket;
    oSocket.setNonBlocking(true);
    oSocket.setNoDelay(true);
    // SocketAddress oAddr(sIP, iPort);
    // 修改代码

    sockaddr_in addr;
    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    addr.sin_port = htons(llNodeID & 0xffffffff);
    addr.sin_addr.s_addr = llNodeID >> 32;
    SocketAddress oAddr(addr);

    oSocket.connect(oAddr);

    MessageEvent * poEvent = new MessageEvent(MessageEventType_SEND, oSocket.detachSocketHandle(), 
            oAddr, m_poEventLoop, m_poNetWork);
    assert(poEvent != nullptr);

    m_mapEvent[llNodeID] = poEvent;
    m_vecEvent.push_back(poEvent);

    // PLImp("ok, ip %s port %d", sIP.c_str(), iPort);
    // 修改代码
    PLImp("ok, nodeid %" PRIu64 "", llNodeID);

    return poEvent;
}

src/communicate/udp.h

class UDPSend : public Thread
{
    // ......
 /*
    int AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage);

    struct QueueData
    {
        std::string m_sIP;
        int m_iPort;
        std::string m_sMessage;
    };

private:
    void SendMessage(const std::string & sIP, const int iPort, const std::string & sMessage);
*/

// 修改代码
    int AddMessage(const nodeid_t llNodeID, const std::string & sMessage);

    struct QueueData
    {
        nodeid_t m_llNodeID;
        std::string m_sMessage;
    };

private:
    void SendMessage(const nodeid_t llNodeID, const std::string & sMessage);   

};

src/communicate/udp.cpp

// void UDPSend :: SendMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
// 修改代码
void UDPSend :: SendMessage(const nodeid_t llNodeID, const std::string & sMessage)
{
    struct sockaddr_in addr;
    int addr_len = sizeof(struct sockaddr_in);
    memset(&addr, 0, sizeof(addr));

    // 新增代码
    int iPort = llNodeID & (0xffffffff);
    
    addr.sin_family = AF_INET;
    addr.sin_port = htons(iPort);
    // addr.sin_addr.s_addr = inet_addr(sIP.c_str());
    // 修改代码
     addr.sin_addr.s_addr = llNodeID >> 32;
    
    int ret = sendto(m_iSockFD, sMessage.data(), (int)sMessage.size(), 0, (struct sockaddr *)&addr, addr_len);
    if (ret > 0)
    {   
        BP->GetNetworkBP()->UDPRealSend(sMessage);
    }
}

void UDPSend :: run()
{
    m_bIsStarted = true;
    
    while(true)
    {   
        QueueData * poData = nullptr;
        
        m_oSendQueue.lock();
        
        bool bSucc = m_oSendQueue.peek(poData, 1000);
        if (bSucc)
        {   
            m_oSendQueue.pop();
        }
        
        m_oSendQueue.unlock();
        
        if (poData != nullptr)
        { 
            // SendMessage(poData->m_sIP, poData->m_iPort, poData->m_sMessage);
            // 修改代码
            SendMessage(poData->m_llNodeID, poData->m_sMessage);
            delete poData;
        }
        
        if (m_bIsEnd)
        {   
            PLHead("UDPSend [END]");
            return;
        }
    }
}

// int UDPSend :: AddMessage(const std::string & sIP, const int iPort, const std::string & sMessage)
// 修改代码
int UDPSend :: AddMessage(const nodeid_t llNodeID, const std::string & sMessage)
{
    m_oSendQueue.lock();

    if ((int)m_oSendQueue.size() > UDP_QUEUE_MAXLEN)
    {
        BP->GetNetworkBP()->UDPQueueFull();
        //PLErr("queue length %d too long, can't enqueue", m_oSendQueue.size());

        m_oSendQueue.unlock();

        return -2;
    }

    QueueData * poData = new QueueData;
    /*
    poData->m_sIP = sIP;
    poData->m_iPort = iPort;
    */
    // 修改代码
    poData->m_llNodeID = llNodeID;

    poData->m_sMessage = sMessage;

    m_oSendQueue.add(poData);
    m_oSendQueue.unlock();

    return 0;
}

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