text:
// Copyright (c) 2106 Klemens D. Morgenstern
//
// Distributed under the Boost Software License, Version 1.0. (See accompanying
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)

#ifndef BOOST_PROCESS_WINDOWS_IS_RUNNING_HPP
#define BOOST_PROCESS_WINDOWS_IS_RUNNING_HPP

#include <boost/process/detail/config.hpp>
#include <system_error>
#include <cstdlib>
#include <boost/winapi/process.hpp>

namespace boost { namespace process { namespace detail { namespace windows {

constexpr static ::boost::winapi::DWORD_ still_active = 259;

struct child_handle;

inline bool is_running(const child_handle &p, int & exit_code)
{
    ::boost::winapi::DWORD_ code;
    //single value, not needed in the winapi.
    if (!::boost::winapi::GetExitCodeProcess(p.process_handle(), &code))
        ::boost::process::detail::throw_last_error("GetExitCodeProcess() failed");

    if (code == still_active)
        return true;
    else
    {
        exit_code = code;
        return false;
    }
}

inline bool is_running(const child_handle &p, int & exit_code, std::error_code &ec) noexcept
{
    ::boost::winapi::DWORD_ code;
    //single value, not needed in the winapi.
    if (!::boost::winapi::GetExitCodeProcess(p.process_handle(), &code))
        ec = ::boost::process::detail::get_last_error();
    else
        ec.clear();

    if (code == still_active)
        return true;
    else
    {
        exit_code = code;
        return false;
    }
}

inline bool is_running(int code)
{
    return code == still_active;
}

inline int eval_exit_status(int in ) {return in;}

}}}}

#endif

ext: hpp

max_stars_repo_name: dilawar/eyesthatblink

max_issues_repo_name: dilawar/eyesthatblink

max_forks_repo_name: dilawar/eyesthatblink


text:
/* Boost.MultiIndex test for allocator awareness.
 *
 * Copyright 2003-2020 Joaquin M Lopez Munoz.
 * Distributed under the Boost Software License, Version 1.0.
 * (See accompanying file LICENSE_1_0.txt or copy at
 * http://www.boost.org/LICENSE_1_0.txt)
 *
 * See http://www.boost.org/libs/multi_index for library home page.
 */

#include "test_alloc_awareness.hpp"

#include <boost/config.hpp> /* keep it first to prevent nasty warns in MSVC */
#include <boost/detail/lightweight_test.hpp>
#include <boost/move/core.hpp>
#include <boost/move/utility_core.hpp>
#include "pre_multi_index.hpp"
#include <boost/multi_index_container.hpp>
#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index/ordered_index.hpp>
#include <boost/multi_index/random_access_index.hpp>
#include <boost/multi_index/ranked_index.hpp>
#include <boost/multi_index/sequenced_index.hpp>
#include "rooted_allocator.hpp"

struct move_tracker
{
  move_tracker(int n):n(n),move_cted(false){}
  move_tracker(const move_tracker& x):n(x.n),move_cted(false){}
  move_tracker(BOOST_RV_REF(move_tracker) x):n(x.n),move_cted(true){}
  move_tracker& operator=(BOOST_COPY_ASSIGN_REF(move_tracker) x)
    {n=x.n;return *this;}
  move_tracker& operator=(BOOST_RV_REF(move_tracker) x){n=x.n;return *this;}

  int  n;
  bool move_cted;

private:
  BOOST_COPYABLE_AND_MOVABLE(move_tracker)
};

inline bool operator==(const move_tracker& x,const move_tracker& y)
{
  return x.n==y.n;
}

inline bool operator<(const move_tracker& x,const move_tracker& y)
{
  return x.n<y.n;
}

#if defined(BOOST_NO_ARGUMENT_DEPENDENT_LOOKUP)
namespace boost{
#endif

inline std::size_t hash_value(const move_tracker& x)
{
  boost::hash<int> h;
  return h(x.n);
}

#if defined(BOOST_NO_ARGUMENT_DEPENDENT_LOOKUP)
} /* namespace boost */
#endif

#if defined(BOOST_NO_CXX17_IF_CONSTEXPR)&&defined(BOOST_MSVC)
#pragma warning(push)
#pragma warning(disable:4127) /* conditional expression is constant */
#endif

template<bool Propagate,bool AlwaysEqual>
void test_allocator_awareness_for()
{
  using namespace boost::multi_index;

  typedef rooted_allocator<move_tracker,Propagate,AlwaysEqual> allocator;
  typedef multi_index_container<
    move_tracker,
    indexed_by<
      hashed_unique<identity<move_tracker> >,
      ordered_unique<identity<move_tracker> >,
      random_access<>,
      ranked_unique<identity<move_tracker> >,
      sequenced<>
    >,
    allocator
  >                                                            container;

  allocator root1(0),root2(0);
  container c(root1);
  for(int i=0;i<10;++i)c.emplace(i);

  BOOST_TEST(c.get_allocator().comes_from(root1));

  {
    container c2(c,root2);
    BOOST_TEST(c2.get_allocator().comes_from(root2));
    BOOST_TEST(c2==c);
  }
  {
    container           c2(c);
    const move_tracker* pfirst=&*c2.begin();
    container           c3(boost::move(c2),root2);
    BOOST_TEST(c3.get_allocator().comes_from(root2));
    BOOST_TEST(c3==c);
    BOOST_TEST(c2.empty());
    BOOST_TEST(AlwaysEqual==(&*c3.begin()==pfirst));
    BOOST_TEST(!AlwaysEqual==(c3.begin()->move_cted));
  }
  {
    container c2(root2);
    c2=c;
    BOOST_TEST(c2.get_allocator().comes_from(Propagate?root1:root2));
    BOOST_TEST(c2==c);
  }
  {
    const bool          element_transfer=Propagate||AlwaysEqual;

    container           c2(c);
    const move_tracker* pfirst=&*c2.begin();
    container           c3(root2);
    c3=boost::move(c2);
    BOOST_TEST(c3.get_allocator().comes_from(Propagate?root1:root2));
    BOOST_TEST(c3==c);
    BOOST_TEST(c2.empty());
    BOOST_TEST(element_transfer==(&*c3.begin()==pfirst));
    BOOST_TEST(!element_transfer==(c3.begin()->move_cted));
  }
  if(Propagate||AlwaysEqual){
    container           c2(c);
    const move_tracker* pfirst=&*c2.begin();
    container           c3(root2);
    c3.swap(c2);
    BOOST_TEST(c2.get_allocator().comes_from(Propagate?root2:root1));
    BOOST_TEST(c3.get_allocator().comes_from(Propagate?root1:root2));
    BOOST_TEST(c3==c);
    BOOST_TEST(c2.empty());
    BOOST_TEST(&*c3.begin()==pfirst);
    BOOST_TEST(!c3.begin()->move_cted);
  }
}

#if defined(BOOST_NO_CXX17_IF_CONSTEXPR)&&defined(BOOST_MSVC)
#pragma warning(pop) /* C4127 */
#endif

void test_allocator_awareness()
{
  test_allocator_awareness_for<false,false>();
  test_allocator_awareness_for<false,true>();

#if !defined(BOOST_NO_CXX11_ALLOCATOR)
  /* only in C+11 onwards are allocators potentially expected to propagate */

  test_allocator_awareness_for<true,false>();
  test_allocator_awareness_for<true,true>();

#endif
}

ext: cpp


text:
#include "node.h"
#include <exception>
#include <boost/algorithm/string.hpp>
#include <boost/format.hpp>
#include "scene.h"

using boost::format;

Node::Node(const boost::property_tree::ptree& attr, const Scene& cont)
  : attributes(attr), container(cont) {}

std::string Node::getString(std::string key) const {
  auto result = attributes.get_optional<std::string>(key);

  if (!result) {
    throw std::runtime_error(
      str(format("Cannot read string property '%1%'") % key)
    );
  }

  return *result;
}

int Node::getInt(std::string key) const {
  auto result = attributes.get_optional<int>(key);

  if (!result) {
    throw std::runtime_error(
      str(format("Cannot read integer property '%1%'") % key)
    );
  }

  return *result;
}

bool Node::getBool(std::string key) const {
  auto result = attributes.get_optional<bool>(key);

  if (!result) {
    throw std::runtime_error(
      str(format("Cannot read boolean property '%1%'") % key)
    );
  }

  return *result;
}

float Node::getFloat(std::string key) const {
  auto result = attributes.get_optional<float>(key);

  if (!result) {
    throw std::runtime_error(
      str(format("Cannot read float property '%1%'") % key)
    );
  }

  return *result;
}

Vec Node::getVec(std::string key) const {
  const NodeVecTranslator t;
  auto result = attributes.get_optional<Vec, NodeVecTranslator>(key, t);

  if (!result) {
    throw std::runtime_error(
      str(format("Cannot read vector property '%1%'") % key)
    );
  }

  return *result;
}

const AreaLight* Node::getLight(std::string key) const {
  using NodeLightTranslator = Node::NodeLookupTranslator<const AreaLight*>;
  const NodeLightTranslator t(container.lights);
  auto result =
    attributes.get_optional<const AreaLight*, NodeLightTranslator>(key, t);

  if (!result) {
    const std::string itemName = attributes.get<std::string>(key);
    const std::string msg =
      "Cannot resolve light reference '%1%' in property '%2%'";
    throw std::runtime_error(str(format(msg) % itemName % key));
  }

  return *result;
}

const Material* Node::getMaterial(std::string key) const {
  using NodeMaterialTranslator = Node::NodeLookupTranslator<const Material*>;
  const NodeMaterialTranslator t(container.materials);
  auto result =
    attributes.get_optional<const Material*, NodeMaterialTranslator>(key, t);

  if (!result) {
    const std::string itemName = attributes.get<std::string>(key);
    const std::string msg =
      "Cannot resolve material reference '%1%' in property '%2%'";
    throw std::runtime_error(str(format(msg) % itemName % key));
  }

  return *result;
}

const Geom* Node::getGeometry(std::string key) const {
  using NodeGeometryTranslator = Node::NodeLookupTranslator<const Geom*>;
  const NodeGeometryTranslator t(container.geometry);
  auto result =
    attributes.get_optional<const Geom*, NodeGeometryTranslator>(key, t);

  if (!result) {
    const std::string itemName = attributes.get<std::string>(key);
    const std::string msg =
      "Cannot resolve geometry reference '%1%' in property '%2%'";
    throw std::runtime_error(str(format(msg) % itemName % key));
  }

  return *result;
}

std::vector<const Geom*> Node::getGeometryList(std::string key) const {
  using NodeGeometryTranslator = Node::NodeLookupTranslator<const Geom*, false>;
  const NodeGeometryTranslator t(container.geometry);
  const auto& listRoot = attributes.get_child_optional(key);

  if (!listRoot) {
    throw std::runtime_error(str(format("Cannot read list '%1%'") % key));
  }

  std::vector<const Geom*> result;
  for (const auto& listItem : *listRoot) {
    const auto item = listItem.second.get_value_optional<const Geom*>(t);

    if (!item) {
      const std::string itemName = listItem.second.get_value<std::string>();
      const std::string msg =
        "Cannot resolve geometry reference '%1%' in list '%2%'";
      throw std::runtime_error(str(format(msg) % itemName % key));
    }

    result.push_back(*item);
  }

  return result;
}

Node::NodeVecTranslator::NodeVecTranslator() {}

boost::optional<Vec> Node::NodeVecTranslator::get_value(
  const std::string& data
) const {;
  std::vector<std::string> tokens;
  boost::algorithm::split(
    tokens,
    data,
    boost::is_space(),
    boost::token_compress_on
  );

  if (tokens.size() != 3) {
    return boost::optional<Vec>();
  }

  Vec result;
  try {
    result[0] = std::stof(tokens[0]);
    result[1] = std::stof(tokens[1]);
    result[2] = std::stof(tokens[2]);
  } catch (...) {
    return boost::optional<Vec>();
  }

  return boost::optional<Vec>(result);
}

template <typename T, bool allowNull>
Node::NodeLookupTranslator<T, allowNull>::NodeLookupTranslator(
  const std::map<std::string, T>& l
) : lookup(l) {}

template <typename T, bool allowNull>
boost::optional<T> Node::NodeLookupTranslator<T, allowNull>::get_value(
  const std::string& data
) const {
  if (data.length() == 0 && allowNull) {
    return boost::optional<T>(nullptr);
  } else if (lookup.count(data) != 0) {
    return boost::optional<T>(lookup.at(data));
  }

  return boost::optional<T>();
}

ext: cc

max_stars_repo_name: sdao/path-tracer-nacl

max_issues_repo_name: sdao/path-tracer-nacl

max_forks_repo_name: sdao/path-tracer-nacl


text:
// Copyright (2013) JDSU T&M Pte Ltd.  All rights reserved.

#include <log4cxx/logger.h>
#include <fstream>
#include <boost/foreach.hpp>
#include <boost/thread/mutex.hpp>
#include <gperftools/malloc_extension.h>
#include "xtreme/iwf/internal/monitor.h"
#include "xtreme/cs/cs.h"
#include "xtreme/config/csparamapi.h"
#include "xtreme/config/csparam_pubsub.h"

