[WIP] Add ReadAllParallelAsync (multi-threaded parsing)

.gitignore

@@ -3,5 +3,6 @@ bin/
 /GameTracking-CSGO/
 *.dem
 .idea/
+.vs/
 *.DotSettings.user
-BenchmarkDotNet.Artifacts/
+BenchmarkDotNet.Artifacts/

README.md

@@ -52,12 +52,15 @@ See also the [examples/](./examples) folder.
 
 ## Benchmarks
 
-On an M1 MacBook Pro, DemoFile.Net can read a full competitive game (just under 1 hour of game time) in 1.5 seconds.
+On an M1 MacBook Pro, DemoFile.Net can read a full competitive game (just under 1 hour of game time) in 1.3 seconds.
+When parsing across multiple threads, using the `ReadAllParallelAsync` method, this drops to nearly 500 milliseconds.
 This includes parsing all entity data (player positions, velocities, weapon tracking, grenades, etc).
 
-| Method    | Runtime  |        Mean |    Error |   StdDev |
-|-----------|----------|------------:|---------:|---------:|
-| ParseDemo | .NET 8.0 | **1.501 s** | 0.0047 s | 0.0042 s |
+| Method            |           Mean | Error    | StdDev   | Allocated |
+|-------------------|---------------:|---------:|---------:|----------:|
+| ParseDemo         | **1,294.6 ms** |  3.68 ms |  2.88 ms | 491.48 MB |
+| ParseDemoParallel |   **540.1 ms** | 23.99 ms | 22.44 ms | 600.67 MB |
+
 
 ## Author and acknowledgements
 

RELEASE_NOTES.md

@@ -1,3 +1,8 @@
+### 0.13.1 (2024-04-25)
+
+- Add `DemoParser.ReadAllParallelAsync` to read a demo across multiple threads. \
+  Many thanks to [@in0finite](https://github.com/in0finite) for the initial implementation.
+
 ### 0.12.2 (2024-04-12)
 
 - Fix player pawn positions occasionally jittering ([#37](https://github.com/saul/demofile-net/issues/37)). Thanks to [@in0finite](https://github.com/in0finite) for spotting the bug.

src/DemoFile.Benchmark/DemoParserBenchmark.cs

@@ -48,4 +48,10 @@ public async Task ParseDemo()
         await _demoParser.ReadAllAsync(_fileStream, default);
 #endif
     }
+
+    [Benchmark]
+    public async Task ParseDemoParallel()
+    {
+        await DemoParser.ReadAllParallelAsync(_demoBytes, _ => { },default);
+    }
 }

src/DemoFile.Test/DemoSnapshot.cs

@@ -0,0 +1,55 @@
+using System.Diagnostics;
+using System.Text;
+
+namespace DemoFile.Test;
+
+[DebuggerDisplay("Count = {Count}")]
+public class DemoSnapshot
+{
+    private Dictionary<DemoTick, List<string>> _items = new();
+
+    public int Count => _items.Count;
+
+    public void Add(DemoTick tick, string details)
+    {
+        if (!_items.TryGetValue(tick, out var tickItems))
+        {
+            _items[tick] = new List<string> {details};
+        }
+        else if (!tickItems.Contains(details))
+        {
+            tickItems.Add(details);
+        }
+    }
+
+    public void MergeFrom(DemoSnapshot other)
+    {
+        foreach (var (tick, items) in other._items)
+        {
+            foreach (var item in items)
+            {
+                Add(tick, item);
+            }
+        }
+    }
+
+    public override string ToString()
+    {
+        var result = new StringBuilder();
+
+        foreach (var (tick, items) in _items.OrderBy(kvp => kvp.Key))
+        {
+            foreach (var item in items)
+            {
+                result.Append($"[{tick}] {item}");
+
+                if (item[^1] != '\n')
+                {
+                    result.AppendLine();
+                }
+            }
+        }
+
+        return result.ToString();
+    }
+}

src/DemoFile.Test/GlobalUtil.cs

@@ -1,26 +1,27 @@
 using System.Collections;
+using System.Text;
 
 namespace DemoFile.Test;
 
 public static class GlobalUtil
 {
     private static readonly string DemoBase = Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "demos");
 
-    public static MemoryStream GotvCompetitiveProtocol13963 => new(File.ReadAllBytes(Path.Combine(DemoBase, "navi-javelins-vs-9-pandas-fearless-m1-mirage.dem")));
+    public static byte[] GotvCompetitiveProtocol13963 { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "navi-javelins-vs-9-pandas-fearless-m1-mirage.dem"));
 
