mindspore2022/tests/ut/cpp/dataset/ir_callback_test.cc

/**
 * Copyright 2020-2021 Huawei Technologies Co., Ltd
 *
 * 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 <memory>
#include <list>

#include "common/common.h"
#include "minddata/dataset/callback/ds_callback.h"
#include "minddata/dataset/core/client.h"
#include "minddata/dataset/engine/datasetops/epoch_ctrl_op.h"
#include "minddata/dataset/engine/datasetops/source/random_data_op.h"
#include "minddata/dataset/engine/tree_adapter.h"
#include "minddata/dataset/include/dataset/datasets.h"
#include "minddata/dataset/include/dataset/transforms.h"
#include "minddata/dataset/kernels/data/no_op.h"
#include "utils/log_adapter.h"

using namespace mindspore::dataset;
using mindspore::LogStream;
using mindspore::MsLogLevel::INFO;

namespace mindspore {
namespace dataset {
namespace test {

class TestCallback : public DSCallback {
 public:
  TestCallback(int32_t step_size)
      : DSCallback(step_size),
        begin_(true),
        epoch_begin_(true),
        step_begin_(true),
        end_(false),
        epoch_end_(true),
        step_end_(true) {
    all_names_.reserve(32);
    all_step_nums_.reserve(32);
    all_ep_nums_.reserve(32);
  }

  Status DSBegin(const CallbackParam &cb_param) override {
    std::lock_guard<std::mutex> guard(lock_);
    all_names_.push_back("BGN");
    all_step_nums_.push_back(cb_param.cur_step_num_);
    all_ep_nums_.push_back(cb_param.cur_epoch_num_);
    return Status::OK();
  }
  Status DSEpochBegin(const CallbackParam &cb_param) override {
    std::lock_guard<std::mutex> guard(lock_);
    all_names_.push_back("EPBGN");
    all_step_nums_.push_back(cb_param.cur_step_num_);
    all_ep_nums_.push_back(cb_param.cur_epoch_num_);
    return Status::OK();
  }
  Status DSNStepBegin(const CallbackParam &cb_param) override {
    std::lock_guard<std::mutex> guard(lock_);
    all_names_.push_back("SPBGN");
    all_step_nums_.push_back(cb_param.cur_step_num_);
    all_ep_nums_.push_back(cb_param.cur_epoch_num_);
    return Status::OK();
  }
  Status DSEnd(const CallbackParam &cb_param) override {
    std::lock_guard<std::mutex> guard(lock_);
    all_names_.push_back("END");
    all_step_nums_.push_back(cb_param.cur_step_num_);
    all_ep_nums_.push_back(cb_param.cur_epoch_num_);
    return Status::OK();
  }
  Status DSEpochEnd(const CallbackParam &cb_param) override {
    std::lock_guard<std::mutex> guard(lock_);
    all_names_.push_back("EPEND");
    all_step_nums_.push_back(cb_param.cur_step_num_);
    all_ep_nums_.push_back(cb_param.cur_epoch_num_);
    return Status::OK();
  }
  Status DSNStepEnd(const CallbackParam &cb_param) override {
    std::lock_guard<std::mutex> guard(lock_);
    all_names_.push_back("SPEND");
    all_step_nums_.push_back(cb_param.cur_step_num_);
    all_ep_nums_.push_back(cb_param.cur_epoch_num_);
    return Status::OK();
  }

  bool IsBeginNeeded() override { return begin_; }
  bool IsEpochBeginNeeded() override { return epoch_begin_; }
  bool IsNStepBeginNeeded() override { return step_begin_; }
  bool IsEndNeeded() override { return end_; }
  bool IsEpochEndNeeded() override { return epoch_end_; }
  bool IsNStepEndNeeded() override { return step_end_; }

  std::vector<std::string> all_names(size_t len) {
    std::vector<std::string> res(all_names_.begin(), all_names_.begin() + len);
    std::sort(res.begin(), res.end());
    return res;
  }

  std::vector<int64_t> all_step_nums(size_t len) {
    std::vector<int64_t> res(all_step_nums_.begin(), all_step_nums_.begin() + len);
    std::sort(res.begin(), res.end());
    return res;
  }

  std::vector<int64_t> all_ep_nums(size_t len) {
    std::vector<int64_t> res(all_ep_nums_.begin(), all_ep_nums_.begin() + len);
    std::sort(res.begin(), res.end());
    return res;
  }

  // flag for turning callback on and off
  bool begin_, epoch_begin_, step_begin_, end_, epoch_end_, step_end_;
  // name of the callback function in sequence, BGN, EPBGN, SPB, END, EPEND, SPEND
  std::vector<std::string> all_names_;
  std::vector<int64_t> all_step_nums_, all_ep_nums_;
  std::mutex lock_;
};

}  // namespace test
}  // namespace dataset
}  // namespace mindspore

class MindDataTestCallback : public UT::DatasetOpTesting {
 public:
  void SetUp() override {
    DatasetOpTesting::SetUp();
    GlobalInit();
  }

  void TestBasicCallback(std::shared_ptr<ExecutionTree> tree, std::shared_ptr<DatasetOp> callback_node,
                         int32_t step_size) {
    // config callback
    Status rc;
    std::shared_ptr<test::TestCallback> tst_cb = std::make_shared<test::TestCallback>(step_size);
    std::shared_ptr<DSCallback> cb1 = tst_cb;
    std::vector<std::shared_ptr<DSCallback>> cbs = {};
    cbs.push_back(cb1);
    callback_node->AddCallbacks(std::move(cbs));

    ASSERT_OK(tree->Prepare());
    ASSERT_OK(tree->Launch());
    // Start the loop of reading tensors from our pipeline
    DatasetIterator di(tree);
    TensorMap tensor_map;
    rc = di.GetNextAsMap(&tensor_map);
    EXPECT_TRUE(rc.IsOk());
    while (!tensor_map.empty()) {
      rc = di.GetNextAsMap(&tensor_map);
      EXPECT_TRUE(rc.IsOk());
    }

    std::vector<std::string> callback_names = {"BGN", "EPBGN", "SPBGN", "SPEND", "SPBGN", "SPEND", "EPEND"};
    std::sort(callback_names.begin(), callback_names.end());
    std::vector<int64_t> all_steps = {0, 0, 1, 1, 65, 65, 88};
    std::vector<int64_t> all_epochs = {0, 1, 1, 1, 1, 1, 1};
    // doing resize to make sure no unexpected epoch_end or extra epoch_begin is called
    size_t len = 7;
    EXPECT_EQ(tst_cb->all_names(len), callback_names);
    EXPECT_EQ(tst_cb->all_step_nums(len), all_steps);
    EXPECT_EQ(tst_cb->all_ep_nums(len), all_epochs);
  }
  std::vector<std::shared_ptr<DatasetOp>> GenerateNodes() {
    // config leaf_op, use random_data to avoid I/O
    std::unique_ptr<DataSchema> schema = std::make_unique<DataSchema>();
    TensorShape shape({});  // empty shape is a 1-value scalar Tensor
    ColDescriptor col("label", DataType(DataType::DE_UINT32), TensorImpl::kFlexible, 0, &shape);
    EXPECT_OK(schema->AddColumn(col));

    std::shared_ptr<ConfigManager> config_manager = GlobalContext::config_manager();
    int32_t op_connector_size = config_manager->op_connector_size();
    int32_t num_workers = config_manager->num_parallel_workers();
    int32_t num_rows = 44;
    std::shared_ptr<RandomDataOp> leaf =
      std::make_shared<RandomDataOp>(num_workers, op_connector_size, num_rows, std::move(schema));
    // config mapOp
    std::vector<std::string> input_columns = {"label"};
    std::vector<std::string> output_columns = {};
    std::vector<std::shared_ptr<TensorOp>> op_list;
    std::shared_ptr<TensorOp> my_no_op = std::make_shared<NoOp>();
    op_list.push_back(my_no_op);
    std::shared_ptr<MapOp> map_op =
      std::make_shared<MapOp>(input_columns, output_columns, std::move(op_list), num_workers, op_connector_size);

    PadInfo pad_map;
    std::shared_ptr<BatchOp> batch_op =
      std::make_shared<BatchOp>(1, false, false, op_connector_size, num_workers, std::vector<std::string>{}, pad_map);

    // config RepeatOp
    int32_t num_repeats = 2;
    std::shared_ptr<RepeatOp> repeat_op = std::make_shared<RepeatOp>(num_repeats);
    // start build then launch tree
    leaf->SetTotalRepeats(num_repeats);
    leaf->SetNumRepeatsPerEpoch(num_repeats);
    map_op->SetTotalRepeats(num_repeats);
    map_op->SetNumRepeatsPerEpoch(num_repeats);
    batch_op->SetTotalRepeats(num_repeats);
    batch_op->SetNumRepeatsPerEpoch(num_repeats);

    return {leaf, map_op, batch_op, repeat_op};
  }
};

/// Feature: Callback
/// Description: Test callbacks with mappable dataset (RandomDataset)
/// Expectation: number and order of callbacks generated are correct
TEST_F(MindDataTestCallback, TestBasicCallback) {
  MS_LOG(INFO) << "Doing: MindDataTestCallback-TestBasicCallback";
  // Test Mapop
  auto nodes = GenerateNodes();
  auto tree = Build(nodes);
  TestBasicCallback(tree, nodes[1], 64);
  // Test LeafOp
  nodes = GenerateNodes();
  tree = Build(nodes);
  TestBasicCallback(tree, nodes[0], 64);
  // Test BatchOp
  nodes = GenerateNodes();
  tree = Build(nodes);
  TestBasicCallback(tree, nodes[2], 64);
}

TEST_F(MindDataTestCallback, TestMultiEpochCallback) {
  MS_LOG(INFO) << "Doing: MindDataTestCallback-TestMultiEpochCallback";
  // config callback
  Status rc;
  std::shared_ptr<test::TestCallback> tst_cb = std::make_shared<test::TestCallback>(4);
  std::shared_ptr<DSCallback> cb1 = tst_cb;
  // config leaf_op, use random_data to avoid I/O
  std::shared_ptr<ConfigManager> config_manager = GlobalContext::config_manager();
  int32_t op_connector_size = config_manager->op_connector_size();
  int32_t num_workers = config_manager->num_parallel_workers();
  std::unique_ptr<DataSchema> schema = std::make_unique<DataSchema>();
  TensorShape shape({});  // empty shape is a 1-value scalar Tensor
  ColDescriptor col("label", DataType(DataType::DE_UINT32), TensorImpl::kFlexible, 0, &shape);
  ASSERT_OK(schema->AddColumn(col));
  std::shared_ptr<RandomDataOp> leaf = std::make_shared<RandomDataOp>(4, op_connector_size, 4, std::move(schema));
  // config mapOp
  std::vector<std::string> input_columns = {"label"};
  std::vector<std::string> output_columns = {};
  std::vector<std::shared_ptr<TensorOp>> op_list;
  std::shared_ptr<TensorOp> my_no_op = std::make_shared<NoOp>();
  op_list.push_back(my_no_op);
  std::shared_ptr<MapOp> map_op =
    std::make_shared<MapOp>(input_columns, output_columns, std::move(op_list), num_workers, op_connector_size);
  std::vector<std::shared_ptr<DSCallback>> cbs = {};
  cbs.push_back(cb1);

  map_op->AddCallbacks(std::move(cbs));
  EXPECT_TRUE(rc.IsOk());
  int32_t num_repeats = 2;
  // config RepeatOp
  std::shared_ptr<RepeatOp> repeat_op = std::make_shared<RepeatOp>(num_repeats);
  // config EpochCtrlOp
  std::shared_ptr<EpochCtrlOp> epoch_ctrl_op = std::make_shared<EpochCtrlOp>(num_repeats);
  // start build then launch tree
  leaf->SetTotalRepeats(4);
  leaf->SetNumRepeatsPerEpoch(2);
  map_op->SetTotalRepeats(4);
  map_op->SetNumRepeatsPerEpoch(2);
  std::shared_ptr<ExecutionTree> tree = Build({leaf, map_op, repeat_op, epoch_ctrl_op});
  rc = tree->Prepare();
  EXPECT_TRUE(rc.IsOk());
  rc = tree->Launch();
  EXPECT_TRUE(rc.IsOk());
  // Start the loop of reading tensors from our pipeline
  DatasetIterator di(tree);
  TensorMap tensor_map;
  size_t num_epochs = 2;
  for (int ep_num = 0; ep_num < num_epochs; ++ep_num) {
    ASSERT_OK(di.GetNextAsMap(&tensor_map));
    EXPECT_TRUE(rc.IsOk());

    while (tensor_map.size() != 0) {
      rc = di.GetNextAsMap(&tensor_map);
      EXPECT_TRUE(rc.IsOk());
    }
  }

  std::vector<std::string> callback_names = {"BGN",   "EPBGN", "SPBGN", "SPEND", "SPBGN", "SPEND", "EPEND",
                                             "EPBGN", "SPBGN", "SPEND", "SPBGN", "SPEND", "EPEND"};
  std::sort(callback_names.begin(), callback_names.end());

  std::vector<int64_t> all_steps = {0, 0, 1, 1, 5, 5, 8, 8, 9, 9, 13, 13, 16};
  std::vector<int64_t> all_epochs = {0, 1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2};

  size_t len = 13;
  EXPECT_EQ(tst_cb->all_names(len), callback_names);
  EXPECT_EQ(tst_cb->all_ep_nums(len), all_epochs);
  EXPECT_EQ(tst_cb->all_step_nums(len), all_steps);
}

TEST_F(MindDataTestCallback, TestSelectedCallback) {
  MS_LOG(INFO) << "Doing: MindDataTestCallback-TestSelectedCallback";
  // config callback
  Status rc;
  std::shared_ptr<test::TestCallback> tst_cb = std::make_shared<test::TestCallback>(4);
  // turn off the epochs
  tst_cb->epoch_begin_ = false;
  tst_cb->epoch_end_ = false;
  std::shared_ptr<SchemaObj> schema = Schema();
  ASSERT_OK(schema->add_column("label", mindspore::DataType::kNumberTypeUInt32, {}));
  std::shared_ptr<Dataset> ds = RandomData(4, schema);
  ASSERT_NE(ds, nullptr);
  ds->SetNumWorkers(1);
  // config mapOp
  ds = ds->Map({std::make_shared<transforms::TypeCast>(mindspore::DataType::kNumberTypeUInt64)}, {"label"}, {}, {},
               nullptr, {tst_cb});
  ds->SetNumWorkers(1);
  ASSERT_NE(ds, nullptr);
  ds = ds->Repeat(2);
  ASSERT_NE(ds, nullptr);
  int32_t num_epochs = 2;
  auto itr = ds->CreateIterator({}, num_epochs);
  for (int ep_num = 0; ep_num < num_epochs; ++ep_num) {
    std::unordered_map<std::string, mindspore::MSTensor> row;
    ASSERT_OK(itr->GetNextRow(&row));
    while (!row.empty()) {
      ASSERT_OK(itr->GetNextRow(&row));
    }
  }

  std::vector<std::string> callback_names = {"BGN",   "SPBGN", "SPEND", "SPBGN", "SPEND",
                                             "SPBGN", "SPEND", "SPBGN", "SPEND"};
  std::sort(callback_names.begin(), callback_names.end());

  std::vector<int64_t> all_steps = {0, 1, 1, 5, 5, 9, 9, 13, 13};
  std::vector<int64_t> all_epochs = {0, 1, 1, 1, 1, 2, 2, 2, 2};

  size_t len = 9;
  EXPECT_EQ(tst_cb->all_names(len), callback_names);
  EXPECT_EQ(tst_cb->all_ep_nums(len), all_epochs);
  EXPECT_EQ(tst_cb->all_step_nums(len), all_steps);
}

TEST_F(MindDataTestCallback, TestCAPICallback) {
  MS_LOG(INFO) << "Doing: MindDataTestCallback-TestCAPICallback";
  // config callback
  std::shared_ptr<test::TestCallback> tst_cb = std::make_shared<test::TestCallback>(64);
  std::shared_ptr<DSCallback> cb1 = tst_cb;
  // Create a RandomDataset.  Use random_data to avoid I/O
  std::shared_ptr<SchemaObj> schema = Schema();
  ASSERT_OK(schema->add_column("label", mindspore::DataType::kNumberTypeUInt32, {}));
  std::shared_ptr<Dataset> ds = RandomData(44, schema);
  ASSERT_NE(ds, nullptr);
  ds = ds->Map({std::make_shared<transforms::TypeCast>(mindspore::DataType::kNumberTypeUInt64)}, {"label"}, {}, {},
               nullptr, {cb1});
  ASSERT_NE(ds, nullptr);
  ds = ds->Repeat(2);
  ASSERT_NE(ds, nullptr);

  auto tree_adapter = std::make_shared<TreeAdapter>();

  // Disable IR optimization pass
  tree_adapter->SetOptimize(false);

  // using tree_adapter to set num_epoch = 1
  ASSERT_OK(tree_adapter->Compile(ds->IRNode(), 1));

  TensorRow row;
  ASSERT_OK(tree_adapter->GetNext(&row));
  while (!row.empty()) {
    ASSERT_OK(tree_adapter->GetNext(&row));
  }
  std::vector<std::string> callback_names = {"BGN", "EPBGN", "SPBGN", "SPEND", "SPBGN", "SPEND", "EPEND"};
  std::sort(callback_names.begin(), callback_names.end());
  std::vector<int64_t> all_steps = {0, 0, 1, 1, 65, 65, 88};
  std::vector<int64_t> all_epochs = {0, 1, 1, 1, 1, 1, 1};
  // doing resize to make sure no unexpected epoch_end or extra epoch_begin is called
  size_t len = 7;
  EXPECT_EQ(tst_cb->all_names(len), callback_names);
  EXPECT_EQ(tst_cb->all_step_nums(len), all_steps);
  EXPECT_EQ(tst_cb->all_ep_nums(len), all_epochs);
}