text: /*
   * The MIT License (MIT)
   *
   * Copyright (c) 2015 Microsoft Corporation
   *
   * -=- Robust Distributed System Nucleus (rDSN) -=-
   *
   * Permission is hereby granted, free of charge, to any person obtaining a copy
   * of this software and associated documentation files (the "Software"), to deal
   * in the Software without restriction, including without limitation the rights
   * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   * copies of the Software, and to permit persons to whom the Software is
   * furnished to do so, subject to the following conditions:
   *
   * The above copyright notice and this permission notice shall be included in
   * all copies or substantial portions of the Software.
   *
   * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
   * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
   * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
   * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
   * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
   * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
   * THE SOFTWARE.
   */
  
  /*
   * Description:
   *     What is this file about?
   *
   * Revision history:
   *     xxxx-xx-xx, author, first version
   *     xxxx-xx-xx, author, fix bug about xxx
   */
  
  #include <dsn/tool-api/task_spec.h>
  #include <dsn/tool-api/command_manager.h>
  #include <dsn/tool-api/threadpool_spec.h>
  #include <dsn/utility/smart_pointers.h>
  
  namespace dsn {
  
  constexpr int TASK_SPEC_STORE_CAPACITY = 512;
  
  // A sequential storage maps task_code to task_spec.
  static std::array<std::unique_ptr<task_spec>, TASK_SPEC_STORE_CAPACITY> s_task_spec_store;
  
  void task_spec::register_task_code(task_code code,
                                     dsn_task_type_t type,
                                     dsn_task_priority_t pri,
                                     dsn::threadpool_code pool)
  {
      dassert(code < TASK_SPEC_STORE_CAPACITY, "code = %d", code);
      if (!s_task_spec_store[code]) {
          s_task_spec_store[code] = make_unique<task_spec>(code, code.to_string(), type, pri, pool);
          auto &spec = s_task_spec_store[code];
  
          if (type == TASK_TYPE_RPC_REQUEST) {
              std::string ack_name = std::string(code.to_string()) + std::string("_ACK");
              // for a rpc request, we firstly register it's ack code to invalid threadpool,
              // then the response code's definition will reassign a proper valid threadpool code.
              // please refer to the DEFINE_TASK_CODE_RPC/DEFINE_STORAGE_RPC_CODE in task_code.h
              // for more details.
              dsn::task_code ack_code(
                  ack_name.c_str(), TASK_TYPE_RPC_RESPONSE, pri, THREAD_POOL_INVALID);
              spec->rpc_paired_code = ack_code;
              task_spec::get(ack_code.code())->rpc_paired_code = code;
          }
      } else {
          auto spec = task_spec::get(code);
          if (spec->type != type) {
              dassert(
                  false,
                  "task code %s registerd for %s, which does not match with previously registered %s",
                  code.to_string(),
                  enum_to_string(type),
                  enum_to_string(spec->type));
              return;
          }
  
          if (spec->priority != pri) {
              dwarn("overwrite priority for task %s from %s to %s",
                    code.to_string(),
                    enum_to_string(spec->priority),
                    enum_to_string(pri));
              spec->priority = pri;
          }
  
          if (spec->pool_code != pool) {
              if (spec->pool_code != THREAD_POOL_INVALID) {
                  dwarn("overwrite default thread pool for task %s from %s to %s",
                        code.to_string(),
                        spec->pool_code.to_string(),
                        pool.to_string());
              }
              spec->pool_code = pool;
          }
      }
  }
  
  void task_spec::register_storage_task_code(task_code code,
                                             dsn_task_type_t type,
                                             dsn_task_priority_t pri,
                                             threadpool_code pool,
                                             bool is_write_operation,
                                             bool allow_batch,
                                             bool is_idempotent)
  {
      register_task_code(code, type, pri, pool);
      task_spec *spec = task_spec::get(code);
      spec->rpc_request_for_storage = true;
      spec->rpc_request_is_write_operation = is_write_operation;
      spec->rpc_request_is_write_allow_batch = allow_batch;
      spec->rpc_request_is_write_idempotent = is_idempotent;
  }
  
  task_spec *task_spec::get(int code)
  {
      return code < TASK_SPEC_STORE_CAPACITY ? s_task_spec_store[code].get() : nullptr;
  }
  
  task_spec::task_spec(int code,
                       const char *name,
                       dsn_task_type_t type,
                       dsn_task_priority_t pri,
                       dsn::threadpool_code pool)
      : code(code),
        type(type),
        name(name),
        rpc_paired_code(TASK_CODE_INVALID),
        rpc_request_for_storage(false),
        rpc_request_is_write_operation(false),
        rpc_request_is_write_allow_batch(false),
        rpc_request_is_write_idempotent(false),
        priority(pri),
        pool_code(pool),
        rpc_call_header_format(NET_HDR_DSN),
        rpc_call_channel(RPC_CHANNEL_TCP),
        rpc_message_crc_required(false),
        on_task_create((std::string(name) + std::string(".create")).c_str()),
        on_task_enqueue((std::string(name) + std::string(".enqueue")).c_str()),
        on_task_begin((std::string(name) + std::string(".begin")).c_str()),
        on_task_end((std::string(name) + std::string(".end")).c_str()),
        on_task_cancelled((std::string(name) + std::string(".cancelled")).c_str()),
  
        on_task_wait_pre((std::string(name) + std::string(".wait.pre")).c_str()),
        on_task_wait_notified((std::string(name) + std::string(".wait.notified")).c_str()),
        on_task_wait_post((std::string(name) + std::string(".wait.post")).c_str()),
        on_task_cancel_post((std::string(name) + std::string(".cancel.post")).c_str()),
  
        on_aio_call((std::string(name) + std::string(".aio.call")).c_str()),
        on_aio_enqueue((std::string(name) + std::string(".aio.enqueue")).c_str()),
  
        on_rpc_call((std::string(name) + std::string(".rpc.call")).c_str()),
        on_rpc_request_enqueue((std::string(name) + std::string(".rpc.request.enqueue")).c_str()),
        on_rpc_reply((std::string(name) + std::string(".rpc.reply")).c_str()),
        on_rpc_response_enqueue((std::string(name) + std::string(".rpc.response.enqueue")).c_str()),
        on_rpc_create_response((std::string(name) + std::string("rpc.create.response")).c_str())
  {
      dassert(strlen(name) < DSN_MAX_TASK_CODE_NAME_LENGTH,
              "task code name '%s' is too long: length must be smaller than "
              "DSN_MAX_TASK_CODE_NAME_LENGTH (%u)",
              name,
              DSN_MAX_TASK_CODE_NAME_LENGTH);
  
      rejection_handler = nullptr;
      rpc_call_channel = RPC_CHANNEL_TCP;
      rpc_timeout_milliseconds = 5 * 1000; // 5 seconds
  }
  
  bool task_spec::init()
  {
      /*
      [task..default]
      is_trace = false
      is_profile = false
  
      [task.RPC_PREPARE]
      pool_code = THREAD_POOL_REPLICATION
      priority = TASK_PRIORITY_HIGH
      is_trace = true
      is_profile = true
      */
  
      task_spec default_spec(
          0, "placeholder", TASK_TYPE_COMPUTE, TASK_PRIORITY_COMMON, THREAD_POOL_DEFAULT);
      if (!read_config("task..default", default_spec))
          return false;
  
      for (int code = 0; code <= dsn::task_code::max(); code++) {
          if (code == TASK_CODE_INVALID)
              continue;
  
          std::string section_name =
              std::string("task.") + std::string(dsn::task_code(code).to_string());
          task_spec *spec = task_spec::get(code);
          dassert(spec != nullptr, "task_spec cannot be null");
  
          if (!read_config(section_name.c_str(), *spec, &default_spec))
              return false;
  
          if (code == TASK_CODE_EXEC_INLINED) {
              spec->allow_inline = true;
          }
  
          dassert(spec->rpc_request_delays_milliseconds.size() == 0 ||
                      spec->rpc_request_delays_milliseconds.size() == 6,
                  "invalid length of rpc_request_delays_milliseconds, must be of length 6");
          if (spec->rpc_request_delays_milliseconds.size() > 0) {
              spec->rpc_request_delayer.initialize(spec->rpc_request_delays_milliseconds);
          }
  
          if (spec->rpc_request_throttling_mode != TM_NONE) {
              if (spec->type != TASK_TYPE_RPC_REQUEST) {
                  derror("%s: only rpc request type can have non TM_NONE throttling_mode",
                         spec->name.c_str());
                  return false;
              }
          }
      }
  
      ::dsn::command_manager::instance().register_command(
          {"task-code"},
          "task-code - query task code containing any given keywords",
          "task-code keyword1 keyword2 ...",
          [](const std::vector<std::string> &args) {
              std::stringstream ss;
  
              for (int code = 0; code <= dsn::task_code::max(); code++) {
                  if (code == TASK_CODE_INVALID)
                      continue;
  
                  std::string codes = dsn::task_code(code).to_string();
                  if (args.size() == 0) {
                      ss << "    " << codes << std::endl;
                  } else {
                      for (auto &arg : args) {
                          if (codes.find(arg.c_str()) != std::string::npos) {
                              ss << "    " << codes << std::endl;
                          }
                      }
                  }
              }
              return ss.str();
          });
  
      return true;
  }
  
  bool threadpool_spec::init(/*out*/ std::vector<threadpool_spec> &specs)
  {
      /*
      [threadpool..default]
      worker_count = 4
      worker_priority = THREAD_xPRIORITY_NORMAL
      partitioned = false
      queue_aspects = xxx
      worker_aspects = xxx
      admission_controller_factory_name = xxx
      admission_controller_arguments = xxx
  
      [threadpool.THREAD_POOL_REPLICATION]
      name = Thr.replication
      run = true
      worker_count = 4
      worker_priority = THREAD_xPRIORITY_NORMAL
      partitioned = false
      queue_aspects = xxx
      worker_aspects = xxx
      admission_controller_factory_name = xxx
      admission_controller_arguments = xxx
      */
  
      threadpool_spec default_spec(THREAD_POOL_INVALID);
      if (false == read_config("threadpool..default", default_spec, nullptr))
          return false;
  
      default_spec.name = "";
      specs.clear();
      for (int code = 0; code <= threadpool_code::max(); code++) {
          std::string code_name = std::string(threadpool_code(code).to_string());
          std::string section_name = std::string("threadpool.") + code_name;
          threadpool_spec spec(default_spec);
          if (false == read_config(section_name.c_str(), spec, &default_spec))
              return false;
  
          spec.pool_code = threadpool_code(code);
  
          if ("" == spec.name)
              spec.name = code_name;
  
          if (false == spec.worker_share_core && 0 == spec.worker_affinity_mask) {
              spec.worker_affinity_mask = (1 << std::thread::hardware_concurrency()) - 1;
          }
  
          specs.push_back(spec);
      }
  
      return true;
  }
  
  } // end namespace
  