namespace {

log4cxx::LoggerPtr g_logger(log4cxx::Logger::getLogger("xtreme.IWF.Monitor"));
log4cxx::LoggerPtr g_telemetry_logger(log4cxx::Logger::getLogger("xtreme.telemetry.IWF-Monitor"));

const std::string IWF_OVERLOAD_CHECK_INTERVAL("IWF/overload-check-interval");
const std::string IWF_PIPELINE_QUEUE_OVERLOAD("IWF/pipeline-queue-overload-percentage");
const std::string IWF_PIPELINE_QUEUE_UNCLOGGED("IWF/pipeline-queue-unclogged-percentage");
const std::string DEBUG_DO_NOT_REPORT_OVERLOAD("DEBUG/do-not-report-overload");
const std::string IWF_MEMORY_ASAP_LEVEL("IWF/memory-asap-level-percentage");
const std::string IWF_MEMORY_ASDP_LEVEL("IWF/memory-asdp-level-percentage");
const std::string IWF_INCLUDE_SWAP_IN_MEMORY_USAGE("IWF/include-swap-in-memory-usage");

uint64_t convert(const std::string& number, const std::string& unit) {
    uint64_t value = strtoll(number.c_str(), nullptr, 0);
    if (unit == "kB") {
        return value * 1024;
    } else if (unit == "Mb") {
        return value * 1024 * 1024;
    } else {
        return value;
    }
}

double get_memory_usage(bool include_swap) {
    std::ifstream mem_info_file("/proc/meminfo");
    if (!mem_info_file) {
        LOG4CXX_WARN(g_logger, "Memory usage cannot be retrieved!");
        return -1.0;
    }

    // First lines of /proc/meminfo
    // MemTotal:       16461832 kB
    // MemFree:        16779260 kB
    // Buffers:          214780 kB
    // Cached:          4257524 kB
    // SwapCached:            0 kB
    // Active:          4134816 kB
    // Inactive:        2452516 kB
    // Active(anon):    2114708 kB
    // Inactive(anon):     4596 kB
    // Active(file):    2020108 kB
    // Inactive(file):  2447920 kB
    // Unevictable:           0 kB
    // Mlocked:               0 kB
    // SwapTotal:      16779260 kB
    // SwapFree:       16779260 kB

    uint64_t total_mem = 0ul;
    uint64_t free_mem = 0ul;
    uint64_t buffers = 0ul;
    uint64_t cached = 0ul;
    uint64_t total_swap = 0ul;
    uint64_t free_swap = 0ul;

    std::string label;
    std::string number;
    std::string unit;

    while (true) {
        mem_info_file >> label >> number >> unit;
        if (label == "MemTotal:") {
            total_mem = convert(number, unit);
        } else if (label == "MemFree:") {
            free_mem = convert(number, unit);
        } else if (label == "Buffers:") {
            buffers = convert(number, unit);
        } else if (label == "Cached:") {
            cached = convert(number, unit);
        } else if (label == "SwapTotal:") {
            total_swap = convert(number, unit);
        } else if (label == "SwapFree:") {
            free_swap = convert(number, unit);
            break;
        } else {
        }
    }

    // Sample output of tcmalloc GetStats()
    //    ------------------------------------------------
    //   MALLOC:     3693322584 ( 3522.2 MiB) Bytes in use by application
    //   MALLOC: +       335872 (    0.3 MiB) Bytes in page heap freelist
    //   MALLOC: +       664928 (    0.6 MiB) Bytes in central cache freelist
    //   MALLOC: +        23040 (    0.0 MiB) Bytes in transfer cache freelist
    //   MALLOC: +      9977672 (    9.5 MiB) Bytes in thread cache freelists
    //   MALLOC: +      5288088 (    5.0 MiB) Bytes in malloc metadata
    //   MALLOC:   ------------
    //   MALLOC: =   3709612184 ( 35

    // We want the sum of 4 values (page heap, central, transfer and thread cache freelists).
    const int kTcMallocStatsStringSize = 500;
    char tcmallocstats[kTcMallocStatsStringSize + 1];
    memset(tcmallocstats, 0, kTcMallocStatsStringSize + 1);
    char *freemem = tcmallocstats;
    uint64_t value = 0;
    uint64_t freetcmallocmem = 0;
    MallocExtension::instance()->GetStats(tcmallocstats, kTcMallocStatsStringSize);
    // std::cout << tcmallocstats << std::endl;
    for (int i = 0; i < 4; ++i) {
        while ('+' != (*freemem) && '\0' != (*freemem)) ++freemem;  // Locate '+' character.
        ++freemem;
        while (' ' == (*freemem)) ++freemem;  // Locate non-space character.
        value = strtoll(freemem, nullptr, 10);  // value in bytes
        freetcmallocmem += value;
        // std::cout << "freevalue: " << value << std::endl;

        if (freemem > (tcmallocstats + kTcMallocStatsStringSize - 2)) {
            // Overrun. Treat value as invalid.
            freetcmallocmem = 0;
            break;
        }
    }
    // std::cout << "freetcmallocmem: " << freetcmallocmem << std::endl;

    uint64_t total = total_mem;
    uint64_t total_free = free_mem + buffers + cached + freetcmallocmem;
    if (include_swap) {
        total += total_swap;
        total_free += free_swap;
    }
    double usage = total - total_free;

    size_t tcm_current_allocated = 0;
    size_t tcm_heap_size = 0;
    size_t tcm_ph_free = 0;
    size_t tcm_ph_unmapped = 0;
    size_t tcm_thread_cache_bytes = 0;
    size_t tcm_thread_cache_bytes_max = 0;

    MallocExtension::instance()->GetNumericProperty("generic.current_allocated_bytes", &tcm_current_allocated);
    MallocExtension::instance()->GetNumericProperty("generic.heap_size", &tcm_heap_size);
    MallocExtension::instance()->GetNumericProperty("tcmalloc.pageheap_free_bytes", &tcm_ph_free);
    MallocExtension::instance()->GetNumericProperty("tcmalloc.pageheap_unmapped_bytes", &tcm_ph_unmapped);
    MallocExtension::instance()->GetNumericProperty("tcmalloc.current_total_thread_cache_bytes", &tcm_thread_cache_bytes);
    MallocExtension::instance()->GetNumericProperty("tcmalloc.max_total_thread_cache_bytes", &tcm_thread_cache_bytes_max);
    LOG4CXX_INFO(g_telemetry_logger, "tcmalloc stats: CurrentAllocatedBytes(AppMemUse)= " << tcm_current_allocated
        << " HeapSize(SysMemReserved) " << tcm_heap_size
        << " PhFreeBytes " << tcm_ph_free
        << " PhUnmapped " << tcm_ph_unmapped
        << " ThreadCachedBytes(SmallObs) " << tcm_thread_cache_bytes
        << " ThreadCachedBytesMax(Limit) " << tcm_thread_cache_bytes_max);

    if (include_swap) {
        LOG4CXX_INFO(g_telemetry_logger, "free=" << free_mem << " buf=" << buffers << " cached=" << cached
                     << " tcmallocfree=" << freetcmallocmem
                     << " swapfree=" << free_swap << " total=" << total_mem << " swap="
                     << total_swap << " usage=" << usage / total);
    } else {
        LOG4CXX_INFO(g_telemetry_logger, "free=" << free_mem << " buf=" << buffers << " cached=" << cached
                     << " tcmallocfree=" << freetcmallocmem
                     << " total=" << total_mem << " usage=" << usage / total);
    }
    return usage / total;   // returned value should be between 0.0 and 1.0.
}

}   // anonymous namespace

namespace xtreme {
namespace iwf {

Monitor::Monitor(std::vector<xtreme::iwf::pipeline::Pipeline*>& cpp_pipelines,
                 std::vector<xtreme::iwf::pipeline::Pipeline*>& upp_pipelines,
                 publish_asap_asdp_event_function publish_asap_asdp_event)
  : Subscriber("IWF_Monitor", 1024)
  , cpp_pipelines_(cpp_pipelines)
  , upp_pipelines_(upp_pipelines)
  , publish_asap_asdp_event_(publish_asap_asdp_event)
{
    xtreme::cs::CS& cs = xtreme::cs::CS::theCS();
    config_ = cs.getCSParamAPI();

    overload_check_interval_ = 30;
    try {
        overload_check_interval_ = config_->get_uint(IWF_OVERLOAD_CHECK_INTERVAL);
    } catch(XtremeException& /* e */) {
    }

    queue_capacity_ = 1000.0;
    try {
        queue_capacity_ = static_cast<double>(config_->get_uint("IWF/PIPELINE_QUEUE_SIZE"));
    } catch(XtremeException& /* e */) {
    }

    pipeline_queue_overload_ = 0.9;
    try {
        std::string value = config_->get_string(IWF_PIPELINE_QUEUE_OVERLOAD);
        pipeline_queue_overload_ = strtod(value.c_str(), nullptr);
    } catch(XtremeException& /* e */) {
    }

    pipeline_queue_unclogged_ = 0.7;
    try {
        std::string value = config_->get_string(IWF_PIPELINE_QUEUE_UNCLOGGED);
        pipeline_queue_unclogged_ = strtod(value.c_str(), nullptr);
    } catch(XtremeException& /* e */) {
    }

    do_not_report_overload_ = false;
    try {
        do_not_report_overload_ = config_->get_bool(DEBUG_DO_NOT_REPORT_OVERLOAD);
    } catch(XtremeException& /* e */) {
    }

    memory_asap_level_ = 0.7;
    try {
        std::string value = config_->get_string(IWF_MEMORY_ASAP_LEVEL);
        memory_asap_level_ = strtod(value.c_str(), nullptr);
    } catch(XtremeException& /* e */) {
    }

    memory_asdp_level_ = 0.5;
    try {
        std::string value = config_->get_string(IWF_MEMORY_ASDP_LEVEL);
        memory_asdp_level_ = strtod(value.c_str(), nullptr);
    } catch(XtremeException& /* e */) {
    }

    include_swap_ = true;
    try {
        include_swap_ = config_->get_bool(IWF_INCLUDE_SWAP_IN_MEMORY_USAGE);
    } catch(XtremeException& /* e */) {
    }

    // If the config parameters are not consistent when xtreme starts up, we will disable the
    // overloading and memory-usage monitoring.
    if (overload_check_interval_ == 0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_OVERLOAD_CHECK_INTERVAL
                      << "' must > 0");
        do_not_report_overload_ = true;
    }
    if (queue_capacity_ == 0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << "IWF/PIPELINE_QUEUE_SIZE"
                      << "' must > 0");
        do_not_report_overload_ = true;
    }
    if (pipeline_queue_overload_ < 0.0 || pipeline_queue_overload_ > 1.0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_OVERLOAD
                      << "' must be between 0 and 1.0");
        do_not_report_overload_ = true;
    }
    if (pipeline_queue_unclogged_ < 0.0 || pipeline_queue_unclogged_ > 1.0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_UNCLOGGED
                      << "' must be between 0 and 1.0");
        do_not_report_overload_ = true;
    }
    if (pipeline_queue_overload_ <= pipeline_queue_unclogged_) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_OVERLOAD
                      << "' must be > config-parameter='" << IWF_PIPELINE_QUEUE_UNCLOGGED << "'");
        do_not_report_overload_ = true;
    }
    if (memory_asap_level_ < 0.0 || memory_asap_level_ > 1.0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASAP_LEVEL
                      << "' must be between 0 and 1.0");
        do_not_report_overload_ = true;
    }
    if (memory_asdp_level_ < 0.0 || memory_asdp_level_ > 1.0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASDP_LEVEL
                      << "' must be between 0 and 1.0");
        do_not_report_overload_ = true;
    }
    if (memory_asap_level_ <= memory_asdp_level_) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASDP_LEVEL
                      << "' must be > config-parameter='" << IWF_MEMORY_ASDP_LEVEL << "'");
        do_not_report_overload_ = true;
    }

    keep_going_ = false;
    IPublisherList publishers = cs.GetPublisherForType(kEVENT_CONFIGURATION);
    config::CSParamCriteria* criteria = new config::CSParamCriteria("IWF_Monitor");
    subscription_ = new xtreme::Subscription(kEVENT_CONFIGURATION, criteria);
    Subscribe(publishers, subscription_);
    MallocExtension::instance()->Initialize();

    if (do_not_report_overload_ == false) {
        start();
    }
}

Monitor::~Monitor() {
    delete subscription_;
    stop();
}

/* virtual */ bool Monitor::Subscribe(IPublisherList publishers, Subscription* subscription) {
    if (publishers.size() != 1) {
        LOG4CXX_ERROR(g_logger, "number of config-publisher=" << publishers.size()
                      << " is not exactly 1");
        return false;
    }

    IPublisherList::iterator iter = publishers.begin();
    IPublisher* publisher = *iter;
    if (publisher->AddSubscription(this, subscription)) {
        LOG4CXX_DEBUG(g_logger, "subscribed to config-publisher");
        return true;
    } else {
        LOG4CXX_DEBUG(g_logger, "failed to subscribe to config-publisher");
        return false;
    }
}

