Refactored MultithreadSteppedDumpProcessor to improve concurrency and thread safety

source/LootDumpProcessor/Process/Processor/DumpProcessor/MultithreadSteppedDumpProcessor.cs

@@ -34,9 +34,7 @@ public class MultithreadSteppedDumpProcessor(
     private readonly ILooseLootProcessor _looseLootProcessor =
         looseLootProcessor ?? throw new ArgumentNullException(nameof(looseLootProcessor));
 
-    private static IJsonSerializer _jsonSerializer = JsonSerializerFactory.GetInstance();
+    private static readonly IJsonSerializer _jsonSerializer = JsonSerializerFactory.GetInstance();
-
-    private static readonly List<Task> Runners = new();
 
     // if we need to, this variable can be moved to use the factory, but since the factory
     // needs a locking mechanism to prevent dictionary access exceptions, its better to keep
@@ -53,44 +51,125 @@ public class MultithreadSteppedDumpProcessor(
         if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
             LoggerFactory.GetInstance().Log("Heavy processing done!", LogLevel.Info);
 
-        var staticContainers = new Dictionary<string, MapStaticLoot>();
+        var staticContainers = new ConcurrentDictionary<string, MapStaticLoot>();
-        var staticContainersLock = new object();
         // We need to count how many dumps we have for each map
-        var mapDumpCounter = new Dictionary<string, int>();
+        var mapDumpCounter = new ConcurrentDictionary<string, int>();
-        var mapDumpCounterLock = new object();
         // dictionary of maps, that has a dictionary of template and hit count
-        var mapStaticContainersAggregated = new Dictionary<string, Dictionary<Template, int>>();
+        var mapStaticContainersAggregated = new ConcurrentDictionary<string, ConcurrentDictionary<Template, int>>();
-        var mapStaticContainersAggregatedLock = new object();
 
-        Runners.Clear();
         // BSG changed the map data so static containers are now dynamic, so we need to scan all dumps for the static containers.
         if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
             LoggerFactory.GetInstance().Log("Queuing dumps for static data processing", LogLevel.Info);
-        foreach (var dumped in dumps)
+
+        var parallelOptions = new ParallelOptions
         {
-            Runners.Add(
+            MaxDegreeOfParallelism = Environment.ProcessorCount
-                Task.Factory.StartNew(() =>
+        };
+        Parallel.ForEachAsync(dumps, parallelOptions,
+            async (partialData, cancellationToken) =>
+                await Process(partialData, staticContainers, mapStaticContainersAggregated, mapDumpCounter));
+
+        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
+            LoggerFactory.GetInstance().Log("All static data processing threads finished", LogLevel.Info);
+        // Aggregate and calculate the probability of a static container
+        mapStaticContainersAggregated.ToDictionary(
+                kv => kv.Key,
+                kv => kv.Value.Select(
+                    td => new StaticDataPoint
+                    {
+                        Template = td.Key,
+                        Probability = GetStaticContainerProbability(kv.Key, td, mapDumpCounter) // kv.Key = map name
+                    }
+                ).ToList()
+            ).ToList()
+            .ForEach(kv =>
+                staticContainers[kv.Key].StaticContainers = kv.Value); // Hydrate staticContainers.StaticContainers
+
+        // Static containers
+        output.Add(OutputFileType.StaticContainer, staticContainers);
+        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
+            LoggerFactory.GetInstance().Log("Processing ammo distribution", LogLevel.Info);
+
+        var staticAmmo = new ConcurrentDictionary<string, IReadOnlyDictionary<string, List<AmmoDistribution>>>();
+        Parallel.ForEach(dumpProcessData.ContainerCounts.Keys, parallelOptions: parallelOptions, mapId =>
+        {
+            var preProcessedStaticLoots = dumpProcessData.ContainerCounts[mapId];
+            var ammoDistribution = _ammoProcessor.CreateAmmoDistribution(mapId, preProcessedStaticLoots);
+            staticAmmo[mapId] = ammoDistribution;
+        });
+        // Ammo distribution
+        output.Add(
+            OutputFileType.StaticAmmo,
+            staticAmmo
+        );
+
+        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
+            LoggerFactory.GetInstance().Log("Processing static loot distribution", LogLevel.Info);
+
+        var staticLoot = new ConcurrentDictionary<string, IReadOnlyDictionary<string, StaticItemDistribution>>();
+        Parallel.ForEach(dumpProcessData.ContainerCounts.Keys, parallelOptions: parallelOptions, mapId =>
+        {
+            var preProcessedStaticLoots = dumpProcessData.ContainerCounts[mapId];
+            var staticLootDistribution =
+                _staticLootProcessor.CreateStaticLootDistribution(mapId, preProcessedStaticLoots);
+            staticLoot[mapId] = staticLootDistribution;
+        });
+        // Static loot distribution
+        output.Add(
+            OutputFileType.StaticLoot,
+            staticLoot
+        );
+
+        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
+            LoggerFactory.GetInstance().Log("Processing loose loot distribution", LogLevel.Info);
+
+        // Loose loot distribution
+        var looseLoot = new ConcurrentDictionary<string, LooseLootRoot>();
+        Parallel.ForEach(dumpProcessData.MapCounts.Keys, parallelOptions: parallelOptions, mapId =>
+        {
+            var mapCount = dumpProcessData.MapCounts[mapId];
+            var looseLootCount = dumpProcessData.LooseLootCounts[mapId];
+            var looseLootDistribution =
+                _looseLootProcessor.CreateLooseLootDistribution(mapId, mapCount, looseLootCount);
+            looseLoot[mapId] = looseLootDistribution;
+        });
+        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
+            LoggerFactory.GetInstance().Log("Collecting loose loot distribution information", LogLevel.Info);
+        var loot = dumpProcessData.MapCounts
+            .Select(mapCount => mapCount.Key)
+            .ToDictionary(mi => mi, mi => looseLoot[mi]);
+
+        output.Add(OutputFileType.LooseLoot, loot);
+        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
+            LoggerFactory.GetInstance().Log("Dump processing fully completed!", LogLevel.Info);
+        return output;
+    }
+
+    private async Task Process(PartialData partialData,
+        ConcurrentDictionary<string, MapStaticLoot> staticContainers,
+        ConcurrentDictionary<string, ConcurrentDictionary<Template, int>> mapStaticContainersAggregated,
+        ConcurrentDictionary<string, int> mapDumpCounter)
     {
         if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Debug))
-                        LoggerFactory.GetInstance().Log($"Processing static data for file {dumped.BasicInfo.FileName}",
+            LoggerFactory.GetInstance().Log($"Processing static data for file {partialData.BasicInfo.FileName}",
                 LogLevel.Debug);
 
-                    var dataDump = _jsonSerializer.Deserialize<RootData>(File.ReadAllText(dumped.BasicInfo.FileName));
+        var fileContent = await File.ReadAllTextAsync(partialData.BasicInfo.FileName);
+        var dataDump = _jsonSerializer.Deserialize<RootData>(fileContent);
 
         if (dataDump == null)
         {
             if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Error))
                 LoggerFactory.GetInstance()
-                                .Log($"Failed to deserialize data from file {dumped.BasicInfo.FileName}",
+                    .Log($"Failed to deserialize data from file {partialData.BasicInfo.FileName}",
                         LogLevel.Error);
-                        return; // Skip processing this dump
+            return;
         }
 
         var mapId = dataDump.Data.LocationLoot.Id.ToLower();
 
         // Because we may use multiple version dumps for map data, merge the static loot between dumps
-                    lock (staticContainersLock)
+
-                    {
         if (!staticContainers.TryGetValue(mapId, out var mapStaticLoot))
         {
             if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
@@ -117,38 +196,34 @@ public class MultithreadSteppedDumpProcessor(
             mapStaticLoot.StaticWeapons.AddRange(newStaticWeapons);
             mapStaticLoot.StaticForced.AddRange(newStaticForced);
         }
-                    }
+
 
         // Takes care of finding how many "dynamic static containers" we have on the map
-                    Dictionary<Template, int> mapAggregatedDataDict;
+        ConcurrentDictionary<Template, int> mapAggregatedDataDict;
-                    lock (mapStaticContainersAggregatedLock)
+
-                    {
         // Init dict if map key doesnt exist
         if (!mapStaticContainersAggregated.TryGetValue(mapId, out mapAggregatedDataDict))
         {
-                            mapAggregatedDataDict = new Dictionary<Template, int>();
+            mapAggregatedDataDict = new ConcurrentDictionary<Template, int>();
-                            mapStaticContainersAggregated.Add(mapId, mapAggregatedDataDict);
+            mapStaticContainersAggregated.TryAdd(mapId, mapAggregatedDataDict);
-                        }
         }
 
+
         // Only process the dump file if the date is higher (after) the configuration date
-                    if (!DumpWasMadeAfterConfigThresholdDate(dumped))
+        if (!DumpWasMadeAfterConfigThresholdDate(partialData))
         {
             return;
         }
 
         // Keep track of how many dumps we have for each map
-                    lock (mapDumpCounterLock)
+
-                    {
         IncrementMapCounterDictionaryValue(mapDumpCounter, mapId);
-                    }
+
 
         var containerIgnoreListExists = LootDumpProcessorContext.GetConfig().ContainerIgnoreList
-                        .TryGetValue(mapId, out string[]? ignoreListForMap);
+            .TryGetValue(mapId, out var ignoreListForMap);
         foreach (var dynamicStaticContainer in _staticContainersProcessor.CreateDynamicStaticContainers(
                      dataDump))
-                    {
-                        lock (mapStaticContainersAggregatedLock)
         {
             if (containerIgnoreListExists && ignoreListForMap.Contains(dynamicStaticContainer.Id))
             {
@@ -162,88 +237,8 @@ public class MultithreadSteppedDumpProcessor(
                 mapAggregatedDataDict[dynamicStaticContainer] += 1;
             }
         }
-                    }
 
         GCHandler.Collect();
-                })
-            );
-        }
-
-        Task.WaitAll(Runners.ToArray());
-        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-            LoggerFactory.GetInstance().Log("All static data processing threads finished", LogLevel.Info);
-        // Aggregate and calculate the probability of a static container
-        mapStaticContainersAggregated.ToDictionary(
-                kv => kv.Key,
-                kv => kv.Value.Select(
-                    td => new StaticDataPoint
-                    {
-                        Template = td.Key,
-                        Probability = GetStaticContainerProbability(kv.Key, td, mapDumpCounter) // kv.Key = map name
-                    }
-                ).ToList()
-            ).ToList()
-            .ForEach(kv =>
-                staticContainers[kv.Key].StaticContainers = kv.Value); // Hydrate staticContainers.StaticContainers
-
-        // Static containers
-        output.Add(OutputFileType.StaticContainer, staticContainers);
-        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-            LoggerFactory.GetInstance().Log("Processing ammo distribution", LogLevel.Info);
-
-        var staticAmmo = new ConcurrentDictionary<string, IReadOnlyDictionary<string, List<AmmoDistribution>>>();
-        Parallel.ForEach(dumpProcessData.ContainerCounts.Keys, mapId =>
-        {
-            var preProcessedStaticLoots = dumpProcessData.ContainerCounts[mapId];
-            var ammoDistribution = _ammoProcessor.CreateAmmoDistribution(mapId, preProcessedStaticLoots);
-            staticAmmo[mapId] = ammoDistribution;
-        });
-        // Ammo distribution
-        output.Add(
-            OutputFileType.StaticAmmo,
-            staticAmmo
-        );
-
-        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-            LoggerFactory.GetInstance().Log("Processing static loot distribution", LogLevel.Info);
-
-        var staticLoot = new ConcurrentDictionary<string, IReadOnlyDictionary<string, StaticItemDistribution>>();
-        Parallel.ForEach(dumpProcessData.ContainerCounts.Keys, mapId =>
-        {
-            var preProcessedStaticLoots = dumpProcessData.ContainerCounts[mapId];
-            var staticLootDistribution =
-                _staticLootProcessor.CreateStaticLootDistribution(mapId, preProcessedStaticLoots);
-            staticLoot[mapId] = staticLootDistribution;
-        });
-        // Static loot distribution
-        output.Add(
-            OutputFileType.StaticLoot,
-            staticLoot
-        );
-
-        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-            LoggerFactory.GetInstance().Log("Processing loose loot distribution", LogLevel.Info);
-        
-        // Loose loot distribution
-        var looseLoot = new ConcurrentDictionary<string, LooseLootRoot>();
-        Parallel.ForEach(dumpProcessData.MapCounts.Keys, mapId =>
-        {
-            var mapCount = dumpProcessData.MapCounts[mapId];
-            var looseLootCount = dumpProcessData.LooseLootCounts[mapId];
-            var looseLootDistribution =
-                _looseLootProcessor.CreateLooseLootDistribution(mapId, mapCount, looseLootCount);
-            looseLoot[mapId] = looseLootDistribution;
-        });
-        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-            LoggerFactory.GetInstance().Log("Collecting loose loot distribution information", LogLevel.Info);
-        var loot = dumpProcessData.MapCounts
-            .Select(mapCount => mapCount.Key)
-            .ToDictionary(mi => mi, mi => looseLoot[mi]);
-
-        output.Add(OutputFileType.LooseLoot, loot);
-        if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-            LoggerFactory.GetInstance().Log("Dump processing fully completed!", LogLevel.Info);
-        return output;
     }
 
     private static bool DumpWasMadeAfterConfigThresholdDate(PartialData dataDump)
