Project
Loading...
Searching...
No Matches
ComputingQuotaEvaluator.cxx
Go to the documentation of this file.
1// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
2// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
3// All rights not expressly granted are reserved.
4//
5// This software is distributed under the terms of the GNU General Public
6// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
7//
8// In applying this license CERN does not waive the privileges and immunities
9// granted to it by virtue of its status as an Intergovernmental Organization
10// or submit itself to any jurisdiction.
11
16#include "Framework/Signpost.h"
17#include <Monitoring/Monitoring.h>
18
19#include <vector>
20#include <uv.h>
21#include <cassert>
22#include <fmt/core.h>
23#include <fmt/format.h>
24#include <fmt/ranges.h>
25
27
28namespace o2::framework
29{
30
32 : mRef(ref)
33{
34 auto& state = mRef.get<DeviceState>();
35 // The first offer is valid, but does not contain any resource
36 // so this will only work with some device which does not require
37 // any CPU. Notice this will have troubles if a given DPL process
38 // runs for more than a year.
39 mOffers[0] = {
40 0,
41 0,
42 0,
43 -1,
44 -1,
46 true};
47 mInfos[0] = {
48 uv_now(state.loop),
49 0,
50 0};
51
52 // Creating a timer to check for expired offers
53 mTimer = (uv_timer_t*)malloc(sizeof(uv_timer_t));
54 uv_timer_init(state.loop, mTimer);
55}
56
58 std::vector<int> invalidOffers;
59 std::vector<int> otherUser;
60 std::vector<int> unexpiring;
61 std::vector<int> selectedOffers;
62 std::vector<int> expired;
63};
64
65bool ComputingQuotaEvaluator::selectOffer(int task, ComputingQuotaRequest const& selector, uint64_t now)
66{
67 O2_SIGNPOST_ID_GENERATE(qid, quota);
68
69 auto selectOffer = [&offers = this->mOffers, &infos = this->mInfos, task](int ref, uint64_t now) {
70 auto& selected = offers[ref];
71 auto& info = infos[ref];
72 selected.user = task;
73 if (info.firstUsed == 0) {
74 info.firstUsed = now;
75 }
76 info.lastUsed = now;
77 };
78
79 ComputingQuotaOffer accumulated;
80 static QuotaEvaluatorStats stats;
81
82 stats.invalidOffers.clear();
83 stats.otherUser.clear();
84 stats.unexpiring.clear();
85 stats.selectedOffers.clear();
86 stats.expired.clear();
87
88 auto summarizeWhatHappended = [ref = mRef](bool enough, std::vector<int> const& result, ComputingQuotaOffer const& totalOffer, QuotaEvaluatorStats& stats) -> bool {
89 auto& dpStats = ref.get<DataProcessingStats>();
90 if (result.size() == 1 && result[0] == 0) {
91 // LOG(LOGLEVEL) << "No particular resource was requested, so we schedule task anyways";
92 return enough;
93 }
94 O2_SIGNPOST_ID_GENERATE(sid, quota);
95 if (enough) {
96 O2_SIGNPOST_START(quota, sid, "summary", "%zu offers were selected for a total of: cpu %d, memory %lli, shared memory %lli",
97 result.size(), totalOffer.cpu, totalOffer.memory, totalOffer.sharedMemory);
98 for (auto& offer : result) {
99 // We pretend each offer id is a pointer, to have a unique id.
100 O2_SIGNPOST_ID_FROM_POINTER(oid, quota, (void*)(int64_t)(offer*8));
101 O2_SIGNPOST_START(quota, oid, "offers", "Offer %d has been selected.", offer);
102 }
103 dpStats.updateStats({static_cast<short>(ProcessingStatsId::RESOURCES_SATISFACTORY), DataProcessingStats::Op::Add, 1});
104 } else {
105 O2_SIGNPOST_START(quota, sid, "summary", "Not enough resources to select offers.");
106 dpStats.updateStats({static_cast<short>(ProcessingStatsId::RESOURCES_MISSING), DataProcessingStats::Op::Add, 1});
107 if (result.size()) {
108 dpStats.updateStats({static_cast<short>(ProcessingStatsId::RESOURCES_INSUFFICIENT), DataProcessingStats::Op::Add, 1});
109 }
110 }
111 if (stats.invalidOffers.size()) {
112 O2_SIGNPOST_EVENT_EMIT(quota, sid, "summary", "The following offers were invalid: %s", fmt::format("{}", fmt::join(stats.invalidOffers, ", ")).c_str());
113 }
114 if (stats.otherUser.size()) {
115 O2_SIGNPOST_EVENT_EMIT(quota, sid, "summary", "The following offers were owned by other users: %s", fmt::format("{}", fmt::join(stats.otherUser, ", ")).c_str());
116 }
117 if (stats.expired.size()) {
118 O2_SIGNPOST_EVENT_EMIT(quota, sid, "summary", "The following offers are expired: %s", fmt::format("{}", fmt::join(stats.expired, ", ")).c_str());
119 }
120 if (stats.unexpiring.size() > 1) {
121 O2_SIGNPOST_EVENT_EMIT(quota, sid, "summary", "The following offers will never expire: %s", fmt::format("{}", fmt::join(stats.unexpiring, ", ")).c_str());
122 }
123 O2_SIGNPOST_END(quota, sid, "summary", "Done selecting offers.");
124
125 return enough;
126 };
127
128 bool enough = false;
129 int64_t minValidity = 0;
130
131 for (int i = 0; i != mOffers.size(); ++i) {
132 auto& offer = mOffers[i];
133 auto& info = mInfos[i];
134 if (enough) {
135 break;
136 }
137 // Ignore:
138 // - Invalid offers
139 // - Offers which belong to another task
140 // - Expired offers
141 if (offer.valid == false) {
142 stats.invalidOffers.push_back(i);
143 continue;
144 }
145 if (offer.user != -1 && offer.user != task) {
146 stats.otherUser.push_back(i);
147 continue;
148 }
149 if (offer.runtime < 0) {
150 stats.unexpiring.push_back(i);
151 } else if (offer.runtime + info.received < now) {
152 O2_SIGNPOST_EVENT_EMIT(quota, qid, "select", "Offer %d expired since %llu milliseconds and holds %llu MB",
153 i, now - offer.runtime - info.received, offer.sharedMemory / 1000000);
155 stats.expired.push_back(i);
156 continue;
157 } else {
158 O2_SIGNPOST_EVENT_EMIT(quota, qid, "select", "Offer %d still valid for %llu milliseconds, providing %llu MB",
159 i, offer.runtime + info.received - now, offer.sharedMemory / 1000000);
160 if (minValidity == 0) {
161 minValidity = offer.runtime + info.received - now;
162 }
163 minValidity = std::min(minValidity, (int64_t)(offer.runtime + info.received - now));
164 }
166 assert(offer.sharedMemory >= 0);
167 auto tmp = accumulated;
168 tmp.cpu += offer.cpu;
169 tmp.memory += offer.memory;
170 tmp.sharedMemory += offer.sharedMemory;
171 offer.score = selector(offer, tmp);
172 switch (offer.score) {
174 continue;
176 continue;
177 case OfferScore::More:
178 selectOffer(i, now);
179 accumulated = tmp;
180 stats.selectedOffers.push_back(i);
181 continue;
183 selectOffer(i, now);
184 accumulated = tmp;
185 stats.selectedOffers.push_back(i);
186 enough = true;
187 break;
188 };
189 }
190
191 if (minValidity != 0) {
192 O2_SIGNPOST_EVENT_EMIT(quota, qid, "select", "Next offer to expire in %llu milliseconds", minValidity);
193 uv_timer_start(mTimer, [](uv_timer_t* handle) {
194 O2_SIGNPOST_ID_GENERATE(tid, quota);
195 O2_SIGNPOST_EVENT_EMIT(quota, tid, "select", "Offer should be expired by now, checking again."); }, minValidity + 100, 0);
196 }
197 // If we get here it means we never got enough offers, so we return false.
198 return summarizeWhatHappended(enough, stats.selectedOffers, accumulated, stats);
199}
200
201void ComputingQuotaEvaluator::consume(int id, ComputingQuotaConsumer& consumer, std::function<void(ComputingQuotaOffer const& accumulatedConsumed, ComputingQuotaStats& reportConsumedOffer)>& reportConsumedOffer)
202{
203 // This will report how much of the offers has to be considered consumed.
204 // Notice that actual memory usage might be larger, because we can over
205 // allocate.
206 consumer(id, mOffers, mStats, reportConsumedOffer);
207}
208
210{
211 for (int oi = 0; oi < mOffers.size(); ++oi) {
212 auto& offer = mOffers[oi];
213 if (offer.user != taskId) {
214 continue;
215 }
216 offer.user = -1;
217 // Disposing the offer so that the resource can be recyled.
220 if (oi == 0) {
221 return;
222 }
223 if (offer.valid == false) {
224 continue;
225 }
226 if (offer.sharedMemory <= 0) {
227 O2_SIGNPOST_ID_FROM_POINTER(oid, quota, (void*)(int64_t)(oi*8));
228 O2_SIGNPOST_END(quota, oid, "offers", "Offer %d back to not needed.", oi);
229 offer.valid = false;
230 offer.score = OfferScore::Unneeded;
231 }
232 }
233}
234
236void ComputingQuotaEvaluator::updateOffers(std::vector<ComputingQuotaOffer>& pending, uint64_t now)
237{
238 for (size_t oi = 0; oi < mOffers.size(); oi++) {
239 auto& storeOffer = mOffers[oi];
240 auto& info = mInfos[oi];
241 if (pending.empty()) {
242 return;
243 }
244 if (storeOffer.valid == true) {
245 continue;
246 }
247 info.received = now;
248 auto& offer = pending.back();
249 storeOffer = offer;
250 storeOffer.valid = true;
251 pending.pop_back();
252 }
253}
254
255void ComputingQuotaEvaluator::handleExpired(std::function<void(ComputingQuotaOffer const&, ComputingQuotaStats const& stats)> expirator)
256{
257 static int nothingToDoCount = mExpiredOffers.size();
258 O2_SIGNPOST_ID_GENERATE(qid, quota);
259 if (mExpiredOffers.size()) {
260 O2_SIGNPOST_EVENT_EMIT(quota, qid, "handleExpired", "Handling %zu expired offers", mExpiredOffers.size());
261 nothingToDoCount = 0;
262 } else {
263 if (nothingToDoCount == 0) {
264 nothingToDoCount++;
265 O2_SIGNPOST_EVENT_EMIT(quota, qid, "handleExpired", "No expired offers");
266 }
267 }
270 for (auto& ref : mExpiredOffers) {
271 auto& offer = mOffers[ref.index];
272 O2_SIGNPOST_ID_FROM_POINTER(oid, quota, (void*)(int64_t)(ref.index*8));
273 if (offer.sharedMemory < 0) {
274 O2_SIGNPOST_END(quota, oid, "handleExpired", "Offer %d does not have any more memory. Marking it as invalid.", ref.index);
275 offer.valid = false;
276 offer.score = OfferScore::Unneeded;
277 continue;
278 }
279 // FIXME: offers should go through the driver client, not the monitoring
280 // api.
281 O2_SIGNPOST_END(quota, oid, "handleExpired", "Offer %d expired. Giving back %llu MB and %d cores",
282 ref.index, offer.sharedMemory / 1000000, offer.cpu);
283 assert(offer.sharedMemory >= 0);
284 mStats.totalExpiredBytes += offer.sharedMemory;
286 expirator(offer, mStats);
287 // driverClient.tell("expired shmem {}", offer.sharedMemory);
288 // driverClient.tell("expired cpu {}", offer.cpu);
289 offer.sharedMemory = -1;
290 offer.valid = false;
291 offer.score = OfferScore::Unneeded;
292 }
293 mExpiredOffers.clear();
294}
295
296} // namespace o2::framework
benchmark::State & state
struct uv_timer_s uv_timer_t
int32_t i
#define O2_DECLARE_DYNAMIC_LOG(name)
Definition Signpost.h:489
#define O2_SIGNPOST_ID_FROM_POINTER(name, log, pointer)
Definition Signpost.h:505
#define O2_SIGNPOST_END(log, id, name, format,...)
Definition Signpost.h:608
#define O2_SIGNPOST_ID_GENERATE(name, log)
Definition Signpost.h:506
#define O2_SIGNPOST_EVENT_EMIT(log, id, name, format,...)
Definition Signpost.h:522
#define O2_SIGNPOST_START(log, id, name, format,...)
Definition Signpost.h:602
void consume(int taskId, ComputingQuotaConsumer &consumed, std::function< void(ComputingQuotaOffer const &accumulatedConsumed, ComputingQuotaStats &)> &reportConsumedOffer)
void updateOffers(std::vector< ComputingQuotaOffer > &offers, uint64_t now)
now the time (e.g. uv_now) when invoked.
std::array< ComputingQuotaInfo, MAX_INFLIGHT_OFFERS > mInfos
Information about a given computing offer (e.g. when it was started to be used)
bool selectOffer(int task, ComputingQuotaRequest const &request, uint64_t now)
void handleExpired(std::function< void(ComputingQuotaOffer const &, ComputingQuotaStats const &)> reportExpired)
void dispose(int taskId)
Dispose offers for a given taskId.
std::array< ComputingQuotaOffer, MAX_INFLIGHT_OFFERS > mOffers
All the available offerts.
std::vector< ComputingQuotaOfferRef > mExpiredOffers
The offers which expired and need to be given back.
GLuint64EXT * result
Definition glcorearb.h:5662
GLint ref
Definition glcorearb.h:291
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
std::function< OfferScore(ComputingQuotaOffer const &offer, ComputingQuotaOffer const &accumulated)> ComputingQuotaRequest
std::function< void(int id, std::array< ComputingQuotaOffer, 16 > &, ComputingQuotaStats &, std::function< void(ComputingQuotaOffer const &, ComputingQuotaStats &stats)>)> ComputingQuotaConsumer
int cpu
How many cores it can use.
Statistics on the offers consumed, expired.
Helper struct to hold statistics about the data processing happening.
@ Add
Update the rate of the metric given the amount since the last time.
Running state information of a given device.
Definition DeviceState.h:34