TCP 套接字异常断开和写入超时

Question

我将尝试用最简短的语言解释这个问题。我正在使用 c++ builder 2010。

我正在使用 TIdTCPServer 并将语音数据包发送到已连接的客户端列表。一切正常，直到任何客户端异常断开连接，例如电源故障等。我可以通过切断已连接客户端的以太网连接来重现类似的断开连接。

现在我们有一个断开连接的套接字，但正如您所知，服务器端尚未检测到它，因此服务器也将继续尝试向该客户端发送数据。

但是，当服务器尝试向断开连接的客户端写入数据时…… Write() 或 WriteLn() 在尝试写入时挂起，就像在等待某种写入超时。这会挂起空洞数据包分发过程，从而导致向所有其他客户端的数据传输出现延迟。几秒钟后，出现“套接字连接关闭”异常，数据流继续。

这是代码

try
{
EnterCriticalSection(&SlotListenersCriticalSection);
for(int i=0;i<SlotListeners->Count;i++)
 {
    try
    {

      //Here the process will HANG for several seconds on a disconnected socket
      ((TIdContext*) SlotListeners->Objects[i])->Connection->IOHandler->WriteLn("Some DATA");

   }catch(Exception &e)
   {
     SlotListeners->Delete(i);
   }
}
}__finally
{
 LeaveCriticalSection(&SlotListenersCriticalSection);
}

好吧，我已经有一个保持活动机制，可以在 n 秒不活动后断开套接字。但正如您可以想象的那样，该机制仍然无法与该广播循环完全同步，因为该广播循环几乎一直在运行。

那么我可以通过 iohandler 或其他方式指定写入超时吗？我已经看到很多关于“检测断开连接的 tcp 套接字”的线程，但我的问题略有不同，我需要在写入尝试期间避免挂起几秒钟。

那么有什么解决办法吗？

或者我应该考虑对此类数据广播使用某种不同的机制，例如广播循环将数据包放入某种 FIFO 缓冲区中，并且客户端线程不断检查可用数据并将其传递给自己？这样，如果一个线程挂起，它不会停止/延迟所有分发线程。

有什么想法吗？感谢您的时间和帮助。

问候

果酱

Answer 1

Indy 中没有实现写入超时。为此，您必须使用 TIdSocketHandle.SetSockOpt() 方法直接设置套接字级超时。

FIFO 缓冲区是一个更好的选择（总体上也是一个更好的设计）。例如：

void __fastcall TForm1::IdTCPServer1Connect(TIdContext *AContext)
{
    ...
    AContext->Data = new TIdThreadSafeStringList;
    ...
}

void __fastcall TForm1::IdTCPServer1Disconnect(TIdContext *AContext)
{
    ...
    delete AContext->Data;
    AContext->Data = NULL;
    ...
}

void __fastcall TForm1::IdTCPServer1Execute(TIdContext *AContext)
{
    TIdThreadSafeStringList *Queue = (TIdThreadSafeStringList*) AContext->Data;
    TStringList *Outbound = NULL;
    TStringList *List = Queue->Lock();
    try
    {
        if( List->Count > 0 )
        {
            Outbound = new TStringList;
            Outbound->Assign(List);
            List->Clear();
        }
    }
    __finally
    {
        Queue->Unlock();
    }

    if( Outbound )
    {
        try
        {
            AContext->Connection->IOHandler->Write(Outbound);
        }
        __finally
        {
            delete Outbound;
        }
    }

    ...
}

...

try
{
    EnterCriticalSection(&SlotListenersCriticalSection);
    int i = 0;
    while( i < SlotListeners->Count )
    {
        try
        {
          TIdContext *Ctx = (TIdContext*) SlotListeners->Objects[i];
          TIdThreadSafeStringList *Queue = (TIdThreadSafeStringList*) Ctx->Data;
          Queue->Add("Some DATA"); 
          ++i;
        }
        catch(const Exception &e) 
        { 
            SlotListeners->Delete(i); 
        } 
    } 
}
__finally 
{ 
    LeaveCriticalSection(&SlotListenersCriticalSection); 
}