ExecutorResourceMgr.h

Go to the documentation of this file.

/*
 * Copyright 2022 HEAVY.AI, Inc.
 *
 * 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.
 */

#pragma once

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <set>
#include <shared_mutex>

#include "ExecutorResourceMgrCommon.h"
#include "ExecutorResourcePool.h"
#include "OutstandingQueueRequests.h"

namespace ExecutorResourceMgr_Namespace {

struct ExecutorStats {
  size_t requests{0};
  size_t cpu_requests{0};
  size_t gpu_requests{0};
  size_t queue_length{0};
  size_t cpu_queue_length{0};
  size_t gpu_queue_length{0};
  size_t total_queue_time_ms{0};
  size_t total_cpu_queue_time_ms{0};
  size_t total_gpu_queue_time_ms{0};
  size_t requests_actually_queued{0};
  size_t cpu_requests_actually_queued{0};
  size_t gpu_requests_actually_queued{0};
  size_t sum_queue_size_at_entry{0};
  size_t sum_cpu_queue_size_at_entry{0};
  size_t sum_gpu_queue_size_at_entry{0};
  size_t requests_executing{0};
  size_t cpu_requests_executing{0};
  size_t gpu_requests_executing{0};
  size_t requests_executed{0};
  size_t cpu_requests_executed{0};
  size_t gpu_requests_executed{0};
  size_t total_execution_time_ms{0};
  size_t total_cpu_execution_time_ms{0};
  size_t total_gpu_execution_time_ms{0};
  size_t total_time_ms{0};
  size_t total_cpu_time_ms{0};
  size_t total_gpu_time_ms{0};
  size_t requests_with_timeouts{0};
  size_t requests_timed_out{0};
};

struct RequestStats {
  const RequestId request_id;
  const RequestInfo request_info;
  const ResourceGrant min_resource_grant;
  const ResourceGrant max_resource_grant;
  ResourceGrant actual_resource_grant;
  std::chrono::steady_clock::time_point enqueue_time;             // in ms
  std::chrono::steady_clock::time_point deque_time;               // in ms
  std::chrono::steady_clock::time_point execution_finished_time;  // in ms
  size_t queue_length_at_entry;
  size_t device_type_queue_length_at_entry;
  bool finished_queueing{false};
  bool finished_executing{false};
  size_t queue_time_ms{0};
  size_t execution_time_ms{0};
  size_t total_time_ms{0};
  size_t timeout_in_ms{0};
  bool timed_out{false};
  // this variable will be filled w/ a corresponding msg when an error is occurred
  // when processing the resource allocation request by ERM
  std::optional<std::string> error;

  RequestStats(const RequestId request_id,
               const RequestInfo& request_info,
               const ResourceGrant& min_resource_grant,
               const ResourceGrant& max_resource_grant,
               const std::chrono::steady_clock::time_point& enqueue_time,
               const size_t queue_length_at_entry,
               const size_t device_type_queue_length_at_entry,
               const size_t timeout_in_ms)
      : request_id(request_id)
      , request_info(request_info)
      , min_resource_grant(min_resource_grant)
      , max_resource_grant(max_resource_grant)
      , enqueue_time(enqueue_time)
      , queue_length_at_entry(queue_length_at_entry)
      , device_type_queue_length_at_entry(device_type_queue_length_at_entry)
      , timeout_in_ms(timeout_in_ms) {}
};

enum ExecutionRequestStage { QUEUED, EXECUTING };

class ExecutorResourceHandle;  // forward declaration

class ExecutorResourceMgr : public std::enable_shared_from_this<ExecutorResourceMgr> {
 public:
  ExecutorResourceMgr(
      const std::vector<std::pair<ResourceType, size_t>>& total_resources,
      const std::vector<ConcurrentResourceGrantPolicy>&
          concurrent_resource_grant_policies,
      const std::vector<ResourceGrantPolicy>& max_per_request_resource_grant_policies,
      const double max_available_resource_use_ratio);

  ~ExecutorResourceMgr();

  std::unique_ptr<ExecutorResourceHandle> request_resources_with_timeout(
      const RequestInfo& request_info,
      const size_t timeout_in_ms);

  std::unique_ptr<ExecutorResourceHandle> request_resources(
      const RequestInfo& request_info);

  void release_resources(const RequestId request_id, const ResourceGrant& resource_grant);

  ExecutorStats get_executor_stats() const;

  void print_executor_stats() const;

  std::pair<size_t, size_t> get_resource_info(const ResourceType resource_type) const {
    return executor_resource_pool_.get_resource_info(resource_type);
  }

  ResourcePoolInfo get_resource_info() const {
    return executor_resource_pool_.get_resource_info();
  }

