ceph/src/dmclock/sim/src/ssched/ssched_server.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3
   4 /*
   5  * Copyright (C) 2016 Red Hat Inc.
   6  */
   7
   8 #pragma once
   9
  10 #include <memory>
  11 #include <mutex>
  12 #include <deque>
  13
  14 #include "boost/variant.hpp"
  15
  16 #include "ssched_recs.h"
  17
  18 #ifdef PROFILE
  19 #include "profile.h"
  20 #endif
  21
  22 namespace crimson {
  23
  24   namespace simple_scheduler {
  25
  26     template<typename C, typename R, typename Time>
  27     class SimpleQueue {
  28
  29     public:
  30
  31       using RequestRef = std::unique_ptr<R>;
  32
  33       // a function to see whether the server can handle another request
  34       using CanHandleRequestFunc = std::function<bool(void)>;
  35
  36       // a function to submit a request to the server; the second
  37       // parameter is a callback when it's completed
  38       using HandleRequestFunc =
  39         std::function<void(const C&,RequestRef,NullData)>;
  40
  41       struct PullReq {
  42         enum class Type { returning, none };
  43
  44         struct Retn {
  45           C           client;
  46           RequestRef  request;
  47         };
  48
  49         Type                 type;
  50         boost::variant<Retn> data;
  51       };
  52
  53     protected:
  54
  55       enum class Mechanism { push, pull };
  56
  57       struct QRequest {
  58         C          client;
  59         RequestRef request;
  60       };
  61
  62       bool finishing = false;
  63       Mechanism mechanism;
  64
  65       CanHandleRequestFunc can_handle_f;
  66       HandleRequestFunc handle_f;
  67
  68       mutable std::mutex queue_mtx;
  69       using DataGuard = std::lock_guard<decltype(queue_mtx)>;
  70
  71       std::deque<QRequest> queue;
  72
  73 #ifdef PROFILE
  74     public:
  75       ProfileTimer<std::chrono::nanoseconds> pull_request_timer;
  76       ProfileTimer<std::chrono::nanoseconds> add_request_timer;
  77       ProfileTimer<std::chrono::nanoseconds> request_complete_timer;
  78     protected:
  79 #endif
  80
  81     public:
  82
  83       // push full constructor
  84       SimpleQueue(CanHandleRequestFunc _can_handle_f,
  85                   HandleRequestFunc _handle_f) :
  86         mechanism(Mechanism::push),
  87         can_handle_f(_can_handle_f),
  88         handle_f(_handle_f)
  89       {
  90         // empty
  91       }
  92
  93       SimpleQueue() :
  94         mechanism(Mechanism::pull)
  95       {
  96         // empty
  97       }
  98
  99       ~SimpleQueue() {
 100         finishing = true;
 101       }
 102
 103       void add_request(R&& request,
 104                        const C& client_id,
 105                        const ReqParams& req_params) {
 106         add_request(RequestRef(new R(std::move(request))),
 107                     client_id, req_params);
 108       }
 109
 110       void add_request(RequestRef&& request,
 111                        const C& client_id,
 112                        const ReqParams& req_params) {
 113         DataGuard g(queue_mtx);
 114
 115 #ifdef PROFILE
 116         add_request_timer.start();
 117 #endif
 118         queue.emplace_back(QRequest{client_id, std::move(request)});
 119
 120         if (Mechanism::push == mechanism) {
 121           schedule_request();
 122         }
 123
 124 #ifdef PROFILE
 125         add_request_timer.stop();
 126 #endif
 127       } // add_request
 128
 129       void request_completed() {
 130         assert(Mechanism::push == mechanism);
 131         DataGuard g(queue_mtx);
 132
 133 #ifdef PROFILE
 134         request_complete_timer.start();
 135 #endif
 136         schedule_request();
 137
 138 #ifdef PROFILE
 139         request_complete_timer.stop();
 140 #endif
 141       } // request_completed
 142
 143       PullReq pull_request() {
 144         assert(Mechanism::pull == mechanism);
 145         PullReq result;
 146         DataGuard g(queue_mtx);
 147
 148 #ifdef PROFILE
 149         pull_request_timer.start();
 150 #endif
 151
 152         if (queue.empty()) {
 153           result.type = PullReq::Type::none;
 154         } else {
 155           auto front = queue.front();
 156           result.type = PullReq::Type::returning;
 157           result.data =
 158             typename PullReq::Retn{front.client, std::move(front.request)};
 159           queue.pop();
 160         }
 161
 162 #ifdef PROFILE
 163         pull_request_timer.stop();
 164 #endif
 165
 166         return result;
 167       }
 168
 169     protected:
 170
 171       // queue_mtx should be held when called; should only be called
 172       // when mechanism is push
 173       void schedule_request() {
 174         if (!queue.empty() && can_handle_f()) {
 175           auto& front = queue.front();
 176           static NullData null_data;
 177           handle_f(front.client, std::move(front.request), null_data);
 178           queue.pop_front();
 179         }
 180       }
 181     };
 182   };
 183 };