/**
 * Copyright (c) 2021 OceanBase
 * OceanBase CE is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *          http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */

#define USING_LOG_PREFIX COMMON

#include <gtest/gtest.h>
#include <gmock/gmock.h>
#define private public
#include "share/io/ob_io_manager.h"
#include "share/io/ob_io_calibration.h"
#include "share/io/io_schedule/ob_io_mclock.h"
#undef private
#include "share/ob_local_device.h"
#include "lib/thread/thread_pool.h"
#include "lib/file/file_directory_utils.h"

#define ASSERT_SUCC(ret) ASSERT_EQ((ret), ::oceanbase::common::OB_SUCCESS)
#define ASSERT_FAIL(ret) ASSERT_NE((ret), ::oceanbase::common::OB_SUCCESS)

using namespace oceanbase::lib;
using namespace oceanbase::common;
using namespace oceanbase::share;

#define TEST_ROOT_DIR "io_test"
#define TEST_DATA_DIR TEST_ROOT_DIR "/data_dir"
#define TEST_SSTABLE_DIR TEST_DATA_DIR "/sstable"

static const int64_t IO_MEMORY_LIMIT = 10L * 1024L * 1024L * 1024L;
static const uint64_t TEST_TENANT_ID = 1001;


int init_device(const int64_t media_id, ObLocalDevice &device)
{
  int ret = OB_SUCCESS;
  const int64_t IO_OPT_COUNT = 6;
  const int64_t block_size = 1024L * 1024L * 2L; // 2MB
  const int64_t data_disk_size = 1024L * 1024L * 1024L; // 1GB
  const int64_t data_disk_percentage = 50L;
  ObIODOpt io_opts[IO_OPT_COUNT];
  io_opts[0].key_ = "data_dir";                   io_opts[0].value_.value_str = TEST_DATA_DIR;
  io_opts[1].key_ = "sstable_dir";                io_opts[1].value_.value_str = TEST_SSTABLE_DIR;
  io_opts[2].key_ = "block_size";                 io_opts[2].value_.value_int64 = block_size;
  io_opts[3].key_ = "datafile_disk_percentage";   io_opts[3].value_.value_int64 = data_disk_percentage;
  io_opts[4].key_ = "datafile_size";              io_opts[4].value_.value_int64 = data_disk_size;
  io_opts[5].key_ = "media_id";                   io_opts[5].value_.value_int64 = media_id;
  ObIODOpts init_opts;
  init_opts.opts_ = io_opts;
  init_opts.opt_cnt_ = IO_OPT_COUNT;
  bool need_format = false;
  if (OB_FAIL(device.init(init_opts))) {
    LOG_WARN("init device failed", K(ret));
  } else {
    int64_t reserved_size = 0;
    ObIODOpts opts_start;
    ObIODOpt opt_start;
    opts_start.opts_ = &(opt_start);
    opts_start.opt_cnt_ = 1;
    opt_start.set("reserved size", reserved_size);
    if (OB_FAIL(device.start(opts_start))) {
      LOG_WARN("start device failed", K(ret));
    }
  }
  return ret;
}


class TestIOStruct : public ::testing::Test
{
public:
static void SetUpTestCase()
{
  // prepare test directory and file
  system("mkdir -p " TEST_DATA_DIR);
  system("mkdir -p " TEST_SSTABLE_DIR);

  // init io device
  static oceanbase::share::ObLocalDevice local_device;
  ASSERT_SUCC(init_device(0, local_device));
  THE_IO_DEVICE = &local_device;
}

static void TearDownTestCase()
{
  THE_IO_DEVICE->destroy();
}

static ObIOInfo get_random_io_info()
{
  ObIOInfo io_info;
  io_info.tenant_id_ = OB_SERVER_TENANT_ID;
  io_info.fd_.first_id_ = ObRandom::rand(0, 10000);
  io_info.fd_.second_id_ = ObRandom::rand(0, 10000);
  io_info.flag_.set_mode(static_cast<ObIOMode>(ObRandom::rand(0, (int)ObIOMode::MAX_MODE - 1)));
  io_info.flag_.set_category(static_cast<ObIOCategory>(ObRandom::rand(0, (int)ObIOCategory::MAX_CATEGORY - 1)));
  io_info.flag_.set_wait_event(ObRandom::rand(1, 9999));
  io_info.offset_ = ObRandom::rand(1, 1000L * 1000L * 1000L);
  io_info.size_ = ObRandom::rand(1, 1000L * 10L);
  io_info.flag_.set_read();
  return io_info;
}

static ObTenantIOConfig default_tenant_io_config()
{
  ObTenantIOConfig tenant_config;
  tenant_config.callback_thread_count_ = 2;
  tenant_config.memory_limit_ = 1024L * 1024L * 1024L;
  tenant_config.unit_config_.min_iops_ = 1000;
  tenant_config.unit_config_.max_iops_ = 1000;
  tenant_config.unit_config_.weight_ = 1000;
  tenant_config.other_config_.min_percent_ = 100;
  tenant_config.other_config_.max_percent_ = 100;
  tenant_config.other_config_.weight_percent_ = 100;
  return tenant_config;
}
};

class TestIOCallback : public ObIOCallback
{
public:
  TestIOCallback()
    : number_(nullptr), allocator_(nullptr), user_offset_(0),
     user_size_(0), raw_buf_(nullptr), user_buf_(nullptr)
  {}
  virtual ~TestIOCallback();
  virtual const char *get_data() override { return (char *)user_buf_; }
  virtual int64_t size() const override { return sizeof(TestIOCallback); }
  virtual int inner_deep_copy(char *buf, const int64_t buf_len, ObIOCallback *&callback) const override;
  virtual int alloc_io_buf(char *&io_buf, int64_t &io_buf_size, int64_t &aligned_offset) override;
  virtual int inner_process(const bool is_success) override;
  TO_STRING_KV(KP(number_), KP(allocator_), K(user_offset_), K(user_size_), KP(raw_buf_), KP(user_buf_));

public:
  int64_t *number_;
  ObIAllocator *allocator_;
  int64_t user_offset_;
  int64_t user_size_;
  void *raw_buf_;
  void *user_buf_;
};

TEST_F(TestIOStruct, IOFlag)
{
  // default invalid
  ObIOFlag flag;
  ObIOFlag flag2;
  ASSERT_FALSE(flag.is_valid());

  // normal usage
  flag.set_mode(ObIOMode::READ);
  flag.set_category(ObIOCategory::USER_IO);
  flag.set_wait_event(99);
  ASSERT_TRUE(flag.is_valid());

  // test io mode
  flag2 = flag;
  ASSERT_TRUE(flag2.is_valid());
  flag2.set_mode(ObIOMode::MAX_MODE);
  ASSERT_FALSE(flag2.is_valid());
  flag2.set_mode((ObIOMode)88);
  ASSERT_FALSE(flag2.is_valid());

  // test io category
  flag2 = flag;
  ASSERT_TRUE(flag2.is_valid());
  flag2.set_category(ObIOCategory::MAX_CATEGORY);
  ASSERT_FALSE(flag2.is_valid());
  flag2.set_category((ObIOCategory)88);
  ASSERT_FALSE(flag2.is_valid());

  // test wait event number
  flag2 = flag;
  ASSERT_TRUE(flag2.is_valid());
  flag2.set_wait_event(0);
  ASSERT_FALSE(flag2.is_valid());
  flag2.set_wait_event(-100);
  ASSERT_FALSE(flag2.is_valid());

  // test reset
  flag2 = flag;
  ASSERT_TRUE(flag2.is_valid());
  flag2.reset();
  ASSERT_FALSE(flag2.is_valid());
}

TEST_F(TestIOStruct, IOInfo)
{
  // default invalid
  ObIOInfo info;
  ASSERT_FALSE(info.is_valid());

  // normal usage
  ObIOFd fd;
  fd.first_id_ = 0;
  fd.second_id_ = 0;
  info.tenant_id_ = OB_SERVER_TENANT_ID;
  info.fd_ = fd;
  info.flag_.set_mode(ObIOMode::READ);
  info.flag_.set_category(ObIOCategory::USER_IO);
  info.flag_.set_wait_event(1);
  info.offset_ = 80;
  info.size_ = 1;
  ASSERT_TRUE(info.is_valid());

  // write io info require write buf
  info.flag_.set_mode(ObIOMode::WRITE);
  ASSERT_FALSE(info.is_valid());
  char tmp_buf[10] = { 0 };
  info.buf_ = tmp_buf;
  ASSERT_TRUE(info.is_valid());

  // test reset
  info.reset();
  ASSERT_FALSE(info.is_valid());
}

TEST_F(TestIOStruct, IOHandle)
{
  ObIORequest req;
  req.inc_ref(); // prevent from free
  // default invalid
  ObIOHandle handle;
  ASSERT_FALSE(handle.is_valid());
  ASSERT_TRUE(handle.is_empty());

  // normal usage
  handle.set_request(req);
  ASSERT_TRUE(handle.is_valid());
  ASSERT_FALSE(handle.is_empty());
  ASSERT_EQ(2, req.ref_cnt_);
  ASSERT_EQ(1, req.out_ref_cnt_);

  // copy assign
  ObIOHandle handle2 = handle;
  ASSERT_EQ(3, req.ref_cnt_);
  ASSERT_EQ(2, req.out_ref_cnt_);

  // reset
  handle2.reset();
  ASSERT_EQ(2, req.ref_cnt_);
  ASSERT_EQ(1, req.out_ref_cnt_);
  ASSERT_FALSE(handle2.is_valid());
  ASSERT_TRUE(handle2.is_empty());

  // test log
  LOG_INFO("test log io handle", K(handle), K(handle2));
}

TEST_F(TestIOStruct, IOAllocator)
{
  ObIOAllocator allocator;
  ASSERT_FALSE(allocator.is_inited_);

  // init
  const int64_t memory_limit = 1024L * 1024L * 1024L; // 1GB
  ASSERT_SUCC(allocator.init(TEST_TENANT_ID, memory_limit));
  ASSERT_TRUE(allocator.is_inited_);

  // alloc and free memory
  void *buf = allocator.alloc(1000);
  ASSERT_NE(buf, nullptr);
  allocator.free(buf);

  // alloc and free object
  ObIORequest *req = nullptr;
  allocator.alloc(req);
  ASSERT_NE(req, nullptr);
  allocator.free(req);

  // destroy
  allocator.destroy();
  ASSERT_FALSE(allocator.is_inited_);
}

