docs: public final DeepLearningModel trainModel(DeepLearningModel model) {
      Frame validScoreFrame = null;
      Frame train, trainScoreFrame;
      try {
        lock_data();
        if (checkpoint == null && !quiet_mode) logStart(); //if checkpoint is given, some Job's params might be uninitialized (but the restarted model's parameters are correct)
        if (model == null) {
          model = UKV.get(dest());
        }
        model.write_lock(self());
        final DeepLearning mp = model.model_info().get_params(); //use the model's parameters for everything below - NOT the job's parameters (can be different after checkpoint restart)
  
        prepareValidationWithModel(model);
        final long model_size = model.model_info().size();
        if (!quiet_mode) Log.info("Number of model parameters (weights/biases): " + String.format("%,d", model_size));
        train = model.model_info().data_info()._adaptedFrame;
        if (mp.force_load_balance) train = updateFrame(train, reBalance(train, mp.replicate_training_data));
        if (mp.classification && mp.balance_classes) {
          float[] trainSamplingFactors = new float[train.lastVec().domain().length]; //leave initialized to 0 -> will be filled up below
          if (class_sampling_factors != null) {
            if (class_sampling_factors.length != train.lastVec().domain().length)
              throw new IllegalArgumentException("class_sampling_factors must have " + train.lastVec().domain().length + " elements");
            trainSamplingFactors = class_sampling_factors.clone(); //clone: don't modify the original
          }
          train = updateFrame(train, sampleFrameStratified(
                  train, train.lastVec(), trainSamplingFactors, (long)(mp.max_after_balance_size*train.numRows()), mp.seed, true, false));
          model.setModelClassDistribution(new MRUtils.ClassDist(train.lastVec()).doAll(train.lastVec()).rel_dist());
        }
        model.training_rows = train.numRows();
        trainScoreFrame = updateFrame(train, sampleFrame(train, mp.score_training_samples, mp.seed)); //training scoring dataset is always sampled uniformly from the training dataset
  
        if (!quiet_mode) Log.info("Number of chunks of the training data: " + train.anyVec().nChunks());
        if (validation != null) {
          model.validation_rows = validation.numRows();
          Frame adaptedValid = getValidation();
          if (getValidAdaptor().needsAdaptation2CM()) {
            adaptedValid.add(getValidAdaptor().adaptedValidationResponse(_responseName), getValidAdaptor().getAdaptedValidationResponse2CM());
          }
          // validation scoring dataset can be sampled in multiple ways from the given validation dataset
          if (mp.classification && mp.balance_classes && mp.score_validation_sampling == ClassSamplingMethod.Stratified) {
            validScoreFrame = updateFrame(adaptedValid, sampleFrameStratified(adaptedValid, adaptedValid.lastVec(), null,
                    mp.score_validation_samples > 0 ? mp.score_validation_samples : adaptedValid.numRows(), mp.seed+1, false /* no oversampling */, false));
          } else {
            validScoreFrame = updateFrame(adaptedValid, sampleFrame(adaptedValid, mp.score_validation_samples, mp.seed+1));
          }
          if (mp.force_load_balance) validScoreFrame = updateFrame(validScoreFrame, reBalance(validScoreFrame, false /*always split up globally since scoring should be distributed*/));
          if (!quiet_mode) Log.info("Number of chunks of the validation data: " + validScoreFrame.anyVec().nChunks());
        }
  
        // Set train_samples_per_iteration size (cannot be done earlier since this depends on whether stratified sampling is done)
        model.actual_train_samples_per_iteration = computeTrainSamplesPerIteration(mp, train.numRows(), model);
        // Determine whether shuffling is enforced
        if(mp.replicate_training_data && (model.actual_train_samples_per_iteration == train.numRows()*(mp.single_node_mode?1:H2O.CLOUD.size())) && !mp.shuffle_training_data && H2O.CLOUD.size() > 1 && !mp.reproducible) {
          Log.warn("Enabling training data shuffling, because all nodes train on the full dataset (replicated training data).");
          mp.shuffle_training_data = true;
        }
  
        model._timeLastScoreEnter = System.currentTimeMillis(); //to keep track of time per iteration, must be called before first call to doScoring
  
  
        if (!mp.quiet_mode) Log.info("Initial model:\n" + model.model_info());
        if (autoencoder) model.doScoring(train, trainScoreFrame, validScoreFrame, self(), getValidAdaptor()); //get the null model reconstruction error
        // put the initial version of the model into DKV
        model.update(self());
        Log.info("Starting to train the Deep Learning model.");
  
        //main loop
        do model.set_model_info(H2O.CLOUD.size() > 1 && mp.replicate_training_data ? ( mp.single_node_mode ?
                new DeepLearningTask2(train, model.model_info(), rowFraction(train, mp, model)).invoke(Key.make()).model_info() : //replicated data + single node mode
                new DeepLearningTask2(train, model.model_info(), rowFraction(train, mp, model)).invokeOnAllNodes().model_info() ) : //replicated data + multi-node mode
                new DeepLearningTask(model.model_info(), rowFraction(train, mp, model)).doAll(train).model_info()); //distributed data (always in multi-node mode)
        while (model.doScoring(train, trainScoreFrame, validScoreFrame, self(), getValidAdaptor()));
  
        // replace the model with the best model so far (if it's better)
        if (!isCancelledOrCrashed() && override_with_best_model && model.actual_best_model_key != null && n_folds == 0) {
          DeepLearningModel best_model = UKV.get(model.actual_best_model_key);
          if (best_model != null && best_model.error() < model.error() && Arrays.equals(best_model.model_info().units, model.model_info().units)) {
            Log.info("Setting the model to be the best model so far (based on scoring history).");
            DeepLearningModel.DeepLearningModelInfo mi = best_model.model_info().deep_clone();
            // Don't cheat - count full amount of training samples, since that's the amount of training it took to train (without finding anything better)
            mi.set_processed_global(model.model_info().get_processed_global());
            mi.set_processed_local(model.model_info().get_processed_local());
            model.set_model_info(mi);
            model.update(self());
            model.doScoring(train, trainScoreFrame, validScoreFrame, self(), getValidAdaptor());
            assert(best_model.error() == model.error());
          }
        }
  
        Log.info(model);
        Log.info("Finished training the Deep Learning model.");
        return model;
      }
      catch(JobCancelledException ex) {
        model = UKV.get(dest());
        state = JobState.CANCELLED; //for JSON REST response
        model.get_params().state = state; //for parameter JSON on the HTML page
        Log.info("Deep Learning model building was cancelled.");
        return model;
      }
      catch(Throwable t) {
        t.printStackTrace();
        model = UKV.get(dest());
        state = JobState.FAILED; //for JSON REST response
        if (model != null) {
          model.get_params().state = state; //for parameter JSON on the HTML page
          Log.info("Deep Learning model building failed.");
        }
        return model;
      }
      finally {
        if (model != null && DKV.get(model._key) != null) model.unlock(self());
        unlock_data();
      }
    }