/* virtual */ void Monitor::Callback(EventPtr& event) {
    xtreme::config::CSParamEvent& e = static_cast<xtreme::config::CSParamEvent&>(*event);
    const std::string& param_name = e.getSrcConfigName();

    // If the config-parameters are inconsistently changed while xtreme is running,
    // we ignored the new values.
    bool check_interval_changed = false;
    bool overload_changed = false;
    bool unclogged_changed = false;
    bool asap_changed = false;
    bool asdp_changed = false;
    bool swap_changed = false;

    uint32_t overload_check_interval = 0;
    double pipeline_queue_overload = -1.0;
    double pipeline_queue_unclogged = -1.0;
    double memory_asap_level = -1.0;
    double memory_asdp_level = -1.0;
    bool include_swap = true;

    if (param_name == IWF_OVERLOAD_CHECK_INTERVAL) {
        try {
            overload_check_interval = config_->get_uint(IWF_OVERLOAD_CHECK_INTERVAL);
            check_interval_changed = true;
        } catch(XtremeException& /* e */) {
        }
    } else if (param_name == IWF_PIPELINE_QUEUE_OVERLOAD) {
        try {
            std::string value = config_->get_string(IWF_PIPELINE_QUEUE_OVERLOAD);
            pipeline_queue_overload = strtod(value.c_str(), nullptr);
            overload_changed = true;
        } catch(XtremeException& /* e */) {
        }
    } else if (param_name == IWF_PIPELINE_QUEUE_UNCLOGGED) {
        try {
            std::string value = config_->get_string(IWF_PIPELINE_QUEUE_UNCLOGGED);
            pipeline_queue_unclogged = strtod(value.c_str(), nullptr);
            unclogged_changed = true;
        } catch(XtremeException& /* e */) {
        }
    } else if (param_name == DEBUG_DO_NOT_REPORT_OVERLOAD) {
        try {
            do_not_report_overload_ = config_->get_bool(DEBUG_DO_NOT_REPORT_OVERLOAD);
        } catch(XtremeException& /* e */) {
        }
    } else if (param_name == IWF_MEMORY_ASAP_LEVEL) {
        try {
            std::string value = config_->get_string(IWF_MEMORY_ASAP_LEVEL);
            memory_asap_level = strtod(value.c_str(), nullptr);
            asap_changed = true;
        } catch(XtremeException& /* e */) {
        }
    } else if (param_name == IWF_MEMORY_ASDP_LEVEL) {
        try {
            std::string value = config_->get_string(IWF_MEMORY_ASDP_LEVEL);
            memory_asdp_level = strtod(value.c_str(), nullptr);
            asdp_changed = true;
        } catch(XtremeException& /* e */) {
        }
    } else if (param_name == IWF_INCLUDE_SWAP_IN_MEMORY_USAGE) {
        try {
            include_swap = config_->get_bool(IWF_INCLUDE_SWAP_IN_MEMORY_USAGE);
            swap_changed = true;
        } catch(XtremeException& /* e */) {
        }
    } else {
    }

    bool error = false;
    if (check_interval_changed && overload_check_interval == 0) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_OVERLOAD_CHECK_INTERVAL
                      << "' must > 0");
        error = true;
    }
    if (overload_changed && (pipeline_queue_overload < 0.0 || pipeline_queue_overload > 1.0)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_OVERLOAD
                      << "' must be between 0 and 1.0");
        error = true;
    }
    if (unclogged_changed && (pipeline_queue_unclogged < 0.0 || pipeline_queue_unclogged > 1.0)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_UNCLOGGED
                      << "' must be between 0 and 1.0");
        error = true;
    }
    if (overload_changed && unclogged_changed && (pipeline_queue_overload <= pipeline_queue_unclogged)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_OVERLOAD
                      << "' must be > config-parameter='" << IWF_PIPELINE_QUEUE_UNCLOGGED << "'");
        error = true;
    }
    if (overload_changed && !unclogged_changed && (pipeline_queue_overload <= pipeline_queue_unclogged_)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_OVERLOAD
                      << "' must be > config-parameter='" << IWF_PIPELINE_QUEUE_UNCLOGGED << "'");
        error = true;
    }
    if (!overload_changed && unclogged_changed && (pipeline_queue_overload_ <= pipeline_queue_unclogged)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_PIPELINE_QUEUE_OVERLOAD
                      << "' must be > config-parameter='" << IWF_PIPELINE_QUEUE_UNCLOGGED << "'");
        error = true;
    }
    if (asap_changed && (memory_asap_level || memory_asap_level > 1.0)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASAP_LEVEL
                      << "' must be between 0 and 1.0");
        error = true;
    }
    if (asdp_changed && (memory_asdp_level < 0.0 || memory_asdp_level > 1.0)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASDP_LEVEL
                      << "' must be between 0 and 1.0");
        error = true;
    }
    if (asap_changed && asdp_changed && (memory_asap_level <= memory_asdp_level)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASDP_LEVEL
                      << "' must be > config-parameter='" << IWF_MEMORY_ASDP_LEVEL << "'");
        error = true;
    }
    if (asap_changed && !asdp_changed && (memory_asap_level <= memory_asdp_level_)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASDP_LEVEL
                      << "' must be > config-parameter='" << IWF_MEMORY_ASDP_LEVEL << "'");
        error = true;
    }
    if (!asap_changed && asdp_changed && (memory_asap_level_ <= memory_asdp_level)) {
        LOG4CXX_ERROR(g_logger, "config-parameter='" << IWF_MEMORY_ASDP_LEVEL
                      << "' must be > config-parameter='" << IWF_MEMORY_ASDP_LEVEL << "'");
        error = true;
    }

    if (error == false) {
        if (check_interval_changed) {
            overload_check_interval_ = overload_check_interval;
        }
        if (overload_changed) {
            pipeline_queue_overload_ = pipeline_queue_overload;
        }
        if (unclogged_changed) {
            pipeline_queue_unclogged_ = pipeline_queue_unclogged;
        }
        if (asap_changed) {
            memory_asap_level_ = memory_asap_level;
        }
        if (asdp_changed) {
            memory_asdp_level_ = memory_asdp_level;
        }
        if (swap_changed) {
            include_swap_ = include_swap;
        }
    }

    if (do_not_report_overload_ == true && keep_going_) {
        stop();
    }
    if (error == false && do_not_report_overload_ == false && keep_going_ == false) {
        start();
    }
}

void Monitor::start() {
    if (do_not_report_overload_) {
        LOG4CXX_DEBUG(g_logger, "the IWF load and memory-usage monitor is not enabled");
        return;
    }
    if (keep_going_) {
        return;     // already running
    }

    keep_going_ = true;
    thread_ = boost::thread(&Monitor::thread_function_, this);
}

void Monitor::stop() {
    keep_going_ = false;
    LOG4CXX_DEBUG(g_logger, "shutting down the IWF load and memory-usage monitor");
    if (thread_.joinable()) {
        thread_.interrupt();
        try {
            thread_.join();
        } catch(const boost::thread_interrupted& /* e */) {
        }
    }
    LOG4CXX_DEBUG(g_logger, "the IWF load and memory-usage monitor is shut down");
}

void Monitor::thread_function_() {
    xtreme::cs::CS& cs = xtreme::cs::CS::theCS();
    pid_t tid;
    tid = syscall(SYS_gettid);
    cs.RegisterThreads("IWF Load and Memory Monitor Thread", tid);

    // Wait for IWF to set the cpp_pipelines_ and upp_pipelines_ data members.
    LOG4CXX_DEBUG(g_logger, "waiting for IWF to set the cpp_pipelines_ and upp_pipelines_");
    while (keep_going_) {
        boost::this_thread::sleep(boost::posix_time::milliseconds(overload_check_interval_ * 1000));
        if (!upp_pipelines_.empty() && !cpp_pipelines_.empty()) {
            break;
        }
    }

    LOG4CXX_DEBUG(g_logger, "cpp-count=" << cpp_pipelines_.size()
                  << " upp-count=" << upp_pipelines_.size());
    size_t pipelines_count = cpp_pipelines_.size() + upp_pipelines_.size();
    std::vector<uint64_t> queue_sizes(pipelines_count);
    std::vector<double> queue_filled_ratio(pipelines_count);

    bool is_overloaded = false;
    bool is_asap = false;
    bool sent_asap = false;
    bool need_to_reset = false;

    while (keep_going_) {
        boost::this_thread::sleep(boost::posix_time::milliseconds(overload_check_interval_ * 1000));

        if (keep_going_ == false) {
            break;
        }

        size_t i = 0;
        BOOST_FOREACH(xtreme::iwf::pipeline::Pipeline* p, cpp_pipelines_) {
            const pipeline::PipelineStats s = p->GetTelemetrySnapshot();
            queue_sizes[i] = s.input_queue_size_;
            queue_filled_ratio[i] = s.input_queue_size_ / queue_capacity_;
            ++i;
        }
        BOOST_FOREACH(xtreme::iwf::pipeline::Pipeline* p, upp_pipelines_) {
            const pipeline::PipelineStats s = p->GetTelemetrySnapshot();
            queue_sizes[i] = s.input_queue_size_;
            queue_filled_ratio[i] = s.input_queue_size_ / queue_capacity_;
            ++i;
        }

        bool overload = false;
        for (size_t i = 0; i < pipelines_count; ++i) {
            if (queue_filled_ratio[i] > pipeline_queue_overload_) {
                LOG4CXX_INFO(g_logger, "pipeline-" << i << " is overloaded, percentage-fill="
                             << queue_filled_ratio[i] * 100 << "%");
                overload = true;
                break;
            }
        }

        bool unclogged = true;
        for (size_t i = 0; i < pipelines_count; ++i) {
            if (queue_filled_ratio[i] > pipeline_queue_unclogged_) {
                LOG4CXX_INFO(g_logger, "pipeline-" << i << " is still not unclogged, percentage-fill="
                             << queue_filled_ratio[i] * 100 << "%");
                unclogged = false;
                break;
            }
        }

        double usage = get_memory_usage(include_swap_);
        bool asap = (usage > memory_asap_level_);
        bool asdp = (usage < memory_asdp_level_);

        if (is_asap == false) {
            if (asap) {
                LOG4CXX_INFO(g_logger, "memory-usage has crossed above the ASAP-level="
                             << memory_asap_level_ * 100 << "%");
                is_asap = true;
                sent_asap = false;
                need_to_reset = true;
            }
        } else if (asdp) {
            LOG4CXX_INFO(g_logger, "memory-usage has settled down to below the ASDP-level="
                         << memory_asdp_level_ * 100 << "%");
            is_asap = false;
        }

        if (is_overloaded == false) {
            if (overload) {
                LOG4CXX_INFO(g_logger, "the pipeline queues have been jammed above "
                             << pipeline_queue_overload_ * 100 << "%");
                is_overloaded = true;
                need_to_reset = true;
            }
        } else if (unclogged == true) {
            LOG4CXX_INFO(g_logger, "the pipeline queues have cleared down to below "
                         << pipeline_queue_unclogged_ * 100 << "%");
            is_overloaded = false;
        }

        if (is_overloaded || is_asap) {
            bool job_killed = cs.UpdateLoadStatus(is_overloaded, is_asap);
            if (job_killed == false) {
                // Last (user) job remaining.  So now need to suspend MDR event production if
                // memory usage is still above the ASAP level.
                if (is_asap && sent_asap == false) {
                    LOG4CXX_INFO(g_logger, "sent an ASAP event to suspend MDR production");
                    publish_asap_asdp_event_(new ASActivateEvent);
                    sent_asap = true;
                } else if (is_asap == false && sent_asap == true) {
                    LOG4CXX_INFO(g_logger, "sent an ASDP event to resume MDR production");
                    publish_asap_asdp_event_(new ASDeactivateEvent);
                    sent_asap = false;
                }
            }
        } else if (is_overloaded == false && is_asap == false && need_to_reset) {
            cs.UpdateLoadStatus(is_overloaded, is_asap);
            need_to_reset = false;
            if (sent_asap) {
                LOG4CXX_INFO(g_logger, "sent an ASDP event to resume MDR production");
                publish_asap_asdp_event_(new ASDeactivateEvent);
                sent_asap = false;
            }
        }
    }
}

}}  // namespace xtreme::iwf


text:
/*
 * RESTKmsConnector.actor.cpp
 *
 * This source file is part of the FoundationDB open source project
 *
 * Copyright 2013-2022 Apple Inc. and the FoundationDB project authors
 *
 * 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.
 */

#include "fdbserver/RESTKmsConnector.h"

#include "fdbclient/FDBTypes.h"
#include "fdbclient/rapidjson/document.h"
#include "fdbclient/rapidjson/rapidjson.h"
#include "fdbclient/rapidjson/stringbuffer.h"
#include "fdbclient/rapidjson/writer.h"
#include "fdbrpc/HTTP.h"
#include "fdbrpc/IAsyncFile.h"
#include "fdbserver/KmsConnectorInterface.h"
#include "fdbserver/Knobs.h"
#include "fdbrpc/RESTClient.h"
#include "flow/Arena.h"
#include "flow/EncryptUtils.h"
#include "flow/Error.h"
#include "flow/FastRef.h"
#include "flow/IRandom.h"
#include "flow/Platform.h"
#include "flow/Trace.h"
#include "flow/UnitTest.h"

#include <boost/algorithm/string.hpp>
#include <memory>
#include <queue>
#include <sstream>
#include <unordered_map>
#include <utility>

#include "flow/actorcompiler.h" // This must be the last #include

