2 * Copyright 2012-present Facebook, Inc.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
19 #include <folly/Likely.h>
20 #include <folly/stats/BucketedTimeSeries.h>
21 #include <glog/logging.h>
27 template <typename VT, typename CT>
28 BucketedTimeSeries<VT, CT>::BucketedTimeSeries(
31 : firstTime_(Duration(1)), latestTime_(), duration_(maxDuration) {
32 // For tracking all-time data we only use total_, and don't need to bother
35 // Round nBuckets down to duration_.count().
37 // There is no point in having more buckets than our timestamp
38 // granularity: otherwise we would have buckets that could never be used.
39 if (nBuckets > size_t(duration_.count())) {
40 nBuckets = size_t(duration_.count());
43 buckets_.resize(nBuckets, Bucket());
47 template <typename VT, typename CT>
48 BucketedTimeSeries<VT, CT>::BucketedTimeSeries(
49 TimePoint theFirstTime,
50 TimePoint theLatestTime,
52 const std::vector<Bucket>& bucketsList)
53 : firstTime_(theFirstTime),
54 latestTime_(theLatestTime),
55 duration_(maxDuration),
56 buckets_(bucketsList) {
57 // Come up with the total_ from buckets_ being passed in
58 for (auto const& bucket : buckets_) {
59 total_.add(bucket.sum, bucket.count);
62 // Verify the integrity of the data
64 // If firstTime is greater than latestTime, the total count should be 0.
65 // (firstTime being greater than latestTime means that no data points have
66 // ever been added to the time series.)
67 if (firstTime_ > latestTime_ && (total_.sum != 0 || total_.count != 0)) {
68 throw std::invalid_argument(
69 "The total should have been 0 "
70 "if firstTime is greater than lastestTime");
73 // If firstTime is less than or equal to latestTime,
74 // latestTime - firstTime should be less than or equal to the duration.
75 if (firstTime_ <= latestTime_ && latestTime_ - firstTime_ > duration_) {
76 throw std::invalid_argument(
77 "The difference between firstTime and latestTime "
78 "should be less than or equal to the duration");
82 template <typename VT, typename CT>
83 bool BucketedTimeSeries<VT, CT>::addValue(TimePoint now, const ValueType& val) {
84 return addValueAggregated(now, val, 1);
87 template <typename VT, typename CT>
88 bool BucketedTimeSeries<VT, CT>::addValue(
92 return addValueAggregated(now, val * ValueType(times), times);
95 template <typename VT, typename CT>
96 bool BucketedTimeSeries<VT, CT>::addValueAggregated(
98 const ValueType& total,
101 if (UNLIKELY(empty())) {
104 } else if (now > latestTime_) {
106 } else if (now < firstTime_) {
109 total_.add(total, nsamples);
114 if (UNLIKELY(empty())) {
115 // First data point we've ever seen
118 bucketIdx = getBucketIdx(now);
119 } else if (now > latestTime_) {
120 // More recent time. Need to update the buckets.
121 bucketIdx = updateBuckets(now);
122 } else if (LIKELY(now == latestTime_)) {
124 bucketIdx = getBucketIdx(now);
126 // An earlier time in the past. We need to check if this time still falls
127 // within our window.
128 if (now < getEarliestTimeNonEmpty()) {
131 bucketIdx = getBucketIdx(now);
134 total_.add(total, nsamples);
135 buckets_[bucketIdx].add(total, nsamples);
139 template <typename VT, typename CT>
140 size_t BucketedTimeSeries<VT, CT>::update(TimePoint now) {
142 // This is the first data point.
146 // For all-time data, all we need to do is update latestTime_
148 latestTime_ = std::max(latestTime_, now);
152 // Make sure time doesn't go backwards.
153 // If the time is less than or equal to the latest time we have already seen,
154 // we don't need to do anything.
155 if (now <= latestTime_) {
156 return getBucketIdx(latestTime_);
159 return updateBuckets(now);
162 template <typename VT, typename CT>
163 size_t BucketedTimeSeries<VT, CT>::updateBuckets(TimePoint now) {
164 // We could cache nextBucketStart as a member variable, so we don't have to
165 // recompute it each time update() is called with a new timestamp value.
166 // This makes things faster when update() (or addValue()) is called once
167 // per second, but slightly slower when update() is called multiple times a
168 // second. We care more about optimizing the cases where addValue() is being
169 // called frequently. If addValue() is only being called once every few
170 // seconds, it doesn't matter as much if it is fast.
172 // Get info about the bucket that latestTime_ points at
173 size_t currentBucket;
174 TimePoint currentBucketStart;
175 TimePoint nextBucketStart;
177 latestTime_, ¤tBucket, ¤tBucketStart, &nextBucketStart);
179 // Update latestTime_
182 if (now < nextBucketStart) {
183 // We're still in the same bucket.
184 // We're done after updating latestTime_.
185 return currentBucket;
186 } else if (now >= currentBucketStart + duration_) {
187 // It's been a while. We have wrapped, and all of the buckets need to be
189 for (Bucket& bucket : buckets_) {
193 return getBucketIdx(latestTime_);
195 // clear all the buckets between the last time and current time, meaning
196 // buckets in the range [(currentBucket+1), newBucket]. Note that
197 // the bucket (currentBucket+1) is always the oldest bucket we have. Since
198 // our array is circular, loop when we reach the end.
199 size_t newBucket = getBucketIdx(now);
200 size_t idx = currentBucket;
201 while (idx != newBucket) {
203 if (idx >= buckets_.size()) {
206 total_ -= buckets_[idx];
207 buckets_[idx].clear();
213 template <typename VT, typename CT>
214 void BucketedTimeSeries<VT, CT>::clear() {
215 for (Bucket& bucket : buckets_) {
219 // Set firstTime_ larger than latestTime_,
220 // to indicate that the timeseries is empty
221 firstTime_ = TimePoint(Duration(1));
222 latestTime_ = TimePoint();
225 template <typename VT, typename CT>
226 typename CT::time_point BucketedTimeSeries<VT, CT>::getEarliestTime() const {
234 // Compute the earliest time we can track
235 TimePoint earliestTime = getEarliestTimeNonEmpty();
237 // We're never tracking data before firstTime_
238 earliestTime = std::max(earliestTime, firstTime_);
243 template <typename VT, typename CT>
244 typename CT::time_point BucketedTimeSeries<VT, CT>::getEarliestTimeNonEmpty()
246 size_t currentBucket;
247 TimePoint currentBucketStart;
248 TimePoint nextBucketStart;
250 latestTime_, ¤tBucket, ¤tBucketStart, &nextBucketStart);
252 // Subtract 1 duration from the start of the next bucket to find the
253 // earliest possible data point we could be tracking.
254 return nextBucketStart - duration_;
257 template <typename VT, typename CT>
258 typename CT::duration BucketedTimeSeries<VT, CT>::elapsed() const {
263 // Add 1 since [latestTime_, earliestTime] is an inclusive interval.
264 return latestTime_ - getEarliestTime() + Duration(1);
267 template <typename VT, typename CT>
268 typename CT::duration BucketedTimeSeries<VT, CT>::elapsed(
270 TimePoint end) const {
274 start = std::max(start, getEarliestTime());
275 end = std::min(end, latestTime_ + Duration(1));
276 end = std::max(start, end);
280 template <typename VT, typename CT>
281 VT BucketedTimeSeries<VT, CT>::sum(TimePoint start, TimePoint end) const {
282 ValueType total = ValueType();
286 [&](const Bucket& bucket,
287 TimePoint bucketStart,
288 TimePoint nextBucketStart) -> bool {
289 total += this->rangeAdjust(
290 bucketStart, nextBucketStart, start, end, bucket.sum);
297 template <typename VT, typename CT>
298 uint64_t BucketedTimeSeries<VT, CT>::count(TimePoint start, TimePoint end)
300 uint64_t sample_count = 0;
304 [&](const Bucket& bucket,
305 TimePoint bucketStart,
306 TimePoint nextBucketStart) -> bool {
307 sample_count += this->rangeAdjust(
308 bucketStart, nextBucketStart, start, end, ValueType(bucket.count));
315 template <typename VT, typename CT>
316 template <typename ReturnType>
317 ReturnType BucketedTimeSeries<VT, CT>::avg(TimePoint start, TimePoint end)
319 ValueType total = ValueType();
320 uint64_t sample_count = 0;
324 [&](const Bucket& bucket,
325 TimePoint bucketStart,
326 TimePoint nextBucketStart) -> bool {
327 total += this->rangeAdjust(
328 bucketStart, nextBucketStart, start, end, bucket.sum);
329 sample_count += this->rangeAdjust(
330 bucketStart, nextBucketStart, start, end, ValueType(bucket.count));
334 if (sample_count == 0) {
335 return ReturnType(0);
338 return detail::avgHelper<ReturnType>(total, sample_count);
342 * A note about some of the bucket index calculations below:
344 * buckets_.size() may not divide evenly into duration_. When this happens,
345 * some buckets will be wider than others. We still want to spread the data
346 * out as evenly as possible among the buckets (as opposed to just making the
347 * last bucket be significantly wider than all of the others).
349 * To make the division work out, we pretend that the buckets are each
350 * duration_ wide, so that the overall duration becomes
351 * buckets.size() * duration_.
353 * To transform a real timestamp into the scale used by our buckets,
354 * we have to multiply by buckets_.size(). To figure out which bucket it goes
355 * into, we then divide by duration_.
358 template <typename VT, typename CT>
359 size_t BucketedTimeSeries<VT, CT>::getBucketIdx(TimePoint time) const {
360 // For all-time data we don't use buckets_. Everything is tracked in total_.
361 DCHECK(!isAllTime());
363 auto timeIntoCurrentCycle = (time.time_since_epoch() % duration_);
364 return timeIntoCurrentCycle.count() * buckets_.size() / duration_.count();
368 * Compute the bucket index for the specified time, as well as the earliest
369 * time that falls into this bucket.
371 template <typename VT, typename CT>
372 void BucketedTimeSeries<VT, CT>::getBucketInfo(
375 TimePoint* bucketStart,
376 TimePoint* nextBucketStart) const {
377 typedef typename Duration::rep TimeInt;
378 DCHECK(!isAllTime());
380 // Keep these two lines together. The compiler should be able to compute
381 // both the division and modulus with a single operation.
382 Duration timeMod = time.time_since_epoch() % duration_;
383 TimeInt numFullDurations = time.time_since_epoch() / duration_;
385 TimeInt scaledTime = timeMod.count() * TimeInt(buckets_.size());
387 // Keep these two lines together. The compiler should be able to compute
388 // both the division and modulus with a single operation.
389 *bucketIdx = size_t(scaledTime / duration_.count());
390 TimeInt scaledOffsetInBucket = scaledTime % duration_.count();
392 TimeInt scaledBucketStart = scaledTime - scaledOffsetInBucket;
393 TimeInt scaledNextBucketStart = scaledBucketStart + duration_.count();
395 Duration bucketStartMod(
396 (scaledBucketStart + buckets_.size() - 1) / buckets_.size());
397 Duration nextBucketStartMod(
398 (scaledNextBucketStart + buckets_.size() - 1) / buckets_.size());
400 TimePoint durationStart(numFullDurations * duration_);
401 *bucketStart = bucketStartMod + durationStart;
402 *nextBucketStart = nextBucketStartMod + durationStart;
405 template <typename VT, typename CT>
406 template <typename Function>
407 void BucketedTimeSeries<VT, CT>::forEachBucket(Function fn) const {
409 fn(total_, firstTime_, latestTime_ + Duration(1));
413 typedef typename Duration::rep TimeInt;
415 // Compute durationStart, latestBucketIdx, and scaledNextBucketStart,
416 // the same way as in getBucketInfo().
417 Duration timeMod = latestTime_.time_since_epoch() % duration_;
418 TimeInt numFullDurations = latestTime_.time_since_epoch() / duration_;
419 TimePoint durationStart(numFullDurations * duration_);
420 TimeInt scaledTime = timeMod.count() * TimeInt(buckets_.size());
421 size_t latestBucketIdx = size_t(scaledTime / duration_.count());
422 TimeInt scaledOffsetInBucket = scaledTime % duration_.count();
423 TimeInt scaledBucketStart = scaledTime - scaledOffsetInBucket;
424 TimeInt scaledNextBucketStart = scaledBucketStart + duration_.count();
426 // Walk through the buckets, starting one past the current bucket.
427 // The next bucket is from the previous cycle, so subtract 1 duration
428 // from durationStart.
429 size_t idx = latestBucketIdx;
430 durationStart -= duration_;
432 TimePoint nextBucketStart =
434 (scaledNextBucketStart + buckets_.size() - 1) / buckets_.size()) +
438 if (idx >= buckets_.size()) {
440 durationStart += duration_;
441 scaledNextBucketStart = duration_.count();
443 scaledNextBucketStart += duration_.count();
446 TimePoint bucketStart = nextBucketStart;
449 (scaledNextBucketStart + buckets_.size() - 1) / buckets_.size()) +
452 // Should we bother skipping buckets where firstTime_ >= nextBucketStart?
453 // For now we go ahead and invoke the function with these buckets.
454 // sum and count should always be 0 in these buckets.
457 bucketStart.time_since_epoch().count(),
458 latestTime_.time_since_epoch().count());
459 bool ret = fn(buckets_[idx], bucketStart, nextBucketStart);
464 if (idx == latestBucketIdx) {
472 * Adjust the input value from the specified bucket to only account
473 * for the desired range.
475 * For example, if the bucket spans time [10, 20), but we only care about the
476 * range [10, 16), this will return 60% of the input value.
478 template <typename VT, typename CT>
479 VT BucketedTimeSeries<VT, CT>::rangeAdjust(
480 TimePoint bucketStart,
481 TimePoint nextBucketStart,
484 ValueType input) const {
485 // If nextBucketStart is greater than latestTime_, treat nextBucketStart as
486 // if it were latestTime_. This makes us more accurate when someone is
487 // querying for all of the data up to latestTime_. Even though latestTime_
488 // may only be partially through the bucket, we don't want to adjust
489 // downwards in this case, because the bucket really only has data up to
491 if (bucketStart <= latestTime_ && nextBucketStart > latestTime_) {
492 nextBucketStart = latestTime_ + Duration(1);
495 if (start <= bucketStart && end >= nextBucketStart) {
496 // The bucket is wholly contained in the [start, end) interval
500 TimePoint intervalStart = std::max(start, bucketStart);
501 TimePoint intervalEnd = std::min(end, nextBucketStart);
502 return input * (intervalEnd - intervalStart) /
503 (nextBucketStart - bucketStart);
506 template <typename VT, typename CT>
507 template <typename Function>
508 void BucketedTimeSeries<VT, CT>::forEachBucket(
514 const Bucket& bucket,
515 TimePoint bucketStart,
516 TimePoint nextBucketStart) -> bool {
517 if (start >= nextBucketStart) {
520 if (end <= bucketStart) {
523 bool ret = fn(bucket, bucketStart, nextBucketStart);