Concurrent simulation runner

src/OSPSuite.Assets/UIConstants.cs

@@ -1212,6 +1212,7 @@ public static class Error
       public static readonly string NaNOnData = "Data contains NaN values at imported columns. Select a different action for NaN values or clean your data.";
       public static readonly string UnsupportedFileFormat = "The file format is not supported";
       public static readonly string InconsistenMoleculeAndMoleWeightException = "Molecule and Molecular Weight do not match. Please either edit your Molecule or your Molecular Weight in your project or remove the Molecular Weight from your mappings";
+      public static readonly string InvalidMixOfSimulationAndSimulationBatch = "You already have Simulation and SimulationBatch objects and should not mix, please invoke Clear to start adding objects from a fresh start";
 
       public static string LinkedParameterIsNotValidInIdentificationParameter(string identificationParameterName) => $"At least one linked parameter is invalid in identification paramter '{identificationParameterName}'";
 

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

@@ -0,0 +1,61 @@
+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<IReadOnlyDictionary<TData, TResult>> RunAsync<TData, TResult>(int numberOfCoresToUse, CancellationToken cancellationToken, IReadOnlyList<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<IReadOnlyDictionary<TData, TResult>> RunAsync<TData, TResult>(int numberOfCoresToUse, CancellationToken cancellationToken, IReadOnlyList<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(async coreIndex =>
+         {
+            //While there is data left
+            while (concurrentData.TryDequeue(out var datum))
+            {
+               cancellationToken.ThrowIfCancellationRequested();
+
+               //Invoke the action on it and store the result
+               var result = await action.Invoke(coreIndex, cancellationToken, datum);
+               results.TryAdd(datum, result);
+            }
+         }).ToList();
+
+         await Task.WhenAll(tasks);
+         //all tasks are completed. Can return results
+
+         return results;
+      }
+   }
+}

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,139 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using OSPSuite.Assets;
+using OSPSuite.Core.Domain;
+using OSPSuite.Core.Domain.Data;
+using OSPSuite.Core.Domain.Services;
+using OSPSuite.R.Domain;
+using OSPSuite.Utility.Exceptions;
+using SimulationRunOptions = OSPSuite.R.Domain.SimulationRunOptions;
+
+namespace OSPSuite.R.Services
+{
+   public interface IConcurrentSimulationRunner : IDisposable
+   {
+      /// <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 readonly IConcurrencyManager _concurrentManager;
+
+      public ConcurrentSimulationRunner(IConcurrencyManager concurrentManager)
+      {
+         _concurrentManager = concurrentManager;
+      }
+
+      public SimulationRunOptions SimulationRunOptions { get; set; }
+
+      public int NumberOfCores { get; set; }
+
+      private readonly List<IModelCoreSimulation> _simulations = new List<IModelCoreSimulation>();
+      private readonly List<SimulationBatch> _simulationBatches = new List<SimulationBatch>();
+      private CancellationTokenSource _cancellationTokenSource;
+
+      public void AddSimulation(IModelCoreSimulation simulation)
+      {
+         _simulations.Add(simulation);
+      }
+
+      public void AddSimulationBatch(SimulationBatch simulationBatch)
+      {
+         _simulationBatches.Add(simulationBatch);
+      }
+
+      public void Clear()
+      {
+         _simulations.Clear();
+         _simulationBatches.Clear();
+         if (_cancellationTokenSource != null)
+            _cancellationTokenSource.Cancel();
+      }
+
+      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 OSPSuiteException(Error.InvalidMixOfSimulationAndSimulationBatch);
+
+         _cancellationTokenSource = new CancellationTokenSource();
+         if (_simulations.Count > 0)
+         { 
+            var results = _concurrentManager.RunAsync(
+               NumberOfCores,
+               _cancellationTokenSource.Token,
+               _simulations,
+               runSimulation
+            ).Result;
+            return _simulations.Select(s => results[s]).ToArray();
+         }
+
+         if (_simulationBatches.Count > 0)
+            return null;
+
+         return Array.Empty<SimulationResults>();
+      }
+
+      private Task<SimulationResults> runSimulation(int coreIndex, CancellationToken cancellationToken, IModelCoreSimulation simulation)
+      {
+         //We want a new instance every time that's why we are not injecting SimulationRunner in constructor
+         return Api.GetSimulationRunner().RunAsync(new SimulationRunArgs {Simulation = simulation, SimulationRunOptions = SimulationRunOptions});
+      }
+
+      #region Disposable properties
+
+      private bool _disposed;
+
+      public void Dispose()
+      {
+         if (_disposed) return;
+
+         Cleanup();
+         GC.SuppressFinalize(this);
+         _disposed = true;
+      }
+
+      ~ConcurrentSimulationRunner()
+      {
+         Cleanup();
+      }
+
+      protected virtual void Cleanup()
+      {
+         Clear();
+      }
+
+      #endregion
+   }
+}

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));
+      }
+   }
+}