namespace {
const char* BASE_CIPHER_ID_TAG = "base_cipher_id";
const char* BASE_CIPHER_TAG = "baseCipher";
const char* CIPHER_KEY_DETAILS_TAG = "cipher_key_details";
const char* ENCRYPT_DOMAIN_ID_TAG = "encrypt_domain_id";
const char* ERROR_TAG = "error";
const char* ERROR_DETAIL_TAG = "details";
const char* KMS_URLS_TAG = "kms_urls";
const char* QUERY_MODE_TAG = "query_mode";
const char* REFRESH_KMS_URLS_TAG = "refresh_kms_urls";
const char* VALIDATION_TOKENS_TAG = "validation_tokens";
const char* VALIDATION_TOKEN_NAME_TAG = "token_name";
const char* VALIDATION_TOKEN_VALUE_TAG = "token_value";
const char* DEBUG_UID_TAG = "debug_uid";

const char* TOKEN_NAME_FILE_SEP = "#";
const char* TOKEN_TUPLE_SEP = ",";
const char DISCOVER_URL_FILE_URL_SEP = '\n';

const char* QUERY_MODE_LOOKUP_BY_DOMAIN_ID = "lookupByDomainId";
const char* QUERY_MODE_LOOKUP_BY_KEY_ID = "lookupByKeyId";

bool canReplyWith(Error e) {
	switch (e.code()) {
	case error_code_encrypt_invalid_kms_config:
	case error_code_encrypt_keys_fetch_failed:
	case error_code_file_not_found:
	case error_code_file_too_large:
	case error_code_http_request_failed:
	case error_code_io_error:
	case error_code_operation_failed:
	case error_code_value_too_large:
		return true;
	default:
		return false;
	}
}
} // namespace

struct KmsUrlCtx {
	std::string url;
	uint64_t nRequests;
	uint64_t nFailedResponses;
	uint64_t nResponseParseFailures;

	KmsUrlCtx() : url(""), nRequests(0), nFailedResponses(0), nResponseParseFailures(0) {}
	explicit KmsUrlCtx(const std::string& u) : url(u), nRequests(0), nFailedResponses(0), nResponseParseFailures(0) {}

	bool operator<(const KmsUrlCtx& toCompare) const {
		if (nFailedResponses != toCompare.nFailedResponses) {
			return nFailedResponses > toCompare.nFailedResponses;
		}
		return nResponseParseFailures > toCompare.nResponseParseFailures;
	}
};

enum class ValidationTokenSource {
	VALIDATION_TOKEN_SOURCE_FILE = 1,
	VALIDATION_TOKEN_SOURCE_LAST // Always the last element
};

struct ValidationTokenCtx {
	std::string name;
	std::string value;
	ValidationTokenSource source;
	Optional<std::string> filePath;

	explicit ValidationTokenCtx(const std::string& n, ValidationTokenSource s)
	  : name(n), value(""), source(s), filePath(Optional<std::string>()), readTS(now()) {}
	double getReadTS() const { return readTS; }

private:
	double readTS; // Approach assists refreshing token based on time of creation
};

using KmsUrlMinHeap = std::priority_queue<std::shared_ptr<KmsUrlCtx>,
                                          std::vector<std::shared_ptr<KmsUrlCtx>>,
                                          std::less<std::vector<std::shared_ptr<KmsUrlCtx>>::value_type>>;

struct RESTKmsConnectorCtx : public ReferenceCounted<RESTKmsConnectorCtx> {
	UID uid;
	KmsUrlMinHeap kmsUrlHeap;
	double lastKmsUrlsRefreshTs;
	RESTClient restClient;
	std::unordered_map<std::string, ValidationTokenCtx> validationTokens;

	RESTKmsConnectorCtx() : uid(deterministicRandom()->randomUniqueID()), lastKmsUrlsRefreshTs(0) {}
	explicit RESTKmsConnectorCtx(const UID& id) : uid(id), lastKmsUrlsRefreshTs(0) {}
};

std::string getEncryptionKeysFullUrl(Reference<RESTKmsConnectorCtx> ctx, const std::string& url) {
	if (SERVER_KNOBS->REST_KMS_CONNECTOR_GET_ENCRYPTION_KEYS_ENDPOINT.empty()) {
		TraceEvent("GetEncryptionKeysFullUrl_EmptyEndpoint", ctx->uid).log();
		throw encrypt_invalid_kms_config();
	}

	std::string fullUrl(url);
	return fullUrl.append("/").append(SERVER_KNOBS->REST_KMS_CONNECTOR_GET_ENCRYPTION_KEYS_ENDPOINT);
}

void dropCachedKmsUrls(Reference<RESTKmsConnectorCtx> ctx) {
	while (!ctx->kmsUrlHeap.empty()) {
		std::shared_ptr<KmsUrlCtx> curUrl = ctx->kmsUrlHeap.top();

		TraceEvent("DropCachedKmsUrls", ctx->uid)
		    .detail("Url", curUrl->url)
		    .detail("NumRequests", curUrl->nRequests)
		    .detail("NumFailedResponses", curUrl->nFailedResponses)
		    .detail("NumRespParseFailures", curUrl->nResponseParseFailures);

		ctx->kmsUrlHeap.pop();
	}
}

bool shouldRefreshKmsUrls(Reference<RESTKmsConnectorCtx> ctx) {
	if (!SERVER_KNOBS->REST_KMS_CONNECTOR_REFRESH_KMS_URLS) {
		return false;
	}

	return (now() - ctx->lastKmsUrlsRefreshTs) > SERVER_KNOBS->REST_KMS_CONNECTOR_REFRESH_KMS_URLS_INTERVAL_SEC;
}

void extractKmsUrls(Reference<RESTKmsConnectorCtx> ctx, rapidjson::Document& doc, Reference<HTTP::Response> httpResp) {
	// Refresh KmsUrls cache
	dropCachedKmsUrls(ctx);
	ASSERT(ctx->kmsUrlHeap.empty());

	for (const auto& url : doc[KMS_URLS_TAG].GetArray()) {
		if (!url.IsString()) {
			TraceEvent("DiscoverKmsUrls_MalformedResp", ctx->uid).detail("ResponseContent", httpResp->content);
			throw operation_failed();
		}

		std::string urlStr;
		urlStr.resize(url.GetStringLength());
		memcpy(urlStr.data(), url.GetString(), url.GetStringLength());

		TraceEvent("DiscoverKmsUrls_AddUrl", ctx->uid).detail("Url", urlStr);

		ctx->kmsUrlHeap.emplace(std::make_shared<KmsUrlCtx>(urlStr));
	}

	// Update Kms URLs refresh timestamp
	ctx->lastKmsUrlsRefreshTs = now();
}

ACTOR Future<Void> parseDiscoverKmsUrlFile(Reference<RESTKmsConnectorCtx> ctx, std::string filename) {
	if (filename.empty() || !fileExists(filename)) {
		TraceEvent("DiscoverKmsUrls_FileNotFound", ctx->uid).log();
		throw encrypt_invalid_kms_config();
	}

	state Reference<IAsyncFile> dFile = wait(IAsyncFileSystem::filesystem()->open(
	    filename, IAsyncFile::OPEN_NO_AIO | IAsyncFile::OPEN_READONLY | IAsyncFile::OPEN_UNCACHED, 0644));
	state int64_t fSize = wait(dFile->size());
	state Standalone<StringRef> buff = makeString(fSize);
	int bytesRead = wait(dFile->read(mutateString(buff), fSize, 0));
	if (bytesRead != fSize) {
		TraceEvent("DiscoveryKmsUrl_FileReadShort", ctx->uid)
		    .detail("Filename", filename)
		    .detail("Expected", fSize)
		    .detail("Actual", bytesRead);
		throw io_error();
	}

	// Acceptable file format (new line character separated URLs):
	// <url1>\n
	// <url2>\n

	std::stringstream ss(buff.toString());
	std::string url;
	while (std::getline(ss, url, DISCOVER_URL_FILE_URL_SEP)) {
		std::string trimedUrl = boost::trim_copy(url);
		if (trimedUrl.empty()) {
			// Empty URL, ignore and continue
			continue;
		}
		TraceEvent(SevDebug, "DiscoverKmsUrls_AddUrl", ctx->uid).detail("Url", url);
		ctx->kmsUrlHeap.emplace(std::make_shared<KmsUrlCtx>(url));
	}

	return Void();
}

ACTOR Future<Void> discoverKmsUrls(Reference<RESTKmsConnectorCtx> ctx, bool refreshPersistedUrls) {
	// KMS discovery needs to be done in two scenarios:
	// 1) Initial cluster bootstrap - first boot.
	// 2) Requests to all cached KMS URLs is failing for some reason.
	//
	// Following steps are followed as part of KMS discovery:
	// 1) Based on the configured KMS URL discovery mode, the KMS URLs are extracted and persited in a DynamicKnob
	// enabled configuration knob. Approach allows relying on the parsing configuration supplied discovery URL mode only
	// during afte the initial boot, from then on, the URLs can periodically refreshed along with encryption key fetch
	// requests (SERVER_KNOBS->REST_KMS_CONNECTOR_REFRESH_KMS_URLS needs to be enabled).
	// 2) Cluster will continue using cached KMS URLs (and refreshing them if needed); however, if for some reason, all
	// cached URLs aren't working, then code re-discovers the URL following step#1 and refresh persisted state as well.

	if (!refreshPersistedUrls) {
		// TODO: request must be satisfied accessing KMS URLs persited using DynamicKnobs. Will be implemented once
		// feature is available
	}

	std::string_view mode{ SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MODE };

	if (mode.compare("file") == 0) {
		wait(parseDiscoverKmsUrlFile(ctx, SERVER_KNOBS->REST_KMS_CONNECTOR_DISCOVER_KMS_URL_FILE));
	} else {
		throw not_implemented();
	}

	return Void();
}

void parseKmsResponse(Reference<RESTKmsConnectorCtx> ctx,
                      Reference<HTTP::Response> resp,
                      Arena* arena,
                      std::vector<EncryptCipherKeyDetails>* outCipherKeyDetails) {
	// Acceptable response payload json format:
	//
	// response_json_payload {
	//   "cipher_key_details" : [
	//     {
	//        "base_cipher_id" : <cipherKeyId>,
	//        "encrypt_domain_id" : <domainId>,
	//        "base_cipher" : <baseCipher>
	//     },
	//     {
	//         ....
	//	   }
	//   ],
	//   "kms_urls" : [
	//         "url1", "url2", ...
	//   ],
	//	 "error" : {					// Optional, populated by the KMS, if present, rest of payload is ignored.
	//		"details": <details>
	// 	  }
	// }

	if (resp->code != HTTP::HTTP_STATUS_CODE_OK) {
		// STATUS_OK is gating factor for REST request success
		throw http_request_failed();
	}

	rapidjson::Document doc;
	doc.Parse(resp->content.c_str());

	// Check if response has error
	if (doc.HasMember(ERROR_TAG)) {
		if (doc[ERROR_TAG].HasMember(ERROR_DETAIL_TAG) && doc[ERROR_TAG][ERROR_DETAIL_TAG].IsString()) {
			Standalone<StringRef> errRef = makeString(doc[ERROR_TAG][ERROR_DETAIL_TAG].GetStringLength());
			memcpy(mutateString(errRef),
			       doc[ERROR_TAG][ERROR_DETAIL_TAG].GetString(),
			       doc[ERROR_TAG][ERROR_DETAIL_TAG].GetStringLength());
			TraceEvent("KMSErrorResponse", ctx->uid).detail("ErrorDetails", errRef.toString());
		} else {
			TraceEvent("KMSErrorResponse_EmptyDetails", ctx->uid).log();
		}

		throw encrypt_keys_fetch_failed();
	}

	// Extract CipherKeyDetails
	if (!doc.HasMember(CIPHER_KEY_DETAILS_TAG) || !doc[CIPHER_KEY_DETAILS_TAG].IsArray()) {
		TraceEvent("ParseKmsResponse_FailureMissingCipherKeyDetails", ctx->uid).log();
		throw operation_failed();
	}

	for (const auto& cipherDetail : doc[CIPHER_KEY_DETAILS_TAG].GetArray()) {
		if (!cipherDetail.IsObject()) {
			TraceEvent("ParseKmsResponse_FailureEncryptKeyDetailsNotObject", ctx->uid)
			    .detail("Type", cipherDetail.GetType());
			throw operation_failed();
		}

		const bool isBaseCipherIdPresent = cipherDetail.HasMember(BASE_CIPHER_ID_TAG);
		const bool isBaseCipherPresent = cipherDetail.HasMember(BASE_CIPHER_TAG);
		const bool isEncryptDomainIdPresent = cipherDetail.HasMember(ENCRYPT_DOMAIN_ID_TAG);
		if (!isBaseCipherIdPresent || !isBaseCipherPresent || !isEncryptDomainIdPresent) {
			TraceEvent("ParseKmsResponse_MalformedKeyDetail", ctx->uid)
			    .detail("BaseCipherIdPresent", isBaseCipherIdPresent)
			    .detail("BaseCipherPresent", isBaseCipherPresent)
			    .detail("EncryptDomainIdPresent", isEncryptDomainIdPresent);
			throw operation_failed();
		}

		const int cipherKeyLen = cipherDetail[BASE_CIPHER_TAG].GetStringLength();
		std::unique_ptr<uint8_t[]> cipherKey = std::make_unique<uint8_t[]>(cipherKeyLen);
		memcpy(cipherKey.get(), cipherDetail[BASE_CIPHER_TAG].GetString(), cipherKeyLen);
		outCipherKeyDetails->emplace_back(cipherDetail[ENCRYPT_DOMAIN_ID_TAG].GetInt64(),
		                                  cipherDetail[BASE_CIPHER_ID_TAG].GetUint64(),
		                                  StringRef(cipherKey.get(), cipherKeyLen),
		                                  *arena);
	}

	if (doc.HasMember(KMS_URLS_TAG)) {
		try {
			extractKmsUrls(ctx, doc, resp);
		} catch (Error& e) {
			TraceEvent("RefreshKmsUrls_Failed", ctx->uid).error(e);
			// Given cipherKeyDetails extraction was done successfully, ignore KmsUrls parsing error
		}
	}
}