  void set_resource(const ResourceType resource_type, const size_t resoure_quantity);

  ConcurrentResourceGrantPolicy get_concurrent_resource_grant_policy(
      const ResourceType resource_type) const;

  void set_concurrent_resource_grant_policy(
      const ConcurrentResourceGrantPolicy& concurrent_resource_grant_policy);

  void pause_process_queue();

  void resume_process_queue();

 private:
  void process_queue_loop();

  RequestStats get_request_for_id(const RequestId request_id) const;

  void mark_request_error(const RequestId request_id, std::string error_msg);

  RequestId choose_next_request();

  RequestId enqueue_request(const RequestInfo& request_info,
                            const size_t timeout_in_ms,
                            const ResourceGrant& min_resource_grant,
                            const ResourceGrant& max_resource_grant);

  void mark_request_dequed(const RequestId request_id);

  void mark_request_timed_out(const RequestId request_id);

  void mark_request_finished(const RequestId request_id);

  void set_process_queue_flag() {
    std::unique_lock<std::mutex> queue_lock(processor_queue_mutex_);
    should_process_queue_ = true;
  }

  void stop_process_queue_thread();

  std::vector<RequestId> get_requests_for_stage(
      const ExecutionRequestStage request_status) const;

  void add_request_to_stage(const RequestId request_id,
                            const ExecutionRequestStage request_status);

  void remove_request_from_stage(const RequestId request_id,
                                 const ExecutionRequestStage request_status);
  ChunkRequestInfo get_chunk_request_info(const RequestId request_id);

  ExecutorResourcePool executor_resource_pool_;

  std::atomic<size_t> requests_count_{0};

  ExecutorStats executor_stats_;

  const size_t ACTUALLY_QUEUED_MIN_MS{2};

  std::thread process_queue_thread_;

  mutable std::mutex processor_queue_mutex_;

  mutable std::mutex pause_processor_queue_mutex_;
  mutable std::mutex print_mutex_;

  mutable std::shared_mutex queue_stats_mutex_;

  mutable std::shared_mutex queued_set_mutex_;

  mutable std::shared_mutex executing_set_mutex_;

  std::condition_variable processor_queue_condition_;
  std::condition_variable pause_processor_queue_condition_;

  bool should_process_queue_{false};
  bool stop_process_queue_thread_{false};
  bool pause_process_queue_{false};
  bool process_queue_is_paused_{false};
  size_t process_queue_counter_{0};

  OutstandingQueueRequests outstanding_queue_requests_;

  std::set<RequestId> queued_requests_;

  std::set<RequestId> executing_requests_;

  std::vector<RequestStats> requests_stats_;

  const bool enable_stats_printing_{false};
  const bool enable_debug_printing_{false};

  const RequestId INVALID_REQUEST_ID{std::numeric_limits<size_t>::max()};

  const double max_available_resource_use_ratio_;
};

std::shared_ptr<ExecutorResourceMgr> generate_executor_resource_mgr(
    const size_t num_cpu_slots,
    const size_t num_gpu_slots,
    const size_t cpu_result_mem,
    const size_t cpu_buffer_pool_mem,
    const size_t gpu_buffer_pool_mem,
    const double per_query_max_cpu_slots_ratio,
    const double per_query_max_cpu_result_mem_ratio,
    const double per_query_max_pinned_cpu_buffer_pool_mem_ratio,
    const double per_query_max_pageable_cpu_buffer_pool_mem_ratio,
    const bool allow_cpu_kernel_concurrency,
    const bool allow_cpu_gpu_kernel_concurrency,
    const bool allow_cpu_slot_oversubscription_concurrency,
    const bool allow_gpu_slot_oversubscription,
    const bool allow_cpu_result_mem_oversubscription_concurrency,
    const double max_available_resource_use_ratio);

class ExecutorResourceHandle {
 public:
  ExecutorResourceHandle(std::shared_ptr<ExecutorResourceMgr> resource_mgr,
                         const RequestId request_id,
                         const ResourceGrant& resource_grant)
      : resource_mgr_(resource_mgr)
      , request_id_(request_id)
      , resource_grant_(resource_grant) {}

  ~ExecutorResourceHandle() {
    resource_mgr_->release_resources(request_id_, resource_grant_);
  }

  inline RequestId get_request_id() const { return request_id_; }
  inline ResourceGrant get_resource_grant() const { return resource_grant_; }

 private:
  std::shared_ptr<ExecutorResourceMgr> resource_mgr_;
  const RequestId request_id_;
  const ResourceGrant resource_grant_;
};

}  // namespace ExecutorResourceMgr_Namespace