Ожидание, когда несколько TcpClients будут иметь доступ к данным - WaitHandle или Thread.Sleep?
Я пишу серверное приложение, которое будет получать данные из нескольких соединений TCP. Мы хотели бы иметь возможность масштабировать до ~200 соединений. Первый алгоритм, который я написал для этого, выглядит следующим образом:
while (keepListening)
{
foreach (TcpClient client in clientList)
{
if (!client.Connected)
{
client.Close();
deleteList.Add(client);
continue;
}
int dataAvail = client.Available;
if (dataAvail > 0)
{
NetworkStream netstr = client.GetStream();
byte[] arry = new byte[dataAvail];
netstr.Read(arry, 0, dataAvail);
MemoryStream ms = new MemoryStream(arry);
try
{
CommData data = dataDeserializer.Deserialize(ms) as CommData;
beaconTable.BeaconReceived(data);
}
catch
{ }
}
}
foreach (TcpClient clientToDelete in deleteList)
clientList.Remove(clientToDelete);
deleteList.Clear();
while (connectionListener.Pending())
clientList.Add(connectionListener.AcceptTcpClient());
Thread.Sleep(20);
}
Это прекрасно работает, хотя я обнаружил, что мне нужно добавить Thread.Sleep, чтобы замедлить цикл, в противном случае он занимает все ядро, независимо от того, сколько или мало соединений. Мне посоветовали, что Thread.Sleep обычно считается плохим, поэтому я искал альтернативы. В вопросе, аналогичном этому, мне было рекомендовано использовать BeginRead и BeginAccept с использованием WaitHandles, поэтому я написал алгоритм, чтобы сделать то же самое с помощью этого, и придумал это:
while (keepListening)
{
int waitResult = WaitHandle.WaitAny(waitList.Select(t => t.AsyncHandle.AsyncWaitHandle).ToArray(), connectionTimeout);
if (waitResult == WaitHandle.WaitTimeout)
continue;
WaitObject waitObject = waitList[waitResult];
Type waitType = waitObject.WaitingObject.GetType();
if (waitType == typeof(TcpListener))
{
TcpClient newClient = (waitObject.WaitingObject as TcpListener).EndAcceptTcpClient(waitObject.AsyncHandle);
waitList.Remove(waitObject);
byte[] newBuffer = new byte[bufferSize];
waitList.Add(new WaitObject(newClient.GetStream().BeginRead(newBuffer, 0, bufferSize, null, null), newClient, newBuffer));
if (waitList.Count < 64)
waitList.Add(new WaitObject(connectionListener.BeginAcceptTcpClient(null, null), connectionListener, null));
else
{
connectionListener.Stop();
listening = false;
}
}
else if (waitType == typeof(TcpClient))
{
TcpClient currentClient = waitObject.WaitingObject as TcpClient;
int bytesRead = currentClient.GetStream().EndRead(waitObject.AsyncHandle);
if (bytesRead > 0)
{
MemoryStream ms = new MemoryStream(waitObject.DataBuffer, 0, bytesRead);
try
{
CommData data = dataDeserializer.Deserialize(ms) as CommData;
beaconTable.BeaconReceived(data);
}
catch
{ }
byte[] newBuffer = new byte[bufferSize];
waitList.Add(new WaitObject(currentClient.GetStream().BeginRead(newBuffer, 0, bufferSize, null, null), currentClient, newBuffer));
}
else
{
currentClient.Close();
}
waitList.Remove(waitObject);
if (!listening && waitList.Count < 64)
{
listening = true;
connectionListener.Start();
waitList.Add(new WaitObject(connectionListener.BeginAcceptTcpClient(null, null), connectionListener, null));
}
}
else
throw new ApplicationException("An unknown type ended up in the wait list somehow: " + waitType.ToString());
}
Это также работает нормально, пока я не набрал 64 клиентов. Я написал ограничение, чтобы не принимать более 64 клиентов, потому что это максимальное количество WaitHandles, которое примет WaitAny. Я не вижу хорошего способа обойти это ограничение, поэтому я не могу поддерживать более 64 таких соединений. Алгоритм Thread.Sleep прекрасно работает с более чем 100 соединениями.
Мне также не очень нравится предварительно выделять принимающий массив произвольного размера, а не назначать его с точным размером полученных данных после получения данных. И я все равно должен дать WaitAny тайм-аут, иначе он не позволит потоку, выполняющему это, присоединиться, когда я закрою приложение, если нет соединений. И как правило, это дольше и сложнее.
Так почему Thread.Sleep - худшее решение? Есть ли какой-нибудь способ, которым я могу по крайней мере заставить версию WaitAny обрабатывать более 64 соединений? Есть ли какой-то совершенно другой способ справиться с этим, которого я не вижу?
1 ответ
Джим дал очевидное предложение использовать асинхронные обратные вызовы вместо WaitHandles. Сначала я думал, что это будет слишком сложно, но стало намного проще, когда я понял, что могу передать ссылку на вызывающий TcpListener или TcpClient в объекте состояния. С этим и несколькими изменениями для безопасности потока, он готов к работе. Он отлично тестирует более 100 соединений и не имеет проблем с чистым выходом. Я все еще хотел бы альтернативу, однако, необходимости предварительно выделить буфер данных. Вот код для тех, кто пытается что-то подобное:
public class NetworkReceiver : IDisposable
{
private IReceiver beaconTable;
private XmlSerializer dataDeserializer;
private HashSet<TcpClient> ClientTable;
private TcpListener connectionListener;
private int bufferSize = 1000;
public NetworkReceiver(IReceiver inputTable)
{
beaconTable = inputTable;
dataDeserializer = new XmlSerializer(typeof(CommData));
ClientTable = new HashSet<TcpClient>();
connectionListener = new TcpListener(IPAddress.Any, SharedData.connectionPort);
connectionListener.Start();
connectionListener.BeginAcceptTcpClient(ListenerCallback, connectionListener);
}
private void ListenerCallback(IAsyncResult callbackResult)
{
TcpListener listener = callbackResult.AsyncState as TcpListener;
TcpClient client;
try
{
client = listener.EndAcceptTcpClient(callbackResult);
lock (ClientTable)
ClientTable.Add(client);
ClientObject clientObj = new ClientObject() { AsyncClient = client, Buffer = new byte[bufferSize] };
client.GetStream().BeginRead(clientObj.Buffer, 0, bufferSize, ClientReadCallback, clientObj);
listener.BeginAcceptTcpClient(ListenerCallback, listener);
}
catch (ObjectDisposedException)
{
return;
}
}
private void ClientReadCallback(IAsyncResult callbackResult)
{
ClientObject clientObj = callbackResult.AsyncState as ClientObject;
TcpClient client = clientObj.AsyncClient;
if (!client.Connected)
return;
try
{
int bytesRead = client.GetStream().EndRead(callbackResult);
if (bytesRead > 0)
{
MemoryStream ms = new MemoryStream(clientObj.Buffer, 0, bytesRead);
try
{
CommData data;
lock (dataDeserializer)
data = dataDeserializer.Deserialize(ms) as CommData;
lock (beaconTable)
beaconTable.BeaconReceived(data);
}
catch
{ }
client.GetStream().BeginRead(clientObj.Buffer, 0, bufferSize, ClientReadCallback, clientObj);
}
else
{
client.Close();
lock (ClientTable)
ClientTable.Remove(client);
}
}
catch (Exception ex)
{
if (ex.GetType() == typeof(ObjectDisposedException) || ex.GetType() == typeof(InvalidOperationException))
return;
else
throw;
}
}
class ClientObject
{
public TcpClient AsyncClient;
public byte[] Buffer;
}
public void Dispose()
{
connectionListener.Stop();
foreach (TcpClient client in ClientTable)
client.Close();
}
}