Problem : I send data from the server to the client (both on the local machine) and the client does not have time to process / receive some of the data into the buffer. Sometimes it happens that it manages to process everything, but more often it does not. Conducted a test with the Windows client built-in telnet, he always manages to accept and withdraw everything. I tried to make a sending delay in the form of Thread.Sleep (50) and as a result the client began to keep up, but is this the way out?

Next code.

Code for sending messages on the server side:

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 is needed to identify a message if data is larger than the buffer size

Client code that accepts incoming messages:

 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(); } }

The code to call the RaiseEventMessageReceive event at the client:

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

I tie this event to the event:

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

And the code itself handler incoming data

 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; } }); }
  • Stop, you, I hope, understand that a message sent in one piece can come in several pieces? - VladD
  • I do not understand this, but if even so, then why does the built-in telnet client normally get everything that comes to it? - adrug
  • blog.stephencleary.com/2009/04/message-framing.html - VladD
  • Because he does not stop reading after the first piece arrived, but reads before CR? - VladD
  • But no, reception is carried out in asynchronous mode. If the sent message comes cut, then the number of calls _stream.ReadAsync will be more so? Without any processing, I counted how many times it came. - adrug

1 answer 1

I see at least two race conditions in your code.

First, your cycle

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

Receives data in the same buffer. Therefore, if new data comes before the end of the processing of old data, they will erase the old data. It makes sense to transfer the buffer allocation inside the loop:

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

Secondly, you expect ReadAsync read the entire data packet sent by WriteAsync . This is not the case; the boundaries of the pieces taken are not necessarily equal to the boundaries of the pieces sent. Your code tries to collect data from several packages in PacketHandler , but it is too late: your code expects a GUID at the beginning of the incoming piece of data, but it may well not be there.

It makes sense to transfer each piece to its length. For example:

(Sending)

 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); }

(Reception)

 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); }

(Related article: TCP / IP .NET Sockets FAQ / Message Framing .)