+++ /dev/null
-// -*- 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 <map>
-#include <deque>
-#include <chrono>
-#include <thread>
-#include <mutex>
-#include <condition_variable>
-
-#include "run_every.h"
-#include "dmclock_util.h"
-#include "dmclock_recs.h"
-
-
-namespace crimson {
- namespace dmclock {
- struct ServerInfo {
- Counter delta_prev_req;
- Counter rho_prev_req;
- uint32_t my_delta;
- uint32_t my_rho;
-
- ServerInfo(Counter _delta_prev_req,
- Counter _rho_prev_req) :
- delta_prev_req(_delta_prev_req),
- rho_prev_req(_rho_prev_req),
- my_delta(0),
- my_rho(0)
- {
- // empty
- }
-
- inline void req_update(Counter delta, Counter rho) {
- delta_prev_req = delta;
- rho_prev_req = rho;
- my_delta = 0;
- my_rho = 0;
- }
-
- inline void resp_update(PhaseType phase) {
- ++my_delta;
- if (phase == PhaseType::reservation) ++my_rho;
- }
- };
-
-
- // S is server identifier type
- template<typename S>
- class ServiceTracker {
- // we don't want to include gtest.h just for FRIEND_TEST
- friend class dmclock_client_server_erase_Test;
-
- using TimePoint = decltype(std::chrono::steady_clock::now());
- using Duration = std::chrono::milliseconds;
- using MarkPoint = std::pair<TimePoint,Counter>;
-
- Counter delta_counter; // # reqs completed
- Counter rho_counter; // # reqs completed via reservation
- std::map<S,ServerInfo> server_map;
- mutable std::mutex data_mtx; // protects Counters and map
-
- using DataGuard = std::lock_guard<decltype(data_mtx)>;
-
- // clean config
-
- std::deque<MarkPoint> clean_mark_points;
- Duration clean_age; // age at which ServerInfo cleaned
-
- // NB: All threads declared at end, so they're destructed firs!
-
- std::unique_ptr<RunEvery> cleaning_job;
-
-
- public:
-
- // we have to start the counters at 1, as 0 is used in the
- // cleaning process
- template<typename Rep, typename Per>
- ServiceTracker(std::chrono::duration<Rep,Per> _clean_every,
- std::chrono::duration<Rep,Per> _clean_age) :
- delta_counter(1),
- rho_counter(1),
- clean_age(std::chrono::duration_cast<Duration>(_clean_age))
- {
- cleaning_job =
- std::unique_ptr<RunEvery>(
- new RunEvery(_clean_every,
- std::bind(&ServiceTracker::do_clean, this)));
- }
-
-
- // the reason we're overloading the constructor rather than
- // using default values for the arguments is so that callers
- // have to either use all defaults or specify all timings; with
- // default arguments they could specify some without others
- ServiceTracker() :
- ServiceTracker(std::chrono::minutes(5), std::chrono::minutes(10))
- {
- // empty
- }
-
-
- /*
- * Incorporates the RespParams received into the various counter.
- */
- void track_resp(const S& server_id, const PhaseType& phase) {
- DataGuard g(data_mtx);
-
- auto it = server_map.find(server_id);
- if (server_map.end() == it) {
- // this code can only run if a request did not precede the
- // response or if the record was cleaned up b/w when
- // the request was made and now
- ServerInfo si(delta_counter, rho_counter);
- si.resp_update(phase);
- server_map.emplace(server_id, si);
- } else {
- it->second.resp_update(phase);
- }
-
- ++delta_counter;
- if (PhaseType::reservation == phase) {
- ++rho_counter;
- }
- }
-
-
- /*
- * Returns the ReqParams for the given server.
- */
- ReqParams get_req_params(const S& server) {
- DataGuard g(data_mtx);
- auto it = server_map.find(server);
- if (server_map.end() == it) {
- server_map.emplace(server, ServerInfo(delta_counter, rho_counter));
- return ReqParams(1, 1);
- } else {
- Counter delta =
- 1 + delta_counter - it->second.delta_prev_req - it->second.my_delta;
- Counter rho =
- 1 + rho_counter - it->second.rho_prev_req - it->second.my_rho;
-
- it->second.req_update(delta_counter, rho_counter);
-
- return ReqParams(uint32_t(delta), uint32_t(rho));
- }
- }
-
- private:
-
- /*
- * 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, delta_counter));
-
- Counter earliest = 0;
- auto point = clean_mark_points.front();
- while (point.first <= now - clean_age) {
- earliest = point.second;
- clean_mark_points.pop_front();
- point = clean_mark_points.front();
- }
-
- if (earliest > 0) {
- for (auto i = server_map.begin();
- i != server_map.end();
- /* empty */) {
- auto i2 = i++;
- if (i2->second.delta_prev_req <= earliest) {
- server_map.erase(i2);
- }
- }
- }
- } // do_clean
- }; // class ServiceTracker
- }
-}
Code Review / stor4nfv.git / blobdiff