/**
 * 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 SQL_EXE

#include "sql/executor/ob_job_control.h"
#include "sql/executor/ob_task_event.h"
#include "lib/utility/ob_tracepoint.h"
#include "sql/engine/ob_exec_context.h"
#include "sql/engine/px/exchange/ob_transmit_op.h"

namespace oceanbase
{
namespace sql
{
using namespace oceanbase::common;

volatile uint64_t ObJobControl::global_job_id_ = 0;

ObJobControl::ObJobControl() : jobs_()
{
}

ObJobControl::~ObJobControl()
{
  for (int64_t i = 0; i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (NULL != job) {
      job->~ObJob();
    }
  }
}

void ObJobControl::reset()
{
  for (int64_t i = 0; i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (NULL != job) {
      job->reset();
    }
  }
  jobs_.reset();
}

int ObJobControl::all_jobs_finished(bool &is_finished) const
{
  int ret = OB_SUCCESS;
  bool finished = true;
  for (int64_t i = 0; OB_SUCC(ret) && true == finished
       && i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (OB_I(t1) OB_ISNULL(job)) {
      ret = OB_ERR_UNEXPECTED;
      finished = false;
      LOG_ERROR("job is NULL", K(i), K(ret));
    } else if (OB_JOB_STATE_FINISHED != job->get_state()) {
      finished = false;
    } else {
      //empty
    }
  }
  is_finished = finished;
  return ret;
}

int ObJobControl::all_jobs_finished_except_root_job(bool &is_finished) const
{
  int ret = OB_SUCCESS;
  bool finished = true;
  if (jobs_.count() < 1) {
    ret = OB_ERR_UNEXPECTED;
    finished = false;
    LOG_ERROR("count of jobs is less than 1",
              K(ret), "job_count", jobs_.count());
  } else {
    for (int64_t i = 0; OB_SUCC(ret) && true == finished
         && i < jobs_.count() - 1; ++i) {
      ObJob *job = jobs_.at(i);
      if (OB_I(t1) OB_ISNULL(job)) {
        ret = OB_ERR_UNEXPECTED;
        finished = false;
        LOG_ERROR("job is NULL", K(i), K(ret));
      } else if (OB_JOB_STATE_FINISHED != job->get_state()) {
        finished = false;
      } else {
        //empty
      }
    }
  }
  is_finished = finished;
  return ret;
}

int ObJobControl::create_job(ObIAllocator &allocator,
                             const ObExecutionID &ob_execution_id,
                             uint64_t root_op_id,
                             ObJob *&job) const
{
  int ret = OB_SUCCESS;
  void *tmp = NULL;
  job = NULL;
  if (OB_I(t1) OB_ISNULL(tmp = allocator.alloc(sizeof(ObJob)))) {
    ret = OB_ALLOCATE_MEMORY_FAILED;
    LOG_ERROR("fail to alloc ObJob", K(ret), K(ob_execution_id));
  } else if (OB_I(t2) OB_ISNULL(job = new(tmp) ObJob)) {
    ret = OB_ERR_UNEXPECTED;
    LOG_WARN("fail to new ObJob", K(ret), K(ob_execution_id));
  } else {
    // job_id之所以全局递增，是因为execution_id是从ObIDMap的assign函数中获取的，
    // 有可能会重复，所以job_id在本进程内必须不能重复。
    ObJobID ob_job_id;
    uint64_t job_id = ATOMIC_FAA(&global_job_id_, 1);
    ob_job_id.set_ob_execution_id(ob_execution_id);
    ob_job_id.set_job_id(job_id);
    ob_job_id.set_root_op_id(root_op_id);
    job->set_ob_job_id(ob_job_id);
  }
  return ret;
}

int ObJobControl::find_job_by_job_id(uint64_t job_id, ObJob *&job) const
{
  int ret = OB_ENTRY_NOT_EXIST;
  for (int64_t i = 0; OB_ENTRY_NOT_EXIST == ret
       && i < jobs_.count(); ++i) {
    ObJob *tmp_job = jobs_.at(i);
    if (OB_I(t1) OB_ISNULL(tmp_job)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_ERROR("job is NULL", K(ret));
    } else if (tmp_job->get_job_id() == job_id) {
      job = tmp_job;
      ret = OB_SUCCESS;
    }
  }
  return ret;
}

int ObJobControl::find_job_by_root_op_id(uint64_t root_op_id, ObJob *&job) const
{
  int ret = OB_ENTRY_NOT_EXIST;
  for (int64_t i = 0; OB_ENTRY_NOT_EXIST == ret
       && i < jobs_.count(); ++i) {
    ObJob *tmp_job = jobs_.at(i);
    if (OB_I(t1) OB_ISNULL(tmp_job)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_ERROR("job is NULL", K(ret));
    } else if (tmp_job->get_root_op_id() == root_op_id) {
      job = tmp_job;
      ret = OB_SUCCESS;
    }
  }
  return ret;
}

int ObJobControl::get_running_jobs(ObIArray<ObJob *> &jobs) const
{
  int ret = OB_SUCCESS;
  jobs.reset();
  for (int64_t i = 0; OB_SUCC(ret) && i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (OB_I(t1) OB_ISNULL(job)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_ERROR("job is NULL", K(ret));
    } else if (OB_JOB_STATE_RUNNING == job->get_state() && OB_FAIL(jobs.push_back(job))) {
      LOG_WARN("fail to push back job", K(ret), K(*job));
    }
  }
  return ret;
}

int ObJobControl::get_all_jobs(ObIArray<ObJob *> &jobs) const
{
  int ret = OB_SUCCESS;
  jobs.reset();
  for (int64_t i = 0; OB_SUCC(ret) && i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (OB_I(t1) OB_ISNULL(job)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_ERROR("job is NULL", K(ret));
    } else if (OB_FAIL(jobs.push_back(job))) {
      LOG_WARN("fail to push back job", K(ret), K(*job));
    }
  }
  return ret;
}

int ObJobControl::get_all_jobs_except_root_job(ObIArray<ObJob *> &jobs) const
{
  int ret = OB_SUCCESS;
  jobs.reset();
  for (int64_t i = 0; OB_SUCC(ret) && i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (OB_I(t1) OB_ISNULL(job)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_ERROR("job is NULL", K(ret));
    } else if (!job->is_root_job() && OB_FAIL(jobs.push_back(job))) {
      LOG_WARN("fail to push back job", K(ret), K(*job));
    }
  }
  return ret;
}

int ObJobControl::sort_job_scan_part_locs(ObExecContext &ctx)
{
  UNUSED(ctx);
  return OB_SUCCESS;
}

int ObJobControl::init_job_finish_queue(ObExecContext &ctx)
{
  UNUSED(ctx);
  return OB_SUCCESS;
}
//int ObJobControl::arrange_jobs()
//{
//  return jobs_quick_sort(jobs_, 0, jobs_.count() - 1);
//}
//
//int ObJobControl::jobs_quick_sort(ObIArray<ObJob *> &jobs,
//                                  int64_t low,
//                                  int64_t high)
//{
//  int ret = OB_SUCCESS;
//  if (low < high) {
//    int64_t i = low;
//    int64_t j = high;
//    ObJob *temp = jobs.at(i);
//    if (OB_ISNULL(temp)) {
//      ret = OB_ERR_UNEXPECTED;
//      LOG_WARN("job is NULL", K(ret), K(i), K(low), K(high));
//    }
//    while (OB_SUCC(ret) && i < j) {
//      while ((jobs.at(j)->get_priority() <= temp->get_priority()) && (i < j)) {
//        j--;
//      }
//      jobs.at(i) = jobs.at(j);
//      while ((jobs.at(i)->get_priority() >= temp->get_priority()) && (i < j)) {
//        i++;
//      }
//      jobs.at(j) = jobs.at(i);
//    }
//    if (OB_SUCC(ret)) {
//      jobs.at(i) = temp;
//      if (OB_FAIL(jobs_quick_sort(jobs, low, i - 1))) {
//        LOG_WARN("fail to jobs quick sort left", K(ret),
//                 K(low), K(high), K(i), K(j));
//      } else if (OB_FAIL(jobs_quick_sort(jobs, j + 1, high))) {
//        LOG_WARN("fail to jobs quick sort right", K(ret),
//                 K(low), K(high), K(i), K(j));
//      }
//    }
//  }
//  return ret;
//}

// 如果超过buf_len则会被截断
int ObJobControl::print_status(char *buf, int64_t buf_len,
                               bool ignore_normal_state/* = false*/) const
{
  int ret = OB_SUCCESS;
  int64_t pos = 0;
  if (OB_FAIL(J_OBJ_START())) {
    LOG_WARN("fail to print obj start", K(ret));
  } else {
    J_KV(N_JOB_COUNT, jobs_.count());
    if (OB_FAIL(J_COMMA())) {
      LOG_WARN("fail to print comma", K(ret));
    }
  }
  for (int64_t i = 0; OB_SUCC(ret) && i < jobs_.count(); ++i) {
    ObJob *job = jobs_.at(i);
    if (OB_ISNULL(job)) {
      ret = OB_ERR_UNEXPECTED;
      LOG_WARN("job is NULL", K(ret), K(i));
    } else if (OB_FAIL(job->print_status(buf, buf_len, pos, ignore_normal_state))) {
      LOG_WARN("fail to print job status", K(ret), K(i), K(*job));
    } else if (i < jobs_.count() - 1 && OB_FAIL(J_COMMA())) {
      LOG_WARN("fail to print comma", K(ret), K(i), K(*job));
    }
  }
  if (OB_FAIL(ret)) {
  } else if (OB_FAIL(J_OBJ_END())) {
    LOG_WARN("fail to print obj end", K(ret));
  }
  if (OB_SIZE_OVERFLOW == ret) {
    LOG_WARN("buf overflow, truncate it", K(ret), K(buf_len), K(pos));
    ret = OB_SUCCESS;
  }
  return ret;
}

DEF_TO_STRING(ObJobControl)
{
  int64_t pos = 0;
  J_OBJ_START();
  J_NAME(N_JOB_TREE);
  J_COLON();
  print_job_tree(buf, buf_len, pos, jobs_.at(0));
  J_OBJ_END();
  return pos;
}

void ObJobControl::print_job_tree(char *buf, const int64_t buf_len, int64_t &pos, ObJob *job) const
{
  J_OBJ_START();
  J_KV(N_JOB, job);
  int64_t child_count = job->get_child_count();
  if (child_count > 0) {
    J_COMMA();
    J_NAME(N_CHILD_JOB);
    J_COLON();
    J_ARRAY_START();
    ObJob *child_job = NULL;
    for (int64_t i = 0; i < child_count; i++) {
      if (i > 0) {
        J_COMMA();
      }
      (void)job->get_child_job(i, child_job);
      print_job_tree(buf, buf_len, pos, child_job);
    }
    J_ARRAY_END();
  }
  J_OBJ_END();
}

}
}