ExecutorResourcePool.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 <array>
#include <map>
#include <shared_mutex>

#include "ExecutorResourceMgrCommon.h"
#include "ResourceGrantPolicy.h"
#include "ResourceRequest.h"

namespace ExecutorResourceMgr_Namespace {

inline ResourceType map_resource_subtype_to_resource_type(
    const ResourceSubtype resource_subtype) {
  switch (resource_subtype) {
    case ResourceSubtype::CPU_SLOTS:
      return ResourceType::CPU_SLOTS;
    case ResourceSubtype::GPU_SLOTS:
      return ResourceType::GPU_SLOTS;
    case ResourceSubtype::CPU_RESULT_MEM:
      return ResourceType::CPU_RESULT_MEM;
    case ResourceSubtype::GPU_RESULT_MEM:
      return ResourceType::GPU_RESULT_MEM;
    case ResourceSubtype::PINNED_CPU_BUFFER_POOL_MEM:
      return ResourceType::CPU_BUFFER_POOL_MEM;
    case ResourceSubtype::PINNED_GPU_BUFFER_POOL_MEM:
      return ResourceType::GPU_BUFFER_POOL_MEM;
    case ResourceSubtype::PAGEABLE_CPU_BUFFER_POOL_MEM:
      return ResourceType::CPU_BUFFER_POOL_MEM;
    case ResourceSubtype::PAGEABLE_GPU_BUFFER_POOL_MEM:
      return ResourceType::GPU_BUFFER_POOL_MEM;
    default:
      UNREACHABLE();
      return ResourceType::INVALID_TYPE;
  }
}

inline std::vector<ResourceSubtype> map_resource_type_to_resource_subtypes(
    const ResourceType resource_type) {
  switch (resource_type) {
    case ResourceType::CPU_SLOTS:
      return {ResourceSubtype::CPU_SLOTS};
    case ResourceType::GPU_SLOTS:
      return {ResourceSubtype::GPU_SLOTS};
    case ResourceType::CPU_RESULT_MEM:
      return {ResourceSubtype::CPU_RESULT_MEM};
    case ResourceType::GPU_RESULT_MEM:
      return {ResourceSubtype::GPU_RESULT_MEM};
    case ResourceType::CPU_BUFFER_POOL_MEM:
      return {ResourceSubtype::PINNED_CPU_BUFFER_POOL_MEM,
              ResourceSubtype::PAGEABLE_CPU_BUFFER_POOL_MEM};
    case ResourceType::GPU_BUFFER_POOL_MEM:
      return {ResourceSubtype::PINNED_GPU_BUFFER_POOL_MEM,
              ResourceSubtype::PAGEABLE_GPU_BUFFER_POOL_MEM};
    default:
      UNREACHABLE();
      return {ResourceSubtype::INVALID_SUBTYPE};
  }
}

struct ResourcePoolInfo {
  size_t total_cpu_slots{0};
  size_t total_gpu_slots{0};
  size_t total_cpu_result_mem{0};
  size_t total_cpu_buffer_pool_mem{0};
  size_t total_gpu_buffer_pool_mem{0};

  size_t allocated_cpu_slots{0};
  size_t allocated_gpu_slots{0};
  size_t allocated_cpu_result_mem{0};
  size_t allocated_cpu_buffer_pool_mem{0};
  size_t allocated_gpu_buffer_pool_mem{0};

  size_t allocated_cpu_buffers{0};
  size_t allocated_gpu_buffers{0};

  size_t allocated_temp_cpu_buffer_pool_mem{0};
  size_t allocated_temp_gpu_buffer_pool_mem{0};

  size_t total_requests{0};
  size_t outstanding_requests{0};
  size_t outstanding_cpu_slots_requests{0};
  size_t outstanding_gpu_slots_requests{0};
  size_t outstanding_cpu_result_mem_requests{0};
  size_t outstanding_cpu_buffer_pool_mem_requests{0};
  size_t outstanding_gpu_buffer_pool_mem_requests{0};