void addQueryModeSection(Reference<RESTKmsConnectorCtx> ctx, rapidjson::Document& doc, const char* mode) {
	rapidjson::Value key(QUERY_MODE_TAG, doc.GetAllocator());
	rapidjson::Value queryMode;
	queryMode.SetString(mode, doc.GetAllocator());

	// Append 'query_mode' object to the parent document
	doc.AddMember(key, queryMode, doc.GetAllocator());
}

void addValidationTokensSectionToJsonDoc(Reference<RESTKmsConnectorCtx> ctx, rapidjson::Document& doc) {
	// Append "validationTokens" as json array
	rapidjson::Value validationTokens(rapidjson::kArrayType);

	for (const auto& token : ctx->validationTokens) {
		rapidjson::Value validationToken(rapidjson::kObjectType);

		// Add "name" - token name
		rapidjson::Value key(VALIDATION_TOKEN_NAME_TAG, doc.GetAllocator());
		rapidjson::Value tokenName(token.second.name.c_str(), doc.GetAllocator());
		validationToken.AddMember(key, tokenName, doc.GetAllocator());

		// Add "value" - token value
		key.SetString(VALIDATION_TOKEN_VALUE_TAG, doc.GetAllocator());
		rapidjson::Value tokenValue;
		tokenValue.SetString(token.second.value.c_str(), token.second.value.size(), doc.GetAllocator());
		validationToken.AddMember(key, tokenValue, doc.GetAllocator());

		validationTokens.PushBack(validationToken, doc.GetAllocator());
	}

	// Append 'validation_token[]' to the parent document
	rapidjson::Value memberKey(VALIDATION_TOKENS_TAG, doc.GetAllocator());
	doc.AddMember(memberKey, validationTokens, doc.GetAllocator());
}

void addRefreshKmsUrlsSectionToJsonDoc(Reference<RESTKmsConnectorCtx> ctx,
                                       rapidjson::Document& doc,
                                       const bool refreshKmsUrls) {
	rapidjson::Value key(REFRESH_KMS_URLS_TAG, doc.GetAllocator());
	rapidjson::Value refreshUrls;
	refreshUrls.SetBool(refreshKmsUrls);

	// Append 'refresh_kms_urls' object to the parent document
	doc.AddMember(key, refreshUrls, doc.GetAllocator());
}

void addDebugUidSectionToJsonDoc(Reference<RESTKmsConnectorCtx> ctx, rapidjson::Document& doc, Optional<UID> dbgId) {
	if (!dbgId.present()) {
		// Debug id not present; do nothing
		return;
	}
	rapidjson::Value key(DEBUG_UID_TAG, doc.GetAllocator());
	rapidjson::Value debugIdVal;
	const std::string dbgIdStr = dbgId.get().toString();
	debugIdVal.SetString(dbgIdStr.c_str(), dbgIdStr.size(), doc.GetAllocator());

	// Append 'debug_uid' object to the parent document
	doc.AddMember(key, debugIdVal, doc.GetAllocator());
}

StringRef getEncryptKeysByKeyIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx,
                                            const KmsConnLookupEKsByKeyIdsReq& req,
                                            const bool refreshKmsUrls,
                                            Arena& arena) {
	// Acceptable request payload json format:
	//
	// request_json_payload {
	//   "query_mode": "lookupByKeyId" / "lookupByDomainId"
	//   "cipher_key_details" = [
	//     {
	//        "base_cipher_id" : <cipherKeyId>
	//        "encrypt_domain_id" : <domainId>
	//     },
	//     {
	//         ....
	//	   }
	//   ],
	//   "validation_tokens" = [
	//     {
	//        "token_name" : <name>,
	//        "token_value": <value>
	//     },
	//     {
	//         ....
	//     }
	//   ]
	//   "refresh_kms_urls" = 1/0
	//   "debug_uid" = <uid-string>   // Optional debug info to trace requests across FDB <--> KMS
	// }

	rapidjson::Document doc;
	doc.SetObject();

	// Append 'query_mode' object
	addQueryModeSection(ctx, doc, QUERY_MODE_LOOKUP_BY_KEY_ID);

	// Append 'cipher_key_details' as json array
	rapidjson::Value keyIdDetails(rapidjson::kArrayType);
	for (const auto& detail : req.encryptKeyIds) {
		rapidjson::Value keyIdDetail(rapidjson::kObjectType);

		// Add 'base_cipher_id'
		rapidjson::Value key(BASE_CIPHER_ID_TAG, doc.GetAllocator());
		rapidjson::Value baseKeyId;
		baseKeyId.SetUint64(detail.first);
		keyIdDetail.AddMember(key, baseKeyId, doc.GetAllocator());

		// Add 'encrypt_domainId'
		key.SetString(ENCRYPT_DOMAIN_ID_TAG, doc.GetAllocator());
		rapidjson::Value domainId;
		domainId.SetInt64(detail.second);
		keyIdDetail.AddMember(key, domainId, doc.GetAllocator());

		// push above object to the array
		keyIdDetails.PushBack(keyIdDetail, doc.GetAllocator());
	}
	rapidjson::Value memberKey(CIPHER_KEY_DETAILS_TAG, doc.GetAllocator());
	doc.AddMember(memberKey, keyIdDetails, doc.GetAllocator());

	// Append 'validation_tokens' as json array
	addValidationTokensSectionToJsonDoc(ctx, doc);

	// Append 'refresh_kms_urls'
	addRefreshKmsUrlsSectionToJsonDoc(ctx, doc, refreshKmsUrls);

	// Append 'debug_uid' section if needed
	addDebugUidSectionToJsonDoc(ctx, doc, req.debugId);

	// Serialize json to string
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	doc.Accept(writer);

	StringRef ref = makeString(sb.GetSize(), arena);
	memcpy(mutateString(ref), sb.GetString(), sb.GetSize());
	return ref;
}

ACTOR
Future<Void> fetchEncryptionKeys_impl(Reference<RESTKmsConnectorCtx> ctx,
                                      StringRef requestBodyRef,
                                      Arena* arena,
                                      std::vector<EncryptCipherKeyDetails>* outCipherKeyDetails) {
	state Reference<HTTP::Response> resp;

	// Follow 2-phase scheme:
	// Phase-1: Attempt to fetch encryption keys by reaching out to cached KmsUrls in the order of
	//          past success requests success counts.
	// Phase-2: For some reason if none of the cached KmsUrls worked, re-discover the KmsUrls and
	//          repeat phase-1.

	state int pass = 1;
	for (; pass <= 2; pass++) {
		state std::stack<std::shared_ptr<KmsUrlCtx>> tempStack;

		// Iterate over Kms URLs
		while (!ctx->kmsUrlHeap.empty()) {
			state std::shared_ptr<KmsUrlCtx> curUrl = ctx->kmsUrlHeap.top();
			ctx->kmsUrlHeap.pop();
			tempStack.push(curUrl);

			try {
				std::string kmsEncryptionFullUrl = getEncryptionKeysFullUrl(ctx, curUrl->url);
				TraceEvent("FetchEncryptionKeys_Start", ctx->uid).detail("KmsEncryptionFullUrl", kmsEncryptionFullUrl);
				Reference<HTTP::Response> _resp =
				    wait(ctx->restClient.doPost(kmsEncryptionFullUrl, requestBodyRef.toString()));
				resp = _resp;
				curUrl->nRequests++;

				try {
					parseKmsResponse(ctx, resp, arena, outCipherKeyDetails);

					// Push urlCtx back on the ctx->urlHeap
					while (!tempStack.empty()) {
						ctx->kmsUrlHeap.emplace(tempStack.top());
						tempStack.pop();
					}

					TraceEvent("FetchEncryptionKeys_Success", ctx->uid).detail("KmsUrl", curUrl->url);
					return Void();
				} catch (Error& e) {
					TraceEvent("FetchEncryptionKeys_RespParseFailure").error(e);
					curUrl->nResponseParseFailures++;
					// attempt to fetch encryption details from next KmsUrl
				}
			} catch (Error& e) {
				TraceEvent("FetchEncryptionKeys_Failed", ctx->uid).error(e);
				curUrl->nFailedResponses++;
				// attempt to fetch encryption details from next KmsUrl
			}
		}

		if (pass == 1) {
			// Re-discover KMS urls and re-attempt to fetch the encryption key details using newer KMS URLs
			wait(discoverKmsUrls(ctx, true));
		}
	}

	// Failed to fetch encryption keys from the remote KMS
	throw encrypt_keys_fetch_failed();
}

ACTOR Future<KmsConnLookupEKsByKeyIdsRep> fetchEncryptionKeysByKeyIds(Reference<RESTKmsConnectorCtx> ctx,
                                                                      KmsConnLookupEKsByKeyIdsReq req) {
	state KmsConnLookupEKsByKeyIdsRep reply;
	bool refreshKmsUrls = shouldRefreshKmsUrls(ctx);
	std::string requestBody;

	StringRef requestBodyRef = getEncryptKeysByKeyIdsRequestBody(ctx, req, refreshKmsUrls, reply.arena);

	wait(fetchEncryptionKeys_impl(ctx, requestBodyRef, &reply.arena, &reply.cipherKeyDetails));

	return reply;
}

StringRef getEncryptKeysByDomainIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx,
                                               const KmsConnLookupEKsByDomainIdsReq& req,
                                               const bool refreshKmsUrls,
                                               Arena& arena) {
	// Acceptable request payload json format:
	//
	// request_json_payload {
	//   "query_mode": "lookupByKeyId" / "lookupByDomainId"
	//   "cipher_key_details" = [
	//     {
	//        "encrypt_domainId" : <domainId>
	//     },
	//     {
	//         ....
	//	   }
	//   ],
	//   "validation_tokens" = [
	//     {
	//        "token_name" : <name>,
	//        "token_value": <value>
	//     },
	//     {
	//         ....
	//     }
	//   ]
	//   "refresh_kms_urls" = 1/0
	//   "debug_uid" = <uid-string>     // Optional debug info to trace requests across FDB <--> KMS
	// }

	rapidjson::Document doc;
	doc.SetObject();

	// Append 'query_mode' object
	addQueryModeSection(ctx, doc, QUERY_MODE_LOOKUP_BY_DOMAIN_ID);

	// Append 'cipher_key_details' as json array
	rapidjson::Value keyIdDetails(rapidjson::kArrayType);
	for (const auto& detail : req.encryptDomainIds) {
		rapidjson::Value keyIdDetail(rapidjson::kObjectType);

		rapidjson::Value key(ENCRYPT_DOMAIN_ID_TAG, doc.GetAllocator());
		rapidjson::Value domainId;
		domainId.SetInt64(detail);
		keyIdDetail.AddMember(key, domainId, doc.GetAllocator());

		// push above object to the array
		keyIdDetails.PushBack(keyIdDetail, doc.GetAllocator());
	}
	rapidjson::Value memberKey(CIPHER_KEY_DETAILS_TAG, doc.GetAllocator());
	doc.AddMember(memberKey, keyIdDetails, doc.GetAllocator());

	// Append 'validation_tokens' as json array
	addValidationTokensSectionToJsonDoc(ctx, doc);

	// Append 'refresh_kms_urls'
	addRefreshKmsUrlsSectionToJsonDoc(ctx, doc, refreshKmsUrls);

	// Append 'debug_uid' section if needed
	addDebugUidSectionToJsonDoc(ctx, doc, req.debugId);

	// Serialize json to string
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	doc.Accept(writer);

	StringRef ref = makeString(sb.GetSize(), arena);
	memcpy(mutateString(ref), sb.GetString(), sb.GetSize());
	return ref;
}

ACTOR Future<KmsConnLookupEKsByDomainIdsRep> fetchEncryptionKeysByDomainIds(Reference<RESTKmsConnectorCtx> ctx,
                                                                            KmsConnLookupEKsByDomainIdsReq req) {
	state KmsConnLookupEKsByDomainIdsRep reply;
	bool refreshKmsUrls = shouldRefreshKmsUrls(ctx);
	StringRef requestBodyRef = getEncryptKeysByDomainIdsRequestBody(ctx, req, refreshKmsUrls, reply.arena);

	wait(fetchEncryptionKeys_impl(ctx, requestBodyRef, &reply.arena, &reply.cipherKeyDetails));

	return reply;
}