@@ -254,7 +249,7 @@ public class MultithreadSteppedDumpProcessor(
                    .SpawnContainerChanceIncludeAfterDate;
     }
 
-    private static void IncrementMapCounterDictionaryValue(Dictionary<string, int> mapDumpCounter, string mapName)
+    private static void IncrementMapCounterDictionaryValue(IDictionary<string, int> mapDumpCounter, string mapName)
     {
         if (!mapDumpCounter.TryAdd(mapName, 1))
         {
@@ -264,7 +259,7 @@ public class MultithreadSteppedDumpProcessor(
     }
 
     private static double GetStaticContainerProbability(string mapName, KeyValuePair<Template, int> td,
-        Dictionary<string, int> mapDumpCounter)
+        IReadOnlyDictionary<string, int> mapDumpCounter)
     {
         return Math.Round((double)((decimal)td.Value / (decimal)mapDumpCounter[mapName]), 2);
     }
@@ -321,19 +316,53 @@ public class MultithreadSteppedDumpProcessor(
                 tuple = null;
                 GCHandler.Collect();
 
-                // The data storage factory has a lock, we dont want the locks to occur when multithreading
+                var parallelOptions = new ParallelOptions
-                for (int i = 0; i < LootDumpProcessorContext.GetConfig().Threads; i++)
                 {
-                    Runners.Add(
+                    MaxDegreeOfParallelism = Environment.ProcessorCount
-                        Task.Factory.StartNew(
+                };
-                            () =>
+                Parallel.ForEach(_partialDataToProcess, parallelOptions,
+                    partialData =>
                     {
-                                while (_partialDataToProcess.TryTake(out var partialData,
+                        ProcessPartialData(partialData, lockObjectContainerCounts, dumpProcessData, mapName,
-                                           TimeSpan.FromMilliseconds(5000)))
+                            lockObjectDictionaryCounts, dictionaryCounts, lockObjectDictionaryItemProperties,
+                            dictionaryItemProperties, actualDictionaryItemProperties, lockObjectCounts,
+                            looseLootCounts);
+                    });
+
+                foreach (var (_, value) in dictionaryItemProperties)
+                {
+                    _dataStorage.Store(value);
+                }
+
+                _dataStorage.Store(dictionaryCounts);
+                dictionaryCounts = null;
+                GCHandler.Collect();
+                /*
+                DataStorageFactory.GetInstance().Store(dictionaryItemCounts);
+                dictionaryItemCounts = null;
+                GC.Collect();
+                */
+                _dataStorage.Store(actualDictionaryItemProperties);
+                actualDictionaryItemProperties = null;
+                GCHandler.Collect();
+                _dataStorage.Store(looseLootCounts);
+                looseLootCounts = null;
+                GCHandler.Collect();
+            });
+        return dumpProcessData;
+    }
+
+    private static void ProcessPartialData(PartialData partialDataToProcess,
+        object lockObjectContainerCounts,
+        DumpProcessData dumpProcessData, string mapName, object lockObjectDictionaryCounts,
+        FlatKeyableDictionary<string, int>? dictionaryCounts, object lockObjectDictionaryItemProperties,
+        FlatKeyableDictionary<string, FlatKeyableList<Template>> dictionaryItemProperties,
+        FlatKeyableDictionary<string, IKey>? actualDictionaryItemProperties,
+        object lockObjectCounts, LooseLootCounts? looseLootCounts)
     {
         try
         {
-                                        var dumpData = _dataStorage.GetItem<ParsedDump>(partialData.ParsedDumpKey);
+            var dumpData = _dataStorage.GetItem<ParsedDump>(partialDataToProcess.ParsedDumpKey);
 
             // Static containers
             lock (lockObjectContainerCounts)
@@ -366,16 +395,6 @@ public class MultithreadSteppedDumpProcessor(
                         dictionaryCounts[uniqueKey] = count;
                 }
 
-                                            /*
-                                            var itemList = dumpData.LooseLoot.Items[k];
-                                            if (!dictionaryItemCounts.TryGetValue(k, out var itemCounts))
-                                            {
-                                                itemCounts = new List<string>();
-                                                dictionaryItemCounts.Add(k, itemCounts);
-                                            }
-                                            itemCounts.AddRange(itemList);
-                                            */
-
                 lock (lockObjectDictionaryItemProperties)
                 {
                     if (!dictionaryItemProperties.TryGetValue(uniqueKey, out var values))
@@ -404,43 +423,4 @@ public class MultithreadSteppedDumpProcessor(
                 );
         }
     }
-                            },
-                            TaskCreationOptions.LongRunning)
-                    );
-                }
-
-                // Wait until all runners are done processing
-                while (!Runners.All(r => r.IsCompleted))
-                {
-                    if (LoggerFactory.GetInstance().CanBeLogged(LogLevel.Info))
-                        LoggerFactory.GetInstance().Log(
-                            $"One or more file processors are still processing files. Waiting {LootDumpProcessorContext.GetConfig().ThreadPoolingTimeoutMs}ms before checking again",
-                            LogLevel.Info
-                        );
-                    Thread.Sleep(
-                        TimeSpan.FromMilliseconds(LootDumpProcessorContext.GetConfig().ThreadPoolingTimeoutMs));
-                }
-
-                foreach (var (_, value) in dictionaryItemProperties)
-                {
-                    _dataStorage.Store(value);
-                }
-
-                _dataStorage.Store(dictionaryCounts);
-                dictionaryCounts = null;
-                GCHandler.Collect();
-                /*
-                DataStorageFactory.GetInstance().Store(dictionaryItemCounts);
-                dictionaryItemCounts = null;
-                GC.Collect();
-                */
-                _dataStorage.Store(actualDictionaryItemProperties);
-                actualDictionaryItemProperties = null;
-                GCHandler.Collect();
-                _dataStorage.Store(looseLootCounts);
-                looseLootCounts = null;
-                GCHandler.Collect();
-            });
-        return dumpProcessData;
-    }
 }

source/LootDumpProcessor/Process/Writer/FileWriter.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using LootDumpProcessor.Model.Output;
 using LootDumpProcessor.Model.Output.LooseLoot;
 using LootDumpProcessor.Model.Output.StaticContainer;
@@ -44,7 +45,7 @@ public class FileWriter : IWriter
         switch (type)
         {
             case OutputFileType.LooseLoot:
-                var looseLootData = (Dictionary<string, LooseLootRoot>)data;
+                var looseLootData = (IReadOnlyDictionary<string, LooseLootRoot>)data;
                 foreach (var (key, value) in looseLootData)
                 {
                     if (!Directory.Exists($@"{_outputPath}\locations\{key}"))
@@ -55,7 +56,7 @@ public class FileWriter : IWriter
 
                 break;
             case OutputFileType.StaticContainer:
-                var staticContainer = (Dictionary<string, MapStaticLoot>)data;
+                var staticContainer = (IReadOnlyDictionary<string, MapStaticLoot>)data;
                 foreach (var (key, value) in staticContainer)
                 {
                         if (!Directory.Exists($@"{_outputPath}\locations\{key}"))