  ResourcePoolInfo() {}

  ResourcePoolInfo(const size_t total_cpu_slots,
                   const size_t total_gpu_slots,
                   const size_t total_cpu_result_mem,
                   const size_t total_cpu_buffer_pool_mem,
                   const size_t total_gpu_buffer_pool_mem,
                   const size_t allocated_cpu_slots,
                   const size_t allocated_gpu_slots,
                   const size_t allocated_cpu_result_mem,
                   const size_t allocated_cpu_buffer_pool_mem,
                   const size_t allocated_gpu_buffer_pool_mem,
                   const size_t allocated_cpu_buffers,
                   const size_t allocated_gpu_buffers,
                   const size_t allocated_temp_cpu_buffer_pool_mem,
                   const size_t allocated_temp_gpu_buffer_pool_mem,
                   const size_t total_requests,
                   const size_t outstanding_requests,
                   const size_t outstanding_cpu_slots_requests,
                   const size_t outstanding_gpu_slots_requests,
                   const size_t outstanding_cpu_result_mem_requests,
                   const size_t outstanding_cpu_buffer_pool_mem_requests,
                   const size_t outstanding_gpu_buffer_pool_mem_requests)
      : total_cpu_slots(total_cpu_slots)
      , total_gpu_slots(total_gpu_slots)
      , total_cpu_result_mem(total_cpu_result_mem)
      , total_cpu_buffer_pool_mem(total_cpu_buffer_pool_mem)
      , total_gpu_buffer_pool_mem(total_gpu_buffer_pool_mem)
      , allocated_cpu_slots(allocated_cpu_slots)
      , allocated_gpu_slots(allocated_gpu_slots)
      , allocated_cpu_result_mem(allocated_cpu_result_mem)
      , allocated_cpu_buffer_pool_mem(allocated_cpu_buffer_pool_mem)
      , allocated_gpu_buffer_pool_mem(allocated_gpu_buffer_pool_mem)
      , allocated_cpu_buffers(allocated_cpu_buffers)
      , allocated_gpu_buffers(allocated_gpu_buffers)
      , allocated_temp_cpu_buffer_pool_mem(allocated_temp_cpu_buffer_pool_mem)
      , allocated_temp_gpu_buffer_pool_mem(allocated_temp_gpu_buffer_pool_mem)
      , total_requests(total_requests)
      , outstanding_requests(outstanding_requests)
      , outstanding_cpu_slots_requests(outstanding_cpu_slots_requests)
      , outstanding_gpu_slots_requests(outstanding_gpu_slots_requests)
      , outstanding_cpu_result_mem_requests(outstanding_cpu_result_mem_requests)
      , outstanding_cpu_buffer_pool_mem_requests(outstanding_cpu_buffer_pool_mem_requests)
      , outstanding_gpu_buffer_pool_mem_requests(
            outstanding_gpu_buffer_pool_mem_requests) {}
};

struct ResourceGrant {
  size_t cpu_slots{0};
  size_t gpu_slots{0};
  size_t cpu_result_mem{0};

  // Below is only relevant if buffer_mem_gated_per_slot is true
  bool buffer_mem_gated_per_slot{false};
  size_t buffer_mem_per_slot{0};
  size_t buffer_mem_for_given_slots{0};

  bool is_empty() const {
    return cpu_slots == 0 && gpu_slots == 0 && cpu_result_mem == 0;
  }

  void print() const;

  std::string to_string() const;
};

using BufferPoolChunkMap =
    std::map<ChunkKey, std::pair<size_t, size_t>>;  // Value is reference count and size

class ExecutorResourcePool {
 public:
  ExecutorResourcePool(
      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);

  void log_parameters() const;

  std::vector<ResourceRequestGrant> calc_static_resource_grant_ranges_for_request(
      const std::vector<ResourceRequest>& resource_requests) const;

