Повторное использование потоков в UDP-сервере

Question

Я пишу UDP-сервер и хочу уменьшить количество распределений, которые происходят на каждый входящий и исходящий пакеты датаграмм. Одна из вещей, которые я хотел бы сделать, - это использовать потоки, которые я выделяю, чтобы читать и писать мои пакеты дейтаграмм.

У меня есть этот метод, который использует экземпляр BinaryReader при получении пакетов для чтения буфера на объект экземпляра.

// Simplified version of the receive data method. 
private void ReceiveClientData(IAsyncResult result) 
{ 
    int receivedData = socket.EndReceiveFrom(result, ref endPoint); 
    if (receivedData == 0) 
    { 
     this.ListenForData(socket); 
     return; 
    } 

    // Read the header in from the buffer first so we know what kind of message and how to route. 
    this.binaryReader.BaseStream.SetLength(receivedData); 
    this.binaryReader.BaseStream.Seek (0, SeekOrigin.Begin); 

    // Can't reset the MemoryStream buffer to the new packet buffer. 
    // How does this get consumed by the stream without allocating a new MemoryStream(buffer)? 
    byte[] buffer = (byte[])result.AsyncState; 

    // Deserialize the bytes into the header and datagram. 
    IClientHeader header = new ClientHeader(); 
    header.GetMessageFromBytes(this.binaryReader); 
    IClientDatagram datagram = this.datagramFactory.CreateDatagramFromClientHeader(header); 
    datagram.GetMessageFromBytes(this.binaryReader); 

    this.ListenForData(socket); 
} 

Безопасно ли повторно использовать тот же BinaryReader и лежащий в основе MemoryStream обеспечения I Seek обратно к началу потока? После некоторого чтения кажется, что Socket не будет одновременно читать или писать пакеты, что потребует от меня использовать Stream за звонок до EndReceiveFrom. Похоже, что вы можете выполнять каждую операцию одновременно, например, читать + писать, но не выполнять одновременные чтения или параллельные записи. Из-за этого, я думаю, я мог бы повторно использовать читателя, но я не уверен, могу ли я это сделать. Для этого мне пришлось бы дать MemoryStream новый буфер для чтения. Можете ли вы это сделать или мне нужно создать новый MemoryBuffer для каждого нового пакета, полученного из сокета?

Я хотел бы использовать этот же подход с отправкой пакетов обратно клиенту. Для этого я попытался установить длину моего MemoryStream на 0 и вернуться к началу. Это, похоже, не очищает буфер, как описано в another answer here. Я все еще вижу старые данные буфера в t ниже.

public void SendMessage(IServerDatagram message) 
{ 
    this.binaryStream.Seek(0, SeekOrigin.Begin); 
    this.binaryStream.SetLength (0); 

    // Contains old buffer data. 
    var t = this.binaryStream.GetBuffer(); 
    message.Serialize (this.binaryWriter); 

    // Contains a mixture of new bytes + old bytes. 
    byte[] data = this.binaryStream.GetBuffer(); 
    this.binaryWriter.Flush(); 

    // Send the datagram packet. 
    this.udpServerSocket.SendTo(data, this.connectedClients.FirstOrDefault().Key); 
} 

Я не знаю заранее, насколько большой буфер должен быть, поэтому я не могу SetLength к размеру моего нового буфера. Нет ли другого способа повторного использования Stream без необходимости экземпляра нового для каждого сообщения? Если я отправляю тысячи сообщений в секунду, что может вызвать некоторое давление памяти, не так ли?

Answer 1

Я столкнулся с проблемами, связанными с использованием потоков, а также с читателями и сетевыми материалами - и решил использовать несколько иной подход. Я изучил реализации BinaryReader, BinaryWriter и BitConverter и написал методы расширения для чтения и записи данных непосредственно в базовом буфере []. Мягкая PITA, но у меня больше нет потоков, читателей и писателей.

Вот, к примеру, является ИНТОМ-писание метод расширения:

[System.Security.SecuritySafeCritical] 
public unsafe static int Write(this byte[ ] array, int value, int offset = 0) 
{ 
    if (offset + sizeof(int) > array.Length) throw new IndexOutOfRangeException(); 
    fixed (byte* b = array) 
    { 
    *((int*) (b + offset)) = value; 
    } 
    return sizeof(int); 
} 

Возвращение может выглядеть вялым, но все методы расширения возвращают количество байт, я «переехал» в массив байт. .. и поэтому я могу постоянно знать, где я нахожусь; псевдо-поток, если хотите. Кроме того, код написания не всегда знает, что он пишет. Для каждого простого типа есть одинаковые переопределения, один для строк и один для байта [].

Моя проблема была на стороне клиента, где у меня была более ограниченная операционная среда ... и я поддерживаю долговечный буфер чтения. Метод UdpClient.Receive создает новые байт-массивы для каждого чтения и только что убил меня. GC сошел с ума ... что, конечно, очень разрушительно для потоковой передачи UDP :-) Итак, я нашел варианты Receive и ReceiveAsync, и обнаружил, что могу контролировать, что базовый носок нужно было кормить, и снова сделал свой собственный RecieveBroadcastToBuffer метод расширения:

const int MaxUdpSize = 0x0000ffff; 
const int AnyPort = 0; 
static EndPoint anyV4Endpoint = new IPEndPoint(IPAddress.Any, AnyPort); 
static EndPoint anyV6Endpoint = new IPEndPoint(IPAddress.IPv6Any, AnyPort); 

/// <summary>Receives a UDP datagram into the specified buffer at the specified offset</summary> 
/// <returns>The length of the received data</returns> 
public static int ReceiveBroadcastToBuffer(this UdpClient client, byte[ ] buffer, int offset = 0) 
{ 
    int received; 
    var socket = client.Client; 
    if (socket.AddressFamily == AddressFamily.InterNetwork) 
    { 
    received = socket.ReceiveFrom(buffer, offset, MaxUdpSize, SocketFlags.None, ref anyV4Endpoint); 
    } 
    else 
    { 
    received = socket.ReceiveFrom(buffer, offset, MaxUdpSize, SocketFlags.None, ref anyV6Endpoint); 
    } 
    return received; 
} 

Это почти именно то, что делает UdpClient.Receive ... за исключением того, что я управлять буфером.Я обнаружил, что UdpClient использует один буфер для чтения, и я просто использую простой ol 'Socket на стороне отправки.