diff --git a/src/Microsoft.ML.AutoML/API/BinaryClassificationExperiment.cs b/src/Microsoft.ML.AutoML/API/BinaryClassificationExperiment.cs index 012b230b3c..f658a6f1aa 100644 --- a/src/Microsoft.ML.AutoML/API/BinaryClassificationExperiment.cs +++ b/src/Microsoft.ML.AutoML/API/BinaryClassificationExperiment.cs @@ -150,6 +150,7 @@ internal BinaryClassificationExperiment(MLContext context, BinaryExperimentSetti { _experiment.SetMaximumMemoryUsageInMegaByte(d); } + _experiment.SetMaxModelToExplore(settings.MaxModels); } public override ExperimentResult Execute(IDataView trainData, ColumnInformation columnInformation, IEstimator preFeaturizer = null, IProgress> progressHandler = null) diff --git a/src/Microsoft.ML.AutoML/API/MulticlassClassificationExperiment.cs b/src/Microsoft.ML.AutoML/API/MulticlassClassificationExperiment.cs index 8a7cba943d..836eddedb5 100644 --- a/src/Microsoft.ML.AutoML/API/MulticlassClassificationExperiment.cs +++ b/src/Microsoft.ML.AutoML/API/MulticlassClassificationExperiment.cs @@ -142,6 +142,7 @@ internal MulticlassClassificationExperiment(MLContext context, MulticlassExperim { _experiment.SetMaximumMemoryUsageInMegaByte(d); } + _experiment.SetMaxModelToExplore(settings.MaxModels); } public override ExperimentResult Execute(IDataView trainData, ColumnInformation columnInformation, IEstimator preFeaturizer = null, IProgress> progressHandler = null) diff --git a/src/Microsoft.ML.AutoML/API/RegressionExperiment.cs b/src/Microsoft.ML.AutoML/API/RegressionExperiment.cs index 50e17822be..a836b6c0c0 100644 --- a/src/Microsoft.ML.AutoML/API/RegressionExperiment.cs +++ b/src/Microsoft.ML.AutoML/API/RegressionExperiment.cs @@ -139,6 +139,7 @@ internal RegressionExperiment(MLContext context, RegressionExperimentSettings se } _experiment.SetTrainingTimeInSeconds(Settings.MaxExperimentTimeInSeconds); + _experiment.SetMaxModelToExplore(Settings.MaxModels); } public override ExperimentResult Execute(IDataView trainData, ColumnInformation columnInformation, IEstimator preFeaturizer = null, IProgress> progressHandler = null) diff --git a/src/Microsoft.ML.AutoML/AutoMLExperiment/AutoMLExperiment.cs b/src/Microsoft.ML.AutoML/AutoMLExperiment/AutoMLExperiment.cs index 4cf264ec7d..60d2790f28 100644 --- a/src/Microsoft.ML.AutoML/AutoMLExperiment/AutoMLExperiment.cs +++ b/src/Microsoft.ML.AutoML/AutoMLExperiment/AutoMLExperiment.cs @@ -25,7 +25,6 @@ public class AutoMLExperiment private double _bestLoss = double.MaxValue; private TrialResult _bestTrialResult = null; private readonly IServiceCollection _serviceCollection; - private CancellationTokenSource _globalCancellationTokenSource; public AutoMLExperiment(MLContext context, AutoMLExperimentSettings settings) { @@ -51,14 +50,15 @@ private void InitializeServiceCollection() _serviceCollection.TryAddTransient((provider) => { var contextManager = provider.GetRequiredService(); + var trainingStopManager = provider.GetRequiredService(); var context = contextManager.CreateMLContext(); - _globalCancellationTokenSource.Token.Register(() => + trainingStopManager.OnStopTraining += (s, e) => { // only force-canceling running trials when there's completed trials. // otherwise, wait for the current running trial to be completed. if (_bestTrialResult != null) context.CancelExecution(); - }); + }; return context; }); @@ -74,6 +74,29 @@ private void InitializeServiceCollection() public AutoMLExperiment SetTrainingTimeInSeconds(uint trainingTimeInSeconds) { _settings.MaxExperimentTimeInSeconds = trainingTimeInSeconds; + _serviceCollection.AddScoped((provider) => + { + var channel = provider.GetRequiredService(); + var timeoutManager = new TimeoutTrainingStopManager(TimeSpan.FromSeconds(trainingTimeInSeconds), channel); + + return timeoutManager; + }); + + return this; + } + + public AutoMLExperiment SetMaxModelToExplore(int maxModel) + { + _context.Assert(maxModel > 0, "maxModel has to be greater than 0"); + _settings.MaxModels = maxModel; + _serviceCollection.AddScoped((provider) => + { + var channel = provider.GetRequiredService(); + var maxModelManager = new MaxModelStopManager(maxModel, channel); + + return maxModelManager; + }); + return this; } @@ -204,19 +227,29 @@ public TrialResult Run() public async Task RunAsync(CancellationToken ct = default) { ValidateSettings(); - _globalCancellationTokenSource = new CancellationTokenSource(); - _settings.CancellationToken = ct; - // use TimeSpan to avoid overflow. - _globalCancellationTokenSource.CancelAfter(TimeSpan.FromSeconds(_settings.MaxExperimentTimeInSeconds)); - _settings.CancellationToken.Register(() => _globalCancellationTokenSource.Cancel()); + _serviceCollection.AddScoped((serviceProvider) => + { + var logger = serviceProvider.GetRequiredService(); + var stopServices = serviceProvider.GetServices(); + var cancellationTrainingStopManager = new CancellationTokenStopTrainingManager(ct, logger); + + // always get the most recent added stop service for each type. + var mostRecentAddedStopServices = stopServices.GroupBy(s => s.GetType()).Select(g => g.Last()).ToList(); + mostRecentAddedStopServices.Add(cancellationTrainingStopManager); + return new AggregateTrainingStopManager(logger, mostRecentAddedStopServices.ToArray()); + }); + var serviceProvider = _serviceCollection.BuildServiceProvider(); - var monitor = serviceProvider.GetService(); + + _settings.CancellationToken = ct; var logger = serviceProvider.GetRequiredService(); + var aggregateTrainingStopManager = serviceProvider.GetRequiredService(); + var monitor = serviceProvider.GetService(); var trialResultManager = serviceProvider.GetService(); var trialNum = trialResultManager?.GetAllTrialResults().Max(t => t.TrialSettings?.TrialId) + 1 ?? 0; var tuner = serviceProvider.GetService(); Contracts.Assert(tuner != null, "tuner can't be null"); - while (!_globalCancellationTokenSource.Token.IsCancellationRequested) + while (!aggregateTrainingStopManager.IsStopTrainingRequested()) { var setting = new TrialSettings() { @@ -227,86 +260,95 @@ public async Task RunAsync(CancellationToken ct = default) setting.Parameter = parameter; monitor?.ReportRunningTrial(setting); - try + using (var trialCancellationTokenSource = new CancellationTokenSource()) { - using (var trialCancellationTokenSource = new CancellationTokenSource()) - using (var deregisterCallback = _globalCancellationTokenSource.Token.Register(() => + void handler(object o, EventArgs e) { // only force-canceling running trials when there's completed trials. // otherwise, wait for the current running trial to be completed. if (_bestTrialResult != null) trialCancellationTokenSource.Cancel(); - })) - using (var performanceMonitor = serviceProvider.GetService()) - using (var runner = serviceProvider.GetRequiredService()) + } + try { - performanceMonitor.MemoryUsageInMegaByte += (o, m) => + using (var performanceMonitor = serviceProvider.GetService()) + using (var runner = serviceProvider.GetRequiredService()) { - if (_settings.MaximumMemoryUsageInMegaByte is double d && m > d && !trialCancellationTokenSource.IsCancellationRequested) - { - logger.Trace($"cancel current trial {setting.TrialId} because it uses {m} mb memory and the maximum memory usage is {d}"); - trialCancellationTokenSource.Cancel(); + aggregateTrainingStopManager.OnStopTraining += handler; - GC.AddMemoryPressure(Convert.ToInt64(m) * 1024 * 1024); - GC.Collect(); + performanceMonitor.MemoryUsageInMegaByte += (o, m) => + { + if (_settings.MaximumMemoryUsageInMegaByte is double d && m > d && !trialCancellationTokenSource.IsCancellationRequested) + { + logger.Trace($"cancel current trial {setting.TrialId} because it uses {m} mb memory and the maximum memory usage is {d}"); + trialCancellationTokenSource.Cancel(); + + GC.AddMemoryPressure(Convert.ToInt64(m) * 1024 * 1024); + GC.Collect(); + } + }; + + performanceMonitor.Start(); + logger.Trace($"trial setting - {JsonSerializer.Serialize(setting)}"); + var trialResult = await runner.RunAsync(setting, trialCancellationTokenSource.Token); + + var peakCpu = performanceMonitor?.GetPeakCpuUsage(); + var peakMemoryInMB = performanceMonitor?.GetPeakMemoryUsageInMegaByte(); + trialResult.PeakCpu = peakCpu; + trialResult.PeakMemoryInMegaByte = peakMemoryInMB; + + monitor?.ReportCompletedTrial(trialResult); + tuner.Update(trialResult); + trialResultManager?.AddOrUpdateTrialResult(trialResult); + aggregateTrainingStopManager.Update(trialResult); + + var loss = trialResult.Loss; + if (loss < _bestLoss) + { + _bestTrialResult = trialResult; + _bestLoss = loss; + monitor?.ReportBestTrial(trialResult); } + } + } + catch (OperationCanceledException ex) when (aggregateTrainingStopManager.IsStopTrainingRequested() == false) + { + monitor?.ReportFailTrial(setting, ex); + var result = new TrialResult + { + TrialSettings = setting, + Loss = double.MaxValue, }; - performanceMonitor.Start(); - logger.Trace($"trial setting - {JsonSerializer.Serialize(setting)}"); - var trialResult = await runner.RunAsync(setting, trialCancellationTokenSource.Token); - - var peakCpu = performanceMonitor?.GetPeakCpuUsage(); - var peakMemoryInMB = performanceMonitor?.GetPeakMemoryUsageInMegaByte(); - trialResult.PeakCpu = peakCpu; - trialResult.PeakMemoryInMegaByte = peakMemoryInMB; - - monitor?.ReportCompletedTrial(trialResult); - tuner.Update(trialResult); - trialResultManager?.AddOrUpdateTrialResult(trialResult); + tuner.Update(result); + continue; + } + catch (OperationCanceledException) when (aggregateTrainingStopManager.IsStopTrainingRequested()) + { + break; + } + catch (Exception ex) + { + monitor?.ReportFailTrial(setting, ex); - var loss = trialResult.Loss; - if (loss < _bestLoss) + if (!aggregateTrainingStopManager.IsStopTrainingRequested() && _bestTrialResult == null) { - _bestTrialResult = trialResult; - _bestLoss = loss; - monitor?.ReportBestTrial(trialResult); + // TODO + // it's questionable on whether to abort the entire training process + // for a single fail trial. We should make it an option and only exit + // when error is fatal (like schema mismatch). + throw; } } - } - catch (OperationCanceledException ex) when (_globalCancellationTokenSource.IsCancellationRequested == false) - { - monitor?.ReportFailTrial(setting, ex); - var result = new TrialResult + finally { - TrialSettings = setting, - Loss = double.MaxValue, - }; + aggregateTrainingStopManager.OnStopTraining -= handler; - tuner.Update(result); - continue; - } - catch (OperationCanceledException) when (_globalCancellationTokenSource.IsCancellationRequested) - { - break; - } - catch (Exception ex) - { - monitor?.ReportFailTrial(setting, ex); - - if (!_globalCancellationTokenSource.IsCancellationRequested && _bestTrialResult == null) - { - // TODO - // it's questionable on whether to abort the entire training process - // for a single fail trial. We should make it an option and only exit - // when error is fatal (like schema mismatch). - throw; } } } trialResultManager?.Save(); - if (_bestTrialResult == null) { throw new TimeoutException("Training time finished without completing a trial run"); diff --git a/src/Microsoft.ML.AutoML/AutoMLExperiment/IStopTrainingManager.cs b/src/Microsoft.ML.AutoML/AutoMLExperiment/IStopTrainingManager.cs new file mode 100644 index 0000000000..e49ceef4ee --- /dev/null +++ b/src/Microsoft.ML.AutoML/AutoMLExperiment/IStopTrainingManager.cs @@ -0,0 +1,155 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading; +using Microsoft.ML.Runtime; + +#nullable enable +namespace Microsoft.ML.AutoML +{ + internal interface IStopTrainingManager + { + bool IsStopTrainingRequested(); + + void Update(TrialResult result); + + public event EventHandler OnStopTraining; + } + + internal class CancellationTokenStopTrainingManager : IStopTrainingManager + { + private readonly CancellationToken _token; + private readonly IChannel? _channel; + public event EventHandler? OnStopTraining; + + public CancellationTokenStopTrainingManager(CancellationToken ct, IChannel? channel) + { + _token = ct; + _channel = channel; + ct.Register(() => + { + _channel?.Info("cancel training because cancellation token is invoked..."); + OnStopTraining?.Invoke(this, EventArgs.Empty); + }); + } + + public bool IsStopTrainingRequested() + { + if (_token.IsCancellationRequested) + { + return true; + } + + return false; + } + + public void Update(TrialResult result) + { + return; + } + } + + internal class TimeoutTrainingStopManager : IStopTrainingManager + { + private readonly CancellationTokenStopTrainingManager _cancellationTokenTrainingStopManager; + private readonly CancellationTokenSource _cts; + + public event EventHandler? OnStopTraining; + + public TimeoutTrainingStopManager(TimeSpan timeoutInSeconds, IChannel? channel) + { + _cts = new CancellationTokenSource(); + _cts.CancelAfter(timeoutInSeconds); + _cancellationTokenTrainingStopManager = new CancellationTokenStopTrainingManager(_cts.Token, channel); + _cancellationTokenTrainingStopManager.OnStopTraining += (o, e) => + { + OnStopTraining?.Invoke(this, e); + }; + } + + public bool IsStopTrainingRequested() + { + return _cancellationTokenTrainingStopManager.IsStopTrainingRequested(); + } + + public void Update(TrialResult result) + { + return; + } + } + + internal class MaxModelStopManager : IStopTrainingManager + { + private readonly int _maxModel; + private int _exploredModel = 0; + public event EventHandler? OnStopTraining; + + public MaxModelStopManager(int maxModel, IChannel? channel) + { + _maxModel = maxModel; + } + + public bool IsStopTrainingRequested() + { + return _exploredModel >= _maxModel; + } + + public void Update(TrialResult result) + { + _exploredModel++; + if (_exploredModel > _maxModel) + { + OnStopTraining?.Invoke(this, EventArgs.Empty); + } + } + } + + /// + /// stop training when any of child training stop manager is stopped. + /// + internal class AggregateTrainingStopManager : IStopTrainingManager + { + private readonly List _managers; + + public event EventHandler? OnStopTraining; + + public AggregateTrainingStopManager(IChannel? channel, params IStopTrainingManager[] managers) + { + _managers = managers.ToList(); + foreach (var manager in _managers) + { + manager.OnStopTraining += (o, e) => + { + OnStopTraining?.Invoke(this, e); + }; + } + } + + public bool IsStopTrainingRequested() + { + return _managers.Any(m => m.IsStopTrainingRequested()); + } + + public void AddTrainingStopManager(IStopTrainingManager manager) + { + _managers.Add(manager); + manager.OnStopTraining += (o, e) => + { + OnStopTraining?.Invoke(this, e); + }; + } + + public void Update(TrialResult result) + { + foreach (var manager in _managers) + { + manager.Update(result); + } + } + } +} diff --git a/test/Microsoft.ML.AutoML.Tests/AutoMLExperimentTests.cs b/test/Microsoft.ML.AutoML.Tests/AutoMLExperimentTests.cs index b5fc42e448..a9dbd14caa 100644 --- a/test/Microsoft.ML.AutoML.Tests/AutoMLExperimentTests.cs +++ b/test/Microsoft.ML.AutoML.Tests/AutoMLExperimentTests.cs @@ -171,6 +171,32 @@ public async Task AutoMLExperiment_finish_training_when_time_is_up_Async() cts.IsCancellationRequested.Should().BeFalse(); } + [Fact] + public async Task AutoMLExperiment_finish_training_when_reach_to_max_model_async() + { + var context = new MLContext(1); + var experiment = context.Auto().CreateExperiment(); + experiment.SetMaxModelToExplore(5) + .SetTrialRunner((serviceProvider) => + { + var channel = serviceProvider.GetService(); + var settings = serviceProvider.GetService(); + return new DummyTrialRunner(settings, 1, channel); + }) + .SetTuner(); + + var runModelCounts = 0; + context.Log += (o, e) => + { + if (e.RawMessage.Contains("Update Completed Trial")) + { + runModelCounts++; + } + }; + await experiment.RunAsync(); + runModelCounts.Should().Be(5); + } + [Fact] public async Task AutoMLExperiment_UCI_Adult_Train_Test_Split_Test() @@ -297,7 +323,7 @@ public async Task AutoMLExperiment_Taxi_Fare_Train_Test_Split_Test() experiment.SetDataset(train, test) .SetRegressionMetric(RegressionMetric.RSquared, label) .SetPipeline(pipeline) - .SetTrainingTimeInSeconds(50); + .SetMaxModelToExplore(1); var result = await experiment.RunAsync(); result.Metric.Should().BeGreaterThan(0.5); @@ -307,13 +333,6 @@ public async Task AutoMLExperiment_Taxi_Fare_Train_Test_Split_Test() public async Task AutoMLExperiment_Taxi_Fare_CV_5_Test() { var context = new MLContext(1); - context.Log += (o, e) => - { - if (e.Source.StartsWith("AutoMLExperiment")) - { - this.Output.WriteLine(e.RawMessage); - } - }; var train = DatasetUtil.GetTaxiFareTrainDataView(); var experiment = context.Auto().CreateExperiment(); var label = DatasetUtil.TaxiFareLabel; @@ -323,7 +342,7 @@ public async Task AutoMLExperiment_Taxi_Fare_CV_5_Test() experiment.SetDataset(train, 5) .SetRegressionMetric(RegressionMetric.RSquared, label) .SetPipeline(pipeline) - .SetTrainingTimeInSeconds(50); + .SetMaxModelToExplore(1); var result = await experiment.RunAsync(); result.Metric.Should().BeGreaterThan(0.5); diff --git a/test/Microsoft.ML.AutoML.Tests/StopTrainingManagerTests.cs b/test/Microsoft.ML.AutoML.Tests/StopTrainingManagerTests.cs new file mode 100644 index 0000000000..4e87bf238d --- /dev/null +++ b/test/Microsoft.ML.AutoML.Tests/StopTrainingManagerTests.cs @@ -0,0 +1,111 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using FluentAssertions; +using Microsoft.ML.TestFramework; +using Xunit; +using Xunit.Abstractions; + +namespace Microsoft.ML.AutoML.Test +{ + public class StopTrainingManagerTests : BaseTestClass + { + public StopTrainingManagerTests(ITestOutputHelper output) + : base(output) + { + } + + [Fact] + public void CancellationTokenStopTrainingManager_isStopTrainingRequested_test() + { + var cts = new CancellationTokenSource(); + var manager = new CancellationTokenStopTrainingManager(cts.Token, null); + + var isOnStopTrainingGetInvoked = false; + manager.OnStopTraining += (o, e) => + { + isOnStopTrainingGetInvoked = true; + }; + + manager.IsStopTrainingRequested().Should().BeFalse(); + + cts.Cancel(); + + manager.IsStopTrainingRequested().Should().BeTrue(); + isOnStopTrainingGetInvoked.Should().BeTrue(); + } + + [Fact] + public async Task TimeoutTrainingStopManager_isStopTrainingRequested_test() + { + var manager = new TimeoutTrainingStopManager(TimeSpan.FromSeconds(1), null); + + var isOnStopTrainingGetInvoked = false; + manager.OnStopTraining += (o, e) => + { + isOnStopTrainingGetInvoked = true; + }; + + var waitForCompleted = Task.Run(async () => + { + while (!isOnStopTrainingGetInvoked) + { + await Task.Delay(1000); + } + }); + + await waitForCompleted; + manager.IsStopTrainingRequested().Should().BeTrue(); + isOnStopTrainingGetInvoked.Should().BeTrue(); + } + + [Fact] + public async Task AggregateTrainingStopManager_isStopTrainingRequested_test() + { + var cts = new CancellationTokenSource(); + var timeoutManager = new TimeoutTrainingStopManager(TimeSpan.FromSeconds(1), null); + var cancellationManager = new CancellationTokenStopTrainingManager(cts.Token, null); + var aggregationManager = new AggregateTrainingStopManager(null, timeoutManager, cancellationManager); + + var isOnStopTrainingGetInvoked = false; + aggregationManager.OnStopTraining += (o, e) => + { + isOnStopTrainingGetInvoked = true; + }; + + var waitForCompleted = Task.Run(async () => + { + while (!isOnStopTrainingGetInvoked) + { + await Task.Delay(1000); + } + }); + + await waitForCompleted; + aggregationManager.IsStopTrainingRequested().Should().BeTrue(); + timeoutManager.IsStopTrainingRequested().Should().BeTrue(); + cancellationManager.IsStopTrainingRequested().Should().BeFalse(); + isOnStopTrainingGetInvoked.Should().BeTrue(); + + aggregationManager = new AggregateTrainingStopManager(null, new TimeoutTrainingStopManager(TimeSpan.FromSeconds(100), null), cancellationManager); + isOnStopTrainingGetInvoked = false; + aggregationManager.OnStopTraining += (o, e) => + { + isOnStopTrainingGetInvoked = true; + }; + cts.Cancel(); + + await waitForCompleted; + aggregationManager.IsStopTrainingRequested().Should().BeTrue(); + cancellationManager.IsStopTrainingRequested().Should().BeTrue(); + isOnStopTrainingGetInvoked.Should().BeTrue(); + } + } +}