Как передать пользовательские данные в рабочий поток с помощью IOCP?

Question

Эй ... Я создал небольшой тестовый сервер, используя порты завершения ввода-вывода и winsock.Я могу успешно подключиться и связать дескриптор сокета с портом завершения.Но я не знаю, как передать пользовательские структуры данных в поток работы ...

До сих пор я пытался передать пользовательскую структуру как (ULONG_PTR)&structure as ключ завершения вВызов ассоциации CreateIoCompletionPort() Но это не сработало.

Теперь я попытался определить собственную OVERLAPPED-структуру и использовать CONTAINING_RECORD (), как описано здесь http://msdn.microsoft.com/en-us/magazine/cc302334.aspx и http://msdn.microsoft.com/en-us/magazine/bb985148.aspx. Ноэто тоже не работает.(Я получаю странные значения для содержимого pHelper)

Итак, мой вопрос: как я могу передать данные в рабочий поток с помощью WSARecv (), GetQueuedCompletionStatus () и пакета Completion или OVERLAPPED-strucutre?

РЕДАКТИРОВАТЬ: Как я могу успешно передавать «данные для каждого соединения»? ... Кажется, я неправильно понял (как описано в двух ссылках выше).

Вот мой код: (Да, уродливый и единственный TEST-код)

struct helper
    {
        SOCKET m_sock;
        unsigned int m_key;
        OVERLAPPED over;
    };


///////

SOCKET newSock = INVALID_SOCKET;
    WSABUF wsabuffer;
    char cbuf[250];
    wsabuffer.buf = cbuf;
    wsabuffer.len = 250;
    DWORD flags, bytesrecvd;


    while(true)
    {
        newSock = accept(AcceptorSock, NULL, NULL);
        if(newSock == INVALID_SOCKET)
            ErrorAbort("could not accept a connection");

        //associate socket with the CP
        if(CreateIoCompletionPort((HANDLE)newSock, hCompletionPort, 3,0) != hCompletionPort)
            ErrorAbort("Wrong port associated with the connection");
        else
            cout << "New Connection made and associated\n";

        helper* pHelper = new helper;
        pHelper->m_key = 3;
        pHelper->m_sock = newSock;
        memset(&(pHelper->over), 0, sizeof(OVERLAPPED));
        flags = 0;
        bytesrecvd = 0;

        if(WSARecv(newSock, &wsabuffer, 1, NULL, &flags, (OVERLAPPED*)pHelper, NULL) != 0)
        {
            if(WSAGetLastError() != WSA_IO_PENDING)
                ErrorAbort("WSARecv didnt work");
        }
    }

    //Cleanup
    CloseHandle(hCompletionPort);
    cin.get();
    return 0;
}

DWORD WINAPI ThreadProc(HANDLE h)
{
    DWORD dwNumberOfBytes = 0;
    OVERLAPPED* pOver = nullptr;
    helper* pHelper = nullptr;
    WSABUF RecvBuf;
    char cBuffer[250];
    RecvBuf.buf = cBuffer;
    RecvBuf.len = 250;
    DWORD dwRecvBytes = 0;
    DWORD dwFlags = 0;
    ULONG_PTR Key = 0;

    GetQueuedCompletionStatus(h, &dwNumberOfBytes, &Key, &pOver, INFINITE);

    //Extract helper
    pHelper = (helper*)CONTAINING_RECORD(pOver, helper, over);


    cout << "Received Overlapped item" << endl;
    if(WSARecv(pHelper->m_sock, &RecvBuf, 1, &dwRecvBytes, &dwFlags, pOver, NULL) != 0)
        cout << "Could not receive data\n";
    else
        cout << "Data Received: " << RecvBuf.buf << endl;

    ExitThread(0);
}

Answer 1

Если вы передадите свою структуру следующим образом, она будет работать нормально:

helper* pHelper = new helper;
CreateIoCompletionPort((HANDLE)newSock, hCompletionPort, (ULONG_PTR)pHelper,0);
...


helper* pHelper=NULL;
GetQueuedCompletionStatus(h, &dwNumberOfBytes, (PULONG_PTR)&pHelper, &pOver, INFINITE);

---

Изменить для добавления данных ввода-вывода:

