IGNITE-7350: Distributed MLP cleanup/refactoring

this closes #3368

examples/src/main/ml/org/apache/ignite/examples/ml/nn/MLPGroupTrainerExample.java

@@ -35,7 +35,7 @@
 import org.apache.ignite.ml.nn.architecture.MLPArchitecture;
 import org.apache.ignite.ml.nn.initializers.RandomInitializer;
 import org.apache.ignite.ml.nn.trainers.distributed.MLPGroupUpdateTrainer;
-import org.apache.ignite.ml.nn.updaters.RPropParameterUpdate;
+import org.apache.ignite.ml.optimization.updatecalculators.RPropParameterUpdate;
 import org.apache.ignite.ml.structures.LabeledVector;
 import org.apache.ignite.thread.IgniteThread;
 

examples/src/main/ml/org/apache/ignite/examples/ml/nn/MLPLocalTrainerExample.java

@@ -25,13 +25,13 @@
 import org.apache.ignite.ml.math.functions.IgniteSupplier;
 import org.apache.ignite.ml.math.impls.matrix.DenseLocalOnHeapMatrix;
 import org.apache.ignite.ml.nn.Activators;
-import org.apache.ignite.ml.nn.LocalBatchTrainerInput;
-import org.apache.ignite.ml.nn.LossFunctions;
+import org.apache.ignite.ml.trainers.local.LocalBatchTrainerInput;
+import org.apache.ignite.ml.optimization.LossFunctions;
 import org.apache.ignite.ml.nn.MultilayerPerceptron;
 import org.apache.ignite.ml.nn.architecture.MLPArchitecture;
 import org.apache.ignite.ml.nn.initializers.RandomInitializer;
 import org.apache.ignite.ml.nn.trainers.local.MLPLocalBatchTrainer;
-import org.apache.ignite.ml.nn.updaters.RPropUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.RPropUpdateCalculator;
 import org.apache.ignite.ml.util.Utils;
 
 /**

modules/ml/src/main/java/org/apache/ignite/ml/nn/MLPGroupUpdateTrainerCacheInput.java

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Random;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.ignite.Ignite;
@@ -56,6 +57,11 @@ public class MLPGroupUpdateTrainerCacheInput extends AbstractMLPGroupUpdateTrain
      */
     private final MultilayerPerceptron mlp;
 
