1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
2 // vim: ts=8 sw=2 smarttab
5 * Copyright (C) 2017 Red Hat Inc.
11 /* COMPILATION OPTIONS
13 * By default we include an optimization over the originally published
14 * dmclock algorithm using not the values of rho and delta that were
15 * sent in with a request but instead the most recent rho and delta
16 * values from the requests's client. To restore the algorithm's
17 * original behavior, define DO_NOT_DELAY_TAG_CALC (i.e., compiler
18 * argument -DDO_NOT_DELAY_TAG_CALC).
20 * The prop_heap does not seem to be necessary. The only thing it
21 * would help with is quickly finding the mininum proportion/prioity
22 * when an idle client became active. To have the code maintain the
23 * proportional heap, define USE_PROP_HEAP (i.e., compiler argument
36 #include <condition_variable>
42 #include <boost/variant.hpp>
44 #include "indirect_intrusive_heap.h"
45 #include "run_every.h"
46 #include "dmclock_util.h"
47 #include "dmclock_recs.h"
53 #include "gtest/gtest_prod.h"
60 namespace c
= crimson
;
62 constexpr double max_tag
= std::numeric_limits
<double>::is_iec559
?
63 std::numeric_limits
<double>::infinity() :
64 std::numeric_limits
<double>::max();
65 constexpr double min_tag
= std::numeric_limits
<double>::is_iec559
?
66 -std::numeric_limits
<double>::infinity() :
67 std::numeric_limits
<double>::lowest();
68 constexpr uint tag_modulo
= 1000000;
71 const double reservation
; // minimum
72 const double weight
; // proportional
73 const double limit
; // maximum
75 // multiplicative inverses of above, which we use in calculations
76 // and don't want to recalculate repeatedly
77 const double reservation_inv
;
78 const double weight_inv
;
79 const double limit_inv
;
81 // order parameters -- min, "normal", max
82 ClientInfo(double _reservation
, double _weight
, double _limit
) :
83 reservation(_reservation
),
86 reservation_inv(0.0 == reservation
? 0.0 : 1.0 / reservation
),
87 weight_inv( 0.0 == weight
? 0.0 : 1.0 / weight
),
88 limit_inv( 0.0 == limit
? 0.0 : 1.0 / limit
)
94 friend std::ostream
& operator<<(std::ostream
& out
,
95 const ClientInfo
& client
) {
97 "{ ClientInfo:: r:" << client
.reservation
<<
98 " w:" << std::fixed
<< client
.weight
<<
99 " l:" << std::fixed
<< client
.limit
<<
100 " 1/r:" << std::fixed
<< client
.reservation_inv
<<
101 " 1/w:" << std::fixed
<< client
.weight_inv
<<
102 " 1/l:" << std::fixed
<< client
.limit_inv
<<
106 }; // class ClientInfo
113 bool ready
; // true when within limit
114 #ifndef DO_NOT_DELAY_TAG_CALC
118 RequestTag(const RequestTag
& prev_tag
,
119 const ClientInfo
& client
,
120 const ReqParams
& req_params
,
122 const double cost
= 0.0) :
123 reservation(cost
+ tag_calc(time
,
124 prev_tag
.reservation
,
125 client
.reservation_inv
,
128 proportion(tag_calc(time
,
139 #ifndef DO_NOT_DELAY_TAG_CALC
143 assert(reservation
< max_tag
|| proportion
< max_tag
);
146 RequestTag(double _res
, double _prop
, double _lim
, const Time
& _arrival
) :
151 #ifndef DO_NOT_DELAY_TAG_CALC
155 assert(reservation
< max_tag
|| proportion
< max_tag
);
158 RequestTag(const RequestTag
& other
) :
159 reservation(other
.reservation
),
160 proportion(other
.proportion
),
163 #ifndef DO_NOT_DELAY_TAG_CALC
164 , arrival(other
.arrival
)
170 static std::string
format_tag_change(double before
, double after
) {
171 if (before
== after
) {
172 return std::string("same");
174 std::stringstream ss
;
175 ss
<< format_tag(before
) << "=>" << format_tag(after
);
180 static std::string
format_tag(double value
) {
181 if (max_tag
== value
) {
182 return std::string("max");
183 } else if (min_tag
== value
) {
184 return std::string("min");
186 return format_time(value
, tag_modulo
);
192 static double tag_calc(const Time
& time
,
195 uint32_t dist_req_val
,
196 bool extreme_is_high
) {
197 if (0.0 == increment
) {
198 return extreme_is_high
? max_tag
: min_tag
;
200 if (0 != dist_req_val
) {
201 increment
*= dist_req_val
;
203 return std::max(time
, prev
+ increment
);
207 friend std::ostream
& operator<<(std::ostream
& out
,
208 const RequestTag
& tag
) {
210 "{ RequestTag:: ready:" << (tag
.ready
? "true" : "false") <<
211 " r:" << format_tag(tag
.reservation
) <<
212 " p:" << format_tag(tag
.proportion
) <<
213 " l:" << format_tag(tag
.limit
) <<
214 #if 0 // try to resolve this to make sure Time is operator<<'able.
215 #ifndef DO_NOT_DELAY_TAG_CALC
216 " arrival:" << tag
.arrival
<<
222 }; // class RequestTag
225 // C is client identifier type, R is request type, B is heap
227 template<typename C
, typename R
, uint B
>
228 class PriorityQueueBase
{
229 FRIEND_TEST(dmclock_server
, client_idle_erase
);
233 using RequestRef
= std::unique_ptr
<R
>;
237 using TimePoint
= decltype(std::chrono::steady_clock::now());
238 using Duration
= std::chrono::milliseconds
;
239 using MarkPoint
= std::pair
<TimePoint
,Counter
>;
241 enum class ReadyOption
{ignore
, lowers
, raises
};
243 // forward decl for friend decls
244 template<double RequestTag::*, ReadyOption
, bool>
245 struct ClientCompare
;
248 friend PriorityQueueBase
;
256 ClientReq(const RequestTag
& _tag
,
258 RequestRef
&& _request
) :
260 client_id(_client_id
),
261 request(std::move(_request
))
266 friend std::ostream
& operator<<(std::ostream
& out
, const ClientReq
& c
) {
267 out
<< "{ ClientReq:: tag:" << c
.tag
<< " client:" <<
271 }; // class ClientReq
275 // NOTE: ClientRec is in the "public" section for compatibility
276 // with g++ 4.8.4, which complains if it's not. By g++ 6.3.1
277 // ClientRec could be "protected" with no issue. [See comments
278 // associated with function submit_top_request.]
280 friend PriorityQueueBase
<C
,R
,B
>;
284 std::deque
<ClientReq
> requests
;
286 // amount added from the proportion tag as a result of
287 // an idle client becoming unidle
288 double prop_delta
= 0.0;
290 c::IndIntruHeapData reserv_heap_data
;
291 c::IndIntruHeapData lim_heap_data
;
292 c::IndIntruHeapData ready_heap_data
;
294 c::IndIntruHeapData prop_heap_data
;
306 const ClientInfo
& _info
,
307 Counter current_tick
) :
309 prev_tag(0.0, 0.0, 0.0, TimeZero
),
312 last_tick(current_tick
),
319 inline const RequestTag
& get_req_tag() const {
323 static inline void assign_unpinned_tag(double& lhs
, const double rhs
) {
324 if (rhs
!= max_tag
&& rhs
!= min_tag
) {
329 inline void update_req_tag(const RequestTag
& _prev
,
330 const Counter
& _tick
) {
331 assign_unpinned_tag(prev_tag
.reservation
, _prev
.reservation
);
332 assign_unpinned_tag(prev_tag
.limit
, _prev
.limit
);
333 assign_unpinned_tag(prev_tag
.proportion
, _prev
.proportion
);
337 inline void add_request(const RequestTag
& tag
,
339 RequestRef
&& request
) {
340 requests
.emplace_back(ClientReq(tag
, client_id
, std::move(request
)));
343 inline const ClientReq
& next_request() const {
344 return requests
.front();
347 inline ClientReq
& next_request() {
348 return requests
.front();
351 inline void pop_request() {
352 requests
.pop_front();
355 inline bool has_request() const {
356 return !requests
.empty();
359 inline size_t request_count() const {
360 return requests
.size();
363 // NB: because a deque is the underlying structure, this
364 // operation might be expensive
365 bool remove_by_req_filter_fw(std::function
<bool(const R
&)> filter_accum
) {
366 bool any_removed
= false;
367 for (auto i
= requests
.begin();
370 if (filter_accum(*i
->request
)) {
372 i
= requests
.erase(i
);
380 // NB: because a deque is the underlying structure, this
381 // operation might be expensive
382 bool remove_by_req_filter_bw(std::function
<bool(const R
&)> filter_accum
) {
383 bool any_removed
= false;
384 for (auto i
= requests
.rbegin();
385 i
!= requests
.rend();
387 if (filter_accum(*i
->request
)) {
389 i
= decltype(i
){ requests
.erase(std::next(i
).base()) };
398 remove_by_req_filter(std::function
<bool(const R
&)> filter_accum
,
399 bool visit_backwards
) {
400 if (visit_backwards
) {
401 return remove_by_req_filter_bw(filter_accum
);
403 return remove_by_req_filter_fw(filter_accum
);
408 operator<<(std::ostream
& out
,
409 const typename PriorityQueueBase
<C
,R
,B
>::ClientRec
& e
) {
410 out
<< "{ ClientRec::" <<
411 " client:" << e
.client
<<
412 " prev_tag:" << e
.prev_tag
<<
413 " req_count:" << e
.requests
.size() <<
415 if (e
.has_request()) {
416 out
<< e
.next_request();
424 }; // class ClientRec
426 using ClientRecRef
= std::shared_ptr
<ClientRec
>;
428 // when we try to get the next request, we'll be in one of three
429 // situations -- we'll have one to return, have one that can
430 // fire in the future, or not have any
431 enum class NextReqType
{ returning
, future
, none
};
433 // specifies which queue next request will get popped from
434 enum class HeapId
{ reservation
, ready
};
436 // this is returned from next_req to tell the caller the situation
446 // a function that can be called to look up client information
447 using ClientInfoFunc
= std::function
<ClientInfo(const C
&)>;
451 DataGuard
g(data_mtx
);
452 return (resv_heap
.empty() || ! resv_heap
.top().has_request());
456 size_t client_count() const {
457 DataGuard
g(data_mtx
);
458 return resv_heap
.size();
462 size_t request_count() const {
463 DataGuard
g(data_mtx
);
465 for (auto i
= resv_heap
.cbegin(); i
!= resv_heap
.cend(); ++i
) {
466 total
+= i
->request_count();
472 bool remove_by_req_filter(std::function
<bool(const R
&)> filter_accum
,
473 bool visit_backwards
= false) {
474 bool any_removed
= false;
475 DataGuard
g(data_mtx
);
476 for (auto i
: client_map
) {
478 i
.second
->remove_by_req_filter(filter_accum
, visit_backwards
);
480 resv_heap
.adjust(*i
.second
);
481 limit_heap
.adjust(*i
.second
);
482 ready_heap
.adjust(*i
.second
);
484 prop_heap
.adjust(*i
.second
);
493 // use as a default value when no accumulator is provide
494 static void request_sink(const R
& req
) {
499 void remove_by_client(const C
& client
,
500 bool reverse
= false,
501 std::function
<void (const R
&)> accum
= request_sink
) {
502 DataGuard
g(data_mtx
);
504 auto i
= client_map
.find(client
);
506 if (i
== client_map
.end()) return;
509 for (auto j
= i
->second
->requests
.rbegin();
510 j
!= i
->second
->requests
.rend();
515 for (auto j
= i
->second
->requests
.begin();
516 j
!= i
->second
->requests
.end();
522 i
->second
->requests
.clear();
524 resv_heap
.adjust(*i
->second
);
525 limit_heap
.adjust(*i
->second
);
526 ready_heap
.adjust(*i
->second
);
528 prop_heap
.adjust(*i
->second
);
533 uint
get_heap_branching_factor() const {
538 friend std::ostream
& operator<<(std::ostream
& out
,
539 const PriorityQueueBase
& q
) {
540 std::lock_guard
<decltype(q
.data_mtx
)> guard(q
.data_mtx
);
542 out
<< "{ PriorityQueue::";
543 for (const auto& c
: q
.client_map
) {
544 out
<< " { client:" << c
.first
<< ", record:" << *c
.second
<<
547 if (!q
.resv_heap
.empty()) {
548 const auto& resv
= q
.resv_heap
.top();
549 out
<< " { reservation_top:" << resv
<< " }";
550 const auto& ready
= q
.ready_heap
.top();
551 out
<< " { ready_top:" << ready
<< " }";
552 const auto& limit
= q
.limit_heap
.top();
553 out
<< " { limit_top:" << limit
<< " }";
555 out
<< " HEAPS-EMPTY";
563 void display_queues(std::ostream
& out
,
564 bool show_res
= true,
565 bool show_lim
= true,
566 bool show_ready
= true,
567 bool show_prop
= true) const {
568 auto filter
= [](const ClientRec
& e
)->bool { return true; };
569 DataGuard
g(data_mtx
);
571 resv_heap
.display_sorted(out
<< "RESER:", filter
);
574 limit_heap
.display_sorted(out
<< "LIMIT:", filter
);
577 ready_heap
.display_sorted(out
<< "READY:", filter
);
581 prop_heap
.display_sorted(out
<< "PROPO:", filter
);
589 // The ClientCompare functor is essentially doing a precedes?
590 // operator, returning true if and only if the first parameter
591 // must precede the second parameter. If the second must precede
592 // the first, or if they are equivalent, false should be
593 // returned. The reason for this behavior is that it will be
594 // called to test if two items are out of order and if true is
595 // returned it will reverse the items. Therefore false is the
596 // default return when it doesn't matter to prevent unnecessary
599 // The template is supporting variations in sorting based on the
600 // heap in question and allowing these variations to be handled
603 // tag_field determines which tag is being used for comparison
605 // ready_opt determines how the ready flag influences the sort
607 // use_prop_delta determines whether the proportional delta is
608 // added in for comparison
609 template<double RequestTag::*tag_field
,
610 ReadyOption ready_opt
,
612 struct ClientCompare
{
613 bool operator()(const ClientRec
& n1
, const ClientRec
& n2
) const {
614 if (n1
.has_request()) {
615 if (n2
.has_request()) {
616 const auto& t1
= n1
.next_request().tag
;
617 const auto& t2
= n2
.next_request().tag
;
618 if (ReadyOption::ignore
== ready_opt
|| t1
.ready
== t2
.ready
) {
619 // if we don't care about ready or the ready values are the same
620 if (use_prop_delta
) {
621 return (t1
.*tag_field
+ n1
.prop_delta
) <
622 (t2
.*tag_field
+ n2
.prop_delta
);
624 return t1
.*tag_field
< t2
.*tag_field
;
626 } else if (ReadyOption::raises
== ready_opt
) {
627 // use_ready == true && the ready fields are different
633 // n1 has request but n2 does not
636 } else if (n2
.has_request()) {
637 // n2 has request but n1 does not
640 // both have none; keep stable w false
646 ClientInfoFunc client_info_f
;
648 mutable std::mutex data_mtx
;
649 using DataGuard
= std::lock_guard
<decltype(data_mtx
)>;
651 // stable mapping between client ids and client queues
652 std::map
<C
,ClientRecRef
> client_map
;
654 c::IndIntruHeap
<ClientRecRef
,
656 &ClientRec::reserv_heap_data
,
657 ClientCompare
<&RequestTag::reservation
,
662 c::IndIntruHeap
<ClientRecRef
,
664 &ClientRec::prop_heap_data
,
665 ClientCompare
<&RequestTag::proportion
,
670 c::IndIntruHeap
<ClientRecRef
,
672 &ClientRec::lim_heap_data
,
673 ClientCompare
<&RequestTag::limit
,
677 c::IndIntruHeap
<ClientRecRef
,
679 &ClientRec::ready_heap_data
,
680 ClientCompare
<&RequestTag::proportion
,
685 // if all reservations are met and all other requestes are under
686 // limit, this will allow the request next in terms of
687 // proportion to still get issued
688 bool allow_limit_break
;
690 std::atomic_bool finishing
;
692 // every request creates a tick
695 // performance data collection
696 size_t reserv_sched_count
= 0;
697 size_t prop_sched_count
= 0;
698 size_t limit_break_sched_count
= 0;
703 std::deque
<MarkPoint
> clean_mark_points
;
705 // NB: All threads declared at end, so they're destructed first!
707 std::unique_ptr
<RunEvery
> cleaning_job
;
710 // COMMON constructor that others feed into; we can accept three
711 // different variations of durations
712 template<typename Rep
, typename Per
>
713 PriorityQueueBase(ClientInfoFunc _client_info_f
,
714 std::chrono::duration
<Rep
,Per
> _idle_age
,
715 std::chrono::duration
<Rep
,Per
> _erase_age
,
716 std::chrono::duration
<Rep
,Per
> _check_time
,
717 bool _allow_limit_break
) :
718 client_info_f(_client_info_f
),
719 allow_limit_break(_allow_limit_break
),
721 idle_age(std::chrono::duration_cast
<Duration
>(_idle_age
)),
722 erase_age(std::chrono::duration_cast
<Duration
>(_erase_age
)),
723 check_time(std::chrono::duration_cast
<Duration
>(_check_time
))
725 assert(_erase_age
>= _idle_age
);
726 assert(_check_time
< _idle_age
);
728 std::unique_ptr
<RunEvery
>(
729 new RunEvery(check_time
,
730 std::bind(&PriorityQueueBase::do_clean
, this)));
734 ~PriorityQueueBase() {
739 // data_mtx must be held by caller
740 void do_add_request(RequestRef
&& request
,
742 const ReqParams
& req_params
,
744 const double cost
= 0.0) {
747 // this pointer will help us create a reference to a shared
748 // pointer, no matter which of two codepaths we take
749 ClientRec
* temp_client
;
751 auto client_it
= client_map
.find(client_id
);
752 if (client_map
.end() != client_it
) {
753 temp_client
= &(*client_it
->second
); // address of obj of shared_ptr
755 ClientInfo info
= client_info_f(client_id
);
756 ClientRecRef client_rec
=
757 std::make_shared
<ClientRec
>(client_id
, info
, tick
);
758 resv_heap
.push(client_rec
);
760 prop_heap
.push(client_rec
);
762 limit_heap
.push(client_rec
);
763 ready_heap
.push(client_rec
);
764 client_map
[client_id
] = client_rec
;
765 temp_client
= &(*client_rec
); // address of obj of shared_ptr
768 // for convenience, we'll create a reference to the shared pointer
769 ClientRec
& client
= *temp_client
;
772 // We need to do an adjustment so that idle clients compete
773 // fairly on proportional tags since those tags may have
774 // drifted from real-time. Either use the lowest existing
775 // proportion tag -- O(1) -- or the client with the lowest
776 // previous proportion tag -- O(n) where n = # clients.
778 // So we don't have to maintain a propotional queue that
779 // keeps the minimum on proportional tag alone (we're
780 // instead using a ready queue), we'll have to check each
783 // The alternative would be to maintain a proportional queue
784 // (define USE_PROP_TAG) and do an O(1) operation here.
786 // Was unable to confirm whether equality testing on
787 // std::numeric_limits<double>::max() is guaranteed, so
788 // we'll use a compile-time calculated trigger that is one
789 // third the max, which should be much larger than any
790 // expected organic value.
791 constexpr double lowest_prop_tag_trigger
=
792 std::numeric_limits
<double>::max() / 3.0;
794 double lowest_prop_tag
= std::numeric_limits
<double>::max();
795 for (auto const &c
: client_map
) {
796 // don't use ourselves (or anything else that might be
797 // listed as idle) since we're now in the map
798 if (!c
.second
->idle
) {
800 // use either lowest proportion tag or previous proportion tag
801 if (c
.second
->has_request()) {
802 p
= c
.second
->next_request().tag
.proportion
+
803 c
.second
->prop_delta
;
805 p
= c
.second
->get_req_tag().proportion
+ c
.second
->prop_delta
;
808 if (p
< lowest_prop_tag
) {
814 // if this conditional does not fire, it
815 if (lowest_prop_tag
< lowest_prop_tag_trigger
) {
816 client
.prop_delta
= lowest_prop_tag
- time
;
819 } // if this client was idle
821 #ifndef DO_NOT_DELAY_TAG_CALC
822 RequestTag
tag(0, 0, 0, time
);
824 if (!client
.has_request()) {
825 tag
= RequestTag(client
.get_req_tag(), client
.info
,
826 req_params
, time
, cost
);
828 // copy tag to previous tag for client
829 client
.update_req_tag(tag
, tick
);
832 RequestTag
tag(client
.get_req_tag(), client
.info
, req_params
, time
, cost
);
833 // copy tag to previous tag for client
834 client
.update_req_tag(tag
, tick
);
837 client
.add_request(tag
, client
.client
, std::move(request
));
838 if (1 == client
.requests
.size()) {
839 // NB: can the following 4 calls to adjust be changed
840 // promote? Can adding a request ever demote a client in the
842 resv_heap
.adjust(client
);
843 limit_heap
.adjust(client
);
844 ready_heap
.adjust(client
);
846 prop_heap
.adjust(client
);
850 client
.cur_rho
= req_params
.rho
;
851 client
.cur_delta
= req_params
.delta
;
853 resv_heap
.adjust(client
);
854 limit_heap
.adjust(client
);
855 ready_heap
.adjust(client
);
857 prop_heap
.adjust(client
);
862 // data_mtx should be held when called; top of heap should have
864 template<typename C1
, IndIntruHeapData
ClientRec::*C2
, typename C3
>
865 void pop_process_request(IndIntruHeap
<C1
, ClientRec
, C2
, C3
, B
>& heap
,
866 std::function
<void(const C
& client
,
867 RequestRef
& request
)> process
) {
868 // gain access to data
869 ClientRec
& top
= heap
.top();
870 ClientReq
& first
= top
.next_request();
871 RequestRef request
= std::move(first
.request
);
873 // pop request and adjust heaps
876 #ifndef DO_NOT_DELAY_TAG_CALC
877 if (top
.has_request()) {
878 ClientReq
& next_first
= top
.next_request();
879 next_first
.tag
= RequestTag(first
.tag
, top
.info
,
880 ReqParams(top
.cur_delta
, top
.cur_rho
),
881 next_first
.tag
.arrival
);
883 // copy tag to previous tag for client
884 top
.update_req_tag(next_first
.tag
, tick
);
888 resv_heap
.demote(top
);
889 limit_heap
.adjust(top
);
891 prop_heap
.demote(top
);
893 ready_heap
.demote(top
);
896 process(top
.client
, request
);
897 } // pop_process_request
900 // data_mtx should be held when called
901 void reduce_reservation_tags(ClientRec
& client
) {
902 for (auto& r
: client
.requests
) {
903 r
.tag
.reservation
-= client
.info
.reservation_inv
;
905 #ifndef DO_NOT_DELAY_TAG_CALC
906 // reduce only for front tag. because next tags' value are invalid
910 // don't forget to update previous tag
911 client
.prev_tag
.reservation
-= client
.info
.reservation_inv
;
912 resv_heap
.promote(client
);
916 // data_mtx should be held when called
917 void reduce_reservation_tags(const C
& client_id
) {
918 auto client_it
= client_map
.find(client_id
);
920 // means the client was cleaned from map; should never happen
921 // as long as cleaning times are long enough
922 assert(client_map
.end() != client_it
);
923 reduce_reservation_tags(*client_it
->second
);
927 // data_mtx should be held when called
928 NextReq
do_next_request(Time now
) {
931 // if reservation queue is empty, all are empty (i.e., no active clients)
932 if(resv_heap
.empty()) {
933 result
.type
= NextReqType::none
;
937 // try constraint (reservation) based scheduling
939 auto& reserv
= resv_heap
.top();
940 if (reserv
.has_request() &&
941 reserv
.next_request().tag
.reservation
<= now
) {
942 result
.type
= NextReqType::returning
;
943 result
.heap_id
= HeapId::reservation
;
947 // no existing reservations before now, so try weight-based
950 // all items that are within limit are eligible based on
952 auto limits
= &limit_heap
.top();
953 while (limits
->has_request() &&
954 !limits
->next_request().tag
.ready
&&
955 limits
->next_request().tag
.limit
<= now
) {
956 limits
->next_request().tag
.ready
= true;
957 ready_heap
.promote(*limits
);
958 limit_heap
.demote(*limits
);
960 limits
= &limit_heap
.top();
963 auto& readys
= ready_heap
.top();
964 if (readys
.has_request() &&
965 readys
.next_request().tag
.ready
&&
966 readys
.next_request().tag
.proportion
< max_tag
) {
967 result
.type
= NextReqType::returning
;
968 result
.heap_id
= HeapId::ready
;
972 // if nothing is schedulable by reservation or
973 // proportion/weight, and if we allow limit break, try to
974 // schedule something with the lowest proportion tag or
975 // alternatively lowest reservation tag.
976 if (allow_limit_break
) {
977 if (readys
.has_request() &&
978 readys
.next_request().tag
.proportion
< max_tag
) {
979 result
.type
= NextReqType::returning
;
980 result
.heap_id
= HeapId::ready
;
982 } else if (reserv
.has_request() &&
983 reserv
.next_request().tag
.reservation
< max_tag
) {
984 result
.type
= NextReqType::returning
;
985 result
.heap_id
= HeapId::reservation
;
990 // nothing scheduled; make sure we re-run when next
991 // reservation item or next limited item comes up
993 Time next_call
= TimeMax
;
994 if (resv_heap
.top().has_request()) {
996 min_not_0_time(next_call
,
997 resv_heap
.top().next_request().tag
.reservation
);
999 if (limit_heap
.top().has_request()) {
1000 const auto& next
= limit_heap
.top().next_request();
1001 assert(!next
.tag
.ready
|| max_tag
== next
.tag
.proportion
);
1002 next_call
= min_not_0_time(next_call
, next
.tag
.limit
);
1004 if (next_call
< TimeMax
) {
1005 result
.type
= NextReqType::future
;
1006 result
.when_ready
= next_call
;
1009 result
.type
= NextReqType::none
;
1012 } // do_next_request
1015 // if possible is not zero and less than current then return it;
1016 // otherwise return current; the idea is we're trying to find
1017 // the minimal time but ignoring zero
1018 static inline const Time
& min_not_0_time(const Time
& current
,
1019 const Time
& possible
) {
1020 return TimeZero
== possible
? current
: std::min(current
, possible
);
1025 * This is being called regularly by RunEvery. Every time it's
1026 * called it notes the time and delta counter (mark point) in a
1027 * deque. It also looks at the deque to find the most recent
1028 * mark point that is older than clean_age. It then walks the
1029 * map and delete all server entries that were last used before
1033 TimePoint now
= std::chrono::steady_clock::now();
1034 DataGuard
g(data_mtx
);
1035 clean_mark_points
.emplace_back(MarkPoint(now
, tick
));
1037 // first erase the super-old client records
1039 Counter erase_point
= 0;
1040 auto point
= clean_mark_points
.front();
1041 while (point
.first
<= now
- erase_age
) {
1042 erase_point
= point
.second
;
1043 clean_mark_points
.pop_front();
1044 point
= clean_mark_points
.front();
1047 Counter idle_point
= 0;
1048 for (auto i
: clean_mark_points
) {
1049 if (i
.first
<= now
- idle_age
) {
1050 idle_point
= i
.second
;
1056 if (erase_point
> 0 || idle_point
> 0) {
1057 for (auto i
= client_map
.begin(); i
!= client_map
.end(); /* empty */) {
1059 if (erase_point
&& i2
->second
->last_tick
<= erase_point
) {
1060 delete_from_heaps(i2
->second
);
1061 client_map
.erase(i2
);
1062 } else if (idle_point
&& i2
->second
->last_tick
<= idle_point
) {
1063 i2
->second
->idle
= true;
1070 // data_mtx must be held by caller
1071 template<IndIntruHeapData
ClientRec::*C1
,typename C2
>
1072 void delete_from_heap(ClientRecRef
& client
,
1073 c::IndIntruHeap
<ClientRecRef
,ClientRec
,C1
,C2
,B
>& heap
) {
1074 auto i
= heap
.rfind(client
);
1079 // data_mtx must be held by caller
1080 void delete_from_heaps(ClientRecRef
& client
) {
1081 delete_from_heap(client
, resv_heap
);
1083 delete_from_heap(client
, prop_heap
);
1085 delete_from_heap(client
, limit_heap
);
1086 delete_from_heap(client
, ready_heap
);
1088 }; // class PriorityQueueBase
1091 template<typename C
, typename R
, uint B
=2>
1092 class PullPriorityQueue
: public PriorityQueueBase
<C
,R
,B
> {
1093 using super
= PriorityQueueBase
<C
,R
,B
>;
1097 // When a request is pulled, this is the return type.
1101 typename
super::RequestRef request
;
1105 typename
super::NextReqType type
;
1106 boost::variant
<Retn
,Time
> data
;
1108 bool is_none() const { return type
== super::NextReqType::none
; }
1110 bool is_retn() const { return type
== super::NextReqType::returning
; }
1112 return boost::get
<Retn
>(data
);
1115 bool is_future() const { return type
== super::NextReqType::future
; }
1116 Time
getTime() const { return boost::get
<Time
>(data
); }
1121 ProfileTimer
<std::chrono::nanoseconds
> pull_request_timer
;
1122 ProfileTimer
<std::chrono::nanoseconds
> add_request_timer
;
1125 template<typename Rep
, typename Per
>
1126 PullPriorityQueue(typename
super::ClientInfoFunc _client_info_f
,
1127 std::chrono::duration
<Rep
,Per
> _idle_age
,
1128 std::chrono::duration
<Rep
,Per
> _erase_age
,
1129 std::chrono::duration
<Rep
,Per
> _check_time
,
1130 bool _allow_limit_break
= false) :
1131 super(_client_info_f
,
1132 _idle_age
, _erase_age
, _check_time
,
1139 // pull convenience constructor
1140 PullPriorityQueue(typename
super::ClientInfoFunc _client_info_f
,
1141 bool _allow_limit_break
= false) :
1142 PullPriorityQueue(_client_info_f
,
1143 std::chrono::minutes(10),
1144 std::chrono::minutes(15),
1145 std::chrono::minutes(6),
1152 inline void add_request(const R
& request
,
1154 const ReqParams
& req_params
,
1155 double addl_cost
= 0.0) {
1156 add_request(typename
super::RequestRef(new R(request
)),
1164 inline void add_request(const R
& request
,
1166 double addl_cost
= 0.0) {
1167 static const ReqParams null_req_params
;
1168 add_request(typename
super::RequestRef(new R(request
)),
1177 inline void add_request_time(const R
& request
,
1179 const ReqParams
& req_params
,
1181 double addl_cost
= 0.0) {
1182 add_request(typename
super::RequestRef(new R(request
)),
1190 inline void add_request(typename
super::RequestRef
&& request
,
1192 const ReqParams
& req_params
,
1193 double addl_cost
= 0.0) {
1194 add_request(request
, req_params
, client_id
, get_time(), addl_cost
);
1198 inline void add_request(typename
super::RequestRef
&& request
,
1200 double addl_cost
= 0.0) {
1201 static const ReqParams null_req_params
;
1202 add_request(request
, null_req_params
, client_id
, get_time(), addl_cost
);
1206 // this does the work; the versions above provide alternate interfaces
1207 void add_request(typename
super::RequestRef
&& request
,
1209 const ReqParams
& req_params
,
1211 double addl_cost
= 0.0) {
1212 typename
super::DataGuard
g(this->data_mtx
);
1214 add_request_timer
.start();
1216 super::do_add_request(std::move(request
),
1221 // no call to schedule_request for pull version
1223 add_request_timer
.stop();
1228 inline PullReq
pull_request() {
1229 return pull_request(get_time());
1233 PullReq
pull_request(Time now
) {
1235 typename
super::DataGuard
g(this->data_mtx
);
1237 pull_request_timer
.start();
1240 typename
super::NextReq next
= super::do_next_request(now
);
1241 result
.type
= next
.type
;
1243 case super::NextReqType::none
:
1246 case super::NextReqType::future
:
1247 result
.data
= next
.when_ready
;
1250 case super::NextReqType::returning
:
1251 // to avoid nesting, break out and let code below handle this case
1257 // we'll only get here if we're returning an entry
1260 [&] (PullReq
& pull_result
, PhaseType phase
) ->
1261 std::function
<void(const C
&,
1262 typename
super::RequestRef
&)> {
1263 return [&pull_result
, phase
](const C
& client
,
1264 typename
super::RequestRef
& request
) {
1266 typename
PullReq::Retn
{client
, std::move(request
), phase
};
1270 switch(next
.heap_id
) {
1271 case super::HeapId::reservation
:
1272 super::pop_process_request(this->resv_heap
,
1273 process_f(result
, PhaseType::reservation
));
1274 ++this->reserv_sched_count
;
1276 case super::HeapId::ready
:
1277 super::pop_process_request(this->ready_heap
,
1278 process_f(result
, PhaseType::priority
));
1279 { // need to use retn temporarily
1280 auto& retn
= boost::get
<typename
PullReq::Retn
>(result
.data
);
1281 super::reduce_reservation_tags(retn
.client
);
1283 ++this->prop_sched_count
;
1290 pull_request_timer
.stop();
1299 // data_mtx should be held when called; unfortunately this
1300 // function has to be repeated in both push & pull
1302 typename
super::NextReq
next_request() {
1303 return next_request(get_time());
1305 }; // class PullPriorityQueue
1309 template<typename C
, typename R
, uint B
=2>
1310 class PushPriorityQueue
: public PriorityQueueBase
<C
,R
,B
> {
1314 using super
= PriorityQueueBase
<C
,R
,B
>;
1318 // a function to see whether the server can handle another request
1319 using CanHandleRequestFunc
= std::function
<bool(void)>;
1321 // a function to submit a request to the server; the second
1322 // parameter is a callback when it's completed
1323 using HandleRequestFunc
=
1324 std::function
<void(const C
&,typename
super::RequestRef
,PhaseType
)>;
1328 CanHandleRequestFunc can_handle_f
;
1329 HandleRequestFunc handle_f
;
1330 // for handling timed scheduling
1331 std::mutex sched_ahead_mtx
;
1332 std::condition_variable sched_ahead_cv
;
1333 Time sched_ahead_when
= TimeZero
;
1337 ProfileTimer
<std::chrono::nanoseconds
> add_request_timer
;
1338 ProfileTimer
<std::chrono::nanoseconds
> request_complete_timer
;
1342 // NB: threads declared last, so constructed last and destructed first
1344 std::thread sched_ahead_thd
;
1348 // push full constructor
1349 template<typename Rep
, typename Per
>
1350 PushPriorityQueue(typename
super::ClientInfoFunc _client_info_f
,
1351 CanHandleRequestFunc _can_handle_f
,
1352 HandleRequestFunc _handle_f
,
1353 std::chrono::duration
<Rep
,Per
> _idle_age
,
1354 std::chrono::duration
<Rep
,Per
> _erase_age
,
1355 std::chrono::duration
<Rep
,Per
> _check_time
,
1356 bool _allow_limit_break
= false) :
1357 super(_client_info_f
,
1358 _idle_age
, _erase_age
, _check_time
,
1361 can_handle_f
= _can_handle_f
;
1362 handle_f
= _handle_f
;
1363 sched_ahead_thd
= std::thread(&PushPriorityQueue::run_sched_ahead
, this);
1367 // push convenience constructor
1368 PushPriorityQueue(typename
super::ClientInfoFunc _client_info_f
,
1369 CanHandleRequestFunc _can_handle_f
,
1370 HandleRequestFunc _handle_f
,
1371 bool _allow_limit_break
= false) :
1372 PushPriorityQueue(_client_info_f
,
1375 std::chrono::minutes(10),
1376 std::chrono::minutes(15),
1377 std::chrono::minutes(6),
1384 ~PushPriorityQueue() {
1385 this->finishing
= true;
1386 sched_ahead_cv
.notify_one();
1387 sched_ahead_thd
.join();
1392 inline void add_request(const R
& request
,
1394 const ReqParams
& req_params
,
1395 double addl_cost
= 0.0) {
1396 add_request(typename
super::RequestRef(new R(request
)),
1404 inline void add_request(typename
super::RequestRef
&& request
,
1406 const ReqParams
& req_params
,
1407 double addl_cost
= 0.0) {
1408 add_request(request
, req_params
, client_id
, get_time(), addl_cost
);
1412 inline void add_request_time(const R
& request
,
1414 const ReqParams
& req_params
,
1416 double addl_cost
= 0.0) {
1417 add_request(typename
super::RequestRef(new R(request
)),
1425 void add_request(typename
super::RequestRef
&& request
,
1427 const ReqParams
& req_params
,
1429 double addl_cost
= 0.0) {
1430 typename
super::DataGuard
g(this->data_mtx
);
1432 add_request_timer
.start();
1434 super::do_add_request(std::move(request
),
1441 add_request_timer
.stop();
1446 void request_completed() {
1447 typename
super::DataGuard
g(this->data_mtx
);
1449 request_complete_timer
.start();
1453 request_complete_timer
.stop();
1459 // data_mtx should be held when called; furthermore, the heap
1460 // should not be empty and the top element of the heap should
1461 // not be already handled
1463 // NOTE: the use of "super::ClientRec" in either the template
1464 // construct or as a parameter to submit_top_request generated
1465 // a compiler error in g++ 4.8.4, when ClientRec was
1466 // "protected" rather than "public". By g++ 6.3.1 this was not
1467 // an issue. But for backwards compatibility
1468 // PriorityQueueBase::ClientRec is public.
1469 template<typename C1
,
1470 IndIntruHeapData
super::ClientRec::*C2
,
1473 C
submit_top_request(IndIntruHeap
<C1
,typename
super::ClientRec
,C2
,C3
,B4
>& heap
,
1476 super::pop_process_request(heap
,
1477 [this, phase
, &client_result
]
1479 typename
super::RequestRef
& request
) {
1480 client_result
= client
;
1481 handle_f(client
, std::move(request
), phase
);
1483 return client_result
;
1487 // data_mtx should be held when called
1488 void submit_request(typename
super::HeapId heap_id
) {
1491 case super::HeapId::reservation
:
1492 // don't need to note client
1493 (void) submit_top_request(this->resv_heap
, PhaseType::reservation
);
1494 // unlike the other two cases, we do not reduce reservation
1496 ++this->reserv_sched_count
;
1498 case super::HeapId::ready
:
1499 client
= submit_top_request(this->ready_heap
, PhaseType::priority
);
1500 super::reduce_reservation_tags(client
);
1501 ++this->prop_sched_count
;
1509 // data_mtx should be held when called; unfortunately this
1510 // function has to be repeated in both push & pull
1512 typename
super::NextReq
next_request() {
1513 return next_request(get_time());
1517 // data_mtx should be held when called; overrides member
1518 // function in base class to add check for whether a request can
1519 // be pushed to the server
1520 typename
super::NextReq
next_request(Time now
) {
1521 if (!can_handle_f()) {
1522 typename
super::NextReq result
;
1523 result
.type
= super::NextReqType::none
;
1526 return super::do_next_request(now
);
1531 // data_mtx should be held when called
1532 void schedule_request() {
1533 typename
super::NextReq next_req
= next_request();
1534 switch (next_req
.type
) {
1535 case super::NextReqType::none
:
1537 case super::NextReqType::future
:
1538 sched_at(next_req
.when_ready
);
1540 case super::NextReqType::returning
:
1541 submit_request(next_req
.heap_id
);
1549 // this is the thread that handles running schedule_request at
1550 // future times when nothing can be scheduled immediately
1551 void run_sched_ahead() {
1552 std::unique_lock
<std::mutex
> l(sched_ahead_mtx
);
1554 while (!this->finishing
) {
1555 if (TimeZero
== sched_ahead_when
) {
1556 sched_ahead_cv
.wait(l
);
1559 while (!this->finishing
&& (now
= get_time()) < sched_ahead_when
) {
1560 long microseconds_l
= long(1 + 1000000 * (sched_ahead_when
- now
));
1561 auto microseconds
= std::chrono::microseconds(microseconds_l
);
1562 sched_ahead_cv
.wait_for(l
, microseconds
);
1564 sched_ahead_when
= TimeZero
;
1565 if (this->finishing
) return;
1568 if (!this->finishing
) {
1569 typename
super::DataGuard
g(this->data_mtx
);
1578 void sched_at(Time when
) {
1579 std::lock_guard
<std::mutex
> l(sched_ahead_mtx
);
1580 if (TimeZero
== sched_ahead_when
|| when
< sched_ahead_when
) {
1581 sched_ahead_when
= when
;
1582 sched_ahead_cv
.notify_one();
1585 }; // class PushPriorityQueue
1587 } // namespace dmclock
1588 } // namespace crimson