  std::pair<ResourceGrant, ResourceGrant> calc_min_max_resource_grants_for_request(
      const RequestInfo& resource_request) const;

  bool can_currently_satisfy_request(const ResourceGrant& min_resource_grant,
                                     const ChunkRequestInfo& chunk_request_info) const;

  std::pair<bool, ResourceGrant> determine_dynamic_resource_grant(
      const ResourceGrant& min_resource_grant,
      const ResourceGrant& max_resource_grant,
      const ChunkRequestInfo& chunk_request_info,
      const double max_request_backoff_ratio) const;

  void allocate_resources(const ResourceGrant& resource_grant,
                          const ChunkRequestInfo& chunk_request_info);

  void deallocate_resources(const ResourceGrant& resource_grant,
                            const ChunkRequestInfo& chunk_request_info);

  std::pair<size_t, size_t> get_resource_info(const ResourceType resource_type) const;

  ResourcePoolInfo get_resource_info() const;

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

  inline ConcurrentResourceGrantPolicy get_concurrent_resource_grant_policy(
      const ResourceType resource_type) const {
    return concurrent_resource_grant_policies_[static_cast<size_t>(resource_type)];
  }

  inline const ResourceGrantPolicy& get_max_resource_grant_per_request_policy(
      const ResourceSubtype resource_subtype) const {
    return max_resource_grants_per_request_policies_[static_cast<size_t>(
        resource_subtype)];
  }

  void set_concurrent_resource_grant_policy(
      const ConcurrentResourceGrantPolicy& concurrent_resource_grant_policy);

 private:
  void init(
      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);

  void init_concurrency_policies();
  void init_max_resource_grants_per_requests();

  void throw_insufficient_resource_error(const ResourceSubtype resource_subtype,
                                         const size_t min_resource_requested) const;

  size_t calc_max_resource_grant_for_request(
      const size_t requested_resource_quantity,
      const size_t min_requested_resource_quantity,
      const size_t max_grantable_resource_quantity) const;

  std::pair<size_t, size_t> calc_min_dependent_resource_grant_for_request(
      const size_t min_requested_dependent_resource_quantity,
      const size_t min_requested_independent_resource_quantity,
      const size_t dependent_to_independent_resource_ratio) const;

  std::pair<size_t, size_t> calc_max_dependent_resource_grant_for_request(
      const size_t requested_dependent_resource_quantity,
      const size_t min_requested_dependent_resource_quantity,
      const size_t max_grantable_dependent_resource_quantity,
      const size_t min_requested_independent_resource_quantity,
      const size_t max_grantable_independent_resource_quantity,
      const size_t dependent_to_independent_resource_ratio) const;

  bool check_request_against_global_policy(
      const size_t resource_total,
      const size_t resource_allocated,
      const ConcurrentResourceGrantPolicy& concurrent_resource_grant_policy) const;

  bool check_request_against_policy(
      const size_t resource_request,
      const size_t resource_total,
      const size_t resource_allocated,
      const size_t global_outstanding_requests,
      const ConcurrentResourceGrantPolicy& concurrent_resource_grant_policy) const;

  // Unlocked internal version
  bool can_currently_satisfy_request_impl(
      const ResourceGrant& min_resource_grant,
      const ChunkRequestInfo& chunk_request_info) const;

  bool can_currently_satisfy_chunk_request(
      const ResourceGrant& min_resource_grant,
      const ChunkRequestInfo& chunk_request_info) const;
  ChunkRequestInfo get_requested_chunks_not_in_pool(
      const ChunkRequestInfo& chunk_request_info) const;
  size_t get_chunk_bytes_not_in_pool(const ChunkRequestInfo& chunk_request_info) const;
  void add_chunk_requests_to_allocated_pool(const ResourceGrant& resource_grant,
                                            const ChunkRequestInfo& chunk_request_info);
  void remove_chunk_requests_from_allocated_pool(
      const ResourceGrant& resource_grant,
      const ChunkRequestInfo& chunk_request_info);

