Process pvs-leave in parallel (#4787)

* Process pvs-leave in parallel

* oops

* Try fix eternal tests

Robust.Server/GameStates/PvsSystem.Ack.cs

@@ -50,23 +50,22 @@ internal sealed partial class PvsSystem
         if (!_async)
         {
             using var _= Histogram.WithLabels("Process Acks").NewTimer();
-            _parallelManager.ProcessNow(_ackJob, _toAck.Count);
+            _parallelManager.ProcessNow(_ackJob, _ackJob.Count);
             return null;
         }
 
-        return _parallelManager.Process(_ackJob, _toAck.Count);
+        return _parallelManager.Process(_ackJob, _ackJob.Count);
     }
 
     private record struct PvsAckJob : IParallelRobustJob
     {
         public int BatchSize => 2;
-
-        public PvsSystem System;
-        public List<ICommonSession> Sessions;
+        public PvsSystem Pvs;
+        public int Count => Pvs._toAck.Count;
 
         public void Execute(int index)
         {
-            System.ProcessQueuedAck(Sessions[index]);
+            Pvs.ProcessQueuedAck(Pvs._toAck[index]);
         }
     }
 

Robust.Server/GameStates/PvsSystem.Leave.cs

@@ -0,0 +1,94 @@
+using System;
+using System.Runtime.InteropServices;
+using System.Runtime.InteropServices.JavaScript;
+using System.Threading;
+using Prometheus;
+using Robust.Shared.Enums;
+using Robust.Shared.Network.Messages;
+using Robust.Shared.Player;
+using Robust.Shared.Threading;
+using Robust.Shared.Utility;
+
+namespace Robust.Server.GameStates;
+
+// Partial class for handling entities leaving a player's pvs range.
+internal sealed partial class PvsSystem
+{
+    private WaitHandle? _leaveTask;
+
+    private void ProcessLeavePvs(ICommonSession[] sessions)
+    {
+        if (!CullingEnabled || sessions.Length == 0)
+            return;
+
+        DebugTools.AssertNull(_leaveTask);
+        _leaveJob.Setup(sessions);
+
+        if (_async)
+        {
+            _leaveTask = _parallelMgr.Process(_leaveJob, _leaveJob.Count);
+            return;
+        }
+
+        using var _ = Histogram.WithLabels("Process Leave").NewTimer();
+        _parallelMgr.ProcessNow(_leaveJob, _leaveJob.Count);
+    }
+
+    /// <summary>
+    /// Figure out what entities are no longer visible to the client. These entities are sent reliably to the client
+    /// in a separate net message. This has to be called after EntityData.LastSent is updated.
+    /// </summary>
+    private void ProcessLeavePvs(PvsSession session)
+    {
+        if (session.DisableCulling || session.Session.Status != SessionStatus.InGame)
+            return;
+
+        if (session.LastSent == null)
+            return;
+
+        var (toTick, lastSent) = session.LastSent.Value;
+        foreach (var data in CollectionsMarshal.AsSpan(lastSent))
+        {
+            if (data.LastSeen == toTick)
+                continue;
+
+            session.LeftView.Add(data.NetEntity);
+            data.LastLeftView = toTick;
+        }
+
+        if (session.LeftView.Count == 0)
+            return;
+
+        var pvsMessage = new MsgStateLeavePvs {Entities = session.LeftView, Tick = toTick};
+
+        // PVS benchmarks use dummy sessions.
+        // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract
+        if (session.Session.Status == SessionStatus.InGame && session.Channel != null)
+            _netMan.ServerSendMessage(pvsMessage, session.Channel);
+
+        session.LeftView.Clear();
+    }
+
+    private record struct PvsLeaveJob : IParallelRobustJob
+    {
+        public int BatchSize => 2;
+        public PvsSystem Pvs;
+        public int Count => _sessions.Length;
+        private PvsSession[] _sessions;
+
+        public void Execute(int index)
+        {
+            Pvs.ProcessLeavePvs(_sessions[index]);
+        }
+
+        public void Setup(ICommonSession[] sessions)
+        {
+            // Copy references to PvsSession, in case players disconnect while the job is running.
+            Array.Resize(ref _sessions, sessions.Length);
+            for (var i = 0; i < sessions.Length; i++)
+            {
+                _sessions[i] = Pvs.PlayerData[sessions[i]];
+            }
+        }
+    }
+}

