<commit_before>/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve.

Licensed 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. */

#include "Trainer.h"

#include <fenv.h>
#include <stdio.h>

#include <iomanip>
#include <iostream>
#include <limits>
#include <sstream>

#include <google/protobuf/text_format.h>

#include "paddle/utils/Excepts.h"
#include "paddle/utils/GlobalConstants.h"
#include "paddle/utils/PythonUtil.h"
#include "paddle/utils/Stat.h"
#include "paddle/utils/Util.h"

#include "RemoteParameterUpdater.h"
#include "TesterConfig.h"
#include "ThreadParameterUpdater.h"
#include "TrainerConfigHelper.h"
#include "paddle/gserver/gradientmachines/GradientMachineMode.h"
#include "paddle/gserver/gradientmachines/NeuralNetwork.h"
#include "paddle/gserver/layers/ValidationLayer.h"

DEFINE_string(config, "", "Trainer config file");

DEFINE_int32(test_period,
             0,
             "if equal 0, do test on all test data at the end of "
             "each pass. While if equal non-zero, do test on all test "
             "data every test_period batches");
DEFINE_bool(test_all_data_in_one_period,
            false,
            "This option was deprecated, since we will always do "
            "test on all test set ");

DEFINE_bool(local, true, "Train in local mode or not");

DEFINE_int32(average_test_period,
             0,
             "Do test on average parameter every so"
             " many batches. MUST be devided by FLAGS_log_period."
             " Default 0 means do not test average parameter");

DEFINE_int32(saving_period, 1, "Save parameteres every so many passes");
DEFINE_int64(saving_period_by_batches,
             0,
             "Save parameters every so many batches in one pass");
DEFINE_string(save_dir, "", "Directory for saving model parameter");
DEFINE_int32(start_pass,
             0,
             "Start training from this pass. "
             "Will load parameter from the previous pass");
DEFINE_int32(test_pass, -1, "Will load parameter start from this pass to test");
DEFINE_int32(test_wait, 0, "Waiting for pass parameter if not exist");
DEFINE_bool(with_cost, true, "enable cost layer or not");
DEFINE_bool(distribute_test, false, "test in distribute mode");

DEFINE_int32(num_passes, 100, "train for so many passes");

DEFINE_string(config_args,
              "",
              "arguments passed to config file."
              "Format: key1=value1,key2=value2");

DEFINE_bool(save_only_one,
            false,
            "Save only parameters in last pass, remove previous.");

DEFINE_string(feat_file, "", "File name of extracted feature.");
DEFINE_string(predict_output_dir,
              "",
              "Directory that saves the predicted results of output layers");
DEFINE_string(model_list, "", "File that saves the model list when evaluation");

