使用基于 IOCP 的客户端的每个 IO 数据

Question

我正在尝试创建一个 IOCP TCP 客户端，我的代码如下所示：

TCPClient.h：

#pragma once

typedef struct 
{ 
    WSAOVERLAPPED Overlapped; 
    SOCKET Socket; 
    WSABUF wsaBuf; 
    char Buffer[1024];
    DWORD Flags;
    DWORD BytesSent;
    DWORD BytesToSend;
} PER_IO_DATA, * LPPER_IO_DATA; 

class TCPClient
{
public:
    TCPClient();
    ~TCPClient();

    bool Connect(const std::string strIpAddress, UINT32 uPort);
    bool Disconnect();
    bool SendCommand(const std::string strCommandName);
    bool ReceiveResponse();

private:
    static DWORD WINAPI ClientWorkerThread(LPVOID lpParameter);

private:
    SOCKET m_socket;
    PER_IO_DATA *m_pPerIoData;
};

TCPClient.cpp：

#include "StdAfx.h"
#include "TCPClient.h"

TCPClient::TCPClient() :
    m_pPerIoData(NULL)
{
}

TCPClient::~TCPClient()
{
}

bool TCPClient::Connect(const std::string strIpAddress, UINT32 uPort)
{
    WSADATA wsaData;
    if (WSAStartup(MAKEWORD(2, 2), &wsaData) != NO_ERROR)
        return false;

    HANDLE hCompletionPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
    if (!hCompletionPort)
        return false;

    SYSTEM_INFO systemInfo;
    GetSystemInfo(&systemInfo);

    for (DWORD dwIndex = 0; dwIndex < systemInfo.dwNumberOfProcessors; dwIndex++)
    {
        HANDLE hThread = CreateThread(NULL, 0, ClientWorkerThread, hCompletionPort, 0, NULL);
        CloseHandle(hThread);
    }

    m_socket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (m_socket == INVALID_SOCKET)
    {
        WSACleanup();

        return false;
    }

    sockaddr_in server;
    server.sin_family = AF_INET;
    server.sin_addr.s_addr = inet_addr(strIpAddress.c_str());
    server.sin_port = htons(uPort);

    CreateIoCompletionPort((HANDLE)m_socket, hCompletionPort, 0, 0);

    if (WSAConnect(m_socket, (LPSOCKADDR)&server, sizeof(server), NULL, NULL, NULL, NULL) == SOCKET_ERROR)
    {
        WSACleanup();

        return false;
    }

    return true;
}

bool TCPClient::Disconnect()
{
    if (m_socket)
        closesocket(m_socket);

    WSACleanup();

    return true;
}

bool TCPClient::SendCommand(const std::string strCommandName)
{
    m_pPerIoData = new PER_IO_DATA;
    ZeroMemory(m_pPerIoData, sizeof(PER_IO_DATA));

    strcpy(m_pPerIoData->Buffer, strCommandName.c_str());

    m_pPerIoData->Overlapped.hEvent = WSACreateEvent();
    m_pPerIoData->Socket = m_socket;
    m_pPerIoData->wsaBuf.buf = m_pPerIoData->Buffer;
    m_pPerIoData->wsaBuf.len = strlen(m_pPerIoData->Buffer);
    m_pPerIoData->BytesToSend = m_pPerIoData->wsaBuf.len;

    DWORD dwNumSent;
    if (WSASend(m_socket, &(m_pPerIoData->wsaBuf), 1, &dwNumSent, 0, &(m_pPerIoData->Overlapped), NULL) == SOCKET_ERROR)
    {
        if (WSAGetLastError() != WSA_IO_PENDING)
        {
            delete m_pPerIoData;
            return 0;
        }
    }

    while (TRUE)
        Sleep(1000);

    return true;
}

bool TCPClient::ReceiveResponse()
{
    return true;
}

DWORD WINAPI TCPClient::ClientWorkerThread(LPVOID lpParameter)
{
    HANDLE hCompletionPort = (HANDLE)lpParameter;
    DWORD NumBytesRecv = 0;
    ULONG CompletionKey;
    LPPER_IO_DATA PerIoData;

    while (GetQueuedCompletionStatus(hCompletionPort, &NumBytesRecv, &CompletionKey, (LPOVERLAPPED*)&PerIoData, INFINITE))
    {
        if (!PerIoData)
            continue;

        if (NumBytesRecv == 0)
        {
            std::cout << "Server disconnected!\r\n\r\n";
        }
        else
        {
            // use PerIoData->Buffer as needed...
            std::cout << std::string(PerIoData->Buffer, NumBytesRecv);

            PerIoData->wsaBuf.len = sizeof(PerIoData->Buffer);
            PerIoData->Flags = 0;

            if (WSARecv(PerIoData->Socket, &(PerIoData->wsaBuf), 1, &NumBytesRecv, &(PerIoData->Flags), &(PerIoData->Overlapped), NULL) == 0)
                continue;

            if (WSAGetLastError() == WSA_IO_PENDING)
                continue;
        }

        closesocket(PerIoData->Socket);
        delete PerIoData;
    }

    return 0;
}