Robust.Server/GameStates/PvsSystem.Session.cs

@@ -3,6 +3,7 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Numerics;
 using System.Runtime.InteropServices;
+using Robust.Shared.Enums;
 using Robust.Shared.GameObjects;
 using Robust.Shared.GameStates;
 using Robust.Shared.Map;
@@ -62,12 +63,6 @@ internal sealed partial class PvsSystem
         }
 
         data.ClearState();
-
-        // TODO parallelize this with system processing.
-        // Before we do that we need to:
-        // - Defer player connection changes untill the start of the nxt PVS tick and  this job has finished
-        // - Defer OnEntityDeleted in pvs system. Or refactor per-session entity data to be stored as arrays on metadaat component
-        ProcessLeavePvs(data);
     }
 
     private PvsSession GetOrNewPvsSession(ICommonSession session)
@@ -174,4 +169,22 @@ internal sealed partial class PvsSystem
         if (session.PreviouslySent.TryGetValue(_gameTiming.CurTick - 1, out var lastSent))
             session.LastSent = (_gameTiming.CurTick, lastSent);
     }
+
+    private void OnPlayerStatusChanged(object? sender, SessionStatusEventArgs e)
+    {
+        if (e.NewStatus != SessionStatus.Disconnected)
+            return;
+
+        if (!PlayerData.Remove(e.Session, out var data))
+            return;
+
+        if (data.Overflow != null)
+            _entDataListPool.Return(data.Overflow.Value.SentEnts);
+        data.Overflow = null;
+
+        foreach (var visSet in data.PreviouslySent.Values)
+        {
+            _entDataListPool.Return(visSet);
+        }
+    }
 }

Robust.Server/GameStates/PvsSystem.ToSendSet.cs

@@ -53,8 +53,10 @@ internal sealed partial class PvsSystem
             : chunk.LodCounts[0];
 
         // Send entities on the chunk.
-        foreach (ref var ent in CollectionsMarshal.AsSpan(chunk.Contents))
+        var span = CollectionsMarshal.AsSpan(chunk.Contents);
+        for (var i = 0; i < count; i++)
         {
+            var ent = span[i];
             if ((mask & ent.Comp.VisibilityMask) == ent.Comp.VisibilityMask)
                 AddEntity(session, ent, fromTick);
         }

Robust.Server/GameStates/PvsSystem.cs

@@ -6,7 +6,6 @@ using System.Numerics;
 using System.Runtime.InteropServices;
 using System.Text;
 using System.Threading.Tasks;
-using FastAccessors;
 using Microsoft.Extensions.ObjectPool;
 using Prometheus;
 using Robust.Server.Configuration;
@@ -15,13 +14,10 @@ using Robust.Server.Player;
 using Robust.Server.Replays;
 using Robust.Shared;
 using Robust.Shared.Configuration;
-using Robust.Shared.Enums;
 using Robust.Shared.GameObjects;
 using Robust.Shared.Log;
 using Robust.Shared.Map;
-using Robust.Shared.Map.Components;
 using Robust.Shared.Network;
-using Robust.Shared.Network.Messages;
 using Robust.Shared.Player;
 using Robust.Shared.Threading;
 using Robust.Shared.Timing;
@@ -77,6 +73,7 @@ internal sealed partial class PvsSystem : EntitySystem
     internal readonly HashSet<ICommonSession> PendingAcks = new();
     private PvsAckJob _ackJob;
     private PvsChunkJob _chunkJob;
+    private PvsLeaveJob _leaveJob;
 
     private EntityQuery<EyeComponent> _eyeQuery;
     private EntityQuery<MetaDataComponent> _metaQuery;
