epic/OpenRA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace server Select loop with individual client threads.

Browse Source 
This guarantees that any unexpected blocking calls due to network
issues cannot stall the main server thread.
This commit is contained in:
Paul Chote
2021-05-25 22:04:47 +01:00
committed by reaperrr
parent 6535411744
commit 2a26ddc622
4 changed files with 292 additions and 257 deletions
Download Patch File Download Diff File Expand all files Collapse all files

130
OpenRA.Game/Server/Connection.cs
View File

@@ -10,21 +10,22 @@
#endregion
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Threading;
namespace OpenRA.Server
{
public class Connection
public class Connection : IDisposable
{
public const int MaxOrderLength = 131072;
public readonly Socket Socket;
public readonly List<byte> Data = new List<byte>();
public readonly int PlayerIndex;
public readonly string AuthToken;
public readonly EndPoint EndPoint;
public long TimeSinceLastResponse => Game.RunTime - lastReceivedTime;
public int MostRecentFrame { get; private set; }
@@ -32,72 +33,61 @@ namespace OpenRA.Server
public bool TimeoutMessageShown;
public bool Validated;
ReceiveState state = ReceiveState.Header;
int expectLength = 8;
int frame = 0;
long lastReceivedTime = 0;
public Connection(Socket socket, int playerIndex, string authToken)
readonly BlockingCollection<byte[]> sendQueue = new BlockingCollection<byte[]>();
public Connection(Socket socket, int playerIndex, string authToken, Action<Connection, int, byte[]> onPacket, Action<Connection> onDisconnect)
{
Socket = socket;
PlayerIndex = playerIndex;
AuthToken = authToken;
EndPoint = socket.RemoteEndPoint;
new Thread(SendReceiveLoop)
{
Name = $"Client communication ({EndPoint}",
IsBackground = true
}.Start((socket, onPacket, onDisconnect));
}
public byte[] PopBytes(int n)
void SendReceiveLoop(object s)
{
var result = Data.GetRange(0, n);
Data.RemoveRange(0, n);
return result.ToArray();
}
var (socket, onPacket, onDisconnect) = (ValueTuple<Socket, Action<Connection, int, byte[]>, Action<Connection>>)s;
socket.Blocking = false;
socket.NoDelay = true;
bool ReadDataInner(Server server)
{
var rx = new byte[1024];
var len = 0;
var receiveBuffer = new byte[1024];
var readBuffer = new List<byte>();
var state = ReceiveState.Header;
var expectLength = 8;
var frame = 0;
while (true)
{
try
{
// Poll the socket first to see if there's anything there.
// This avoids the exception with SocketErrorCode == `SocketError.WouldBlock` thrown
// from `socket.Receive(rx)`.
if (!Socket.Poll(0, SelectMode.SelectRead)) break;
if ((len = Socket.Receive(rx)) > 0)
Data.AddRange(rx.Take(len));
else
while (true)
{
if (len == 0)
server.DropClient(this);
break;
}
}
catch (SocketException e)
// Wait up to 100ms for data to arrive before checking for data to send
if (socket.Poll(100000, SelectMode.SelectRead))
{
// This should no longer be needed with the socket.Poll call above.
if (e.SocketErrorCode == SocketError.WouldBlock) break;
server.DropClient(this);
Log.Write("server", "Dropping client {0} because reading the data failed: {1}", PlayerIndex, e);
return false;
}
var read = socket.Receive(receiveBuffer);
if (read == 0)
{
// Empty packet signals that the client has been dropped
return;
}
if (read > 0)
{
readBuffer.AddRange(receiveBuffer.Take(read));
lastReceivedTime = Game.RunTime;
TimeoutMessageShown = false;
return true;
}
public void ReadData(Server server)
while (readBuffer.Count >= expectLength)
{
if (ReadDataInner(server))
{
while (Data.Count >= expectLength)
{
var bytes = PopBytes(expectLength);
var bytes = readBuffer.GetRange(0, expectLength).ToArray();
readBuffer.RemoveRange(0, expectLength);
switch (state)
{
case ReceiveState.Header:
@@ -108,8 +98,7 @@ namespace OpenRA.Server
if (expectLength < 0 || expectLength > MaxOrderLength)
{
server.DropClient(this);
Log.Write("server", "Dropping client {0} for excessive order length = {1}", PlayerIndex, expectLength);
Log.Write("server", $"Closing socket connection to {EndPoint} because of excessive order length: {expectLength}");
return;
}
@@ -121,7 +110,7 @@ namespace OpenRA.Server
if (MostRecentFrame < frame)
MostRecentFrame = frame;
server.DispatchOrders(this, frame, bytes);
onPacket(this, frame, bytes);
expectLength = 8;
state = ReceiveState.Header;
@@ -130,9 +119,13 @@ namespace OpenRA.Server
}
}
}
}
public void SendData(byte[] data)
// Client has been dropped by the server
if (sendQueue.IsCompleted)
return;
// Send all data immediately, we will block again on read
while (sendQueue.TryTake(out var data, 0))
{
var start = 0;
var length = data.Length;
@@ -140,13 +133,13 @@ namespace OpenRA.Server
// Non-blocking sends are free to send only part of the data
while (start < length)
{
var sent = Socket.Send(data, start, length - start, SocketFlags.None, out var error);
var sent = socket.Send(data, start, length - start, SocketFlags.None, out var error);
if (error == SocketError.WouldBlock)
{
Log.Write("server", "Non-blocking send of {0} bytes failed. Falling back to blocking send.", length - start);
Socket.Blocking = true;
sent = Socket.Send(data, start, length - start, SocketFlags.None);
Socket.Blocking = false;
socket.Blocking = true;
sent = socket.Send(data, start, length - start, SocketFlags.None);
socket.Blocking = false;
}
else if (error != SocketError.Success)
throw new SocketException((int)error);
@@ -154,8 +147,29 @@ namespace OpenRA.Server
start += sent;
}
}
}
}
catch (SocketException e)
{
Log.Write("server", $"Closing socket connection to {EndPoint} because of socket error: {e}");
}
finally
{
onDisconnect(this);
socket.Dispose();
}
}
public EndPoint EndPoint => Socket.RemoteEndPoint;
public void SendData(byte[] data)
{
sendQueue.Add(data);
}
public void Dispose()
{
// Tell the sendReceiveThread that the socket should be closed
sendQueue.CompleteAdding();
}
}
public enum ReceiveState { Header, Data }