TEST_F(TestIOStruct, IORequest)
{
  ObTenantIOManager tenant_io_mgr;
  tenant_io_mgr.inc_ref();
  ASSERT_SUCC(tenant_io_mgr.io_allocator_.init(TEST_TENANT_ID, IO_MEMORY_LIMIT));
  ObRefHolder<ObTenantIOManager> holder(&tenant_io_mgr);
  ObIOFd fd;
  fd.first_id_ = 0;
  fd.second_id_ = 1;

  // default invalid
  ObIORequest req;
  ASSERT_FALSE(req.is_inited_);


  // prepare read request
  req.destroy();
  req.tenant_io_mgr_.hold(&tenant_io_mgr);
  ObIOInfo read_info;
  read_info.tenant_id_ = OB_SERVER_TENANT_ID;
  read_info.fd_ = fd;
  read_info.flag_.set_mode(ObIOMode::READ);
  read_info.flag_.set_category(ObIOCategory::USER_IO);
  read_info.flag_.set_wait_event(1);
  read_info.offset_ = 89;
  read_info.size_ = 1;
  ASSERT_TRUE(read_info.is_valid());
  ASSERT_SUCC(req.init(read_info));
  ASSERT_TRUE(req.is_inited_);
  ASSERT_EQ(req.io_buf_, nullptr); // read buf allocation is delayed
  ASSERT_SUCC(req.prepare());
  ASSERT_NE(req.io_buf_, nullptr);

  // prepare write request
  req.destroy();
  req.tenant_io_mgr_.hold(&tenant_io_mgr);
  ObIOInfo write_info = read_info;
  write_info.flag_.set_mode(ObIOMode::WRITE);
  ASSERT_FAIL(req.init(write_info)); // not aligned
  write_info.offset_ = DIO_READ_ALIGN_SIZE * 2;
  ASSERT_FAIL(req.init(write_info)); // only offset aligned, size not aligned
  write_info.size_ = DIO_READ_ALIGN_SIZE * 4;
  ASSERT_FAIL(req.init(write_info)); // offset and size aligned, but write buf is null

  write_info.buf_ = "test_write";
  ASSERT_SUCC(req.init(write_info)); // normal usage
  ASSERT_TRUE(req.is_inited_);
  ASSERT_NE(req.io_buf_, nullptr); // write buf need copy immediately

  // test finish
  req.finish(OB_CANCELED);
  ASSERT_EQ(req.ret_code_.io_ret_, OB_CANCELED);
  req.finish(OB_IO_ERROR);
  ASSERT_NE(req.ret_code_.io_ret_, OB_IO_ERROR); // finish only once
}

TEST_F(TestIOStruct, IOAbility)
{
  ObIOBenchResult item, item2;
  ASSERT_FALSE(item.is_valid());
  ASSERT_TRUE(item == item2);

  ObIOAbility io_ability;
  ASSERT_FALSE(io_ability.is_valid());
  ASSERT_FAIL(io_ability.add_measure_item(item)); // invalid measure item

  // normal usage
  item.mode_ = ObIOMode::READ;
  item.size_ = 10;
  item.rt_us_ = 10;
  item.iops_ = 100;
  ASSERT_TRUE(item.is_valid());
  ASSERT_SUCC(io_ability.add_measure_item(item));
  item.mode_ = ObIOMode::WRITE;
  ASSERT_SUCC(io_ability.add_measure_item(item));
  ASSERT_TRUE(io_ability.is_valid());

  // test equal
  ASSERT_FALSE(item == item2);
  item2 = item;
  ASSERT_TRUE(item == item2);
  item2.iops_ += 0.0001;
  ASSERT_FALSE(item == item2);

  // test sort
  item.size_ = 2;
  item.rt_us_ = 2;
  item.iops_ = 500;
  ASSERT_SUCC(io_ability.add_measure_item(item));
  item.size_ = 5;
  item.rt_us_ = 5;
  item.iops_ = 200;
  ASSERT_SUCC(io_ability.add_measure_item(item));
  ASSERT_EQ(2, io_ability.measure_items_[(int)ObIOMode::WRITE].at(0).size_);
  ASSERT_EQ(5, io_ability.measure_items_[(int)ObIOMode::WRITE].at(1).size_);
  ASSERT_EQ(10, io_ability.measure_items_[(int)ObIOMode::WRITE].at(2).size_);

  // test calculate smallest
  double iops = 0;
  double rt = 0;
  ASSERT_SUCC(io_ability.get_iops(ObIOMode::WRITE, 1, iops));
  ASSERT_SUCC(io_ability.get_rt(ObIOMode::WRITE, 1, rt));
  ASSERT_GE(iops, 500);
  ASSERT_LE(rt, 2);

  // test calculate middle
  ASSERT_SUCC(io_ability.get_iops(ObIOMode::WRITE, 8, iops));
  ASSERT_SUCC(io_ability.get_rt(ObIOMode::WRITE, 8, rt));
  ASSERT_GE(iops, 100); ASSERT_LE(iops, 200);
  ASSERT_GE(rt, 5);     ASSERT_LE(rt, 10);

  // test calculate biggest
  ASSERT_SUCC(io_ability.get_iops(ObIOMode::WRITE, 100, iops));
  ASSERT_SUCC(io_ability.get_rt(ObIOMode::WRITE, 100, rt));
  ASSERT_LE(iops, 100);
  ASSERT_GE(rt, 10);

  // test assign and equal
  ObIOAbility io_ability2;
  ASSERT_SUCC(io_ability2.assign(io_ability));
  ASSERT_TRUE(io_ability2.is_valid());
  ASSERT_TRUE(io_ability == io_ability2);

  // test destroy
  io_ability2.reset();
  ASSERT_FALSE(io_ability2.is_valid());
}


