Concurrent simulation runner

src/OSPSuite.Core/Domain/Services/ConcurrencyManager.cs

@@ -0,0 +1,64 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace OSPSuite.Core.Domain.Services
+{
+   public interface IConcurrencyManager
+   {
+      /// <summary>
+      /// 
+      /// </summary>
+      /// <typeparam name="TData">Data type to consume by the worker function</typeparam>
+      /// <typeparam name="TResult">Data produced by the worker function</typeparam>
+      /// <param name="numberOfCoresToUse">Number of cores to use. Use 0 or negative to take all cores</param>
+      /// <param name="cancellationToken">Cancellation token to cancel the threads</param>
+      /// <param name="data">List of data to consume by the workers</param>
+      /// <param name="action">A function to run on each worker on each piece of data</param>
+      /// <returns>Dictionary binding a result for each input data after running the action on it</returns>
+      Task<Dictionary<TData, TResult>> RunAsync<TData, TResult>(int numberOfCoresToUse, CancellationToken cancellationToken, IEnumerable<TData> data, Func<int, CancellationToken, TData, Task<TResult>> action);
+   }
+
+   public class ConcurrencyManager : IConcurrencyManager
+   {
+      private readonly ICoreUserSettings _coreUserSettings;
+
+      public ConcurrencyManager(ICoreUserSettings coreUserSettings)
+      {
+         _coreUserSettings = coreUserSettings;
+      }
+
+      public async Task<Dictionary<TData, TResult>> RunAsync<TData, TResult>(int numberOfCoresToUse, CancellationToken cancellationToken, IEnumerable<TData> data, Func<int, CancellationToken, TData, Task<TResult>> action)
+      {
+         if (numberOfCoresToUse <= 0)
+            numberOfCoresToUse = _coreUserSettings.MaximumNumberOfCoresToUse;
+         var concurrentData = new ConcurrentQueue<TData>(data);
+         numberOfCoresToUse = Math.Min(numberOfCoresToUse, concurrentData.Count);
+
+         var results = new ConcurrentDictionary<TData, TResult>();
+         //Starts one task per core
+         var tasks = Enumerable.Range(0, numberOfCoresToUse).Select(coreIndex =>
+            Task.Run(async () =>
+            {
+               //While there is data left
+               while (concurrentData.TryDequeue(out var datum))
+               {
+                  if (cancellationToken.IsCancellationRequested) return;
+
+                  //Invoke the action on it and store the result
+                  var result = await action.Invoke(coreIndex, cancellationToken, datum);
+                  results.TryAdd(datum, result);
+               }
+            }, cancellationToken)
+         ).ToList();
+
+         await Task.WhenAll(tasks);
+         //all tasks are completed. Can return results
+
+         return results.ToDictionary(kvp => kvp.Key, kvp => kvp.Value);
+      }
+   }
+}

src/OSPSuite.R/Domain/SimulationBatch.cs

@@ -48,6 +48,17 @@ public class SimulationBatchRunValues
       public double[] MoleculeValues => InitialValues.ToNetArray(InitialValue);
    }
 
