消息模式下的 NamedPipeClientStream 从不将 IsMessageComplete 设置为 true答案

【问题标题】：NamedPipeClientStream in Message Mode never sets IsMessageComplete to true消息模式下的 NamedPipeClientStream 从不将 IsMessageComplete 设置为 true
【发布时间】：2021-12-19 03:02:56
【问题描述】：

我遇到了一个问题，这几天我一直在苦苦挣扎，但无法通过调试或搜索互联网找到任何解释。所以作为最后的手段，我在这里问。

我有一个用 c++ 编写的简单命名管道服务器和一个在 Windows 上运行的 c# 中的 NamedPipeStreamClient。在尝试制定一个在消息前面传递消息大小（以字节为单位）的协议之前，我想尝试使用消息模式来分隔各个消息。

相关代码片段为：

创建 C++ NamedPipe

hOutputPipe = CreateNamedPipeA(
    lOutputPipeName,               // pipe name
    PIPE_ACCESS_OUTBOUND,         // only write access
    PIPE_TYPE_MESSAGE |         // message type pipe
        PIPE_READMODE_MESSAGE | // message-read mode
        PIPE_WAIT,              // blocking mode
    PIPE_UNLIMITED_INSTANCES,   // max. instances
    1024,                    // output buffer size
    1024,                    // input buffer size
    0,                          // client time-out
    NULL);                      // default security attribute
if (hOutputPipe == INVALID_HANDLE_VALUE)
{
    std::cout << "CreateNamedPipe failed, GLE=" << GetLastError() << std::endl;
    return -1;
}

// Wait for the client to connect; if it succeeds,
// the function returns a nonzero value. If the function
// returns zero, GetLastError returns ERROR_PIPE_CONNECTED.

BOOL fConnected = ConnectNamedPipe(hOutputPipe, NULL) ? TRUE : (GetLastError() == ERROR_PIPE_CONNECTED);

if (!fConnected)
{
    // The client could not connect, so close the pipe.
    CloseHandle(hOutputPipe);
    return -1;
}

C# NamedPipeClientStream 的创建：

fromagent_pipe = new NamedPipeClientStream(".", pipeName + "_output", PipeDirection.In);
// timeout after 2 seconds to prevent blocking
fromagent_pipe.Connect(2000);

fromagent_pipe.ReadMode = PipeTransmissionMode.Message;

在 while(true) 循环中调用的 C++ 函数：

bool writeOutputMessage(HANDLE hPipe, const CubesExample::OutputMessage& outputMessage)
{
    size_t messageBytes = outputMessage.ByteSizeLong();
    char buffer[messageBytes];
    DWORD bytesWritten = 0;
    outputMessage.SerializeToArray(&buffer, messageBytes);

    std::cout << std::string("Writing ") + std::to_string(messageBytes) + " bytes" << std::endl;

    BOOL fSuccess = WriteFile(
        hPipe,        // handle to pipe
        buffer,     // buffer to write from
        messageBytes, // number of bytes to write
        &bytesWritten,   // number of bytes written
        NULL);        // not overlapped I/O
    if (!fSuccess || bytesWritten != messageBytes)
    {
        std::cout << "InstanceThread WriteFile failed, GLE=" << GetLastError() << std::endl;
        return false;
    }

    return true;
}

从管道读取完整消息并返回字节[]的C#方法：

public byte[] readOutputMessage()
{        
    int offset = 0;
    int readBytes = 0;
    do{
        readBytes = fromagent_pipe.Read(inputBuffer, offset, 1);
        offset++;
        Debug.Log("Reading from output pipe! isComplete: " + fromagent_pipe.IsMessageComplete + " readBytes: " + readBytes);
    }while(!fromagent_pipe.IsMessageComplete && readBytes > 0);

    Debug.Log("Read " + offset + " bytes from agent pipe");

    byte[] buffer = new byte[offset];
    Array.Copy(inputBuffer, buffer, offset);
    return buffer;
}

上面的 C# 方法在 Task<byte[]> 中运行，因此在等待 PipeStream.Read() 时不会阻塞主线程。 inputBuffer 是该代码所在类中的一个字段，大小为 4096，因此我不必在每次读取之前分配它。

现在的问题是，fromagent_pipe.IsMessageComplete 永远不会设置为 true，无论它读取多少字节。我发送的消息大小为 7 个字节以供参考，所以我希望 do while 循环迭代 7 次，并且在读取第 7 个字节后 IsMessageComplete 应该设置为 true，对吗？我是命名管道和 IPC 的新手，所以我可能会遗漏一些明显的东西，但这让我发疯，因为我从文档中设置了所有可能的标志，并像互联网上的其他人一样使用 IsMessageComplete 标志，但我的似乎永远不会切换到真的。