Одна из часто используемых функций асинхронного асинхронногоapis - они не копируют структуру OVERLAPPED, они просто используют предоставленную структуру - следовательно, перекрывающаяся структура, возвращаемая из GetQueuedCompletionStatus, указывает на изначально предоставленную структуру.Итак:

struct helper {
  OVERLAPPED m_over;
  SOCKET     m_socket;
  UINT       m_key;
};

if(WSARecv(newSock, &wsabuffer, 1, NULL, &flags, &pHelper->m_over, NULL) != 0)

Обратите внимание, что, опять же, в исходном образце вы ошиблись в кастинге.(OVERLAPPED *) pHelper передавал указатель на START вспомогательной структуры, но часть OVERLAPPED была объявлена ​​последней.Я изменил его, чтобы передать адрес фактической перекрывающейся части, что означает, что код компилируется без преобразования, что позволяет нам знать , что мы делаем правильные вещи.Я также переместил перекрывающуюся структуру в качестве первого члена структуры.

Чтобы перехватить данные на другой стороне:

OVERLAPPED* pOver;
ULONG_PTR key;
if(GetQueuedCompletionStatus(h,&dw,&key,&pOver,INFINITE))
{
  // c cast
  helper* pConnData = (helper*)pOver;

На этой стороне особенно важно, чтобы перекрывающаяся структураявляется первым членом вспомогательной структуры, так как это облегчает приведение обратно из OVERLAPPED *, который нам дает API, и помощника *, которого мы на самом деле хотим.

Answer 2

Вы можете отправить собственные данные специального назначения в порт завершения через PostQueuedCompletionStatus .

Пакет завершения ввода / вывода удовлетворит выдающийся призыв к Функция GetQueuedCompletionStatus. Эта функция возвращается с тремя значения передаются как второе, третье, и четвертый параметр вызова PostQueuedCompletionStatus. Система не использует и не проверяет эти значения. В частности, lpOverlapped параметр не должен указывать на Перекрытая структура.

Answer 3

Я использую стандартные процедуры сокетов (socket, closesocket, bind, accept, connect ...) для создания / уничтожения и ReadFile / WriteFile для ввода-вывода, поскольку они позволяют использовать структуру OVERLAPPED.

После того, как ваш сокет принял или подключился, вы должны связать его с контекстом сеанса, который он обслуживает. Затем вы связываете свой сокет с IOCP и (в третьем параметре) предоставляете ему ссылку на контекст сеанса. IOCP не знает, что это за ссылка, и не заботится ни об этом. Ссылка предназначена для ВАШЕГО использования, поэтому, когда вы получаете IOC через GetQueuedCompletionStatus, переменная, на которую указывает параметр 3, будет заполнена ссылкой, чтобы вы сразу нашли контекст, связанный с событием сокета, и могли начать обслуживание события. Я обычно использую индексированную структуру, содержащую (среди прочего) объявление сокета, перекрывающуюся структуру, а также другие специфичные для сеанса данные. Ссылка, которую я передаю CreateIoCompletionPort в параметре 3, будет индексом элемента структуры, содержащего сокет.

Вам необходимо проверить, вернул ли GetQueuedCompletionStatus завершение или время ожидания. С тайм-аутом вы можете пройти через свою проиндексированную структуру и посмотреть (например), истек ли один из них или что-то еще, и предпринять соответствующие действия по ведению домашнего хозяйства.

Необходимо также проверить перекрывающуюся структуру, чтобы убедиться, что ввод / вывод выполнен правильно.

Функция, обслуживающая IOCP, должна быть отдельной многопоточной сущностью. Используйте столько же потоков, сколько у вас ядер в вашей системе, или, по крайней мере, не больше, поскольку это приводит к потере системных ресурсов (у вас нет больше ресурсов для обслуживания события, чем количество ядер в вашей системе, верно?) .

IOCP действительно лучшие из всех миров (слишком хороши, чтобы быть правдой), и любой, кто говорит «один поток на сокет» или «ждать в списке с несколькими сокетами в одной функции», не знает, о чем идет речь. Первый подчеркивает ваш планировщик, а второй - опрос, а опрос ВСЕГДА крайне бесполезен.