TcpClient не поспевает за данными от сервера

330

28 апреля 2017, 16:33

Проблема: отправляю данные с сервера на клиент (оба на локальной машине) и часть данных клиент не успевает обработать/принять в буфер. Иногда случается так, что успевает все обработать, но чаще нет. Провел тест со встроенным telnet клиентом винды, он же всегда все успевает принять и вывести. Попробовал сделать задержку отправки в виде Thread.Sleep(50) и в итоге клиент стал успевать, но это ли выход?

Далее код.

Код отправки сообщений на стороне сервера:

public async Task Send<T, U>(Packet<T, U> message)
{
    var buffer = message?.ToBytes();
    if (buffer?.Length > 0)
    {
        var guid = Guid.NewGuid();
        var countSemgment = Math.Ceiling(buffer.Length / (double)BufferSize);
        for (int index = 0; index < countSemgment; index++)
        {
            var bytes = AddGuid(buffer.Skip(index * BufferSize).Take(BufferSize).ToArray(), guid);
            await _stream.WriteAsync(bytes, 0, bytes.Length);
        }
    }
}

Guid нужен для идентификации сообщения если данные больше размера буфера.

Код клиента, который принимает входящие сообщения:

private async void Receive()
{
    byte[] buffer = new byte[BufferSize];
    try
    {
        while (true)
        {
            var count = await _stream.ReadAsync(buffer, 0, buffer.Length);
            RaiseEventMessegaReceive(buffer, count);
        }
    }
    catch (Exception e)
    {
        Dispose();
    }
}

Код вызов события RaiseEventMessageReceive у клиента:

protected virtual void RaiseEventMessegaReceive(byte[] data, int count)
{
    MsgReceivEvent?.Invoke(this, new MessageReceiveEventArgs(data, count));
}

Привязываю к событию я этот метод:

public async void PacketExecute(object o, MessageReceiveEventArgs msg)
{
    await PacketHandler(msg.Data, msg.Count);
}

И сам код обработчик входящих данных

private async Task PacketHandler(byte[] data, int count)
{
    await Task.Run(() =>
    {
        try
        {
            Packet packet;
            using (BinaryReader br = new BinaryReader(new MemoryStream(data.Take(count).ToArray())))
            {
                byte[] msg;
                var guid = new Guid(br.ReadBytes(16));
                Console.WriteLine($@"Message Guid = {guid}");
                if (PacketDictionary.TryGetValue(guid, out packet))
                {
                    msg = br.ReadBytes(count - guid.ToByteArray().Length);
                    packet.Add(msg);
                }
                else
                {
                    var groupCommand = br.ReadByte();
                    var command = br.ReadByte();
                    var dataLen = br.ReadInt32();
                    msg = dataLen > bufferSize - _service_len ? br.ReadBytes(bufferSize - _service_len) : br.ReadBytes(dataLen);
                    packet = new Packet(guid, groupCommand, command, msg, dataLen);
                    packet.PacketReadyEvent += PacketReadyExecute;
                    packet.PacketFailEvent += PacketFailExecute;
                    packet.StatusChange();
                }
                PacketDictionary.AddOrUpdate(guid, packet, (g, p) => packet);
            }
        }
        catch (Exception e)
        {
            Console.WriteLine(e);
            throw;
        }
    });
}

Answer 1

Я вижу по крайней мере две «гонки» (race condition) в вашем коде.

Во-первых, ваш цикл

byte[] buffer = new byte[BufferSize];
try
{
    while (true)
    {
        var count = await _stream.ReadAsync(buffer, 0, buffer.Length);
        RaiseEventMessegaReceive(buffer, count);
    }
}
catch (Exception e)
{
    Dispose();
}

получает данные в один и тот же буфер. Поэтому если придут новые данные до окончания обработки старых, они затрут старые данные. Имеет смысл перенести выделение буфера внутрь цикла:

    while (true)
    {
        byte[] buffer = new byte[BufferSize];
        var count = await _stream.ReadAsync(buffer, 0, buffer.Length);
        RaiseEventMessegaReceive(buffer, count);
    }

Во-вторых, вы ожидаете, что ReadAsync прочитает весь отправленный при помощи WriteAsync пакет данных. Это не так, границы принятых кусков вовсем не обязаны совпадать с границами отправленных кусков. Ваш код пробует собрать данные из нескольких пакетов в PacketHandler, но это слишком поздно: ваш код ожидает GUID в начале пришедшего куска данных, а его там вполне может и не оказаться.

Имеет смысл каждый передаваемый кусок предварять его длиной. Например, так:

(Отправка)

async Task SendAsync(byte[] buffer)
{
    byte[] lengthBytes = BitConverter.GetBytes(buffer.Length);
    await _stream.WriteAsync(lengthBytes, 0, lengthBytes.Length);
    await _stream.WriteAsync(bytes, 0, bytes.Length);
}

for (int index = 0; index < countSemgment; index++)
{
    var bytes = AddGuid(buffer.Skip(index * BufferSize).Take(BufferSize).ToArray(), guid);
    await SendAsync(bytes);
}

(Приём)

async Task<byte[]> ReceiveAsync(int nBytesExact)
{
    var buf = new byte[nBytesExact];
    var readpos = 0;
    while (readpos < nBytesExact)
    {
        var actuallyRead = await _stream.ReadAsync(buf, readpos, nBytesExact - readpos);
        if (actuallyRead == 0)
            throw new EndOfStreamException();
        readpos += actuallyRead;
    }
    return buf;
}

while (true)
{
    var messageLenBytes = await ReceiveAsync(4);
    var messageLen = BitConverter.ToInt32(messageLenBytes, 0);
    var buffer = await ReceiveAsync(messageLen);
    RaiseEventMessegaReceive(buffer, buffer.Length);
}