ACTOR Future<Void> procureValidationTokensFromFiles(Reference<RESTKmsConnectorCtx> ctx, std::string details) {
	Standalone<StringRef> detailsRef(details);
	if (details.empty()) {
		TraceEvent("ValidationToken_EmptyFileDetails", ctx->uid).log();
		throw encrypt_invalid_kms_config();
	}

	TraceEvent("ValidationToken", ctx->uid).detail("DetailsStr", details);

	state std::unordered_map<std::string, std::string> tokenFilePathMap;
	while (!details.empty()) {
		StringRef name = detailsRef.eat(TOKEN_NAME_FILE_SEP);
		if (name.empty()) {
			break;
		}
		StringRef path = detailsRef.eat(TOKEN_TUPLE_SEP);
		if (path.empty()) {
			TraceEvent("ValidationToken_FileDetailsMalformed", ctx->uid).detail("FileDetails", details);
			throw operation_failed();
		}

		std::string tokenName = boost::trim_copy(name.toString());
		std::string tokenFile = boost::trim_copy(path.toString());
		if (!fileExists(tokenFile)) {
			TraceEvent("ValidationToken_FileNotFound", ctx->uid)
			    .detail("TokenName", tokenName)
			    .detail("Filename", tokenFile);
			throw encrypt_invalid_kms_config();
		}

		tokenFilePathMap.emplace(tokenName, tokenFile);
		TraceEvent("ValidationToken", ctx->uid).detail("FName", tokenName).detail("Filename", tokenFile);
	}

	// Clear existing cached validation tokens
	ctx->validationTokens.clear();

	// Enumerate all token files and extract details
	state uint64_t tokensPayloadSize = 0;
	for (const auto& item : tokenFilePathMap) {
		state std::string tokenName = item.first;
		state std::string tokenFile = item.second;
		state Reference<IAsyncFile> tFile = wait(IAsyncFileSystem::filesystem()->open(
		    tokenFile, IAsyncFile::OPEN_NO_AIO | IAsyncFile::OPEN_READONLY | IAsyncFile::OPEN_UNCACHED, 0644));

		state int64_t fSize = wait(tFile->size());
		if (fSize > SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MAX_SIZE) {
			TraceEvent("ValidationToken_FileTooLarge", ctx->uid)
			    .detail("FileName", tokenFile)
			    .detail("Size", fSize)
			    .detail("MaxAllowedSize", SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MAX_SIZE);
			throw file_too_large();
		}

		tokensPayloadSize += fSize;
		if (tokensPayloadSize > SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKENS_MAX_PAYLOAD_SIZE) {
			TraceEvent("ValidationToken_PayloadTooLarge", ctx->uid)
			    .detail("MaxAllowedSize", SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKENS_MAX_PAYLOAD_SIZE);
			throw value_too_large();
		}

		state Standalone<StringRef> buff = makeString(fSize);
		int bytesRead = wait(tFile->read(mutateString(buff), fSize, 0));
		if (bytesRead != fSize) {
			TraceEvent("DiscoveryKmsUrl_FileReadShort", ctx->uid)
			    .detail("Filename", tokenFile)
			    .detail("Expected", fSize)
			    .detail("Actual", bytesRead);
			throw io_error();
		}

		// Populate validation token details
		ValidationTokenCtx tokenCtx =
		    ValidationTokenCtx(tokenName, ValidationTokenSource::VALIDATION_TOKEN_SOURCE_FILE);
		tokenCtx.value.resize(fSize);
		memcpy(tokenCtx.value.data(), buff.begin(), fSize);
		tokenCtx.filePath = tokenFile;

		// NOTE: avoid logging token-value to prevent token leaks in log files..
		TraceEvent("ValidationToken_ReadFile", ctx->uid)
		    .detail("TokenName", tokenCtx.name)
		    .detail("TokenSize", tokenCtx.value.size())
		    .detail("TokenFilePath", tokenCtx.filePath.get())
		    .detail("TotalPayloadSize", tokensPayloadSize);

		ctx->validationTokens.emplace(tokenName, std::move(tokenCtx));
	}

	return Void();
}

ACTOR Future<Void> procureValidationTokens(Reference<RESTKmsConnectorCtx> ctx) {
	std::string_view mode{ SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MODE };

	if (mode.compare("file") == 0) {
		wait(procureValidationTokensFromFiles(ctx, SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_DETAILS));
	} else {
		throw not_implemented();
	}

	return Void();
}

ACTOR Future<Void> connectorCore_impl(KmsConnectorInterface interf) {
	state Reference<RESTKmsConnectorCtx> self = makeReference<RESTKmsConnectorCtx>(interf.id());

	TraceEvent("RESTKmsConnector_Init", self->uid).log();

	wait(discoverKmsUrls(self, false /* refreshPersistedUrls */));
	wait(procureValidationTokens(self));

	loop {
		choose {
			when(KmsConnLookupEKsByKeyIdsReq req = waitNext(interf.ekLookupByIds.getFuture())) {
				state KmsConnLookupEKsByKeyIdsReq byKeyIdReq = req;
				state KmsConnLookupEKsByKeyIdsRep byKeyIdResp;
				try {
					KmsConnLookupEKsByKeyIdsRep _rByKeyId = wait(fetchEncryptionKeysByKeyIds(self, byKeyIdReq));
					byKeyIdResp = _rByKeyId;
					byKeyIdReq.reply.send(byKeyIdResp);
				} catch (Error& e) {
					TraceEvent("LookupEKsByKeyIds_Failed", self->uid).error(e);
					if (!canReplyWith(e)) {
						throw e;
					}
					byKeyIdReq.reply.sendError(e);
				}
			}
			when(KmsConnLookupEKsByDomainIdsReq req = waitNext(interf.ekLookupByDomainIds.getFuture())) {
				state KmsConnLookupEKsByDomainIdsReq byDomainIdReq = req;
				state KmsConnLookupEKsByDomainIdsRep byDomainIdResp;
				try {
					KmsConnLookupEKsByDomainIdsRep _rByDomainId =
					    wait(fetchEncryptionKeysByDomainIds(self, byDomainIdReq));
					byDomainIdResp = _rByDomainId;
					byDomainIdReq.reply.send(byDomainIdResp);
				} catch (Error& e) {
					TraceEvent("LookupEKsByDomainIds_Failed", self->uid).error(e);
					if (!canReplyWith(e)) {
						throw e;
					}
					byDomainIdReq.reply.sendError(e);
				}
			}
		}
	}
}

Future<Void> RESTKmsConnector::connectorCore(KmsConnectorInterface interf) {
	return connectorCore_impl(interf);
}

// Only used to link unit tests
void forceLinkRESTKmsConnectorTest() {}

namespace {
std::string_view KMS_URL_NAME_TEST = "http://foo/bar";
uint8_t BASE_CIPHER_KEY_TEST[32];

std::shared_ptr<platform::TmpFile> prepareTokenFile(const uint8_t* buff, const int len) {
	std::shared_ptr<platform::TmpFile> tmpFile = std::make_shared<platform::TmpFile>("/tmp");
	ASSERT(fileExists(tmpFile->getFileName()));
	tmpFile->write(buff, len);
	return tmpFile;
}

std::shared_ptr<platform::TmpFile> prepareTokenFile(const int tokenLen) {
	Standalone<StringRef> buff = makeString(tokenLen);
	generateRandomData(mutateString(buff), tokenLen);

	return prepareTokenFile(buff.begin(), tokenLen);
}

ACTOR Future<Void> testEmptyValidationFileDetails(Reference<RESTKmsConnectorCtx> ctx) {
	try {
		wait(procureValidationTokensFromFiles(ctx, ""));
		ASSERT(false);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_encrypt_invalid_kms_config);
	}
	return Void();
}

ACTOR Future<Void> testMalformedFileValidationTokenDetails(Reference<RESTKmsConnectorCtx> ctx) {
	try {
		wait(procureValidationTokensFromFiles(ctx, "abdc/tmp/foo"));
		ASSERT(false);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_operation_failed);
	}

	return Void();
}

ACTOR Future<Void> testValidationTokenFileNotFound(Reference<RESTKmsConnectorCtx> ctx) {
	try {
		wait(procureValidationTokensFromFiles(ctx, "foo#/imaginary-dir/dream/phantom-file"));
		ASSERT(false);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_encrypt_invalid_kms_config);
	}
	return Void();
}

ACTOR Future<Void> testTooLargeValidationTokenFile(Reference<RESTKmsConnectorCtx> ctx) {
	std::string name("foo");
	const int tokenLen = SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MAX_SIZE + 1;

	state std::shared_ptr<platform::TmpFile> tmpFile = prepareTokenFile(tokenLen);

	std::string details;
	details.append(name).append(TOKEN_NAME_FILE_SEP).append(tmpFile->getFileName());

	try {
		wait(procureValidationTokensFromFiles(ctx, details));
		ASSERT(false);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_file_too_large);
	}

	return Void();
}

ACTOR Future<Void> testValidationFileTokenPayloadTooLarge(Reference<RESTKmsConnectorCtx> ctx) {
	const int tokenLen = SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MAX_SIZE;
	const int nTokens = SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKENS_MAX_PAYLOAD_SIZE /
	                        SERVER_KNOBS->REST_KMS_CONNECTOR_VALIDATION_TOKEN_MAX_SIZE +
	                    2;
	Standalone<StringRef> buff = makeString(tokenLen);
	generateRandomData(mutateString(buff), tokenLen);

	std::string details;
	state std::vector<std::shared_ptr<platform::TmpFile>> tokenfiles;
	for (int i = 0; i < nTokens; i++) {
		std::shared_ptr<platform::TmpFile> tokenfile = prepareTokenFile(buff.begin(), tokenLen);

		details.append(std::to_string(i)).append(TOKEN_NAME_FILE_SEP).append(tokenfile->getFileName());
		if (i < nTokens)
			details.append(TOKEN_TUPLE_SEP);
		tokenfiles.emplace_back(tokenfile);
	}

	try {
		wait(procureValidationTokensFromFiles(ctx, details));
		ASSERT(false);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_value_too_large);
	}

	return Void();
}

ACTOR Future<Void> testMultiValidationFileTokenFiles(Reference<RESTKmsConnectorCtx> ctx) {
	state int numFiles = deterministicRandom()->randomInt(2, 5);
	state int tokenLen = deterministicRandom()->randomInt(26, 75);
	state Standalone<StringRef> buff = makeString(tokenLen);
	state std::unordered_map<std::string, std::shared_ptr<platform::TmpFile>> tokenFiles;
	state std::unordered_map<std::string, std::string> tokenNameValueMap;
	state std::string tokenDetailsStr;

	generateRandomData(mutateString(buff), tokenLen);

	for (int i = 1; i <= numFiles; i++) {
		std::string tokenName = std::to_string(i);
		std::shared_ptr<platform::TmpFile> tokenfile = prepareTokenFile(buff.begin(), tokenLen);

		std::string token((char*)buff.begin(), tokenLen);
		tokenFiles.emplace(tokenName, tokenfile);
		tokenNameValueMap.emplace(std::to_string(i), token);
		tokenDetailsStr.append(tokenName).append(TOKEN_NAME_FILE_SEP).append(tokenfile->getFileName());
		if (i < numFiles)
			tokenDetailsStr.append(TOKEN_TUPLE_SEP);
	}

	wait(procureValidationTokensFromFiles(ctx, tokenDetailsStr));

	ASSERT_EQ(ctx->validationTokens.size(), tokenNameValueMap.size());

	for (const auto& token : ctx->validationTokens) {
		const auto& itr = tokenNameValueMap.find(token.first);
		const ValidationTokenCtx& tokenCtx = token.second;

		ASSERT(itr != tokenNameValueMap.end());
		ASSERT_EQ(token.first.compare(itr->first), 0);
		ASSERT_EQ(tokenCtx.value.compare(itr->second), 0);
		ASSERT_EQ(tokenCtx.source, ValidationTokenSource::VALIDATION_TOKEN_SOURCE_FILE);
		ASSERT(tokenCtx.filePath.present());
		ASSERT_EQ(tokenCtx.filePath.compare(tokenFiles[tokenCtx.name]->getFileName()), 0);
		ASSERT_NE(tokenCtx.getReadTS(), 0);
	}

	return Void();
}

EncryptCipherDomainId getRandomDomainId() {
	const int lottery = deterministicRandom()->randomInt(0, 100);
	if (lottery < 10) {
		return SYSTEM_KEYSPACE_ENCRYPT_DOMAIN_ID;
	} else if (lottery >= 10 && lottery < 25) {
		return ENCRYPT_HEADER_DOMAIN_ID;
	} else {
		return lottery;
	}
}