ma​​in.cpp：

#include "stdafx.h"
#include "TCPClient.h"

int main()
{
    TCPClient client;

    client.Connect("127.0.0.1", 8888);

    client.SendCommand("Hello command\r\n");

    return 0;
}

我对“m_pPerIoData”的使用显然是错误的，因为我每次执行 SendCommand() 时都在更新，并且没有正确删除它。

• 第一季度。我应该在哪里做 m_pPerIoData = new PER_IO_DATA？
• 第二季度。将指向 PER_IO_DATA 的指针作为成员变量有意义吗？

编辑 2：

我已经对上面的现有代码进行了一些重命名（客户端 -> 连接），因为这让我感到困惑。

一些背景：

• 我正在创建一个 DLL，用于控制通过 LAN（或串行端口）连接的 ECR（电子收银机）设备。
• DLL 提供易于使用的接口，例如 Connect()、Disconnect() 以及一些 ECR 特定命令，包括 Logon()、Logoff()、ReadCard() 等。
• （可能对我的应用程序来说有点矫枉过正，但是...）我想在我的 DLL 中使用 IOCP 来异步向 ECR 发送/接收数据。

我的顶级课程看起来像这样：

#pragma once

#include "Connection.h"
#include "Uncopyable.h"
#include "ConnectionFactory.h"
#include "CommandName.h"

class Ecr : private Uncopyable
{
public:
    Ecr(const std::string& rstrConnectionInfo)
        : m_spConnection(ConnectionFactory::CreateConnection(rstrConnectionInfo))
    {
        //Initialise();
    }

    ~Ecr()
    {
        //Shutdown();
    }

    bool Initialise()
    {
        if (!m_spConnection)
            return false;

        m_spConnection->Initialise();

        return true;
    }

    bool Shutdown()
    {
        if (!m_spConnection)
            return false;

        m_spConnection->Shutdown();

        return true;
    }

    bool Connect()
    {
        if (!m_spConnection)
            return false;

        if (!m_spConnection->Connect())
            return false;

        return true;
    }

    bool Disconnect()
    {
        m_spConnection->Disconnect();

        return true;
    }

    bool Logon(const std::vector<BYTE>& rvecCommandOptions)
    {
        m_spConnection->SendCommand(CommandName::Logon(), rvecCommandOptions);

        return true;
    }

    bool Logoff()
    {
        m_spConnection->SendCommand(CommandName::Logoff());

        return true;
    }

    // ... more commands follow.

private:
    ConnectionPtr m_spConnection;
};

我有 TcpConnection 类，它是由 ECR 类创建的，可以完成所有的艰苦工作。 Ecr 和 Connection 类都是不可复制的。

Connection.h：

#pragma once

#include "Uncopyable.h"

class CommandName;

class Connection : private Uncopyable
{
public:
    Connection(const std::string& rstrConnectionInfo);
    virtual ~Connection();

    virtual bool Initialise() = 0;
    virtual bool Shutdown() = 0;
    virtual bool Connect() = 0;
    virtual bool Disconnect() = 0;
    virtual bool SendCommand(const CommandName& rCommandName, const std::vector<BYTE>& rvecCommandOptions) = 0;
    virtual bool ReceiveResponse() = 0;

    bool SendCommand(const CommandName& rCommandName);

private:
    std::string m_strConnectionInfo;
};

typedef std::tr1::shared_ptr<Connection> ConnectionPtr;

TcpConnection.h：

#pragma once
#include "connection.h"

class TcpConnection : public Connection
{
public:
    TcpConnection(const std::string& rstrConnectionInfo);
    ~TcpConnection();

    // Connection
    bool Initialise();
    bool Shutdown();
    bool Connect();
    bool Disconnect();
    bool SendCommand(const CommandName& rCommandName, const std::vector<BYTE>& rvecCommandOptions);
    bool ReceiveResponse();

    static DWORD WINAPI WorkerThread(LPVOID lpParam);

private:
    SOCKET m_socket;
    HANDLE m_hIocp;
};

TcpConnection.cpp：

#include "StdAfx.h"
#include "TcpConnection.h"
#include "CommandBuilderTcp.h"

TcpConnection::TcpConnection(const std::string& rstrConnectionInfo)
    : Connection(rstrConnectionInfo)
    , m_socket(INVALID_SOCKET)
    , m_hIocp(INVALID_HANDLE_VALUE)
{
    WSADATA wsaData;
    WSAStartup(MAKEWORD(2, 2), &wsaData);

    m_socket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
}