TEST_F(TestIOStruct, IOCalibration)
{
  ObIOCalibration bench;
  ASSERT_FALSE(bench.is_inited_);
  ASSERT_SUCC(bench.init());
  ASSERT_TRUE(bench.is_inited_);
  ObIOAbility io_ability;
  ASSERT_FALSE(io_ability.is_valid());
  ASSERT_FAIL(bench.update_io_ability(io_ability));
  ASSERT_SUCC(bench.reset_io_ability());
  ASSERT_SUCC(bench.get_io_ability(io_ability));
  ASSERT_FALSE(io_ability.is_valid());
  bench.destroy();
  ASSERT_FALSE(bench.is_inited_);

  ObIOBenchResult item;
  ASSERT_SUCC(ObIOCalibration::parse_calibration_string("READ:4096:120:100000", item));
  ASSERT_TRUE(item.is_valid());
  ASSERT_SUCC(ObIOCalibration::parse_calibration_string("read:4K:10ms:1000", item));
  ASSERT_TRUE(item.is_valid());
  ASSERT_SUCC(ObIOCalibration::parse_calibration_string("r:2M:1s:100", item));
  ASSERT_TRUE(item.is_valid());
  ASSERT_SUCC(ObIOCalibration::parse_calibration_string("write:2M:1s:100", item));
  ASSERT_TRUE(item.is_valid());
  ASSERT_SUCC(ObIOCalibration::parse_calibration_string("W:12KB:20us:10000", item));
  ASSERT_TRUE(item.is_valid());

  ASSERT_FAIL(ObIOCalibration::parse_calibration_string(":12KB:20us:10000", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("W::20us:10000", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("W:12KB::10000", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("W:12KB:20us:", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("W:12KB:20us:X1000", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("A:12KB:20us:1000", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("W:12KB:20ks:1000", item));
  ASSERT_FAIL(ObIOCalibration::parse_calibration_string("W:12U:20us:1000", item));
}

TEST_F(TestIOStruct, IOStat)
{
  ObIOStat stat;
  stat.accumulate(100, 200, 300);
  ObIOStatDiff io_diff;
  ObCpuUsage cpu_diff;
  double avg_iops = 0;
  double avg_size = 0;
  double avg_rt = 0;
  double avg_cpu = 0;
  io_diff.diff(stat, avg_iops, avg_size, avg_rt);
  cpu_diff.get_cpu_usage(avg_cpu);
//  ASSERT_DOUBLE_EQ(avg_iops, 0);
  ASSERT_DOUBLE_EQ(avg_size, 2);
  ASSERT_DOUBLE_EQ(avg_rt, 3);
  ASSERT_DOUBLE_EQ(avg_cpu, 0);

  usleep(1000L * 1000L); // sleep 1s
  stat.accumulate(200, 600, 4000);
  io_diff.diff(stat, avg_iops, avg_size, avg_rt);
  ASSERT_NEAR(avg_iops, 200, 1);
  ASSERT_NEAR(avg_size, 3, 0.1);
  ASSERT_NEAR(avg_rt, 20, 0.1);
  ASSERT_GE(avg_cpu, 0);
}

TEST_F(TestIOStruct, IOScheduler)
{
  ObIOCalibration::get_instance().init();
  // test init
  ObIOConfig io_config = ObIOConfig::default_config();
  ObIOAllocator io_allocator;
  ASSERT_SUCC(io_allocator.init(TEST_TENANT_ID, IO_MEMORY_LIMIT));
  ASSERT_TRUE(io_config.is_valid());
  ObIOScheduler scheduler(io_config, io_allocator);
  ASSERT_FALSE(scheduler.is_inited_);
  ASSERT_SUCC(scheduler.init(2, 10000));
  ASSERT_TRUE(scheduler.is_inited_);

  // test schedule
  ObIORequest req;
  ObIOInfo io_info = get_random_io_info();
  ASSERT_TRUE(io_info.is_valid());
  ASSERT_SUCC(req.init(io_info));
  ObTenantIOConfig tenant_config = default_tenant_io_config();
  ObTenantIOClock io_clock;
  ObIOUsage io_usage;
  ASSERT_SUCC(io_clock.init(tenant_config, &io_usage));
  //ASSERT_SUCC(scheduler.schedule_request(io_clock, req));

  // test destroy
  scheduler.destroy();
  ASSERT_FALSE(scheduler.is_inited_);

  ObIOCalibration::get_instance().destroy();
}  

TEST_F(TestIOStruct, MClockQueue)
{
  ObTenantIOConfig io_config;
  io_config.callback_thread_count_ = 2;
  io_config.memory_limit_ = 1024L * 1024L * 1024L;
  io_config.unit_config_.min_iops_ = 100;
  io_config.unit_config_.max_iops_ = 10000L;
  io_config.unit_config_.weight_ = 1000;
  io_config.category_configs_[0].min_percent_ = 1;
  io_config.category_configs_[0].max_percent_ = 90;
  io_config.category_configs_[0].weight_percent_ = 60;
  io_config.category_configs_[1].min_percent_ = 1;
  io_config.category_configs_[1].max_percent_ = 90;
  io_config.category_configs_[1].weight_percent_ = 30;
  io_config.other_config_.min_percent_ = 98;
  io_config.other_config_.max_percent_ = 100;
  io_config.other_config_.weight_percent_ = 10;
  ASSERT_TRUE(io_config.is_valid());
  ObTenantIOClock tenant_clock;
  ObIOUsage io_usage;
  ASSERT_SUCC(tenant_clock.init(io_config, &io_usage));
  ObMClockQueue mqueue1;
  ObMClockQueue mqueue2;
  const int64_t test_count = 6000L;
  ASSERT_SUCC(mqueue1.init());
  ASSERT_SUCC(mqueue2.init());
  ObArenaAllocator arena;
  for (int64_t i = 0; i < test_count; ++i) {
    char *buf = (char *)arena.alloc(sizeof(ObPhyQueue) * 2);
    ASSERT_TRUE(nullptr != buf);
    ObPhyQueue *phy_req = new (buf) ObPhyQueue;
    ObPhyQueue *phy_req2 = new (buf + sizeof(ObPhyQueue)) ObPhyQueue;
    ASSERT_SUCC(mqueue1.push_phyqueue(phy_req));
    ASSERT_SUCC(mqueue2.push_phyqueue(phy_req2));
  }
}

TEST_F(TestIOStruct, IOCallbackManager)
{
  // test init
  ObIOCallbackManager callback_mgr;
  ASSERT_FALSE(callback_mgr.is_inited_);
  ASSERT_FAIL(callback_mgr.init(0, 1000, nullptr));
  ObIOAllocator io_allocator;
  ASSERT_SUCC(io_allocator.init(TEST_TENANT_ID, IO_MEMORY_LIMIT));
  ASSERT_SUCC(callback_mgr.init(2, 1000, &io_allocator));
  ASSERT_TRUE(callback_mgr.is_inited_);

  // test enqueue and dequeue
  ObIORequest req;
  req.inc_ref();
  ObIOInfo io_info = get_random_io_info();
  ASSERT_TRUE(io_info.is_valid());
  ASSERT_SUCC(req.init(io_info));
  ASSERT_FAIL(callback_mgr.enqueue_callback(req));
  char buf[32] = "test";
  req.io_buf_ = buf;
  char callback_buf_[ObIOCallback::CALLBACK_BUF_SIZE] __attribute__ ((aligned (16)));
  req.copied_callback_ = new (callback_buf_) TestIOCallback();
//  ObIOManager::get_instance().io_config_ = ObIOConfig::default_config();
  ASSERT_SUCC(callback_mgr.enqueue_callback(req));

  // test destroy
  callback_mgr.destroy();
  ASSERT_FALSE(callback_mgr.is_inited_);
  req.copied_callback_ = nullptr;
}

TEST_F(TestIOStruct, IOFaultDetector)
{
  // test init
  ObIOConfig io_config = ObIOConfig::default_config();
  ObIOFaultDetector detector(io_config);
  ASSERT_FALSE(detector.is_inited_);
  ASSERT_SUCC(detector.init());
  ASSERT_TRUE(detector.is_inited_);

  // test get device health
  ObDeviceHealthStatus dhs = DEVICE_HEALTH_NORMAL;
  int64_t disk_abnormal_time = 0;
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_NORMAL == dhs);
  ASSERT_TRUE(0 == disk_abnormal_time);

  // test start
  ASSERT_SUCC(detector.start());
  ObIOManager::get_instance().is_working_ = true;

  // test write failure detection
  ObIOInfo io_info = get_random_io_info();
  ObIORequest req;
  req.inc_ref();
  ASSERT_SUCC(req.init(io_info));
  req.io_info_.flag_.set_mode(ObIOMode::WRITE);
  req.finish(OB_CANCELED);
  for (int64_t i = 0; i < 10000L; ++i) {
    detector.record_failure(req);
  }
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_NORMAL == dhs);
  ASSERT_TRUE(0 == disk_abnormal_time);
  req.ret_code_.io_ret_ = OB_IO_ERROR;
  for (int64_t i = 0; i < 10000L; ++i) {
    detector.record_failure(req);
  }
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_ERROR == dhs);
  ASSERT_TRUE(disk_abnormal_time > 0);

  // test read failure detection
  detector.reset_device_health();
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_NORMAL == dhs);
  ASSERT_TRUE(0 == disk_abnormal_time);
  req.io_info_.flag_.set_mode(ObIOMode::READ);
  io_config.data_storage_warning_tolerance_time_ = 1000L * 1000L;
  io_config.data_storage_error_tolerance_time_ = 3000L * 1000L;
  // io manager not init, skip this test
//  detector.record_failure(req);
//  usleep(2000L * 1000L);
//  ASSERT_SUCC(detector.get_device_health(is_device_warning, is_device_error));
//  ASSERT_TRUE(is_device_warning);
//  ASSERT_FALSE(is_device_error);
//  usleep(2000L * 1000L);
//  ASSERT_SUCC(detector.get_device_health(is_device_warning, is_device_error));
//  ASSERT_TRUE(is_device_warning);
//  ASSERT_TRUE(is_device_error);

  // test auto clean device warning, but not clean device error
  detector.reset_device_health();
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_NORMAL == dhs);
  ASSERT_TRUE(0 == disk_abnormal_time);
  io_config.read_failure_black_list_interval_ = 1000L * 100L; // 100ms
  detector.set_device_warning();
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_WARNING == dhs);
  ASSERT_TRUE(disk_abnormal_time > 0);
  usleep(io_config.read_failure_black_list_interval_ * 2);
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_NORMAL == dhs);
  ASSERT_TRUE(0 == disk_abnormal_time);
  detector.set_device_error();
  usleep(io_config.read_failure_black_list_interval_ * 2);
  ASSERT_SUCC(detector.get_device_health_status(dhs, disk_abnormal_time));
  ASSERT_TRUE(DEVICE_HEALTH_ERROR == dhs);
  ASSERT_TRUE(disk_abnormal_time > 0);

  // test destroy
  detector.destroy();
  ASSERT_FALSE(detector.is_inited_);
}

TEST_F(TestIOStruct, IOManager)
{
  ObIOManager io_mgr;
  ASSERT_FALSE(io_mgr.is_inited_);
  ASSERT_SUCC(io_mgr.init());
  ASSERT_TRUE(io_mgr.is_inited_);
  ASSERT_SUCC(io_mgr.start());
  io_mgr.stop();
  io_mgr.destroy();
  ASSERT_FALSE(io_mgr.is_inited_);
}


class TestIOManager : public TestIOStruct
{
public:
  virtual void SetUp()
  {
    ObIOManager::get_instance().destroy();
    const int64_t memory_limit = 10L * 1024L * 1024L * 1024L; // 4GB
    ASSERT_SUCC(ObIOManager::get_instance().init(memory_limit));
    ASSERT_SUCC(ObIOManager::get_instance().start());

    // add io device
    ASSERT_SUCC(OB_IO_MANAGER.add_device_channel(THE_IO_DEVICE, 16, 2, 1024));

    // add tenant io manager
    const uint64_t tenant_id = 500;
    ObTenantIOConfig io_config;
    io_config.memory_limit_ = memory_limit;
    io_config.callback_thread_count_ = 2;
    io_config.other_config_.min_percent_ = 100;
    io_config.other_config_.max_percent_ = 100;
    io_config.other_config_.weight_percent_ = 100;
    io_config.unit_config_.min_iops_ = 10000;
    io_config.unit_config_.max_iops_ = 100000;
    io_config.unit_config_.weight_ = 100;
    ASSERT_SUCC(OB_IO_MANAGER.add_tenant_io_manager(tenant_id, io_config));
  }
  virtual void TearDown()
  {
    ObIOManager::get_instance().stop();
    ObIOManager::get_instance().destroy();
  }
};

