Project
Loading...
Searching...
No Matches
CCDBHelpers.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
12#include "CCDBHelpers.h"
14#include "Framework/Logger.h"
20#include "CCDB/CcdbApi.h"
22#include "Framework/Signpost.h"
23#include <TError.h>
24#include <TMemFile.h>
25
27
28namespace o2::framework
29{
30
31namespace {
32struct CCDBFetcherHelper {
33 struct CCDBCacheInfo {
34 std::string etag;
35 size_t cacheValidUntil = 0;
36 size_t cachePopulatedAt = 0;
37 size_t cacheMiss = 0;
38 size_t cacheHit = 0;
39 size_t minSize = -1ULL;
40 size_t maxSize = 0;
42 };
43
44 struct RemapMatcher {
45 std::string path;
46 };
47
48 struct RemapTarget {
49 std::string url;
50 };
51
52 std::unordered_map<std::string, CCDBCacheInfo> mapURL2UUID;
53 std::unordered_map<std::string, DataAllocator::CacheId> mapURL2DPLCache;
54 std::string createdNotBefore = "0";
55 std::string createdNotAfter = "3385078236000";
56 std::unordered_map<std::string, o2::ccdb::CcdbApi> apis;
57 std::vector<OutputRoute> routes;
58 std::unordered_map<std::string, std::string> remappings;
59 uint32_t lastCheckedTFCounterOrbReset = 0; // last checkecked TFcounter for bulk check
62 int64_t timeToleranceMS = 5000;
63
64 o2::ccdb::CcdbApi& getAPI(const std::string& path)
65 {
66 // find the first = sign in the string. If present drop everything after it
67 // and between it and the previous /.
68 auto pos = path.find('=');
69 if (pos == std::string::npos) {
70 auto entry = remappings.find(path);
71 return apis[entry == remappings.end() ? "" : entry->second];
72 }
73 auto pos2 = path.rfind('/', pos);
74 if (pos2 == std::string::npos || pos2 == pos - 1 || pos2 == 0) {
75 throw runtime_error_f("Malformed path %s", path.c_str());
76 }
77 auto entry = remappings.find(path.substr(0, pos2));
78 return apis[entry == remappings.end() ? "" : entry->second];
79 }
80};
81}
82
83bool isPrefix(std::string_view prefix, std::string_view full)
84{
85 return prefix == full.substr(0, prefix.size());
86}
87
89{
90 std::unordered_map<std::string, std::string> remappings;
91 std::string currentUrl = "";
92
93 enum ParsingStates {
94 IN_BEGIN,
95 IN_BEGIN_URL,
96 IN_BEGIN_TARGET,
97 IN_END_TARGET,
98 IN_END_URL
99 };
100 ParsingStates state = IN_BEGIN;
101
102 while (true) {
103 switch (state) {
104 case IN_BEGIN: {
105 if (*str == 0) {
106 return {remappings, ""};
107 }
108 state = IN_BEGIN_URL;
109 }
110 case IN_BEGIN_URL: {
111 if ((strncmp("http://", str, 7) != 0) && (strncmp("https://", str, 8) != 0 && (strncmp("file://", str, 7) != 0))) {
112 return {remappings, "URL should start with either http:// or https:// or file://"};
113 }
114 state = IN_END_URL;
115 } break;
116 case IN_END_URL: {
117 char const* c = strchr(str, '=');
118 if (c == nullptr) {
119 return {remappings, "Expecting at least one target path, missing `='?"};
120 }
121 if ((c - str) == 0) {
122 return {remappings, "Empty url"};
123 }
124 currentUrl = std::string_view(str, c - str);
125 state = IN_BEGIN_TARGET;
126 str = c + 1;
127 } break;
128 case IN_BEGIN_TARGET: {
129 if (*str == 0) {
130 return {remappings, "Empty target"};
131 }
132 state = IN_END_TARGET;
133 } break;
134 case IN_END_TARGET: {
135 char const* c = strpbrk(str, ",;");
136 if (c == nullptr) {
137 if (remappings.count(str)) {
138 return {remappings, fmt::format("Path {} requested more than once.", str)};
139 }
140 remappings[std::string(str)] = currentUrl;
141 return {remappings, ""};
142 }
143 if ((c - str) == 0) {
144 return {remappings, "Empty target"};
145 }
146 auto key = std::string(str, c - str);
147 if (remappings.count(str)) {
148 return {remappings, fmt::format("Path {} requested more than once.", key)};
149 }
150 remappings[key] = currentUrl;
151 if (*c == ';') {
152 state = IN_BEGIN_URL;
153 } else {
154 state = IN_BEGIN_TARGET;
155 }
156 str = c + 1;
157 } break;
158 }
159 }
160}
161
162void initialiseHelper(CCDBFetcherHelper& helper, ConfigParamRegistry const& options, std::vector<o2::framework::OutputRoute> const& outputRoutes)
163{
164 std::unordered_map<std::string, bool> accountedSpecs;
165 auto defHost = options.get<std::string>("condition-backend");
166 auto checkRate = options.get<int>("condition-tf-per-query");
167 auto checkMult = options.get<int>("condition-tf-per-query-multiplier");
168 helper.timeToleranceMS = options.get<int64_t>("condition-time-tolerance");
169 helper.queryPeriodGlo = checkRate > 0 ? checkRate : std::numeric_limits<int>::max();
170 helper.queryPeriodFactor = checkMult > 0 ? checkMult : 1;
171 LOGP(info, "CCDB Backend at: {}, validity check for every {} TF{}", defHost, helper.queryPeriodGlo, helper.queryPeriodFactor == 1 ? std::string{} : fmt::format(", (query for high-rate objects downscaled by {})", helper.queryPeriodFactor));
172 LOGP(info, "Hook to enable signposts for CCDB messages at {}", (void*)&private_o2_log_ccdb->stacktrace);
173 auto remapString = options.get<std::string>("condition-remap");
175 if (!result.error.empty()) {
176 throw runtime_error_f("Error while parsing remapping string %s", result.error.c_str());
177 }
178 helper.remappings = result.remappings;
179 helper.apis[""].init(defHost); // default backend
180 LOGP(info, "Initialised default CCDB host {}", defHost);
181 //
182 for (auto& entry : helper.remappings) { // init api instances for every host seen in the remapping
183 if (helper.apis.find(entry.second) == helper.apis.end()) {
184 helper.apis[entry.second].init(entry.second);
185 LOGP(info, "Initialised custom CCDB host {}", entry.second);
186 }
187 LOGP(info, "{} is remapped to {}", entry.first, entry.second);
188 }
189 helper.createdNotBefore = std::to_string(options.get<int64_t>("condition-not-before"));
190 helper.createdNotAfter = std::to_string(options.get<int64_t>("condition-not-after"));
191
192 for (auto& route : outputRoutes) {
193 if (route.matcher.lifetime != Lifetime::Condition) {
194 continue;
195 }
196 auto specStr = DataSpecUtils::describe(route.matcher);
197 if (accountedSpecs.find(specStr) != accountedSpecs.end()) {
198 continue;
199 }
200 accountedSpecs[specStr] = true;
201 helper.routes.push_back(route);
202 LOGP(info, "The following route is a condition {}", DataSpecUtils::describe(route.matcher));
203 for (auto& metadata : route.matcher.metadata) {
204 if (metadata.type == VariantType::String) {
205 LOGP(info, "- {}: {}", metadata.name, metadata.defaultValue.asString());
206 }
207 }
208 }
209}
210
212{
213 Int_t previousErrorLevel = gErrorIgnoreLevel;
214 gErrorIgnoreLevel = kFatal;
215 TMemFile memFile("name", const_cast<char*>(v.data()), v.size(), "READ");
216 gErrorIgnoreLevel = previousErrorLevel;
217 if (memFile.IsZombie()) {
218 throw runtime_error("CTP is Zombie");
219 }
220 TClass* tcl = TClass::GetClass(typeid(std::vector<Long64_t>));
221 void* result = ccdb::CcdbApi::extractFromTFile(memFile, tcl);
222 if (!result) {
223 throw runtime_error_f("Couldn't retrieve object corresponding to %s from TFile", tcl->GetName());
224 }
225 memFile.Close();
226 auto* ctp = (std::vector<Long64_t>*)result;
227 return (*ctp)[0];
228};
229
234
235auto populateCacheWith(std::shared_ptr<CCDBFetcherHelper> const& helper,
236 int64_t timestamp,
237 TimingInfo& timingInfo,
239 DataAllocator& allocator) -> void
240{
241 std::string ccdbMetadataPrefix = "ccdb-metadata-";
242 int objCnt = -1;
243 // We use the timeslice, so that we hook into the same interval as the rest of the
244 // callback.
245 static bool isOnline = isOnlineRun(dtc);
246
247 auto sid = _o2_signpost_id_t{(int64_t)timingInfo.timeslice};
248 O2_SIGNPOST_START(ccdb, sid, "populateCacheWith", "Starting to populate cache with CCDB objects");
249 for (auto& route : helper->routes) {
250 int64_t timestampToUse = timestamp;
251 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Fetching object for route %{public}s", DataSpecUtils::describe(route.matcher).data());
252 objCnt++;
253 auto concrete = DataSpecUtils::asConcreteDataMatcher(route.matcher);
254 Output output{concrete.origin, concrete.description, concrete.subSpec};
255 auto&& v = allocator.makeVector<char>(output);
256 std::map<std::string, std::string> metadata;
257 std::map<std::string, std::string> headers;
258 std::string path = "";
259 std::string etag = "";
260 int chRate = helper->queryPeriodGlo;
261 bool checkValidity = false;
262 for (auto& meta : route.matcher.metadata) {
263 if (meta.name == "ccdb-path") {
264 path = meta.defaultValue.get<std::string>();
265 } else if (meta.name == "ccdb-run-dependent" && meta.defaultValue.get<int>() > 0) {
266 if (meta.defaultValue.get<int>() == 1) {
267 metadata["runNumber"] = dtc.runNumber;
268 } else if (meta.defaultValue.get<int>() == 2) {
269 timestampToUse = std::stoi(dtc.runNumber);
270 } else {
271 LOGP(fatal, "Undefined ccdb-run-dependent option {} for spec {}/{}/{}", meta.defaultValue.get<int>(), concrete.origin.as<std::string>(), concrete.description.as<std::string>(), int(concrete.subSpec));
272 }
273 } else if (isPrefix(ccdbMetadataPrefix, meta.name)) {
274 std::string key = meta.name.substr(ccdbMetadataPrefix.size());
275 auto value = meta.defaultValue.get<std::string>();
276 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Adding metadata %{public}s: %{public}s to the request", key.data(), value.data());
277 metadata[key] = value;
278 } else if (meta.name == "ccdb-query-rate") {
279 chRate = meta.defaultValue.get<int>() * helper->queryPeriodFactor;
280 }
281 }
282 const auto url2uuid = helper->mapURL2UUID.find(path);
283 if (url2uuid != helper->mapURL2UUID.end()) {
284 etag = url2uuid->second.etag;
285 // We check validity every chRate timeslices or if the cache is expired
286 uint64_t validUntil = url2uuid->second.cacheValidUntil;
287 // When the cache was populated. If the cache was populated after the timestamp, we need to check validity.
288 uint64_t cachePopulatedAt = url2uuid->second.cachePopulatedAt;
289 // If timestamp is before the time the element was cached or after the claimed validity, we need to check validity, again
290 // when online.
291 bool cacheExpired = (validUntil <= timestampToUse) || (timestamp < cachePopulatedAt);
292 checkValidity = (std::abs(int(timingInfo.tfCounter - url2uuid->second.lastCheckedTF)) >= chRate) && (isOnline || cacheExpired);
293 } else {
294 checkValidity = true; // never skip check if the cache is empty
295 }
296
297 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "checkValidity is %{public}s for tfID %d of %{public}s", checkValidity ? "true" : "false", timingInfo.tfCounter, path.data());
298
299 const auto& api = helper->getAPI(path);
300 if (checkValidity && (!api.isSnapshotMode() || etag.empty())) { // in the snapshot mode the object needs to be fetched only once
301 LOGP(detail, "Loading {} for timestamp {}", path, timestampToUse);
302 api.loadFileToMemory(v, path, metadata, timestampToUse, &headers, etag, helper->createdNotAfter, helper->createdNotBefore);
303 if ((headers.count("Error") != 0) || (etag.empty() && v.empty())) {
304 LOGP(fatal, "Unable to find CCDB object {}/{}", path, timestampToUse);
305 // FIXME: I should send a dummy message.
306 continue;
307 }
308 // printing in case we find a default entry
309 if (headers.find("default") != headers.end()) {
310 LOGP(detail, "******** Default entry used for {} ********", path);
311 }
312 helper->mapURL2UUID[path].lastCheckedTF = timingInfo.tfCounter;
313 if (etag.empty()) {
314 helper->mapURL2UUID[path].etag = headers["ETag"]; // update uuid
315 helper->mapURL2UUID[path].cachePopulatedAt = timestampToUse;
316 helper->mapURL2UUID[path].cacheMiss++;
317 helper->mapURL2UUID[path].minSize = std::min(v.size(), helper->mapURL2UUID[path].minSize);
318 helper->mapURL2UUID[path].maxSize = std::max(v.size(), helper->mapURL2UUID[path].maxSize);
319 api.appendFlatHeader(v, headers);
320 auto cacheId = allocator.adoptContainer(output, std::move(v), DataAllocator::CacheStrategy::Always, header::gSerializationMethodCCDB);
321 helper->mapURL2DPLCache[path] = cacheId;
322 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Caching %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value);
323 continue;
324 }
325 if (v.size()) { // but should be overridden by fresh object
326 // somewhere here pruneFromCache should be called
327 helper->mapURL2UUID[path].etag = headers["ETag"]; // update uuid
328 helper->mapURL2UUID[path].cachePopulatedAt = timestampToUse;
329 helper->mapURL2UUID[path].cacheValidUntil = headers["Cache-Valid-Until"].empty() ? 0 : std::stoul(headers["Cache-Valid-Until"]);
330 helper->mapURL2UUID[path].cacheMiss++;
331 helper->mapURL2UUID[path].minSize = std::min(v.size(), helper->mapURL2UUID[path].minSize);
332 helper->mapURL2UUID[path].maxSize = std::max(v.size(), helper->mapURL2UUID[path].maxSize);
333 api.appendFlatHeader(v, headers);
334 auto cacheId = allocator.adoptContainer(output, std::move(v), DataAllocator::CacheStrategy::Always, header::gSerializationMethodCCDB);
335 helper->mapURL2DPLCache[path] = cacheId;
336 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Caching %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value);
337 // one could modify the adoptContainer to take optional old cacheID to clean:
338 // mapURL2DPLCache[URL] = ctx.outputs().adoptContainer(output, std::move(outputBuffer), DataAllocator::CacheStrategy::Always, mapURL2DPLCache[URL]);
339 continue;
340 } else {
341 // Only once the etag is actually used, we get the information on how long the object is valid
342 helper->mapURL2UUID[path].cacheValidUntil = headers["Cache-Valid-Until"].empty() ? 0 : std::stoul(headers["Cache-Valid-Until"]);
343 }
344 }
345 // cached object is fine
346 auto cacheId = helper->mapURL2DPLCache[path];
347 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "populateCacheWith", "Reusing %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value);
348 helper->mapURL2UUID[path].cacheHit++;
349 allocator.adoptFromCache(output, cacheId, header::gSerializationMethodCCDB);
350 // the outputBuffer was not used, can we destroy it?
351 }
352 O2_SIGNPOST_END(ccdb, sid, "populateCacheWith", "Finished populating cache with CCDB objects");
353};
354
356{
357 return adaptStateful([](CallbackService& callbacks, ConfigParamRegistry const& options, DeviceSpec const& spec) {
358 std::shared_ptr<CCDBFetcherHelper> helper = std::make_shared<CCDBFetcherHelper>();
359 initialiseHelper(*helper, options, spec.outputs);
362 callbacks.set<CallbackService::Id::Stop>([helper]() {
363 LOGP(info, "CCDB cache miss/hit ratio:");
364 for (auto& entry : helper->mapURL2UUID) {
365 LOGP(info, " {}: {}/{} ({}-{} bytes)", entry.first, entry.second.cacheMiss, entry.second.cacheHit, entry.second.minSize, entry.second.maxSize);
366 }
367 });
368
369 return adaptStateless([helper](DataTakingContext& dtc, DataAllocator& allocator, TimingInfo& timingInfo) {
370 auto sid = _o2_signpost_id_t{(int64_t)timingInfo.timeslice};
371 O2_SIGNPOST_START(ccdb, sid, "fetchFromCCDB", "Fetching CCDB objects for timeslice %" PRIu64, (uint64_t)timingInfo.timeslice);
372 static Long64_t orbitResetTime = -1;
373 static size_t lastTimeUsed = -1;
375 LOGP(info, "Dummy creation time is not supported for CCDB objects. Setting creation to last one used {}.", lastTimeUsed);
376 timingInfo.creation = lastTimeUsed;
377 }
378 lastTimeUsed = timingInfo.creation;
379 // Fetch the CCDB object for the CTP
380 {
381 const std::string path = "CTP/Calib/OrbitReset";
382 std::map<std::string, std::string> metadata;
383 std::map<std::string, std::string> headers;
384 std::string etag;
385 bool checkValidity = std::abs(int(timingInfo.tfCounter - helper->lastCheckedTFCounterOrbReset)) >= helper->queryPeriodGlo;
386 const auto url2uuid = helper->mapURL2UUID.find(path);
387 if (url2uuid != helper->mapURL2UUID.end()) {
388 etag = url2uuid->second.etag;
389 } else {
390 checkValidity = true; // never skip check if the cache is empty
391 }
392 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "fetchFromCCDB", "checkValidity is %{public}s for tfID %d of %{public}s",
393 checkValidity ? "true" : "false", timingInfo.tfCounter, path.data());
394 Output output{"CTP", "OrbitReset", 0};
395 Long64_t newOrbitResetTime = orbitResetTime;
396 auto&& v = allocator.makeVector<char>(output);
397 const auto& api = helper->getAPI(path);
398 if (checkValidity && (!api.isSnapshotMode() || etag.empty())) { // in the snapshot mode the object needs to be fetched only once
399 helper->lastCheckedTFCounterOrbReset = timingInfo.tfCounter;
400 api.loadFileToMemory(v, path, metadata, timingInfo.creation, &headers, etag, helper->createdNotAfter, helper->createdNotBefore);
401 if ((headers.count("Error") != 0) || (etag.empty() && v.empty())) {
402 LOGP(fatal, "Unable to find CCDB object {}/{}", path, timingInfo.creation);
403 // FIXME: I should send a dummy message.
404 return;
405 }
406 if (etag.empty()) {
407 helper->mapURL2UUID[path].etag = headers["ETag"]; // update uuid
408 helper->mapURL2UUID[path].cacheMiss++;
409 helper->mapURL2UUID[path].minSize = std::min(v.size(), helper->mapURL2UUID[path].minSize);
410 helper->mapURL2UUID[path].maxSize = std::max(v.size(), helper->mapURL2UUID[path].maxSize);
411 newOrbitResetTime = getOrbitResetTime(v);
412 api.appendFlatHeader(v, headers);
414 helper->mapURL2DPLCache[path] = cacheId;
415 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "fetchFromCCDB", "Caching %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value);
416 } else if (v.size()) { // but should be overridden by fresh object
417 // somewhere here pruneFromCache should be called
418 helper->mapURL2UUID[path].etag = headers["ETag"]; // update uuid
419 helper->mapURL2UUID[path].cacheMiss++;
420 helper->mapURL2UUID[path].minSize = std::min(v.size(), helper->mapURL2UUID[path].minSize);
421 helper->mapURL2UUID[path].maxSize = std::max(v.size(), helper->mapURL2UUID[path].maxSize);
422 newOrbitResetTime = getOrbitResetTime(v);
423 api.appendFlatHeader(v, headers);
425 helper->mapURL2DPLCache[path] = cacheId;
426 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "fetchFromCCDB", "Caching %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value);
427 // one could modify the adoptContainer to take optional old cacheID to clean:
428 // mapURL2DPLCache[URL] = ctx.outputs().adoptContainer(output, std::move(outputBuffer), DataAllocator::CacheStrategy::Always, mapURL2DPLCache[URL]);
429 }
430 // cached object is fine
431 }
432 auto cacheId = helper->mapURL2DPLCache[path];
433 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "fetchFromCCDB", "Reusing %{public}s for %{public}s (DPL id %" PRIu64 ")", path.data(), headers["ETag"].data(), cacheId.value);
434 helper->mapURL2UUID[path].cacheHit++;
436
437 if (newOrbitResetTime != orbitResetTime) {
438 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "fetchFromCCDB", "Orbit reset time changed from %lld to %lld", orbitResetTime, newOrbitResetTime);
439 orbitResetTime = newOrbitResetTime;
440 dtc.orbitResetTimeMUS = orbitResetTime;
441 }
442 }
443
444 int64_t timestamp = ceil((timingInfo.firstTForbit * o2::constants::lhc::LHCOrbitNS / 1000 + orbitResetTime) / 1000); // RS ceilf precision is not enough
445 if (std::abs(int64_t(timingInfo.creation) - timestamp) > helper->timeToleranceMS) {
446 static bool notWarnedYet = true;
447 if (notWarnedYet) {
448 LOGP(warn, "timestamp {} for orbit {} and orbit reset time {} differs by >{} from the TF creation time {}, use the latter", timestamp, timingInfo.firstTForbit, orbitResetTime / 1000, helper->timeToleranceMS, timingInfo.creation);
449 notWarnedYet = false;
450 // apparently the orbit reset time from the CTP object makes no sense (i.e. orbit was reset for this run w/o create an object, as it happens for technical runs)
451 dtc.orbitResetTimeMUS = 1000 * timingInfo.creation - timingInfo.firstTForbit * o2::constants::lhc::LHCOrbitNS / 1000;
452 }
453 timestamp = timingInfo.creation;
454 }
455 // Fetch the rest of the objects.
456 O2_SIGNPOST_EVENT_EMIT(ccdb, sid, "fetchFromCCDB", "Fetching objects. Run %{public}s. OrbitResetTime %lld. Creation %lld. Timestamp %lld. firstTForbit %" PRIu32,
457 dtc.runNumber.data(), orbitResetTime, timingInfo.creation, timestamp, timingInfo.firstTForbit);
458
459 populateCacheWith(helper, timestamp, timingInfo, dtc, allocator);
460 O2_SIGNPOST_END(ccdb, _o2_signpost_id_t{(int64_t)timingInfo.timeslice}, "fetchFromCCDB", "Fetching CCDB objects");
461 }); });
462}
463
464} // namespace o2::framework
benchmark::State & state
std::unordered_map< std::string, std::string > remappings
size_t cachePopulatedAt
std::string etag
Header to collect LHC related constants.
void output(const std::map< std::string, ChannelStat > &channels)
Definition rawdump.cxx:197
uint16_t pos
Definition RawData.h:3
uint32_t c
Definition RawData.h:2
#define O2_DECLARE_DYNAMIC_LOG(name)
Definition Signpost.h:489
#define O2_SIGNPOST_END(log, id, name, format,...)
Definition Signpost.h:608
#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
StringRef key
static void * extractFromTFile(TFile &file, TClass const *cl, const char *what=CCDBOBJECT_ENTRY)
Definition CcdbApi.cxx:895
o2::pmr::vector< T > makeVector(const Output &spec, Args &&... args)
void adoptFromCache(Output const &spec, CacheId id, header::SerializationMethod method=header::gSerializationMethodNone)
Adopt an already cached message, using an already provided CacheId.
CacheId adoptContainer(const Output &, ContainerT &, CacheStrategy, o2::header::SerializationMethod)
GLuint64EXT * result
Definition glcorearb.h:5662
GLuint entry
Definition glcorearb.h:5735
const GLdouble * v
Definition glcorearb.h:832
GLsizei const GLfloat * value
Definition glcorearb.h:819
GLsizei const GLchar *const * path
Definition glcorearb.h:3591
constexpr double LHCOrbitNS
Defining PrimaryVertex explicitly as messageable.
Definition TFIDInfo.h:20
RuntimeErrorRef runtime_error(const char *)
auto populateCacheWith(std::shared_ptr< CCDBFetcherHelper > const &helper, int64_t timestamp, TimingInfo &timingInfo, DataTakingContext &dtc, DataAllocator &allocator) -> void
bool isOnlineRun(DataTakingContext const &dtc)
bool isPrefix(std::string_view prefix, std::string_view full)
AlgorithmSpec::ProcessCallback adaptStateless(LAMBDA l)
RuntimeErrorRef runtime_error_f(const char *,...)
auto getOrbitResetTime(o2::pmr::vector< char > const &v) -> Long64_t
void initialiseHelper(CCDBFetcherHelper &helper, ConfigParamRegistry const &options, std::vector< o2::framework::OutputRoute > const &outputRoutes)
AlgorithmSpec::InitCallback adaptStateful(LAMBDA l)
constexpr o2::header::SerializationMethod gSerializationMethodNone
Definition DataHeader.h:327
constexpr o2::header::SerializationMethod gSerializationMethodCCDB
Definition DataHeader.h:329
std::vector< T, fair::mq::pmr::polymorphic_allocator< T > > vector
std::string to_string(gsl::span< T, Size > span)
Definition common.h:52
o2::ccdb::CcdbApi & getAPI(const std::string &path)
std::vector< OutputRoute > routes
std::unordered_map< std::string, std::string > remappings
std::unordered_map< std::string, DataAllocator::CacheId > mapURL2DPLCache
std::unordered_map< std::string, o2::ccdb::CcdbApi > apis
std::unordered_map< std::string, CCDBCacheInfo > mapURL2UUID
static ParserResult parseRemappings(char const *)
static AlgorithmSpec fetchFromCCDB()
static constexpr uint64_t DUMMY_CREATION_TIME_OFFSET
static std::string describe(InputSpec const &spec)
static ConcreteDataMatcher asConcreteDataMatcher(InputSpec const &input)
long orbitResetTimeMUS
The start time of the first orbit in microseconds(!)
DeploymentMode deploymentMode
Where we thing this is running.
std::string runNumber
The current run number.
std::vector< OutputRoute > outputs
Definition DeviceSpec.h:63
header::DataOrigin origin
Definition Output.h:28
uint32_t tfCounter
the orbit the TF begins
Definition TimingInfo.h:32
const std::string str