void getFakeKmsResponse(StringRef jsonReqRef, const bool baseCipherIdPresent, Reference<HTTP::Response> httpResponse) {
	rapidjson::Document reqDoc;
	reqDoc.Parse(jsonReqRef.toString().c_str());

	rapidjson::Document resDoc;
	resDoc.SetObject();

	ASSERT(reqDoc.HasMember(CIPHER_KEY_DETAILS_TAG) && reqDoc[CIPHER_KEY_DETAILS_TAG].IsArray());

	rapidjson::Value cipherKeyDetails(rapidjson::kArrayType);
	for (const auto& detail : reqDoc[CIPHER_KEY_DETAILS_TAG].GetArray()) {
		rapidjson::Value keyDetail(rapidjson::kObjectType);

		ASSERT(detail.HasMember(ENCRYPT_DOMAIN_ID_TAG));

		rapidjson::Value key(ENCRYPT_DOMAIN_ID_TAG, resDoc.GetAllocator());
		rapidjson::Value domainId;
		domainId.SetInt64(detail[ENCRYPT_DOMAIN_ID_TAG].GetInt64());
		keyDetail.AddMember(key, domainId, resDoc.GetAllocator());

		key.SetString(BASE_CIPHER_ID_TAG, resDoc.GetAllocator());
		rapidjson::Value baseCipherId;
		if (detail.HasMember(BASE_CIPHER_ID_TAG)) {
			domainId.SetUint64(detail[BASE_CIPHER_ID_TAG].GetUint64());
		} else {
			ASSERT(!baseCipherIdPresent);
			domainId.SetUint(1234);
		}
		keyDetail.AddMember(key, domainId, resDoc.GetAllocator());

		key.SetString(BASE_CIPHER_TAG, resDoc.GetAllocator());
		rapidjson::Value baseCipher;
		baseCipher.SetString((char*)&BASE_CIPHER_KEY_TEST[0], sizeof(BASE_CIPHER_KEY_TEST), resDoc.GetAllocator());
		keyDetail.AddMember(key, baseCipher, resDoc.GetAllocator());

		cipherKeyDetails.PushBack(keyDetail, resDoc.GetAllocator());
	}
	rapidjson::Value memberKey(CIPHER_KEY_DETAILS_TAG, resDoc.GetAllocator());
	resDoc.AddMember(memberKey, cipherKeyDetails, resDoc.GetAllocator());

	ASSERT(reqDoc.HasMember(REFRESH_KMS_URLS_TAG));
	if (reqDoc[REFRESH_KMS_URLS_TAG].GetBool()) {
		rapidjson::Value kmsUrls(rapidjson::kArrayType);
		for (int i = 0; i < 3; i++) {
			rapidjson::Value url;
			url.SetString(KMS_URL_NAME_TEST.data(), resDoc.GetAllocator());
			kmsUrls.PushBack(url, resDoc.GetAllocator());
		}
		memberKey.SetString(KMS_URLS_TAG, resDoc.GetAllocator());
		resDoc.AddMember(memberKey, kmsUrls, resDoc.GetAllocator());
	}

	// Serialize json to string
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	resDoc.Accept(writer);
	httpResponse->content.resize(sb.GetSize(), '\0');
	memcpy(httpResponse->content.data(), sb.GetString(), sb.GetSize());
}

void validateKmsUrls(Reference<RESTKmsConnectorCtx> ctx) {
	ASSERT_EQ(ctx->kmsUrlHeap.size(), 3);
	std::shared_ptr<KmsUrlCtx> urlCtx = ctx->kmsUrlHeap.top();
	ASSERT_EQ(urlCtx->url.compare(KMS_URL_NAME_TEST), 0);
}

void testGetEncryptKeysByKeyIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx, Arena arena) {
	KmsConnLookupEKsByKeyIdsReq req;
	std::unordered_map<EncryptCipherBaseKeyId, EncryptCipherDomainId> keyMap;
	const int nKeys = deterministicRandom()->randomInt(7, 8);
	for (int i = 1; i < nKeys; i++) {
		EncryptCipherDomainId domainId = getRandomDomainId();
		req.encryptKeyIds.push_back(std::make_pair(i, domainId));
		keyMap[i] = domainId;
	}

	bool refreshKmsUrls = deterministicRandom()->randomInt(0, 100) < 50;
	if (deterministicRandom()->randomInt(0, 100) < 40) {
		req.debugId = deterministicRandom()->randomUniqueID();
	}

	StringRef requestBodyRef = getEncryptKeysByKeyIdsRequestBody(ctx, req, refreshKmsUrls, arena);
	TraceEvent("FetchKeysByKeyIds", ctx->uid).setMaxFieldLength(100000).detail("JsonReqStr", requestBodyRef.toString());
	Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>();
	httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
	getFakeKmsResponse(requestBodyRef, true, httpResp);
	TraceEvent("FetchKeysByKeyIds", ctx->uid).setMaxFieldLength(100000).detail("HttpRespStr", httpResp->content);

	std::vector<EncryptCipherKeyDetails> cipherDetails;
	parseKmsResponse(ctx, httpResp, &arena, &cipherDetails);
	ASSERT_EQ(cipherDetails.size(), keyMap.size());
	for (const auto& detail : cipherDetails) {
		ASSERT(keyMap.find(detail.encryptKeyId) != keyMap.end());
		ASSERT_EQ(keyMap[detail.encryptKeyId], detail.encryptDomainId);
		ASSERT_EQ(detail.encryptKey.size(), sizeof(BASE_CIPHER_KEY_TEST));
		ASSERT_EQ(memcmp(detail.encryptKey.begin(), &BASE_CIPHER_KEY_TEST[0], sizeof(BASE_CIPHER_KEY_TEST)), 0);
	}
	if (refreshKmsUrls) {
		validateKmsUrls(ctx);
	}
}

void testGetEncryptKeysByDomainIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx, Arena arena) {
	KmsConnLookupEKsByDomainIdsReq req;
	std::unordered_set<EncryptCipherDomainId> domainIdsSet;
	const int nKeys = deterministicRandom()->randomInt(7, 25);
	for (int i = 1; i < nKeys; i++) {
		domainIdsSet.emplace(getRandomDomainId());
	}
	req.encryptDomainIds.insert(req.encryptDomainIds.begin(), domainIdsSet.begin(), domainIdsSet.end());

	bool refreshKmsUrls = deterministicRandom()->randomInt(0, 100) < 50;

	StringRef jsonReqRef = getEncryptKeysByDomainIdsRequestBody(ctx, req, refreshKmsUrls, arena);
	TraceEvent("FetchKeysByDomainIds", ctx->uid).detail("JsonReqStr", jsonReqRef.toString());
	Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>();
	httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
	getFakeKmsResponse(jsonReqRef, false, httpResp);
	TraceEvent("FetchKeysByDomainIds", ctx->uid).detail("HttpRespStr", httpResp->content);

	std::vector<EncryptCipherKeyDetails> cipherDetails;
	parseKmsResponse(ctx, httpResp, &arena, &cipherDetails);
	ASSERT_EQ(domainIdsSet.size(), cipherDetails.size());
	for (const auto& detail : cipherDetails) {
		ASSERT(domainIdsSet.find(detail.encryptDomainId) != domainIdsSet.end());
		ASSERT_EQ(detail.encryptKey.size(), sizeof(BASE_CIPHER_KEY_TEST));
		ASSERT_EQ(memcmp(detail.encryptKey.begin(), &BASE_CIPHER_KEY_TEST[0], sizeof(BASE_CIPHER_KEY_TEST)), 0);
	}
	if (refreshKmsUrls) {
		validateKmsUrls(ctx);
	}
}

void testMissingCipherDetailsTag(Reference<RESTKmsConnectorCtx> ctx) {
	Arena arena;
	std::vector<EncryptCipherKeyDetails> cipherDetails;

	rapidjson::Document doc;
	doc.SetObject();

	rapidjson::Value key(KMS_URLS_TAG, doc.GetAllocator());
	rapidjson::Value refreshUrl;
	refreshUrl.SetBool(true);
	doc.AddMember(key, refreshUrl, doc.GetAllocator());

	Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>();
	httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	doc.Accept(writer);
	httpResp->content.resize(sb.GetSize(), '\0');
	memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize());

	try {
		parseKmsResponse(ctx, httpResp, &arena, &cipherDetails);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_operation_failed);
	}
}

void testMalformedCipherDetails(Reference<RESTKmsConnectorCtx> ctx) {
	Arena arena;
	std::vector<EncryptCipherKeyDetails> cipherDetails;

	rapidjson::Document doc;
	doc.SetObject();

	rapidjson::Value key(CIPHER_KEY_DETAILS_TAG, doc.GetAllocator());
	rapidjson::Value details;
	details.SetBool(true);
	doc.AddMember(key, details, doc.GetAllocator());

	Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>();
	httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	doc.Accept(writer);
	httpResp->content.resize(sb.GetSize(), '\0');
	memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize());

	try {
		parseKmsResponse(ctx, httpResp, &arena, &cipherDetails);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_operation_failed);
	}
}

void testMalfromedCipherDetailObj(Reference<RESTKmsConnectorCtx> ctx) {
	Arena arena;
	std::vector<EncryptCipherKeyDetails> cipherDetails;

	rapidjson::Document doc;
	doc.SetObject();

	rapidjson::Value cDetails(rapidjson::kArrayType);
	rapidjson::Value detail(rapidjson::kObjectType);
	rapidjson::Value key(BASE_CIPHER_ID_TAG, doc.GetAllocator());
	rapidjson::Value id;
	id.SetUint(12345);
	detail.AddMember(key, id, doc.GetAllocator());
	cDetails.PushBack(detail, doc.GetAllocator());
	key.SetString(CIPHER_KEY_DETAILS_TAG, doc.GetAllocator());
	doc.AddMember(key, cDetails, doc.GetAllocator());

	Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>();
	httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	doc.Accept(writer);
	httpResp->content.resize(sb.GetSize(), '\0');
	memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize());

	try {
		parseKmsResponse(ctx, httpResp, &arena, &cipherDetails);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_operation_failed);
	}
}

void testKMSErrorResponse(Reference<RESTKmsConnectorCtx> ctx) {
	Arena arena;
	std::vector<EncryptCipherKeyDetails> cipherDetails;

	rapidjson::Document doc;
	doc.SetObject();

	// Construct fake response, it should get ignored anyways
	rapidjson::Value cDetails(rapidjson::kArrayType);
	rapidjson::Value detail(rapidjson::kObjectType);
	rapidjson::Value key(BASE_CIPHER_ID_TAG, doc.GetAllocator());
	rapidjson::Value id;
	id.SetUint(12345);
	detail.AddMember(key, id, doc.GetAllocator());
	cDetails.PushBack(detail, doc.GetAllocator());
	key.SetString(CIPHER_KEY_DETAILS_TAG, doc.GetAllocator());
	doc.AddMember(key, cDetails, doc.GetAllocator());

	// Add error tag
	rapidjson::Value errorTag(rapidjson::kObjectType);

	// Add 'error_detail'
	rapidjson::Value eKey(ERROR_DETAIL_TAG, doc.GetAllocator());
	rapidjson::Value detailInfo;
	detailInfo.SetString("Foo is always bad", doc.GetAllocator());
	errorTag.AddMember(eKey, detailInfo, doc.GetAllocator());

	key.SetString(ERROR_TAG, doc.GetAllocator());
	doc.AddMember(key, errorTag, doc.GetAllocator());

	Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>();
	httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
	rapidjson::StringBuffer sb;
	rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
	doc.Accept(writer);
	httpResp->content.resize(sb.GetSize(), '\0');
	memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize());

	try {
		parseKmsResponse(ctx, httpResp, &arena, &cipherDetails);
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_encrypt_keys_fetch_failed);
	}
}

ACTOR Future<Void> testParseDiscoverKmsUrlFileNotFound(Reference<RESTKmsConnectorCtx> ctx) {
	try {
		wait(parseDiscoverKmsUrlFile(ctx, "/imaginary-dir/dream/phantom-file"));
	} catch (Error& e) {
		ASSERT_EQ(e.code(), error_code_encrypt_invalid_kms_config);
	}
	return Void();
}

ACTOR Future<Void> testParseDiscoverKmsUrlFile(Reference<RESTKmsConnectorCtx> ctx) {
	state std::shared_ptr<platform::TmpFile> tmpFile = std::make_shared<platform::TmpFile>("/tmp");
	ASSERT(fileExists(tmpFile->getFileName()));

	state std::unordered_set<std::string> urls;
	urls.emplace("https://127.0.0.1/foo");
	urls.emplace("https://127.0.0.1/foo1");
	urls.emplace("https://127.0.0.1/foo2");

	std::string content;
	for (auto& url : urls) {
		content.append(url);
		content.push_back(DISCOVER_URL_FILE_URL_SEP);
	}
	tmpFile->write((const uint8_t*)content.c_str(), content.size());
	wait(parseDiscoverKmsUrlFile(ctx, tmpFile->getFileName()));

	ASSERT_EQ(ctx->kmsUrlHeap.size(), urls.size());
	while (!ctx->kmsUrlHeap.empty()) {
		std::shared_ptr<KmsUrlCtx> urlCtx = ctx->kmsUrlHeap.top();
		ctx->kmsUrlHeap.pop();

		ASSERT(urls.find(urlCtx->url) != urls.end());
		ASSERT_EQ(urlCtx->nFailedResponses, 0);
		ASSERT_EQ(urlCtx->nRequests, 0);
		ASSERT_EQ(urlCtx->nResponseParseFailures, 0);
	}

	return Void();
}

} // namespace

TEST_CASE("/KmsConnector/REST/ParseKmsDiscoveryUrls") {
	state Reference<RESTKmsConnectorCtx> ctx = makeReference<RESTKmsConnectorCtx>();
	state Arena arena;

	// initialize cipher key used for testing
	generateRandomData(&BASE_CIPHER_KEY_TEST[0], 32);

	wait(testParseDiscoverKmsUrlFileNotFound(ctx));
	wait(testParseDiscoverKmsUrlFile(ctx));

	return Void();
}

