D语言的IOCP（完成端口）例子

在WindowsNT平台上，最具有伸缩性和吞吐量的网络服务器程序都使用了完成端口。为了在D中使用完成端口，我写了这个简单的例子。希望大家指正！

在DMD1.020－1.022，WindowsXP，编译测试通过。

没有使用std.socket，因为std里面的socket实现不能使用重叠IO。

D 代码

// D Program Language IOCP
// write by [e-mail]
// complie: dmd ic ws2_32.lib

import std.c.windows.windows, std.c.windows.winsock;
import std.string, std.stdint, std.c.string, std.c.stdlib;
import std.stdio;
import std.thread;




alias HANDLE WSAEVENT;
alias OVERLAPPED WSAOVERLAPPED;
alias OVERLAPPED* LPWSAOVERLAPPED;
alias OVERLAPPED* POVERLAPPED, LPOVERLAPPED;

struct GUID {
align(1):
    DWORD Data1;
    WORD  Data2;
    WORD  Data3;
    BYTE  Data4[8];
}

struct WSAPROTOCOLCHAIN {
    int                       ChainLen;
    DWORD[7] ChainEntries;
}

alias WSAPROTOCOLCHAIN* LPWSAPROTOCOLCHAIN;

const WSAPROTOCOL_LEN = 255;
const ERROR_IO_PENDING = 997;

struct WSAPROTOCOL_INFOW {
    DWORD dwServiceFlags1;
    DWORD dwServiceFlags2;
    DWORD dwServiceFlags3;
    DWORD dwServiceFlags4;
    DWORD dwProviderFlags;
    GUID ProviderId;
    DWORD dwCatalogEntryId;
    WSAPROTOCOLCHAIN ProtocolChain;
    int iVersion;
    int iAddressFamily;
    int iMaxSockAddr;
    int iMinSockAddr;
    int iSocketType;
    int iProtocol;
    int iProtocolMaxOffset;
    int iNetworkByteOrder;
    int iSecurityScheme;
    DWORD dwMessageSize;
    DWORD dwProviderReserved;
    WCHAR[WSAPROTOCOL_LEN+1] szProtocol;
}

alias WSAPROTOCOL_INFOW* LPWSAPROTOCOL_INFOW;
const WSA_FLAG_OVERLAPPED = 0x01;

struct WSABUF {
    uint  len;
    char* buf;
}

struct SOCKADDR {
    ushort  sa_family;
    char[14] sa_data;
}
alias SOCKADDR* PSOCKADDR, LPSOCKADDR;

alias WSABUF* LPWSABUF;
alias uint GROUP;


extern(Windows)
{
    alias void function(DWORD, DWORD, LPWSAOVERLAPPED, DWORD) LPWSAOVERLAPPED_COMPLETION_ROUTINE;
    SOCKET WSASocketW(int, int, int, LPWSAPROTOCOL_INFOW, GROUP, DWORD);
    bool GetQueuedCompletionStatus(HANDLE, PDWORD, PDWORD, LPOVERLAPPED*, DWORD);
    HANDLE CreateIoCompletionPort(HANDLE, HANDLE, DWORD, DWORD);
    int WSASend(SOCKET, LPWSABUF, DWORD, LPDWORD, DWORD, LPWSAOVERLAPPED, LPWSAOVERLAPPED_COMPLETION_ROUTINE);
    int WSARecv(SOCKET, LPWSABUF, DWORD, LPDWORD, LPDWORD, LPWSAOVERLAPPED, LPWSAOVERLAPPED_COMPLETION_ROUTINE);
}

alias WSASocketW WSASocket;

// Maximum Buffer Size
const int BUFFERSIZE =  128;

enum IO_OPERATION
{   ACCEPT,READ,WRITE };

struct PIO_DATA
{
    WSAOVERLAPPED               ol;
    char                        Buffer[BUFFERSIZE];
    WSABUF                      wsabuf;
    int                         nTotalBytes;
    int                         nSentBytes;
    IO_OPERATION                opCode;
    SOCKET                        activeSocket;
}



int max_ThreadCount;
HANDLE iocpHandle = INVALID_HANDLE_VALUE;
SOCKET serverSock;


int WorkerThread (void * context)
{
    LPWSAOVERLAPPED lpol = null;
  PIO_DATA* lpIOContext  = null;
  WSABUF buffSend;
  uint dwRecvNumBytes = 0;
  uint dwSendNumBytes = 0;
  uint dwFlags = 0;
  uint dwIoSize = 0;
  bool bSuccess = false;
  int nRet = 0;

   while( 1 ) {
       void * lpCompletionKey = null;
       bSuccess = GetQueuedCompletionStatus(iocpHandle, &dwIoSize,cast(LPDWORD)&lpCompletionKey,cast(LPOVERLAPPED *)&lpol, INFINITE);
       if( !bSuccess )
       {
          writefln("GetQueuedCompletionStatus() failed: %s.",GetLastError());
          break;
       }
       lpIOContext = cast(PIO_DATA * )lpol;
       if(dwIoSize == 0) //socket closed?
       {
           writefln("ClientSocket Disconnect!" );
           closesocket(lpIOContext.activeSocket);
          // delete lpIOContext;
           continue;
       }

       if(lpIOContext.opCode == IO_OPERATION.READ) // a read operation complete
       {
                    char[] s = "Echo:" ~  std.string.toString(lpIOContext.wsabuf.buf);
                    lpIOContext.wsabuf.buf = std.string.toStringz(s);
              lpIOContext.nTotalBytes  = lpIOContext.wsabuf.len;
              lpIOContext.nSentBytes   = 0;
              lpIOContext.opCode = IO_OPERATION.WRITE;
              dwFlags = 0;
              nRet = WSASend(
                            lpIOContext.activeSocket,
                            &lpIOContext.wsabuf, 1, &dwSendNumBytes,
                            dwFlags,
                            &(lpIOContext.ol) , null);
              if( nRet == SOCKET_ERROR && (ERROR_IO_PENDING != WSAGetLastError()) )
              {
                      writefln("1.WASSend Failed,Ret:%s." ,WSAGetLastError() );
                      closesocket(lpIOContext.activeSocket);
                      continue;
              }
       }
       else if(lpIOContext.opCode == IO_OPERATION.WRITE) //a write operation complete
       {
              lpIOContext.nSentBytes  += dwIoSize;
              dwFlags = 0;
              if( lpIOContext.nSentBytes < lpIOContext.nTotalBytes ) {
                  lpIOContext.opCode = IO_OPERATION.WRITE;
                  // A Write operation has not completed yet, so post another
                  // Write operation to post remaining data.
                    buffSend.buf = lpIOContext.Buffer.ptr + lpIOContext.nSentBytes; //offset.
                  buffSend.len = lpIOContext.nTotalBytes - lpIOContext.nSentBytes;
                  nRet = WSASend (lpIOContext.activeSocket,
                                 &buffSend, 1, &dwSendNumBytes,
                                 dwFlags,
                                 &(lpIOContext.ol), null);

                  if( nRet == SOCKET_ERROR && (ERROR_IO_PENDING != WSAGetLastError()) ) {
                      writefln("2.WASSend Failed,Ret:%s.",WSAGetLastError());
                      closesocket(lpIOContext.activeSocket);
                      continue;
                  }
              }
              else
              {
                  // Write operation completed, so post Read operation.
                  lpIOContext.opCode = IO_OPERATION.READ;
                  dwRecvNumBytes = 0;
                  dwFlags = 0;
                  lpIOContext.wsabuf.buf = lpIOContext.Buffer.ptr,
                  lpIOContext.ol.Internal = 0;
                  lpIOContext.ol.InternalHigh = 0;
                  lpIOContext.ol.Offset = 0;
                  lpIOContext.ol.OffsetHigh = 0;
                  lpIOContext.ol.hEvent = null;
                  lpIOContext.wsabuf.len = BUFFERSIZE;
                  nRet = WSARecv(
                                lpIOContext.activeSocket,
                                &lpIOContext.wsabuf, 1, &dwRecvNumBytes,
                                &dwFlags,
                                &lpIOContext.ol, null);
                  if( nRet == SOCKET_ERROR && (ERROR_IO_PENDING != WSAGetLastError()) ) {
                      writefln("1.WASRecv Failed,Ret:%s.",WSAGetLastError()  );
                      closesocket(lpIOContext.activeSocket);
                      continue;
                  }
              }
       }
  }

    return 0;
}