TEST_F(TestIOManager, simple)
{
  ObIOFd fd;
  ASSERT_SUCC(THE_IO_DEVICE->open(TEST_ROOT_DIR "/test_io_file", O_CREAT | O_DIRECT | O_TRUNC | O_RDWR, 0644, fd));
  ASSERT_TRUE(fd.is_valid());
  ObIOManager &io_mgr = ObIOManager::get_instance();

  // fallocate
  const int64_t FILE_SIZE = 4 * 1024 * 1024;
  ASSERT_SUCC(THE_IO_DEVICE->fallocate(fd, 0, 0, FILE_SIZE)); // 4M
  {
    oceanbase::common::ObIODFileStat stat_buf;
    ASSERT_SUCC(THE_IO_DEVICE->stat(TEST_ROOT_DIR "/test_io_file", stat_buf));
    ASSERT_EQ(FILE_SIZE, stat_buf.size_);
  }

  // sync write
  const int64_t io_timeout_ms = 1000L * 5L;
  const int64_t write_io_size = DIO_READ_ALIGN_SIZE * 2;
  ObIOInfo io_info;
  io_info.tenant_id_ = 500;
  io_info.fd_ = fd;
  io_info.flag_.set_write();
  io_info.flag_.set_category(ObIOCategory::USER_IO);
  io_info.flag_.set_wait_event(100);
  io_info.offset_ = 0;
  io_info.size_ = write_io_size;
  char buf[write_io_size] = { 0 };
  const int64_t user_offset = write_io_size / 2;
  memset(buf, 'a', user_offset);
  memset(buf + user_offset, 'b', write_io_size - user_offset);
  io_info.buf_ = buf;
  ASSERT_SUCC(io_mgr.write(io_info, io_timeout_ms));

  // check size
  {
    oceanbase::common::ObIODFileStat stat_buf;
    ASSERT_SUCC(THE_IO_DEVICE->stat(TEST_ROOT_DIR "/test_io_file", stat_buf));
    ASSERT_EQ(FILE_SIZE, stat_buf.size_);
  }

  // sync read and compare data buffer
  io_info.flag_.set_read();
  ObIOHandle io_handle;
  ASSERT_SUCC(io_mgr.read(io_info, io_handle, io_timeout_ms));
  ASSERT_NE(nullptr, io_handle.get_buffer());
  ASSERT_EQ(io_info.size_, io_handle.get_data_size());
  ASSERT_EQ(0, memcmp(buf, io_handle.get_buffer(), write_io_size));

  // read end
  io_handle.reset();
  io_info.offset_ = FILE_SIZE;
  io_info.size_ = DIO_READ_ALIGN_SIZE;
  ASSERT_NE(OB_SUCCESS, io_mgr.read(io_info, io_handle, io_timeout_ms));
  ASSERT_EQ(0, io_handle.get_data_size());

  // read tail
  io_handle.reset();
  io_info.offset_ = FILE_SIZE - DIO_READ_ALIGN_SIZE;
  io_info.size_ = DIO_READ_ALIGN_SIZE;
  ASSERT_SUCC(io_mgr.read(io_info, io_handle, io_timeout_ms));
  ASSERT_EQ(DIO_READ_ALIGN_SIZE, io_handle.get_data_size());

  // read tail part
  io_handle.reset();
  io_info.offset_ = FILE_SIZE - DIO_READ_ALIGN_SIZE;
  io_info.size_ = DIO_READ_ALIGN_SIZE * 2;
  ASSERT_NE(OB_SUCCESS, io_mgr.read(io_info, io_handle, io_timeout_ms));
  ASSERT_EQ(DIO_READ_ALIGN_SIZE, io_handle.get_data_size());

  // read with callback
  io_handle.reset();
  io_info.offset_ = user_offset;
  io_info.size_ = write_io_size - user_offset;
  ObArenaAllocator allocator;
  int64_t tmp_number = 0;
  TestIOCallback callback;
  callback.number_ = &tmp_number;
  callback.allocator_ = &allocator;
  callback.user_offset_ = io_info.offset_;
  callback.user_size_ = io_info.size_;
  io_info.callback_ = &callback;
  ASSERT_SUCC(io_mgr.read(io_info, io_handle, io_timeout_ms));
  ASSERT_NE(nullptr, io_handle.get_buffer());
  ASSERT_EQ(io_info.size_, io_handle.get_data_size());
  ASSERT_EQ(0, memcmp(buf + user_offset, io_handle.get_buffer(), write_io_size - user_offset));
  ASSERT_EQ(100, tmp_number); // callback process called
  io_handle.reset();
  ASSERT_EQ(10, tmp_number); // callback destructor called

  ASSERT_SUCC(THE_IO_DEVICE->close(fd));
}


struct IOPerfDevice
{
  IOPerfDevice() : device_id_(0), media_id_(0), async_channel_count_(0), sync_channel_count_(0), max_io_depth_(0),
                   file_size_(0), file_path_(), fd_(0), device_handle_(nullptr) {
    memset(file_path_, 0, sizeof(file_path_));
  }
  bool is_valid() const {
    return device_id_ > 0 && media_id_ >= 0 && async_channel_count_ > 0 && sync_channel_count_ > 0  && max_io_depth_ > 0
      && file_size_ > 0 && strlen(file_path_) > 0;
  }
  TO_STRING_KV(K(device_id_), K(media_id_), K(async_channel_count_), K(sync_channel_count_), K(max_io_depth_),
      K(file_size_), K(file_path_), K(fd_), KP(device_handle_));
  int32_t device_id_;
  int32_t media_id_;
  int32_t async_channel_count_;
  int32_t sync_channel_count_;
  int32_t max_io_depth_;
  int64_t file_size_;
  char file_path_[OB_MAX_FILE_NAME_LENGTH];
  int32_t fd_;
  ObLocalDevice *device_handle_;
};

struct IOPerfTenant
{
  IOPerfTenant() : tenant_id_(0), config_() {}
  bool is_valid() const { return tenant_id_ > 0 && config_.is_valid(); }
  TO_STRING_KV(K(tenant_id_), K(config_));
  int32_t tenant_id_;
  ObTenantIOConfig config_;
};

enum class IOPerfMode
{
  ROLLING,
  BATCH,
  ASYNC,
  UNKNOWN
};

struct IOPerfLoad
{
  IOPerfLoad()
    : tenant_id_(0), category_(ObIOCategory::MAX_CATEGORY), device_id_(0),
      mode_(ObIOMode::MAX_MODE), size_(0), depth_(0), iops_(0),
      thread_count_(0), is_sequence_(false), start_delay_ts_(0), stop_delay_ts_(0),
      device_(nullptr), perf_mode_(IOPerfMode::UNKNOWN)
  {}
  TO_STRING_KV(K(tenant_id_), "category", get_io_category_name(category_), K(device_id_),
      "mode", ObIOMode::READ == mode_ ? "read" : ObIOMode::WRITE == mode_ ? "write" : "unknown",
      "io_size", size_, "io_depth", depth_, "target_iops", iops_,
      K(thread_count_), K(is_sequence_), K(start_delay_ts_), K(stop_delay_ts_), KP(device_), K(perf_mode_));
  bool is_valid() const {
    return tenant_id_ > 0 && category_ < ObIOCategory::MAX_CATEGORY && device_id_ > 0
      && mode_ < ObIOMode::MAX_MODE && size_ > 0 && depth_ > 0 && iops_ >= 0
      && thread_count_ > 0 && start_delay_ts_ >= 0 && stop_delay_ts_ > start_delay_ts_ && size_ > 0
      && (ObIOMode::WRITE == mode_ ? is_io_aligned(size_) : true)
      && perf_mode_ != IOPerfMode::UNKNOWN;
  }
  int32_t tenant_id_;
  ObIOCategory category_;
  int32_t device_id_;
  ObIOMode mode_;
  int32_t size_;
  int32_t depth_;
  int64_t iops_;
  int32_t thread_count_;
  bool is_sequence_;
  int64_t start_delay_ts_;
  int64_t stop_delay_ts_;
  IOPerfDevice *device_;
  IOPerfMode perf_mode_;
};

struct IOPerfScheduler
{
  IOPerfScheduler() : sender_count_(0), schedule_media_id_(0), io_greed_(0) {}
  int64_t sender_count_;
  int64_t schedule_media_id_;
  int64_t io_greed_;
  TO_STRING_KV(K(sender_count_), K(schedule_media_id_), K(io_greed_));
};

struct IOPerfResult
{
  IOPerfResult()
    : succ_count_(0), fail_count_(0), sum_rt_(0), start_delay_ts_(0), stop_delay_ts_(0), disk_rt_(0)
  {}
  TO_STRING_KV(K(succ_count_), K(fail_count_), K(sum_rt_), K(start_delay_ts_), K(stop_delay_ts_));
  int64_t succ_count_;
  int64_t fail_count_;
  int64_t sum_rt_;
  int64_t start_delay_ts_;
  int64_t stop_delay_ts_;
  int64_t disk_rt_;
};

class IOPerfRunner : public ThreadPool
{
public:
  IOPerfRunner()
    : abs_ts_(0), last_offset_(0), write_buf_(nullptr), io_count_(0), report_count_(0), total_io_count_(0)
  {}
  int init(const int64_t absolute_ts, const IOPerfLoad &load);
  void destroy();
  virtual void run1() override;
  int do_batch_io();
  int do_perf_batch();
  int do_perf_rolling();
  int wait_and_count(ObIOHandle &io_handle);
  int wait_handles(ObFixedQueue<ObIOHandle> &handles);
  int print_result();
  TO_STRING_KV(K(load_), K(result_), K(last_offset_), KP(write_buf_), K(fd_));
public:
  int64_t abs_ts_;
  IOPerfLoad load_;
  IOPerfResult result_;
  int64_t last_offset_;
  const char *write_buf_;
  ObIOFd fd_;
  ObConcurrentFIFOAllocator allocator_;
  ObFixedQueue<ObIOHandle> handle_queue_;
  int64_t io_count_;
  int64_t report_count_;
  int64_t total_io_count_;
};

class IOConfModify : public ThreadPool
{
public:
  IOConfModify()
    : modify_init_ts_(0)
  {}
  int init(int64_t modify_init_ts, int64_t modify_delay_ts, const IOPerfTenant &curr_tenant);
  void destroy();
  virtual void run1() override;
  int modify_tenant_io(const int64_t min_iops,
                       const int64_t max_iops,
                       const int64_t weight,
                       IOPerfTenant &curr_tenant);
  TO_STRING_KV(K(modify_delay_ts_), K(fd_), K(curr_tenant_));
public:
  int64_t modify_init_ts_;
  int64_t modify_delay_ts_;
  IOPerfTenant curr_tenant_;
  ObConcurrentFIFOAllocator allocator_;
  IOPerfLoad load_;
  ObIOFd fd_;
};

#define PERF_CONFIG_FILE "io_perf.conf"

void write_perf_config();
int parse_perf_config(const char *config_file_path,
                      IOPerfScheduler &scheduler_config,
                      ObIArray<IOPerfDevice> &perf_devices,
                      ObIArray<IOPerfTenant> &perf_tenants,
                      ObIArray<IOPerfLoad> &perf_loads);


