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

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions