c - 没有完成键的 IOCP 通知

Question

我正在构建 IOCP/RIO Winsock 服务器，并且当客户端触发我的 AcceptEx() 调用时，一直在努力从完成端口获取正确的通知要求。

当我在发送客户端请求后调用 GetQueuedCompletionStatus() 时，我将获得成功返回，但没有包含任何数据的完成 key 。

我已经使用 PostQueuedCompletionStatus() 来获取完成键，但是当我的 AcceptEx() 调用 post 时通知，我没有收到完成 key ，只是收到通知。

我确信我没有正确设置或调用某些东西，我只是找不到它在哪里。问题here出奇的相似，但最终答案并没有解决我的问题。

为了将问题提炼为最简单的形式，我启动了一个新项目，并复制了必要的部分来复制问题。所有变量都是全局变量，因此没有任何内容超出范围，为了发布代码，我删除了错误检查和任何循环。我们只需创建一个套接字并给客户端几秒钟的时间来请求某些内容，然后我们检查是否发送了完成 key 。这就是这个错误证明项目。

全局变量的注释按照它们在代码中出现的顺序排列，并复制代码注释以供引用。

//server main variables
const int maxSockets = 10;
const int bufferSlicesPerSocket = 3;
const int bufferSliceSize = 400;
const int numRecvToPostPerSocket = 2;
atomic<bool> runApp(true);
bool bResult;
int iResult;

//initWinsock
WSADATA wsaData;
struct addrinfo *ptr = NULL;
struct addrinfo hints;

//resolveAddress
struct addrinfo *result = NULL;

//createIOCP1
HANDLE hCompPort1;

//createListenSocket
SOCKET listenSocket = INVALID_SOCKET;

//associateListenSocketWithIOCP

//bindListenSocket

//startListenOnSocket

//getAcceptExPointer
LPFN_ACCEPTEX pAcceptEx = NULL;
GUID guidForAcceptEx = WSAID_ACCEPTEX;
DWORD dwBytes = 0;

//getRIO
RIO_EXTENSION_FUNCTION_TABLE rio;
GUID functionTableId = WSAID_MULTIPLE_RIO;

//registerBigBuffer
RIO_BUFFERID bigBufferId;
const int totalBufferSlices = maxSockets * bufferSlicesPerSocket;
const int bigBufferSize = totalBufferSlices * bufferSliceSize;
char bigBuffer[bigBufferSize];
char * pBigBuffer = bigBuffer;

//fillBufferSliceQueue
RIO_BUF bufferSliceArray[totalBufferSlices];
concurrent_queue<int> availableBufferSliceIdQueue;

//createCompletionQueue
RIO_CQ recvCQ;
RIO_CQ sendCQ;
const int RQMaxRecvs = 5;
const int RQMaxSends = 5;
DWORD recvCQsize = maxSockets * RQMaxRecvs;
DWORD sendCQsize = maxSockets * RQMaxSends;
RIO_NOTIFICATION_COMPLETION notify;
WSAOVERLAPPED ol;

//fill empty socket queue with all Ids
concurrent_queue<int> availableEmptySocketIdQueue;

//set per-handle data
ULONG_PTR cKeyArray[maxSockets];

//check for an empty socketId and create a socket
int socketId;

//create a socket
SOCKET socketArray[maxSockets] = { INVALID_SOCKET };

//then post an accept on the socket

//associate the socket with the completion port
HANDLE hCompPort2;

//empty overlapped structure
WSAOVERLAPPED olArray[maxSockets];

//post an acceptEx on socket
const int acceptExAddrBufferLength = (sizeof(sockaddr) + 16) * 2 * maxSockets;
char acceptExAddrBuffer[acceptExAddrBufferLength];

//then create a request queue on the socket
RIO_RQ requestQueueArray[maxSockets];

//then post a receive on the socket

//get next available buffer slice id
int bufferSliceId;
int postRecv_counter;

//post a recv

//zero memory for overlapped struct
LPOVERLAPPED pol;

//check for completion
ULONG_PTR pKey;

//record socket as being in use
bool socketInUse[maxSockets];
atomic<int> numSocketsInUse = 0;

//update new socket with the listening socket's options
int numResults;
const int maxDequeue = 10;
RIORESULT rioResultArray[maxDequeue];

//calculate contexts
int socketContext;
int requestContext;

这是main()和初始化:

int _tmain(int argc, _TCHAR* argv[]) {
    //initWinsock
    iResult = WSAStartup(MAKEWORD(2, 2), &wsaData);

    ZeroMemory(&hints, sizeof(hints));

    hints.ai_family = AF_INET;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_protocol = IPPROTO_TCP;
    hints.ai_flags = AI_PASSIVE;

    //resolve local address and port to be used by the server
    iResult = getaddrinfo(NULL, DEFAULT_PORT, &hints, &result);

    //create a handle for the first completion port
    hCompPort1 = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, (u_long)0, 0);

    //create a socket for the server to listen to for client connections
    listenSocket = ::WSASocket(
        AF_INET,
        SOCK_STREAM,
        IPPROTO_TCP,
        NULL,
        0,
        WSA_FLAG_REGISTERED_IO | WSA_FLAG_OVERLAPPED);

    //associate the listening socket with the first completion port
    CreateIoCompletionPort((HANDLE)listenSocket, hCompPort1, (u_long)0, 0);

    //bind the listening socket
    iResult = ::bind(listenSocket, result->ai_addr, (int)result->ai_addrlen);

    //start listen on socket
    iResult = listen(listenSocket, SOMAXCONN);

    //get the AcceptEx pointer
    iResult = WSAIoctl(
        listenSocket,
        SIO_GET_EXTENSION_FUNCTION_POINTER,
        &guidForAcceptEx,
        sizeof(guidForAcceptEx),
        &pAcceptEx,
        sizeof(pAcceptEx),
        &dwBytes,
        NULL,
        NULL);

    //get RIO
    if (0 != WSAIoctl(
        listenSocket,
        SIO_GET_MULTIPLE_EXTENSION_FUNCTION_POINTER,
        &functionTableId,
        sizeof(GUID),
        (void**)&rio,
        sizeof(rio),
        &dwBytes,
        0,
        0
        )) {}

    //register big buffer
    bigBufferId = rio.RIORegisterBuffer(pBigBuffer, bigBufferSize);

    //fill bufferSlice queue
    for (int i = 0; i < totalBufferSlices; i++) {
        bufferSliceArray[i].BufferId = bigBufferId;
        bufferSliceArray[i].Offset = i * bufferSliceSize;
        bufferSliceArray[i].Length = bufferSliceSize;
        availableBufferSliceIdQueue.push(i);
    }

    //createCompletionQueue
    notify.Type = RIO_IOCP_COMPLETION;
    notify.Iocp.IocpHandle = hCompPort1;
    notify.Iocp.Overlapped = &ol;
    notify.Iocp.CompletionKey = NULL;

    //create completion queue
    recvCQ = rio.RIOCreateCompletionQueue(
        recvCQsize, //size of queue
        &notify);   //notification mechanism
    sendCQ = rio.RIOCreateCompletionQueue(
        sendCQsize, //size of queue
        &notify);   //notification mechanism

    socketId = 1; //simplified for demo

    //create a socket
    socketArray[socketId] = ::WSASocket(
        AF_INET,                                        //af
        SOCK_STREAM,                                    //type
        IPPROTO_TCP,                                    //protocol
        NULL,                                           //lpProtocolInfo
        0,                                              //group
        WSA_FLAG_REGISTERED_IO | WSA_FLAG_OVERLAPPED);  //dwFlags

    //set completion key
    cKeyArray[socketId] = socketId;

    //then post an accept on the socket

    //empty overlapped structure
    memset(&olArray[socketId], 0, sizeof(WSAOVERLAPPED));

我们现在将套接字与完成端口关联起来:

//associate the socket with the completion port
hCompPort2 = CreateIoCompletionPort(
    (HANDLE)socketArray[socketId],  //file handle
    hCompPort1,                     //completion port
    cKeyArray[socketId],            //completion key
    NULL);                          //number of threads

//post an acceptEx on socket
bResult = pAcceptEx(
    listenSocket,                                                   //listen socket
    socketArray[socketId],                                          //accept socket
    &acceptExAddrBuffer[socketId * (sizeof(sockaddr) + 16) * 2],    //output buffer
    0,                                                              //data length
    sizeof(sockaddr) + 16,                                          //local address length
    sizeof(sockaddr) + 16,                                          //remote address length
    &dwBytes,                                                       //bytes received
    &olArray[socketId]);                                            //overlapped struct

//then create a request queue on the socket
requestQueueArray[socketId] = rio.RIOCreateRequestQueue(
    socketArray[socketId],  //socket
    RQMaxRecvs,             //max RECVs on queue
    1,                      //max recv buffers, set to 1
    RQMaxSends,             //max outstanding sends
    1,                      //max send buffers, set to 1
    recvCQ,                 //recv completion queue
    sendCQ,                 //send completion queue
    &socketArray[socketId]);//socket context

//then post a receive on the socket

//get next available buffer slice id
bufferSliceId = 1; //simplified for demo

//post a recv
bResult = rio.RIOReceive(
    requestQueueArray[socketId],        //socket request queue
    &bufferSliceArray[bufferSliceId],   //buffer slice
    1,                                  //set to 1
    0,                                  //flags
    &bufferSliceArray[bufferSliceId]);  //request context

wprintf(L"Please send request now...\n");

std::this_thread::sleep_for(std::chrono::milliseconds(3500));

最后，使通知出队，由于某种原因，通知不包含完成键。

//send a dummy message to check the IOCP
//bResult = PostQueuedCompletionStatus(hCompPort2,(DWORD)0,(ULONG_PTR)socketId,NULL);   

//zero memory for overlapped struct
ZeroMemory(&pol, sizeof(LPOVERLAPPED));

//check for completion
iResult = GetQueuedCompletionStatus(
    hCompPort2, //completion port
    &dwBytes,   //number of bytes transferred
    &pKey,      //completion key pointer
    &pol,       //overlapped pointer
    0);         //milliseconds to wait

//update new socket with the listening socket's options
iResult = setsockopt(
    socketArray[pKey],          //socket
    SOL_SOCKET,                 //level
    SO_UPDATE_ACCEPT_CONTEXT,   //optname
    (char*)&listenSocket,       //*optval
    sizeof(listenSocket));      //optlen

wprintf(L"accepted socketId: %d\n", pKey);

return 0;
}

完成 key 没有数据，但是当我手动发送 PostQueuedCompletionStatus() 时，我将通过通知获取 key 。我在 AcceptEx() 或 CreateIoCompletionPort() 中未能执行什么操作？？？

Answer 1

您正在将 0 指定为监听套接字和 RIO 完成队列的完成键。您正在为接受客户端时 AcceptEx() 填充的客户端套接字分配其他完成键。 GetQueuedCompletionStatus() 报告正在执行排队 IOCP 操作的套接字的完成键。

当AcceptEx()完成时，GetQueuedCompletionStatus()报告监听套接字的完成键(为0)。

当RIOReceive()完成时，GetQueuedCompletionStatus()报告从中读取的客户端套接字/队列的完成键。

因此，您需要为每种类型的套接字(监听与客户端)分配唯一的完成键，并仅在收到监听套接字的完成键时调用 setsockopt(SO_UPDATE_ACCEPT_CONTEXT)。如果您不以这种方式使用完成键，则必须使用 OVERLAPPED 结构来传递您自己的自定义每个套接字上下文信息，因此您仍然可以区分 AcceptEx( ) 来自其他操作的操作。