int prepare_file(const char *file_path, const int64_t file_size, int32_t &fd)
{
  int ret = OB_SUCCESS;
  fd = -1;
  bool need_create_perf_file = true;
  bool is_file_exist = false;
  int64_t exist_file_size = 0;
  if (OB_FAIL(FileDirectoryUtils::is_exists(file_path, is_file_exist))) {
    LOG_WARN("check file exist failed", K(ret));
  } else if (is_file_exist) {
    if (OB_FAIL(FileDirectoryUtils::get_file_size(file_path, exist_file_size))) {
      LOG_WARN("get file size failed", K(ret));
    } else if (exist_file_size == file_size) {
      if ((fd = ::open(file_path, O_DIRECT | O_RDWR)) < 0) {
        ret = OB_IO_ERROR;
        LOG_WARN("fail to open file", K(ret));
      } else {
        need_create_perf_file = false;
      }
    }
  }
  if (need_create_perf_file) {
    char cmd[1024] = { 0 };
    sprintf(cmd, "rm -rf %s", file_path);
    system(cmd);
    if ((fd = ::open(file_path, O_CREAT | O_TRUNC | O_DIRECT | O_RDWR, 0644)) < 0) {
      ret = OB_IO_ERROR;
      LOG_WARN("fail to create file", K(ret));
    } else {
      if (fallocate(fd, 0, 0, file_size) < 0) {
        ret = OB_IO_ERROR;
        LOG_WARN("fail to allocate file", K(ret), K(fd), K(file_size));
      } else {
        const int64_t buf_size = 1024L * 1024L; // 1MB
        const int64_t write_count = file_size / buf_size;
        sprintf(cmd, "dd if=/dev/zero of=%s bs=1M count=%ld", file_path, write_count);
        system(cmd);
        fsync(fd);
        LOG_INFO("prepare file finished", K(cmd), "file_size_mb", write_count);
      }
    }
  }
  return ret;
}

TEST_F(TestIOManager, tenant)
{
  ObTenantIOConfig default_config = ObTenantIOConfig::default_instance();
  default_config.unit_config_.max_iops_ = 20000L;
  ASSERT_SUCC(OB_IO_MANAGER.add_tenant_io_manager(1001, default_config));
  ASSERT_SUCC(OB_IO_MANAGER.add_tenant_io_manager(1002, default_config));
  int64_t current_ts = ObTimeUtility::fast_current_time();
  IOPerfLoad load;
  load.category_ = ObIOCategory::USER_IO;
  load.depth_ = 1;
  IOPerfDevice device;
  device.device_id_ = 1;
  strcpy(device.file_path_, "./perf_test");
  device.file_size_ = 1024L * 1024L * 1024L;
  device.device_handle_ = static_cast<ObLocalDevice *>(THE_IO_DEVICE);
  prepare_file(device.file_path_, device.file_size_, device.fd_);
  load.device_ = &device;
  load.device_id_ = 1; // unused
  load.iops_ = 0;
  load.is_sequence_ = false;
  load.mode_ = ObIOMode::READ;
  load.perf_mode_ = IOPerfMode::ROLLING;
  load.size_ = 8192;
  load.start_delay_ts_ = 0;
  load.stop_delay_ts_ = 3L * 1000L * 1000L; // 3s
  load.tenant_id_ = 1001;
  load.thread_count_ = 16;
  IOPerfRunner runner;
  ASSERT_SUCC(runner.init(current_ts, load));
  usleep(2L * 1000L * 1000L); // 2s
  ASSERT_SUCC(OB_IO_MANAGER.remove_tenant_io_manager(1002));
  ASSERT_SUCC(OB_IO_MANAGER.remove_tenant_io_manager(1001));
  runner.wait();
  runner.destroy();
}

TEST_F(TestIOManager, perf)
{
  // uset multi thread to do some io stress, maybe use test_io_performance
  bool is_perf_config_exist = false;
  ASSERT_SUCC(FileDirectoryUtils::is_exists(PERF_CONFIG_FILE, is_perf_config_exist));
  if (!is_perf_config_exist) {
    write_perf_config();
  }
  // parse configs
  IOPerfScheduler scheduler_config;
  ObArray<IOPerfDevice> perf_devices;
  ObArray<IOPerfTenant> perf_tenants;
  ObArray<IOPerfLoad> perf_loads;
  ASSERT_SUCC(parse_perf_config(PERF_CONFIG_FILE, scheduler_config, perf_devices, perf_tenants, perf_loads));
  ASSERT_TRUE(perf_devices.count() > 0);
  ASSERT_TRUE(perf_tenants.count() > 0);
  ASSERT_TRUE(perf_loads.count() > 0);

  ObIOManager::get_instance().destroy();
  const int64_t memory_limit = 30L * 1024L * 1024L * 1024L; // 10GB
  const int64_t queue_depth = 100L;
  ASSERT_SUCC(ObIOManager::get_instance().init(memory_limit, queue_depth, scheduler_config.sender_count_, scheduler_config.schedule_media_id_));
  ASSERT_SUCC(ObIOManager::get_instance().start());

  // prepare devices and files
  char *device_buf = (char *)malloc(sizeof(ObLocalDevice) * perf_devices.count());
  ASSERT_TRUE(nullptr != device_buf);
  for (int64_t i = 0; i < perf_devices.count(); ++i) {
    IOPerfDevice &curr_config = perf_devices.at(i);
    ASSERT_SUCC(prepare_file(curr_config.file_path_, curr_config.file_size_, curr_config.fd_));
    ObLocalDevice *device = new (device_buf + sizeof(ObLocalDevice) * i) ObLocalDevice;
    ASSERT_SUCC(init_device(curr_config.media_id_, *device));
    ASSERT_SUCC(OB_IO_MANAGER.add_device_channel(device, curr_config.async_channel_count_, curr_config.sync_channel_count_, curr_config.max_io_depth_));
    curr_config.device_handle_ = device;
  }
  // prepare tenant io manager
  for (int64_t i = 0; i < perf_tenants.count(); ++i) {
    IOPerfTenant &curr_config = perf_tenants.at(i);
    LOG_INFO("wenqu: tenant config", K(curr_config), K(i));
    ASSERT_SUCC(OB_IO_MANAGER.add_tenant_io_manager(curr_config.tenant_id_, curr_config.config_));
  }
  // prepare perf runners
  char *runner_buf = (char *)malloc(perf_loads.count() * sizeof(IOPerfRunner));
  ObArray<IOPerfRunner *> runners;
  const int64_t start_ts = ObTimeUtility::current_time() + 10000L;
  for (int64_t i = 0; i < perf_loads.count(); ++i) {
    IOPerfRunner *runner = new (runner_buf + i * sizeof(IOPerfRunner)) IOPerfRunner();
    const IOPerfLoad &cur_load = perf_loads.at(i);
    ASSERT_SUCC(runner->init(start_ts, cur_load));
    ASSERT_SUCC(runners.push_back(runner));
  }
  // wait perf finished
  for (int64_t i = 0; i < runners.count(); ++i) {
    IOPerfRunner *runner = runners.at(i);
    runner->wait();
    ASSERT_SUCC(runner->print_result());
    runner->destroy();
  }
  free(runner_buf);

  ObIOManager::get_instance().stop();
  ObIOManager::get_instance().destroy();
  for (int64_t i = 0; i < perf_devices.count(); ++i) {
    ObLocalDevice *device_handle = perf_devices.at(i).device_handle_;
//    ASSERT_SUCC(OB_IO_MANAGER.remove_device_channel(device_handle));
    device_handle->destroy();
  }
  free(device_buf);
  LOG_INFO("wenqu: perf finished");
}

TEST_F(TestIOManager, modify)
{
  // uset multi thread to do some io stress, maybe use test_io_performance
  bool is_perf_config_exist = false;
  ASSERT_SUCC(FileDirectoryUtils::is_exists(PERF_CONFIG_FILE, is_perf_config_exist));
  if (!is_perf_config_exist) {
    write_perf_config();
  }
  // parse configs
  IOPerfScheduler scheduler_config;
  ObArray<IOPerfDevice> perf_devices;
  ObArray<IOPerfTenant> perf_tenants;
  ObArray<IOPerfLoad> perf_loads;
  ASSERT_SUCC(parse_perf_config(PERF_CONFIG_FILE, scheduler_config, perf_devices, perf_tenants, perf_loads));
  ASSERT_TRUE(perf_devices.count() > 0);
  ASSERT_TRUE(perf_tenants.count() > 0);
  ASSERT_TRUE(perf_loads.count() > 0);

  ObIOManager::get_instance().destroy();
  const int64_t memory_limit = 30L * 1024L * 1024L * 1024L; // 10GB
  const int64_t queue_depth = 100L;
  ASSERT_SUCC(ObIOManager::get_instance().init(memory_limit, queue_depth, scheduler_config.sender_count_, scheduler_config.schedule_media_id_));
  ASSERT_SUCC(ObIOManager::get_instance().start());

  // prepare devices and files
  char *device_buf = (char *)malloc(sizeof(ObLocalDevice) * perf_devices.count());
  ASSERT_TRUE(nullptr != device_buf);
  for (int64_t i = 0; i < perf_devices.count(); ++i) {
    IOPerfDevice &curr_config = perf_devices.at(i);
    ASSERT_SUCC(prepare_file(curr_config.file_path_, curr_config.file_size_, curr_config.fd_));
    ObLocalDevice *device = new (device_buf + sizeof(ObLocalDevice) * i) ObLocalDevice;
    ASSERT_SUCC(init_device(curr_config.media_id_, *device));
    ASSERT_SUCC(OB_IO_MANAGER.add_device_channel(device, curr_config.async_channel_count_, curr_config.sync_channel_count_, curr_config.max_io_depth_));
    curr_config.device_handle_ = device;
  }
  // prepare tenant io manager
  for (int64_t i = 0; i < perf_tenants.count(); ++i) {
    IOPerfTenant &curr_config = perf_tenants.at(i);
    LOG_INFO("wenqu: tenant config", K(curr_config), K(i));
    ASSERT_SUCC(OB_IO_MANAGER.add_tenant_io_manager(curr_config.tenant_id_, curr_config.config_));
  }
  // prepare perf runners
  char *runner_buf = (char *)malloc(perf_loads.count() * sizeof(IOPerfRunner));
  char *modifyer_buf = (char *)malloc(perf_loads.count() * sizeof(IOConfModify));  
  ObArray<IOPerfRunner *> runners;
  ObArray<IOConfModify *> modifyers;  
  const int64_t start_ts = ObTimeUtility::current_time() + 10000L;
  for (int64_t i = 0; i < perf_loads.count(); ++i) {
    IOPerfRunner *runner = new (runner_buf + i * sizeof(IOPerfRunner)) IOPerfRunner();
    const IOPerfLoad &cur_load = perf_loads.at(i);
    ASSERT_SUCC(runner->init(start_ts, cur_load));
    ASSERT_SUCC(runners.push_back(runner));
    LOG_INFO("runner start now");
    IOConfModify *modifyer=new (modifyer_buf + i * sizeof(IOConfModify)) IOConfModify();
    IOPerfTenant &curr_tenant = perf_tenants.at(i);
    int64_t modify_init_ts = start_ts+cur_load.start_delay_ts_;
    int64_t modify_delay_ts = 3000000L;
    ASSERT_SUCC(modifyer->init(modify_init_ts, modify_delay_ts, curr_tenant));
    ASSERT_SUCC(modifyers.push_back(modifyer));  
  }
  // wait perf finished
  for (int64_t i = 0; i < runners.count(); ++i) {
    IOPerfRunner *runner = runners.at(i);
    runner->wait();
    ASSERT_SUCC(runner->print_result());
    runner->destroy();
  }
  free(runner_buf);
  free(modifyer_buf);  

  ObIOManager::get_instance().stop();
  ObIOManager::get_instance().destroy();
  for (int64_t i = 0; i < perf_devices.count(); ++i) {
    ObLocalDevice *device_handle = perf_devices.at(i).device_handle_;
//    ASSERT_SUCC(OB_IO_MANAGER.remove_device_channel(device_handle));
    device_handle->destroy();
  }
  free(device_buf);
  LOG_INFO("wenqu: modify finished");
}