TcpConnection::~TcpConnection()
{
}

bool TcpConnection::Initialise()
{
    // Set up threads for using IOCP.

    m_hIocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);

    SYSTEM_INFO systemInfo;
    GetSystemInfo(&systemInfo);

    for (DWORD dwIndex = 0; dwIndex < systemInfo.dwNumberOfProcessors; dwIndex++)
    {
        HANDLE hThread = CreateThread(NULL, 0, WorkerThread, m_hIocp, 0, NULL);
        CloseHandle(hThread);
    }

    CreateIoCompletionPort((HANDLE)m_socket, m_hIocp, 0, 0);

    return true;
}

bool TcpConnection::Shutdown()
{
    // Release threads.

    return true;
}

bool TcpConnection::Connect()
{
    if (m_socket)
        return true;

    // Hard-coding IP address and port number for now.
    sockaddr_in server;
    server.sin_family = AF_INET;
    server.sin_addr.s_addr = inet_addr("10.0.9.80");
    server.sin_port = htons(22000);

    WSAConnect(m_socket, (LPSOCKADDR)&server, sizeof(server), NULL, NULL, NULL, NULL);

    return true;
}

bool TcpConnection::Disconnect()
{
    if (m_socket)
        closesocket(m_socket);

    WSACleanup();

    return true;
}

bool TcpConnection::SendCommand(const CommandName& rCommandName, const std::vector<BYTE>& rvecCommandOptions)
{
    // Build full command from rCommandName and rvecCommandOptions and send to server.
    return true;
}

bool TcpConnection::ReceiveResponse()
{
    return true;
}

DWORD WINAPI TcpConnection::WorkerThread(LPVOID lpParam)
{
    // Call GetQueuedCompletionStatus in a loop
    return 0;
}

然后主机应用程序为每个 ECR 设备创建一个 DLL 实例：

Ecr ecr("ipaddress+port");

ecr.Initialise(); // Or do this in Ecr's ctor?

ecr.Connect();

BYTE arrCommandOptions[] = {0x00, 0x00, 0x00, 0x01, 0x18, 0xA6, 0x00, 0x01, 0x49, 0x08, 0x26};
std::vector<BYTE> vecCommandOptions(arrCommandOptions, arrCommandOptions + sizeof(arrCommandOptions) / sizeof(arrCommandOptions[0]));
ecr.Logon(vecCommandOptions);

ecr.Logoff();

ecr.Disconnect();

ecr.Shutdown(); // Or do this in Ecr's dtor?

return 0;

我想要 TcpConnection 级别的所有 IOCP 相关内容，而不是 Ecr 级别，因为我不希望 Ecr 关心数据传输是如何在下面完成的。

我的想法行不通吗？

Answer 1

每个 I/O 数据必须在相关 I/O 操作的生命周期内存在。因此，您可能希望动态分配它，并在完成后将其池化以供重用。我使用引用计数系统，但这只是因为我可以让每个 I/O 数据保持比单个 I/O 操作更长的时间。 I/O 特定数据的生命周期如下：

• 在发出 I/O 操作之前分配（显然）
• 在操作成功或出错完成时释放。

将每个 I/O 作为数据成员实际上没有任何意义，因为它由 I/O 操作“拥有”...您可能有一个 I/O 数据“分配器”可以分配这些对象并在 I/O 操作完成后获得它们的所有权（将它们池化以供重用）。

您可能想查看一些执行此操作的 IOCP 代码（并以可扩展的方式处理事物的线程方面），请参阅 here。

Answer 2

您展示的内容基于我在another question 中给您的 IOCP 代码。但是，该代码旨在跨同一个套接字上的多个 IOCP 操作重用单个 PER_IO_DATA 实例，因为您只在客户端读取而只在服务器中写入。现在您在客户端中混合读取和写入，因此您需要通过合并我之前给您的所有代码来调整工作线程以支持两种类型的 IOCP 操作，然后在PER_IO_DATA 中添加一个附加标志所以完成的操作知道它是读操作还是写操作。

至于你的问题：

我应该在哪里做 m_pPerIoData = new PER_IO_DATA？

使用类范围的m_pPerIoData 成员根本没有多大意义。 SendCommand() 正在创建一个新的 PER_IO_DATA，这很好，当工作线程检测到所有 PER_IO_DATA::Buffer 数据已发送完毕或套接字已关闭时，您只需要 delete 即可。

将指向 PER_IO_DATA 的指针作为成员变量有意义吗？

不用于写操作，不。不过，你可以有一个班级成员来阅读。除非SendCommand() 在准备好读取单个响应时创建一个新的PER_IO_DATA。当收到完整响应或套接字已关闭时，工作线程将需要delete。