@@ -110,8 +107,9 @@ internal sealed partial class PvsSystem : EntitySystem
     {
         base.Initialize();
 
+        _leaveJob = new PvsLeaveJob { Pvs = this };
         _chunkJob = new PvsChunkJob { Pvs = this };
-        _ackJob = new PvsAckJob { System = this, Sessions = _toAck,};
+        _ackJob = new PvsAckJob { Pvs = this };
 
         _eyeQuery = GetEntityQuery<EyeComponent>();
         _metaQuery = GetEntityQuery<MetaDataComponent>();
@@ -157,6 +155,8 @@ internal sealed partial class PvsSystem : EntitySystem
         _serverGameStateManager.ClientRequestFull -= OnClientRequestFull;
 
         ShutdownDirty();
+        _leaveTask?.WaitOne();
+        _leaveTask = null;
     }
 
     /// <summary>
@@ -172,6 +172,8 @@ internal sealed partial class PvsSystem : EntitySystem
 
         // Cull deletion history
         AfterSendState(players);
+
+        ProcessLeavePvs(players);
     }
 
     private void SendStates(ICommonSession[] players)
@@ -286,28 +288,6 @@ internal sealed partial class PvsSystem : EntitySystem
         _mapManager.CullDeletionHistory(oldestAck);
     }
 
-    #region PVSCollection Event Updates
-
-    private void OnPlayerStatusChanged(object? sender, SessionStatusEventArgs e)
-    {
-        if (e.NewStatus != SessionStatus.Disconnected)
-            return;
-
-        if (!PlayerData.Remove(e.Session, out var data))
-            return;
-
-        if (data.Overflow != null)
-            _entDataListPool.Return(data.Overflow.Value.SentEnts);
-        data.Overflow = null;
-
-        foreach (var visSet in data.PreviouslySent.Values)
-        {
-            _entDataListPool.Return(visSet);
-        }
-    }
-
-    #endregion
-
     private void GetEntityStates(PvsSession session)
     {
         // First, we send the client's own viewers. we want to ALWAYS send these, regardless of any pvs budget.
@@ -397,45 +377,6 @@ internal sealed partial class PvsSystem : EntitySystem
         }
     }
 
-    /// <summary>
-    /// Figure out what entities are no longer visible to the client. These entities are sent reliably to the client
-    /// in a separate net message. This has to be called after EntityData.LastSent is updated.
-    /// </summary>
-    private void ProcessLeavePvs(PvsSession session)
-    {
-        if (!CullingEnabled || session.DisableCulling)
-            return;
-
-        if (session.LastSent == null)
-            return;
-
-        var (toTick, lastSent) = session.LastSent.Value;
-        foreach (var data in CollectionsMarshal.AsSpan(lastSent))
-        {
-            if (data.LastSeen == toTick)
-                continue;
-
-            session.LeftView.Add(data.NetEntity);
-            data.LastLeftView = toTick;
-
-            // TODO PVS make this not required. I.e., hide maps/grids from clients.
-            DebugTools.Assert(!TryGetEntity(data.NetEntity, out var uid)
-                              || (!HasComp<MapGridComponent>(uid) && !HasComp<MapComponent>(uid)));
-        }
-
-        if (session.LeftView.Count == 0)
-            return;
-
-        var pvsMessage = new MsgStateLeavePvs {Entities = session.LeftView, Tick = toTick};
-
-        // PVS benchmarks use dummy sessions.
-        // ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract
-        if (session.Channel != null)
-            _netMan.ServerSendMessage(pvsMessage, session.Channel);
-
-        session.LeftView.Clear();
-    }
-
     private (Vector2 worldPos, float range, EntityUid? map) CalcViewBounds(Entity<TransformComponent, EyeComponent?> eye)
     {
         var size = Math.Max(eye.Comp2?.PvsSize ?? _viewSize, 1);
@@ -477,6 +418,8 @@ internal sealed partial class PvsSystem : EntitySystem
         }
 
         ackJob?.WaitOne();
+        _leaveTask?.WaitOne();
+        _leaveTask = null;
     }
 
     private void AfterSendState(ICommonSession[] players)

Robust.Shared/Threading/ParallelManager.cs

@@ -162,6 +162,12 @@ internal sealed class ParallelManager : IParallelManagerInternal
 
         // Need to set this up front to avoid firing too early.
         tracker.Event.Reset();
+        if (amount <= 0)
+        {
+            tracker.Event.Set();
+            return tracker;
+        }
+
         tracker.PendingTasks = batches;
 
         for (var i = 0; i < batches; i++)