(参考Ghost Cheng的模型封装)完成端口模型.封装的不好.

lustskyboy

//MemoryPool.h
#pragma once
#include <iostream>
#include <list>
#include <queue>
#include <windows.h>
using namespace std;

//关键区锁
class CLock
{
    CRITICAL_SECTION _crisection;
public:
    CLock()   
    {
        InitializeCriticalSection( &_crisection );
    }
    ~CLock()   
    {
        DeleteCriticalSection( &_crisection );     
    }
    void lock()
    {
        EnterCriticalSection( &_crisection );      
    }
    void unlock()
    {
        LeaveCriticalSection( &_crisection );      
    }
};
//链表模板
template <class T>
class CList : public list<T>
{
   
public:
    CList(void){};
    virtual ~CList(void){};
    CLock _guard;

    void Clear()
    {
        _guard.lock();
        clear();
        _guard.unlock();
    }
    void Remove(T lpData)
    {
        _guard.lock();
        remove(lpData);
        _guard.unlock();
    }
    void Push_Back(T lpData)
    {
        _guard.lock();
        push_back(lpData);
        _guard.unlock();
    }
    DWORD Size()
    {
        DWORD dwSize = 0;
        _guard.lock();
        dwSize = size();
        _guard.unlock();
        return dwSize;
    };
};



template <class particle>
class particle_allocator
{
    enum
    {
        chunk_size =1024
    };
    typedef unsigned char byte;

    list<byte *>  _chunks;
    queue<particle *> _free_list;

    CLock _guard;
public:
    particle_allocator()
    {
        _used_particle_number = 0;
        _free_particle_number = 0;
    }
    ~particle_allocator()
    {
    }
    void Destroy()
    {
        while(!_free_list.empty())
        {
            _free_list.pop();
        }
        for(std::list<byte *>::iterator all_iter = _chunks.begin();all_iter!=_chunks.end();++all_iter)
        {
            delete [](*all_iter);
        }
        _chunks.clear();
        _used_particle_number = 0;
        _free_particle_number = 0;
    }
    particle* alloc_particle()
    {
        _guard.lock();
        byte *memory;
        if(_free_list.empty())
        {
            memory = new byte[chunk_size*sizeof(particle)];
            _chunks.push_front(memory);
            for(int i =0;i< chunk_size;i++,_free_particle_number++)
            {
                particle *newnode = (particle *)(memory + i*sizeof(particle));
                _free_list.push(newnode);

            }
        }
        particle  *redata = NULL;
        if(_free_list.size())
        {
            redata = _free_list.front();
            _free_list.pop();
        }
        ++_used_particle_number;
        --_free_particle_number;
        _guard.unlock();
        return redata;
    }
    void free_particle(particle  *p)
    {
        _guard.lock();
        _free_list.push(p);
        --_used_particle_number;
        ++_free_particle_number;
        _guard.unlock();
    }
    void getallocator()
    {
        cout << "_used_particle_number is "
            << _used_particle_number
            << " _free_particle_number is "
            << _free_particle_number << endl;
    }
private:
    size_t _used_particle_number;
    size_t _free_particle_number;
};

lustskyboy

//#include CIOCPServer.h
#pragma once
#include <winsock2.h>
#include <Mswsock.h>
#include <process.h>
#include ".\CMemoryPool.h"
#define NET_DATA_LONGTH 2048
#define USE_DATA_LONGTH 4096
#define IP_MAX_LEN                15                //IP地址的最大长度，15个字节，255.255.255.255
#define IP_STRING_LEN                16                //存放IP地址的字符串的长度，最后一位为0x00

typedef unsigned (WINAPI *PBEGINTHREADX_THREADFUN)(LPVOID lpThreadParameter);        //线程函数原型

//只有连接断开，或者服务器关闭的时候才释放
typedef struct tagPER_HANDLE_DATA
{
    SOCKET  Socket;
    DWORD   m_dwIP;                                //Socket的IP地址
    USHORT  m_usPort;                                //Socket的端口号
    DWORD   m_dwLastTime;                        //最后一次正常通讯的时间
    BYTE*   m_lpBufBegin;                        //拼包缓冲区头指针
    BYTE*   m_lpBufEnd;                                //拼包缓冲区尾指针
    BYTE    m_arrayDataBuf[USE_DATA_LONGTH];        //拼包缓冲区
}PER_HANDLE_DATA, *LPPER_HANDLE_DATA;