-    public static MemoryStream GotvCompetitiveProtocol13992 => new(File.ReadAllBytes(Path.Combine(DemoBase, "virtus-pro-vs-natus-vincere-m1-ancient.dem")));
+    public static byte[] GotvCompetitiveProtocol13992 { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "virtus-pro-vs-natus-vincere-m1-ancient.dem"));
 
     public static MemoryStream MatchmakingProtocol13968 => new(File.ReadAllBytes(Path.Combine(DemoBase, "93n781.dem")));
 
-    public static MemoryStream GotvProtocol13978 => new(File.ReadAllBytes(Path.Combine(DemoBase, "13978.dem")));
+    public static byte[] GotvProtocol13978 { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "13978.dem"));
 
-    public static MemoryStream GotvProtocol13980 => new(File.ReadAllBytes(Path.Combine(DemoBase, "13980.dem")));
+    public static byte[] GotvProtocol13980 { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "13980.dem"));
 
-    public static MemoryStream GotvProtocol13987 => new(File.ReadAllBytes(Path.Combine(DemoBase, "13987.dem")));
+    public static byte[] GotvProtocol13987 { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "13987.dem"));
 
-    public static MemoryStream GotvProtocol13990ArmsRace => new(File.ReadAllBytes(Path.Combine(DemoBase, "13990_armsrace.dem")));
+    public static byte[] GotvProtocol13990ArmsRace { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "13990_armsrace.dem"));
 
-    public static MemoryStream GotvProtocol13990Deathmatch => new(File.ReadAllBytes(Path.Combine(DemoBase, "13990_dm.dem")));
+    public static byte[] GotvProtocol13990Deathmatch { get; } = File.ReadAllBytes(Path.Combine(DemoBase, "13990_dm.dem"));
 
     public static byte[] ToBitStream(string input)
     {
@@ -40,4 +41,49 @@ public static byte[] ToBitStream(string input)
         bitArray.CopyTo(bytes, 0);
         return bytes;
     }
+
+    public static ParseMode[] ParseModes => Enum.GetValues<ParseMode>();
+
+    public static async Task<string> Parse(ParseMode mode, byte[] demoFileBytes, Func<DemoParser, DemoSnapshot> parseSection)
+    {
+        if (mode == ParseMode.ReadAll)
+        {
+            var demo = new DemoParser();
+            var stream = new MemoryStream(demoFileBytes);
+
+            var result = parseSection(demo);
+
+            await demo.ReadAllAsync(stream, default);
+            return result.ToString();
+        }
+        else if (mode == ParseMode.ByTick)
+        {
+            var demo = new DemoParser();
+            var stream = new MemoryStream(demoFileBytes);
+
+            var result = parseSection(demo);
+
+            await demo.StartReadingAsync(stream, default);
+            while (await demo.MoveNextAsync(default))
+            {
+            }
+
+            return result.ToString();
+        }
+        else if (mode == ParseMode.ReadAllParallel)
+        {
+            var results = await DemoParser.ReadAllParallelAsync(demoFileBytes, parseSection, default);
+
+            var acc = results.Aggregate(new DemoSnapshot(), (acc, snapshot) =>
+            {
+                acc.MergeFrom(snapshot);
+                return acc;
+            });
+            return acc.ToString();
+        }
+        else
+        {
+            throw new NotImplementedException();
+        }
+    }
 }

src/DemoFile.Test/Integration/DemoEventsIntegrationTest.cs

@@ -1,51 +1,42 @@
-using System.Text;
-using System.Text.Json;
+using System.Text.Json;
 
 namespace DemoFile.Test.Integration;
 