  size_t determine_dynamic_single_resource_grant(
      const size_t min_resource_requested,
      const size_t max_resource_requested,
      const size_t resource_allocated,
      const size_t total_resource,
      const double max_request_backoff_ratio) const;

  void sanity_check_requests_against_allocations() const;

  inline size_t get_total_allocated_buffer_pool_mem_for_level(
      const ExecutorDeviceType memory_pool_type) const {
    return memory_pool_type == ExecutorDeviceType::CPU
               ? get_allocated_resource_of_type(ResourceType::CPU_BUFFER_POOL_MEM)
               : get_allocated_resource_of_type(ResourceType::GPU_BUFFER_POOL_MEM);
  }

  inline bool is_resource_valid(const ResourceType resource_type) const {
    return resource_type_validity_[static_cast<size_t>(resource_type)];
  }

  inline size_t get_total_resource(const ResourceType resource_type) const {
    return total_resources_[static_cast<size_t>(resource_type)];
  }

  inline size_t get_allocated_resource_of_subtype(
      const ResourceSubtype resource_subtype) const {
    return allocated_resources_[static_cast<size_t>(resource_subtype)];
  }

  size_t get_allocated_resource_of_type(const ResourceType resource_type) const;

  inline size_t get_max_resource_grant_per_request(
      const ResourceSubtype resource_subtype) const {
    return max_resource_grants_per_request_[static_cast<size_t>(resource_subtype)];
  }

  inline size_t get_total_per_resource_num_requests(
      const ResourceType resource_type) const {
    return total_per_resource_num_requests_[static_cast<size_t>(resource_type)];
  }

  inline size_t increment_total_per_resource_num_requests(
      const ResourceType resource_type) {
    return ++total_per_resource_num_requests_[static_cast<size_t>(resource_type)];
  }

  inline size_t decrement_total_per_resource_num_requests(
      const ResourceType resource_type) {
    return --total_per_resource_num_requests_[static_cast<size_t>(resource_type)];
  }

  inline size_t get_outstanding_per_resource_num_requests(
      const ResourceType resource_type) const {
    return outstanding_per_resource_num_requests_[static_cast<size_t>(resource_type)];
  }

  inline size_t increment_outstanding_per_resource_num_requests(
      const ResourceType resource_type) {
    return ++outstanding_per_resource_num_requests_[static_cast<size_t>(resource_type)];
  }

  inline size_t decrement_outstanding_per_resource_num_requests(
      const ResourceType resource_type) {
    return --outstanding_per_resource_num_requests_[static_cast<size_t>(resource_type)];
  }

  std::array<size_t, ResourceTypeSize>
      total_resources_{};  // Will be value initialized to 0s
  std::array<bool, ResourceTypeSize> resource_type_validity_{
      false};  // Will be value initialized to false
  std::array<size_t, ResourceSubtypeSize>
      allocated_resources_{};  // Will be value initialized to 0s
  std::array<ResourceGrantPolicy, ResourceSubtypeSize>
      max_resource_grants_per_request_policies_{};
  std::array<size_t, ResourceSubtypeSize> max_resource_grants_per_request_{};
  std::array<ConcurrentResourceGrantPolicy, ResourceTypeSize>
      concurrent_resource_grant_policies_;

  size_t total_num_requests_{0};
  size_t outstanding_num_requests_{0};

  std::array<size_t, ResourceTypeSize>
      total_per_resource_num_requests_{};  // Will be value initialized to 0s

  std::array<size_t, ResourceTypeSize>
      outstanding_per_resource_num_requests_{};  // Will be value initialized to 0s

  BufferPoolChunkMap allocated_cpu_buffer_pool_chunks_;
  BufferPoolChunkMap allocated_gpu_buffer_pool_chunks_;

  const bool sanity_check_pool_state_on_deallocations_{false};
  mutable std::shared_mutex resource_mutex_;
};

}  // namespace ExecutorResourceMgr_Namespace