// 单I/O操作数据。每次收发数据的时候
//收发数据操作完成数之后释放。
//需要注意的是OVERLAPPED Overlapped一定要放在最前
typedef struct tagPER_IO_DATA
{
    OVERLAPPED        Overlapped;
    WSABUF  wDataBuf;
    char    arrayDataBuf[NET_DATA_LONGTH];
    DWORD   dwBytes;
    DWORD   dwOpCode;
    DWORD   dwFlags;
    SOCKET  Socket;
    LPPER_HANDLE_DATA lpSession;
}PER_IO_DATA, *LPPER_IO_DATA;

class CIOCPServer
{
    enum
    {
        OP_READ,
        OP_WRITE,
        OP_ACCEPT,
    };
    //监听socket

    SOCKET m_ListenSocket;
    //完成端口句柄
    HANDLE m_hCompletionPort;
    //创建监听socket
    bool    CreateSocket(void);
    //绑定指定端口与ip
    int            Bind(LPCTSTR lpszHostAddress,const USHORT usPort);
    //开始监听
    int            Listen(int nBacklog);
    //关闭监听socket
    void    Close(void);
    void CloseAllClient(void);
    //得到AcceptEx地址
    void GetAcceptEx(void);
    //得到GetAcceptExSockaddrs地址
    void GetAcceptExSockaddrsFun(void);
    //完成端口工作线程数

    DWORD  m_dwIOCPThreadCount;
    //工作线程句柄数组指针
    HANDLE *m_arrayIOCPThreadHandle;
    //工作线程
    static DWORD WINAPI IOCPWorkThread(LPVOID lpParam);
    //创建工作线程
    bool CreateWorkThread(void);
    void CloseWorkThread(void);
    //连接信号同步

    HANDLE  m_hAcceptExOverEvent;//AcceptEx动作耗尽的信号
    //连接线程句柄
    HANDLE  m_hAcceptEx;
    //结束标记
    bool    m_bAcceptExCreateThread;
    //AcceptEx函数指针
    LPFN_ACCEPTEX   m_lpAcceptExFun;
    //GetAcceptExSockaddrs函数指针
    LPFN_GETACCEPTEXSOCKADDRS        m_lpGetAcceptExSockaddrsFun;
    //开启发送接受消息线程
    bool CreateAcceptExThread(void);
    void CloseAcceptExThread(void);
    //发送接受消息线程
    static DWORD WINAPI AcceptExThread(LPVOID lpParam);
    //创建一个接受消息
    void CreateAcceptEx(const DWORD dwPostCount = 10);
    //得到接受socket的ip与端口
    void GetAcceptExSockaddrsInfo(LPPER_IO_DATA lpOvplus, DWORD* lpdwIP, USHORT* lpusPort);
    //重叠数据内存池
    particle_allocator&ltER_IO_DATA> m_Net_Data;
    //连接池
    particle_allocator<PER_HANDLE_DATA> m_Client_Context;
    //当前连接客户
    CList<LPPER_HANDLE_DATA> m_listSession;
private:
    void OnIOCPAccept(const DWORD dwByteCount, LPPER_IO_DATA lpOvplus);
    void OnIOCPRecvFromClient(const DWORD dwByteCount, LPPER_HANDLE_DATA lpSession, LPPER_IO_DATA lpOvplus);
    //创建一个接收的 I/O 异步请求
    bool CreateRecv(LPPER_IO_DATA lpOvplus);

    //创建一个发送的 I/O 异步请求
    bool CreateSend(LPPER_IO_DATA lpOvplus);

public:
    CIOCPServer(void);
    virtual ~CIOCPServer(void);
public:
    bool BeginServer(void);
    void StopServer(void);
    //IP过滤～
    virtual bool OnIpFilter(const DWORD dwClientIP){return true;};
    virtual void OnAccept(void){};
    virtual void OnDisconnectFromClient(const LPPER_HANDLE_DATA lpSession){};

};

