Files
AIC-Project/Server/Gateway/ServerLoop.cs
T

619 lines
25 KiB
C#
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
using System;
using System.Collections.Generic;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using ChannelsNS = System.Threading.Channels;
using XWorld.Framework;
using XWorld.Framework.Protocol;
using XWorld.Server.Host;
namespace XWorld.Server.Gateway
{
// 单线程 actor:独占 RoomHost 与 SessionManager;命令入队、周期性排空+推进。
public sealed class ServerLoop
{
private readonly RoomHost _host;
private readonly SessionManager _sessions = new SessionManager();
private readonly ILogger _logger;
private readonly long _reconnectWindowTicks;
private readonly string _gamesRoot;
private readonly Matchmaker _matchmaker;
private readonly ChannelsNS.Channel<ServerCommand> _cmds =
ChannelsNS.Channel.CreateUnbounded<ServerCommand>();
public const long DefaultReconnectWindowTicks = 1800;
private const int MaxWorldChatTextLength = 200;
private readonly Dictionary<string, List<int>> _roomPlayers = new Dictionary<string, List<int>>();
private readonly Dictionary<string, ActiveRoomInfo> _activeRooms = new Dictionary<string, ActiveRoomInfo>();
private readonly Dictionary<int, LobbyPlayerMsg> _lobbyPlayers = new Dictionary<int, LobbyPlayerMsg>();
private readonly Dictionary<string, int> _playerCountCache = new Dictionary<string, int>(); // "gameId@version" -> playerCount
private readonly HashSet<int> _queuedPids = new HashSet<int>(); // pids currently in matchmaker queue
private readonly Dictionary<int, int> _queuedRequestIds = new Dictionary<int, int>();
private readonly Dictionary<int, int> _activeRoomRequestIds = new Dictionary<int, int>();
private readonly Random _random = new Random(1);
private long _tick;
private int _roomSeq;
private sealed class ActiveRoomInfo
{
public string RoomId;
public string GameId;
public int Version;
public List<PlayerInfo> Players;
public Dictionary<int, int> RequestIds = new Dictionary<int, int>();
public bool HasLastGameMessage;
public NetMessage LastGameMessage;
}
public ServerLoop(string gamesRoot, ILogger logger, long reconnectWindowTicks = DefaultReconnectWindowTicks,
long matchTimeoutTicks = 150, string publicKeyPem = null)
{
_logger = logger;
_reconnectWindowTicks = reconnectWindowTicks;
_gamesRoot = gamesRoot;
_host = new RoomHost(new GameModuleRegistry(new GameModuleLoader(publicKeyPem), gamesRoot), logger);
_matchmaker = new Matchmaker(matchTimeoutTicks);
}
// 线程安全:网络线程调用,仅入队
public void Submit(ServerCommand cmd) => _cmds.Writer.TryWrite(cmd);
// 仅 actor 线程调用(_sessions 无锁);测试在单线程下使用
public bool HasSession(int pid) => _sessions.Get(pid) != null;
// 测试直调;真实循环周期性调它。
// 两步排空:先处理所有 Connect/Disconnect/Framework(含 MatchRequest 入队),
// 再 Poll 匹配器建房,再处理 Game 帧(此时 RoomId 已就位)。
public void DrainAndTick(float dt)
{
// 第一遍:收集所有命令,立刻处理非游戏帧(connect/disconnect/framework
var deferred = new List<FrameCommand>(); // 本批游戏帧,延后到建房后处理
while (_cmds.Reader.TryRead(out var cmd))
{
if (cmd is FrameCommand fc && fc.Frame.Channel == Channel.Game)
deferred.Add(fc);
else
Process(cmd);
}
// Poll 匹配器 → 对本 tick 内已凑齐或超时的队列建房
var matches = _matchmaker.Poll(_tick);
for (int i = 0; i < matches.Count; i++)
{
_logger.Info($"match formed {matches[i].GameId}@{matches[i].Version} seats={matches[i].Seats.Count} tick={_tick}");
CreateRoomForSeats(matches[i]);
}
// 第二遍:处理本批游戏帧(此时 RoomId 已就位)
for (int i = 0; i < deferred.Count; i++)
{
var fc = deferred[i];
Route(fc.PlayerId, fc.Frame);
}
_tick++;
var ended = _host.TickAll(dt);
for (int i = 0; i < ended.Count; i++) OnRoomEnded(ended[i].RoomId, ended[i].Result);
var expired = _sessions.SweepExpired(_tick, _reconnectWindowTicks);
for (int i = 0; i < expired.Count; i++)
_logger.Info($"session {expired[i].PlayerId} 重连窗口过期,清除");
}
public async Task RunAsync(int tickIntervalMs, CancellationToken ct)
=> await RunAsync(tickIntervalMs, tickIntervalMs / 1000f, ct);
public async Task RunAsync(int tickIntervalMs, float logicalDt, CancellationToken ct)
{
while (!ct.IsCancellationRequested)
{
DrainAndTick(logicalDt);
try { await Task.Delay(tickIntervalMs, ct).ConfigureAwait(false); }
catch (OperationCanceledException) { break; }
}
}
private void Process(ServerCommand cmd)
{
switch (cmd)
{
case ConnectCommand cc:
_sessions.OnConnect(cc.PlayerId, cc.Connection);
_logger.Info($"session connect pid={cc.PlayerId}");
SendRoomResumeIfNeeded(cc.PlayerId);
break;
case DisconnectCommand dc:
_sessions.OnDisconnect(dc.PlayerId, dc.Connection, _tick);
_logger.Info($"session disconnect pid={dc.PlayerId}");
CancelQueuedMatch(dc.PlayerId);
if (_lobbyPlayers.Remove(dc.PlayerId)) BroadcastLobbyState();
break;
case FrameCommand fc: Route(fc.PlayerId, fc.Frame); break;
}
}
private void Route(int pid, Frame frame)
{
if (frame.Channel == Channel.Framework) HandleFramework(pid, frame);
else
{
var s = _sessions.Get(pid);
if (s?.RoomId != null)
_host.DeliverTo(s.RoomId, pid, new NetMessage(frame.Opcode, frame.Payload));
}
}
private void HandleFramework(int pid, Frame frame)
{
FrameworkOpcode op = (FrameworkOpcode)frame.Opcode;
if (op != FrameworkOpcode.LobbyMove)
{
_logger.Info($"framework pid={pid} op={op} bytes={frame.Payload?.Length ?? 0}");
}
switch (op)
{
case FrameworkOpcode.Heartbeat:
SendFramework(pid, FrameworkOpcode.Heartbeat, frame.Payload); // 原样回显
break;
case FrameworkOpcode.MatchRequest:
var req = MatchRequestMsg.Decode(frame.Payload);
EnqueueMatch(pid, req.GameId, req.Version, req.RequestId);
break;
case FrameworkOpcode.GameListRequest:
SendFramework(pid, FrameworkOpcode.GameListResponse, BuildGameList().Encode());
break;
case FrameworkOpcode.RandomMatchRequest:
EnqueueRandomMatch(pid, MatchControlMsg.Decode(frame.Payload).RequestId);
break;
case FrameworkOpcode.MatchCancel:
CancelMatchState(pid, MatchControlMsg.Decode(frame.Payload).RequestId);
break;
case FrameworkOpcode.LobbyJoin:
HandleLobbyJoin(pid, frame.Payload);
break;
case FrameworkOpcode.LobbyMove:
HandleLobbyMove(pid, frame.Payload);
break;
case FrameworkOpcode.LobbyLeave:
CancelQueuedMatch(pid);
if (_lobbyPlayers.Remove(pid)) BroadcastLobbyState();
break;
case FrameworkOpcode.WorldChatSend:
HandleWorldChatSend(pid, frame.Payload);
break;
}
}
private void HandleLobbyJoin(int pid, byte[] payload)
{
var req = LobbyJoinMsg.Decode(payload);
string name = string.IsNullOrWhiteSpace(req.Name) ? $"P{pid}" : req.Name;
int figure = req.Figure <= 0 ? 1 : req.Figure;
if (!_lobbyPlayers.TryGetValue(pid, out var player))
{
float lane = (_lobbyPlayers.Count % 6) - 2.5f;
player = new LobbyPlayerMsg
{
PlayerId = pid,
Name = name,
Figure = figure,
X = lane * 1.5f,
Y = 0f,
Z = 0f,
DirX = 0f,
DirZ = 1f,
Moving = false
};
_lobbyPlayers[pid] = player;
}
else
{
player.Name = name;
player.Figure = figure;
}
_logger.Info($"lobby join pid={pid} name={name} figure={figure} players={_lobbyPlayers.Count} pids={string.Join(",", _lobbyPlayers.Keys)}");
BroadcastLobbyState();
}
private void HandleLobbyMove(int pid, byte[] payload)
{
if (!_lobbyPlayers.TryGetValue(pid, out var player))
{
_logger.Warn($"lobby move ignored pid={pid} reason=not_joined players={_lobbyPlayers.Count} pids={string.Join(",", _lobbyPlayers.Keys)}");
return;
}
var move = LobbyMoveMsg.Decode(payload);
player.X = move.X;
player.Y = move.Y;
player.Z = move.Z;
player.DirX = move.DirX;
player.DirZ = move.DirZ;
player.Moving = move.Moving;
move.PlayerId = pid;
byte[] encoded = move.Encode();
foreach (int targetPid in _lobbyPlayers.Keys)
{
SendFramework(targetPid, FrameworkOpcode.LobbyMove, encoded);
}
}
private void HandleWorldChatSend(int pid, byte[] payload)
{
if (!_sessions.IsConnected(pid))
{
_logger.Warn($"world chat ignored pid={pid} reason=not_connected");
return;
}
var req = WorldChatSendMsg.Decode(payload);
string text = (req.Text ?? string.Empty).Trim();
if (text.Length == 0)
{
_logger.Warn($"world chat ignored pid={pid} reason=empty");
return;
}
if (text.Length > MaxWorldChatTextLength)
{
text = text.Substring(0, MaxWorldChatTextLength);
}
string name = _lobbyPlayers.TryGetValue(pid, out var player) && !string.IsNullOrWhiteSpace(player.Name)
? player.Name
: $"P{pid}";
var msg = new WorldChatMessageMsg
{
PlayerId = pid,
PlayerName = name,
Text = text,
ServerTimeMs = DateTimeOffset.UtcNow.ToUnixTimeMilliseconds()
};
byte[] encoded = msg.Encode();
IReadOnlyList<int> targets = _sessions.ConnectedPlayerIds();
_logger.Info($"world chat pid={pid} name={name} length={text.Length} targets={targets.Count}");
foreach (int targetPid in targets)
{
SendFramework(targetPid, FrameworkOpcode.WorldChatMessage, encoded);
}
}
private void BroadcastLobbyState()
{
if (_lobbyPlayers.Count == 0) return;
var state = new LobbyStateMsg();
foreach (var player in _lobbyPlayers.Values)
{
state.Players.Add(player);
}
byte[] payload = state.Encode();
_logger.Info($"lobby state players={_lobbyPlayers.Count} pids={string.Join(",", _lobbyPlayers.Keys)}");
foreach (int pid in _lobbyPlayers.Keys)
{
SendFramework(pid, FrameworkOpcode.LobbyState, payload);
}
}
private void EnqueueRandomMatch(int pid, int requestId)
{
var games = BuildGameList().Games;
if (games.Count == 0)
{
SendFramework(pid, FrameworkOpcode.Error,
new ErrorMsg { Code = 1, Message = "no minigame available", RequestId = requestId }.Encode());
return;
}
var selected = games[_random.Next(games.Count)];
if (EnqueueMatch(pid, selected.GameId, selected.Version, requestId))
{
_logger.Info($"random match pid={pid} assigned {selected.GameId}@{selected.Version}");
SendFramework(pid, FrameworkOpcode.MatchAssigned,
new MatchAssignedMsg { GameId = selected.GameId, Version = selected.Version, RequestId = requestId }.Encode());
}
}
private GameListResponseMsg BuildGameList()
{
var latest = new Dictionary<string, GameInfoMsg>(StringComparer.OrdinalIgnoreCase);
if (!Directory.Exists(_gamesRoot)) return new GameListResponseMsg();
foreach (var gameDir in Directory.GetDirectories(_gamesRoot))
{
foreach (var versionDir in Directory.GetDirectories(gameDir))
{
string manifestPath = Path.Combine(versionDir, "game.json");
if (!File.Exists(manifestPath)) continue;
try
{
var manifest = GameManifest.Load(manifestPath);
if (string.Equals(manifest.GameId, "lobby", StringComparison.OrdinalIgnoreCase)) continue;
if (manifest.PlayerCount < 2 || manifest.PlayerCount > 8) continue;
if (!latest.TryGetValue(manifest.GameId, out var old) || manifest.Version > old.Version)
{
latest[manifest.GameId] = new GameInfoMsg
{
GameId = manifest.GameId,
Version = manifest.Version,
DisplayName = manifest.GameId,
MinPlayers = 2,
MaxPlayers = Math.Max(2, Math.Min(8, manifest.PlayerCount)),
};
}
}
catch (Exception ex)
{
_logger.Warn($"skip invalid game manifest {manifestPath}: {ex.Message}");
}
}
}
var msg = new GameListResponseMsg();
foreach (var game in latest.Values) msg.Games.Add(game);
msg.Games.Sort((a, b) => string.Compare(a.GameId, b.GameId, StringComparison.OrdinalIgnoreCase));
return msg;
}
private bool EnqueueMatch(int pid, string gameId, int version, int requestId = 0)
{
var s = _sessions.Get(pid);
if (s == null) return false; // 未建立会话,忽略
if (s.RoomId != null || _queuedPids.Contains(pid))
{
// 已在房间内或已在队列中:拒绝
SendFramework(pid, FrameworkOpcode.Error,
new ErrorMsg { Code = 2, Message = "already in room or queue", RequestId = requestId }.Encode());
return false;
}
// 读取游戏的 playerCount(缓存)
string cacheKey = $"{gameId}@{version}";
if (!_playerCountCache.TryGetValue(cacheKey, out int playerCount))
{
string manifestPath = Path.Combine(_gamesRoot, gameId, version.ToString(), "game.json");
try
{
var manifest = GameManifest.Load(manifestPath);
playerCount = manifest.PlayerCount;
_playerCountCache[cacheKey] = playerCount;
}
catch (Exception ex)
{
_logger.Error($"读取 game.json 失败 {gameId}@{version}: {ex.Message}");
SendFramework(pid, FrameworkOpcode.Error,
new ErrorMsg { Code = 1, Message = "game not found", RequestId = requestId }.Encode());
return false;
}
}
_queuedPids.Add(pid);
_queuedRequestIds[pid] = requestId;
_matchmaker.Enqueue(pid, gameId, version, playerCount, _tick, requestId);
_logger.Info($"enqueue match pid={pid} req={requestId} game={gameId}@{version} playerCount={playerCount} tick={_tick}");
return true;
}
private void CancelQueuedMatch(int pid, int requestId = 0)
{
if (requestId != 0 &&
_queuedRequestIds.TryGetValue(pid, out int queuedRequestId) &&
queuedRequestId != requestId)
{
return;
}
if (_queuedPids.Remove(pid))
{
_queuedRequestIds.Remove(pid);
_matchmaker.Remove(pid);
_logger.Info($"cancel queued match pid={pid} req={requestId}");
}
}
private void CancelMatchState(int pid, int requestId)
{
CancelQueuedMatch(pid, requestId);
CancelActiveRoom(pid, requestId);
}
private void CancelActiveRoom(int pid, int requestId)
{
var s = _sessions.Get(pid);
if (s?.RoomId == null) return;
if (requestId != 0 &&
_activeRoomRequestIds.TryGetValue(pid, out int activeRequestId) &&
activeRequestId != requestId)
{
return;
}
string roomId = s.RoomId;
if (_roomPlayers.TryGetValue(roomId, out var players))
{
var roomPids = new List<int>(players);
_roomPlayers.Remove(roomId);
_activeRooms.Remove(roomId);
for (int i = 0; i < roomPids.Count; i++)
{
int roomPid = roomPids[i];
var roomSession = _sessions.Get(roomPid);
if (roomSession?.RoomId == roomId) roomSession.RoomId = null;
_activeRoomRequestIds.Remove(roomPid);
}
_host.EndRoom(roomId);
}
else
{
_activeRooms.Remove(roomId);
s.RoomId = null;
_activeRoomRequestIds.Remove(pid);
}
_logger.Info($"cancel active match pid={pid} req={requestId} room={roomId}");
}
private void CreateRoomForSeats(Matchmaker.Match match)
{
string roomId = $"r{++_roomSeq}";
var seats = match.Seats;
int aiCount = 0;
for (int i = 0; i < seats.Count; i++)
{
if (seats[i].IsAi) aiCount++;
}
_logger.Info($"create room {roomId} game={match.GameId}@{match.Version} real={seats.Count - aiCount} ai={aiCount}");
// Build PlayerInfo[] for all seats
var playerInfos = new PlayerInfo[seats.Count];
var realPids = new List<int>();
for (int i = 0; i < seats.Count; i++)
{
var seat = seats[i];
playerInfos[i] = new PlayerInfo
{
PlayerId = seat.PlayerId,
Name = seat.IsAi ? "AI" : $"P{seat.PlayerId}",
IsAI = seat.IsAi
};
if (!seat.IsAi)
{
realPids.Add(seat.PlayerId);
_queuedPids.Remove(seat.PlayerId);
_queuedRequestIds.Remove(seat.PlayerId);
}
}
var allPlayers = new List<PlayerInfo>(playerInfos);
var roomInfo = new ActiveRoomInfo
{
RoomId = roomId,
GameId = match.GameId,
Version = match.Version,
Players = allPlayers
};
var sink = new RoomOutputSink(roomId, realPids, _sessions, message =>
{
roomInfo.HasLastGameMessage = true;
roomInfo.LastGameMessage = new NetMessage(message.Opcode, message.Payload ?? Array.Empty<byte>());
});
var cfg = new RoomConfig
{
GameId = match.GameId,
Version = match.Version,
Seed = (ulong)_roomSeq,
PlayerCount = seats.Count
};
_activeRooms[roomId] = roomInfo;
Room room;
try
{
room = _host.CreateRoom(roomId, match.GameId, match.Version, cfg, playerInfos, sink);
}
catch (Exception ex)
{
_logger.Error($"建房失败 {match.GameId}@{match.Version}: {ex.Message}");
_activeRooms.Remove(roomId);
// Notify real players of error
var errPayload = new ErrorMsg { Code = 1, Message = "create room failed" }.Encode();
foreach (int rpid in realPids)
SendFramework(rpid, FrameworkOpcode.Error, errPayload);
return;
}
// For each real player: set session.RoomId + send MatchFound
foreach (int rpid in realPids)
{
var s = _sessions.Get(rpid);
if (s != null) s.RoomId = roomId;
int requestId = 0;
for (int i = 0; i < seats.Count; i++)
{
if (!seats[i].IsAi && seats[i].PlayerId == rpid)
{
requestId = seats[i].RequestId;
break;
}
}
_activeRoomRequestIds[rpid] = requestId;
roomInfo.RequestIds[rpid] = requestId;
var found = new MatchFoundMsg
{
GameId = match.GameId,
RoomId = roomId,
Version = match.Version,
SelfPlayerId = rpid,
RequestId = requestId
};
foreach (var pi in allPlayers) found.Players.Add(pi);
SendFramework(rpid, FrameworkOpcode.MatchFound, found.Encode());
}
_roomPlayers[roomId] = realPids;
if (!room.IsActive) OnRoomEnded(roomId, room.TrustedEndResult); // 极端:Start 即结束
}
private void OnRoomEnded(string roomId, RoomEndResult result)
{
if (!_roomPlayers.TryGetValue(roomId, out var players)) return;
_roomPlayers.Remove(roomId);
_activeRooms.Remove(roomId);
var msg = new RoomEndMsg
{
WinnerPlayerId = result?.WinnerPlayerId ?? 0,
ResultBlob = result?.ResultBlob ?? Array.Empty<byte>()
};
byte[] payload = msg.Encode();
for (int i = 0; i < players.Count; i++)
{
int pid = players[i];
SendFramework(pid, FrameworkOpcode.RoomEnd, payload);
var s = _sessions.Get(pid);
if (s != null) s.RoomId = null;
_activeRoomRequestIds.Remove(pid);
}
}
private void SendFramework(int pid, FrameworkOpcode op, byte[] payload)
{
_sessions.GetConnection(pid)?.Send(
FrameCodec.Encode(new Frame(Channel.Framework, (ushort)op, payload)));
}
private void SendGame(int pid, NetMessage message)
{
_sessions.GetConnection(pid)?.Send(
FrameCodec.Encode(new Frame(Channel.Game, message.Opcode, message.Payload)));
}
private void SendRoomResumeIfNeeded(int pid)
{
var session = _sessions.Get(pid);
if (session?.RoomId == null) return;
if (!_activeRooms.TryGetValue(session.RoomId, out var room))
{
session.RoomId = null;
_activeRoomRequestIds.Remove(pid);
return;
}
var found = new MatchFoundMsg
{
GameId = room.GameId,
RoomId = room.RoomId,
Version = room.Version,
SelfPlayerId = pid,
RequestId = room.RequestIds.TryGetValue(pid, out int requestId) ? requestId : 0
};
for (int i = 0; i < room.Players.Count; i++) found.Players.Add(room.Players[i]);
SendFramework(pid, FrameworkOpcode.MatchFound, found.Encode());
if (room.HasLastGameMessage) SendGame(pid, room.LastGameMessage);
_logger.Info($"resume room pid={pid} room={room.RoomId} game={room.GameId}@{room.Version}");
}
}
}