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

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

До сих пор я пытался передать пользовательскую структуру как (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() и пакет завершения или структуру OVERLAPPED?

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

Вот мой код: (Да, он уродлив и это единственный 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);
}

c++ user-defined-types worker-thread iocp
person Juarrow    schedule 03.12.2010    source источник

Ответы (3)


arrow_upward
2
arrow_downward

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

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

person Steve Townsend    schedule 03.12.2010
comment
Эй, спасибо за помощь. Не могли бы Вы помочь мне и со вторым вопросом? Как я могу правильно отправить эти данные для каждого соединения через структуру OVERLAPPED?... Как упоминалось в моем OP, я не заставил его правильно работать с макросом CONTAINING_RECORD() - person Juarrow; 03.12.2010
comment
Почему бы вам не переместить член OVERLAPPED over; в верхнюю часть struct, а затем просто привести указатель к helper*? - person Steve Townsend; 03.12.2010
comment
Даже если это работает, мне это кажется неправильным — зависимость от порядка структур внутри структур — … - person Juarrow; 03.12.2010
comment
Это было предположение, потому что пример кода, который вы упомянули в MSDN, имеет именно это. Так это сработало? - person Steve Townsend; 03.12.2010
comment
Да, это, наконец, работает... Но - извините - но это ОТСТОЙ, что методы, упомянутые в MSDN (независимо от того, сколько им лет - макросы все еще существуют, так что я думаю, что остальные тоже должны работать) не работают вообще ^^ - person Juarrow; 04.12.2010
comment
Соглашусь, что статья немного сумбурна. Однако это сложная область, и я уверен, что она помогла вам начать работу. - person Steve Townsend; 04.12.2010
comment
Вы можете отправлять собственные данные специального назначения в порт завершения через PortQueuedCompletionStatus. Это запутанное описание. В нем должно быть написано, что вы можете отправить IOC, не связанный с событием, в IOCP через PostQueuedCompletionStatus. По сути, вы можете посылать фиктивные IOC, чтобы ваша программа обслуживания создавала впечатление, что IOC произошел. То, что вы хотите, чтобы МОК делал, зависит от вас и вашего контекста. - person Olof Forshell; 14.12.2010

arrow_upward
4
arrow_downward

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

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


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

Изменить, чтобы добавить данные для каждого ввода-вывода:

Одной из часто злоупотребляемых функций асинхронного API является то, что они не копируют структуру 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 передавал указатель на начало вспомогательной структуры, но часть OVERLAPPED была объявлена ​​последней. Я изменил его, чтобы передать адрес фактической перекрывающейся части, что означает, что код компилируется без приведения, что позволяет нам знать, что мы делаем правильно. Я также переместил перекрывающуюся структуру, чтобы она стала первым членом структуры.

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

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

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

person Chris Becke    schedule 03.12.2010
comment
Вау... Я уже пробовал это, и это не сработало О_о... Кажется, я напортачил с разыменованием... - person Juarrow; 03.12.2010
comment
Трагически легко ошибиться в таком разыменовании. Компилятор, заставляющий нас все время приводить, не помогает. - person Chris Becke; 03.12.2010
comment
хе-хе... и я всегда доказываю это ^^... Знаете ли вы, как получить данные для каждого вызова ввода-вывода, отправленные со структурой OVERLAPPED в поток, который обрабатывает, например, WSARecv? - person Juarrow; 03.12.2010

arrow_upward
0
arrow_downward

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

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

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

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

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

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

person Olof Forshell    schedule 14.12.2010