lustskyboy

#include ".\CIOCPServer.h"

#pragma comment(lib,"ws2_32.lib")

CIOCPServer::CIOCPServer(void)
{
    //初始化Winsock
    WSADATA wsd;
    WSAStartup(MAKEWORD(2, 2), &wsd);
    SYSTEM_INFO SystemInfo;
    GetSystemInfo(&SystemInfo);
    m_dwIOCPThreadCount = 2*SystemInfo.dwNumberOfProcessors+2;
   
}
CIOCPServer::~CIOCPServer(void)
{

}
bool CIOCPServer::CreateSocket(void)
{
    m_ListenSocket = WSASocket(AF_INET,
        SOCK_STREAM,
        0,
        NULL,
        0,
        WSA_FLAG_OVERLAPPED);
    if (m_ListenSocket == INVALID_SOCKET) return false;
    return true;
}
int CIOCPServer::Bind(LPCTSTR lpszHostAddress,const USHORT usPort)
{
    SOCKADDR_IN InternetAddr;
    InternetAddr.sin_family = PF_INET;
    InternetAddr.sin_port = htons(usPort);
    InternetAddr.sin_addr.s_addr = inet_addr(lpszHostAddress);
    //帮定到指定端口
    return bind(m_ListenSocket, (SOCKADDR*)&InternetAddr, sizeof(InternetAddr));
}
int CIOCPServer:isten(int nBacklog)
{
    return listen(m_ListenSocket, nBacklog);
}
bool CIOCPServer::CreateWorkThread(void)
{
    m_arrayIOCPThreadHandle = new HANDLE[m_dwIOCPThreadCount];
    for(DWORD i = 0; i < m_dwIOCPThreadCount; i++)
    {
        m_arrayIOCPThreadHandle = (HANDLE)_beginthreadex(NULL, 0, (PBEGINTHREADX_THREADFUN)IOCPWorkThread, (LPVOID)this, 0, NULL);
        if(!m_arrayIOCPThreadHandle)
            return false;
    }
    return true;
}
void CIOCPServer::CloseWorkThread(void)
{
    //向IOCP发送结束线程信号
    for(DWORD i = 0; i < m_dwIOCPThreadCount; i++)
        PostQueuedCompletionStatus(m_hCompletionPort, 0, NULL, NULL);
    //等待工作线程结束，等待时间10秒

    if(WaitForMultipleObjects(m_dwIOCPThreadCount, m_arrayIOCPThreadHandle, TRUE, 10000) != WAIT_OBJECT_0)
    {
        //如果10秒内没有结束所有线程，就强制结束
        for(DWORD i = 0; i < m_dwIOCPThreadCount; i++)
            TerminateThread(m_arrayIOCPThreadHandle, 0);
    }
    //关闭所有工作线程句柄
    for(DWORD i = 0; i < m_dwIOCPThreadCount; i++)
        CloseHandle(m_arrayIOCPThreadHandle);
    //释放线程句柄数组
    delete [] m_arrayIOCPThreadHandle;
    m_arrayIOCPThreadHandle = NULL;

}
DWORD WINAPI CIOCPServer::IOCPWorkThread(LPVOID lpParam)
{
    CIOCPServer* lpThis = (CIOCPServer*)lpParam;
    HANDLE hCompletionPort = lpThis->m_hCompletionPort;
   

    LPVOID lpCompletionKey = NULL;
    LPPER_IO_DATA lpOvplus = NULL;
    DWORD dwByteCount = 0;

    BOOL bGetFunResult = TRUE;

    while(1)
    {
        //线程挂起，等待IOCP消息
        bGetFunResult = GetQueuedCompletionStatus(hCompletionPort, &dwByteCount, (PULONG_PTR)&lpCompletionKey, (LPOVERLAPPED*)&lpOvplus, INFINITE);

        if(!bGetFunResult)
        {
            if(!lpOvplus)
                continue;
        }

        //收到线程结束信号
        if(!lpCompletionKey)
            return 0;

        switch(lpOvplus->dwOpCode)
        {
        case OP_ACCEPT:
            if(!bGetFunResult)
            {
                closesocket(lpOvplus->Socket);                        //关闭socket
                lpThis->m_Net_Data.free_particle(lpOvplus);        //归还结构体到内存池
                break;
            }
            lpThis->OnIOCPAccept(dwByteCount, lpOvplus);
            break;
        case OP_READ:
            lpThis->OnIOCPRecvFromClient(dwByteCount, (LPPER_HANDLE_DATA)lpCompletionKey, lpOvplus);
            break;
        case OP_WRITE:
            lpThis->m_Net_Data.free_particle(lpOvplus);
            break;
        }
    }
    return 0;
}
bool CIOCPServer::CreateAcceptExThread(void)
{
    m_hAcceptExOverEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
    if(!m_hAcceptExOverEvent)
        return false;
    if(WSAEventSelect(m_ListenSocket, m_hAcceptExOverEvent, FD_ACCEPT) == SOCKET_ERROR)
        return false;
    m_bAcceptExCreateThread = true;
    m_hAcceptEx = (HANDLE)_beginthreadex(NULL, 0, (PBEGINTHREADX_THREADFUN)AcceptExThread, (LPVOID)this, 0, NULL);
    if(!m_hAcceptEx)
        return false;
    return true;
}
void CIOCPServer::CloseAcceptExThread(void)
{
    m_bAcceptExCreateThread = false;
    SetEvent(m_hAcceptExOverEvent);
    if(WaitForSingleObject(m_hAcceptEx,10000)!= WAIT_OBJECT_0)
        TerminateThread(m_hAcceptEx, 0);
    //关闭m_hAcceptExOverEvent事件句柄
    CloseHandle(m_hAcceptExOverEvent);
    m_hAcceptExOverEvent = NULL;

}
DWORD WINAPI CIOCPServer::AcceptExThread(LPVOID lpParam)
{
    CIOCPServer* lpThis = (CIOCPServer*)lpParam;
    while(lpThis->m_bAcceptExCreateThread)
    {
        //无限等待，如果有AcceptEx耗尽的信号就创建新的AcceptEx
        if(WaitForSingleObject(lpThis->m_hAcceptExOverEvent, INFINITE) == WAIT_FAILED)
            continue;
        //创建新的AcceptEx消息
        lpThis->CreateAcceptEx();
    }

    return 0;
}
void CIOCPServer::CreateAcceptEx(const DWORD dwPostCount)
{
    LINGER lingerStruct = { 0x01, 0x00 };

    DWORD dwAddrLen = sizeof(sockaddr_in) + IP_STRING_LEN;
    LPPER_IO_DATA lpOvplus = NULL;
    BOOL bNodelay = TRUE;

    for(DWORD i = 0; i < dwPostCount; i++)
    {
        //先从OV结构池取一个可用的结构
        lpOvplus = m_Net_Data.alloc_particle();
        lpOvplus->dwOpCode = OP_ACCEPT;
        lpOvplus->lpSession = NULL;
        lpOvplus->dwBytes = 0;
        ZeroMemory(&lpOvplus->Overlapped,sizeof(OVERLAPPED));

        //创建一个SOCKET
        lpOvplus->Socket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
        if(lpOvplus->Socket == INVALID_SOCKET)
        {
            m_Net_Data.free_particle(lpOvplus);
            continue;
        }
       

        setsockopt(lpOvplus->Socket, IPPROTO_TCP, TCP_NODELAY, (char*)&bNodelay, sizeof(BOOL));
        setsockopt(lpOvplus->Socket, SOL_SOCKET, SO_LINGER, (char*)&lingerStruct, sizeof(LINGER));

        if(!m_lpAcceptExFun(m_ListenSocket, lpOvplus->Socket, lpOvplus->arrayDataBuf, 0, dwAddrLen, dwAddrLen, &lpOvplus->dwBytes, &lpOvplus->Overlapped))
        {
            if(WSAGetLastError() != ERROR_IO_PENDING)
            {
                closesocket(lpOvplus->Socket);
                m_Net_Data.free_particle(lpOvplus);        //归还结构体到内存池
                continue;
            }
        }
    }

}
void CIOCPServer::Close(void)
{
    if (m_ListenSocket != INVALID_SOCKET)
    {
        shutdown(m_ListenSocket, 2);
        closesocket(m_ListenSocket);
    }
}
void CIOCPServer::GetAcceptEx(void)
{
    DWORD dwResult;
    GUID guidAcceptEx = WSAID_ACCEPTEX;
    WSAIoctl(m_ListenSocket, SIO_GET_EXTENSION_FUNCTION_POINTER, &guidAcceptEx, sizeof(GUID), &m_lpAcceptExFun, sizeof(LPFN_ACCEPTEX), &dwResult, NULL, NULL);
}
void CIOCPServer::GetAcceptExSockaddrsFun(void)
{
    DWORD dwResult;
    GUID guidGetAcceptExSockaddrs = WSAID_GETACCEPTEXSOCKADDRS;
    WSAIoctl(m_ListenSocket, SIO_GET_EXTENSION_FUNCTION_POINTER, &guidGetAcceptExSockaddrs, sizeof(GUID), &m_lpGetAcceptExSockaddrsFun, sizeof(LPFN_GETACCEPTEXSOCKADDRS), &dwResult, NULL, NULL);

}
void CIOCPServer::GetAcceptExSockaddrsInfo(LPPER_IO_DATA lpOvplus, DWORD* lpdwIP, USHORT* lpusPort)
{
    int                        nLen = sizeof(sockaddr);
    int                        sLen = sizeof(sockaddr_in) + IP_STRING_LEN;
    sockaddr_in        SockAddr;
    sockaddr*        lpRemoteAddr;
    sockaddr*        lpLocalAddr;
    m_lpGetAcceptExSockaddrsFun(lpOvplus->arrayDataBuf, 0, sLen, sLen, &lpLocalAddr, &nLen, &lpRemoteAddr, &nLen);
    memcpy(&SockAddr, lpRemoteAddr, sizeof(SOCKADDR));
    *lpdwIP = SockAddr.sin_addr.S_un.S_addr;
    *lpusPort = ntohs(SockAddr.sin_port);
}
bool CIOCPServer::BeginServer(void)
{
    if(!CreateSocket())
        return false;
    //绑定
    Bind("127.0.0.1",10000);
    //监听
    Listen(5);
    //创建完成端口句柄

    m_hCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, NULL, m_dwIOCPThreadCount);
    if(!m_hCompletionPort)
        return false;

    //将ListenSocket绑定到IOCP
    if(!CreateIoCompletionPort((HANDLE)m_ListenSocket, m_hCompletionPort,(ULONG_PTR)&m_ListenSocket, 0))
        return false;
    //创建工作线程
    if(!CreateWorkThread())
        return false;
    GetAcceptEx();
    GetAcceptExSockaddrsFun();
    //开始投递AcceptEx
    if(CreateAcceptExThread())
        return false;
    return true;
}
void CIOCPServer::StopServer(void)
{
    Close();
    CloseAllClient();
    CloseAcceptExThread();
    CloseWorkThread();
    //关闭完成端口句柄
    CloseHandle(m_hCompletionPort);
    m_hCompletionPort = NULL;
    //销毁结构池
    m_Net_Data.Destroy();
    m_Client_Context.Destroy();
    m_listSession.Clear();
}
void CIOCPServer::CloseAllClient(void)
{
    m_listSession._guard.lock();
    for(CList<LPPER_HANDLE_DATA>::iterator lpNode = m_listSession.begin(); lpNode != m_listSession.end(); lpNode++)
        closesocket((*lpNode)->Socket);
    m_listSession._guard.unlock();
}
void CIOCPServer::OnIOCPAccept(const DWORD dwByteCount, LPPER_IO_DATA lpOvplus)
{
    DWORD dwIP = 0;
    USHORT usPort = 0;
    GetAcceptExSockaddrsInfo(lpOvplus, &dwIP, &usPort);
    if(!OnIpFilter(dwIP))        //检查该IP是否需要过滤，该函数由派生类实现
    {
        closesocket(lpOvplus->Socket);        //关闭socket
        m_Net_Data.free_particle(lpOvplus);        //归还OV结构体到内存池
        return;
    }
    LPPER_HANDLE_DATA lpSession = m_Client_Context.alloc_particle();
    lpSession->Socket = lpOvplus->Socket;
    lpSession->m_dwLastTime = GetTickCount();
    lpSession->m_dwIP = dwIP;
    lpSession->m_usPort = usPort;
    //添加到Session链表
    m_listSession.Push_Back(lpSession);

    //将完成键与socket绑定到IOCP
    if(CreateIoCompletionPort((HANDLE)lpSession->Socket, m_hCompletionPort, (ULONG_PTR)lpSession, 0))
    {
        ZeroMemory(&lpOvplus->Overlapped, sizeof(OVERLAPPED));
        lpOvplus->dwOpCode = OP_READ;
        lpOvplus->wDataBuf.len = NET_DATA_LONGTH;
        lpOvplus->wDataBuf.buf =lpOvplus->arrayDataBuf;
        lpOvplus->dwBytes = 0;
        lpOvplus->lpSession = lpSession;
        lpOvplus->dwFlags = 0;
        if(CreateRecv(lpOvplus))
        {
            OnAccept();
            return;
        }  
    }
    closesocket(lpOvplus->Socket);                        //关闭socket
    m_listSession.Remove(lpSession);//从Session链表中删除
    m_Client_Context.free_particle(lpSession);                        //将Session归还到池
    m_Net_Data.free_particle(lpOvplus);                        //归还OV结构体到内存池

}
void CIOCPServer::OnIOCPRecvFromClient(const DWORD dwByteCount, LPPER_HANDLE_DATA lpSession, LPPER_IO_DATA lpOvplus)
{
    if(dwByteCount)
    {
        //接收到数据
        lpSession->m_dwLastTime = GetTickCount();
        lpOvplus->dwBytes = dwByteCount;
        //数据处理
        for (int i =0;i<dwByteCount;i++)
        {
            printf("%x",lpOvplus->arrayDataBuf);
        }
        printf("\n");
       
        //
        ZeroMemory(&lpOvplus->Overlapped, sizeof(OVERLAPPED));
        lpOvplus->Socket = lpSession->Socket;
        lpOvplus->dwOpCode = OP_READ;
        lpOvplus->wDataBuf.buf =lpOvplus->arrayDataBuf;
        lpOvplus->lpSession = lpSession;
        lpOvplus->dwFlags = 0;
        if(CreateRecv(lpOvplus))
            return;
    }
    closesocket(lpSession->Socket);                //关闭Socket
    OnDisconnectFromClient(lpSession);                        //通知派生类，一个Session断开了
    m_listSession.Remove(lpSession);//从Session链表中删除
    m_Client_Context.free_particle(lpSession);                        //断开后将Session归还到结构池
    m_Net_Data.free_particle(lpOvplus);        //归还OV结构到结构池

}

bool CIOCPServer::CreateRecv(LPPER_IO_DATA lpOvplus)
{
    if(WSARecv(lpOvplus->Socket, &lpOvplus->wDataBuf, 1, &lpOvplus->dwBytes, &lpOvplus->dwFlags, &lpOvplus->Overlapped, NULL) == SOCKET_ERROR)
    {
        if(WSAGetLastError() != ERROR_IO_PENDING)
            return false;
    }
    return true;
}
bool CIOCPServer::CreateSend(LPPER_IO_DATA lpOvplus)
{
    if(WSASend(lpOvplus->Socket, &lpOvplus->wDataBuf, 1, &lpOvplus->dwBytes, MSG_PARTIAL, &lpOvplus->Overlapped, NULL) == SOCKET_ERROR)
    {
        if(WSAGetLastError() != ERROR_IO_PENDING)
            return false;
    }
    return true;
}

ken1984

有个说明详解就好了。

lustskyboy

注释的很明白啊.

lustskyboy

谁有好的模型出来分享下............