+    /**
+     * Random number generator.
+     */
+    private final Random rand;
+
     /**
      * Construct instance of this class with given parameters.
      *
@@ -64,15 +70,32 @@ public class MLPGroupUpdateTrainerCacheInput extends AbstractMLPGroupUpdateTrain
      * @param networksCnt Count of networks to be trained in parallel by {@link MLPGroupUpdateTrainer}.
      * @param cache Cache with labeled vectors.
      * @param batchSize Size of batch to return on each training iteration.
+     * @param rand RNG.
      */
     public MLPGroupUpdateTrainerCacheInput(MLPArchitecture arch, MLPInitializer init,
         int networksCnt, IgniteCache<Integer, LabeledVector<Vector, Vector>> cache,
-        int batchSize) {
+        int batchSize, Random rand) {
         super(networksCnt);
 
         this.batchSize = batchSize;
         this.cache = cache;
         this.mlp = new MultilayerPerceptron(arch, init);
+        this.rand = rand;
+    }
+
+    /**
+     * Construct instance of this class with given parameters.
+     *
+     * @param arch Architecture of multilayer perceptron.
+     * @param init Initializer of multilayer perceptron.
+     * @param networksCnt Count of networks to be trained in parallel by {@link MLPGroupUpdateTrainer}.
+     * @param cache Cache with labeled vectors.
+     * @param batchSize Size of batch to return on each training iteration.
+     */
+    public MLPGroupUpdateTrainerCacheInput(MLPArchitecture arch, MLPInitializer init,
+        int networksCnt, IgniteCache<Integer, LabeledVector<Vector, Vector>> cache,
+        int batchSize) {
+        this(arch, init, networksCnt, cache, batchSize, new Random());
     }
 
     /**
@@ -93,6 +116,7 @@ public MLPGroupUpdateTrainerCacheInput(MLPArchitecture arch, int networksCnt,
     @Override public IgniteSupplier<IgniteBiTuple<Matrix, Matrix>> batchSupplier() {
         String cName = cache.getName();
         int bs = batchSize;
+        Random r = rand; // IMPL NOTE this is intended to make below lambda more lightweight.
 
         return () -> {
             Ignite ignite = Ignition.localIgnite();
@@ -105,7 +129,7 @@ public MLPGroupUpdateTrainerCacheInput(MLPArchitecture arch, int networksCnt,
 
             int locKeysCnt = keys.size();
 
-            int[] selected = Utils.selectKDistinct(locKeysCnt, Math.min(bs, locKeysCnt));
+            int[] selected = Utils.selectKDistinct(locKeysCnt, Math.min(bs, locKeysCnt), r);
 
             // Get dimensions of vectors in cache. We suppose that every feature vector has
             // same dimension d 1 and every label has the same dimension d2.

modules/ml/src/main/java/org/apache/ignite/ml/nn/MultilayerPerceptron.java

@@ -35,7 +35,7 @@
 import org.apache.ignite.ml.nn.architecture.TransformationLayerArchitecture;
 import org.apache.ignite.ml.nn.initializers.MLPInitializer;
 import org.apache.ignite.ml.nn.initializers.RandomInitializer;
-import org.apache.ignite.ml.nn.updaters.SmoothParametrized;
+import org.apache.ignite.ml.optimization.SmoothParametrized;
 
 import static org.apache.ignite.ml.math.util.MatrixUtil.elementWiseTimes;
 

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/distributed/AbstractMLPGroupUpdateTrainerInput.java

@@ -20,7 +20,7 @@
 import java.util.UUID;
 import java.util.stream.Stream;
 import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.nn.LocalBatchTrainerInput;
+import org.apache.ignite.ml.trainers.local.LocalBatchTrainerInput;
 import org.apache.ignite.ml.nn.MultilayerPerceptron;
 import org.apache.ignite.ml.trainers.group.GroupTrainerCacheKey;
 import org.apache.ignite.ml.trainers.group.GroupTrainerInput;

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/distributed/MLPGroupUpdateTrainer.java

@@ -33,11 +33,11 @@
 import org.apache.ignite.ml.math.functions.IgniteFunction;
 import org.apache.ignite.ml.math.functions.IgniteSupplier;
 import org.apache.ignite.ml.math.util.MatrixUtil;
-import org.apache.ignite.ml.nn.LossFunctions;
+import org.apache.ignite.ml.optimization.LossFunctions;
 import org.apache.ignite.ml.nn.MultilayerPerceptron;
-import org.apache.ignite.ml.nn.updaters.ParameterUpdateCalculator;
-import org.apache.ignite.ml.nn.updaters.RPropParameterUpdate;
-import org.apache.ignite.ml.nn.updaters.RPropUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.ParameterUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.RPropParameterUpdate;
+import org.apache.ignite.ml.optimization.updatecalculators.RPropUpdateCalculator;
 import org.apache.ignite.ml.trainers.group.GroupTrainerCacheKey;
 import org.apache.ignite.ml.trainers.group.MetaoptimizerGroupTrainer;
 import org.apache.ignite.ml.trainers.group.ResultAndUpdates;
@@ -227,8 +227,7 @@ MLPGroupUpdateTrainingContext<U>>, MLPGroupUpdateTrainingLoopData<U>> trainingLo
         UUID uuid = ctx.trainingUUID();
 
         return () -> {
-            MLPGroupUpdateTrainingData<U> data = MLPGroupUpdateTrainerDataCache
-                .getOrCreate(Ignition.localIgnite()).get(uuid);
+            MLPGroupUpdateTrainingData<U> data = MLPGroupUpdateTrainerDataCache.getOrCreate(Ignition.localIgnite()).get(uuid);
             return new MLPGroupUpdateTrainingContext<>(data, prevUpdate);
         };
     }

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/distributed/MLPGroupUpdateTrainerDataCache.java

@@ -55,7 +55,7 @@ public static IgniteCache<UUID, MLPGroupUpdateTrainingData> getOrCreate(Ignite i
         CacheConfiguration<UUID, MLPGroupUpdateTrainingData> cfg = new CacheConfiguration<>();
 
         // Write to primary.
-        cfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.PRIMARY_SYNC);
+        cfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
 
         // Atomic transactions only.
         cfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/distributed/MLPGroupUpdateTrainingData.java

@@ -25,21 +25,36 @@
 import org.apache.ignite.ml.math.functions.IgniteFunction;
 import org.apache.ignite.ml.math.functions.IgniteSupplier;
 import org.apache.ignite.ml.nn.MultilayerPerceptron;
-import org.apache.ignite.ml.nn.updaters.ParameterUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.ParameterUpdateCalculator;
 
 /** Multilayer perceptron group update training data. */
 public class MLPGroupUpdateTrainingData<U> {
-    /** */
+    /** {@link ParameterUpdateCalculator}. */
     private final ParameterUpdateCalculator<MultilayerPerceptron, U> updateCalculator;
-    /** */
+
+    /**
+     * Count of steps which should be done by each of parallel trainings before sending it's update for combining with
+     * other parallel trainings updates.
+     */
     private final int stepsCnt;
-    /** */
+
+    /**
+     * Function used to reduce updates in one training (for example, sum all sequential gradient updates to get one
+     * gradient update).
+     */
     private final IgniteFunction<List<U>, U> updateReducer;
-    /** */
+
+    /**
+     * Supplier of batches in the form of (inputs, groundTruths).
+     */
     private final IgniteSupplier<IgniteBiTuple<Matrix, Matrix>> batchSupplier;
-    /** */
+
+    /**
+     * Loss function.
+     */
     private final IgniteFunction<Vector, IgniteDifferentiableVectorToDoubleFunction> loss;
-    /** */
+
+    /** Error tolerance. */
     private final double tolerance;
 
     /** Construct multilayer perceptron group update training data with all parameters provided. */

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/distributed/MLPGroupUpdateTrainingLoopData.java

@@ -26,29 +26,39 @@
 import org.apache.ignite.ml.math.functions.IgniteFunction;
 import org.apache.ignite.ml.math.functions.IgniteSupplier;
 import org.apache.ignite.ml.nn.MultilayerPerceptron;
-import org.apache.ignite.ml.nn.updaters.ParameterUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.ParameterUpdateCalculator;
 import org.apache.ignite.ml.trainers.group.GroupTrainerCacheKey;
 
 /** Multilayer perceptron group update training loop data. */
 public class MLPGroupUpdateTrainingLoopData<P> implements Serializable {
-    /** */
+    /** {@link ParameterUpdateCalculator}. */
     private final ParameterUpdateCalculator<MultilayerPerceptron, P> updateCalculator;
-    /** */
+
+    /**
+     * Count of steps which should be done by each of parallel trainings before sending it's update for combining with
+     * other parallel trainings updates.
+     */
     private final int stepsCnt;
-    /** */
+
+    /** Function used to reduce updates of all steps of given parallel training. */
     private final IgniteFunction<List<P>, P> updateReducer;
-    /** */
+
+    /** Previous update. */
     private final P previousUpdate;
-    /** */
+
+    /** Supplier of batches. */
     private final IgniteSupplier<IgniteBiTuple<Matrix, Matrix>> batchSupplier;
-    /** */
+
+    /** Loss function. */
     private final IgniteFunction<Vector, IgniteDifferentiableVectorToDoubleFunction> loss;
-    /** */
+
+    /** Error tolerance. */
     private final double tolerance;
 
-    /** */
+    /** Key. */
     private final GroupTrainerCacheKey<Void> key;
-    /** */
+
+    /** MLP. */
     private final MultilayerPerceptron mlp;
 
     /** Create multilayer perceptron group update training loop data. */

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/distributed/MLPMetaoptimizer.java

@@ -26,7 +26,7 @@
 /** Meta-optimizer for multilayer perceptron. */
 public class MLPMetaoptimizer<P> implements Metaoptimizer<MLPGroupUpdateTrainerLocalContext,
     MLPGroupUpdateTrainingLoopData<P>, P, P, P, ArrayList<P>> {
-    /** */
+    /** Function used for reducing updates produced by parallel trainings. */
     private final IgniteFunction<List<P>, P> allUpdatesReducer;
 
     /** Construct metaoptimizer. */

modules/ml/src/main/java/org/apache/ignite/ml/nn/trainers/local/MLPLocalBatchTrainer.java

@@ -21,11 +21,12 @@
 import org.apache.ignite.ml.math.functions.IgniteDifferentiableVectorToDoubleFunction;
 import org.apache.ignite.ml.math.functions.IgniteFunction;
 import org.apache.ignite.ml.math.functions.IgniteSupplier;
-import org.apache.ignite.ml.nn.LossFunctions;
+import org.apache.ignite.ml.optimization.LossFunctions;
 import org.apache.ignite.ml.nn.MultilayerPerceptron;
-import org.apache.ignite.ml.nn.updaters.ParameterUpdateCalculator;
-import org.apache.ignite.ml.nn.updaters.RPropParameterUpdate;
-import org.apache.ignite.ml.nn.updaters.RPropUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.ParameterUpdateCalculator;
+import org.apache.ignite.ml.optimization.updatecalculators.RPropParameterUpdate;
+import org.apache.ignite.ml.optimization.updatecalculators.RPropUpdateCalculator;
+import org.apache.ignite.ml.trainers.local.LocalBatchTrainer;
 
 /**
  * Local batch trainer for MLP.

modules/ml/src/main/java/org/apache/ignite/ml/optimization/BarzilaiBorweinUpdater.java

@@ -34,8 +34,11 @@ public class BarzilaiBorweinUpdater implements Updater {
     /**
      * {@inheritDoc}
      */
-    @Override public Vector compute(Vector oldWeights, Vector oldGradient, Vector weights, Vector gradient, int iteration) {
-        double learningRate = computeLearningRate(oldWeights != null ? oldWeights.copy() : null, oldGradient != null ? oldGradient.copy() : null, weights.copy(), gradient.copy());
+    @Override public Vector compute(Vector oldWeights, Vector oldGradient, Vector weights, Vector gradient,
+        int iteration) {
+        double learningRate = computeLearningRate(oldWeights != null ? oldWeights.copy() : null,
+            oldGradient != null ? oldGradient.copy() : null, weights.copy(), gradient.copy());
+
         return weights.copy().minus(gradient.copy().times(learningRate));
     }
 
@@ -45,6 +48,7 @@ private double computeLearningRate(Vector oldWeights, Vector oldGradient, Vector
             return INITIAL_LEARNING_RATE;
         else {
             Vector gradientDiff = gradient.minus(oldGradient);
+
             return weights.minus(oldWeights).dot(gradientDiff) / Math.pow(gradientDiff.kNorm(2.0), 2.0);
         }
     }

modules/ml/src/main/java/org/apache/ignite/ml/optimization/BaseParametrized.java

@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.ml.optimization;
+
+import org.apache.ignite.ml.math.Vector;
+import org.apache.ignite.ml.util.Utils;
+
+/**
+ * Base interface for parametrized models.
+ *
+ * @param <M> Model class.
+ */
+interface BaseParametrized<M extends BaseParametrized<M>> {
+    /**
+     * Get parameters vector.
+     *
+     * @return Parameters vector.
+     */
+    Vector parameters();
+
+    /**
+     * Set parameters.
+     *
+     * @param vector Parameters vector.
+     */
+    M setParameters(Vector vector);
+
+    /**
+     * Return new model with given parameters vector.
+     *
+     * @param vector Parameters vector.
+     */
+    default M withParameters(Vector vector) {
+        return Utils.copy(this).setParameters(vector);
+    }
+
+    /**
+     * Get count of parameters of this model.
+     *
+     * @return Count of parameters of this model.
+     */
+    int parametersCount();
+}

modules/ml/src/main/java/org/apache/ignite/ml/optimization/GradientDescent.java

@@ -141,6 +141,7 @@ private Vector calculateDistributedGradient(SparseDistributedMatrix data, Vector
                         cnt++;
                     }
                 }
+
                 return resGradient.divide(cnt);
             },
             weights);
@@ -187,7 +188,7 @@ private Matrix extractInputs(Matrix data) {
     /** Makes carrying of the gradient function and fixes data matrix. */
     private IgniteFunction<Vector, Vector> getLossGradientFunction(Matrix data) {
         if (data instanceof SparseDistributedMatrix) {
-            SparseDistributedMatrix distributedMatrix = (SparseDistributedMatrix) data;
+            SparseDistributedMatrix distributedMatrix = (SparseDistributedMatrix)data;
 
             if (distributedMatrix.getStorage().storageMode() == StorageConstants.ROW_STORAGE_MODE)
                 return weights -> calculateDistributedGradient(distributedMatrix, weights);

modules/ml/src/main/java/org/apache/ignite/ml/optimization/GradientFunction.java

@@ -27,5 +27,5 @@
 @FunctionalInterface
 public interface GradientFunction extends Serializable {
     /** */
-    Vector compute(Matrix inputs, Vector groundTruth, Vector point);
+    Vector compute(Matrix inputs, Vector groundTruth, Vector pnt);
 }

modules/ml/src/main/java/org/apache/ignite/ml/nn/LossFunctions.java → modules/ml/src/main/java/org/apache/ignite/ml/optimization/LossFunctions.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.ignite.ml.nn;
+package org.apache.ignite.ml.optimization;
 
 import org.apache.ignite.ml.math.Vector;
 import org.apache.ignite.ml.math.functions.IgniteDifferentiableVectorToDoubleFunction;