TEST_F(TestIOManager, abnormal)
{
  // simulate submit failure
  // simulate get_event failure
  // simulate device hang
}

#define LOG_FILE_PATH "./test_io_manager.log"

int main(int argc, char **argv)
{
  set_memory_limit(20L * 1024L * 1024L * 1024L);
  system("rm -rf " LOG_FILE_PATH "*");
  oceanbase::common::ObLogger::get_logger().set_log_level("INFO");
  oceanbase::common::ObLogger::get_logger().set_file_name(LOG_FILE_PATH, true);
  testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}

/****************                 test io callback              ******************/
TestIOCallback::~TestIOCallback()
{
  if (nullptr != number_) {
    *number_ -= 90;
  }
  if (nullptr != allocator_ && nullptr != raw_buf_) {
    allocator_->free(raw_buf_);
  }
  number_ = nullptr;
  allocator_ = nullptr;
  user_offset_ = 0;
  user_size_ = 0;
  raw_buf_ = nullptr;
  user_buf_ = nullptr;
}


int TestIOCallback::inner_deep_copy(char *buf, const int64_t buf_len, ObIOCallback *&callback) const
{
  int ret = OB_SUCCESS;
  callback = nullptr;
  if (OB_UNLIKELY(nullptr == buf || buf_len <= 0)) {
    ret = OB_INVALID_ARGUMENT;
    LOG_WARN("invalid argument", K(ret), KP(buf), K(buf_len));
  } else if (buf_len < size()) {
    ret = OB_BUF_NOT_ENOUGH;
    LOG_WARN("buf_len not enough", K(ret), K(buf_len), K(size()));
  } else {
    TestIOCallback *tmp_callback = new (buf) TestIOCallback();
    tmp_callback->number_ = number_;
    tmp_callback->allocator_ = allocator_;
    tmp_callback->user_offset_ = user_offset_;
    tmp_callback->user_size_ = user_size_;
    callback = tmp_callback;
  }
  return ret;
}

int TestIOCallback::alloc_io_buf(char *&io_buf, int64_t &io_buf_size, int64_t &aligned_offset)
{
  int ret = OB_SUCCESS;
  align_offset_size(user_offset_, user_size_, aligned_offset, io_buf_size);
  if (OB_ISNULL(raw_buf_ = allocator_->alloc(io_buf_size + DIO_READ_ALIGN_SIZE))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_WARN("allocate memory failed", K(ret), K(user_size_));
  } else {
    io_buf = reinterpret_cast<char *>(upper_align(reinterpret_cast<int64_t>(raw_buf_), DIO_READ_ALIGN_SIZE));
    user_buf_ = io_buf + user_offset_ - aligned_offset;
  }
  return ret;
}

int TestIOCallback::inner_process(const bool is_success)
{
  if (nullptr != number_) {
    is_success ? *number_ += 100 : *number_ -=100;
  }
  return OB_SUCCESS;
}

/****************                io perf              ******************/
void write_perf_config()
{
  int fd = -1;
  const char *file_name = PERF_CONFIG_FILE;
  if (0 > (fd = ::open(file_name, O_RDWR | O_CREAT | O_TRUNC, 0644))) {
    LOG_WARN("open perf config file failed", K(fd), K(file_name));
  } else {
    const char *file_buf =
      "sender_count      schedule_media_id     io_greed\n"
      "8                 0                     0\n"
      "\n"
      "device_id   media_id    async_channel   sync_channel    max_io_depth    file_size_gb    file_path\n"
      "1           0           8               1               64              1               ./perf_test\n"
      "\n"
      "tenant_id   min_iops    max_iops    weight          category\n"
      "1002        5000        50000       700             user: 90, 100, 80; large: 1, 90, 10; other: 9, 100, 10;\n"
      "1001        50000       100000      1000            other: 100, 100, 100;\n"
      "\n"
      "tenant_id   device_id     category    io_mode     io_size_byte    io_depth    perf_mode     target_iops     thread_count    is_sequence     start_s    stop_s\n"
      "1001        1             user        r           16384           10          rolling       0               16              0               0          2\n"
      "1002        1             large       r           16384           10          rolling       4000            16              0               1          3\n"
      "#1002        1             sys         w           2097152         10          rolling       40000           16              0               2          3\n"
      ;
    const int64_t file_len = strlen(file_buf);
    int write_ret = ::write(fd, file_buf, file_len);
    if (write_ret < file_len) {
      LOG_WARN("write file content failed", K(write_ret), K(file_len));
    }
    close(fd);
  }
}

