CrossValidator: isolates used

gyrdym · gyrdym · commit ecedbe2879cd · 2020-12-25T18:18:10.000+02:00
diff --git a/lib/src/di/common/init_common_module.dart b/lib/src/di/common/init_common_module.dart
@@ -24,6 +24,8 @@ import 'package:ml_algo/src/model_selection/model_assessor/classifier_assessor.d
 import 'package:ml_algo/src/model_selection/model_assessor/model_assessor.dart';
 import 'package:ml_algo/src/model_selection/model_assessor/regressor_assessor.dart';
 import 'package:ml_algo/src/predictor/predictor.dart';
+import 'package:ml_algo/src/service/worker_manager/worker_manager.dart';
+import 'package:ml_algo/src/service/worker_manager/worker_manager_impl.dart';
 import 'package:ml_dataframe/ml_dataframe.dart';
 import 'package:ml_preprocessing/ml_preprocessing.dart';
 
@@ -76,5 +78,8 @@ void initCommonModule() {
         RegressorAssessor(
           injector.get<MetricFactory>(),
           featuresTargetSplit,
-        ));
+        ))
+
+    ..registerSingletonIf<WorkerManager>(
+            () => WorkerManagerImpl());
 }
diff --git a/lib/src/model_selection/_init_module.dart b/lib/src/model_selection/_init_module.dart
@@ -1,11 +1,13 @@
+import 'package:ml_algo/src/di/common/init_common_module.dart';
 import 'package:ml_algo/src/model_selection/_injector.dart';
 import 'package:ml_algo/src/model_selection/split_indices_provider/split_indices_provider_factory.dart';
 import 'package:ml_algo/src/model_selection/split_indices_provider/split_indices_provider_factory_impl.dart';
+import 'package:ml_algo/src/extensions/injector.dart';
 
 void initModelSelectionModule() {
-  if (!modelSelectionInjector.exists<SplitIndicesProviderFactory>()) {
-    modelSelectionInjector
-      ..registerSingleton<SplitIndicesProviderFactory>(
-              () => const SplitIndicesProviderFactoryImpl());
-  }
+  initCommonModule();
+
+  modelSelectionInjector
+    ..registerSingletonIf<SplitIndicesProviderFactory>(
+            () => const SplitIndicesProviderFactoryImpl());
 }
