qasyntcpserver.cpp

BCB下的高效开发包。包含大并发的网络库

#include "QAsynTcpServer.h"
#include "../QNetBase/QNetBase.h"
#include "../QStdlib/QStdlib.h"
#include "Winsock2.h"
#include "Windows.h"
#include "../QDEBUGLOGER/QDebugLoger.h"
using namespace N_QAsynTcpServer;
//------------------------------------------------------------------------------
IQAsynTcpServer* __stdcall IQAsynTcpServer::Create(void)
{
        return new QAsynTcpServer;
}
//------------------------------------------------------------------------------
__stdcall QAsynTcpServer::QAsynTcpServer()
{
        this->SetTimeOut(DEFAULT_TIMEOUT);
}
//------------------------------------------------------------------------------
__stdcall QAsynTcpServer::~QAsynTcpServer()
{
        //全面清理
        SOCKET_MAP::const_iterator iter = m_SocketMap.begin();
        const SOCKET_MAP::const_iterator iend = m_SocketMap.end();
        for(; iter != iend; ++iter) {
                _QClose(iter->first);
        }
        m_SocketMap.clear();
}
//------------------------------------------------------------------------------
void __stdcall QAsynTcpServer::SetTimeOut(unsigned int TimeOutMs)
{
        m_TimeOut = TimeOutMs;
}
//------------------------------------------------------------------------------
bool __stdcall QAsynTcpServer::Send(int Sock, const std::string& Data)
{
        SOCKET_MAP::iterator iter = m_SocketMap.find(Sock);
        if (m_SocketMap.end() == iter)
                return false;
        iter->second.SendStoreBuf += Data;
        return true;
}
//------------------------------------------------------------------------------
void __stdcall QAsynTcpServer::Recv(int Sock, std::string* pData)
{
        //QDEBUG_LOG("RECV1",1);
        if (!pData)
                return;
        //QDEBUG_LOG("RECV2",1);
        SOCKET_MAP::iterator iter = m_SocketMap.find(Sock);
        if (m_SocketMap.end() == iter)
                return;
        //QDEBUG_LOG("RECV3",1);
        if (iter->second.RecvBuf.empty()) {
                //QDEBUG_LOG("RECV3+",1);
                return;
        }
        //QDEBUG_LOG("RECV4",1);
        pData->swap(iter->second.RecvBuf);
        iter->second.RecvBuf.clear();
        //QDEBUG_LOG("RECV5",1);
}
//------------------------------------------------------------------------------
unsigned int __stdcall QAsynTcpServer::GetSendStore(int Sock)const
{
        const SOCKET_MAP::const_iterator iter = m_SocketMap.find(Sock);
        if (m_SocketMap.end() == iter)
                return 0;
        return iter->second.SendBuf.empty() && iter->second.SendStoreBuf.empty()
            && 0 == iter->second.SendBufLoc;
}
//------------------------------------------------------------------------------
void __stdcall QAsynTcpServer::Process(void)
{
        this->ProcessCheckRecv();
        this->ProcessCheckSend();
        this->ProcessTimeOut(false);
}
//------------------------------------------------------------------------------
void __stdcall QAsynTcpServer::ShutDown(int Sock)
{
        SOCKET_MAP::iterator iter = m_SocketMap.find(Sock);
        if (m_SocketMap.end() == iter)
                return;
        iter->second.GraceShutDown = true;
}
//------------------------------------------------------------------------------
void __stdcall QAsynTcpServer::ShutDownForce(int Sock)
{
        SOCKET_MAP::iterator iter = m_SocketMap.find(Sock);
        if (m_SocketMap.end() == iter)
                return;
        iter->second.Alive = false;
        _QShutdown(iter->first);
}
//------------------------------------------------------------------------------
bool __stdcall QAsynTcpServer::Insert(int Sock)
{
        if (this->IsExist(Sock))
                return false;

        ITEM Item;
        Item.SendBufLoc = 0;
        Item.Alive = true;
        Item.GraceShutDown = false;
        Item.LastTime = GetTickCount();
        m_SocketMap[Sock] = Item;

        _QSetBlock(Sock, false);//非阻塞模式
        return true;
}
//------------------------------------------------------------------------------
void __stdcall QAsynTcpServer::Delete(int Sock)
{
        if (!this->IsExist(Sock))
                return;
        m_SocketMap.erase(Sock);//注意先清除再CloseSocket,防止相同值新句柄
        _QClose(Sock);
}
//------------------------------------------------------------------------------
bool __stdcall QAsynTcpServer::IsAlive(int Sock)const
{
        const SOCKET_MAP::const_iterator iter = m_SocketMap.find(Sock);
        if (m_SocketMap.end() == iter){
                return false;
        }
        return iter->second.Alive;
}
//------------------------------------------------------------------------------
bool __stdcall QAsynTcpServer::IsExist(int Sock)const
{
        return m_SocketMap.find(Sock) != m_SocketMap.end();
}
//------------------------------------------------------------------------------
unsigned int __stdcall QAsynTcpServer::GetCount(void)const
{
        return m_SocketMap.size();
}
//------------------------------------------------------------------------------
void QAsynTcpServer::ProcessCheckRecv(void)
{
        SOCKET_MAP::const_iterator iter = m_SocketMap.begin();
        const SOCKET_MAP::const_iterator iend = m_SocketMap.end();
        m_SelectSrc.clear();
        for (; iter != iend; ++iter) {
                if (!iter->second.Alive)
                        continue;
                if (!iter->second.RecvBuf.empty())//数据没取走则不继续读
                        continue;
                m_SelectSrc.push_back(iter->first);
        }
        m_SelectOut.clear();
        _QSelectRead(m_SelectSrc, &m_SelectOut);
        std::vector<int>::const_iterator iter2 = m_SelectOut.begin();
        const std::vector<int>::const_iterator iend2 = m_SelectOut.end();
        for(; iter2 != iend2; ++iter2) {
                this->ProcessRecv(*iter2);
        }
        m_SelectSrc.clear();
        m_SelectOut.clear();
}
//------------------------------------------------------------------------------
void QAsynTcpServer::ProcessCheckSend(void)
{
        SOCKET_MAP::const_iterator iter = m_SocketMap.begin();
        const SOCKET_MAP::const_iterator iend = m_SocketMap.end();
        m_SelectSrc.clear();
        for (; iter != iend; ++iter) {
                if (!iter->second.Alive)
                        continue;
                if (iter->second.SendStoreBuf.empty() && iter->second.SendBuf.empty()) {//两个发送缓冲都空则不需要发数据
                        if(iter->second.GraceShutDown) {   //如果标记了优雅关闭,则此时应该shutdown了
                                this->ShutDownForce(iter->first);
                        }
                        continue;
                }
                m_SelectSrc.push_back(iter->first);
        }
        m_SelectOut.clear();
        _QSelectWrite(m_SelectSrc, &m_SelectOut);        //执行select
        std::vector<int>::const_iterator iter2 = m_SelectOut.begin();
        const std::vector<int>::const_iterator iend2 = m_SelectOut.end();
        for(; iter2 != iend2; ++iter2) {
                this->ProcessSend(*iter2);
        }
        m_SelectSrc.clear();
        m_SelectOut.clear();
}
//------------------------------------------------------------------------------
void QAsynTcpServer::ProcessTimeOut(bool DoNow)
{
        if (!DoNow && GetRealTickDifference(m_TimeOutCheckTime, GetTickCount()) < TIMEOUT_CHECK_TIME)
                return;
        m_TimeOutCheckTime = GetTickCount();
        
        SOCKET_MAP::const_iterator iter = m_SocketMap.begin();
        const SOCKET_MAP::const_iterator iend = m_SocketMap.end();
        for(; iter != iend; ++iter) {
                if (!iter->second.Alive)
                        continue;
                if (GetRealTickDifference(iter->second.LastTime, GetTickCount()) < m_TimeOut)
                        continue;
                this->ShutDownForce(iter->first);
        }
}
//------------------------------------------------------------------------------
void QAsynTcpServer::ProcessRecv(int Sock)
{
        SOCKET_MAP::iterator iter = m_SocketMap.find(Sock);
        if ( m_SocketMap.end() == iter) {
                MsgBox("错误:不在容器中","void QAsynTcpServer::ProcessRecv(int Sock)");
        }
        if (!iter->second.RecvBuf.empty()) {
                MsgBox("错误:缓冲非空","void QAsynTcpServer::ProcessRecv(int Sock)");
        }

        //iter->second.RecvBuf.resize(RECV_BUF_SIZE);
        char Buf[RECV_BUF_SIZE];
        int Ret = recv(iter->first, Buf, sizeof(Buf), 0);
        if (0 == Ret) {                 //断开
                iter->second.Alive = false;
                shutdown(iter->first, SD_BOTH);
                iter->second.RecvBuf.clear();
                return;
        }else if (SOCKET_ERROR == Ret) {//出错
                assert(WSAGetLastError() != WSAEWOULDBLOCK);//这种错误值不可能存在,除非没有select就进来了
                iter->second.Alive = false;
                shutdown(iter->first, SD_BOTH);
                iter->second.RecvBuf.clear();
                return;
        }
        //iter->second.RecvBuf.resize(Ret);
        std::string tmp(Buf, Ret);
        iter->second.RecvBuf.swap(tmp);
        iter->second.LastTime = GetTickCount();
        return;
}
//------------------------------------------------------------------------------
void QAsynTcpServer::ProcessSend(int Sock)
{
        SOCKET_MAP::iterator iter = m_SocketMap.find(Sock);
        if ( m_SocketMap.end() == iter) {
                MsgBox("错误:不在容器中","void QAsynTcpServer::ProcessSend(int Sock)");
        }

        //QDEBUG_LOG(iter->second.SendStoreBuf.c_str(),1);

        if (iter->second.SendBuf.empty()) {
                if (!iter->second.SendStoreBuf.empty()) {
                        iter->second.SendBuf.swap(iter->second.SendStoreBuf);
                }else{
                        assert("QAsynTcpServer::DoSocketSendm, Sendbufs double empty.assert down.");
                }
        }

        //QDEBUG_LOG((QIntToStr(iter->second.SendBufLoc) + QIntToStr(iter->second.SendBuf.size())).c_str(),1);
        assert(iter->second.SendBufLoc < iter->second.SendBuf.size());
        int Ret = send(iter->first, &iter->second.SendBuf[iter->second.SendBufLoc]
                , iter->second.SendBuf.size() - iter->second.SendBufLoc, 0);
        if (0 == Ret) {//断开
                iter->second.Alive = false;
                shutdown(iter->first, SD_BOTH);
                return;
        }else if (SOCKET_ERROR == Ret) {
                if (WSAGetLastError() != WSAEWOULDBLOCK) { //错误
                        iter->second.Alive = false;
                        shutdown(iter->first, SD_BOTH);
                        return;
                }else{
                        //正常的,发送了一部分数据
                }
        }

        iter->second.SendBufLoc += Ret;
        assert(iter->second.SendBufLoc >= iter->second.SendBuf.size());//不可能超出
        if (iter->second.SendBuf.size() == iter->second.SendBufLoc) {
                iter->second.SendBufLoc = 0;
                iter->second.SendBuf.clear();
        }
        iter->second.LastTime = GetTickCount();
        return;
}
//------------------------------------------------------------------------------