-[TestFixture(true)]
-[TestFixture(false)]
+[TestFixtureSource(typeof(GlobalUtil), nameof(ParseModes))]
 public class DemoEventsIntegrationTest
 {
-    private readonly bool _readAll;
+    private readonly ParseMode _mode;
 
-    public DemoEventsIntegrationTest(bool readAll)
+    public DemoEventsIntegrationTest(ParseMode mode)
     {
-        _readAll = readAll;
+        _mode = mode;
     }
 
     [Test]
     public async Task DemoFileInfo()
     {
         // Arrange
-        var snapshot = new StringBuilder();
-        var demo = new DemoParser();
-
-        demo.DemoEvents.DemoFileInfo += e =>
+        DemoSnapshot ParseSection(DemoParser demo)
         {
-            snapshot.AppendLine($"[{demo.CurrentDemoTick}/{demo.CurrentGameTick}] TickCount={demo.TickCount}");
-            snapshot.AppendLine($"[{demo.CurrentDemoTick}/{demo.CurrentGameTick}] DemoFileInfo: {JsonSerializer.Serialize(e, DemoJson.SerializerOptions)}");
-        };
+            var snapshot = new DemoSnapshot();
 
-        demo.PacketEvents.SvcServerInfo += e =>
-        {
-            snapshot.AppendLine($"[{demo.CurrentDemoTick}/{demo.CurrentGameTick}] SvcServerInfo: {JsonSerializer.Serialize(e, DemoJson.SerializerOptions)}");
-        };
+            demo.DemoEvents.DemoFileInfo += e =>
+            {
+                snapshot.Add(demo.CurrentDemoTick, $"GameTick: {demo.CurrentGameTick}, TickCount: {demo.TickCount}, DemoFileInfo: {JsonSerializer.Serialize(e, DemoJson.SerializerOptions)}");
+            };
 
-        // Act
-        if (_readAll)
-        {
-            await demo.ReadAllAsync(GotvCompetitiveProtocol13963, default);
-        }
-        else
-        {
-            await demo.StartReadingAsync(GotvCompetitiveProtocol13963, default);
-            while (await demo.MoveNextAsync(default))
+            demo.PacketEvents.SvcServerInfo += e =>
             {
-            }
+                snapshot.Add(demo.CurrentDemoTick, $"GameTick: {demo.CurrentGameTick}, SvcServerInfo: {JsonSerializer.Serialize(e, DemoJson.SerializerOptions)}");
+            };
+
+            return snapshot;
         }
 
+        // Act
+        var snapshot = await Parse(_mode, GotvCompetitiveProtocol13963, ParseSection);
+
         // Assert
-        Snapshot.Assert(snapshot.ToString());
+        Snapshot.Assert(snapshot);
     }
 }

src/DemoFile.Test/Integration/DemoParserIntegrationTest.cs

@@ -7,7 +7,7 @@ public class DemoParserIntegrationTest
     public async Task ReadAll()
     {
         var demo = new DemoParser();
-        await demo.ReadAllAsync(GotvCompetitiveProtocol13963, default);
+        await demo.ReadAllAsync(new MemoryStream(GotvCompetitiveProtocol13963), default);
         Assert.That(demo.CurrentDemoTick.Value, Is.EqualTo(217866));
     }
 
@@ -19,7 +19,7 @@ public async Task ByTick()
         var tick = demo.CurrentDemoTick;
 
         // Act
-        await demo.StartReadingAsync(GotvCompetitiveProtocol13963, default);
+        await demo.StartReadingAsync(new MemoryStream(GotvCompetitiveProtocol13963), default);
         while (await demo.MoveNextAsync(default))
         {
             // Tick is monotonic
@@ -31,7 +31,7 @@ public async Task ByTick()
         Assert.That(demo.CurrentDemoTick.Value, Is.EqualTo(217866));
     }
 
-    private static readonly KeyValuePair<string, Stream>[] CompatibilityCases =
+    private static readonly KeyValuePair<string, byte[]>[] CompatibilityCases =
     {
         new("v13978", GotvProtocol13978),
         new("v13980", GotvProtocol13980),
@@ -40,11 +40,18 @@ public async Task ByTick()
         new("v13990_dm", GotvProtocol13990Deathmatch),
     };
 
-    [TestCaseSource(nameof(CompatibilityCases))]
-    public async Task ReadAll_Compatibility(KeyValuePair<string, Stream> testCase)
+    [Test]
+    public async Task Compatibility(
+        [Values] ParseMode mode,
+        [ValueSource(nameof(CompatibilityCases))] KeyValuePair<string, byte[]> testCase)
     {
-        var demo = new DemoParser();
-        await demo.ReadAllAsync(testCase.Value, default);
+        DemoSnapshot ParseSection(DemoParser demo)
+        {
+            // no-op - we're just parsing the demo to the end
+            return new DemoSnapshot();
+        }
+
+        await Parse(mode, testCase.Value, ParseSection);
     }
 
     [Test]