diff --git a/lib/src/model_selection/cross_validator/_helpers/assess_predictor.dart b/lib/src/model_selection/cross_validator/_helpers/assess_predictor.dart
@@ -0,0 +1,64 @@
+import 'package:ml_algo/src/common/exception/invalid_test_data_columns_number_exception.dart';
+import 'package:ml_algo/src/common/exception/invalid_train_data_columns_number_exception.dart';
+import 'package:ml_algo/src/model_selection/cross_validator/cross_validator_assessing_data_model.dart';
+import 'package:ml_dataframe/ml_dataframe.dart';
+import 'package:ml_linalg/dtype.dart';
+import 'package:ml_linalg/matrix.dart';
+import 'package:ml_linalg/vector.dart';
+import 'package:quiver/iterables.dart';
+
+num assessPredictor(Map<String, dynamic> assessingDataJson) {
+  final assessingDataModel = CrossValidatorAssessingDataModel
+      .fromJson(assessingDataJson);
+  final predictorPrototype = assessingDataModel.predictorPrototype;
+  final testRowsIndices = assessingDataModel.testRowsIndices;
+  final dtype = predictorPrototype.dtype;
+  final samples = assessingDataModel.samplesPrototype;
+  final discreteColumns = enumerate(samples.series)
+      .where((indexedSeries) => indexedSeries.value.isDiscrete)
+      .map((indexedSeries) => indexedSeries.index);
+  final samplesAsMatrix = samples.toMatrix(dtype);
+  final sourceColumnsNum = samplesAsMatrix.columnsNum;
+  final testRowsIndicesAsSet = Set<int>.from(testRowsIndices);
+  final trainSamples = List<Vector>(
+      samplesAsMatrix.rowsNum - testRowsIndicesAsSet.length);
+  final testSamples = List<Vector>(testRowsIndicesAsSet.length);
+
+  var trainSamplesCounter = 0;
+  var testSamplesCounter = 0;
+
+  samplesAsMatrix.rowIndices.forEach((i) {
+    if (testRowsIndicesAsSet.contains(i)) {
+      testSamples[testSamplesCounter++] = samplesAsMatrix[i];
+    } else {
+      trainSamples[trainSamplesCounter++] = samplesAsMatrix[i];
+    }
+  });
+
+  final transformedData = [
+    DataFrame.fromMatrix(
+      Matrix.fromRows(trainSamples, dtype: dtype),
+      header: samples.header,
+      discreteColumns: discreteColumns,
+    ),
+    DataFrame.fromMatrix(
+      Matrix.fromRows(testSamples, dtype: dtype),
+      header: samples.header,
+      discreteColumns: discreteColumns,
+    ),
+  ];
+  final transformedTrainData = transformedData[0];
+  final transformedTestData = transformedData[1];
+
+  if (transformedTrainData.header.length != sourceColumnsNum) {
+    throw InvalidTrainDataColumnsNumberException(sourceColumnsNum,
+        transformedTrainData.header.length);
+  }
+
+  if (transformedTestData.header.length != sourceColumnsNum) {
+    throw InvalidTestDataColumnsNumberException(sourceColumnsNum,
+        transformedTestData.header.length);
+  }
+
+  return 0.3;
+}
diff --git a/lib/src/model_selection/cross_validator/_helpers/fromSerializablePredictorJson.dart b/lib/src/model_selection/cross_validator/_helpers/fromSerializablePredictorJson.dart
@@ -0,0 +1,5 @@
+import 'package:ml_algo/src/model_selection/serializable_predictor.dart';
+
+SerializablePredictor fromSerializablePredictorJson(Map<String, dynamic> json) {
+  return null;
+}
diff --git a/lib/src/model_selection/cross_validator/_init_module.dart b/lib/src/model_selection/cross_validator/_init_module.dart
@@ -0,0 +1,5 @@
+import 'package:ml_algo/src/di/common/init_common_module.dart';
+
+void initCrossValidatorModule() {
+  initCommonModule();
+}
diff --git a/lib/src/model_selection/cross_validator/_injector.dart b/lib/src/model_selection/cross_validator/_injector.dart
@@ -0,0 +1,3 @@
+import 'package:injector/injector.dart';
+
+final crossValidatorInjector = Injector();
diff --git a/lib/src/model_selection/cross_validator/cross_validator.dart b/lib/src/model_selection/cross_validator/cross_validator.dart
@@ -1,15 +1,17 @@
+import 'package:ml_algo/src/di/injector.dart';
 import 'package:ml_algo/src/metric/metric_type.dart';
 import 'package:ml_algo/src/model_selection/_init_module.dart';
 import 'package:ml_algo/src/model_selection/_injector.dart';
-import 'package:ml_algo/src/model_selection/assessable.dart';
 import 'package:ml_algo/src/model_selection/cross_validator/cross_validator_impl.dart';
+import 'package:ml_algo/src/model_selection/serializable_predictor.dart';
 import 'package:ml_algo/src/model_selection/split_indices_provider/split_indices_provider_factory.dart';
 import 'package:ml_algo/src/model_selection/split_indices_provider/split_indices_provider_type.dart';
+import 'package:ml_algo/src/service/worker_manager/worker_manager.dart';
 import 'package:ml_dataframe/ml_dataframe.dart';
 import 'package:ml_linalg/dtype.dart';
 import 'package:ml_linalg/linalg.dart';
 
-typedef PredictorFactory = Assessable Function(DataFrame observations);
+typedef PredictorFactoryFn = SerializablePredictor Function(DataFrame observations);
 
 typedef DataPreprocessFn = List<DataFrame> Function(DataFrame trainData,
     DataFrame testData);
