命名管道服务器，如何中断或超时等待客户端连接和传入数据

Question

我正在为 Windows 编写一个简单的命名管道服务器，调用 Windows API（在 Java 中使用 JNA，但这不相关）。

我试图弄清楚如何避免服务器永远卡住等待客户端连接或来自客户端的数据。

服务器代码执行以下操作：

1) 它通过调用CreateNamedPipe 创建管道，在dwPipeMode 参数中使用PIPE_WAIT。

2) 它调用ConnectNamedPipe，直到客户端连接后才会返回。

3) 它进入一个循环，它通过调用ReadFile 从客户端重复读取一条消息，直到读取数据才返回，并且对于每条接收到的消息，它通过调用将一条消息发送回客户端作为响应WriteFile.

4) 在多次这样的对话之后，客户端和服务器将断开管道。

我只是希望能够在第 2 步等待 ConnectNamedPipe 和在第 3 步等待 ReadFile 时设置超时，但我看不到在哪里设置超时。 CreateNamedPipe 函数中有 nDefaultTimeOut 参数，但听起来并不适合这样做； API 文档说：

默认超时值，以毫秒为单位，如果WaitNamedPipe 函数指定NMPWAIT_USE_DEFAULT_WAIT。

所以CreateNamedPipe 中的nDefaultTimeOut arg 听起来像是客户端 将连接到管道的默认超时将用于他们的操作，并且只有当他们调用WaitNamedPipe 函数时。事实上，在我的测试中，值 0 或 1000 并没有什么区别，对 ConnectNamedPipe 的调用永远不会返回（除非客户端连接）。我正在寻找的是服务器超时，而不是调用ConnectNamedPipe 和ReadFile。

作为CreateNamedPipe 的文档，对于带有PIPE_WAIT 的dwPipeMode 参数说，Blocking mode is enabled. When the pipe handle is specified in the ReadFile, WriteFile, or ConnectNamedPipe function, the operations are not completed until there is data to read, all data is written, or a client is connected. Use of this mode can mean waiting indefinitely in some situations for a client process to perform an action.

因此，实现此类超时的方法可能是以非阻塞模式创建管道（使用PIPE_NOWAIT 而不是PIPE_WAIT），以便对ReadFile、WriteFile 和ConnectNamedPipe 的调用立即返回，然后以某种方式在循环中监控自己的事件（客户端连接或接收数据），并在循环中检查自己是否超时或发生了另一个中断事件（如用户单击取消按钮）？

添加：对于ReadFile 调用，我可以使用立即返回的PeekNamedPipe 检查是否有要读取的数据，然后才调用ReadFile .我会试试的。但是我对ConnectNamedPipe的调用仍然有同样的问题。

添加：正如我所怀疑和答案所证实的那样，作为管道的新手，我从某种偏斜的角度看待它们，从这个角度来看，对超时的需求似乎比实际更大。

F.ex.想要超时调用ReadFile 背后的原因是，如果我（服务器）在它里面从客户端读取数据并且客户端突然关闭，有时我可能最终会卡在ReadFile 中。但现在我知道如果ReadFile 正在从管道读取并且客户端关闭，ReadFile 将总是出错，因此执行不会卡在其中。

Answer 1

我建议你设置FILE_FLAG_OVERLAPPED 并使用事件来检查/等待完成。

虽然这最初是为异步 IO 设计的，但您可以改为将事件计时到您预定义的生存时间。

如果您想取消 I/O 操作，可以使用 CancelIo() 函数。如果您只是想做一些工作然后继续等待，您也可以这样做 - 超时等待不会自动取消 I/O，因此您不需要再次调用 ConnectNamedPipe。

您也可以按照您自己的建议设置PIPE_NOWAIT 并轮询连接直到成功，在此用例中，无论哪种方式都应该带来相同的结果。但请注意，这是旧功能，Microsoft 不鼓励使用此选项。

Answer 2

在 GUI 应用程序中演示异步使用管道服务器端的一些真实代码：

void wait_for_object(HANDLE object)
{
  DWORD dw;
  MSG msg;

  for (;;) 
  {
    dw = MsgWaitForMultipleObjectsEx(1, &object, INFINITE, QS_ALLINPUT, 0);

    if (dw == WAIT_OBJECT_0) break;
    if (dw == WAIT_OBJECT_0 + 1) 
    {
      while (PeekMessage(&msg, NULL, 0, 0, PM_REMOVE)) DispatchMessage(&msg);
      continue;
    }
    srvfail(L"sleep() messageloop", GetLastError());
  }
}

