--- /dev/null
- set(CMAKE_CXX_FLAGS "-std=c++11 -Wno-write-strings ${CMAKE_CXX_FLAGS}")
+cmake_minimum_required(VERSION 2.8.11)
+
- set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/cmake/modules/")
-
- if(DO_NOT_DELAY_TAG_CALC)
- set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DDO_NOT_DELAY_TAG_CALC")
- endif()
-
- if(K_WAY_HEAP)
- if(K_WAY_HEAP LESS 2)
- message(FATAL_ERROR "K_WAY_HEAP value should be at least 2")
- else()
- set(CMAKE_CXX_SIM_FLAGS "-DK_WAY_HEAP=${K_WAY_HEAP}")
++list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/cmake/modules")
+
- if (NOT(TARGET gtest AND TARGET gtest_main))
- find_package(gtest REQUIRED)
- include_directories(${GTEST_INCLUDE_DIRS})
++if (NOT(TARGET gtest AND TARGET gtest_main))
++ if(NOT(GTEST_FOUND))
++ find_package(GTest REQUIRED)
+ endif()
+endif()
+
- find_package(Boost REQUIRED)
- include_directories(${Boost_INCLUDE_DIRS})
-
++if (NOT(BOOST_FOUND))
++ find_package(Boost REQUIRED)
+endif()
+
- add_subdirectory(support)
++# add_subdirectory(support/src)
+add_subdirectory(src)
+add_subdirectory(sim)
+
+enable_testing()
+add_subdirectory(test)
++add_subdirectory(support/test)
++add_test(NAME dmclock-tests
++ COMMAND $<TARGET_FILE:dmclock-tests>)
++add_test(NAME dmclock-data-struct-tests
++ COMMAND $<TARGET_FILE:dmclock-data-struct-tests>)
--- /dev/null
++if(K_WAY_HEAP)
++ if(K_WAY_HEAP LESS 2)
++ message(FATAL_ERROR "K_WAY_HEAP value should be at least 2")
++ else()
++ set(CMAKE_CXX_SIM_FLAGS "-DK_WAY_HEAP=${K_WAY_HEAP}")
++ endif()
++endif()
++
+add_subdirectory(src)
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+/*
+ * Copyright (C) 2016 Red Hat Inc.
+ */
+
+
+#pragma once
+
+
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <chrono>
+#include <deque>
+
+#include "sim_recs.h"
+
+
+namespace crimson {
+ namespace qos_simulation {
+
+ template<typename Q, typename ReqPm, typename RespPm, typename Accum>
+ class SimulatedServer {
+
+ struct QueueItem {
+ ClientId client;
+ std::unique_ptr<TestRequest> request;
+ RespPm additional;
+
+ QueueItem(const ClientId& _client,
+ std::unique_ptr<TestRequest>&& _request,
+ const RespPm& _additional) :
+ client(_client),
+ request(std::move(_request)),
+ additional(_additional)
+ {
+ // empty
+ }
+ }; // QueueItem
+
+ public:
+
+ struct InternalStats {
+ std::mutex mtx;
+ std::chrono::nanoseconds add_request_time;
+ std::chrono::nanoseconds request_complete_time;
+ uint32_t add_request_count;
+ uint32_t request_complete_count;
+
+ InternalStats() :
+ add_request_time(0),
+ request_complete_time(0),
+ add_request_count(0),
+ request_complete_count(0)
+ {
+ // empty
+ }
+ };
+
+ using ClientRespFunc = std::function<void(ClientId,
+ const TestResponse&,
+ const ServerId&,
+ const RespPm&)>;
+
+ using ServerAccumFunc = std::function<void(Accum& accumulator,
+ const RespPm& additional)>;
+
+ protected:
+
+ const ServerId id;
+ Q* priority_queue;
+ ClientRespFunc client_resp_f;
+ int iops;
+ size_t thread_pool_size;
+
+ bool finishing;
+ std::chrono::microseconds op_time;
+
+ std::mutex inner_queue_mtx;
+ std::condition_variable inner_queue_cv;
+ std::deque<QueueItem> inner_queue;
+
+ std::thread* threads;
+
+ using InnerQGuard = std::lock_guard<decltype(inner_queue_mtx)>;
+ using Lock = std::unique_lock<std::mutex>;
+
+ // data collection
+
+ ServerAccumFunc accum_f;
+ Accum accumulator;
+
+ InternalStats internal_stats;
+
+ public:
+
+ using CanHandleRequestFunc = std::function<bool(void)>;
+ using HandleRequestFunc =
+ std::function<void(const ClientId&,std::unique_ptr<TestRequest>,const RespPm&)>;
+ using CreateQueueF = std::function<Q*(CanHandleRequestFunc,HandleRequestFunc)>;
+
+
+ SimulatedServer(ServerId _id,
+ int _iops,
+ size_t _thread_pool_size,
+ const ClientRespFunc& _client_resp_f,
+ const ServerAccumFunc& _accum_f,
+ CreateQueueF _create_queue_f) :
+ id(_id),
+ priority_queue(_create_queue_f(std::bind(&SimulatedServer::has_avail_thread,
+ this),
+ std::bind(&SimulatedServer::inner_post,
+ this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ std::placeholders::_3))),
+ client_resp_f(_client_resp_f),
+ iops(_iops),
+ thread_pool_size(_thread_pool_size),
+ finishing(false),
+ accum_f(_accum_f)
+ {
+ op_time =
+ std::chrono::microseconds((int) (0.5 +
+ thread_pool_size * 1000000.0 / iops));
+ std::chrono::milliseconds delay(1000);
+ threads = new std::thread[thread_pool_size];
+ for (size_t i = 0; i < thread_pool_size; ++i) {
+ threads[i] = std::thread(&SimulatedServer::run, this, delay);
+ }
+ }
+
+ virtual ~SimulatedServer() {
+ Lock l(inner_queue_mtx);
+ finishing = true;
+ inner_queue_cv.notify_all();
+ l.unlock();
+
+ for (size_t i = 0; i < thread_pool_size; ++i) {
+ threads[i].join();
+ }
+
+ delete[] threads;
++
++ delete priority_queue;
+ }
+
+ void post(const TestRequest& request,
+ const ClientId& client_id,
+ const ReqPm& req_params)
+ {
+ time_stats(internal_stats.mtx,
+ internal_stats.add_request_time,
+ [&](){
+ priority_queue->add_request(request, client_id, req_params);
+ });
+ count_stats(internal_stats.mtx,
+ internal_stats.add_request_count);
+ }
+
+ bool has_avail_thread() {
+ InnerQGuard g(inner_queue_mtx);
+ return inner_queue.size() <= thread_pool_size;
+ }
+
+ const Accum& get_accumulator() const { return accumulator; }
+ const Q& get_priority_queue() const { return *priority_queue; }
+ const InternalStats& get_internal_stats() const { return internal_stats; }
+
+ protected:
+
+ void inner_post(const ClientId& client,
+ std::unique_ptr<TestRequest> request,
+ const RespPm& additional) {
+ Lock l(inner_queue_mtx);
+ assert(!finishing);
+ accum_f(accumulator, additional);
+ inner_queue.emplace_back(QueueItem(client,
+ std::move(request),
+ additional));
+ inner_queue_cv.notify_one();
+ }
+
+ void run(std::chrono::milliseconds check_period) {
+ Lock l(inner_queue_mtx);
+ while(true) {
+ while(inner_queue.empty() && !finishing) {
+ inner_queue_cv.wait_for(l, check_period);
+ }
+ if (!inner_queue.empty()) {
+ auto& front = inner_queue.front();
+ auto client = front.client;
+ auto req = std::move(front.request);
+ auto additional = front.additional;
+ inner_queue.pop_front();
+
+ l.unlock();
+
+ // simulation operation by sleeping; then call function to
+ // notify server of completion
+ std::this_thread::sleep_for(op_time);
+
+ TestResponse resp(req->epoch);
+ // TODO: rather than assuming this constructor exists, perhaps
+ // pass in a function that does this mapping?
+ client_resp_f(client, resp, id, additional);
+
+ time_stats(internal_stats.mtx,
+ internal_stats.request_complete_time,
+ [&](){
+ priority_queue->request_completed();
+ });
+ count_stats(internal_stats.mtx,
+ internal_stats.request_complete_count);
+
+ l.lock(); // in prep for next iteration of loop
+ } else {
+ break;
+ }
+ }
+ }
+ }; // class SimulatedServer
+
+ }; // namespace qos_simulation
+}; // namespace crimson
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+/*
+ * Copyright (C) 2016 Red Hat Inc.
+ */
+
+
+#pragma once
+
+
+#include <assert.h>
+
+#include <memory>
+#include <chrono>
+#include <map>
+#include <random>
+#include <iostream>
+#include <iomanip>
+#include <string>
+
+
+namespace crimson {
+ namespace qos_simulation {
+
+ template<typename ServerId, typename ClientId, typename TS, typename TC>
+ class Simulation {
+
+ public:
+
+ using TimePoint = std::chrono::time_point<std::chrono::steady_clock>;
+
+ protected:
+
+ using ClientMap = std::map<ClientId,TC*>;
+ using ServerMap = std::map<ServerId,TS*>;
+
+ uint server_count = 0;
+ uint client_count = 0;
+
+ ServerMap servers;
+ ClientMap clients;
+ std::vector<ServerId> server_ids;
+
+ TimePoint early_time;
+ TimePoint servers_created_time;
+ TimePoint clients_created_time;
+ TimePoint clients_finished_time;
+ TimePoint late_time;
+
+ std::default_random_engine prng;
+
+ bool has_run = false;
+
+
+ public:
+
+ double fmt_tp(const TimePoint& t) {
+ auto c = t.time_since_epoch().count();
+ return uint64_t(c / 1000000.0 + 0.5) % 100000 / 1000.0;
+ }
+
+ TimePoint now() {
+ return std::chrono::steady_clock::now();
+ }
+
+ using ClientBasedServerSelectFunc =
+ std::function<const ServerId&(uint64_t, uint16_t)>;
+
+ using ClientFilter = std::function<bool(const ClientId&)>;
+
+ using ServerFilter = std::function<bool(const ServerId&)>;
+
+ using ServerDataOutF =
+ std::function<void(std::ostream& out,
+ Simulation* sim, ServerFilter,
+ int header_w, int data_w, int data_prec)>;
+
+ using ClientDataOutF =
+ std::function<void(std::ostream& out,
+ Simulation* sim, ClientFilter,
+ int header_w, int data_w, int data_prec)>;
+
+ Simulation() :
+ early_time(now()),
+ prng(std::chrono::system_clock::now().time_since_epoch().count())
+ {
+ // empty
+ }
+
++ ~Simulation() {
++ for (auto c : clients) {
++ TC* cp = c.second;
++ delete cp;
++ }
++
++ for (auto s : servers) {
++ delete s.second;
++ }
++ }
++
+ uint get_client_count() const { return client_count; }
+ uint get_server_count() const { return server_count; }
+ TC& get_client(ClientId id) { return *clients[id]; }
+ TS& get_server(ServerId id) { return *servers[id]; }
+ const ServerId& get_server_id(uint index) const {
+ return server_ids[index];
+ }
+
+
+ void add_servers(uint count,
+ std::function<TS*(ServerId)> create_server_f) {
+ uint i = server_count;
+
+ // increment server_count before creating servers since they
+ // will start running immediately and may use the server_count
+ // value; NB: this could still be an issue if servers are
+ // added with multiple add_servers calls; consider using a
+ // separate start function after all servers (and clients?)
+ // have been added
+ server_count += count;
+
+ for (; i < server_count; ++i) {
+ server_ids.push_back(i);
+ servers[i] = create_server_f(i);
+ }
+
+ servers_created_time = now();
+ }
+
+
+ void add_clients(uint count,
+ std::function<TC*(ClientId)> create_client_f) {
+ uint i = client_count;
+
+ // increment client_count before creating clients since they
+ // will start running immediately and may use the client_count
+ // value (e.g., in the server selection function); NB: this could
+ // still be an issue if clients are added with multiple
+ // add_clients calls; consider using a separate start function
+ // after all clients have been added
+ client_count += count;
+
+ for (; i < client_count; ++i) {
+ clients[i] = create_client_f(i);
+ }
+
+ clients_created_time = now();
+ }
+
+
+ void run() {
+ assert(server_count > 0);
+ assert(client_count > 0);
+
+ std::cout << "simulation started" << std::endl;
+
+ // clients are now running; wait for all to finish
+
+ for (auto const &i : clients) {
+ i.second->wait_until_done();
+ }
+
+ late_time = clients_finished_time = now();
+
+ std::cout << "simulation completed in " <<
+ std::chrono::duration_cast<std::chrono::milliseconds>(clients_finished_time - servers_created_time).count() <<
+ " millisecs" << std::endl;
+
+ has_run = true;
+ } // run
+
+
+ void display_stats(std::ostream& out,
+ ServerDataOutF server_out_f, ClientDataOutF client_out_f,
+ ServerFilter server_filter =
+ [] (const ServerId&) { return true; },
+ ClientFilter client_filter =
+ [] (const ClientId&) { return true; },
+ int head_w = 12, int data_w = 7, int data_prec = 2) {
+ assert(has_run);
+
+ // skip first 2 secondsd of data
+ const std::chrono::seconds skip_amount(0);
+ // calculate in groups of 5 seconds
+ const std::chrono::seconds measure_unit(2);
+ // unit to output reports in
+ const std::chrono::seconds report_unit(1);
+
+ // compute and display stats
+
+ TimePoint earliest_start = late_time;
+ TimePoint latest_start = early_time;
+ TimePoint earliest_finish = late_time;
+ TimePoint latest_finish = early_time;
+
+ for (auto const &c : clients) {
+ auto start = c.second->get_op_times().front();
+ auto end = c.second->get_op_times().back();
+
+ if (start < earliest_start) { earliest_start = start; }
+ if (start > latest_start) { latest_start = start; }
+ if (end < earliest_finish) { earliest_finish = end; }
+ if (end > latest_finish) { latest_finish = end; }
+ }
+
+ double ops_factor =
+ std::chrono::duration_cast<std::chrono::duration<double>>(measure_unit) /
+ std::chrono::duration_cast<std::chrono::duration<double>>(report_unit);
+
+ const auto start_edge = clients_created_time + skip_amount;
+
+ std::map<ClientId,std::vector<double>> ops_data;
+
+ for (auto const &c : clients) {
+ auto it = c.second->get_op_times().begin();
+ const auto end = c.second->get_op_times().end();
+ while (it != end && *it < start_edge) { ++it; }
+
+ for (auto time_edge = start_edge + measure_unit;
+ time_edge <= latest_finish + measure_unit;
+ time_edge += measure_unit) {
+ int count = 0;
+ for (; it != end && *it < time_edge; ++count, ++it) { /* empty */ }
+ double ops_per_second = double(count) / ops_factor;
+ ops_data[c.first].push_back(ops_per_second);
+ }
+ }
+
+ out << "==== Client Data ====" << std::endl;
+
+ out << std::setw(head_w) << "client:";
+ for (auto const &c : clients) {
+ if (!client_filter(c.first)) continue;
+ out << " " << std::setw(data_w) << c.first;
+ }
+ out << std::setw(data_w) << "total" << std::endl;
+
+ {
+ bool has_data;
+ size_t i = 0;
+ do {
+ std::string line_header = "t_" + std::to_string(i) + ":";
+ out << std::setw(head_w) << line_header;
+ has_data = false;
+ double total = 0.0;
+ for (auto const &c : clients) {
+ double data = 0.0;
+ if (i < ops_data[c.first].size()) {
+ data = ops_data[c.first][i];
+ has_data = true;
+ }
+ total += data;
+
+ if (!client_filter(c.first)) continue;
+
+ out << " " << std::setw(data_w) << std::setprecision(data_prec) <<
+ std::fixed << data;
+ }
+ out << " " << std::setw(data_w) << std::setprecision(data_prec) <<
+ std::fixed << total << std::endl;
+ ++i;
+ } while(has_data);
+ }
+
+ client_out_f(out, this, client_filter, head_w, data_w, data_prec);
+
+ display_client_internal_stats<std::chrono::nanoseconds>(out,
+ "nanoseconds");
+
+ out << std::endl << "==== Server Data ====" << std::endl;
+
+ out << std::setw(head_w) << "server:";
+ for (auto const &s : servers) {
+ if (!server_filter(s.first)) continue;
+ out << " " << std::setw(data_w) << s.first;
+ }
+ out << " " << std::setw(data_w) << "total" << std::endl;
+
+ server_out_f(out, this, server_filter, head_w, data_w, data_prec);
+
+ display_server_internal_stats<std::chrono::nanoseconds>(out,
+ "nanoseconds");
+
+ // clean up clients then servers
+
+ for (auto i = clients.begin(); i != clients.end(); ++i) {
+ delete i->second;
+ i->second = nullptr;
+ }
+
+ for (auto i = servers.begin(); i != servers.end(); ++i) {
+ delete i->second;
+ i->second = nullptr;
+ }
+ } // display_stats
+
+
+ template<typename T>
+ void display_server_internal_stats(std::ostream& out,
+ std::string time_unit) {
+ T add_request_time(0);
+ T request_complete_time(0);
+ uint32_t add_request_count = 0;
+ uint32_t request_complete_count = 0;
+
+ for (uint i = 0; i < get_server_count(); ++i) {
+ const auto& server = get_server(i);
+ const auto& is = server.get_internal_stats();
+ add_request_time +=
+ std::chrono::duration_cast<T>(is.add_request_time);
+ request_complete_time +=
+ std::chrono::duration_cast<T>(is.request_complete_time);
+ add_request_count += is.add_request_count;
+ request_complete_count += is.request_complete_count;
+ }
+
+ double add_request_time_per_unit =
+ double(add_request_time.count()) / add_request_count ;
+ out << "total time to add requests: " <<
+ std::fixed << add_request_time.count() << " " << time_unit <<
+ ";" << std::endl <<
+ " count: " << add_request_count << ";" << std::endl <<
+ " average: " << add_request_time_per_unit <<
+ " " << time_unit << " per request/response" << std::endl;
+
+ double request_complete_time_unit =
+ double(request_complete_time.count()) / request_complete_count ;
+ out << "total time to note requests complete: " << std::fixed <<
+ request_complete_time.count() << " " << time_unit << ";" <<
+ std::endl <<
+ " count: " << request_complete_count << ";" << std::endl <<
+ " average: " << request_complete_time_unit <<
+ " " << time_unit << " per request/response" << std::endl;
+
+ out << std::endl;
+
+ assert(add_request_count == request_complete_count);
+ out << "server timing for QOS algorithm: " <<
+ add_request_time_per_unit + request_complete_time_unit <<
+ " " << time_unit << " per request/response" << std::endl;
+ }
+
+
+ template<typename T>
+ void display_client_internal_stats(std::ostream& out,
+ std::string time_unit) {
+ T track_resp_time(0);
+ T get_req_params_time(0);
+ uint32_t track_resp_count = 0;
+ uint32_t get_req_params_count = 0;
+
+ for (uint i = 0; i < get_client_count(); ++i) {
+ const auto& client = get_client(i);
+ const auto& is = client.get_internal_stats();
+ track_resp_time +=
+ std::chrono::duration_cast<T>(is.track_resp_time);
+ get_req_params_time +=
+ std::chrono::duration_cast<T>(is.get_req_params_time);
+ track_resp_count += is.track_resp_count;
+ get_req_params_count += is.get_req_params_count;
+ }
+
+ double track_resp_time_unit =
+ double(track_resp_time.count()) / track_resp_count;
+ out << "total time to track responses: " <<
+ std::fixed << track_resp_time.count() << " " << time_unit << ";" <<
+ std::endl <<
+ " count: " << track_resp_count << ";" << std::endl <<
+ " average: " << track_resp_time_unit << " " << time_unit <<
+ " per request/response" << std::endl;
+
+ double get_req_params_time_unit =
+ double(get_req_params_time.count()) / get_req_params_count;
+ out << "total time to get request parameters: " <<
+ std::fixed << get_req_params_time.count() << " " << time_unit <<
+ ";" << std::endl <<
+ " count: " << get_req_params_count << ";" << std::endl <<
+ " average: " << get_req_params_time_unit << " " << time_unit <<
+ " per request/response" << std::endl;
+
+ out << std::endl;
+
+ assert(track_resp_count == get_req_params_count);
+ out << "client timing for QOS algorithm: " <<
+ track_resp_time_unit + get_req_params_time_unit << " " <<
+ time_unit << " per request/response" << std::endl;
+ }
+
+
+ // **** server selection functions ****
+
+
+ const ServerId& server_select_alternate(uint64_t seed,
+ uint16_t client_idx) {
+ uint index = (client_idx + seed) % server_count;
+ return server_ids[index];
+ }
+
+
+ // returns a lambda using the range specified as servers_per (client)
+ ClientBasedServerSelectFunc
+ make_server_select_alt_range(uint16_t servers_per) {
+ return [servers_per,this](uint64_t seed, uint16_t client_idx)
+ -> const ServerId& {
+ double factor = double(server_count) / client_count;
+ uint offset = seed % servers_per;
+ uint index = (uint(0.5 + client_idx * factor) + offset) % server_count;
+ return server_ids[index];
+ };
+ }
+
+
+ // function to choose a server randomly
+ const ServerId& server_select_random(uint64_t seed, uint16_t client_idx) {
+ uint index = prng() % server_count;
+ return server_ids[index];
+ }
+
+
+ // function to choose a server randomly
+ ClientBasedServerSelectFunc
+ make_server_select_ran_range(uint16_t servers_per) {
+ return [servers_per,this](uint64_t seed, uint16_t client_idx)
+ -> const ServerId& {
+ double factor = double(server_count) / client_count;
+ uint offset = prng() % servers_per;
+ uint index = (uint(0.5 + client_idx * factor) + offset) % server_count;
+ return server_ids[index];
+ };
+ }
+
+
+ // function to always choose the first server
+ const ServerId& server_select_0(uint64_t seed, uint16_t client_idx) {
+ return server_ids[0];
+ }
+ }; // class Simulation
+
+ }; // namespace qos_simulation
+}; // namespace crimson
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+/*
+ * Copyright (C) 2016 Red Hat Inc.
+ */
+
+
+#include "test_dmclock.h"
+#include "config.h"
+
+#ifdef PROFILE
+#include "profile.h"
+#endif
+
+
+namespace dmc = crimson::dmclock;
+namespace test = crimson::test_dmc;
+namespace sim = crimson::qos_simulation;
+
+using namespace std::placeholders;
+
+
+namespace crimson {
+ namespace test_dmc {
+ void server_data(std::ostream& out,
+ test::MySim* sim,
+ test::MySim::ServerFilter server_disp_filter,
+ int head_w, int data_w, int data_prec);
+
+ void client_data(std::ostream& out,
+ test::MySim* sim,
+ test::MySim::ClientFilter client_disp_filter,
+ int head_w, int data_w, int data_prec);
+ }
+}
+
+
+int main(int argc, char* argv[]) {
+ std::vector<const char*> args;
+ for (int i = 1; i < argc; ++i) {
+ args.push_back(argv[i]);
+ }
+
+ std::string conf_file_list;
+ sim::ceph_argparse_early_args(args, &conf_file_list);
+
+ sim::sim_config_t g_conf;
+ std::vector<sim::cli_group_t> &cli_group = g_conf.cli_group;
+ std::vector<sim::srv_group_t> &srv_group = g_conf.srv_group;
+
+ if (!conf_file_list.empty()) {
+ int ret;
+ ret = sim::parse_config_file(conf_file_list, g_conf);
+ if (ret) {
+ // error
+ _exit(1);
+ }
+ } else {
+ // default simulation parameter
+ g_conf.client_groups = 2;
+
+ sim::srv_group_t st;
+ srv_group.push_back(st);
+
+ sim::cli_group_t ct1(99, 0);
+ cli_group.push_back(ct1);
+
+ sim::cli_group_t ct2(1, 10);
+ cli_group.push_back(ct2);
+ }
+
+ const uint server_groups = g_conf.server_groups;
+ const uint client_groups = g_conf.client_groups;
+ const bool server_random_selection = g_conf.server_random_selection;
+ const bool server_soft_limit = g_conf.server_soft_limit;
+ uint server_total_count = 0;
+ uint client_total_count = 0;
+
+ for (uint i = 0; i < client_groups; ++i) {
+ client_total_count += cli_group[i].client_count;
+ }
+
+ for (uint i = 0; i < server_groups; ++i) {
+ server_total_count += srv_group[i].server_count;
+ }
+
+ std::vector<test::dmc::ClientInfo> client_info;
+ for (uint i = 0; i < client_groups; ++i) {
+ client_info.push_back(test::dmc::ClientInfo
+ { cli_group[i].client_reservation,
+ cli_group[i].client_weight,
+ cli_group[i].client_limit } );
+ }
+
+ auto ret_client_group_f = [&](const ClientId& c) -> uint {
+ uint group_max = 0;
+ uint i = 0;
+ for (; i < client_groups; ++i) {
+ group_max += cli_group[i].client_count;
+ if (c < group_max) {
+ break;
+ }
+ }
+ return i;
+ };
+
+ auto ret_server_group_f = [&](const ServerId& s) -> uint {
+ uint group_max = 0;
+ uint i = 0;
+ for (; i < server_groups; ++i) {
+ group_max += srv_group[i].server_count;
+ if (s < group_max) {
+ break;
+ }
+ }
+ return i;
+ };
+
+ auto client_info_f = [=](const ClientId& c) -> test::dmc::ClientInfo {
+ return client_info[ret_client_group_f(c)];
+ };
+
+ auto client_disp_filter = [=] (const ClientId& i) -> bool {
+ return i < 3 || i >= (client_total_count - 3);
+ };
+
+ auto server_disp_filter = [=] (const ServerId& i) -> bool {
+ return i < 3 || i >= (server_total_count - 3);
+ };
+
+
+ test::MySim *simulation;
+
+
+ // lambda to post a request to the identified server; called by client
+ test::SubmitFunc server_post_f =
+ [&simulation](const ServerId& server,
+ const sim::TestRequest& request,
+ const ClientId& client_id,
+ const test::dmc::ReqParams& req_params) {
+ test::DmcServer& s = simulation->get_server(server);
+ s.post(request, client_id, req_params);
+ };
+
+ std::vector<std::vector<sim::CliInst>> cli_inst;
+ for (uint i = 0; i < client_groups; ++i) {
+ if (cli_group[i].client_wait == std::chrono::seconds(0)) {
+ cli_inst.push_back(
+ { { sim::req_op,
+ (uint32_t)cli_group[i].client_total_ops,
+ (double)cli_group[i].client_iops_goal,
+ (uint16_t)cli_group[i].client_outstanding_ops } } );
+ } else {
+ cli_inst.push_back(
+ { { sim::wait_op, cli_group[i].client_wait },
+ { sim::req_op,
+ (uint32_t)cli_group[i].client_total_ops,
+ (double)cli_group[i].client_iops_goal,
+ (uint16_t)cli_group[i].client_outstanding_ops } } );
+ }
+ }
+
+ simulation = new test::MySim();
+
+ test::DmcServer::ClientRespFunc client_response_f =
+ [&simulation](ClientId client_id,
+ const sim::TestResponse& resp,
+ const ServerId& server_id,
+ const dmc::PhaseType& phase) {
+ simulation->get_client(client_id).receive_response(resp,
+ server_id,
+ phase);
+ };
+
+ test::CreateQueueF create_queue_f =
+ [&](test::DmcQueue::CanHandleRequestFunc can_f,
+ test::DmcQueue::HandleRequestFunc handle_f) -> test::DmcQueue* {
+ return new test::DmcQueue(client_info_f, can_f, handle_f, server_soft_limit);
+ };
+
+
+ auto create_server_f = [&](ServerId id) -> test::DmcServer* {
+ uint i = ret_server_group_f(id);
+ return new test::DmcServer(id,
+ srv_group[i].server_iops,
+ srv_group[i].server_threads,
+ client_response_f,
+ test::dmc_server_accumulate_f,
+ create_queue_f);
+ };
+
+ auto create_client_f = [&](ClientId id) -> test::DmcClient* {
+ uint i = ret_client_group_f(id);
+ test::MySim::ClientBasedServerSelectFunc server_select_f;
+ uint client_server_select_range = cli_group[i].client_server_select_range;
+ if (!server_random_selection) {
+ server_select_f = simulation->make_server_select_alt_range(client_server_select_range);
+ } else {
+ server_select_f = simulation->make_server_select_ran_range(client_server_select_range);
+ }
+ return new test::DmcClient(id,
+ server_post_f,
+ std::bind(server_select_f, _1, id),
+ test::dmc_client_accumulate_f,
+ cli_inst[i]);
+ };
+
+#if 1
+ std::cout << "[global]" << std::endl << g_conf << std::endl;
+ for (uint i = 0; i < client_groups; ++i) {
+ std::cout << std::endl << "[client." << i << "]" << std::endl;
+ std::cout << cli_group[i] << std::endl;
+ }
+ for (uint i = 0; i < server_groups; ++i) {
+ std::cout << std::endl << "[server." << i << "]" << std::endl;
+ std::cout << srv_group[i] << std::endl;
+ }
+ std::cout << std::endl;
+#endif
+
+ simulation->add_servers(server_total_count, create_server_f);
+ simulation->add_clients(client_total_count, create_client_f);
+
+ simulation->run();
+ simulation->display_stats(std::cout,
+ &test::server_data, &test::client_data,
+ server_disp_filter, client_disp_filter);
++
++ delete simulation;
+} // main
+
+
+void test::client_data(std::ostream& out,
+ test::MySim* sim,
+ test::MySim::ClientFilter client_disp_filter,
+ int head_w, int data_w, int data_prec) {
+ // report how many ops were done by reservation and proportion for
+ // each client
+
+ int total_r = 0;
+ out << std::setw(head_w) << "res_ops:";
+ for (uint i = 0; i < sim->get_client_count(); ++i) {
+ const auto& client = sim->get_client(i);
+ auto r = client.get_accumulator().reservation_count;
+ total_r += r;
+ if (!client_disp_filter(i)) continue;
+ out << " " << std::setw(data_w) << r;
+ }
+ out << " " << std::setw(data_w) << std::setprecision(data_prec) <<
+ std::fixed << total_r << std::endl;
+
+ int total_p = 0;
+ out << std::setw(head_w) << "prop_ops:";
+ for (uint i = 0; i < sim->get_client_count(); ++i) {
+ const auto& client = sim->get_client(i);
+ auto p = client.get_accumulator().proportion_count;
+ total_p += p;
+ if (!client_disp_filter(i)) continue;
+ out << " " << std::setw(data_w) << p;
+ }
+ out << " " << std::setw(data_w) << std::setprecision(data_prec) <<
+ std::fixed << total_p << std::endl;
+}
+
+
+void test::server_data(std::ostream& out,
+ test::MySim* sim,
+ test::MySim::ServerFilter server_disp_filter,
+ int head_w, int data_w, int data_prec) {
+ out << std::setw(head_w) << "res_ops:";
+ int total_r = 0;
+ for (uint i = 0; i < sim->get_server_count(); ++i) {
+ const auto& server = sim->get_server(i);
+ auto rc = server.get_accumulator().reservation_count;
+ total_r += rc;
+ if (!server_disp_filter(i)) continue;
+ out << " " << std::setw(data_w) << rc;
+ }
+ out << " " << std::setw(data_w) << std::setprecision(data_prec) <<
+ std::fixed << total_r << std::endl;
+
+ out << std::setw(head_w) << "prop_ops:";
+ int total_p = 0;
+ for (uint i = 0; i < sim->get_server_count(); ++i) {
+ const auto& server = sim->get_server(i);
+ auto pc = server.get_accumulator().proportion_count;
+ total_p += pc;
+ if (!server_disp_filter(i)) continue;
+ out << " " << std::setw(data_w) << pc;
+ }
+ out << " " << std::setw(data_w) << std::setprecision(data_prec) <<
+ std::fixed << total_p << std::endl;
+
+ const auto& q = sim->get_server(0).get_priority_queue();
+ out << std::endl <<
+ " k-way heap: " << q.get_heap_branching_factor() << std::endl
+ << std::endl;
+
+#ifdef PROFILE
+ crimson::ProfileCombiner<std::chrono::nanoseconds> art_combiner;
+ crimson::ProfileCombiner<std::chrono::nanoseconds> rct_combiner;
+ for (uint i = 0; i < sim->get_server_count(); ++i) {
+ const auto& q = sim->get_server(i).get_priority_queue();
+ const auto& art = q.add_request_timer;
+ art_combiner.combine(art);
+ const auto& rct = q.request_complete_timer;
+ rct_combiner.combine(rct);
+ }
+ out << "Server add_request_timer: count:" << art_combiner.get_count() <<
+ ", mean:" << art_combiner.get_mean() <<
+ ", std_dev:" << art_combiner.get_std_dev() <<
+ ", low:" << art_combiner.get_low() <<
+ ", high:" << art_combiner.get_high() << std::endl;
+ out << "Server request_complete_timer: count:" << rct_combiner.get_count() <<
+ ", mean:" << rct_combiner.get_mean() <<
+ ", std_dev:" << rct_combiner.get_std_dev() <<
+ ", low:" << rct_combiner.get_low() <<
+ ", high:" << rct_combiner.get_high() << std::endl;
+ out << "Server combined mean: " <<
+ (art_combiner.get_mean() + rct_combiner.get_mean()) <<
+ std::endl;
+#endif
+}
--- /dev/null
- include_directories(${BOOST_INCLUDE_DIR})
++include_directories(${GTEST_INCLUDE_DIRS})
++include_directories(${Boost_INCLUDE_DIRS})
+include_directories(../support/src)
- set(local_flags "-Wall -pthread")
+
++set(local_flags "-std=c++11 -Wno-write-strings -Wall -pthread")
++
++if(DO_NOT_DELAY_TAG_CALC)
++ set(local_flags "${local_flags} -DDO_NOT_DELAY_TAG_CALC")
++endif()
+
+set(dmc_srcs dmclock_util.cc ../support/src/run_every.cc)
+
+set_source_files_properties(${dmc_srcs}
+ PROPERTIES
+ COMPILE_FLAGS "${local_flags}"
+ )
+
+if ("${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang")
+ set(warnings_off " -Wno-unused-variable -Wno-unused-function")
+elseif ("${CMAKE_CXX_COMPILER_ID}" STREQUAL "GNU")
+ set(warnings_off " -Wno-unused-but-set-variable -Wno-unused-function")
+endif()
+
+add_library(dmclock STATIC ${dmc_srcs})
--- /dev/null
- const ReqParams& req_params,
- const Time& time,
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+/*
+ * Copyright (C) 2017 Red Hat Inc.
+ */
+
+
+#pragma once
+
+/* COMPILATION OPTIONS
+ *
+ * By default we include an optimization over the originally published
+ * dmclock algorithm using not the values of rho and delta that were
+ * sent in with a request but instead the most recent rho and delta
+ * values from the requests's client. To restore the algorithm's
+ * original behavior, define DO_NOT_DELAY_TAG_CALC (i.e., compiler
+ * argument -DDO_NOT_DELAY_TAG_CALC).
+ *
+ * The prop_heap does not seem to be necessary. The only thing it
+ * would help with is quickly finding the mininum proportion/prioity
+ * when an idle client became active. To have the code maintain the
+ * proportional heap, define USE_PROP_HEAP (i.e., compiler argument
+ * -DUSE_PROP_HEAP).
+ */
+
+#include <assert.h>
+
+#include <cmath>
+#include <memory>
+#include <map>
+#include <deque>
+#include <queue>
+#include <atomic>
+#include <mutex>
+#include <condition_variable>
+#include <thread>
+#include <iostream>
+#include <sstream>
+#include <limits>
+
+#include <boost/variant.hpp>
+
+#include "indirect_intrusive_heap.h"
+#include "run_every.h"
+#include "dmclock_util.h"
+#include "dmclock_recs.h"
+
+#ifdef PROFILE
+#include "profile.h"
+#endif
+
+#include "gtest/gtest_prod.h"
+
+
+namespace crimson {
+
+ namespace dmclock {
+
+ namespace c = crimson;
+
+ constexpr double max_tag = std::numeric_limits<double>::is_iec559 ?
+ std::numeric_limits<double>::infinity() :
+ std::numeric_limits<double>::max();
+ constexpr double min_tag = std::numeric_limits<double>::is_iec559 ?
+ -std::numeric_limits<double>::infinity() :
+ std::numeric_limits<double>::lowest();
+ constexpr uint tag_modulo = 1000000;
+
+ struct ClientInfo {
+ const double reservation; // minimum
+ const double weight; // proportional
+ const double limit; // maximum
+
+ // multiplicative inverses of above, which we use in calculations
+ // and don't want to recalculate repeatedly
+ const double reservation_inv;
+ const double weight_inv;
+ const double limit_inv;
+
+ // order parameters -- min, "normal", max
+ ClientInfo(double _reservation, double _weight, double _limit) :
+ reservation(_reservation),
+ weight(_weight),
+ limit(_limit),
+ reservation_inv(0.0 == reservation ? 0.0 : 1.0 / reservation),
+ weight_inv( 0.0 == weight ? 0.0 : 1.0 / weight),
+ limit_inv( 0.0 == limit ? 0.0 : 1.0 / limit)
+ {
+ // empty
+ }
+
+
+ friend std::ostream& operator<<(std::ostream& out,
+ const ClientInfo& client) {
+ out <<
+ "{ ClientInfo:: r:" << client.reservation <<
+ " w:" << std::fixed << client.weight <<
+ " l:" << std::fixed << client.limit <<
+ " 1/r:" << std::fixed << client.reservation_inv <<
+ " 1/w:" << std::fixed << client.weight_inv <<
+ " 1/l:" << std::fixed << client.limit_inv <<
+ " }";
+ return out;
+ }
+ }; // class ClientInfo
+
+
+ struct RequestTag {
+ double reservation;
+ double proportion;
+ double limit;
+ bool ready; // true when within limit
+#ifndef DO_NOT_DELAY_TAG_CALC
+ Time arrival;
+#endif
+
+ RequestTag(const RequestTag& prev_tag,
+ const ClientInfo& client,
- req_params.rho,
++ const uint32_t delta,
++ const uint32_t rho,
++ const Time time,
+ const double cost = 0.0) :
+ reservation(cost + tag_calc(time,
+ prev_tag.reservation,
+ client.reservation_inv,
- req_params.delta,
++ rho,
+ true)),
+ proportion(tag_calc(time,
+ prev_tag.proportion,
+ client.weight_inv,
- req_params.delta,
++ delta,
+ true)),
+ limit(tag_calc(time,
+ prev_tag.limit,
+ client.limit_inv,
- RequestTag(double _res, double _prop, double _lim, const Time& _arrival) :
++ delta,
+ false)),
+ ready(false)
+#ifndef DO_NOT_DELAY_TAG_CALC
+ , arrival(time)
+#endif
+ {
+ assert(reservation < max_tag || proportion < max_tag);
+ }
+
- static double tag_calc(const Time& time,
++ RequestTag(const RequestTag& prev_tag,
++ const ClientInfo& client,
++ const ReqParams req_params,
++ const Time time,
++ const double cost = 0.0) :
++ RequestTag(prev_tag, client, req_params.delta, req_params.rho, time, cost)
++ { /* empty */ }
++
++ RequestTag(double _res, double _prop, double _lim, const Time _arrival) :
+ reservation(_res),
+ proportion(_prop),
+ limit(_lim),
+ ready(false)
+#ifndef DO_NOT_DELAY_TAG_CALC
+ , arrival(_arrival)
+#endif
+ {
+ assert(reservation < max_tag || proportion < max_tag);
+ }
+
+ RequestTag(const RequestTag& other) :
+ reservation(other.reservation),
+ proportion(other.proportion),
+ limit(other.limit),
+ ready(other.ready)
+#ifndef DO_NOT_DELAY_TAG_CALC
+ , arrival(other.arrival)
+#endif
+ {
+ // empty
+ }
+
+ static std::string format_tag_change(double before, double after) {
+ if (before == after) {
+ return std::string("same");
+ } else {
+ std::stringstream ss;
+ ss << format_tag(before) << "=>" << format_tag(after);
+ return ss.str();
+ }
+ }
+
+ static std::string format_tag(double value) {
+ if (max_tag == value) {
+ return std::string("max");
+ } else if (min_tag == value) {
+ return std::string("min");
+ } else {
+ return format_time(value, tag_modulo);
+ }
+ }
+
+ private:
+
- void do_add_request(RequestRef&& request,
- const C& client_id,
++ static double tag_calc(const Time time,
+ double prev,
+ double increment,
+ uint32_t dist_req_val,
+ bool extreme_is_high) {
+ if (0.0 == increment) {
+ return extreme_is_high ? max_tag : min_tag;
+ } else {
+ if (0 != dist_req_val) {
+ increment *= dist_req_val;
+ }
+ return std::max(time, prev + increment);
+ }
+ }
+
+ friend std::ostream& operator<<(std::ostream& out,
+ const RequestTag& tag) {
+ out <<
+ "{ RequestTag:: ready:" << (tag.ready ? "true" : "false") <<
+ " r:" << format_tag(tag.reservation) <<
+ " p:" << format_tag(tag.proportion) <<
+ " l:" << format_tag(tag.limit) <<
+#if 0 // try to resolve this to make sure Time is operator<<'able.
+#ifndef DO_NOT_DELAY_TAG_CALC
+ " arrival:" << tag.arrival <<
+#endif
+#endif
+ " }";
+ return out;
+ }
+ }; // class RequestTag
+
+
+ // C is client identifier type, R is request type, B is heap
+ // branching factor
+ template<typename C, typename R, uint B>
+ class PriorityQueueBase {
+ FRIEND_TEST(dmclock_server, client_idle_erase);
+
+ public:
+
+ using RequestRef = std::unique_ptr<R>;
+
+ protected:
+
+ using TimePoint = decltype(std::chrono::steady_clock::now());
+ using Duration = std::chrono::milliseconds;
+ using MarkPoint = std::pair<TimePoint,Counter>;
+
+ enum class ReadyOption {ignore, lowers, raises};
+
+ // forward decl for friend decls
+ template<double RequestTag::*, ReadyOption, bool>
+ struct ClientCompare;
+
+ class ClientReq {
+ friend PriorityQueueBase;
+
+ RequestTag tag;
+ C client_id;
+ RequestRef request;
+
+ public:
+
+ ClientReq(const RequestTag& _tag,
+ const C& _client_id,
+ RequestRef&& _request) :
+ tag(_tag),
+ client_id(_client_id),
+ request(std::move(_request))
+ {
+ // empty
+ }
+
+ friend std::ostream& operator<<(std::ostream& out, const ClientReq& c) {
+ out << "{ ClientReq:: tag:" << c.tag << " client:" <<
+ c.client_id << " }";
+ return out;
+ }
+ }; // class ClientReq
+
+ public:
+
+ // NOTE: ClientRec is in the "public" section for compatibility
+ // with g++ 4.8.4, which complains if it's not. By g++ 6.3.1
+ // ClientRec could be "protected" with no issue. [See comments
+ // associated with function submit_top_request.]
+ class ClientRec {
+ friend PriorityQueueBase<C,R,B>;
+
+ C client;
+ RequestTag prev_tag;
+ std::deque<ClientReq> requests;
+
+ // amount added from the proportion tag as a result of
+ // an idle client becoming unidle
+ double prop_delta = 0.0;
+
+ c::IndIntruHeapData reserv_heap_data;
+ c::IndIntruHeapData lim_heap_data;
+ c::IndIntruHeapData ready_heap_data;
+#if USE_PROP_HEAP
+ c::IndIntruHeapData prop_heap_data;
+#endif
+
+ public:
+
+ ClientInfo info;
+ bool idle;
+ Counter last_tick;
+ uint32_t cur_rho;
+ uint32_t cur_delta;
+
+ ClientRec(C _client,
+ const ClientInfo& _info,
+ Counter current_tick) :
+ client(_client),
+ prev_tag(0.0, 0.0, 0.0, TimeZero),
+ info(_info),
+ idle(true),
+ last_tick(current_tick),
+ cur_rho(1),
+ cur_delta(1)
+ {
+ // empty
+ }
+
+ inline const RequestTag& get_req_tag() const {
+ return prev_tag;
+ }
+
+ static inline void assign_unpinned_tag(double& lhs, const double rhs) {
+ if (rhs != max_tag && rhs != min_tag) {
+ lhs = rhs;
+ }
+ }
+
+ inline void update_req_tag(const RequestTag& _prev,
+ const Counter& _tick) {
+ assign_unpinned_tag(prev_tag.reservation, _prev.reservation);
+ assign_unpinned_tag(prev_tag.limit, _prev.limit);
+ assign_unpinned_tag(prev_tag.proportion, _prev.proportion);
+ last_tick = _tick;
+ }
+
+ inline void add_request(const RequestTag& tag,
+ const C& client_id,
+ RequestRef&& request) {
+ requests.emplace_back(ClientReq(tag, client_id, std::move(request)));
+ }
+
+ inline const ClientReq& next_request() const {
+ return requests.front();
+ }
+
+ inline ClientReq& next_request() {
+ return requests.front();
+ }
+
+ inline void pop_request() {
+ requests.pop_front();
+ }
+
+ inline bool has_request() const {
+ return !requests.empty();
+ }
+
+ inline size_t request_count() const {
+ return requests.size();
+ }
+
+ // NB: because a deque is the underlying structure, this
+ // operation might be expensive
+ bool remove_by_req_filter_fw(std::function<bool(const R&)> filter_accum) {
+ bool any_removed = false;
+ for (auto i = requests.begin();
+ i != requests.end();
+ /* no inc */) {
+ if (filter_accum(*i->request)) {
+ any_removed = true;
+ i = requests.erase(i);
+ } else {
+ ++i;
+ }
+ }
+ return any_removed;
+ }
+
+ // NB: because a deque is the underlying structure, this
+ // operation might be expensive
+ bool remove_by_req_filter_bw(std::function<bool(const R&)> filter_accum) {
+ bool any_removed = false;
+ for (auto i = requests.rbegin();
+ i != requests.rend();
+ /* no inc */) {
+ if (filter_accum(*i->request)) {
+ any_removed = true;
+ i = decltype(i){ requests.erase(std::next(i).base()) };
+ } else {
+ ++i;
+ }
+ }
+ return any_removed;
+ }
+
+ inline bool
+ remove_by_req_filter(std::function<bool(const R&)> filter_accum,
+ bool visit_backwards) {
+ if (visit_backwards) {
+ return remove_by_req_filter_bw(filter_accum);
+ } else {
+ return remove_by_req_filter_fw(filter_accum);
+ }
+ }
+
+ friend std::ostream&
+ operator<<(std::ostream& out,
+ const typename PriorityQueueBase<C,R,B>::ClientRec& e) {
+ out << "{ ClientRec::" <<
+ " client:" << e.client <<
+ " prev_tag:" << e.prev_tag <<
+ " req_count:" << e.requests.size() <<
+ " top_req:";
+ if (e.has_request()) {
+ out << e.next_request();
+ } else {
+ out << "none";
+ }
+ out << " }";
+
+ return out;
+ }
+ }; // class ClientRec
+
+ using ClientRecRef = std::shared_ptr<ClientRec>;
+
+ // when we try to get the next request, we'll be in one of three
+ // situations -- we'll have one to return, have one that can
+ // fire in the future, or not have any
+ enum class NextReqType { returning, future, none };
+
+ // specifies which queue next request will get popped from
+ enum class HeapId { reservation, ready };
+
+ // this is returned from next_req to tell the caller the situation
+ struct NextReq {
+ NextReqType type;
+ union {
+ HeapId heap_id;
+ Time when_ready;
+ };
+ };
+
+
+ // a function that can be called to look up client information
+ using ClientInfoFunc = std::function<ClientInfo(const C&)>;
+
+
+ bool empty() const {
+ DataGuard g(data_mtx);
+ return (resv_heap.empty() || ! resv_heap.top().has_request());
+ }
+
+
+ size_t client_count() const {
+ DataGuard g(data_mtx);
+ return resv_heap.size();
+ }
+
+
+ size_t request_count() const {
+ DataGuard g(data_mtx);
+ size_t total = 0;
+ for (auto i = resv_heap.cbegin(); i != resv_heap.cend(); ++i) {
+ total += i->request_count();
+ }
+ return total;
+ }
+
+
+ bool remove_by_req_filter(std::function<bool(const R&)> filter_accum,
+ bool visit_backwards = false) {
+ bool any_removed = false;
+ DataGuard g(data_mtx);
+ for (auto i : client_map) {
+ bool modified =
+ i.second->remove_by_req_filter(filter_accum, visit_backwards);
+ if (modified) {
+ resv_heap.adjust(*i.second);
+ limit_heap.adjust(*i.second);
+ ready_heap.adjust(*i.second);
+#if USE_PROP_HEAP
+ prop_heap.adjust(*i.second);
+#endif
+ any_removed = true;
+ }
+ }
+ return any_removed;
+ }
+
+
+ // use as a default value when no accumulator is provide
+ static void request_sink(const R& req) {
+ // do nothing
+ }
+
+
+ void remove_by_client(const C& client,
+ bool reverse = false,
+ std::function<void (const R&)> accum = request_sink) {
+ DataGuard g(data_mtx);
+
+ auto i = client_map.find(client);
+
+ if (i == client_map.end()) return;
+
+ if (reverse) {
+ for (auto j = i->second->requests.rbegin();
+ j != i->second->requests.rend();
+ ++j) {
+ accum(*j->request);
+ }
+ } else {
+ for (auto j = i->second->requests.begin();
+ j != i->second->requests.end();
+ ++j) {
+ accum(*j->request);
+ }
+ }
+
+ i->second->requests.clear();
+
+ resv_heap.adjust(*i->second);
+ limit_heap.adjust(*i->second);
+ ready_heap.adjust(*i->second);
+#if USE_PROP_HEAP
+ prop_heap.adjust(*i->second);
+#endif
+ }
+
+
+ uint get_heap_branching_factor() const {
+ return B;
+ }
+
+
+ friend std::ostream& operator<<(std::ostream& out,
+ const PriorityQueueBase& q) {
+ std::lock_guard<decltype(q.data_mtx)> guard(q.data_mtx);
+
+ out << "{ PriorityQueue::";
+ for (const auto& c : q.client_map) {
+ out << " { client:" << c.first << ", record:" << *c.second <<
+ " }";
+ }
+ if (!q.resv_heap.empty()) {
+ const auto& resv = q.resv_heap.top();
+ out << " { reservation_top:" << resv << " }";
+ const auto& ready = q.ready_heap.top();
+ out << " { ready_top:" << ready << " }";
+ const auto& limit = q.limit_heap.top();
+ out << " { limit_top:" << limit << " }";
+ } else {
+ out << " HEAPS-EMPTY";
+ }
+ out << " }";
+
+ return out;
+ }
+
+ // for debugging
+ void display_queues(std::ostream& out,
+ bool show_res = true,
+ bool show_lim = true,
+ bool show_ready = true,
+ bool show_prop = true) const {
+ auto filter = [](const ClientRec& e)->bool { return true; };
+ DataGuard g(data_mtx);
+ if (show_res) {
+ resv_heap.display_sorted(out << "RESER:", filter);
+ }
+ if (show_lim) {
+ limit_heap.display_sorted(out << "LIMIT:", filter);
+ }
+ if (show_ready) {
+ ready_heap.display_sorted(out << "READY:", filter);
+ }
+#if USE_PROP_HEAP
+ if (show_prop) {
+ prop_heap.display_sorted(out << "PROPO:", filter);
+ }
+#endif
+ } // display_queues
+
+
+ protected:
+
+ // The ClientCompare functor is essentially doing a precedes?
+ // operator, returning true if and only if the first parameter
+ // must precede the second parameter. If the second must precede
+ // the first, or if they are equivalent, false should be
+ // returned. The reason for this behavior is that it will be
+ // called to test if two items are out of order and if true is
+ // returned it will reverse the items. Therefore false is the
+ // default return when it doesn't matter to prevent unnecessary
+ // re-ordering.
+ //
+ // The template is supporting variations in sorting based on the
+ // heap in question and allowing these variations to be handled
+ // at compile-time.
+ //
+ // tag_field determines which tag is being used for comparison
+ //
+ // ready_opt determines how the ready flag influences the sort
+ //
+ // use_prop_delta determines whether the proportional delta is
+ // added in for comparison
+ template<double RequestTag::*tag_field,
+ ReadyOption ready_opt,
+ bool use_prop_delta>
+ struct ClientCompare {
+ bool operator()(const ClientRec& n1, const ClientRec& n2) const {
+ if (n1.has_request()) {
+ if (n2.has_request()) {
+ const auto& t1 = n1.next_request().tag;
+ const auto& t2 = n2.next_request().tag;
+ if (ReadyOption::ignore == ready_opt || t1.ready == t2.ready) {
+ // if we don't care about ready or the ready values are the same
+ if (use_prop_delta) {
+ return (t1.*tag_field + n1.prop_delta) <
+ (t2.*tag_field + n2.prop_delta);
+ } else {
+ return t1.*tag_field < t2.*tag_field;
+ }
+ } else if (ReadyOption::raises == ready_opt) {
+ // use_ready == true && the ready fields are different
+ return t1.ready;
+ } else {
+ return t2.ready;
+ }
+ } else {
+ // n1 has request but n2 does not
+ return true;
+ }
+ } else if (n2.has_request()) {
+ // n2 has request but n1 does not
+ return false;
+ } else {
+ // both have none; keep stable w false
+ return false;
+ }
+ }
+ };
+
+ ClientInfoFunc client_info_f;
+
+ mutable std::mutex data_mtx;
+ using DataGuard = std::lock_guard<decltype(data_mtx)>;
+
+ // stable mapping between client ids and client queues
+ std::map<C,ClientRecRef> client_map;
+
+ c::IndIntruHeap<ClientRecRef,
+ ClientRec,
+ &ClientRec::reserv_heap_data,
+ ClientCompare<&RequestTag::reservation,
+ ReadyOption::ignore,
+ false>,
+ B> resv_heap;
+#if USE_PROP_HEAP
+ c::IndIntruHeap<ClientRecRef,
+ ClientRec,
+ &ClientRec::prop_heap_data,
+ ClientCompare<&RequestTag::proportion,
+ ReadyOption::ignore,
+ true>,
+ B> prop_heap;
+#endif
+ c::IndIntruHeap<ClientRecRef,
+ ClientRec,
+ &ClientRec::lim_heap_data,
+ ClientCompare<&RequestTag::limit,
+ ReadyOption::lowers,
+ false>,
+ B> limit_heap;
+ c::IndIntruHeap<ClientRecRef,
+ ClientRec,
+ &ClientRec::ready_heap_data,
+ ClientCompare<&RequestTag::proportion,
+ ReadyOption::raises,
+ true>,
+ B> ready_heap;
+
+ // if all reservations are met and all other requestes are under
+ // limit, this will allow the request next in terms of
+ // proportion to still get issued
+ bool allow_limit_break;
+
+ std::atomic_bool finishing;
+
+ // every request creates a tick
+ Counter tick = 0;
+
+ // performance data collection
+ size_t reserv_sched_count = 0;
+ size_t prop_sched_count = 0;
+ size_t limit_break_sched_count = 0;
+
+ Duration idle_age;
+ Duration erase_age;
+ Duration check_time;
+ std::deque<MarkPoint> clean_mark_points;
+
+ // NB: All threads declared at end, so they're destructed first!
+
+ std::unique_ptr<RunEvery> cleaning_job;
+
+
+ // COMMON constructor that others feed into; we can accept three
+ // different variations of durations
+ template<typename Rep, typename Per>
+ PriorityQueueBase(ClientInfoFunc _client_info_f,
+ std::chrono::duration<Rep,Per> _idle_age,
+ std::chrono::duration<Rep,Per> _erase_age,
+ std::chrono::duration<Rep,Per> _check_time,
+ bool _allow_limit_break) :
+ client_info_f(_client_info_f),
+ allow_limit_break(_allow_limit_break),
+ finishing(false),
+ idle_age(std::chrono::duration_cast<Duration>(_idle_age)),
+ erase_age(std::chrono::duration_cast<Duration>(_erase_age)),
+ check_time(std::chrono::duration_cast<Duration>(_check_time))
+ {
+ assert(_erase_age >= _idle_age);
+ assert(_check_time < _idle_age);
+ cleaning_job =
+ std::unique_ptr<RunEvery>(
+ new RunEvery(check_time,
+ std::bind(&PriorityQueueBase::do_clean, this)));
+ }
+
+
+ ~PriorityQueueBase() {
+ finishing = true;
+ }
+
+
+ // data_mtx must be held by caller
- const Time time,
- const double cost = 0.0) {
++ void do_add_request(RequestRef&& request,
++ const C& client_id,
+ const ReqParams& req_params,
- tag = RequestTag(client.get_req_tag(), client.info,
- req_params, time, cost);
++ const Time time,
++ const double cost = 0.0) {
+ ++tick;
+
+ // this pointer will help us create a reference to a shared
+ // pointer, no matter which of two codepaths we take
+ ClientRec* temp_client;
+
+ auto client_it = client_map.find(client_id);
+ if (client_map.end() != client_it) {
+ temp_client = &(*client_it->second); // address of obj of shared_ptr
+ } else {
+ ClientInfo info = client_info_f(client_id);
+ ClientRecRef client_rec =
+ std::make_shared<ClientRec>(client_id, info, tick);
+ resv_heap.push(client_rec);
+#if USE_PROP_HEAP
+ prop_heap.push(client_rec);
+#endif
+ limit_heap.push(client_rec);
+ ready_heap.push(client_rec);
+ client_map[client_id] = client_rec;
+ temp_client = &(*client_rec); // address of obj of shared_ptr
+ }
+
+ // for convenience, we'll create a reference to the shared pointer
+ ClientRec& client = *temp_client;
+
+ if (client.idle) {
+ // We need to do an adjustment so that idle clients compete
+ // fairly on proportional tags since those tags may have
+ // drifted from real-time. Either use the lowest existing
+ // proportion tag -- O(1) -- or the client with the lowest
+ // previous proportion tag -- O(n) where n = # clients.
+ //
+ // So we don't have to maintain a propotional queue that
+ // keeps the minimum on proportional tag alone (we're
+ // instead using a ready queue), we'll have to check each
+ // client.
+ //
+ // The alternative would be to maintain a proportional queue
+ // (define USE_PROP_TAG) and do an O(1) operation here.
+
+ // Was unable to confirm whether equality testing on
+ // std::numeric_limits<double>::max() is guaranteed, so
+ // we'll use a compile-time calculated trigger that is one
+ // third the max, which should be much larger than any
+ // expected organic value.
+ constexpr double lowest_prop_tag_trigger =
+ std::numeric_limits<double>::max() / 3.0;
+
+ double lowest_prop_tag = std::numeric_limits<double>::max();
+ for (auto const &c : client_map) {
+ // don't use ourselves (or anything else that might be
+ // listed as idle) since we're now in the map
+ if (!c.second->idle) {
+ double p;
+ // use either lowest proportion tag or previous proportion tag
+ if (c.second->has_request()) {
+ p = c.second->next_request().tag.proportion +
+ c.second->prop_delta;
+ } else {
+ p = c.second->get_req_tag().proportion + c.second->prop_delta;
+ }
+
+ if (p < lowest_prop_tag) {
+ lowest_prop_tag = p;
+ }
+ }
+ }
+
+ // if this conditional does not fire, it
+ if (lowest_prop_tag < lowest_prop_tag_trigger) {
+ client.prop_delta = lowest_prop_tag - time;
+ }
+ client.idle = false;
+ } // if this client was idle
+
+#ifndef DO_NOT_DELAY_TAG_CALC
+ RequestTag tag(0, 0, 0, time);
+
+ if (!client.has_request()) {
- ClientReq& first = top.next_request();
- RequestRef request = std::move(first.request);
++ tag = RequestTag(client.get_req_tag(),
++ client.info,
++ req_params,
++ time,
++ cost);
+
+ // copy tag to previous tag for client
+ client.update_req_tag(tag, tick);
+ }
+#else
+ RequestTag tag(client.get_req_tag(), client.info, req_params, time, cost);
+ // copy tag to previous tag for client
+ client.update_req_tag(tag, tick);
+#endif
+
+ client.add_request(tag, client.client, std::move(request));
+ if (1 == client.requests.size()) {
+ // NB: can the following 4 calls to adjust be changed
+ // promote? Can adding a request ever demote a client in the
+ // heaps?
+ resv_heap.adjust(client);
+ limit_heap.adjust(client);
+ ready_heap.adjust(client);
+#if USE_PROP_HEAP
+ prop_heap.adjust(client);
+#endif
+ }
+
+ client.cur_rho = req_params.rho;
+ client.cur_delta = req_params.delta;
+
+ resv_heap.adjust(client);
+ limit_heap.adjust(client);
+ ready_heap.adjust(client);
+#if USE_PROP_HEAP
+ prop_heap.adjust(client);
+#endif
+ } // add_request
+
+
+ // data_mtx should be held when called; top of heap should have
+ // a ready request
+ template<typename C1, IndIntruHeapData ClientRec::*C2, typename C3>
+ void pop_process_request(IndIntruHeap<C1, ClientRec, C2, C3, B>& heap,
+ std::function<void(const C& client,
+ RequestRef& request)> process) {
+ // gain access to data
+ ClientRec& top = heap.top();
- next_first.tag = RequestTag(first.tag, top.info,
- ReqParams(top.cur_delta, top.cur_rho),
++
++ RequestRef request = std::move(top.next_request().request);
++ RequestTag tag = top.next_request().tag;
+
+ // pop request and adjust heaps
+ top.pop_request();
+
+#ifndef DO_NOT_DELAY_TAG_CALC
+ if (top.has_request()) {
+ ClientReq& next_first = top.next_request();
- const C& client_id,
++ next_first.tag = RequestTag(tag, top.info,
++ top.cur_delta, top.cur_rho,
+ next_first.tag.arrival);
+
+ // copy tag to previous tag for client
+ top.update_req_tag(next_first.tag, tick);
+ }
+#endif
+
+ resv_heap.demote(top);
+ limit_heap.adjust(top);
+#if USE_PROP_HEAP
+ prop_heap.demote(top);
+#endif
+ ready_heap.demote(top);
+
+ // process
+ process(top.client, request);
+ } // pop_process_request
+
+
+ // data_mtx should be held when called
+ void reduce_reservation_tags(ClientRec& client) {
+ for (auto& r : client.requests) {
+ r.tag.reservation -= client.info.reservation_inv;
+
+#ifndef DO_NOT_DELAY_TAG_CALC
+ // reduce only for front tag. because next tags' value are invalid
+ break;
+#endif
+ }
+ // don't forget to update previous tag
+ client.prev_tag.reservation -= client.info.reservation_inv;
+ resv_heap.promote(client);
+ }
+
+
+ // data_mtx should be held when called
+ void reduce_reservation_tags(const C& client_id) {
+ auto client_it = client_map.find(client_id);
+
+ // means the client was cleaned from map; should never happen
+ // as long as cleaning times are long enough
+ assert(client_map.end() != client_it);
+ reduce_reservation_tags(*client_it->second);
+ }
+
+
+ // data_mtx should be held when called
+ NextReq do_next_request(Time now) {
+ NextReq result;
+
+ // if reservation queue is empty, all are empty (i.e., no active clients)
+ if(resv_heap.empty()) {
+ result.type = NextReqType::none;
+ return result;
+ }
+
+ // try constraint (reservation) based scheduling
+
+ auto& reserv = resv_heap.top();
+ if (reserv.has_request() &&
+ reserv.next_request().tag.reservation <= now) {
+ result.type = NextReqType::returning;
+ result.heap_id = HeapId::reservation;
+ return result;
+ }
+
+ // no existing reservations before now, so try weight-based
+ // scheduling
+
+ // all items that are within limit are eligible based on
+ // priority
+ auto limits = &limit_heap.top();
+ while (limits->has_request() &&
+ !limits->next_request().tag.ready &&
+ limits->next_request().tag.limit <= now) {
+ limits->next_request().tag.ready = true;
+ ready_heap.promote(*limits);
+ limit_heap.demote(*limits);
+
+ limits = &limit_heap.top();
+ }
+
+ auto& readys = ready_heap.top();
+ if (readys.has_request() &&
+ readys.next_request().tag.ready &&
+ readys.next_request().tag.proportion < max_tag) {
+ result.type = NextReqType::returning;
+ result.heap_id = HeapId::ready;
+ return result;
+ }
+
+ // if nothing is schedulable by reservation or
+ // proportion/weight, and if we allow limit break, try to
+ // schedule something with the lowest proportion tag or
+ // alternatively lowest reservation tag.
+ if (allow_limit_break) {
+ if (readys.has_request() &&
+ readys.next_request().tag.proportion < max_tag) {
+ result.type = NextReqType::returning;
+ result.heap_id = HeapId::ready;
+ return result;
+ } else if (reserv.has_request() &&
+ reserv.next_request().tag.reservation < max_tag) {
+ result.type = NextReqType::returning;
+ result.heap_id = HeapId::reservation;
+ return result;
+ }
+ }
+
+ // nothing scheduled; make sure we re-run when next
+ // reservation item or next limited item comes up
+
+ Time next_call = TimeMax;
+ if (resv_heap.top().has_request()) {
+ next_call =
+ min_not_0_time(next_call,
+ resv_heap.top().next_request().tag.reservation);
+ }
+ if (limit_heap.top().has_request()) {
+ const auto& next = limit_heap.top().next_request();
+ assert(!next.tag.ready || max_tag == next.tag.proportion);
+ next_call = min_not_0_time(next_call, next.tag.limit);
+ }
+ if (next_call < TimeMax) {
+ result.type = NextReqType::future;
+ result.when_ready = next_call;
+ return result;
+ } else {
+ result.type = NextReqType::none;
+ return result;
+ }
+ } // do_next_request
+
+
+ // if possible is not zero and less than current then return it;
+ // otherwise return current; the idea is we're trying to find
+ // the minimal time but ignoring zero
+ static inline const Time& min_not_0_time(const Time& current,
+ const Time& possible) {
+ return TimeZero == possible ? current : std::min(current, possible);
+ }
+
+
+ /*
+ * This is being called regularly by RunEvery. Every time it's
+ * called it notes the time and delta counter (mark point) in a
+ * deque. It also looks at the deque to find the most recent
+ * mark point that is older than clean_age. It then walks the
+ * map and delete all server entries that were last used before
+ * that mark point.
+ */
+ void do_clean() {
+ TimePoint now = std::chrono::steady_clock::now();
+ DataGuard g(data_mtx);
+ clean_mark_points.emplace_back(MarkPoint(now, tick));
+
+ // first erase the super-old client records
+
+ Counter erase_point = 0;
+ auto point = clean_mark_points.front();
+ while (point.first <= now - erase_age) {
+ erase_point = point.second;
+ clean_mark_points.pop_front();
+ point = clean_mark_points.front();
+ }
+
+ Counter idle_point = 0;
+ for (auto i : clean_mark_points) {
+ if (i.first <= now - idle_age) {
+ idle_point = i.second;
+ } else {
+ break;
+ }
+ }
+
+ if (erase_point > 0 || idle_point > 0) {
+ for (auto i = client_map.begin(); i != client_map.end(); /* empty */) {
+ auto i2 = i++;
+ if (erase_point && i2->second->last_tick <= erase_point) {
+ delete_from_heaps(i2->second);
+ client_map.erase(i2);
+ } else if (idle_point && i2->second->last_tick <= idle_point) {
+ i2->second->idle = true;
+ }
+ } // for
+ } // if
+ } // do_clean
+
+
+ // data_mtx must be held by caller
+ template<IndIntruHeapData ClientRec::*C1,typename C2>
+ void delete_from_heap(ClientRecRef& client,
+ c::IndIntruHeap<ClientRecRef,ClientRec,C1,C2,B>& heap) {
+ auto i = heap.rfind(client);
+ heap.remove(i);
+ }
+
+
+ // data_mtx must be held by caller
+ void delete_from_heaps(ClientRecRef& client) {
+ delete_from_heap(client, resv_heap);
+#if USE_PROP_HEAP
+ delete_from_heap(client, prop_heap);
+#endif
+ delete_from_heap(client, limit_heap);
+ delete_from_heap(client, ready_heap);
+ }
+ }; // class PriorityQueueBase
+
+
+ template<typename C, typename R, uint B=2>
+ class PullPriorityQueue : public PriorityQueueBase<C,R,B> {
+ using super = PriorityQueueBase<C,R,B>;
+
+ public:
+
+ // When a request is pulled, this is the return type.
+ struct PullReq {
+ struct Retn {
+ C client;
+ typename super::RequestRef request;
+ PhaseType phase;
+ };
+
+ typename super::NextReqType type;
+ boost::variant<Retn,Time> data;
+
+ bool is_none() const { return type == super::NextReqType::none; }
+
+ bool is_retn() const { return type == super::NextReqType::returning; }
+ Retn& get_retn() {
+ return boost::get<Retn>(data);
+ }
+
+ bool is_future() const { return type == super::NextReqType::future; }
+ Time getTime() const { return boost::get<Time>(data); }
+ };
+
+
+#ifdef PROFILE
+ ProfileTimer<std::chrono::nanoseconds> pull_request_timer;
+ ProfileTimer<std::chrono::nanoseconds> add_request_timer;
+#endif
+
+ template<typename Rep, typename Per>
+ PullPriorityQueue(typename super::ClientInfoFunc _client_info_f,
+ std::chrono::duration<Rep,Per> _idle_age,
+ std::chrono::duration<Rep,Per> _erase_age,
+ std::chrono::duration<Rep,Per> _check_time,
+ bool _allow_limit_break = false) :
+ super(_client_info_f,
+ _idle_age, _erase_age, _check_time,
+ _allow_limit_break)
+ {
+ // empty
+ }
+
+
+ // pull convenience constructor
+ PullPriorityQueue(typename super::ClientInfoFunc _client_info_f,
+ bool _allow_limit_break = false) :
+ PullPriorityQueue(_client_info_f,
+ std::chrono::minutes(10),
+ std::chrono::minutes(15),
+ std::chrono::minutes(6),
+ _allow_limit_break)
+ {
+ // empty
+ }
+
+
+ inline void add_request(const R& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ double addl_cost = 0.0) {
+ add_request(typename super::RequestRef(new R(request)),
+ client_id,
+ req_params,
+ get_time(),
+ addl_cost);
+ }
+
+
+ inline void add_request(const R& request,
+ const C& client_id,
+ double addl_cost = 0.0) {
+ static const ReqParams null_req_params;
+ add_request(typename super::RequestRef(new R(request)),
+ client_id,
+ null_req_params,
+ get_time(),
+ addl_cost);
+ }
+
+
+
+ inline void add_request_time(const R& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ const Time time,
+ double addl_cost = 0.0) {
+ add_request(typename super::RequestRef(new R(request)),
+ client_id,
+ req_params,
+ time,
+ addl_cost);
+ }
+
+
+ inline void add_request(typename super::RequestRef&& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ double addl_cost = 0.0) {
+ add_request(request, req_params, client_id, get_time(), addl_cost);
+ }
+
+
+ inline void add_request(typename super::RequestRef&& request,
+ const C& client_id,
+ double addl_cost = 0.0) {
+ static const ReqParams null_req_params;
+ add_request(request, null_req_params, client_id, get_time(), addl_cost);
+ }
+
+
+ // this does the work; the versions above provide alternate interfaces
+ void add_request(typename super::RequestRef&& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ const Time time,
+ double addl_cost = 0.0) {
+ typename super::DataGuard g(this->data_mtx);
+#ifdef PROFILE
+ add_request_timer.start();
+#endif
+ super::do_add_request(std::move(request),
+ client_id,
+ req_params,
+ time,
+ addl_cost);
+ // no call to schedule_request for pull version
+#ifdef PROFILE
+ add_request_timer.stop();
+#endif
+ }
+
+
+ inline PullReq pull_request() {
+ return pull_request(get_time());
+ }
+
+
+ PullReq pull_request(Time now) {
+ PullReq result;
+ typename super::DataGuard g(this->data_mtx);
+#ifdef PROFILE
+ pull_request_timer.start();
+#endif
+
+ typename super::NextReq next = super::do_next_request(now);
+ result.type = next.type;
+ switch(next.type) {
+ case super::NextReqType::none:
+ return result;
+ break;
+ case super::NextReqType::future:
+ result.data = next.when_ready;
+ return result;
+ break;
+ case super::NextReqType::returning:
+ // to avoid nesting, break out and let code below handle this case
+ break;
+ default:
+ assert(false);
+ }
+
+ // we'll only get here if we're returning an entry
+
+ auto process_f =
+ [&] (PullReq& pull_result, PhaseType phase) ->
+ std::function<void(const C&,
+ typename super::RequestRef&)> {
+ return [&pull_result, phase](const C& client,
+ typename super::RequestRef& request) {
+ pull_result.data =
+ typename PullReq::Retn{client, std::move(request), phase};
+ };
+ };
+
+ switch(next.heap_id) {
+ case super::HeapId::reservation:
+ super::pop_process_request(this->resv_heap,
+ process_f(result, PhaseType::reservation));
+ ++this->reserv_sched_count;
+ break;
+ case super::HeapId::ready:
+ super::pop_process_request(this->ready_heap,
+ process_f(result, PhaseType::priority));
+ { // need to use retn temporarily
+ auto& retn = boost::get<typename PullReq::Retn>(result.data);
+ super::reduce_reservation_tags(retn.client);
+ }
+ ++this->prop_sched_count;
+ break;
+ default:
+ assert(false);
+ }
+
+#ifdef PROFILE
+ pull_request_timer.stop();
+#endif
+ return result;
+ } // pull_request
+
+
+ protected:
+
+
+ // data_mtx should be held when called; unfortunately this
+ // function has to be repeated in both push & pull
+ // specializations
+ typename super::NextReq next_request() {
+ return next_request(get_time());
+ }
+ }; // class PullPriorityQueue
+
+
+ // PUSH version
+ template<typename C, typename R, uint B=2>
+ class PushPriorityQueue : public PriorityQueueBase<C,R,B> {
+
+ protected:
+
+ using super = PriorityQueueBase<C,R,B>;
+
+ public:
+
+ // a function to see whether the server can handle another request
+ using CanHandleRequestFunc = std::function<bool(void)>;
+
+ // a function to submit a request to the server; the second
+ // parameter is a callback when it's completed
+ using HandleRequestFunc =
+ std::function<void(const C&,typename super::RequestRef,PhaseType)>;
+
+ protected:
+
+ CanHandleRequestFunc can_handle_f;
+ HandleRequestFunc handle_f;
+ // for handling timed scheduling
+ std::mutex sched_ahead_mtx;
+ std::condition_variable sched_ahead_cv;
+ Time sched_ahead_when = TimeZero;
+
+#ifdef PROFILE
+ public:
+ ProfileTimer<std::chrono::nanoseconds> add_request_timer;
+ ProfileTimer<std::chrono::nanoseconds> request_complete_timer;
+ protected:
+#endif
+
+ // NB: threads declared last, so constructed last and destructed first
+
+ std::thread sched_ahead_thd;
+
+ public:
+
+ // push full constructor
+ template<typename Rep, typename Per>
+ PushPriorityQueue(typename super::ClientInfoFunc _client_info_f,
+ CanHandleRequestFunc _can_handle_f,
+ HandleRequestFunc _handle_f,
+ std::chrono::duration<Rep,Per> _idle_age,
+ std::chrono::duration<Rep,Per> _erase_age,
+ std::chrono::duration<Rep,Per> _check_time,
+ bool _allow_limit_break = false) :
+ super(_client_info_f,
+ _idle_age, _erase_age, _check_time,
+ _allow_limit_break)
+ {
+ can_handle_f = _can_handle_f;
+ handle_f = _handle_f;
+ sched_ahead_thd = std::thread(&PushPriorityQueue::run_sched_ahead, this);
+ }
+
+
+ // push convenience constructor
+ PushPriorityQueue(typename super::ClientInfoFunc _client_info_f,
+ CanHandleRequestFunc _can_handle_f,
+ HandleRequestFunc _handle_f,
+ bool _allow_limit_break = false) :
+ PushPriorityQueue(_client_info_f,
+ _can_handle_f,
+ _handle_f,
+ std::chrono::minutes(10),
+ std::chrono::minutes(15),
+ std::chrono::minutes(6),
+ _allow_limit_break)
+ {
+ // empty
+ }
+
+
+ ~PushPriorityQueue() {
+ this->finishing = true;
+ sched_ahead_cv.notify_one();
+ sched_ahead_thd.join();
+ }
+
+ public:
+
+ inline void add_request(const R& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ double addl_cost = 0.0) {
+ add_request(typename super::RequestRef(new R(request)),
+ client_id,
+ req_params,
+ get_time(),
+ addl_cost);
+ }
+
+
+ inline void add_request(typename super::RequestRef&& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ double addl_cost = 0.0) {
+ add_request(request, req_params, client_id, get_time(), addl_cost);
+ }
+
+
+ inline void add_request_time(const R& request,
+ const C& client_id,
+ const ReqParams& req_params,
+ const Time time,
+ double addl_cost = 0.0) {
+ add_request(typename super::RequestRef(new R(request)),
+ client_id,
+ req_params,
+ time,
+ addl_cost);
+ }
+
+
+ void add_request(typename super::RequestRef&& request,
- const Time time,
- double addl_cost = 0.0) {
++ const C& client_id,
+ const ReqParams& req_params,
++ const Time time,
++ double addl_cost = 0.0) {
+ typename super::DataGuard g(this->data_mtx);
+#ifdef PROFILE
+ add_request_timer.start();
+#endif
+ super::do_add_request(std::move(request),
+ client_id,
+ req_params,
+ time,
+ addl_cost);
+ schedule_request();
+#ifdef PROFILE
+ add_request_timer.stop();
+#endif
+ }
+
+
+ void request_completed() {
+ typename super::DataGuard g(this->data_mtx);
+#ifdef PROFILE
+ request_complete_timer.start();
+#endif
+ schedule_request();
+#ifdef PROFILE
+ request_complete_timer.stop();
+#endif
+ }
+
+ protected:
+
+ // data_mtx should be held when called; furthermore, the heap
+ // should not be empty and the top element of the heap should
+ // not be already handled
+ //
+ // NOTE: the use of "super::ClientRec" in either the template
+ // construct or as a parameter to submit_top_request generated
+ // a compiler error in g++ 4.8.4, when ClientRec was
+ // "protected" rather than "public". By g++ 6.3.1 this was not
+ // an issue. But for backwards compatibility
+ // PriorityQueueBase::ClientRec is public.
+ template<typename C1,
+ IndIntruHeapData super::ClientRec::*C2,
+ typename C3,
+ uint B4>
+ C submit_top_request(IndIntruHeap<C1,typename super::ClientRec,C2,C3,B4>& heap,
+ PhaseType phase) {
+ C client_result;
+ super::pop_process_request(heap,
+ [this, phase, &client_result]
+ (const C& client,
+ typename super::RequestRef& request) {
+ client_result = client;
+ handle_f(client, std::move(request), phase);
+ });
+ return client_result;
+ }
+
+
+ // data_mtx should be held when called
+ void submit_request(typename super::HeapId heap_id) {
+ C client;
+ switch(heap_id) {
+ case super::HeapId::reservation:
+ // don't need to note client
+ (void) submit_top_request(this->resv_heap, PhaseType::reservation);
+ // unlike the other two cases, we do not reduce reservation
+ // tags here
+ ++this->reserv_sched_count;
+ break;
+ case super::HeapId::ready:
+ client = submit_top_request(this->ready_heap, PhaseType::priority);
+ super::reduce_reservation_tags(client);
+ ++this->prop_sched_count;
+ break;
+ default:
+ assert(false);
+ }
+ } // submit_request
+
+
+ // data_mtx should be held when called; unfortunately this
+ // function has to be repeated in both push & pull
+ // specializations
+ typename super::NextReq next_request() {
+ return next_request(get_time());
+ }
+
+
+ // data_mtx should be held when called; overrides member
+ // function in base class to add check for whether a request can
+ // be pushed to the server
+ typename super::NextReq next_request(Time now) {
+ if (!can_handle_f()) {
+ typename super::NextReq result;
+ result.type = super::NextReqType::none;
+ return result;
+ } else {
+ return super::do_next_request(now);
+ }
+ } // next_request
+
+
+ // data_mtx should be held when called
+ void schedule_request() {
+ typename super::NextReq next_req = next_request();
+ switch (next_req.type) {
+ case super::NextReqType::none:
+ return;
+ case super::NextReqType::future:
+ sched_at(next_req.when_ready);
+ break;
+ case super::NextReqType::returning:
+ submit_request(next_req.heap_id);
+ break;
+ default:
+ assert(false);
+ }
+ }
+
+
+ // this is the thread that handles running schedule_request at
+ // future times when nothing can be scheduled immediately
+ void run_sched_ahead() {
+ std::unique_lock<std::mutex> l(sched_ahead_mtx);
+
+ while (!this->finishing) {
+ if (TimeZero == sched_ahead_when) {
+ sched_ahead_cv.wait(l);
+ } else {
+ Time now;
+ while (!this->finishing && (now = get_time()) < sched_ahead_when) {
+ long microseconds_l = long(1 + 1000000 * (sched_ahead_when - now));
+ auto microseconds = std::chrono::microseconds(microseconds_l);
+ sched_ahead_cv.wait_for(l, microseconds);
+ }
+ sched_ahead_when = TimeZero;
+ if (this->finishing) return;
+
+ l.unlock();
+ if (!this->finishing) {
+ typename super::DataGuard g(this->data_mtx);
+ schedule_request();
+ }
+ l.lock();
+ }
+ }
+ }
+
+
+ void sched_at(Time when) {
+ std::lock_guard<std::mutex> l(sched_ahead_mtx);
++ if (this->finishing) return;
+ if (TimeZero == sched_ahead_when || when < sched_ahead_when) {
+ sched_ahead_when = when;
+ sched_ahead_cv.notify_one();
+ }
+ }
+ }; // class PushPriorityQueue
+
+ } // namespace dmclock
+} // namespace crimson
--- /dev/null
- remove(0);
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+/*
+ * Copyright (C) 2016 Red Hat Inc.
+ */
+
+
+#pragma once
+
+
+#include <memory>
+#include <vector>
+#include <string>
+#include <iostream>
+#include <functional>
+#include <algorithm>
+
+#include "assert.h"
+
+
+namespace crimson {
+ using IndIntruHeapData = size_t;
+
+ /* T is the ultimate data that's being stored in the heap, although
+ * through indirection.
+ *
+ * I is the indirect type that will actually be stored in the heap
+ * and that must allow dereferencing (via operator*) to yield a
+ * T&.
+ *
+ * C is a functor when given two T&'s will return true if the first
+ * must precede the second.
+ *
+ * heap_info is a data member pointer as to where the heap data in T
+ * is stored.
+ *
+ * K is the branching factor of the heap, default is 2 (binary heap).
+ */
+ template<typename I,
+ typename T,
+ IndIntruHeapData T::*heap_info,
+ typename C,
+ uint K = 2>
+ class IndIntruHeap {
+
+ // shorthand
+ using HeapIndex = IndIntruHeapData;
+
+ static_assert(
+ std::is_same<T,typename std::pointer_traits<I>::element_type>::value,
+ "class I must resolve to class T by indirection (pointer dereference)");
+
+ static_assert(
+ std::is_same<bool,
+ typename std::result_of<C(const T&,const T&)>::type>::value,
+ "class C must define operator() to take two const T& and return a bool");
+
+ static_assert(K >= 2, "K (degree of branching) must be at least 2");
+
+ class Iterator {
+ friend IndIntruHeap<I, T, heap_info, C, K>;
+
+ IndIntruHeap<I, T, heap_info, C, K>& heap;
+ HeapIndex index;
+
+ Iterator(IndIntruHeap<I, T, heap_info, C, K>& _heap, HeapIndex _index) :
+ heap(_heap),
+ index(_index)
+ {
+ // empty
+ }
+
+ public:
+
+ Iterator(Iterator&& other) :
+ heap(other.heap),
+ index(other.index)
+ {
+ // empty
+ }
+
+ Iterator(const Iterator& other) :
+ heap(other.heap),
+ index(other.index)
+ {
+ // empty
+ }
+
+ Iterator& operator=(Iterator&& other) {
+ std::swap(heap, other.heap);
+ std::swap(index, other.index);
+ return *this;
+ }
+
+ Iterator& operator=(const Iterator& other) {
+ heap = other.heap;
+ index = other.index;
+ }
+
+ Iterator& operator++() {
+ if (index <= heap.count) {
+ ++index;
+ }
+ return *this;
+ }
+
+ bool operator==(const Iterator& other) const {
+ return &heap == &other.heap && index == other.index;
+ }
+
+ bool operator!=(const Iterator& other) const {
+ return !(*this == other);
+ }
+
+ T& operator*() {
+ return *heap.data[index];
+ }
+
+ T* operator->() {
+ return &(*heap.data[index]);
+ }
+
+#if 0
+ // the item this iterator refers to
+ void increase() {
+ heap.sift_up(index);
+ }
+#endif
+ }; // class Iterator
+
+
+ class ConstIterator {
+ friend IndIntruHeap<I, T, heap_info, C, K>;
+
+ const IndIntruHeap<I, T, heap_info, C, K>& heap;
+ HeapIndex index;
+
+ ConstIterator(const IndIntruHeap<I, T, heap_info, C, K>& _heap,
+ HeapIndex _index) :
+ heap(_heap),
+ index(_index)
+ {
+ // empty
+ }
+
+ public:
+
+ ConstIterator(ConstIterator&& other) :
+ heap(other.heap),
+ index(other.index)
+ {
+ // empty
+ }
+
+ ConstIterator(const ConstIterator& other) :
+ heap(other.heap),
+ index(other.index)
+ {
+ // empty
+ }
+
+ ConstIterator& operator=(ConstIterator&& other) {
+ std::swap(heap, other.heap);
+ std::swap(index, other.index);
+ return *this;
+ }
+
+ ConstIterator& operator=(const ConstIterator& other) {
+ heap = other.heap;
+ index = other.index;
+ }
+
+ ConstIterator& operator++() {
+ if (index <= heap.count) {
+ ++index;
+ }
+ return *this;
+ }
+
+ bool operator==(const ConstIterator& other) const {
+ return &heap == &other.heap && index == other.index;
+ }
+
+ bool operator!=(const ConstIterator& other) const {
+ return !(*this == other);
+ }
+
+ const T& operator*() {
+ return *heap.data[index];
+ }
+
+ const T* operator->() {
+ return &(*heap.data[index]);
+ }
+ }; // class ConstIterator
+
+
+ protected:
+
+ std::vector<I> data;
+ HeapIndex count;
+ C comparator;
+
+ public:
+
+ IndIntruHeap() :
+ count(0)
+ {
+ // empty
+ }
+
+ IndIntruHeap(const IndIntruHeap<I,T,heap_info,C,K>& other) :
+ count(other.count)
+ {
+ for (HeapIndex i = 0; i < other.count; ++i) {
+ data.push_back(other.data[i]);
+ }
+ }
+
+ bool empty() const { return 0 == count; }
+
+ size_t size() const { return (size_t) count; }
+
+ T& top() { return *data[0]; }
+
+ const T& top() const { return *data[0]; }
+
+ I& top_ind() { return data[0]; }
+
+ const I& top_ind() const { return data[0]; }
+
+ void push(I&& item) {
+ HeapIndex i = count++;
+ intru_data_of(item) = i;
+ data.emplace_back(std::move(item));
+ sift_up(i);
+ }
+
+ void push(const I& item) {
+ I copy(item);
+ push(std::move(copy));
+ }
+
+ void pop() {
++ remove(HeapIndex(0));
+ }
+
+ void remove(Iterator& i) {
+ remove(i.index);
+ i = end();
+ }
+
+ Iterator find(const I& ind_item) {
+ for (HeapIndex i = 0; i < count; ++i) {
+ if (data[i] == ind_item) {
+ return Iterator(*this, i);
+ }
+ }
+ return end();
+ }
+
+ // when passing in value we do a comparison via operator==
+ Iterator find(const T& item) {
+ for (HeapIndex i = 0; i < count; ++i) {
+ if (*data[i] == item) {
+ return Iterator(*this, i);
+ }
+ }
+ return end();
+ }
+
+ // reverse find -- start looking from bottom of heap
+ Iterator rfind(const I& ind_item) {
+ // HeapIndex is unsigned, so we can't allow to go negative; so
+ // we'll keep it one more than actual index
+ for (HeapIndex i = count; i > 0; --i) {
+ if (data[i-1] == ind_item) {
+ return Iterator(*this, i-1);
+ }
+ }
+ return end();
+ }
+
+ // reverse find -- start looking from bottom of heap
+ Iterator rfind(const T& item) {
+ // HeapIndex is unsigned, so we can't allow to go negative; so
+ // we'll keep it one more than actual index
+ for (HeapIndex i = count; i > 0; --i) {
+ if (*data[i-1] == item) {
+ return Iterator(*this, i-1);
+ }
+ }
+ return end();
+ }
+
+ ConstIterator find(const I& ind_item) const {
+ for (HeapIndex i = 0; i < count; ++i) {
+ if (data[i] == ind_item) {
+ return ConstIterator(*this, i);
+ }
+ }
+ return cend();
+ }
+
+ // when passing in value we do a comparison via operator==
+ ConstIterator find(const T& item) const {
+ for (HeapIndex i = 0; i < count; ++i) {
+ if (*data[i] == item) {
+ return ConstIterator(*this, i);
+ }
+ }
+ return cend();
+ }
+
+ // reverse find -- start looking from bottom of heap
+ ConstIterator rfind(const I& ind_item) const {
+ // HeapIndex is unsigned, so we can't allow to go negative; so
+ // we'll keep it one more than actual index
+ for (HeapIndex i = count; i > 0; --i) {
+ if (data[i-1] == ind_item) {
+ return ConstIterator(*this, i-1);
+ }
+ }
+ return cend();
+ }
+
+ // reverse find -- start looking from bottom of heap
+ ConstIterator rfind(const T& item) const {
+ // HeapIndex is unsigned, so we can't allow to go negative; so
+ // we'll keep it one more than actual index
+ for (HeapIndex i = count; i > 0; --i) {
+ if (*data[i-1] == item) {
+ return ConstIterator(*this, i-1);
+ }
+ }
+ return cend();
+ }
+
+ void promote(T& item) {
+ sift_up(item.*heap_info);
+ }
+
+ void demote(T& item) {
+ sift_down(item.*heap_info);
+ }
+
+ void adjust(T& item) {
+ sift(item.*heap_info);
+ }
+
+ Iterator begin() {
+ return Iterator(*this, 0);
+ }
+
+ Iterator end() {
+ return Iterator(*this, count);
+ }
+
+ ConstIterator cbegin() const {
+ return ConstIterator(*this, 0);
+ }
+
+ ConstIterator cend() const {
+ return ConstIterator(*this, count);
+ }
+
+ friend std::ostream& operator<<(std::ostream& out, const IndIntruHeap& h) {
+ auto i = h.data.cbegin();
+ if (i != h.data.cend()) {
+ out << **i;
+ ++i;
+ while (i != h.data.cend()) {
+ out << ", " << **i;
+ }
+ }
+ return out;
+ }
+
+ // can only be called if I is copyable; copies heap into a vector
+ // and sorts it before displaying it
+ std::ostream&
+ display_sorted(std::ostream& out,
+ std::function<bool(const T&)> filter = all_filter) const {
+ static_assert(std::is_copy_constructible<I>::value,
+ "cannot call display_sorted when class I is not copy"
+ " constructible");
+ auto compare = [this] (const I first, const I second) -> bool {
+ return this->comparator(*first, *second);
+ };
+ std::vector<I> copy(data);
+ std::sort(copy.begin(), copy.end(), compare);
+
+ bool first = true;
+ for (auto c = copy.begin(); c != copy.end(); ++c) {
+ if (filter(**c)) {
+ if (!first) {
+ out << ", ";
+ } else {
+ first = false;
+ }
+ out << **c;
+ }
+ }
+
+ return out;
+ }
+
+
+ protected:
+
+ static IndIntruHeapData& intru_data_of(I& item) {
+ return (*item).*heap_info;
+ }
+
+ void remove(HeapIndex i) {
+ std::swap(data[i], data[--count]);
+ intru_data_of(data[i]) = i;
+ data.pop_back();
+
+ // the following needs to be sift (and not sift_down) as it can
+ // go up or down the heap; imagine the heap vector contains 0,
+ // 10, 100, 20, 30, 200, 300, 40; then 200 is removed, and 40
+ // would have to be sifted upwards
+ // sift(i);
+ sift(i);
+ }
+
+ // default value of filter parameter to display_sorted
+ static bool all_filter(const T& data) { return true; }
+
+ // when i is negative?
+ static inline HeapIndex parent(HeapIndex i) {
+ assert(0 != i);
+ return (i - 1) / K;
+ }
+
+ // index of left child when K==2, index of left-most child when K>2
+ static inline HeapIndex lhs(HeapIndex i) { return K*i + 1; }
+
+ // index of right child when K==2, index of right-most child when K>2
+ static inline HeapIndex rhs(HeapIndex i) { return K*i + K; }
+
+ void sift_up(HeapIndex i) {
+ while (i > 0) {
+ HeapIndex pi = parent(i);
+ if (!comparator(*data[i], *data[pi])) {
+ break;
+ }
+
+ std::swap(data[i], data[pi]);
+ intru_data_of(data[i]) = i;
+ intru_data_of(data[pi]) = pi;
+ i = pi;
+ }
+ } // sift_up
+
+ // use this sift_down definition when K>2; it's more general and
+ // uses a loop; EnableBool insures template uses a template
+ // parameter
+ template<bool EnableBool=true>
+ typename std::enable_if<(K>2)&&EnableBool,void>::type sift_down(HeapIndex i) {
+ if (i >= count) return;
+ while (true) {
+ HeapIndex li = lhs(i);
+
+ if (li < count) {
+ HeapIndex ri = std::min(rhs(i), count - 1);
+
+ // find the index of min. child
+ HeapIndex min_i = li;
+ for (HeapIndex k = li + 1; k <= ri; ++k) {
+ if (comparator(*data[k], *data[min_i])) {
+ min_i = k;
+ }
+ }
+
+ if (comparator(*data[min_i], *data[i])) {
+ std::swap(data[i], data[min_i]);
+ intru_data_of(data[i]) = i;
+ intru_data_of(data[min_i]) = min_i;
+ i = min_i;
+ } else {
+ // no child is smaller
+ break;
+ }
+ } else {
+ // no children
+ break;
+ }
+ }
+ } // sift_down
+
+ // use this sift_down definition when K==2; EnableBool insures
+ // template uses a template parameter
+ template<bool EnableBool=true>
+ typename std::enable_if<K==2&&EnableBool,void>::type sift_down(HeapIndex i) {
+ if (i >= count) return;
+ while (true) {
+ const HeapIndex li = lhs(i);
+ const HeapIndex ri = 1 + li;
+
+ if (li < count) {
+ if (comparator(*data[li], *data[i])) {
+ if (ri < count && comparator(*data[ri], *data[li])) {
+ std::swap(data[i], data[ri]);
+ intru_data_of(data[i]) = i;
+ intru_data_of(data[ri]) = ri;
+ i = ri;
+ } else {
+ std::swap(data[i], data[li]);
+ intru_data_of(data[i]) = i;
+ intru_data_of(data[li]) = li;
+ i = li;
+ }
+ } else if (ri < count && comparator(*data[ri], *data[i])) {
+ std::swap(data[i], data[ri]);
+ intru_data_of(data[i]) = i;
+ intru_data_of(data[ri]) = ri;
+ i = ri;
+ } else {
+ // no child is smaller
+ break;
+ }
+ } else {
+ // no children
+ break;
+ }
+ } // while
+ } // sift_down
+
+ void sift(HeapIndex i) {
+ if (i == 0) {
+ // if we're at top, can only go down
+ sift_down(i);
+ } else {
+ HeapIndex pi = parent(i);
+ if (comparator(*data[i], *data[pi])) {
+ // if we can go up, we will
+ sift_up(i);
+ } else {
+ // otherwise we'll try to go down
+ sift_down(i);
+ }
+ }
+ } // sift
+ }; // class IndIntruHeap
+
+} // namespace crimson
--- /dev/null
- * Copyright (C) 2016 Red Hat Inc.
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
- #include <iostream>
-
++ * Copyright (C) 2017 Red Hat Inc.
+ */
+
+
- finishing = true;
- cv.notify_all();
+#include "run_every.h"
+
+
+// can define ADD_MOVE_SEMANTICS, although not fully debugged and tested
+
+
+namespace chrono = std::chrono;
+
+
+#ifdef ADD_MOVE_SEMANTICS
+crimson::RunEvery::RunEvery()
+{
+ // empty
+}
+
+
+crimson::RunEvery& crimson::RunEvery::operator=(crimson::RunEvery&& other)
+{
+ // finish run every thread
+ {
+ Guard g(mtx);
+ finishing = true;
+ cv.notify_one();
+ }
+ if (thd.joinable()) {
+ thd.join();
+ }
+
+ // transfer info over from previous thread
+ finishing.store(other.finishing);
+ wait_period = other.wait_period;
+ body = other.body;
+
+ // finish other thread
+ other.finishing.store(true);
+ other.cv.notify_one();
+
+ // start this thread
+ thd = std::thread(&RunEvery::run, this);
+
+ return *this;
+}
+#endif
+
+
+crimson::RunEvery::~RunEvery() {
++ join();
++}
++
++
++void crimson::RunEvery::join() {
++ {
++ Guard l(mtx);
++ if (finishing) return;
++ finishing = true;
++ cv.notify_all();
++ }
+ thd.join();
+}
+
+
+void crimson::RunEvery::run() {
+ Lock l(mtx);
+ while(!finishing) {
+ TimePoint until = chrono::steady_clock::now() + wait_period;
+ while (!finishing && chrono::steady_clock::now() < until) {
+ cv.wait_until(l, until);
+ }
+ if (!finishing) {
+ body();
+ }
+ }
+}
--- /dev/null
- * Copyright (C) 2016 Red Hat Inc.
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
++ * Copyright (C) 2017 Red Hat Inc.
+ */
+
+
+#pragma once
+
+#include <chrono>
+#include <mutex>
+#include <condition_variable>
+#include <thread>
+#include <functional>
+
+namespace crimson {
+ using std::chrono::duration_cast;
+ using std::chrono::milliseconds;
+
+ // runs a given simple function object waiting wait_period
+ // milliseconds between; the destructor stops the other thread
+ // immediately
+ class RunEvery {
+ using Lock = std::unique_lock<std::mutex>;
+ using Guard = std::lock_guard<std::mutex>;
+ using TimePoint = std::chrono::steady_clock::time_point;
+
+ bool finishing = false;
+ std::chrono::milliseconds wait_period;
+ std::function<void()> body;
+ std::mutex mtx;
+ std::condition_variable cv;
+
+ // put threads last so all other variables are initialized first
+
+ std::thread thd;
+
+ public:
+
+#ifdef ADD_MOVE_SEMANTICS
+ RunEvery();
+#endif
+
+ template<typename D>
+ RunEvery(D _wait_period,
+ std::function<void()> _body) :
+ wait_period(duration_cast<milliseconds>(_wait_period)),
+ body(_body)
+ {
+ thd = std::thread(&RunEvery::run, this);
+ }
+
+ RunEvery(const RunEvery& other) = delete;
+ RunEvery& operator=(const RunEvery& other) = delete;
+ RunEvery(RunEvery&& other) = delete;
+#ifdef ADD_MOVE_SEMANTICS
+ RunEvery& operator=(RunEvery&& other);
+#else
+ RunEvery& operator=(RunEvery&& other) = delete;
+#endif
+
+ ~RunEvery();
+
++ void join();
++
+ protected:
+
+ void run();
+ };
+}
--- /dev/null
-
- add_test(NAME dmclock-data-struct-tests
- COMMAND $<TARGET_FILE:dmclock-data-struct-tests>)
+include_directories(../src)
+
+set(local_flags "-Wall -pthread")
+
+# dmclock does not use intrusive heap (but it does use indirect
+# intrusive heap), so we won't use this code
+if(false)
+ set(srcs
+ test_intrusive_heap.cc)
+ add_executable(test_intru_heap test_intrusive_heap.cc)
+ set_source_files_properties(${srcs}
+ PROPERTIES
+ COMPILE_FLAGS "${local_flags}")
+endif(false)
+
+set(test_srcs test_indirect_intrusive_heap.cc)
+
+set_source_files_properties(${test_srcs}
+ PROPERTIES
+ COMPILE_FLAGS "${local_flags}"
+ )
+
+add_executable(dmclock-data-struct-tests ${test_srcs})
+
+target_link_libraries(dmclock-data-struct-tests
+ LINK_PRIVATE gtest gtest_main pthread)
--- /dev/null
- include_directories(${BOOST_INCLUDE_DIR})
+include_directories(../src)
+include_directories(../support/src)
+include_directories(../sim/src)
- LINK_PRIVATE $<TARGET_FILE:dmclock> pthread ${GTEST_LIBRARY} ${GTEST_MAIN_LIBRARY})
+
+set(support_srcs ../sim/src/test_dmclock.cc)
+set(test_srcs
+ test_test_client.cc
+ test_dmclock_server.cc
+ test_dmclock_client.cc
+ )
+
+set_source_files_properties(${core_srcs} ${test_srcs}
+ PROPERTIES
+ COMPILE_FLAGS "${local_flags}"
+ )
+
+add_executable(dmclock-tests ${test_srcs} ${support_srcs})
+
+if (TARGET gtest AND TARGET gtest_main)
+ add_dependencies(dmclock-tests gtest gtest_main)
+ target_link_libraries(dmclock-tests
+ LINK_PRIVATE $<TARGET_FILE:dmclock>
+ pthread
+ $<TARGET_FILE:gtest>
+ $<TARGET_FILE:gtest_main>)
+else()
+ target_link_libraries(dmclock-tests
-
- add_test(NAME dmclock-tests
- COMMAND $<TARGET_FILE:dmclock-tests>)
++ LINK_PRIVATE $<TARGET_FILE:dmclock> pthread ${GTEST_LIBRARIES} ${GTEST_MAIN_LIBRARIES})
+endif()
+
+add_dependencies(dmclock-tests dmclock)
--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Copyright (C) 2016 Red Hat Inc.
+ */
+
+#include <atomic>
+#include <thread>
+#include <chrono>
+#include <iostream>
+
+#include "gtest/gtest.h"
+
+#include "sim_recs.h"
+#include "sim_client.h"
+
+#include "test_dmclock.h"
+
+
+using namespace std::placeholders;
+
+namespace dmc = crimson::dmclock;
+namespace test = crimson::test_dmc;
+namespace sim = crimson::qos_simulation;
+
+using TimePoint = std::chrono::time_point<std::chrono::system_clock>;
+
+static TimePoint now() { return std::chrono::system_clock::now(); }
+
+
+TEST(test_client, full_bore_timing) {
+ std::atomic_ulong count(0);
+
+ ServerId server_id = 3;
+
+ sim::TestResponse resp(0);
+ dmc::PhaseType resp_params = dmc::PhaseType::priority;
+ test::DmcClient* client;
+
+ auto start = now();
+ client =
+ new test::DmcClient(ClientId(0),
+ [&] (const ServerId& server,
+ const sim::TestRequest& req,
+ const ClientId& client_id,
+ const dmc::ReqParams& req_params) {
+ ++count;
+ client->receive_response(resp, client_id, resp_params);
+ },
+ [&] (const uint64_t seed) -> ServerId& {
+ return server_id;
+ },
+ test::dmc_client_accumulate_f,
+ 1000, // ops to run
+ 100, // iops goal
+ 5); // outstanding ops allowed
+ client->wait_until_done();
+ auto end = now();
+ EXPECT_EQ(1000u, count) << "didn't get right number of ops";
+
+ int milliseconds = (end - start) / std::chrono::milliseconds(1);
+ EXPECT_LT(10000, milliseconds) << "timing too fast to be correct";
+ EXPECT_GT(12000, milliseconds) << "timing suspiciously slow";
++
++ delete client;
+}
+
+
+TEST(test_client, paused_timing) {
+ std::atomic_ulong count(0);
+ std::atomic_ulong unresponded_count(0);
+ std::atomic_bool auto_respond(false);
+
+ ClientId my_client_id = 0;
+ ServerId server_id = 3;
+
+ sim::TestResponse resp(0);
+ dmc::PhaseType resp_params = dmc::PhaseType::priority;
+ test::DmcClient* client;
+
+ auto start = now();
+ client =
+ new test::DmcClient(my_client_id,
+ [&] (const ServerId& server,
+ const sim::TestRequest& req,
+ const ClientId& client_id,
+ const dmc::ReqParams& req_params) {
+ ++count;
+ if (auto_respond.load()) {
+ client->receive_response(resp, client_id, resp_params);
+ } else {
+ ++unresponded_count;
+ }
+ },
+ [&] (const uint64_t seed) -> ServerId& {
+ return server_id;
+ },
+ test::dmc_client_accumulate_f,
+
+ 1000, // ops to run
+ 100, // iops goal
+ 50); // outstanding ops allowed
+ std::thread t([&]() {
+ std::this_thread::sleep_for(std::chrono::seconds(5));
+ EXPECT_EQ(50u, unresponded_count.load()) <<
+ "should have 50 unresponded calls";
+ auto_respond = true;
+ // respond to those 50 calls
+ for(int i = 0; i < 50; ++i) {
+ client->receive_response(resp, my_client_id, resp_params);
+ --unresponded_count;
+ }
+ });
+
+ client->wait_until_done();
+ auto end = now();
+ int milliseconds = (end - start) / std::chrono::milliseconds(1);
+
+ // the 50 outstanding ops allowed means the first half-second of
+ // requests get responded to during the 5 second pause. So we have
+ // to adjust our expectations by a half-second.
+ EXPECT_LT(15000 - 500, milliseconds) << "timing too fast to be correct";
+ EXPECT_GT(17000 - 500, milliseconds) << "timing suspiciously slow";
+ t.join();
++
++ delete client;
+}
projects / ceph.git / commitdiff