delphi - TIdTCPServer 如何在 60Hz 计时器中向所有客户端多播？

Question

我是 Indy 的新手。这是我第一次在这里发布问题。

我的项目必须以 60Hz 的频率向所有客户端发送数据。为此，我正在使用 TIdTCPServer 来监控保持事件状态。我的工具很旧，在 WinXP 上运行，使用 C++Builder 6 和 Indy 8。有一个潜在的超时问题，有没有人想过如何处理它？

这是我的代码:

服务器端

typedef struct
{
    AnsiString PeerIP;     //{ Cleint IP address }
    AnsiString HostName;   //{ Hostname }
    int Id;        // {Cleint ID}
} TClient;


// This is Multimedia timer callback function, on 60Hz

void CALLBACK mmTimerProc(UINT wTimerID, UINT msg, DWORD dwUser, DWORD dw1, DWORD dw2)
{
    DWORD T1, T2;
    TfmMain *pMain = (TfmMain *)dwUser;
    int Index;
    double dT;
    TClient *pClient;

    if (pMain->IsClosing) return;


    if (pMain->Shutdown)
    {
        return;
    }


    pMain->UpdateServer1Data();

    TList *pList = pMain->IdTCPServer1->Threads->LockList();
    try
    {
        for(int X = 0; X < pList->Count; X++)
        {
            if (!pMain->IsClosing)
            {
                TIdPeerThread *AThread = (TIdPeerThread *)pList->Items[X];

                if(AThread != NULL)
                {
                    pClient = (TClient *)AThread->Data;
                    try
                    {
                        if(!AThread->Connection->ClosedGracefully)
                        {
                            // Send data to ALL Clients
                            AThread->Connection->WriteBuffer(&pMain->Data2Client, sizeof(pMain->Data2Client), true);
                        }
                    }
                    catch(Exception &E)
                    {
                        if(!AThread->Stopped)
                        {
                            AThread->Stop();
                            AThread->Connection->Disconnect();
                        }
                    }
                }
            }
        }
    }
    __finally
    {
        pMain->IdTCPServer1->Threads->UnlockList();
    }

    // Shutdown computer or close application

    if(pMain->SimIos.Status.iSimStatus == 11)
    {
        pMain->Shutdown = true;
        pMain->CloseApp();
    }
}


void __fastcall TfmMain::IdTCPServer1Connect(TIdPeerThread *AThread)
{
    TClient *pClient = NULL;
    AnsiString ABuffer, text;
    AnsiString PeerIP   = AThread->Connection->Binding->PeerIP;
    TDateTime TimeConnected = Now();


    ABuffer = AThread->Connection->ReadLn();
    if((ABuffer.Pos("TT") == 0) && (ABuffer.Pos("IG") == 0) && (ABuffer.Pos("RR") == 0))
    {
        text = AnsiString().sprintf("1>>> Unknown(%s) on %s connected illegal!...",
                            PeerIP, DateTimeToStr(TimeConnected));
        WriteMsg(text);
        AThread->Connection->Disconnect();
        return;
    }

    if(ABuffer.Pos("IG") != 0)
    {
        pClient = new TClient;
        pClient->PeerIP     = PeerIP;
        pClient->HostName   = Clients[eIG];
        pClient->Id         = eIG;

        AThread->Data = (TObject *)pClient;
        AThread->FreeOnTerminate = true;

        // Report client is on line
    }

    text = AnsiString().sprintf("1>>>%s(%s) on %s on line!...",
                        pClient->HostName, PeerIP, DateTimeToStr(TimeConnected));
    WriteMsg(text);    
}
//---------------------------------------------------------------------------

void __fastcall TfmMain::IdTCPServer1Disconnect(TIdPeerThread *AThread)
{
    TClient *pClient = NULL;
    AnsiString Msg;


    if (IsClosing) return;


    pClient = (TClient *)AThread->Data;

    if(pClient->Id == 1)
    {
        // Report client is off line
        Msg = AnsiString().sprintf("1>>>%s(%s) on %s off line...",
        pClient->HostName, pClient->PeerIP, DateTimeToStr(Now()));
        WriteMsg(Msg);
    }


    delete pClient;
    AThread->Data = NULL;
    AThread->Terminate();

    try
    {
        IdTCPServer1->Threads->LockList()->Remove(AThread);
    }
    __finally
    {
        IdTCPServer1->Threads->UnlockList();
    }

}
//---------------------------------------------------------------------------

void __fastcall TfmMain::IdTCPServer1Execute(TIdPeerThread *AThread)
{
    TClient *pClient;


    if (!AThread->Terminated && !IsClosing)
    {
        pClient = (TClient *)AThread->Data;
        if((pClient != NULL) && (pClient->Id != 0))
        {
            try
            {
                if(pClient->Id == 1)
                {
                    // Report client still alive
                }
            }
            catch(Exception &E)
            {
                if (!IsClosing)
                {
                    if(!AThread->Stopped)
                    {
                        AThread->Stop();
                        AThread->Connection->Disconnect();
                    }
                }
            }
        }
    }
}

客户端

