Читать SSL через PipeReader в .NET

В настоящее время у меня есть рабочая реализация, использующая поток SSL, обернутая в поток с буферизацией, и просто вызывающая чтение/запись в потоке с использованием байтовых массивов.

Я хочу сделать это быстрее, и из некоторого прочтения похоже, что System.IO.Pipelines - это путь к высокопроизводительному вводу-выводу.

Многие статьи/демонстрации, которые я только читаю, демонстрируют код, использующий сокет напрямую - который, кажется, не работает со мной, так как я использую SSL.

Я нашел несколько расширений, чтобы получить конвейер/писатель из потока> Stream.UsePipeReader() или Stream.UsePipeWriter(), поэтому я попытался вызвать SSLStream.UsePipeReader()

Однако я постоянно получаю ошибку:

System.NotSupportedException :  The ReadAsync method cannot be called when another read operation is pending.
   at System.Net.Security.SslStreamInternal.ReadAsyncInternal[TReadAdapter](TReadAdapter adapter, Memory'1 buffer)
   at Nerdbank.Streams.PipeExtensions.<>c__DisplayClass5_0.<<UsePipeReader>b__1>d.MoveNext() in D:\a\1\s\src\Nerdbank.Streams\PipeExtensions.cs:line 92
--- End of stack trace from previous location where exception was thrown ---
   at System.IO.Pipelines.PipeCompletion.ThrowLatchedException()
   at System.IO.Pipelines.Pipe.GetReadResult(ReadResult& result)
   at System.IO.Pipelines.Pipe.GetReadAsyncResult()

Мой код для чтения из канала:

private async Task<string> ReadAsync(PipeReader reader)
        {

            var stringBuilder = new StringBuilder();

            while (true)
            {
                var result = await reader.ReadAsync();

                var buffer = result.Buffer;
                SequencePosition? position;

                do
                {
                    // Look for a EOL in the buffer
                    position = buffer.PositionOf((byte) '\n');

                    if (position != null)
                    {
                        // Process the line
                        ProcessLine(buffer.Slice(0, position.Value), stringBuilder);
                        buffer = buffer.Slice(buffer.GetPosition(1, position.Value));
                    }
                } while (position != null);

                reader.AdvanceTo(buffer.Start, buffer.End);

                if (result.IsCompleted)
                {
                    reader.Complete();
                    break;
                }
            }

            return stringBuilder.ToString();
        }

Это не вызывается никакими другими потоками, так как я тестировал его с замком вокруг него. И я делаю только один звонок за один раз для целей тестирования.

Может кто-нибудь сказать мне, что я делаю не так?

Спасибо!

Ответ 1

Я верю, что могу знать ответ.

Ваша проблема может быть в этой строке:

var result = await reader.ReadAsync();

Видите ли, ожидание блокирует выполнение только в методе Send, но при этом код в вызывающей программе продолжает выполняться, даже если ReadAsync() еще не завершен.

Теперь, если метод Send вызывается снова до завершения ReadAsync(), вы получите исключение, которое вы опубликовали, поскольку SslStream не разрешает множественные операции чтения/записи, а код, который вы опубликовали, не предотвращает этого.

В этом случае вам нужно написать свой собственный механизм сбоя безопасности, чтобы избежать второго вызова до завершения первого.

Поскольку вы используете цикл здесь:

while (true)
{
  //rest of code...
}

Я полагаю, что именно здесь вы должны решить свою проблему. Вы достигаете второй итерации, пока первая еще не завершена, в результате чего:

System.NotSupportedException

Ответ 2

вы не можете выполнить readasync внутри другого readasync. Я думаю, что вы, производные от pipereader, уже изменили имя метода на другое.

private async Task<string> **ReadAsync**(PipeReader reader)
        {

            var stringBuilder = new StringBuilder();

            while (true)
            {
                var result = await reader.**ReadAsync**();

Попробуйте изменить это так.

**

async ValueTask ReadSomeDataAsync(PipeReader reader)
{
    while (true)
    {
        // await some data being available
        ReadResult read = await reader.ReadAsync();
        ReadOnlySequence<byte> buffer = read.Buffer;
        // check whether we've reached the end
        // and processed everything
        if (buffer.IsEmpty && read.IsCompleted)
            break; // exit loop
        // process what we received
        foreach (Memory<byte> segment in buffer)
        {
            string s = Encoding.ASCII.GetString(
                segment.Span);
            Console.Write(s);
        }
        // tell the pipe that we used everything
        reader.AdvanceTo(buffer.End);
    }
}

**

В любом случае, вы все еще можете сделать поток памяти тоже.