int main ()
{

   { // Init winsock2.2
        WSADATA wsaData;
        int retVal = -1;
        if( (retVal = WSAStartup(0x2020, &wsaData)) != 0 ) {
            writefln("WSAStartup Failed,Ret: %s" ,retVal);
            return 1;
        }
    }
    writefln("WSAStartup Init OK!" );

    {  //Create socket
        serverSock = WSASocket(AF_INET,SOCK_STREAM, IPPROTO_TCP, null,0,WSA_FLAG_OVERLAPPED);

        if( serverSock == INVALID_SOCKET ) {
            writefln("Server Socket Creation Failed,Ret:%s." , WSAGetLastError() );
            return 1;
        }
    }

    writefln("Create socket OK!" );

    {   //bind
        sockaddr_in service;
        service.sin_family = AF_INET;
        service.sin_addr.s_addr = htonl(INADDR_ANY);
        service.sin_port = htons(9001);
        int retVal = bind(serverSock,cast(sockaddr *)&service,service.sizeof);
        if( retVal == SOCKET_ERROR ) {
            writefln("Server Soket Bind Failed,Ret:%s." , WSAGetLastError() );;
            return 1;
        }
    }

    writefln("Binding ServerSocket OK!" );

    {   //listen
        int retVal = listen(serverSock, 8);
        if( retVal == SOCKET_ERROR ) {
            writefln("Server Socket Listen Failed,Ret:%s." , WSAGetLastError() );;
            return 1;
        }
    }

        writefln("ServerSocket listen OK!" );

        //create iocp & binding serverSocket to iocp
    {   // Create IOCP

        max_ThreadCount = 1 * 2;
        iocpHandle = CreateIoCompletionPort(INVALID_HANDLE_VALUE,null,0,max_ThreadCount);
        if (iocpHandle == null) {
            writefln("CreateIoCompletionPort() Failed,Ret:%s." , GetLastError() );
            return 1;
        }

        if (CreateIoCompletionPort(cast(HANDLE)serverSock,iocpHandle,0,0) == null){
            writefln("Binding Server Socket to IO Completion Port Failed,Ret:%s." , GetLastError() );
            return 1;
        }
    }

        writefln("Create IOCP & binding ServerSocket to IOCP OK!" );

    {
        Thread worker = new Thread(&WorkerThread, cast(void *)0);
        worker.start();
    }

    writefln("Create Worker threads OK, Waitting Client Connect..." );

    { //accept new connection
        while(1)
        {
            SOCKET clientsock = accept( serverSock, null, null );
            if(clientsock == SOCKET_ERROR)  break;
            writefln("Client connected." );

            { //diable buffer to improve performance
                int nZero = 0;
                setsockopt(clientsock, SOL_SOCKET, SO_SNDBUF, cast(char *)&nZero, nZero.sizeof);
            }

                        //binding ClientSocket to IOCP
            if (CreateIoCompletionPort(cast(HANDLE)clientsock,iocpHandle,0,0) == null){
                writefln("Binding Client Socket to IO Completion Port Failed,Ret:%s." , GetLastError() );
                closesocket(clientsock);
            }
            else {
            writefln("binding ClientSocket to IOCP OK!" );
            //post a recv request
                PIO_DATA data;
                data.opCode = IO_OPERATION.READ;
                data.nTotalBytes = 0;
                data.nSentBytes  = 0;
                data.wsabuf.buf  = data.Buffer.ptr;
                data.wsabuf.len  = data.Buffer.sizeof;
                data.activeSocket = clientsock;
                uint dwRecvNumBytes=0,dwFlags=0;
                int nRet = WSARecv(clientsock,&data.wsabuf, 1, &dwRecvNumBytes,&dwFlags,&data.ol, null);

                if(nRet == SOCKET_ERROR  && (ERROR_IO_PENDING != WSAGetLastError())){
                    writefln("3.WASRecv Failed,Ret:%s." , WSAGetLastError() );;
                    closesocket(clientsock);
                    //delete data;
                }

               writefln("Post a recv request IOCP OK!" );
            }
        }
    }
        //close & Cleanup
    closesocket(serverSock);
    WSACleanup();
        return 0;
}

分享到：

火车站滞留旅客带来的思考 | 那个美国老太忽悠了十三亿中国人(转帖)

• 2007-10-20 20:43
• 浏览 6189
• 评论(9)
• 查看更多

评论

9 楼 ahadf 2007-10-22  

"iocp的使用例子(哪怕是C方式的)在dsource上都找不到一个，仅此一条ideage的做法就值得称道."
---不想抬杠，不过偶手边的资料有5，6本书上都有类似的。网上一搜IOCP更是一大把。也许写个boost.asio D 语言简化版本还比较有价值。

8 楼 h_rain 2007-10-22  

不错了.都这个样子了,进行封装也很简单了.
如果需要一个好的封装,需要考虑一些方面:IOCP句柄与状态查询的封装;线程的完善处理("逻辑"暂停,启动...);Socket句柄的重叠IO操作封装等;
如果想要一个完整的封装,这些东西的处理其实还是很复杂的.

7 楼 tomqyp 2007-10-22  

iocp的使用例子(哪怕是C方式的)在dsource上都找不到一个，仅此一条ideage的做法就值得称道。

6 楼 ahadf 2007-10-22  

偶还是要说，又没用到D的特性，又没用到D的库，用C不挺好，为啥还要用D写呢？演示？这种程序书上网上一抓一大把，学C的比学D的人多多了，直接copy下来看的人指不定还多些。

5 楼 ideage 2007-10-22  

这个例子关键是演示了IOCP在D语言中的应用：过程，方法。

已经写的IOCP的类，用在公司的程序，不能公开

4 楼 dayn9 2007-10-22  

这只能说是批着D马甲的C程序吧？

oldrev老大，这点我持有异议。D也应该是多范式的语言，不能只有OO,OB,GP，否则会损失潜在用户,SP范式恰恰是用户最大的群体。

3 楼 ahadf 2007-10-21  

我还以为是一个IOCP的类。。。

2 楼 oldrev 2007-10-21  

这只能说是批着D马甲的C程序吧？

1 楼 tomqyp 2007-10-20  

收藏~