namespace paddle {

void Trainer::init(int argc, char** argv) {
  initMain(argc, argv);
  initPython(argc, argv);

  auto config = TrainerConfigHelper::createFromFlagConfig();
  feenableexcept(FE_INVALID | FE_DIVBYZERO | FE_OVERFLOW);

  init(config);
}

void Trainer::init(const std::shared_ptr<TrainerConfigHelper>& config,
                   bool testing,
                   const std::shared_ptr<GradientMachine>& gradientMachine,
                   const std::shared_ptr<DataProvider>& dataProvider,
                   const std::shared_ptr<DataProvider>& testDataProvider) {
  this->stats_ = std::make_shared<TrainerStats>();

  config_ = config;

  config_->updateConfigFromFlags();

  testing_ = testing;

  // in testing, mode_ may GradientMachine::kTesting or
  // GradientMachine::kSgdSparseCpuTraining

  if (FLAGS_local) {
    CHECK(!FLAGS_loadsave_parameters_in_pserver)
        << "local and loadsave_parameters_in_pserver can not both true";
    if (config_->getOptConfig().use_sparse_remote_updater()) {
      config_->disableRemoteSparseUpdaterForEachParams();
      LOG(INFO) << "ignore sparse_remote_update=true due to  --local=true";
    }
  }
  if (FLAGS_loadsave_parameters_in_pserver) {
    CHECK(config_->getOptConfig().use_sparse_remote_updater())
        << "no parameter to load from pserver, please check network config";
  }
  if (testing && !FLAGS_loadsave_parameters_in_pserver) {
    if (config_->getOptConfig().use_sparse_remote_updater()) {
      config_->disableRemoteSparseUpdater();
      LOG(INFO) << "because parameter is loaded local,"
                << "tester ignore sparse_remote_update flag";
    }
  }

  CHECK(TrainAlgorithm::isValid(config_->getOptConfig().algorithm()))
      << "invalid algorithm configuration: "
      << config_->getOptConfig().algorithm();

  bool useSparseUpdater = false;
  for (auto& paraConfig : config_->getModelConfig().parameters()) {
    if (paraConfig.sparse_update() || paraConfig.sparse_remote_update()) {
      useSparseUpdater = true;
    }
  }

  if (testing) {
    LOG(INFO) << "trainer: in testing mode";
    if (config_->getOptConfig().use_sparse_remote_updater() ||
        FLAGS_trainer_count > 1) {
      mode_ = GradientMachine::kSgdSparseCpuTraining;
      LOG(INFO) << "trainer mode: SgdSparseCpuTraining";
    } else {
      mode_ = GradientMachine::kTesting;
      LOG(INFO) << "trainer mode: Testing";
    }
  } else if (IGradientMachineMode::tryGetMode(
                 (int*)&mode_,
                 config_->getOptConfig().algorithm(),
                 FLAGS_trainer_count,
                 FLAGS_local,
                 FLAGS_use_gpu)) {
    LOG(INFO) << "Custom trainer mode.";
  } else if ((config_->getOptConfig().algorithm() == TrainAlgorithm::SGD ||
              config_->getOptConfig().algorithm() ==
                  TrainAlgorithm::AsyncSGD) &&
             useSparseUpdater) {
    mode_ = GradientMachine::kSgdSparseCpuTraining;
    LOG(INFO) << "trainer mode: SgdSparseCpuTraining";
  } else {
    mode_ = GradientMachine::kNormal;
    LOG(INFO) << "trainer mode: Normal";
  }

  // initialize trainer internal
  trainerInternal_.init(config_,
                        gradientMachine,
                        TrainerInternalConfig::createFromMode(mode_),
                        stats_,
                        testing);
  std::unique_ptr<ParameterUtilConfig> paramConfig(
      new ParameterUtilConfig(FLAGS_save_only_one,
                              FLAGS_saving_period,
                              FLAGS_loadsave_parameters_in_pserver,
                              FLAGS_config));

  paramUtil_.reset(
      new paddle::ParameterUtil(config_,
                                std::move(paramConfig),
                                trainerInternal_.getGradientMachine(),
                                trainerInternal_.getParameterUpdater()));

  bool gpuData =
      FLAGS_use_gpu && (!FLAGS_parallel_nn) &&
      (!IGradientMachineMode::dataMustInCpu(mode_, FLAGS_trainer_count));

  dataProvider_ = dataProvider;
  if (!dataProvider_ && config_->hasDataConfig() && !testing_) {
    dataProvider_.reset(DataProvider::create(*config_, *config_, gpuData));
  }
  if (!testDataProvider_) {
    // No evaluator_ if there is testDataProvider but no dataProvider.
    evaluator_.reset(trainerInternal_.getGradientMachine()->makeEvaluator());
    currentEvaluator_.reset(
        trainerInternal_.getGradientMachine()->makeEvaluator());
    if (FLAGS_average_test_period > 0 && FLAGS_trainer_id == 0 &&
        config_->getOptConfig().average_window() > 0) {
      CHECK_EQ(FLAGS_average_test_period % FLAGS_log_period, 0)
          << "FLAGS_average_test_period must be divided by FALGS_log_period";
      averageEvaluator_.reset(
          trainerInternal_.getGradientMachine()->makeEvaluator());
    }
  }

  testDataProvider_ = testDataProvider;
  if (!testDataProvider_ && config_->hasTestDataConfig()) {
    testDataProvider_.reset(
        DataProvider::create(config_->getTestDataConfig(), *config_, gpuData));
  }
  if (testDataProvider_) {
    createTester();
  }

  if (!testing &&
      (trainerInternal_.getGradientMachine()->hasStaticParameters())) {
    CHECK(!FLAGS_loadsave_parameters_in_pserver)
        << "is_static and loadsave_parameters_in_pserver can not both true";
  }
  if (testing) {
    // will load per pass for tester
  } else if (paramUtil_->tryLoadParametersFromConfig()) {
    // load from config already.
  } else {
    trainerInternal_.getGradientMachine()->randParameters();
  }

  // Only non static parameters need to be updated
  std::vector<ParameterPtr>& parameters =
      trainerInternal_.getGradientMachine()->getNonStaticParameters();
  if (trainerInternal_.getParameterUpdater()) {
    trainerInternal_.getParameterUpdater()->init(parameters);<commit_msg>fix bug: if test and sparse_remote_update can not co-exsit, crash trainer if necessary<commit_after>/* Copyright (c) 2016 PaddlePaddle Authors. All Rights Reserve.

Licensed 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. */

#include "Trainer.h"

#include <fenv.h>
#include <stdio.h>

#include <iomanip>
#include <iostream>
#include <limits>
#include <sstream>

#include <google/protobuf/text_format.h>

#include "paddle/utils/Excepts.h"
#include "paddle/utils/GlobalConstants.h"
#include "paddle/utils/PythonUtil.h"
#include "paddle/utils/Stat.h"
#include "paddle/utils/Util.h"

#include "RemoteParameterUpdater.h"
#include "TesterConfig.h"
#include "ThreadParameterUpdater.h"
#include "TrainerConfigHelper.h"
#include "paddle/gserver/gradientmachines/GradientMachineMode.h"
#include "paddle/gserver/gradientmachines/NeuralNetwork.h"
#include "paddle/gserver/layers/ValidationLayer.h"

DEFINE_string(config, "", "Trainer config file");

DEFINE_int32(test_period,
             0,
             "if equal 0, do test on all test data at the end of "
             "each pass. While if equal non-zero, do test on all test "
             "data every test_period batches");
DEFINE_bool(test_all_data_in_one_period,
            false,
            "This option was deprecated, since we will always do "
            "test on all test set ");

DEFINE_bool(local, true, "Train in local mode or not");

DEFINE_int32(average_test_period,
             0,
             "Do test on average parameter every so"
             " many batches. MUST be devided by FLAGS_log_period."
             " Default 0 means do not test average parameter");

DEFINE_int32(saving_period, 1, "Save parameteres every so many passes");
DEFINE_int64(saving_period_by_batches,
             0,
             "Save parameters every so many batches in one pass");
DEFINE_string(save_dir, "", "Directory for saving model parameter");
DEFINE_int32(start_pass,
             0,
             "Start training from this pass. "
             "Will load parameter from the previous pass");
DEFINE_int32(test_pass, -1, "Will load parameter start from this pass to test");
DEFINE_int32(test_wait, 0, "Waiting for pass parameter if not exist");
DEFINE_bool(with_cost, true, "enable cost layer or not");
DEFINE_bool(distribute_test, false, "test in distribute mode");

DEFINE_int32(num_passes, 100, "train for so many passes");

DEFINE_string(config_args,
              "",
              "arguments passed to config file."
              "Format: key1=value1,key2=value2");

DEFINE_bool(save_only_one,
            false,
            "Save only parameters in last pass, remove previous.");

DEFINE_string(feat_file, "", "File name of extracted feature.");
DEFINE_string(predict_output_dir,
              "",
              "Directory that saves the predicted results of output layers");
DEFINE_string(model_list, "", "File that saves the model list when evaluation");

namespace paddle {

void Trainer::init(int argc, char** argv) {
  initMain(argc, argv);
  initPython(argc, argv);

  auto config = TrainerConfigHelper::createFromFlagConfig();
  feenableexcept(FE_INVALID | FE_DIVBYZERO | FE_OVERFLOW);

  init(config);
}

void Trainer::init(const std::shared_ptr<TrainerConfigHelper>& config,
                   bool testing,
                   const std::shared_ptr<GradientMachine>& gradientMachine,
                   const std::shared_ptr<DataProvider>& dataProvider,
                   const std::shared_ptr<DataProvider>& testDataProvider) {
  this->stats_ = std::make_shared<TrainerStats>();

  config_ = config;

  config_->updateConfigFromFlags();

  testing_ = testing;

  // in testing, mode_ may GradientMachine::kTesting or
  // GradientMachine::kSgdSparseCpuTraining

  if (FLAGS_local) {
    CHECK(!FLAGS_loadsave_parameters_in_pserver)
        << "local and loadsave_parameters_in_pserver can not both true";
    if (config_->getOptConfig().use_sparse_remote_updater()) {
      config_->disableRemoteSparseUpdaterForEachParams();
      LOG(INFO) << "ignore sparse_remote_update=true due to  --local=true";
    }
  }
  if (FLAGS_loadsave_parameters_in_pserver) {
    CHECK(config_->getOptConfig().use_sparse_remote_updater())
        << "no parameter to load from pserver, please check network config";
  }
  if (testing && !FLAGS_loadsave_parameters_in_pserver) {
    if (config_->getOptConfig().use_sparse_remote_updater()) {
      config_->disableRemoteSparseUpdater();
      LOG(INFO) << "because parameter is loaded local,"
                << "tester ignore sparse_remote_update flag";
    }
  }

  CHECK(TrainAlgorithm::isValid(config_->getOptConfig().algorithm()))
      << "invalid algorithm configuration: "
      << config_->getOptConfig().algorithm();

  bool useSparseUpdater = false;
  for (auto& paraConfig : config_->getModelConfig().parameters()) {
    if (paraConfig.sparse_update() || paraConfig.sparse_remote_update()) {
      useSparseUpdater = true;
    }
  }

  if (testing) {
    LOG(INFO) << "trainer: in testing mode";
    if (config_->getOptConfig().use_sparse_remote_updater() ||
        FLAGS_trainer_count > 1) {
      mode_ = GradientMachine::kSgdSparseCpuTraining;
      LOG(INFO) << "trainer mode: SgdSparseCpuTraining";
    } else {
      mode_ = GradientMachine::kTesting;
      LOG(INFO) << "trainer mode: Testing";
    }
  } else if (IGradientMachineMode::tryGetMode(
                 (int*)&mode_,
                 config_->getOptConfig().algorithm(),
                 FLAGS_trainer_count,
                 FLAGS_local,
                 FLAGS_use_gpu)) {
    LOG(INFO) << "Custom trainer mode.";
  } else if ((config_->getOptConfig().algorithm() == TrainAlgorithm::SGD ||
              config_->getOptConfig().algorithm() ==
                  TrainAlgorithm::AsyncSGD) &&
             useSparseUpdater) {
    mode_ = GradientMachine::kSgdSparseCpuTraining;
    LOG(INFO) << "trainer mode: SgdSparseCpuTraining";
  } else {
    mode_ = GradientMachine::kNormal;
    LOG(INFO) << "trainer mode: Normal";
  }

  // initialize trainer internal
  trainerInternal_.init(config_,
                        gradientMachine,
                        TrainerInternalConfig::createFromMode(mode_),
                        stats_,
                        testing);
  std::unique_ptr<ParameterUtilConfig> paramConfig(
      new ParameterUtilConfig(FLAGS_save_only_one,
                              FLAGS_saving_period,
                              FLAGS_loadsave_parameters_in_pserver,
                              FLAGS_config));

  paramUtil_.reset(
      new paddle::ParameterUtil(config_,
                                std::move(paramConfig),
                                trainerInternal_.getGradientMachine(),
                                trainerInternal_.getParameterUpdater()));

  bool gpuData =
      FLAGS_use_gpu && (!FLAGS_parallel_nn) &&
      (!IGradientMachineMode::dataMustInCpu(mode_, FLAGS_trainer_count));

  dataProvider_ = dataProvider;
  if (!dataProvider_ && config_->hasDataConfig() && !testing_) {
    dataProvider_.reset(DataProvider::create(*config_, *config_, gpuData));
  }
  if (!testDataProvider_) {
    // No evaluator_ if there is testDataProvider but no dataProvider.
    evaluator_.reset(trainerInternal_.getGradientMachine()->makeEvaluator());
    currentEvaluator_.reset(
        trainerInternal_.getGradientMachine()->makeEvaluator());
    if (FLAGS_average_test_period > 0 && FLAGS_trainer_id == 0 &&
        config_->getOptConfig().average_window() > 0) {
      CHECK_EQ(FLAGS_average_test_period % FLAGS_log_period, 0)
          << "FLAGS_average_test_period must be divided by FALGS_log_period";
      averageEvaluator_.reset(
          trainerInternal_.getGradientMachine()->makeEvaluator());
    }
  }

  testDataProvider_ = testDataProvider;
  if (!testDataProvider_ && config_->hasTestDataConfig()) {
    testDataProvider_.reset(
        DataProvider::create(config_->getTestDataConfig(), *config_, gpuData));
  }
  if (testDataProvider_) {
    if (config_->getOptConfig().use_sparse_remote_updater()) {
      LOG(FATAL) << "It's prohibited to set sparse_remote_update "
                 << "in some layers if testing will be under going "
                 << "in the middle of training. You can do testing "
                 << "within separate process.";
    }
    createTester();
  }

  if (!testing &&
      (trainerInternal_.getGradientMachine()->hasStaticParameters())) {
    CHECK(!FLAGS_loadsave_parameters_in_pserver)
        << "is_static and loadsave_parameters_in_pserver can not both true";
  }
  if (testing) {
    // will load per pass for tester
  } else if (paramUtil_->tryLoadParametersFromConfig()) {
    // load from config already.
  } else {
    trainerInternal_.getGradientMachine()->randParameters();
  }

  // Only non static parameters need to be updated
  std::vector<ParameterPtr>& parameters =
      trainerInternal_.getGradientMachine()->getNonStaticParameters();
  if (trainerInternal_.getParameterUpdater()) {
    trainerInternal_.getParameterUpdater()->init(parameters);
{'fraction_non_alphanumeric': 0.10560240640275018, 'fraction_numerical': 0.004834291239189987, 'mean_word_length': 3.2381060778511266, 'pattern_counts': {'":': 0, '<': 99, '<?xml version=': 0, '>': 90, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4908003', 'n_tokens_mistral': 5470, 'n_tokens_neox': 5236, 'n_words': 1282}