HANDLE server_pipe;
HANDLE io_event;

void pipe_connection(void)
{
    OVERLAPPED overlapped;
    DWORD dw, err;

    SecureZeroMemory(&overlapped, sizeof(overlapped));
    overlapped.hEvent = io_event;

    if (!ReadFile(server_pipe, input_buffer, sizeof(input_buffer) - 1, NULL, &overlapped))
    {
        err = GetLastError();
        if (err == ERROR_IO_PENDING)
        {
            wait_for_object(io_event);
            if (!GetOverlappedResult(server_pipe, &overlapped, &dw, FALSE)) 
            {
                srvfail(L"Read from pipe failed asynchronously.", GetLastError());
            }
        }
        else
        {
            srvfail(L"Read from pipe failed synchronously.", GetLastError());
        }
    }
    else
    {
        if (!GetOverlappedResult(server_pipe, &overlapped, &dw, FALSE)) 
        {
            srvfail(L"GetOverlappedResult failed reading from pipe.", GetLastError());
        }
    }

    input_buffer[dw] = '\0';

    process_command();

    if (!WriteFile(server_pipe, &output_struct, 
        ((char *)&output_struct.output_string - (char *)&output_struct) + output_struct.string_length, 
        NULL, &overlapped))
    {
        err = GetLastError();
        if (err == ERROR_IO_PENDING)
        {
            wait_for_object(io_event);
            if (!GetOverlappedResult(server_pipe, &overlapped, &dw, FALSE)) 
            {
                srvfail(L"Write to pipe failed asynchronously.", GetLastError());
            }
        }
        else
        {
            srvfail(L"Write to pipe failed synchronously.", GetLastError());
        }
    }
    else
    {
        if (!GetOverlappedResult(server_pipe, &overlapped, &dw, FALSE)) 
        {
            srvfail(L"GetOverlappedResult failed writing to pipe.", GetLastError());
        }
    }

    if (!FlushFileBuffers(server_pipe)) srvfail(L"FlushFileBuffers failed.", GetLastError());
    if (!DisconnectNamedPipe(server_pipe)) srvfail(L"DisconnectNamedPipe failed.", GetLastError());
}

void server(void)
{
    OVERLAPPED overlapped;
    DWORD err, dw; 

    // Create the named pipe

    server_pipe = CreateNamedPipe(pipe_name, PIPE_ACCESS_DUPLEX | FILE_FLAG_FIRST_PIPE_INSTANCE | FILE_FLAG_OVERLAPPED, PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE, 1, buffer_size, buffer_size, 0, NULL);
    if (server_pipe == INVALID_HANDLE_VALUE) srvfail(L"CreateNamedPipe failed.", GetLastError());

    // Wait for connections

    io_event = CreateEvent(NULL, FALSE, FALSE, NULL);
    if (io_event == NULL) srvfail(L"CreateEvent(io_event) failed.", GetLastError());

    for (;;)
    {
        SecureZeroMemory(&overlapped, sizeof(overlapped));
        overlapped.hEvent = io_event;

        if (!ConnectNamedPipe(server_pipe, &overlapped))
        {
            err = GetLastError();
            if (err == ERROR_PIPE_CONNECTED)
            {
                pipe_connection();
            }
            else if (err == ERROR_IO_PENDING)
            {
                wait_for_object(io_event);
                if (!GetOverlappedResult(server_pipe, &overlapped, &dw, FALSE)) 
                {
                    srvfail(L"Pipe connection failed asynchronously.", GetLastError());
                }
                pipe_connection();
            }
            else
            {
                srvfail(L"Pipe connection failed synchronously.", GetLastError());
            }
        }
        else
        {
            if (!GetOverlappedResult(server_pipe, &overlapped, &dw, FALSE)) 
            {
                srvfail(L"GetOverlappedResult failed connecting pipe.", GetLastError());
            }
            pipe_connection();
        }
    }
}

(这段代码已经从原文中删减了多余的逻辑。我没有尝试编译编辑后的版本，所以可能会有一些小问题。还要注意全局变量的使用，这在我的情况下是可以的因为应用程序非常小，但应该通常避免使用。）

使用 MsgWaitForMultipleObjectsEx() 允许在等待 I/O 完成时处理窗口消息。如果您还在等待其他事情发生，您可以向它传递一个句柄数组，而不仅仅是一个句柄 - 例如，如果您想监视一个子进程并在它退出时执行某些操作，您可以传递一个包含io_event 和进程句柄。如果你只是不得不定期做一些其他的工作，你可以为等待设置一个超时，或者使用一个窗口计时器。