int parse_perf_config(const char *config_file_path,
                      IOPerfScheduler &scheduler_config,
                      ObIArray<IOPerfDevice> &perf_devices,
                      ObIArray<IOPerfTenant> &perf_tenants,
                      ObIArray<IOPerfLoad> &perf_loads)
{
  int ret = OB_SUCCESS;
  bool is_file_exist = false;
  FILE *file = nullptr;
  if (OB_FAIL(FileDirectoryUtils::is_exists(config_file_path, is_file_exist))) {
    LOG_WARN("failed to check file exists", K(ret));
  } else if (!is_file_exist) {
    ret = OB_FILE_NOT_EXIST;
    LOG_WARN("file not exist!");
  } else if (OB_UNLIKELY(nullptr == (file = ::fopen(config_file_path, "r")))) {
    ret = OB_IO_ERROR;
    LOG_WARN("failed to open iops data file, ", K(ret), K(errno), KERRMSG);
  } else {
    char curr_line[1024] = { 0 };
    const char *scheduler_header = "sender_count      schedule_media_id     io_greed";
    const char *device_header = "device_id   media_id    async_channel   sync_channel    max_io_depth    file_size_gb    file_path";
    const char *tenant_header = "tenant_id   min_iops    max_iops    weight          category";
    const char *load_header = "tenant_id   device_id     category    io_mode     io_size_byte    io_depth    perf_mode     target_iops     thread_count    is_sequence     start_s    stop_s";
    enum class PerfConfigType { SCHEDULER, DEVICE, TENANT, LOAD, MAX };
    PerfConfigType config_type = PerfConfigType::MAX;
    while (OB_SUCC(ret)) {
      if (OB_UNLIKELY(nullptr == fgets(curr_line, sizeof(curr_line), file))) {
        ret = OB_ITER_END;
      } else if (' ' == curr_line[0] || '\0' == curr_line[0] || '#' == curr_line[0]) {
        // ignore empty line
      } else if (0 == atoi(curr_line)) {
        if (0 == strncmp(curr_line, scheduler_header, strlen(scheduler_header))) {
          config_type = PerfConfigType::SCHEDULER;
        } else if (0 == strncmp(curr_line, device_header, strlen(device_header))) {
          config_type = PerfConfigType::DEVICE;
        } else if (0 == strncmp(curr_line, tenant_header, strlen(tenant_header))) {
          config_type = PerfConfigType::TENANT;
        } else if (0 == strncmp(curr_line, load_header, strlen(load_header))) {
          config_type = PerfConfigType::LOAD;
        }
      } else if (PerfConfigType::SCHEDULER == config_type) {
        int scan_ret = sscanf(curr_line, "%ld%ld%ld\n",
            &scheduler_config.sender_count_, &scheduler_config.schedule_media_id_, &scheduler_config.io_greed_);
        if (OB_UNLIKELY(3 != scan_ret)) {
          ret = OB_ERR_UNEXPECTED;
          LOG_WARN("scan config file failed", K(ret), K(scan_ret));
        }
        LOG_INFO("wenqu: parse scheduler config", K(ret), K(scheduler_config));
      } else if (PerfConfigType::DEVICE == config_type) {
        IOPerfDevice item;
        int scan_ret = sscanf(curr_line, "%d%d%d%d%d%ld%s\n",
            &item.device_id_, &item.media_id_, &item.async_channel_count_, &item.sync_channel_count_, &item.max_io_depth_, &item.file_size_, item.file_path_);
        if (OB_UNLIKELY(7 != scan_ret)) {
          ret = OB_ERR_UNEXPECTED;
          LOG_WARN("scan config file failed", K(ret), K(scan_ret));
        } else {
          item.file_size_ *= 1024L * 1024L * 1024L;
          if (OB_UNLIKELY(!item.is_valid())) {
            ret = OB_INVALID_DATA;
            LOG_WARN("invalid data", K(ret), K(item));
          } else if (OB_FAIL(perf_devices.push_back(item))) {
            LOG_WARN("add item failed", K(ret), K(item));
          }
          LOG_INFO("wenqu: parse device", K(ret), K(item));
        }
      } else if (PerfConfigType::TENANT == config_type) {
        IOPerfTenant item;
        char category_config[1024] = { 0 };
        int scan_ret = sscanf(curr_line, "%d%ld%ld%ld%[^\n]\n",
            &item.tenant_id_, &item.config_.unit_config_.min_iops_, &item.config_.unit_config_.max_iops_,
            &item.config_.unit_config_.weight_, category_config);
        if (OB_UNLIKELY(5 != scan_ret)) {
          ret = OB_ERR_UNEXPECTED;
          LOG_WARN("scan config file failed", K(ret), K(scan_ret));
        } else {
          item.config_.memory_limit_ = IO_MEMORY_LIMIT;
          item.config_.callback_thread_count_ = 1;
          // parse category config
          if (OB_FAIL(item.config_.parse_category_config(category_config))) {
            LOG_WARN("parse category config failed", K(ret), K(category_config));
          } else if (OB_UNLIKELY(!item.is_valid())) {
            ret = OB_INVALID_DATA;
            LOG_WARN("invalid data", K(ret), K(item));
          } else if (OB_FAIL(perf_tenants.push_back(item))) {
            LOG_WARN("add item failed", K(ret), K(item));
          }
          LOG_INFO("wenqu: parse tenant", K(ret), K(item), K(category_config));
        }
      } else if (PerfConfigType::LOAD == config_type) {
        IOPerfLoad item;
        char category_name[64] = { 0 };
        char io_mode[16] = { 0 };
        char perf_mode[16] = { 0 };
        int scan_ret = sscanf(curr_line, "%d%d%s%s%d%d%s%ld%d%d%ld%ld\n",
            &item.tenant_id_, &item.device_id_, category_name, io_mode,
            &item.size_, &item.depth_, perf_mode, &item.iops_, &item.thread_count_,
            (int *)&item.is_sequence_, &item.start_delay_ts_, &item.stop_delay_ts_);
        if (OB_UNLIKELY(12 != scan_ret)) {
          ret = OB_ERR_UNEXPECTED;
          LOG_WARN("scan config file failed", K(ret), K(scan_ret));
        } else {
          item.start_delay_ts_ *= 1000000L;
          item.stop_delay_ts_ *= 1000000L;
          switch (io_mode[0]) {
            case 'r' :
              item.mode_ = ObIOMode::READ;
              break;
            case 'w':
              item.mode_ = ObIOMode::WRITE;
              break;
            default:
              break;
          }
          const char *rolling_str = "rolling";
          const char *batch_str = "batch";
          const char *async_str = "async";
          if (0 == strncmp(perf_mode, rolling_str, strlen(rolling_str))) {
            item.perf_mode_ = IOPerfMode::ROLLING;
          } else if (0 == strncmp(perf_mode, batch_str, strlen(batch_str))) {
            item.perf_mode_ = IOPerfMode::BATCH;
          } else if (0 == strncmp(perf_mode, async_str, strlen(async_str))) {
            item.perf_mode_ = IOPerfMode::ASYNC;
          }
          for (int64_t i = 0; OB_SUCC(ret) && i < perf_devices.count(); ++i) {
            if (item.device_id_ == perf_devices.at(i).device_id_) {
              item.device_ = &perf_devices.at(i);
              break;
            }
          }
          item.category_ = get_io_category_enum(category_name);
          if (OB_UNLIKELY(!item.is_valid())) {
            ret = OB_INVALID_DATA;
            LOG_WARN("invalid data", K(ret), K(item));
          } else if (OB_FAIL(perf_loads.push_back(item))) {
            LOG_WARN("add item failed", K(ret), K(item));
          }
          LOG_INFO("wenqu: parse load", K(ret), K(item), K(io_mode), K(category_name));
        }
      } else {
        ret = OB_ERR_UNEXPECTED;
        LOG_WARN("config type not match", K(ret), K(config_type));
      }
    }
    if (OB_ITER_END == ret) {
      ret = OB_SUCCESS;
    }
    ::fclose(file);
  }
  return ret;
}

int IOPerfRunner::init(const int64_t absolute_ts, const IOPerfLoad &load)
{
  int ret = OB_SUCCESS;
  if (OB_UNLIKELY(!load.is_valid())) {
    ret = OB_INVALID_ARGUMENT;
    LOG_WARN("invalid argument", K(ret), K(load));
  } else if (OB_FAIL(allocator_.init(OB_MALLOC_BIG_BLOCK_SIZE, "perf runner", OB_SERVER_TENANT_ID, 1024L * 1024L * 1024L * 10L))) {
    LOG_WARN("init allocator failed", K(ret));
  } else if (OB_FAIL(handle_queue_.init(1000L * 10000L, &allocator_))) {
    LOG_WARN("init handle queue failed", K(ret));
  } else {
    abs_ts_ = absolute_ts;
    load_ = load;
    // prepare file
    fd_.first_id_ = ObIOFd::NORMAL_FILE_ID;
    fd_.second_id_ = load.device_->fd_;
    fd_.device_handle_ = load.device_->device_handle_;

    // prepare write buffer
    if (OB_SUCC(ret) && ObIOMode::WRITE == load_.mode_ && nullptr == write_buf_) {
      const int64_t buf_size = load_.size_ + DIO_READ_ALIGN_SIZE;
      void *tmp_buf = ob_malloc(buf_size);
      if (nullptr == tmp_buf) {
        ret = OB_ALLOCATE_MEMORY_FAILED;
        LOG_WARN("allocate memory failed", K(ret), K(buf_size));
      } else {
        memset(tmp_buf, 'a', buf_size);
        write_buf_ = reinterpret_cast<const char *>(tmp_buf);
      }
    }
  }
  if (OB_SUCC(ret)) {
    if (OB_FAIL(set_thread_count(load_.thread_count_ + 2))) {
      LOG_WARN("set thread count failed", K(ret), K(load_)); 
    } else if (OB_FAIL(start())) {
      LOG_WARN("start thread pool failed", K(ret), K(load_));
    }
  }
  if (OB_FAIL(ret)) {
    destroy();
  }
  return ret;
}

void IOPerfRunner::destroy()
{
  stop();
  wait();
  load_ = IOPerfLoad();
  result_ = IOPerfResult();
  last_offset_ = 0;
  if (nullptr != write_buf_) {
    ob_free((void *)write_buf_);
    write_buf_ = nullptr;
  }
  fd_.reset();
}

void IOPerfRunner::run1()
{
  int ret = OB_SUCCESS;
  const int64_t thread_idx = get_thread_idx();
  if (thread_idx < load_.thread_count_) {
  const int64_t current_ts = ObTimeUtility::current_time();
  if (abs_ts_ + load_.start_delay_ts_ > current_ts) {
    usleep(abs_ts_ + load_.start_delay_ts_ - current_ts);
  }
  ATOMIC_CAS(&result_.start_delay_ts_, 0, ObTimeUtility::current_time());
    if (IOPerfMode::ROLLING == load_.perf_mode_)  {
      do_perf_rolling();
    } else {
      do_perf_batch();
    }
  } else if (thread_idx == load_.thread_count_) {
    const int64_t current_ts = ObTimeUtility::current_time();
    if (abs_ts_ + load_.stop_delay_ts_ > current_ts) {
      usleep(abs_ts_ + load_.stop_delay_ts_ - current_ts);
    }
    stop();
    usleep(100L * 1000L);
    wait_handles(handle_queue_);
    result_.stop_delay_ts_ = ObTimeUtility::current_time();
  } else {
    while (!has_set_stop()) {
      usleep(10L * 1000L);
      wait_handles(handle_queue_);
    }
  }
}

int IOPerfRunner::do_perf_batch()
{
  int ret = OB_SUCCESS;
  static const int64_t check_interval_ms = 50L; // 50ms
  int64_t io_count = 0;
  const int64_t check_count = load_.iops_ / load_.thread_count_ / (1000L / check_interval_ms);
  int64_t last_check_ts = ObTimeUtility::fast_current_time();
  const bool need_control_io_speed = 0 != load_.iops_;
  LOG_INFO("perf start", K(load_.tenant_id_), K(load_.category_));
  while (!has_set_stop()) {
    (void) do_batch_io();
    if (need_control_io_speed) {
      io_count += load_.depth_;
      if (io_count > check_count) {
        const int64_t sleep_us = check_interval_ms * 1000L - (ObTimeUtility::fast_current_time() - last_check_ts);
        if (sleep_us > 0) {
          usleep(sleep_us);
        }
        io_count -= check_count;
        last_check_ts = ObTimeUtility::fast_current_time();
      }
    }
  }
  return ret;
}

