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(const R& request,
 104                        const C& client_id,
 105                        const ReqParams& req_params) {
 106         add_request(RequestRef(new R(request)), client_id, req_params);
 107       }
 108
 109       void add_request(RequestRef&& request,
 110                        const C& client_id,
 111                        const ReqParams& req_params) {
 112         DataGuard g(queue_mtx);
 113
 114 #ifdef PROFILE
 115         add_request_timer.start();
 116 #endif
 117         queue.emplace_back(QRequest{client_id, std::move(request)});
 118
 119         if (Mechanism::push == mechanism) {
 120           schedule_request();
 121         }
 122
 123 #ifdef PROFILE
 124         add_request_timer.stop();
 125 #endif
 126       } // add_request
 127
 128       void request_completed() {
 129         assert(Mechanism::push == mechanism);
 130         DataGuard g(queue_mtx);
 131
 132 #ifdef PROFILE
 133         request_complete_timer.start();
 134 #endif
 135         schedule_request();
 136
 137 #ifdef PROFILE
 138         request_complete_timer.stop();
 139 #endif
 140       } // request_completed
 141
 142       PullReq pull_request() {
 143         assert(Mechanism::pull == mechanism);
 144         PullReq result;
 145         DataGuard g(queue_mtx);
 146
 147 #ifdef PROFILE
 148         pull_request_timer.start();
 149 #endif
 150
 151         if (queue.empty()) {
 152           result.type = PullReq::Type::none;
 153         } else {
 154           auto front = queue.front();
 155           result.type = PullReq::Type::returning;
 156           result.data =
 157             typename PullReq::Retn{front.client, std::move(front.request)};
 158           queue.pop();
 159         }
 160
 161 #ifdef PROFILE
 162         pull_request_timer.stop();
 163 #endif
 164
 165         return result;
 166       }
 167
 168     protected:
 169
 170       // queue_mtx should be held when called; should only be called
 171       // when mechanism is push
 172       void schedule_request() {
 173         if (!queue.empty() && can_handle_f()) {
 174           auto& front = queue.front();
 175           static NullData null_data;
 176           handle_f(front.client, std::move(front.request), null_data);
 177           queue.pop_front();
 178         }
 179       }
 180     };
 181   };
 182 };