235 void ValueMetricProducer::onStateChanged(int64_t eventTimeNs, int32_t atomId,  in onStateChanged()  argument
239          (long long)mMetricId, (long long)eventTimeNs, atomId, primaryKey.toString().c_str(),  in onStateChanged()
258     bool isEventLate = eventTimeNs < mCurrentBucketStartTimeNs;  in onStateChanged()
260         VLOG("Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,  in onStateChanged()
262         invalidateCurrentBucket(eventTimeNs, BucketDropReason::EVENT_IN_WRONG_BUCKET);  in onStateChanged()
268         pullAndMatchEventsLocked(eventTimeNs);  in onStateChanged()
272     flushIfNeededLocked(eventTimeNs);  in onStateChanged()
476 void ValueMetricProducer::onActiveStateChangedLocked(const int64_t& eventTimeNs) {  in onActiveStateChangedLocked()  argument
477     bool isEventTooLate  = eventTimeNs < mCurrentBucketStartTimeNs;  in onActiveStateChangedLocked()
481         invalidateCurrentBucket(eventTimeNs, BucketDropReason::EVENT_IN_WRONG_BUCKET);  in onActiveStateChangedLocked()
485     MetricProducer::onActiveStateChangedLocked(eventTimeNs);  in onActiveStateChangedLocked()
494             pullAndMatchEventsLocked(eventTimeNs);  in onActiveStateChangedLocked()
503     flushIfNeededLocked(eventTimeNs);  in onActiveStateChangedLocked()
506     mConditionTimer.onConditionChanged(mIsActive, eventTimeNs);  in onActiveStateChangedLocked()
508     updateCurrentSlicedBucketConditionTimers(mIsActive, eventTimeNs);  in onActiveStateChangedLocked()
512                                                    const int64_t eventTimeNs) {  in onConditionChangedLocked()  argument
514     bool isEventTooLate  = eventTimeNs < mCurrentBucketStartTimeNs;  in onConditionChangedLocked()
524         VLOG("Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,  in onConditionChangedLocked()
528         invalidateCurrentBucket(eventTimeNs, BucketDropReason::EVENT_IN_WRONG_BUCKET);  in onConditionChangedLocked()
530         mConditionTimer.onConditionChanged(mCondition, eventTimeNs);  in onConditionChangedLocked()
532         updateCurrentSlicedBucketConditionTimers(mCondition, eventTimeNs);  in onConditionChangedLocked()
543         invalidateCurrentBucket(eventTimeNs, BucketDropReason::CONDITION_UNKNOWN);  in onConditionChangedLocked()
558         pullAndMatchEventsLocked(eventTimeNs);  in onConditionChangedLocked()
572     flushIfNeededLocked(eventTimeNs);  in onConditionChangedLocked()
573     mConditionTimer.onConditionChanged(mCondition, eventTimeNs);  in onConditionChangedLocked()
575     updateCurrentSlicedBucketConditionTimers(mCondition, eventTimeNs);  in onConditionChangedLocked()
579                                                                    int64_t eventTimeNs) {  in updateCurrentSlicedBucketConditionTimers()  argument
595                 .conditionTimer.onConditionChanged(newCondition && inPulledData, eventTimeNs);  in updateCurrentSlicedBucketConditionTimers()
842     int64_t eventTimeNs = event.GetElapsedTimestampNs();  in onMatchedLogEventInternalLocked()  local
843     if (eventTimeNs < mCurrentBucketStartTimeNs) {  in onMatchedLogEventInternalLocked()
844         VLOG("Skip event due to late arrival: %lld vs %lld", (long long)eventTimeNs,  in onMatchedLogEventInternalLocked()
852         flushIfNeededLocked(eventTimeNs);  in onMatchedLogEventInternalLocked()
1011                 .conditionTimer.onConditionChanged(false, eventTimeNs);  in onMatchedLogEventInternalLocked()
1015                 .conditionTimer.onConditionChanged(true, eventTimeNs);  in onMatchedLogEventInternalLocked()
1020     if (useAnomalyDetection && !multipleBucketsSkipped(calcBucketsForwardCount(eventTimeNs))) {  in onMatchedLogEventInternalLocked()
1028             tracker->detectAndDeclareAnomaly(eventTimeNs, mCurrentBucketNum, mMetricId, eventKey,  in onMatchedLogEventInternalLocked()
1036 void ValueMetricProducer::flushIfNeededLocked(const int64_t& eventTimeNs) {  in flushIfNeededLocked()  argument
1038     if (eventTimeNs < currentBucketEndTimeNs) {  in flushIfNeededLocked()
1039         VLOG("eventTime is %lld, less than current bucket end time %lld", (long long)eventTimeNs,  in flushIfNeededLocked()
1043     int64_t numBucketsForward = calcBucketsForwardCount(eventTimeNs);  in flushIfNeededLocked()
1045     flushCurrentBucketLocked(eventTimeNs, nextBucketStartTimeNs);  in flushIfNeededLocked()
1048 int64_t ValueMetricProducer::calcBucketsForwardCount(const int64_t& eventTimeNs) const {  in calcBucketsForwardCount()
1050     if (eventTimeNs < currentBucketEndTimeNs) {  in calcBucketsForwardCount()
1053     return 1 + (eventTimeNs - currentBucketEndTimeNs) / mBucketSizeNs;  in calcBucketsForwardCount()
1056 void ValueMetricProducer::flushCurrentBucketLocked(const int64_t& eventTimeNs,  in flushCurrentBucketLocked()  argument
1060         invalidateCurrentBucketWithoutResetBase(eventTimeNs, BucketDropReason::CONDITION_UNKNOWN);  in flushCurrentBucketLocked()
1068     int64_t numBucketsForward = calcBucketsForwardCount(eventTimeNs);  in flushCurrentBucketLocked()
1075         invalidateCurrentBucketWithoutResetBase(eventTimeNs,  in flushCurrentBucketLocked()
1079     } else if (eventTimeNs < fullBucketEndTimeNs) {  in flushCurrentBucketLocked()
1080         bucketEndTime = eventTimeNs;  in flushCurrentBucketLocked()
1087         skipCurrentBucket(eventTimeNs, BucketDropReason::BUCKET_TOO_SMALL);  in flushCurrentBucketLocked()
1109             skipCurrentBucket(eventTimeNs, BucketDropReason::NO_DATA);  in flushCurrentBucketLocked()
1128                 buildDropEvent(eventTimeNs, BucketDropReason::NO_DATA));  in flushCurrentBucketLocked()
1131     appendToFullBucket(eventTimeNs > fullBucketEndTimeNs);  in flushCurrentBucketLocked()