int IOPerfRunner::do_perf_rolling()
{
  int ret = OB_SUCCESS;
  ObIOInfo info;
  info.tenant_id_ = load_.tenant_id_;
  info.flag_.set_category(load_.category_);
  info.flag_.set_mode(load_.mode_);
  info.flag_.set_wait_event(ObWaitEventIds::DB_FILE_DATA_READ);
  info.fd_ = fd_;
  info.size_ = load_.size_;
  if (ObIOMode::WRITE == load_.mode_) {
    info.buf_ = reinterpret_cast<const char *>(upper_align(reinterpret_cast<int64_t>(write_buf_), DIO_READ_ALIGN_SIZE));
  }
  ObArray<ObIOHandle *> handles;
  RLOCAL(int64_t, local_io_count);
  local_io_count = 0;
  const int64_t max_offset = load_.device_->file_size_ - load_.size_;
  for (int32_t i = 0; OB_SUCC(ret) && i < load_.depth_; ++i) {
    void *buf = nullptr;
    if (OB_ISNULL(buf = allocator_.alloc(sizeof(ObIOHandle)))) {
      ret = OB_ALLOCATE_MEMORY_FAILED;
      LOG_WARN("allocate memory failed", K(ret));
    } else {
      ObIOHandle *cur_handle  = new (buf) ObIOHandle;
      if (load_.is_sequence_) {
        last_offset_ += load_.size_;
        info.offset_ = last_offset_ % (max_offset);
      } else {
        info.offset_ = lower_align(ObRandom::rand(0, max_offset), DIO_READ_ALIGN_SIZE);
      }
      if (info.flag_.is_read()) {
        if (OB_FAIL(ObIOManager::get_instance().aio_read(info, *cur_handle))) {
          LOG_WARN("fail to aio read", K(ret), K(i));
        }
      } else if (info.flag_.is_write()) {
        if (OB_FAIL(ObIOManager::get_instance().aio_write(info, *cur_handle))) {
          LOG_WARN("fail to aio write", K(ret), K(i));
        }
      }
      if (OB_SUCC(ret)) {
        ++local_io_count;
        if (OB_FAIL(handles.push_back(cur_handle))) {
          LOG_WARN("push back handle failed", K(ret), KPC(cur_handle));
        }
      }
    }
  }
  int64_t pos = 0;
  while (!has_set_stop() && OB_SUCC(ret)) {
    if (TC_REACH_TIME_INTERVAL(1000L * 1000L)) {
      ATOMIC_FAA(&io_count_, local_io_count);
      ATOMIC_FAA(&total_io_count_, local_io_count);
      local_io_count = 0;
      ATOMIC_INC(&report_count_);
      if (load_.thread_count_ == ATOMIC_LOAD(&report_count_)) {
//        LOG_INFO("IO STATUS: perf io count", K(io_count_), K(load_.category_), K(total_io_count_));
        report_count_ = 0;
        io_count_ = 0;
      }
    }
    pos = pos % handles.count();
    ObIOHandle *cur_handle = handles[pos];
    wait_and_count(*cur_handle);
    if (load_.is_sequence_) {
      last_offset_ += load_.size_;
      info.offset_ = last_offset_ % (max_offset);
    } else {
      info.offset_ = lower_align(ObRandom::rand(0, max_offset), DIO_READ_ALIGN_SIZE);
    }
    if (info.flag_.is_read()) {
      if (OB_FAIL(ObIOManager::get_instance().aio_read(info, *cur_handle))) {
        LOG_WARN("fail to aio read", K(ret));
      }
    } else if (info.flag_.is_write()) {
      if (OB_FAIL(ObIOManager::get_instance().aio_write(info, *cur_handle))) {
        LOG_WARN("fail to aio write", K(ret));
      }
    }
    if (OB_SUCC(ret)) {
      ++local_io_count;
    }
    ++pos;
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < handles.count(); ++i) {
    int64_t tmp_pos = (pos + i) % handles.count();
    ObIOHandle *cur_handle = handles[tmp_pos];
    wait_and_count(*cur_handle);
    cur_handle->~ObIOHandle();
    allocator_.free(cur_handle);
  }
  return ret;
}

int IOPerfRunner::do_batch_io()
{
  int ret = OB_SUCCESS;
  ObIOInfo info;
  info.tenant_id_ = load_.tenant_id_;
  info.flag_.set_category(load_.category_);
  info.flag_.set_mode(load_.mode_);
  info.flag_.set_wait_event(ObWaitEventIds::DB_FILE_DATA_READ);
  info.fd_ = fd_;
  info.size_ = load_.size_;
  if (ObIOMode::WRITE == load_.mode_) {
    info.buf_ = reinterpret_cast<const char *>(upper_align(reinterpret_cast<int64_t>(write_buf_), DIO_READ_ALIGN_SIZE));
  }
  ObFixedQueue<ObIOHandle> local_handles;
  if (OB_FAIL(local_handles.init(load_.depth_ * 2))) {
    LOG_WARN("init handles array failed", K(ret));
  }
  ObFixedQueue<ObIOHandle> &handles = IOPerfMode::ASYNC == load_.perf_mode_ ? handle_queue_ : local_handles;
  const int64_t max_offset = load_.device_->file_size_ - load_.size_;
  for (int32_t i = 0; OB_SUCC(ret) && i < load_.depth_; ++i) {
    void *buf = nullptr;
    if (OB_ISNULL(buf = allocator_.alloc(sizeof(ObIOHandle)))) {
      ret = OB_ALLOCATE_MEMORY_FAILED;
      LOG_WARN("allocate memory failed", K(ret));
    } else {
      ObIOHandle *cur_handle  = new (buf) ObIOHandle;
      if (load_.is_sequence_) {
        last_offset_ += load_.size_;
        info.offset_ = last_offset_ % (max_offset);
      } else {
        info.offset_ = lower_align(ObRandom::rand(0, max_offset), DIO_READ_ALIGN_SIZE);
      }
      if (info.flag_.is_read()) {
        if (OB_FAIL(ObIOManager::get_instance().aio_read(info, *cur_handle))) {
          LOG_WARN("fail to aio read", K(ret), K(i));
        }
      } else if (info.flag_.is_write()) {
        if (OB_FAIL(ObIOManager::get_instance().aio_write(info, *cur_handle))) {
          LOG_WARN("fail to aio write", K(ret), K(i));
        }
      }
      if (OB_FAIL(ret)) {
      } else if (OB_FAIL(handles.push(cur_handle))) {
        LOG_WARN("push back handle failed", K(ret), KPC(cur_handle));
      } else if (local_handles.get_total() > 0) {
        wait_handles(local_handles);
      }
    }
  }
  return ret;
}

int IOPerfRunner::wait_and_count(ObIOHandle &io_handle)
{
  int ret = OB_SUCCESS;
  if (OB_FAIL(io_handle.wait(DEFAULT_IO_WAIT_TIME_MS))) {
    LOG_WARN("fail to wait read io", K(ret));
    ATOMIC_INC(&result_.fail_count_);
  } else {
    ATOMIC_INC(&result_.succ_count_);
  }
  const int64_t delay = max(0, io_handle.get_rt());
  const int64_t disk_delay = max(0, io_handle.req_->time_log_.return_ts_ - io_handle.req_->time_log_.submit_ts_);
  ATOMIC_FAA(&result_.sum_rt_, delay);
  ATOMIC_FAA(&result_.disk_rt_, disk_delay);
  return ret;
}

int IOPerfRunner::wait_handles(ObFixedQueue<ObIOHandle> &handles)
{
  int ret = OB_SUCCESS;
  ObIOHandle *cur_handle = nullptr;
  while (OB_SUCC(handles.pop(cur_handle))) {
    wait_and_count(*cur_handle);
    cur_handle->~ObIOHandle();
    allocator_.free(cur_handle);
  }
  return ret;
}

int IOPerfRunner::print_result()
{
  int ret = OB_SUCCESS;
  if (!(result_.start_delay_ts_ > 0 && result_.stop_delay_ts_ > 0 && result_.stop_delay_ts_ > result_.start_delay_ts_)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("io perf not finished", K(ret), K(result_));
  } else {
    LOG_INFO("io perf result",
        "perf_config", load_,
        "succ_cnt", result_.succ_count_,
        "fail_cnt", result_.fail_count_,
        "time_ms", (result_.stop_delay_ts_ - result_.start_delay_ts_) / 1000,
        "iops", result_.succ_count_ * 1000000L / max(1, result_.stop_delay_ts_ - result_.start_delay_ts_),
        "rt", result_.sum_rt_ / max(1, result_.succ_count_),
        "disk_rt", result_.disk_rt_ / max(1, result_.succ_count_));
  }
  return ret;
}

int IOConfModify::init(int64_t modify_init_ts, int64_t modify_delay_ts, const IOPerfTenant &curr_tenant)
{
  int ret = OB_SUCCESS;
  if (OB_UNLIKELY(!curr_tenant.is_valid())) {
    ret = OB_INVALID_ARGUMENT;
    LOG_WARN("invalid argument", K(ret), K(curr_tenant));
  } else if (OB_FAIL(allocator_.init(OB_MALLOC_BIG_BLOCK_SIZE, "perf runner", OB_SERVER_TENANT_ID, 1024L * 1024L * 1024L * 10L))) {
    LOG_WARN("init allocator failed", K(ret));
  } else {
    curr_tenant_ = curr_tenant;
    modify_init_ts_ = modify_init_ts;
    modify_delay_ts_ = modify_delay_ts;
  }
  if (OB_SUCC(ret)) {
    if (OB_FAIL(set_thread_count(load_.thread_count_ + 1))) {
      LOG_WARN("set thread count failed", K(ret), K(modify_init_ts_), K(curr_tenant_)); 
    } else if (OB_FAIL(start())) {
      LOG_WARN("start thread failed", K(ret), K(modify_init_ts_), K(curr_tenant_));
    }
  }
  if (OB_FAIL(ret)) {
    destroy();
  }
  return ret;
}

void IOConfModify::destroy()
{
  stop();
  wait();
  curr_tenant_ = IOPerfTenant();
}

void IOConfModify::run1()
{
  int ret = OB_SUCCESS;
  const int64_t thread_idx = get_thread_idx();
  LOG_INFO("modify thread start");
  const int64_t current_ts = ObTimeUtility::current_time();
  if (modify_init_ts_ + modify_delay_ts_ > current_ts) {
    usleep(modify_init_ts_ + modify_delay_ts_ - current_ts);
  }
  int64_t min_test = 50000;
  int64_t max_test = 1000000; 
  int64_t weight_test = 1000;
  if (OB_FAIL(modify_tenant_io(min_test, max_test, weight_test, curr_tenant_))) {
    LOG_WARN("modify config failed", K(ret), K(curr_tenant_));
  }
}
int IOConfModify::modify_tenant_io( const int64_t min_iops,
                                    const int64_t max_iops,
                                    const int64_t weight,
                                    IOPerfTenant &curr_tenant)
{
  int ret = OB_SUCCESS;
  //ObTenantIOConfig io_config;
  curr_tenant.config_.unit_config_.min_iops_ = min_iops;
  curr_tenant.config_.unit_config_.max_iops_ = max_iops;
  curr_tenant.config_.unit_config_.weight_ = weight;

  if (OB_FAIL(OB_IO_MANAGER.refresh_tenant_io_config(curr_tenant.tenant_id_, curr_tenant.config_))) {
    LOG_WARN("refresh tenant io config failed", K(ret), K(curr_tenant.tenant_id_), K(curr_tenant.config_));
  }
  return ret;
}