Use steady clock for job tracker load sampling.

7 files changed, 65 insertions, 53 deletions

diff --git a/searchcore/src/tests/proton/metrics/job_load_sampler/job_load_sampler_test.cpp b/searchcore/src/tests/proton/metrics/job_load_sampler/job_load_sampler_test.cpp
index b8fa728927f..c27de8cd7a4 100644
--- a/searchcore/src/tests/proton/metrics/job_load_sampler/job_load_sampler_test.cpp
+++ b/searchcore/src/tests/proton/metrics/job_load_sampler/job_load_sampler_test.cpp
@@ -5,28 +5,34 @@ LOG_SETUP("job_load_sampler_test");
 
 #include <vespa/searchcore/proton/metrics/job_load_sampler.h>
 #include <vespa/vespalib/testkit/testapp.h>
+#include <chrono>
 
 using namespace proton;
 
 constexpr double EPS = 0.000001;
 
+std::chrono::time_point<std::chrono::steady_clock> fakeTime(double now)
+{
+    return std::chrono::time_point<std::chrono::steady_clock>(std::chrono::duration_cast<std::chrono::steady_clock::duration>(std::chrono::duration<double>(now)));
+}
+
 struct Fixture
 {
     JobLoadSampler _sampler;
     Fixture()
-        : _sampler(10)
+        : _sampler(fakeTime(10))
     {
     }
     Fixture &start(double now) {
-        _sampler.startJob(now);
+        _sampler.startJob(fakeTime(now));
         return *this;
     }
     Fixture &end(double now) {
-        _sampler.endJob(now);
+        _sampler.endJob(fakeTime(now));
         return *this;
     }
     double sample(double now) {
-        return _sampler.sampleLoad(now);
+        return _sampler.sampleLoad(fakeTime(now));
     }
 };
 
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.cpp b/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.cpp
index cb4d29af497..85819cad53c 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.cpp
+++ b/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.cpp
@@ -14,15 +14,15 @@ namespace proton {
 
 DocumentDBJobTrackers::DocumentDBJobTrackers()
     : _lock(),
-      _now(fastos::ClockSystem::now()),
-      _attributeFlush(new JobTracker(_now.sec(), _lock)),
-      _memoryIndexFlush(new JobTracker(_now.sec(), _lock)),
-      _diskIndexFusion(new JobTracker(_now.sec(), _lock)),
-      _documentStoreFlush(new JobTracker(_now.sec(), _lock)),
-      _documentStoreCompact(new JobTracker(_now.sec(), _lock)),
-      _bucketMove(new JobTracker(_now.sec(), _lock)),
-      _lidSpaceCompact(new JobTracker(_now.sec(), _lock)),
-      _removedDocumentsPrune(new JobTracker(_now.sec(), _lock))
+      _now(std::chrono::steady_clock::now()),
+      _attributeFlush(new JobTracker(_now, _lock)),
+      _memoryIndexFlush(new JobTracker(_now, _lock)),
+      _diskIndexFusion(new JobTracker(_now, _lock)),
+      _documentStoreFlush(new JobTracker(_now, _lock)),
+      _documentStoreCompact(new JobTracker(_now, _lock)),
+      _bucketMove(new JobTracker(_now, _lock)),
+      _lidSpaceCompact(new JobTracker(_now, _lock)),
+      _removedDocumentsPrune(new JobTracker(_now, _lock))
 {
 }
 
@@ -68,10 +68,10 @@ namespace {
 double
 updateMetric(metrics::DoubleAverageMetric &metric,
              JobTracker &tracker,
-             double nowInSec,
-             const vespalib::LockGuard &guard)
+             std::chrono::time_point<std::chrono::steady_clock> now,
+             const std::lock_guard<std::mutex> &guard)
 {
-    double load = tracker.sampleLoad(nowInSec, guard);
+    double load = tracker.sampleLoad(now, guard);
     metric.addValue(load);
     return load;
 }
@@ -81,18 +81,17 @@ updateMetric(metrics::DoubleAverageMetric &metric,
 void
 DocumentDBJobTrackers::updateMetrics(DocumentDBTaggedMetrics::JobMetrics &metrics)
 {
-    vespalib::LockGuard guard(_lock);
-    _now = fastos::ClockSystem::now();
-    double nowInSec = _now.sec();
+    std::lock_guard<std::mutex> guard(_lock);
+    _now = std::chrono::steady_clock::now();
     double load = 0.0;
-    load += updateMetric(metrics.attributeFlush, *_attributeFlush, nowInSec, guard);
-    load += updateMetric(metrics.memoryIndexFlush, *_memoryIndexFlush, nowInSec, guard);
-    load += updateMetric(metrics.diskIndexFusion, *_diskIndexFusion, nowInSec, guard);
-    load += updateMetric(metrics.documentStoreFlush, *_documentStoreFlush, nowInSec, guard);
-    load += updateMetric(metrics.documentStoreCompact, *_documentStoreCompact, nowInSec, guard);
-    load += updateMetric(metrics.bucketMove, *_bucketMove, nowInSec, guard);
-    load += updateMetric(metrics.lidSpaceCompact, *_lidSpaceCompact, nowInSec, guard);
-    load += updateMetric(metrics.removedDocumentsPrune, *_removedDocumentsPrune, nowInSec, guard);
+    load += updateMetric(metrics.attributeFlush, *_attributeFlush, _now, guard);
+    load += updateMetric(metrics.memoryIndexFlush, *_memoryIndexFlush, _now, guard);
+    load += updateMetric(metrics.diskIndexFusion, *_diskIndexFusion, _now, guard);
+    load += updateMetric(metrics.documentStoreFlush, *_documentStoreFlush, _now, guard);
+    load += updateMetric(metrics.documentStoreCompact, *_documentStoreCompact, _now, guard);
+    load += updateMetric(metrics.bucketMove, *_bucketMove, _now, guard);
+    load += updateMetric(metrics.lidSpaceCompact, *_lidSpaceCompact, _now, guard);
+    load += updateMetric(metrics.removedDocumentsPrune, *_removedDocumentsPrune, _now, guard);
     metrics.total.addValue(load);
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.h b/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.h
index 4fd0b20688e..c95cdcf680a 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.h
+++ b/searchcore/src/vespa/searchcore/proton/metrics/documentdb_job_trackers.h
@@ -6,6 +6,8 @@
 #include <vespa/fastos/timestamp.h>
 #include <vespa/searchcorespi/flush/iflushtarget.h>
 #include <vespa/vespalib/util/sync.h>
+#include <chrono>
+#include <mutex>
 
 namespace proton {
 
@@ -16,8 +18,8 @@ namespace proton {
 class DocumentDBJobTrackers
 {
 private:
-    vespalib::Lock    _lock;
-    fastos::TimeStamp _now;
+    std::mutex        _lock;
+    std::chrono::time_point<std::chrono::steady_clock> _now;
     JobTracker::SP    _attributeFlush;
     JobTracker::SP    _memoryIndexFlush;
     JobTracker::SP    _diskIndexFusion;
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.cpp b/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.cpp
index cf70fb8a4a8..caa93e4c43b 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.cpp
+++ b/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.cpp
@@ -8,14 +8,15 @@ LOG_SETUP(".proton.metrics.job_load_sampler");
 namespace proton {
 
 void
-JobLoadSampler::updateIntegral(double now, uint32_t jobCnt)
+JobLoadSampler::updateIntegral(std::chrono::time_point<std::chrono::steady_clock> now, uint32_t jobCnt)
 {
     assert(now >= _lastUpdateTime);
-    _loadIntegral += (now - _lastUpdateTime) * jobCnt;
+    std::chrono::duration<double> duration = now - _lastUpdateTime;
+    _loadIntegral += duration.count() * jobCnt;
     _lastUpdateTime = now;
 }
 
-JobLoadSampler::JobLoadSampler(double now)
+JobLoadSampler::JobLoadSampler(std::chrono::time_point<std::chrono::steady_clock> now)
     : _lastSampleTime(now),
       _lastUpdateTime(now),
       _currJobCnt(0),
@@ -24,23 +25,24 @@ JobLoadSampler::JobLoadSampler(double now)
 }
 
 void
-JobLoadSampler::startJob(double now)
+JobLoadSampler::startJob(std::chrono::time_point<std::chrono::steady_clock> now)
 {
     updateIntegral(now, _currJobCnt++);
 }
 
 void
-JobLoadSampler::endJob(double now)
+JobLoadSampler::endJob(std::chrono::time_point<std::chrono::steady_clock> now)
 {
     updateIntegral(now, _currJobCnt--);
 }
 
 double
-JobLoadSampler::sampleLoad(double now)
+JobLoadSampler::sampleLoad(std::chrono::time_point<std::chrono::steady_clock> now)
 {
     assert(now >= _lastSampleTime);
     updateIntegral(now, _currJobCnt);
-    double load = (now - _lastSampleTime > 0) ? (_loadIntegral / (now - _lastSampleTime)) : 0;
+    std::chrono::duration<double> duration = now - _lastSampleTime;
+    double load = (duration.count() > 0) ? (_loadIntegral / duration.count()) : 0;
     _lastSampleTime = now;
     _loadIntegral = 0;
     return load;
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.h b/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.h
index 6f078a8d9b2..f56c3309ef3 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.h
+++ b/searchcore/src/vespa/searchcore/proton/metrics/job_load_sampler.h
@@ -2,6 +2,8 @@
 
 #pragma once
 
+#include <chrono>
+
 namespace proton {
 
 /**
@@ -13,33 +15,33 @@ namespace proton {
 class JobLoadSampler
 {
 private:
-    double _lastSampleTime;
-    double _lastUpdateTime;
+    std::chrono::time_point<std::chrono::steady_clock> _lastSampleTime;
+    std::chrono::time_point<std::chrono::steady_clock> _lastUpdateTime;
     uint32_t _currJobCnt;
     double _loadIntegral;
 
-    void updateIntegral(double now, uint32_t jobCnt);
+    void updateIntegral(std::chrono::time_point<std::chrono::steady_clock> now, uint32_t jobCnt);
 
 public:
     /**
      * Start the sampler with now (in seconds).
      */
-    JobLoadSampler(double now);
+    JobLoadSampler(std::chrono::time_point<std::chrono::steady_clock> now);
 
     /**
      * Signal that a job starts now (in seconds).
      */
-    void startJob(double now);
+    void startJob(std::chrono::time_point<std::chrono::steady_clock> now);
 
     /**
      * Signal that a job ends now (in seconds).
      */
-    void endJob(double now);
+    void endJob(std::chrono::time_point<std::chrono::steady_clock> now);
 
     /**
      * Samples the average load from previous sample time to now (in seconds).
      */
-    double sampleLoad(double now);
+    double sampleLoad(std::chrono::time_point<std::chrono::steady_clock> now);
 };
 
 } // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.cpp b/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.cpp
index 5dc3acd1f89..a3248b302a1 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.cpp
+++ b/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.cpp
@@ -5,20 +5,21 @@
 LOG_SETUP(".proton.metrics.job_tracker");
 #include "job_tracker.h"
 #include <vespa/fastos/timestamp.h>
+#include <chrono>
 
 using fastos::TimeStamp;
 using fastos::ClockSystem;
 
 namespace proton {
 
-JobTracker::JobTracker(double now, vespalib::Lock &lock)
+JobTracker::JobTracker(std::chrono::time_point<std::chrono::steady_clock> now, std::mutex &lock)
     : _sampler(now),
       _lock(lock)
 {
 }
 
 double
-JobTracker::sampleLoad(double now, const vespalib::LockGuard &guard)
+JobTracker::sampleLoad(std::chrono::time_point<std::chrono::steady_clock> now, const std::lock_guard<std::mutex> &guard)
 {
     (void) guard;
     return _sampler.sampleLoad(now);
@@ -27,15 +28,15 @@ JobTracker::sampleLoad(double now, const vespalib::LockGuard &guard)
 void
 JobTracker::start()
 {
-    vespalib::LockGuard guard(_lock);
-    _sampler.startJob(TimeStamp(ClockSystem::now()).sec());
+    std::unique_lock<std::mutex> guard(_lock);
+    _sampler.startJob(std::chrono::steady_clock::now());
 }
 
 void
 JobTracker::end()
 {
-    vespalib::LockGuard guard(_lock);
-    _sampler.endJob(TimeStamp(ClockSystem::now()).sec());
+    std::unique_lock<std::mutex> guard(_lock);
+    _sampler.endJob(std::chrono::steady_clock::now());
 }
 
 } // namespace proton
diff --git a/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.h b/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.h
index 5d1ba336a71..a3be41ea89d 100644
--- a/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.h
+++ b/searchcore/src/vespa/searchcore/proton/metrics/job_tracker.h
@@ -4,7 +4,7 @@
 
 #include "job_load_sampler.h"
 #include "i_job_tracker.h"
-#include <vespa/vespalib/util/sync.h>
+#include <mutex>
 
 namespace proton {
 
@@ -15,18 +15,18 @@ class JobTracker : public IJobTracker
 {
 private:
     JobLoadSampler  _sampler;
-    vespalib::Lock &_lock;
+    std::mutex     &_lock;
 
 public:
     typedef std::shared_ptr<JobTracker> SP;
 
-    JobTracker(double now, vespalib::Lock &lock);
+    JobTracker(std::chrono::time_point<std::chrono::steady_clock> now, std::mutex &lock);
 
     /**
      * Samples the average job load from previous sample time to now (in seconds).
      * The caller of this function must take the guard on the lock referenced by this class.
      */
-    double sampleLoad(double now, const vespalib::LockGuard &guard);
+    double sampleLoad(std::chrono::time_point<std::chrono::steady_clock> now, const std::lock_guard<std::mutex> &guard);
 
     // Implements IJobTracker
     virtual void start();