223
OpenRA.Game/Server/Server.cs
View File

@@ -10,6 +10,7 @@
#endregion
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Globalization;
using System.IO;
@@ -48,12 +49,8 @@ namespace OpenRA.Server
public readonly MersenneTwister Random = new MersenneTwister();
public readonly ServerType Type;
// Valid player connections
public List<Connection> Conns = new List<Connection>();
// Pre-verified player connections
public List<Connection> PreConns = new List<Connection>();
public Session LobbyInfo;
public ServerSettings Settings;
public ModData ModData;
@@ -71,8 +68,7 @@ namespace OpenRA.Server
protected volatile ServerState internalState = ServerState.WaitingPlayers;
volatile ActionQueue delayedActions = new ActionQueue();
int waitingForAuthenticationCallback = 0;
readonly BlockingCollection<IServerEvent> events = new BlockingCollection<IServerEvent>();
ReplayRecorder recorder;
GameInformation gameInfo;
@@ -164,7 +160,6 @@ namespace OpenRA.Server
Log.AddChannel("server", "server.log", true);
SocketException lastException = null;
var checkReadServer = new List<Socket>();
foreach (var endpoint in endpoints)
{
var listener = new TcpListener(endpoint);
@@ -184,7 +179,32 @@ namespace OpenRA.Server
listener.Start();
listeners.Add(listener);
checkReadServer.Add(listener.Server);
new Thread(() =>
{
while (true)
{
if (State != ServerState.WaitingPlayers)
{
listener.Stop();
return;
}
// Use a 1s timeout so we can stop listening once the game starts
if (listener.Server.Poll(1000000, SelectMode.SelectRead))
{
try
{
events.Add(new ClientConnectEvent(listener.AcceptSocket()));
}
catch (Exception)
{
// Ignore the exception that may be generated if the connection
// drops while we are trying to connect
}
}
}
}) { Name = $"Connection listener ({listener.LocalEndpoint})", IsBackground = true }.Start();
}
catch (SocketException ex)
{
@@ -257,42 +277,10 @@ namespace OpenRA.Server
while (true)
{
var checkRead = new List<Socket>();
if (State == ServerState.WaitingPlayers)
checkRead.AddRange(checkReadServer);
checkRead.AddRange(Conns.Select(c => c.Socket));
checkRead.AddRange(PreConns.Select(c => c.Socket));
// Block for at most 1 second in order to guarantee a minimum tick rate for ServerTraits
// Decrease this to 100ms to improve responsiveness if we are waiting for an authentication query
var localTimeout = waitingForAuthenticationCallback > 0 ? 100000 : 1000000;
if (checkRead.Count > 0)
Socket.Select(checkRead, null, null, localTimeout);
if (State != ServerState.ShuttingDown)
{
foreach (var s in checkRead)
{
var serverIndex = checkReadServer.IndexOf(s);
if (serverIndex >= 0)
{
AcceptConnection(listeners[serverIndex]);
continue;
}
var preConn = PreConns.SingleOrDefault(c => c.Socket == s);
if (preConn != null)
{
preConn.ReadData(this);
continue;
}
var conn = Conns.SingleOrDefault(c => c.Socket == s);
conn?.ReadData(this);
}
delayedActions.PerformActions(0);
if (events.TryTake(out var e, 1000))
e.Invoke(this);
// PERF: Dedicated servers need to drain the action queue to remove references blocking the GC from cleaning up disposed objects.
if (Type == ServerType.Dedicated)
@@ -314,14 +302,7 @@ namespace OpenRA.Server
foreach (var t in serverTraits.WithInterface<INotifyServerShutdown>())
t.ServerShutdown(this);
PreConns.Clear();
Conns.Clear();
foreach (var listener in listeners)
{
try { listener.Stop(); }
catch { }
}
})
{ IsBackground = true }.Start();
}
@@ -332,44 +313,34 @@ namespace OpenRA.Server
return nextPlayerIndex++;
}
void AcceptConnection(TcpListener listener)
void OnClientPacket(Connection conn, int frame, byte[] data)
{
Socket newSocket;
try
{
if (!listener.Server.IsBound)
return;
newSocket = listener.AcceptSocket();
}
catch (Exception e)
{
/* TODO: Could have an exception here when listener 'goes away' when calling AcceptConnection! */
/* Alternative would be to use locking but the listener doesn't go away without a reason. */
Log.Write("server", "Accepting the connection failed.", e);
return;
events.Add(new ClientPacketEvent(conn, frame, data));
}
void OnClientDisconnect(Connection conn)
{
events.Add(new ClientDisconnectEvent(conn));
}
void AcceptConnection(Socket socket)
{
if (State != ServerState.WaitingPlayers)
return;
// Validate player identity by asking them to sign a random blob of data
// which we can then verify against the player public key database
var token = Convert.ToBase64String(OpenRA.Exts.MakeArray(256, _ => (byte)Random.Next()));
var newConn = new Connection(newSocket, ChooseFreePlayerIndex(), token);
var newConn = new Connection(socket, ChooseFreePlayerIndex(), token, OnClientPacket, OnClientDisconnect);
try
{
newConn.Socket.Blocking = false;
newConn.Socket.NoDelay = true;
// Send handshake and client index.
var ms = new MemoryStream(8);
ms.WriteArray(BitConverter.GetBytes(ProtocolVersion.Handshake));
ms.WriteArray(BitConverter.GetBytes(newConn.PlayerIndex));
newConn.SendData(ms.ToArray());
PreConns.Add(newConn);
// Dispatch a handshake order
var request = new HandshakeRequest
{
@@ -387,9 +358,10 @@ namespace OpenRA.Server
}
catch (Exception e)
{
DropClient(newConn);
Log.Write("server", "Dropping client {0} because handshake failed: {1}", newConn.PlayerIndex.ToString(CultureInfo.InvariantCulture), e);
Log.Write("server", $"Handshake for client {newConn.EndPoint} failed: {e}");
}
Conns.Add(newConn);
}
void ValidateClient(Connection newConn, string data)
@@ -491,8 +463,6 @@ namespace OpenRA.Server
client.Color = Color.White;
// Promote connection to a valid client
PreConns.Remove(newConn);
Conns.Add(newConn);
LobbyInfo.Clients.Add(client);
newConn.Validated = true;
@@ -546,8 +516,6 @@ namespace OpenRA.Server
}
else if (!string.IsNullOrEmpty(handshake.Fingerprint) && !string.IsNullOrEmpty(handshake.AuthSignature))
{
waitingForAuthenticationCallback++;
Task.Run(async () =>
{
var httpClient = HttpClientFactory.Create();
@@ -593,7 +561,7 @@ namespace OpenRA.Server
Log.Write("server", ex.ToString());
}
delayedActions.Add(() =>
events.Add(new CallbackEvent(() =>
{
var notAuthenticated = Type == ServerType.Dedicated && profile == null && (Settings.RequireAuthentication || Settings.ProfileIDWhitelist.Any());
var blacklisted = Type == ServerType.Dedicated && profile != null && Settings.ProfileIDBlacklist.Contains(profile.ProfileID);
@@ -618,9 +586,7 @@ namespace OpenRA.Server
}
else
completeConnection();
waitingForAuthenticationCallback--;
}, 0);
}));
});
}
else
@@ -771,9 +737,10 @@ namespace OpenRA.Server
public void DispatchOrdersToClients(Connection conn, int frame, byte[] data)
{
var from = conn != null ? conn.PlayerIndex : 0;
var from = conn?.PlayerIndex ?? 0;
var frameData = CreateFrame(from, frame, data);
foreach (var c in Conns.Except(conn).ToList())
foreach (var c in Conns.ToList())
if (c != conn && c.Validated)
DispatchFrameToClient(c, from, frameData);
if (recorder != null)
@@ -1022,14 +989,15 @@ namespace OpenRA.Server
public Session.Client GetClient(Connection conn)
{
if (conn == null)
return null;
return LobbyInfo.ClientWithIndex(conn.PlayerIndex);
}
public void DropClient(Connection toDrop)
{
lock (LobbyInfo)
{
if (!PreConns.Remove(toDrop))
{
Conns.Remove(toDrop);
@@ -1074,23 +1042,18 @@ namespace OpenRA.Server
DispatchOrders(toDrop, toDrop.MostRecentFrame, new[] { (byte)OrderType.Disconnect });
// All clients have left: clean up
if (!Conns.Any())
if (!Conns.Any(c => c.Validated))
foreach (var t in serverTraits.WithInterface<INotifyServerEmpty>())
t.ServerEmpty(this);
if (Conns.Any() || Type == ServerType.Dedicated)
if (Conns.Any(c => c.Validated) || Type == ServerType.Dedicated)
SyncLobbyClients();
if (Type != ServerType.Dedicated && dropClient.IsAdmin)
Shutdown();
}
}
try
{
toDrop.Socket.Disconnect(false);
}
catch { }
toDrop.Dispose();
}
public void SyncLobbyInfo()
@@ -1171,17 +1134,10 @@ namespace OpenRA.Server
{
lock (LobbyInfo)
{
foreach (var listener in listeners)
listener.Stop();
Console.WriteLine("[{0}] Game started", DateTime.Now.ToString(Settings.TimestampFormat));
// Drop any unvalidated clients
foreach (var c in PreConns.ToArray())
DropClient(c);
// Drop any players who are not ready
foreach (var c in Conns.Where(c => GetClient(c).IsInvalid).ToArray())
foreach (var c in Conns.Where(c => !c.Validated || GetClient(c).IsInvalid).ToArray())
{
SendOrderTo(c, "ServerError", "You have been kicked from the server!");
DropClient(c);
@@ -1249,6 +1205,7 @@ namespace OpenRA.Server
GameSave.ParseOrders(LobbyInfo, (frame, client, data) =>
{
foreach (var c in Conns)
if (c.Validated)
DispatchOrdersToClient(c, client, frame, data);
});
}
@@ -1271,5 +1228,69 @@ namespace OpenRA.Server
return new ConnectionTarget(endpoints);
}
interface IServerEvent { void Invoke(Server server); }
class ClientConnectEvent : IServerEvent
{
readonly Socket socket;
public ClientConnectEvent(Socket socket)
{
this.socket = socket;
}
void IServerEvent.Invoke(Server server)
{
server.AcceptConnection(socket);
}
}
class ClientDisconnectEvent : IServerEvent
{
readonly Connection connection;
public ClientDisconnectEvent(Connection connection)
{
this.connection = connection;
}
void IServerEvent.Invoke(Server server)
{
server.DropClient(connection);
}
}
class ClientPacketEvent : IServerEvent
{
readonly Connection connection;
readonly int frame;
readonly byte[] data;
public ClientPacketEvent(Connection connection, int frame, byte[] data)
{
this.connection = connection;
this.frame = frame;
this.data = data;
}
void IServerEvent.Invoke(Server server)
{
server.DispatchOrders(connection, frame, data);
}
}
class CallbackEvent : IServerEvent
{
readonly Action action;
public CallbackEvent(Action action)
{
this.action = action;
}
void IServerEvent.Invoke(Server server)
{
action();
}
}
}
}