另一条信息是 c++ 服务器的运行速度比 c# 循环快得多，因此管道消耗数据的速度比获取数据的速度要慢。我最终会丢弃在单次读取之间传递的所有消息，但现在我什至无法读取一条消息。

PS。在有人指出之前，是的，inputBuffer 在读取 4096 个字节后确实会溢出。我想在处理重置之前解决 IsMessageComplete 问题。

感谢任何提前阅读并...请发送帮助

【问题讨论】：

标签： c# c++ windows io named-pipes

【解决方案1】：

通过将管道方向设置为双工，我能够使消息模式正常工作。这个accepted answer 解释了为什么我必须这样做。

这是我快速制作的示例。我在C# 端提供了一个异步和同步示例。我没有提供在C++ 端使用OVERLAPPED 的异步示例。

服务器

int main( ) 
{
    std::cout << "Press any key to begin\n";
    std::cin.ignore( );

    const auto print_error{ [ ]( DWORD code ) 
    { 
        const std::error_code ec{ static_cast<int>( code ),
            std::system_category( ) };

        std::cerr << ec.message( ) << '\n';
    } };

    const auto pipe{ CreateNamedPipeA( 
        R"(\\.\pipe\testpipe)",
        PIPE_ACCESS_DUPLEX,       
        PIPE_TYPE_MESSAGE | PIPE_READMODE_MESSAGE | PIPE_WAIT,              
        PIPE_UNLIMITED_INSTANCES,  
        1024,                    
        1024,                    
        0,                          
        nullptr ) };

    if ( pipe == INVALID_HANDLE_VALUE ) 
    {
        print_error( GetLastError( ) );
        return EXIT_FAILURE;
    }

    if ( !ConnectNamedPipe( pipe, nullptr ) ) 
    {        
        if ( const auto ec{ GetLastError( ) }; ec != ERROR_PIPE_CONNECTED ) 
        {
            print_error( ec );
            CloseHandle( pipe );
            return EXIT_FAILURE;
        }        
    }

    std::string_view message{ "Hello friend!" };
    const auto length{ static_cast<DWORD>( message.size( ) ) };

    std::cout << "Sending message\n";

    DWORD sent{ 0 };
    const auto result{ WriteFile(
        pipe, message.data( ), length , &sent, nullptr ) };

    if ( !result || sent != length ) 
    {
        print_error( GetLastError( ) );
        CloseHandle( pipe );
        return EXIT_FAILURE;
    }

    std::cout << "Message sent, press enter to exit\n";
    std::cin.ignore( );

    std::cout << "Closing handle\n";
    CloseHandle( pipe );
}

同步客户端

static void Main( string[ ] args )
{
    Console.WriteLine( "Press any key to begin" );
    Console.ReadKey( );

    using var client = new NamedPipeClientStream( ".", "testpipe", PipeDirection.InOut );
    Console.WriteLine( "Connecting to server" );

    client.Connect( 5000 );
    client.ReadMode = PipeTransmissionMode.Message;

    Console.WriteLine( "Connected" );

    var buffer = new byte[ 1024 ];
    var nBytes = 0;

    do
    {
        nBytes += client.Read( buffer.AsSpan( )[ nBytes.. ] );

    } while ( !client.IsMessageComplete );


    var message = Encoding.ASCII.GetString( buffer.AsSpan( )[ ..nBytes ] );
    Console.WriteLine( $"Received message: {message}" );

    Console.WriteLine( "Press any key to exit" );
    Console.ReadKey( );
}

异步客户端

static async Task Main( string[ ] args )
{
    Console.WriteLine( "Press any key to begin" );
    Console.ReadKey( );

    using var client = new NamedPipeClientStream( ".", "testpipe", PipeDirection.InOut );
    Console.WriteLine( "Connecting to server" );

    using ( var cts = new CancellationTokenSource( TimeSpan.FromSeconds( 5 ) ) )
    {
        await client.ConnectAsync( cts.Token ).ConfigureAwait( false );
        client.ReadMode = PipeTransmissionMode.Message;
    }
        
    Console.WriteLine( "Connected" );

    var buffer = new byte[ 1024 ];
    var nBytes = 0;

    do
    {                
        nBytes += await client.ReadAsync( 
            buffer.AsMemory( )[ nBytes.. ] ).ConfigureAwait( false );

    } while ( !client.IsMessageComplete );


    var message = Encoding.ASCII.GetString( buffer.AsSpan( )[ ..nBytes ] );
    Console.WriteLine( $"Received message: {message}" );

    Console.WriteLine( "Press any key to exit" );
    Console.ReadKey( );
}

【讨论】：