TEST_CASE("/KmsConnector/REST/ParseValidationTokenFile") {
	state Reference<RESTKmsConnectorCtx> ctx = makeReference<RESTKmsConnectorCtx>();
	state Arena arena;

	// initialize cipher key used for testing
	generateRandomData(&BASE_CIPHER_KEY_TEST[0], 32);

	wait(testEmptyValidationFileDetails(ctx));
	wait(testMalformedFileValidationTokenDetails(ctx));
	wait(testValidationTokenFileNotFound(ctx));
	wait(testTooLargeValidationTokenFile(ctx));
	wait(testValidationFileTokenPayloadTooLarge(ctx));
	wait(testMultiValidationFileTokenFiles(ctx));

	return Void();
}

TEST_CASE("/KmsConnector/REST/ParseKmsResponse") {
	state Reference<RESTKmsConnectorCtx> ctx = makeReference<RESTKmsConnectorCtx>();
	state Arena arena;

	// initialize cipher key used for testing
	generateRandomData(&BASE_CIPHER_KEY_TEST[0], 32);

	testMissingCipherDetailsTag(ctx);
	testMalformedCipherDetails(ctx);
	testMalfromedCipherDetailObj(ctx);
	testKMSErrorResponse(ctx);
	return Void();
}

TEST_CASE("/KmsConnector/REST/GetEncryptionKeyOps") {
	state Reference<RESTKmsConnectorCtx> ctx = makeReference<RESTKmsConnectorCtx>();
	state Arena arena;

	// initialize cipher key used for testing
	generateRandomData(&BASE_CIPHER_KEY_TEST[0], 32);

	// Prepare KmsConnector context details
	wait(testParseDiscoverKmsUrlFile(ctx));
	wait(testMultiValidationFileTokenFiles(ctx));

	const int numIterations = deterministicRandom()->randomInt(512, 786);
	for (int i = 0; i < numIterations; i++) {
		testGetEncryptKeysByKeyIdsRequestBody(ctx, arena);
		testGetEncryptKeysByDomainIdsRequestBody(ctx, arena);
	}
	return Void();
}

max_stars_repo_name: sfc-gh-bvr/foundationdb

max_issues_repo_name: sfc-gh-bvr/foundationdb

max_forks_repo_name: sfc-gh-bvr/foundationdb


text:
/*
   Copyright 2018 Simon Vogl <svogl@voxel.at>
                  Angel Merino-Sastre <amerino@voxel.at>

   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.
*/
#include <iostream>

#include <boost/log/trivial.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#if OCV_VERSION_MAJOR >= 3
#  include <opencv2/imgproc.hpp>
#  include <opencv2/highgui.hpp>
#else
#  include <opencv2/imgproc/imgproc.hpp>
#  include <opencv2/highgui/highgui.hpp>
#endif

#include "toffy/tracking/tracker.hpp"

using namespace toffy;
using namespace toffy::tracking;
using namespace cv;
using namespace std;

#ifdef CM_DEBUG
const bool dbg=true;
#else
const bool dbg=false;
#endif

std::size_t Tracker::_filter_counter = 1;
const std::string Tracker::id_name = "tracker";

Tracker::Tracker(): Filter(Tracker::id_name, _filter_counter),
    _in_vec("dect"), _in_fc("fc"), _in_img("img"),
    _out_img(_in_img), _out_objects("objects"),
    _out_count("count"), maxMergeDistance(20.),
    _render_image(false)
{
    _filter_counter++;

}

void Tracker::updateConfig(const boost::property_tree::ptree &pt) {
    BOOST_LOG_TRIVIAL(debug) << __FUNCTION__ <<  " " << id();

    using namespace boost::property_tree;

    Filter::updateConfig(pt);

    _in_vec = pt.get<string>("inputs.vec",_in_vec);
    _in_fc = pt.get<string>("inputs.fc",_in_fc);
    _in_img = pt.get<string>("inputs.img",_in_img);

    _out_img = pt.get<string>("outputs.img",_out_img);
    _out_objects = pt.get<string>("outputs.objects",_out_objects);
    _out_count = pt.get<string>("outputs.count",_out_count);

    maxMergeDistance = pt.get<double>("options.maxMergeDistance",maxMergeDistance);
    _render_image = pt.get<bool>("options.renderImage",_render_image);

}

boost::property_tree::ptree Tracker::getConfig() const {
    boost::property_tree::ptree pt;

    pt = Filter::getConfig();
    //pt.put("options.max_x", _max_x);

    pt.put("inputs.vec", _in_vec);
    pt.put("inputs.fc", _in_fc);
    pt.put("inputs.img", _in_img);
    pt.put("outputs.img", _out_img);
    pt.put("outputs.objects", _out_objects);
    pt.put("outputs.count", _out_count);

    pt.put("options.renderImage", _render_image);
    pt.put("options.maxMergeDistance", maxMergeDistance);

    return pt;
}

bool Tracker::filter(const Frame& in, Frame& out) {
    BOOST_LOG_TRIVIAL(debug) << __FUNCTION__ << " " << id();

    toffy::Filter::setLoggingLvl();

    //Get list of detected blobs
    boost::shared_ptr<std::vector<detection::DetectedObject*> > blobs;
    try {
        blobs = boost::any_cast<boost::shared_ptr<std::vector<detection::DetectedObject*> > >(in.getData(_in_vec));
    } catch(const boost::bad_any_cast &) {
        BOOST_LOG_TRIVIAL(warning) << "Could not find object vector: "
                                   << _in_vec << ". " << id();
        return true;
    }

    BOOST_LOG_TRIVIAL(debug) << "Got _in_vec size: " << blobs->size();

    //Current frame counter
    try {
        _fc = in.getUInt(_in_fc);
    } catch(const boost::bad_any_cast &) {
        BOOST_LOG_TRIVIAL(warning) << "Could not find framecounter. " << id();
        _fc = -1;
    }

    //Current Timestamp
    try {
        _ts = in.getUInt("ts");
    } catch(const boost::bad_any_cast &) {
        BOOST_LOG_TRIVIAL(warning) << "Could not find Timestamp. " << id();
        _ts = -1;
    }

    // List of tracked objects
    boost::shared_ptr<std::vector<detection::DetectedObject*> > tracked;
    try {
        tracked = boost::any_cast<boost::shared_ptr<std::vector<detection::DetectedObject*> > >(out.getData(_out_objects));
    } catch(const boost::bad_any_cast &) {
        BOOST_LOG_TRIVIAL(warning) << id() << "Could not find object vector. Creating a new one";
        tracked.reset(new std::vector<detection::DetectedObject*>);
    }

    vector<detection::DetectedObject*> newState; // the new state..

    // 1. compare all old objects against the candidates, extract best matches
    // 2. for all candidates that could not be matched, add them if applicable
    // 3. eliminate all old detObjs ( not seen >5 frames)
    // 4. save the new state.

    ///// 1.: compare all old objects against the candidates, extract best matches

    BOOST_LOG_TRIVIAL(debug) << "detObjs: " << detObjs.size();
    BOOST_LOG_TRIVIAL(debug) << "blobs: " << blobs->size();

    vector<detection::DetectedObject*>::iterator objDetIter = detObjs.begin();
    while ( objDetIter != detObjs.end() ) {
        detection::DetectedObject* obj = *objDetIter;
        detection::DetectedObject* blob = NULL;

        //Compare the object massCenter distances, find best match
        int candidate = -1; //Flag for potential match
        double minDis = maxMergeDistance; // Distance to merge
        for (size_t j = 0; j < blobs->size();j++) {
            blob = (*blobs)[j];
            if ( !blob )
                continue; //For removed blob from input list

            double ndis = norm(obj->massCenter - blob->massCenter);

            if (ndis < minDis) {
                minDis = ndis;
                candidate = j;
            }
        }
        // if no near candidate has been found assume a new obj (step 2).
        if ( minDis > maxMergeDistance ) {
            candidate = -1;
            blob = 0;
        }
        if (candidate >= 0  ) {
            blob = (*blobs)[candidate];
            blob->size = contourArea(blob->contour);
            /*
        BOOST_LOG_TRIVIAL(debug) <<
            "Found candidate object id = " <<
            blob->id <<
            "\t" << minDis <<
            "\t" << blob->size <<
            "\t" << blob->contour.size() <<
            "\t" << blob->massCenter;
        */
            // found a match, update obj data:
            obj->update(*blob);

            // TODO We destroy the input blob list!!!
            // remove blob from candidates & delete
            delete blob;
            (*blobs)[candidate] = NULL;

            // move obj to new state
            newState.push_back(obj);
            objDetIter = detObjs.erase(objDetIter);
        } else {
            objDetIter++;
        }
    }

    /////// 2.: add all valid new candidates:
    vector<detection::DetectedObject*>::iterator candIter = blobs->begin();
    while ( candIter != blobs->end() ) {
        detection::DetectedObject* blob = *candIter;
        if (blob == NULL) {
            candIter++;
            continue;
        }

        //TODO Parameter?
        if (blob->contour.size() < 4) {
            delete blob;
            blob = NULL;
            //candIter = candidates.erase(candIter);
            candIter++;
        } else {
            // init new detected object
            blob->init();
            /*blob->first_fc = blob->fc;
            blob->first_cts = blob->cts;
            blob->first_ts = blob->ts;
            blob->firstCenter = blob->massCenter;
            blob->size = contourArea(blob->contour);
            blob->record.reset(new boost::circular_buffer<detection::DetectedObject* >(10));*/

            newState.push_back(blob);

            candIter++;
        }
    }
    //TODO do not delete input blobs!
    blobs->clear();

    /////// 3. eliminate all old detObjs ( not seen >5 frames)
    while ( objDetIter != detObjs.end() ) {
        detection::DetectedObject* obj = *objDetIter;

        //TODO Parameter!
        if (abs(_fc - obj->fc) > 5 ) {
            delete obj;
        } else {
            newState.push_back( obj );
        }
    }
    // at this point, the contents of detObjs is not touched any more and will be overwritten.

    ///// 4. save the new state.
    detObjs = newState;
    out.addData(_out_count, detObjs.size());

    *tracked = newState;

    out.addData(_out_objects, tracked);

#if 0
    // TODO Where come img from?, why we mask it with the objects.
    BOOST_LOG_TRIVIAL(debug) << "Getting got _in_img: " << _in_img;
    boost::shared_ptr<cv::Mat> img;
    try {
        img = in.getMatPtr(_in_img);
    } catch(const boost::bad_any_cast &) {
        BOOST_LOG_TRIVIAL(warning) <<
                                      "Could not cast input " << _in_img <<
                                      ", filter  " << id() <<" does not show objects.";
        return true;
    }

    detection::DetectedObject *d = tracked->at(0);
    Mat mask = Mat::zeros(img->size(),CV_8U);
    for (size_t i = 0; i < tracked->size(); i++) {
        drawContours( mask, *(tracked->at(i)->contours), tracked->at(i)->idx, Scalar(255),
                      FILLED, LINE_AA, *(tracked->at(i)->hierarchy));
        //circle( imgCopy, o->massCenter, 2, Scalar( 0, 255, 255 ), FILLED, LINE_AA);
    }
    //imshow("<<<<<", imgCopy);
    *img = img->setTo(0,~mask);
#endif
    //Debug, shows image with tracked objects
    if (_render_image) {

        BOOST_LOG_TRIVIAL(debug) << "Getting got _in_img: " << _in_img;
        boost::shared_ptr<cv::Mat> img;
        try {
            img = in.getMatPtr(_in_img);
        } catch(const boost::bad_any_cast &) {
            BOOST_LOG_TRIVIAL(warning) <<
                                          "Could not cast input " << _in_img <<
                                          ", filter  " << id() <<" does not show objects.";
            return true;
        }

        BOOST_LOG_TRIVIAL(debug) << "Getting got _out_img: " << _out_img;
        boost::shared_ptr<cv::Mat> img_out;
        try {
            img_out = in.getMatPtr(_out_img);
        } catch(const boost::bad_any_cast &) {
            BOOST_LOG_TRIVIAL(warning) <<
                                          "Could not cast output " << _out_img;
            img_out.reset(new Mat());
            out.addData(_out_img,img_out);
        }

        img->copyTo(*img_out);
        showObjects(*img_out);
    }
    return true;
}

void Tracker::showObjects(cv::Mat& depth) {
    double max, min;
    minMaxIdx(depth, &min, &max);
    depth.convertTo(depth,CV_8U,255.0/(max-min),-255.0*min/(max-min));
    cvtColor(depth,depth, COLOR_GRAY2RGB);

    //BOOST_LOG_TRIVIAL(debug) << "showObjects " << detObjs.size();
    for( size_t i = 0; i< detObjs.size(); i++ )
    {
        if (detObjs[i]->first_fc == _fc) {
            circle(depth, detObjs[i]->massCenter, 4, detObjs[i]->color);
            continue;
        }
        try {
            if (detObjs[i]->fc == _fc) {
                vector<vector<Point> > contours;
                contours.push_back(detObjs[i]->contour);
                drawContours( depth, contours, 0, detObjs[i]->color, 1, LINE_AA, noArray(), 0, Point() );
                circle(depth, detObjs[i]->massCenter, 2, detObjs[i]->color);
            }
        } catch (std::exception &e) {
            LOG(warning) << "auweh - could not paint detObjs.s=" << detObjs.size() << " i= " << i ;
        }

    }

    if (dbg) cv::imshow("objectsDetc", depth);

    return;
}

max_stars_repo_name: voxel-dot-at/toffy

max_issues_repo_name: voxel-dot-at/toffy

max_forks_repo_name: voxel-dot-at/toffy