From 284b230163ad86a9eebe0092363c096761e698b2 Mon Sep 17 00:00:00 2001 From: "DESKTOP-O6U9TBJ\\Lilith" Date: Fri, 27 Oct 2023 18:01:36 -0700 Subject: [PATCH] fix parallel overloads; add tests --- src/Arch.Samples/Game.cs | 3 +- .../Queries/InlineParallelQuery.cs | 23 ++ src/Arch.SourceGen/QueryGenerator.cs | 2 +- src/Arch.Tests/QueryTest.cs | 5 +- src/Arch.Tests/ScheduledQueryTest.cs | 271 ++++++++++++++++++ src/Arch/Core/Jobs/Jobs.cs | 2 +- src/Arch/Core/Jobs/World.Jobs.cs | 14 +- 7 files changed, 308 insertions(+), 12 deletions(-) create mode 100644 src/Arch.Tests/ScheduledQueryTest.cs diff --git a/src/Arch.Samples/Game.cs b/src/Arch.Samples/Game.cs index 278f8853..173d41df 100644 --- a/src/Arch.Samples/Game.cs +++ b/src/Arch.Samples/Game.cs @@ -2,6 +2,7 @@ using Microsoft.Xna.Framework; using Microsoft.Xna.Framework.Graphics; using Microsoft.Xna.Framework.Input; +using Schedulers; namespace Arch.Samples; @@ -12,7 +13,7 @@ public class Game : Microsoft.Xna.Framework.Game { // The world and a job scheduler for multithreading. private World _world; - private JobScheduler.JobScheduler _jobScheduler; + private JobScheduler _jobScheduler; // Our systems processing entities. private MovementSystem _movementSystem; diff --git a/src/Arch.SourceGen/Queries/InlineParallelQuery.cs b/src/Arch.SourceGen/Queries/InlineParallelQuery.cs index 5f00d979..cce9c091 100644 --- a/src/Arch.SourceGen/Queries/InlineParallelQuery.cs +++ b/src/Arch.SourceGen/Queries/InlineParallelQuery.cs @@ -28,6 +28,17 @@ public static void AppendHpParallelQuery(this StringBuilder builder, int amount) }; return InlineParallelChunkQuery(in queryDescription, in innerJob, in dependency, batchSize); } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public JobHandle InlineParallelQuery(in QueryDescription queryDescription, in JobHandle? dependency = null, int batchSize = 16) + where T : struct, IForEach<{{generics}}> + { + var innerJob = new IForEachJob() + { + ForEach = new() + }; + return InlineParallelChunkQuery(in queryDescription, in innerJob, in dependency, batchSize); + } """; builder.AppendLine(template); @@ -60,6 +71,18 @@ public static void AppendHpeParallelQuery(this StringBuilder builder, int amount }; return InlineParallelChunkQuery(in queryDescription, in innerJob, in dependency, batchSize); } + + [MethodImpl(MethodImplOptions.AggressiveInlining)] + public JobHandle InlineParallelEntityQuery(in QueryDescription queryDescription, + in JobHandle? dependency = null, int batchSize = 16) + where T : struct, IForEachWithEntity<{{generics}}> + { + var innerJob = new IForEachWithEntityJob() + { + ForEach = new() + }; + return InlineParallelChunkQuery(in queryDescription, in innerJob, in dependency, batchSize); + } """; builder.AppendLine(template); diff --git a/src/Arch.SourceGen/QueryGenerator.cs b/src/Arch.SourceGen/QueryGenerator.cs index 23b8f9e6..ea6cbd9e 100644 --- a/src/Arch.SourceGen/QueryGenerator.cs +++ b/src/Arch.SourceGen/QueryGenerator.cs @@ -56,7 +56,7 @@ public void Initialize(IncrementalGeneratorInitializationContext context) var accessors = new StringBuilder(); accessors.AppendLine("using System;"); accessors.AppendLine("using System.Runtime.CompilerServices;"); - accessors.AppendLine("using JobScheduler;"); + accessors.AppendLine("using Schedulers;"); accessors.AppendLine("using Arch.Core.Utils;"); accessors.AppendLine("using System.Diagnostics.Contracts;"); accessors.AppendLine("using Arch.Core.Extensions;"); diff --git a/src/Arch.Tests/QueryTest.cs b/src/Arch.Tests/QueryTest.cs index 40490567..c5c43c0a 100644 --- a/src/Arch.Tests/QueryTest.cs +++ b/src/Arch.Tests/QueryTest.cs @@ -1,5 +1,6 @@ using Arch.Core; using Arch.Core.Utils; +using Schedulers; using static NUnit.Framework.Assert; namespace Arch.Tests; @@ -7,7 +8,7 @@ namespace Arch.Tests; [TestFixture] public partial class QueryTest { - private JobScheduler.JobScheduler _jobScheduler; + private JobScheduler _jobScheduler; private World? _world; private static readonly ComponentType[] _entityGroup = { typeof(Transform), typeof(Rotation) }; @@ -19,7 +20,7 @@ public partial class QueryTest [OneTimeSetUp] public void Setup() { - _jobScheduler = new JobScheduler.JobScheduler(new() { + _jobScheduler = new(new() { ThreadPrefixName = nameof(QueryTest) }); } diff --git a/src/Arch.Tests/ScheduledQueryTest.cs b/src/Arch.Tests/ScheduledQueryTest.cs new file mode 100644 index 00000000..4b39fbfb --- /dev/null +++ b/src/Arch.Tests/ScheduledQueryTest.cs @@ -0,0 +1,271 @@ +using System.Diagnostics; +using Arch.Core; +using Arch.Core.Extensions; +using Schedulers; + +namespace Arch.Tests; +[TestFixture] +[System.Diagnostics.CodeAnalysis.SuppressMessage("Style", "IDE0061:Use block body for local function", Justification = "Space efficiency")] +[System.Diagnostics.CodeAnalysis.SuppressMessage("Style", "IDE0022:Use block body for method", Justification = "Space efficiency")] +internal class ScheduledQueryTest +{ + public enum InlineMode + { + NotInline, + Inline, + InlineWithStructRef + } + + [Test] + public void ParallelQueriesPoolCorrectly() + { + var world = World.Create(); + world.AttachScheduler(new JobScheduler(new() + { + ThreadPrefixName = nameof(ScheduledQueryTest) + })); + var dep = new SleepJob(); + for (int i = 0; i < 1000; i++) + { + // entity to query + world.Create(); + // entity to ignore + world.Create(); + } + + int nCounter = 0; + int counter = 0; + + // Grab from pool, ensure it actually allocates memory + var mem = GC.GetAllocatedBytesForCurrentThread(); + var job1 = ChunkIterationJobPool>.Get(); + var job2 = ChunkIterationJobPool>.Get(); + var mem2 = GC.GetAllocatedBytesForCurrentThread(); + Assert.That(mem, Is.Not.EqualTo(mem2)); + + // Since we never return, grab 2 more new ones from the pool. + // They should pool themselves. + var handle1 = world.ParallelQuery(new QueryDescription().WithAll().WithNone(), (ref S1 s1) => Interlocked.Increment(ref counter)); + var handle2 = world.ParallelQuery(new QueryDescription().WithAll(), (ref S1 s1) => Interlocked.Increment(ref nCounter)); + + world.Scheduler?.Flush(); + handle1.Complete(); + handle2.Complete(); + + // Ensure they actually ran + Assert.Multiple(() => + { + Assert.That(nCounter, Is.EqualTo(1000)); + Assert.That(counter, Is.EqualTo(1000)); + }); + + // Ensure that grabbing from the pool is now free + mem = GC.GetAllocatedBytesForCurrentThread(); + job1 = ChunkIterationJobPool>.Get(); + job2 = ChunkIterationJobPool>.Get(); + mem2 = GC.GetAllocatedBytesForCurrentThread(); + Assert.That(mem, Is.EqualTo(mem2)); + } + + /// + /// Test all the various parallel queries and their overloads, including dependency validation. + /// Only chooses a subset of the many generic overloads: up to 3 generics. + /// + /// + /// + /// + [Test, Combinatorial] + public void GenericParallelQueriesFunction( + [Values(-1, 0, 1, 2)] int genericCount, + [Values(InlineMode.NotInline, InlineMode.Inline, InlineMode.InlineWithStructRef)] InlineMode inlineMode, + [Values(true, false)] bool includeEntity) + { + // skip this one, it doesn't mean anything + if (!includeEntity && genericCount == -1) + { + return; + } + + var world = World.Create(); + world.AttachScheduler(new JobScheduler(new() + { + ThreadPrefixName = nameof(ScheduledQueryTest) + })); + + var dep = new SleepJob(); + + for (int i = 0; i < 1000; i++) + { + // entity to query + world.Create(new() { SleepJob = dep }); + // entity to ignore + world.Create(new() { SleepJob = dep }); + } + + static void fe0(ref S0 s0) => s0.Update(); + static void fe1(ref S0 s0, ref S1 s1) => s0.Update(); + static void fe2(ref S0 s0, ref S1 s1, ref S2 s2) => s0.Update(); + static void fee(Entity e) + { + ref var s0 = ref e.Get(); + s0.Update(); + } + + static void fee0(Entity e, ref S0 s0) => s0.Update(); + static void fee1(Entity e, ref S0 s0, ref S1 s1) => s0.Update(); + static void fee2(Entity e, ref S0 s0, ref S1 s1, ref S2 s2) => s0.Update(); + + FE0 sfe0 = new(); + FE1 sfe1 = new(); + FE2 sfe2 = new(); + FEE sfee = new(); + FEE0 sfee0 = new(); + FEE1 sfee1 = new(); + FEE2 sfee2 = new(); + + QueryDescription qd = genericCount switch + { + -1 => new QueryDescription().WithAll().WithNone(), + 0 => new QueryDescription().WithAll().WithNone(), + 1 => new QueryDescription().WithAll().WithNone(), + _ => new QueryDescription().WithAll().WithNone(), + }; + + // setup a dependency + var dependency = world.Scheduler?.Schedule(dep); + + JobHandle? handle = null; + + if (inlineMode == InlineMode.NotInline && includeEntity) + { + handle = genericCount switch + { + -1 => world.ParallelQuery(qd, fee, dependency), + 0 => world.ParallelQuery(qd, (ForEachWithEntity)fee0, dependency), + 1 => world.ParallelQuery(qd, (ForEachWithEntity)fee1, dependency), + _ => world.ParallelQuery(qd, (ForEachWithEntity)fee2, dependency), + }; + } + else if (inlineMode == InlineMode.NotInline && !includeEntity) + { + + handle = genericCount switch + { + 0 => world.ParallelQuery(qd, (ForEach)fe0, dependency), + 1 => world.ParallelQuery(qd, (ForEach)fe1, dependency), + _ => world.ParallelQuery(qd, (ForEach)fe2, dependency), + }; + } + else if (inlineMode == InlineMode.InlineWithStructRef && includeEntity) + { + handle = genericCount switch + { + -1 => world.InlineParallelQuery(qd, ref sfee, dependency), + 0 => world.InlineParallelEntityQuery(qd, ref sfee0, dependency), + 1 => world.InlineParallelEntityQuery(qd, ref sfee1, dependency), + _ => world.InlineParallelEntityQuery(qd, ref sfee2, dependency), + }; + + } + else if (inlineMode == InlineMode.Inline && includeEntity) + { + handle = genericCount switch + { + -1 => world.InlineParallelQuery(qd, dependency), + 0 => world.InlineParallelEntityQuery(qd, dependency), + 1 => world.InlineParallelEntityQuery(qd, dependency), + _ => world.InlineParallelEntityQuery(qd, dependency), + }; + } + else if (inlineMode == InlineMode.InlineWithStructRef && !includeEntity) + { + handle = genericCount switch + { + 0 => world.InlineParallelQuery(qd, ref sfe0, dependency), + 1 => world.InlineParallelQuery(qd, ref sfe1, dependency), + _ => world.InlineParallelQuery(qd, ref sfe2, dependency), + }; + + } + else if (inlineMode == InlineMode.Inline && !includeEntity) + { + handle = genericCount switch + { + 0 => world.InlineParallelQuery(qd, dependency), + 1 => world.InlineParallelQuery(qd, dependency), + _ => world.InlineParallelQuery(qd, dependency), + }; + } + + Debug.Assert(handle is not null); + + world.Scheduler?.Flush(); + handle.Value.Complete(); + + world.Query(qd, (ref S0 s0) => Assert.That(s0.Counter, Is.EqualTo(1))); + world.Query(new QueryDescription().WithAll(), (ref S0 s0) => Assert.That(s0.Counter, Is.EqualTo(0))); + } + + private class SleepJob : IJob + { + public bool Complete { get; private set; } = false; + public void Execute() + { + Thread.Sleep(5); + Complete = true; + } + } + + private struct N { } + private struct S0 + { + public int Counter; + public SleepJob SleepJob; + + public void Update() + { + // we can't use NUnit.Assert because it's slow + if (!SleepJob.Complete) + { + throw new InvalidOperationException("Dependency didn't work!"); + } + + Interlocked.Increment(ref Counter); + } + } + private struct S1 { } + private struct S2 { } + + private struct FE0 : IForEach + { + public readonly void Update(ref S0 s0) => s0.Update(); + } + private struct FE1 : IForEach + { + public readonly void Update(ref S0 s0, ref S1 s1) => s0.Update(); + } + private struct FE2 : IForEach + { + public readonly void Update(ref S0 s0, ref S1 s1, ref S2 s2) => s0.Update(); + } + private struct FEE : IForEach + { + public readonly void Update(Entity e) + { + ref var s0 = ref e.Get(); + s0.Update(); + } + } + private struct FEE0 : IForEachWithEntity + { + public readonly void Update(Entity e, ref S0 s0) => s0.Update(); + } + private struct FEE1 : IForEachWithEntity + { + public readonly void Update(Entity e, ref S0 s0, ref S1 s1) => s0.Update(); + } + private struct FEE2 : IForEachWithEntity + { + public readonly void Update(Entity e, ref S0 s0, ref S1 s1, ref S2 s2) => s0.Update(); + } +} diff --git a/src/Arch/Core/Jobs/Jobs.cs b/src/Arch/Core/Jobs/Jobs.cs index ba5bb0e8..654b9bcd 100644 --- a/src/Arch/Core/Jobs/Jobs.cs +++ b/src/Arch/Core/Jobs/Jobs.cs @@ -1,6 +1,6 @@ using System.Collections.Concurrent; using CommunityToolkit.HighPerformance; -using JobScheduler; +using Schedulers; namespace Arch.Core; diff --git a/src/Arch/Core/Jobs/World.Jobs.cs b/src/Arch/Core/Jobs/World.Jobs.cs index 70903a55..d249a0a1 100644 --- a/src/Arch/Core/Jobs/World.Jobs.cs +++ b/src/Arch/Core/Jobs/World.Jobs.cs @@ -1,4 +1,4 @@ -using JobScheduler; +using Schedulers; // ReSharper disable once CheckNamespace namespace Arch.Core; @@ -8,7 +8,7 @@ namespace Arch.Core; public partial class World { /// - /// Thrown when the has not been assigned a . + /// Thrown when the has not been assigned a . /// public class JobSchedulerNotAssignedException : Exception { @@ -23,20 +23,20 @@ public class NotOnMainThreadException : Exception { internal NotOnMainThreadException() : base($"A scheduling method cannot be called on a different thread than {nameof(World.Scheduler)} was created on. " + - $"Either create the {nameof(JobScheduler.JobScheduler)} on a different thread, or only schedule queries on the main thread.") { } + $"Either create the {nameof(JobScheduler)} on a different thread, or only schedule queries on the main thread.") { } } /// - /// The attached to this , or null if none has been attached. + /// The attached to this , or null if none has been attached. /// - public JobScheduler.JobScheduler? Scheduler { get; private set; } + public JobScheduler? Scheduler { get; private set; } /// - /// Attach a to this . Only one scheduler can be attached, and it cannot + /// Attach a to this . Only one scheduler can be attached, and it cannot /// be changed once set. /// /// The scheduler to assign. - public void AttachScheduler(JobScheduler.JobScheduler scheduler) + public void AttachScheduler(JobScheduler scheduler) { if (Scheduler is not null) {