6
OpenRA.Mods.Common/ServerTraits/LobbyCommands.cs
View File

@@ -645,7 +645,7 @@ namespace OpenRA.Mods.Common.Server
Exts.TryParseIntegerInvariant(split[0], out var kickClientID);
var kickConn = server.Conns.SingleOrDefault(c => server.GetClient(c) != null && server.GetClient(c).Index == kickClientID);
var kickConn = server.Conns.SingleOrDefault(c => server.GetClient(c)?.Index == kickClientID);
if (kickConn == null)
{
server.SendOrderTo(conn, "Message", "No-one in that slot.");
@@ -690,7 +690,7 @@ namespace OpenRA.Mods.Common.Server
}
Exts.TryParseIntegerInvariant(s, out var newAdminId);
var newAdminConn = server.Conns.SingleOrDefault(c => server.GetClient(c) != null && server.GetClient(c).Index == newAdminId);
var newAdminConn = server.Conns.SingleOrDefault(c => server.GetClient(c)?.Index == newAdminId);
if (newAdminConn == null)
{
@@ -727,7 +727,7 @@ namespace OpenRA.Mods.Common.Server
}
Exts.TryParseIntegerInvariant(s, out var targetId);
var targetConn = server.Conns.SingleOrDefault(c => server.GetClient(c) != null && server.GetClient(c).Index == targetId);
var targetConn = server.Conns.SingleOrDefault(c => server.GetClient(c)?.Index == targetId);
if (targetConn == null)
{

10
OpenRA.Mods.Common/ServerTraits/PlayerPinger.cs
View File

@@ -41,13 +41,16 @@ namespace OpenRA.Mods.Common.Server
nonBotClientCount = server.LobbyInfo.NonBotClients.Count();
if (nonBotClientCount < 2 && server.Type != ServerType.Dedicated)
{
foreach (var c in server.Conns.ToList())
if (c.Validated)
server.SendOrderTo(c, "Ping", Game.RunTime.ToString());
}
else
{
foreach (var c in server.Conns.ToList())
{
if (c == null || c.Socket == null)
if (!c.Validated)
continue;
var client = server.GetClient(c);
@@ -79,14 +82,11 @@ namespace OpenRA.Mods.Common.Server
lastConnReport = Game.RunTime;
var timeouts = server.Conns
.Where(c => c.TimeSinceLastResponse > ConnReportInterval && c.TimeSinceLastResponse < ConnTimeout)
.Where(c => c.Validated && c.TimeSinceLastResponse > ConnReportInterval && c.TimeSinceLastResponse < ConnTimeout)
.OrderBy(c => c.TimeSinceLastResponse);
foreach (var c in timeouts)
{
if (c == null || c.Socket == null)
continue;
var client = server.GetClient(c);
if (client != null)
server.SendMessage($"{client.Name} will be dropped in {(ConnTimeout - c.TimeSinceLastResponse) / 1000} seconds.");