@@ -39,11 +41,13 @@ abstract class CrossValidator {
 
     final dataSplitterFactory = modelSelectionInjector
         .get<SplitIndicesProviderFactory>();
+    final workerManager = injector.get<WorkerManager>();
     final dataSplitter = dataSplitterFactory
         .createByType(SplitIndicesProviderType.kFold, numberOfFolds: numberOfFolds);
 
     return CrossValidatorImpl(
       samples,
+      workerManager,
       dataSplitter,
       dtype,
     );
@@ -71,11 +75,13 @@ abstract class CrossValidator {
 
     final dataSplitterFactory = modelSelectionInjector
         .get<SplitIndicesProviderFactory>();
+    final workerManager = injector.get<WorkerManager>();
     final dataSplitter = dataSplitterFactory
         .createByType(SplitIndicesProviderType.lpo, p: p);
 
     return CrossValidatorImpl(
       samples,
+      workerManager,
       dataSplitter,
       dtype,
     );
@@ -133,7 +139,7 @@ abstract class CrossValidator {
   /// print(averageScore);
   /// ````
   Future<Vector> evaluate(
-      PredictorFactory predictorFactory,
+      PredictorFactoryFn predictorFactory,
       MetricType metricType,
       {
         DataPreprocessFn onDataSplit,
diff --git a/lib/src/model_selection/cross_validator/cross_validator_assessing_data_model.dart b/lib/src/model_selection/cross_validator/cross_validator_assessing_data_model.dart
@@ -0,0 +1,34 @@
+import 'package:json_annotation/json_annotation.dart';
+import 'package:ml_algo/src/model_selection/cross_validator/_helpers/fromSerializablePredictorJson.dart';
+import 'package:ml_algo/src/model_selection/cross_validator/cross_validator_assessing_data_model_json_keys.dart';
+import 'package:ml_algo/src/model_selection/serializable_predictor.dart';
+import 'package:ml_dataframe/ml_dataframe.dart';
+
+part 'cross_validator_assessing_data_model.g.dart';
+
+@JsonSerializable()
+class CrossValidatorAssessingDataModel {
+  @JsonKey(
+    name: predictorPrototypeJsonKey,
+    fromJson: fromSerializablePredictorJson,
+  )
+  final SerializablePredictor predictorPrototype;
+
+  @JsonKey(name: samplesJsonKey)
+  final DataFrame samplesPrototype;
+
+  @JsonKey(name: testRowsIndicesJsonKey)
+  final List<int> testRowsIndices;
+
+  CrossValidatorAssessingDataModel(
+    this.predictorPrototype,
+    this.samplesPrototype,
+    this.testRowsIndices,
+  );
+
+  factory CrossValidatorAssessingDataModel.fromJson(Map<String, dynamic> json) =>
+      _$CrossValidatorAssessingDataModelFromJson(json);
+
+  Map<String, dynamic> toJson() =>
+      _$CrossValidatorAssessingDataModelToJson(this);
+}
diff --git a/lib/src/model_selection/cross_validator/cross_validator_assessing_data_model.g.dart b/lib/src/model_selection/cross_validator/cross_validator_assessing_data_model.g.dart
diff --git a/lib/src/model_selection/cross_validator/cross_validator_assessing_data_model_json_keys.dart b/lib/src/model_selection/cross_validator/cross_validator_assessing_data_model_json_keys.dart
@@ -0,0 +1,3 @@
+const predictorPrototypeJsonKey = 'P';
+const samplesJsonKey = 'S';
+const testRowsIndicesJsonKey = 'R';
diff --git a/lib/src/model_selection/cross_validator/cross_validator_impl.dart b/lib/src/model_selection/cross_validator/cross_validator_impl.dart
@@ -1,8 +1,11 @@
 import 'package:ml_algo/src/common/exception/invalid_test_data_columns_number_exception.dart';
 import 'package:ml_algo/src/common/exception/invalid_train_data_columns_number_exception.dart';
 import 'package:ml_algo/src/metric/metric_type.dart';
+import 'package:ml_algo/src/model_selection/cross_validator/_helpers/assess_predictor.dart';
 import 'package:ml_algo/src/model_selection/cross_validator/cross_validator.dart';
+import 'package:ml_algo/src/model_selection/cross_validator/cross_validator_assessing_data_model.dart';
 import 'package:ml_algo/src/model_selection/split_indices_provider/split_indices_provider.dart';
+import 'package:ml_algo/src/service/worker_manager/worker_manager.dart';
 import 'package:ml_dataframe/ml_dataframe.dart';
 import 'package:ml_linalg/dtype.dart';
 import 'package:ml_linalg/matrix.dart';
@@ -12,38 +15,47 @@ import 'package:quiver/iterables.dart';
 class CrossValidatorImpl implements CrossValidator {
   CrossValidatorImpl(
       this.samples,
+      this._workerManager,
       this._splitter,
       this.dtype,
   );
 
   final DataFrame samples;
-  final DType dtype;
+  final WorkerManager _workerManager;
   final SplitIndicesProvider _splitter;
+  final DType dtype;
 
   @override
   Future<Vector> evaluate(
-      PredictorFactory predictorFactory,
+      PredictorFactoryFn predictorFactory,
       MetricType metricType,
       {
         DataPreprocessFn onDataSplit,
       }
-  ) {
+  ) async {
     final samplesAsMatrix = samples.toMatrix(dtype);
     final sourceColumnsNum = samplesAsMatrix.columnsNum;
     final discreteColumns = enumerate(samples.series)
         .where((indexedSeries) => indexedSeries.value.isDiscrete)
         .map((indexedSeries) => indexedSeries.index);
     final allIndicesGroups = _splitter.getIndices(samplesAsMatrix.rowsNum);
+
+    // if onDataSplit callback is null, we can easily do our cross validation
+    // in parallel using several isolates, since isolates don't support
+    // functions as parameters, and that means we can't pass onDataSplit
+    // callback to an isolate
+    if (onDataSplit == null) {
+      return _evaluateInParallel(allIndicesGroups, predictorFactory);
+    }
+
     final scores = allIndicesGroups
         .map((testRowsIndices) {
           final split = _makeSplit(testRowsIndices, discreteColumns);
           final trainDataFrame = split[0];
           final testDataFrame = split[1];
-          final splits = onDataSplit != null
-              ? onDataSplit(trainDataFrame, testDataFrame)
-              : [trainDataFrame, testDataFrame];
-          final transformedTrainData = splits[0];
-          final transformedTestData = splits[1];
+          final transformedData = onDataSplit(trainDataFrame, testDataFrame);
+          final transformedTrainData = transformedData[0];
+          final transformedTestData = transformedData[1];
           final transformedTrainDataColumnsNum = transformedTrainData.header.length;
           final transformedTestDataColumnsNum = transformedTestData.header.length;
 
@@ -65,6 +77,31 @@ class CrossValidatorImpl implements CrossValidator {
     return Future.value(Vector.fromList(scores, dtype: dtype));
   }
 
+  Future<Vector> _evaluateInParallel(
+      Iterable<Iterable<int>> allIndicesGroups,
+      PredictorFactoryFn predictorFactoryFn,
+  ) async {
+    await _workerManager.init();
+
+    final samplesPrototype = samples.sampleFromRows([0]);
+    final predictorPrototype = predictorFactoryFn(samplesPrototype);
+    final scores = await Future.wait(
+        allIndicesGroups.map((testRowsIndices) =>
+            _workerManager
+                .executor
+                .execute(
+                  arg1: CrossValidatorAssessingDataModel(
+                    predictorPrototype,
+                    samplesPrototype,
+                    testRowsIndices
+                  ).toJson(),
+                  fun1: assessPredictor,
+                ),
+        ));
+
+    return Future.value(Vector.fromList(scores, dtype: dtype));
+  }
+
   List<DataFrame> _makeSplit(Iterable<int> testRowsIndices,
       Iterable<int> discreteColumns) {
     final samplesAsMatrix = samples.toMatrix(dtype);
diff --git a/lib/src/model_selection/serializable_predictor.dart b/lib/src/model_selection/serializable_predictor.dart
@@ -0,0 +1,8 @@
+import 'package:ml_algo/src/common/serializable/serializable.dart';
+import 'package:ml_algo/src/model_selection/assessable.dart';
+import 'package:ml_algo/src/predictor/predictor.dart';
+
+abstract class SerializablePredictor implements
+    Assessable,
+    Serializable,
+    Predictor {}
diff --git a/lib/src/service/worker_manager/worker_manager.dart b/lib/src/service/worker_manager/worker_manager.dart
@@ -0,0 +1,6 @@
+import 'package:worker_manager/worker_manager.dart';
+
+abstract class WorkerManager {
+  Future<void> init();
+  Executor get executor;
+}
diff --git a/lib/src/service/worker_manager/worker_manager_impl.dart b/lib/src/service/worker_manager/worker_manager_impl.dart
@@ -0,0 +1,19 @@
+import 'package:ml_algo/src/service/worker_manager/worker_manager.dart';
+import 'package:worker_manager/src/executor.dart';
+
+class WorkerManagerImpl implements WorkerManager {
+  bool _isInitialized = false;
+
+  @override
+  Executor get executor => Executor();
+
+  @override
+  Future<void> init() async {
+    if (_isInitialized) {
+      return;
+    }
+
+    await Executor().warmUp();
+    _isInitialized = true;
+  }
+}
diff --git a/pubspec.yaml b/pubspec.yaml
@@ -14,6 +14,7 @@ dependencies:
   ml_linalg: ^12.17.8
   ml_preprocessing: ^5.2.1
   quiver: ^2.0.2
+  worker_manager: ^3.1.8
   xrange: ^0.0.8
 
 dev_dependencies:
diff --git a/test/mocks.dart b/test/mocks.dart
diff --git a/test/model_selection/cross_validator/cross_validator_impl_test.dart b/test/model_selection/cross_validator/cross_validator_impl_test.dart

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+import 'package:injector/injector.dart';`
	`2`	`+`
	`3`	`+final crossValidatorInjector = Injector();`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	`+const predictorPrototypeJsonKey = 'P';`
	`2`	`+const samplesJsonKey = 'S';`
	`3`	`+const testRowsIndicesJsonKey = 'R';`