--- /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 <atomic>
+#include <mutex>
+#include <condition_variable>
+#include <thread>
+#include <chrono>
+#include <vector>
+#include <deque>
+#include <iostream>
+
+#include "sim_recs.h"
+
+
+namespace crimson {
+ namespace qos_simulation {
+
+ struct req_op_t {};
+ struct wait_op_t {};
+ constexpr struct req_op_t req_op {};
+ constexpr struct wait_op_t wait_op {};
+
+
+ enum class CliOp { req, wait };
+ struct CliInst {
+ CliOp op;
+ union {
+ std::chrono::milliseconds wait_time;
+ struct {
+ uint32_t count;
+ std::chrono::microseconds time_bw_reqs;
+ uint16_t max_outstanding;
+ } req_params;
+ } args;
+
+ // D is a duration type
+ template<typename D>
+ CliInst(wait_op_t, D duration) :
+ op(CliOp::wait)
+ {
+ args.wait_time =
+ std::chrono::duration_cast<std::chrono::milliseconds>(duration);
+ }
+
+ CliInst(req_op_t,
+ uint32_t count, double ops_per_sec, uint16_t max_outstanding) :
+ op(CliOp::req)
+ {
+ args.req_params.count = count;
+ args.req_params.max_outstanding = max_outstanding;
+ uint32_t us = uint32_t(0.5 + 1.0 / ops_per_sec * 1000000);
+ args.req_params.time_bw_reqs = std::chrono::microseconds(us);
+ }
+ };
+
+
+ using ServerSelectFunc = std::function<const ServerId&(uint64_t seed)>;
+
+
+ template<typename SvcTrk, typename ReqPm, typename RespPm, typename Accum>
+ class SimulatedClient {
+ public:
+
+ struct InternalStats {
+ std::mutex mtx;
+ std::chrono::nanoseconds track_resp_time;
+ std::chrono::nanoseconds get_req_params_time;
+ uint32_t track_resp_count;
+ uint32_t get_req_params_count;
+
+ InternalStats() :
+ track_resp_time(0),
+ get_req_params_time(0),
+ track_resp_count(0),
+ get_req_params_count(0)
+ {
+ // empty
+ }
+ };
+
+ using SubmitFunc =
+ std::function<void(const ServerId&,
+ TestRequest&&,
+ const ClientId&,
+ const ReqPm&)>;
+
+ using ClientAccumFunc = std::function<void(Accum&,const RespPm&)>;
+
+ typedef std::chrono::time_point<std::chrono::steady_clock> TimePoint;
+
+ static TimePoint now() { return std::chrono::steady_clock::now(); }
+
+ protected:
+
+ struct RespQueueItem {
+ TestResponse response;
+ ServerId server_id;
+ RespPm resp_params;
+ };
+
+ const ClientId id;
+ const SubmitFunc submit_f;
+ const ServerSelectFunc server_select_f;
+ const ClientAccumFunc accum_f;
+
+ std::vector<CliInst> instructions;
+
+ SvcTrk service_tracker;
+
+ // TODO: use lock rather than atomic???
+ std::atomic_ulong outstanding_ops;
+ std::atomic_bool requests_complete;
+
+ std::deque<RespQueueItem> resp_queue;
+
+ std::mutex mtx_req;
+ std::condition_variable cv_req;
+
+ std::mutex mtx_resp;
+ std::condition_variable cv_resp;
+
+ using RespGuard = std::lock_guard<decltype(mtx_resp)>;
+ using Lock = std::unique_lock<std::mutex>;
+
+ // data collection
+
+ std::vector<TimePoint> op_times;
+ Accum accumulator;
+ InternalStats internal_stats;
+
+ std::thread thd_req;
+ std::thread thd_resp;
+
+ public:
+
+ SimulatedClient(ClientId _id,
+ const SubmitFunc& _submit_f,
+ const ServerSelectFunc& _server_select_f,
+ const ClientAccumFunc& _accum_f,
+ const std::vector<CliInst>& _instrs) :
+ id(_id),
+ submit_f(_submit_f),
+ server_select_f(_server_select_f),
+ accum_f(_accum_f),
+ instructions(_instrs),
+ service_tracker(),
+ outstanding_ops(0),
+ requests_complete(false)
+ {
+ size_t op_count = 0;
+ for (auto i : instructions) {
+ if (CliOp::req == i.op) {
+ op_count += i.args.req_params.count;
+ }
+ }
+ op_times.reserve(op_count);
+
+ thd_resp = std::thread(&SimulatedClient::run_resp, this);
+ thd_req = std::thread(&SimulatedClient::run_req, this);
+ }
+
+
+ SimulatedClient(ClientId _id,
+ const SubmitFunc& _submit_f,
+ const ServerSelectFunc& _server_select_f,
+ const ClientAccumFunc& _accum_f,
+ uint16_t _ops_to_run,
+ double _iops_goal,
+ uint16_t _outstanding_ops_allowed) :
+ SimulatedClient(_id,
+ _submit_f, _server_select_f, _accum_f,
+ {{req_op, _ops_to_run, _iops_goal, _outstanding_ops_allowed}})
+ {
+ // empty
+ }
+
+
+ SimulatedClient(const SimulatedClient&) = delete;
+ SimulatedClient(SimulatedClient&&) = delete;
+ SimulatedClient& operator=(const SimulatedClient&) = delete;
+ SimulatedClient& operator=(SimulatedClient&&) = delete;
+
+ virtual ~SimulatedClient() {
+ wait_until_done();
+ }
+
+ void receive_response(const TestResponse& resp,
+ const ServerId& server_id,
+ const RespPm& resp_params) {
+ RespGuard g(mtx_resp);
+ resp_queue.push_back(RespQueueItem{resp, server_id, resp_params});
+ cv_resp.notify_one();
+ }
+
+ const std::vector<TimePoint>& get_op_times() const { return op_times; }
+
+ void wait_until_done() {
+ if (thd_req.joinable()) thd_req.join();
+ if (thd_resp.joinable()) thd_resp.join();
+ }
+
+ const Accum& get_accumulator() const { return accumulator; }
+
+ const InternalStats& get_internal_stats() const { return internal_stats; }
+
+ protected:
+
+ void run_req() {
+ size_t ops_count = 0;
+ for (auto i : instructions) {
+ if (CliOp::wait == i.op) {
+ std::this_thread::sleep_for(i.args.wait_time);
+ } else if (CliOp::req == i.op) {
+ Lock l(mtx_req);
+ for (uint64_t o = 0; o < i.args.req_params.count; ++o) {
+ while (outstanding_ops >= i.args.req_params.max_outstanding) {
+ cv_req.wait(l);
+ }
+
+ l.unlock();
+ auto now = std::chrono::steady_clock::now();
+ const ServerId& server = server_select_f(o);
+
+ ReqPm rp =
+ time_stats_w_return<decltype(internal_stats.get_req_params_time),
+ ReqPm>(internal_stats.mtx,
+ internal_stats.get_req_params_time,
+ [&]() -> ReqPm {
+ return service_tracker.get_req_params(server);
+ });
+ count_stats(internal_stats.mtx,
+ internal_stats.get_req_params_count);
+
+ submit_f(server,
+ TestRequest{server, static_cast<uint32_t>(o), 12},
+ id, rp);
+ ++outstanding_ops;
+ l.lock(); // lock for return to top of loop
+
+ auto delay_time = now + i.args.req_params.time_bw_reqs;
+ while (std::chrono::steady_clock::now() < delay_time) {
+ cv_req.wait_until(l, delay_time);
+ } // while
+ } // for
+ ops_count += i.args.req_params.count;
+ } else {
+ assert(false);
+ }
+ } // for loop
+
+ requests_complete = true;
+
+ // all requests made, thread ends
+ }
+
+
+ void run_resp() {
+ std::chrono::milliseconds delay(1000);
+ int op = 0;
+
+ Lock l(mtx_resp);
+
+ // since the following code would otherwise be repeated (except for
+ // the call to notify_one) in the two loops below; let's avoid
+ // repetition and define it once.
+ const auto proc_resp = [this, &op, &l](const bool notify_req_cv) {
+ if (!resp_queue.empty()) {
+ RespQueueItem item = resp_queue.front();
+ resp_queue.pop_front();
+
+ l.unlock();
+
+ // data collection
+
+ op_times.push_back(now());
+ accum_f(accumulator, item.resp_params);
+
+ // processing
+
+#if 0 // not needed
+ TestResponse& resp = item.response;
+#endif
+
+ time_stats(internal_stats.mtx,
+ internal_stats.track_resp_time,
+ [&](){
+ service_tracker.track_resp(item.server_id, item.resp_params);
+ });
+ count_stats(internal_stats.mtx,
+ internal_stats.track_resp_count);
+
+ --outstanding_ops;
+ if (notify_req_cv) {
+ cv_req.notify_one();
+ }
+
+ l.lock();
+ }
+ };
+
+ while(!requests_complete.load()) {
+ while(resp_queue.empty() && !requests_complete.load()) {
+ cv_resp.wait_for(l, delay);
+ }
+ proc_resp(true);
+ }
+
+ while(outstanding_ops.load() > 0) {
+ while(resp_queue.empty() && outstanding_ops.load() > 0) {
+ cv_resp.wait_for(l, delay);
+ }
+ proc_resp(false); // don't call notify_one as all requests are complete
+ }
+
+ // all responses received, thread ends
+ }
+ }; // class SimulatedClient
+
+
+ }; // namespace qos_simulation
+}; // namespace crimson
Code Review / stor4nfv.git / blobdiff