+   public class SimulationBatchRunConcurrentlyOptions
+   {
+      private List<SimulationBatchRunValues> _simulationBatchRunValues = new List<SimulationBatchRunValues>();
+      public IReadOnlyList<SimulationBatchRunValues> SimulationBatchRunValues { get => _simulationBatchRunValues; }
+
+      public void AddSimulationBatchRunValues(SimulationBatchRunValues runValues)
+      {
+         _simulationBatchRunValues.Add(runValues);
+      }
+   }
+
    public class SimulationBatch : IDisposable
    {
       private readonly ISimModelBatch _simModelBatch;

src/OSPSuite.R/Services/ConcurrentSimulationRunner.cs

@@ -0,0 +1,110 @@
+using OSPSuite.Core.Domain;
+using OSPSuite.Core.Domain.Data;
+using OSPSuite.Core.Domain.Services;
+using OSPSuite.R.Domain;
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using SimulationRunOptions = OSPSuite.R.Domain.SimulationRunOptions;
+
+namespace OSPSuite.R.Services
+{
+   public interface IConcurrentSimulationRunner
+   {
+      /// <summary>
+      /// General simulation options
+      /// </summary>
+      SimulationRunOptions SimulationRunOptions { get; set; }
+
+      /// <summary>
+      /// Number of cores to use concurrently. Use 0 or a negative value for using the maximum available.
+      /// </summary>
+      int NumberOfCores { get; set; }
+
+      /// <summary>
+      /// Adds a simulation to the list of Simulations
+      /// </summary>
+      void AddSimulation(IModelCoreSimulation simulation);
+
+      /// <summary>
+      /// Adds a SimulationBatch to the list of SimulationBatches
+      /// </summary>
+      /// <param name="simulationBatch"></param>
+      void AddSimulationBatch(SimulationBatch simulationBatch);
+
+      void Clear();
+
+      /// <summary>
+      /// Runs all preset settings concurrently
+      /// </summary>
+      /// <returns></returns>
+      SimulationResults[] RunConcurrently();
+   }
+
+   public class ConcurrentSimulationRunner : IConcurrentSimulationRunner
+   {
+      private IConcurrencyManager _concurrentManager;
+
+      public ConcurrentSimulationRunner(IConcurrencyManager concurrentManager)
+      {
+         _concurrentManager = concurrentManager;
+      }
+
+      public SimulationRunOptions SimulationRunOptions { get; set; }
+
+      public int NumberOfCores { get; set; }
+
+      private List<IModelCoreSimulation> _simulations = new List<IModelCoreSimulation>();
+      private List<SimulationBatch> _simulationBatches = new List<SimulationBatch>();
+
+      public void AddSimulation(IModelCoreSimulation simulation)
+      {
+         _simulations.Add(simulation);
+      }
+
+      public void AddSimulationBatch(SimulationBatch simulationBatch)
+      {
+         _simulationBatches.Add(simulationBatch);
+      }
+
+      public void Clear()
+      {
+         _simulations.Clear();
+         _simulationBatches.Clear();
+      }
+
+      public SimulationResults[] RunConcurrently()
+      {
+         //Currently we only allow for running simulations or simulation batches, but not both
+         if (_simulationBatches.Count > 0 && _simulations.Count > 0) 
+            //Temporal Exception. We should allow for mixing both use cases but we need to discuss first
+            throw new Exception("You already have Simulation and SimulationBatch objects and should not mix, please invoke Clear to start adding objects from a fresh start");
+
+         if (_simulations.Count > 0)
+         {
+            var results = _concurrentManager.RunAsync(
+               NumberOfCores,
+               new CancellationToken(false),
+               new List<IModelCoreSimulation>(_simulations),
+               runSimulation
+            ).Result;
+            return _simulations.Select(s => results[s]).ToArray();
+         }
+
+         if (_simulationBatches.Count > 0)
+            return null;
+
+         return Enumerable.Empty<SimulationResults>().ToArray();
+      }
+
+      private async Task<SimulationResults> runSimulation(int coreIndex, CancellationToken cancellationToken, IModelCoreSimulation simulation)
+      {
+         return await Api.GetSimulationRunner().RunAsync(new SimulationRunArgs() { Simulation = simulation, SimulationRunOptions = SimulationRunOptions });
+      }
+   }
+
+
+}

tests/OSPSuite.R.Tests/Services/ConcurrentSimulationRunnerSpecs.cs

@@ -0,0 +1,46 @@
+using NUnit.Framework;
+using OSPSuite.BDDHelper;
+using OSPSuite.Core;
+using OSPSuite.Core.Domain.Data;
+using OSPSuite.Core.Domain.Services;
+using System.Linq;
+
+namespace OSPSuite.R.Services
+{
+   class CoreUserSettings : ICoreUserSettings
+   {
+      public int MaximumNumberOfCoresToUse { get; set; } = 4;
+      public int NumberOfBins { get; set; }
+      public int NumberOfIndividualsPerBin { get; set; }
+   }
+
+   public class When_running_simulations_concurrently : ContextForIntegration<IConcurrentSimulationRunner>
+   {
+      private ISimulationPersister _simulationPersister;
+      private SimulationResults[] _results;
+
+      protected override void Context()
+      {
+         base.Context();
+
+         _simulationPersister = Api.GetSimulationPersister();
+         sut = new ConcurrentSimulationRunner(new ConcurrencyManager(new CoreUserSettings()));
+         sut.AddSimulation(_simulationPersister.LoadSimulation(HelperForSpecs.DataFile("S1.pkml")));
+         sut.AddSimulation(_simulationPersister.LoadSimulation(HelperForSpecs.DataFile("simple.pkml")));
+         sut.AddSimulation(_simulationPersister.LoadSimulation(HelperForSpecs.DataFile("simple.pkml")));
+         sut.AddSimulation(_simulationPersister.LoadSimulation(HelperForSpecs.DataFile("multiple_dosing.pkml")));
+      }
+
+      protected override void Because()
+      {
+         _results = sut.RunConcurrently();
+      }
+
+      [Observation]
+      public void should_run_the_simulations()
+      {
+         Assert.IsNotNull(_results);
+         Assert.IsTrue(_results.All(r => r.ElementAt(0).AllValues.SelectMany(v => v.Values).Count() > 0));
+      }
+   }
+}