void __fastcall TSocketThread::Execute()
{
    unsigned long ulCheckSum;
    SIM_SVR1_ACK_STRUCT   Ack2Svr;
    SIM_SVR_DATA   DataFromSvr;
    int Counter;


    memset(&DataFromSvr, 0, sizeof(DataFromSvr));
    memset(&Ack2Svr,  0, sizeof(Ack2Svr));
    Counter = 0;

    // fetch and process commands until the connection or thread is terminated
    while (!this->Terminated && FSocket->Connected())
    {
        try
        {
            // recieve data from server
            FSocket->ReadBuffer(&DataFromSvr, sizeof(DataFromSvr));

            // check CRC
            ulCheckSum = CRC_32((unsigned char*)&DataFromSvr.SimData, sizeof(DataFromSvr.SimData));

            if (ulCheckSum == DataFromSvr.uiCheckSum)
            {
                FSmIpcUtil->Writeto(&DataFromSvr.SimData, DATA_OFFSET, sizeof(DataFromSvr.SimData));
            }

            else
            {
                // counter to record error  
                Synchronize(UpdateCaption);
            }

            // read return from local SM
            FSmIpcUtil->Readfrom(&Ack2Svr, ACK_OFFSET, sizeof(Ack2Svr));

            FSocket->WriteBuffer(&Ack2Svr, sizeof(Ack2Svr));

            if (DataFromSvr.SimData.SimIgTgt.Acdata.iSimStatus == 11)
            {
                Terminate();
                FSocket->Disconnect();

                PostMessage(Application->Handle, WM_SHUTDOWN, 0, 0);

                Sleep(500);
            }
        }

        catch (EIdException& E)
        {
            this->Terminate();
            FSocket->Disconnect();
        }
    }
}

Answer 1

您的代码有几个问题。

多媒体计时器回调在允许执行的操作方面非常受限:

Applications should not call any system-defined functions from inside a callback function, except for PostMessage, timeGetSystemTime, timeGetTime, timeSetEvent, timeKillEvent, midiOutShortMsg, midiOutLongMsg, and OutputDebugString.

如果传输速度很重要，则根本不要让计时器回调进行广播。将数据保存在安全的地方，然后让每个 TIdTCPServer 线程按自己的时间获取最新数据。这也使每个连接线程保持隔离，因此如果出现问题，一个连接不会影响任何其他连接。

不要将 TIdPeerThread::FreeOnTerminate 设置为 true，不要调用 TIdPeerThread::Stop()，不要手动从 TIdTCPServer 中删除线程: :Threads 属性。您不拥有线程，TIdTCPServer 拥有线程，它会为您管理它们。如果你想停止一个给定的线程，你需要做的就是关闭线程的套接字。但是通过将所有发送逻辑移动到它所属的 OnExecute 中，您可以让 TIdTCPServer 处理任何错误并为您关闭套接字。

只有当 IG 客户端连接时，您的 OnConnect 事件处理程序才会设置 AThread->Data，但是您的 OnConnect 和 OnDisconnect 处理程序在尝试访问 TClient 对象之前不检查该条件。

如果 IsClosing 为真，则您的 OnDisconnect 事件处理程序正在泄漏内存。它调用 Threads->UnlockList() 而不是先调用 Threads->LockList()。在列表未被调用线程锁定时尝试解锁列表将导致错误和死锁。

尝试更像这样的东西:

struct TClient
{
    AnsiString PeerIP;     //{ Client IP address }
    AnsiString HostName;   //{ Hostname }
    int Id;                //{ Client ID }
};

void CALLBACK mmTimerProc(UINT wTimerID, UINT msg, DWORD dwUser, DWORD dw1, DWORD dw2)
{
    TfmMain *pMain = (TfmMain *)dwUser;

    if (pMain->IsClosing || pMain->Shutdown) return;

    pMain->UpdateServer1Data();    
    // make sure pMain->Data2Client is thread-safe...
    // set a signal that Data2Client has been updated...

    // Shutdown computer or close application

    if (pMain->SimIos.Status.iSimStatus == 11)
    {
        pMain->Shutdown = true;
        pMain->CloseApp();
    }
}

void __fastcall TfmMain::IdTCPServer1Connect(TIdPeerThread *AThread)
{
    TClient *pClient;
    AnsiString ABuffer, text;
    AnsiString PeerIP = AThread->Connection->Binding->PeerIP;
    TDateTime TimeConnected = Now();

    ABuffer = AThread->Connection->ReadLn();
    if ((ABuffer.Pos("TT") == 0) && (ABuffer.Pos("IG") == 0) && (ABuffer.Pos("RR") == 0))
    {
        text = AnsiString().sprintf("1>>> Unknown(%s) on %s connected illegal!...", PeerIP.c_str(), DateTimeToStr(TimeConnected).c_str());
        WriteMsg(text);
        AThread->Connection->Disconnect();
        return;
    }

    pClient = new TClient;
    pClient->PeerIP = PeerIP;

    if (ABuffer.Pos("IG") != 0)
    {
        pClient->HostName   = Clients[eIG];
        pClient->Id         = eIG;
    }
    else
        pClient->Id         = 0;

    AThread->Data = (TObject *)pClient;

    // Report client is on line

    text = AnsiString().sprintf("1>>>%s(%s) on %s on line!...", pClient->HostName.c_str(), PeerIP.c_str(), DateTimeToStr(TimeConnected).c_str());
    WriteMsg(text);    
}

void __fastcall TfmMain::IdTCPServer1Disconnect(TIdPeerThread *AThread)
{
    TClient *pClient = (TClient *)AThread->Data;
    AnsiString Msg;

    AThread->Data = NULL;

    if (pClient)
    {
        // Report client is off line
        Msg = AnsiString().sprintf("1>>>%s(%s) on %s off line...",
            pClient->HostName.c_str(), pClient->PeerIP.c_str(), DateTimeToStr(Now()).c_str());
        WriteMsg(Msg);

        delete pClient;
    }
}

void __fastcall TfmMain::IdTCPServer1Execute(TIdPeerThread *AThread)
{
    TClient *pClient;

    if (IsClosing) return;

    // make sure pMain->Data2Client is thread-safe...
    if (Data2Client has been updated since last event)
    {
        AThread->Connection->WriteBuffer(&pMain->Data2Client, sizeof(pMain->Data2Client), true);
    }

    pClient = (TClient *)AThread->Data;
    // Report client still alive
}