SERVER-23243 Replace Listener::getElapsedTimeMillis() in dur_journal.cpp

8 files changed, 56 insertions, 32 deletions

diff --git a/src/mongo/db/storage/mmap_v1/dur.cpp b/src/mongo/db/storage/mmap_v1/dur.cpp
index fee1652b440..0bfb6d44e45 100644
--- a/src/mongo/db/storage/mmap_v1/dur.cpp
+++ b/src/mongo/db/storage/mmap_v1/dur.cpp
@@ -93,6 +93,7 @@
 #include "mongo/stdx/condition_variable.h"
 #include "mongo/stdx/mutex.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/util/clock_source.h"
 #include "mongo/util/concurrency/synchronization.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/log.h"
@@ -227,7 +228,7 @@ public:
     virtual void closingFileNotification();
     virtual void commitAndStopDurThread();
 
-    void start();
+    void start(ClockSource* cs, int64_t serverStartMs);
 
 private:
     stdx::thread _durThreadHandle;
@@ -408,14 +409,17 @@ static stdx::mutex journalListenerMutex;
 
 
 // Declared in dur_preplogbuffer.cpp
-void PREPLOGBUFFER(JSectHeader& outHeader, AlignedBuilder& outBuffer);
+void PREPLOGBUFFER(JSectHeader& outHeader,
+                   AlignedBuilder& outBuffer,
+                   ClockSource* cs,
+                   int64_t serverStartMs);
 
 // Declared in dur_journal.cpp
 boost::filesystem::path getJournalDir();
 void preallocateFiles();
 
 // Forward declaration
-static void durThread();
+static void durThread(ClockSource* cs, int64_t serverStartMs);
 
 // Durability activity statistics
 Stats stats;
@@ -606,9 +610,9 @@ void DurableImpl::commitAndStopDurThread() {
     _durThreadHandle.join();
 }
 
-void DurableImpl::start() {
+void DurableImpl::start(ClockSource* cs, int64_t serverStartMs) {
     // Start the durability thread
-    stdx::thread t(durThread);
+    stdx::thread t(durThread, cs, serverStartMs);
     _durThreadHandle.swap(t);
 }
 
@@ -654,7 +658,7 @@ static void remapPrivateView(double fraction) {
 /**
  * The main durability thread loop. There is a single instance of this function running.
  */
-static void durThread() {
+static void durThread(ClockSource* cs, int64_t serverStartMs) {
     Client::initThread("durability");
 
     log() << "Durability thread started";
@@ -741,7 +745,7 @@ static void durThread() {
             } else {
                 // This copies all the in-memory changes into the journal writer's buffer.
                 JournalWriter::Buffer* const buffer = journalWriter.newBuffer();
-                PREPLOGBUFFER(buffer->getHeader(), buffer->getBuilder());
+                PREPLOGBUFFER(buffer->getHeader(), buffer->getBuilder(), cs, serverStartMs);
 
                 estimatedPrivateMapSize += commitJob.bytes();
                 commitCounter++;
@@ -867,12 +871,12 @@ static void durThread() {
  * Invoked at server startup. Recovers the database by replaying journal files and then
  * starts the durability thread.
  */
-void startup() {
+void startup(ClockSource* cs, int64_t serverStartMs) {
     if (!storageGlobalParams.dur) {
         return;
     }
 
-    journalMakeDir();
+    journalMakeDir(cs, serverStartMs);
 
     try {
         replayJournalFilesAtStartup();
@@ -889,7 +893,7 @@ void startup() {
 
     preallocateFiles();
 
-    durableImpl.start();
+    durableImpl.start(cs, serverStartMs);
     DurableInterface::_impl = &durableImpl;
 }
 
diff --git a/src/mongo/db/storage/mmap_v1/dur.h b/src/mongo/db/storage/mmap_v1/dur.h
index 43f2ebe7c23..cf5e4ec2b14 100644
--- a/src/mongo/db/storage/mmap_v1/dur.h
+++ b/src/mongo/db/storage/mmap_v1/dur.h
@@ -36,6 +36,7 @@
 
 namespace mongo {
 
+class ClockSource;
 class OperationContext;
 
 namespace dur {
@@ -136,7 +137,7 @@ protected:
     DurableInterface();
 
 private:
-    friend void startup();
+    friend void startup(ClockSource* cs, int64_t serverStartMs);
 
     static DurableInterface* _impl;
 };
@@ -146,7 +147,7 @@ private:
  * Called during startup to startup the durability module.
  * Does nothing if storageGlobalParams.dur is false
  */
-void startup();
+void startup(ClockSource* cs, int64_t serverStartMs);
 
 // Sets a new JournalListener, which is used to alert the rest of the system about
 // journaled write progress.
diff --git a/src/mongo/db/storage/mmap_v1/dur_journal.cpp b/src/mongo/db/storage/mmap_v1/dur_journal.cpp
index 38b0505ba08..ea3b02d9476 100644
--- a/src/mongo/db/storage/mmap_v1/dur_journal.cpp
+++ b/src/mongo/db/storage/mmap_v1/dur_journal.cpp
@@ -52,6 +52,7 @@
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/platform/random.h"
 #include "mongo/util/checksum.h"
+#include "mongo/util/clock_source.h"
 #include "mongo/util/exit.h"
 #include "mongo/util/file.h"
 #include "mongo/util/hex.h"
@@ -193,11 +194,7 @@ Journal j;
 
 const unsigned long long LsnShutdownSentinel = ~((unsigned long long)0);
 
-Journal::Journal() {
-    _written = 0;
-    _nextFileNumber = 0;
-    _curLogFile = 0;
-    _curFileId = 0;
+Journal::Journal() : _written(0), _nextFileNumber(0), _curLogFile(0), _curFileId(0) {
     _lastSeqNumberWrittenToSharedView.store(0);
     _preFlushTime.store(0);
     _lastFlushTime.store(0);
@@ -498,8 +495,8 @@ boost::filesystem::path findPrealloced() {
 }
 
 /** assure journal/ dir exists. throws. call during startup. */
-void journalMakeDir() {
-    j.init();
+void journalMakeDir(ClockSource* cs, int64_t serverStartMs) {
+    j.init(cs, serverStartMs);
 
     boost::filesystem::path p = getJournalDir();
     j.dir = p.string();
@@ -552,8 +549,10 @@ void Journal::_open() {
     }
 }
 
-void Journal::init() {
+void Journal::init(ClockSource* cs, int64_t serverStartMs) {
     verify(_curLogFile == 0);
+    _clock = cs;
+    _serverStartMs = serverStartMs;
 }
 
 void Journal::open() {
@@ -661,8 +660,9 @@ stdx::mutex lastGeneratedSeqNumberMutex;
 uint64_t lastGeneratedSeqNumber = 0;
 }
 
-uint64_t generateNextSeqNumber() {
-    const uint64_t now = Listener::getElapsedTimeMillis();
+uint64_t generateNextSeqNumber(ClockSource* cs, int64_t serverStartMs) {
+    const uint64_t now = cs->now().toMillisSinceEpoch() - serverStartMs;
+
     stdx::lock_guard<stdx::mutex> lock(lastGeneratedSeqNumberMutex);
     if (now > lastGeneratedSeqNumber) {
         lastGeneratedSeqNumber = now;
@@ -693,7 +693,7 @@ void Journal::closeCurrentJournalFile() {
 
     JFile jf;
     jf.filename = _curLogFile->_name;
-    jf.lastEventTimeMs = generateNextSeqNumber();
+    jf.lastEventTimeMs = generateNextSeqNumber(_clock, _serverStartMs);
     _oldJournalFiles.push_back(jf);
 
     delete _curLogFile;  // close
diff --git a/src/mongo/db/storage/mmap_v1/dur_journal.h b/src/mongo/db/storage/mmap_v1/dur_journal.h
index 2477eb846ce..e1da1b65818 100644
--- a/src/mongo/db/storage/mmap_v1/dur_journal.h
+++ b/src/mongo/db/storage/mmap_v1/dur_journal.h
@@ -35,6 +35,7 @@
 namespace mongo {
 
 class AlignedBuilder;
+class ClockSource;
 class JSectHeader;
 
 namespace dur {
@@ -51,13 +52,13 @@ extern bool okToCleanUp;
 void journalCleanup(bool log = false);
 
 /** assure journal/ dir exists. throws */
-void journalMakeDir();
+void journalMakeDir(ClockSource* cs, int64_t serverStartMs);
 
 /**
  * Generates the next sequence number for use in the journal, guaranteed to be greater than all
  * prior sequence numbers.
  */
-uint64_t generateNextSeqNumber();
+uint64_t generateNextSeqNumber(ClockSource* cs, int64_t serverStartMs);
 
 /**
  * Informs the journaling system that all writes on or before the passed in sequence number have
diff --git a/src/mongo/db/storage/mmap_v1/dur_journalimpl.h b/src/mongo/db/storage/mmap_v1/dur_journalimpl.h
index 62d28db7036..5a7a6a63e7f 100644
--- a/src/mongo/db/storage/mmap_v1/dur_journalimpl.h
+++ b/src/mongo/db/storage/mmap_v1/dur_journalimpl.h
@@ -39,6 +39,9 @@
 #include "mongo/util/concurrency/mutex.h"
 
 namespace mongo {
+
+class ClockSource;
+
 namespace dur {
 
 /** the writeahead journal for durability */
@@ -49,7 +52,7 @@ public:
     Journal();
 
     /** call during startup by journalMakeDir() */
-    void init();
+    void init(ClockSource* cs, int64_t serverStartMs);
 
     /** check if time to rotate files.  assure a file is open.
         done separately from the journal() call as we can do this part
@@ -118,6 +121,9 @@ private:
     // data <= this time is fsynced in the datafiles (unless hard drive controller is caching)
     AtomicUInt64 _lastFlushTime;
     AtomicWord<bool> _writeToLSNNeeded;
+
+    ClockSource* _clock;
+    int64_t _serverStartMs;
 };
 }
 }
diff --git a/src/mongo/db/storage/mmap_v1/dur_preplogbuffer.cpp b/src/mongo/db/storage/mmap_v1/dur_preplogbuffer.cpp
index 9a68deb3752..d31b883b9c7 100644
--- a/src/mongo/db/storage/mmap_v1/dur_preplogbuffer.cpp
+++ b/src/mongo/db/storage/mmap_v1/dur_preplogbuffer.cpp
@@ -46,6 +46,7 @@
 #include "mongo/db/storage/mmap_v1/durable_mapped_file.h"
 #include "mongo/db/storage/storage_options.h"
 #include "mongo/stdx/thread.h"
+#include "mongo/util/clock_source.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
 #include "mongo/util/stacktrace.h"
@@ -170,12 +171,15 @@ static void prepBasicWrites(AlignedBuilder& bb, const std::vector<WriteIntent>&
     caller handles locking
     @return partially populated sectheader and _ab set
 */
-static void _PREPLOGBUFFER(JSectHeader& h, AlignedBuilder& bb) {
+static void _PREPLOGBUFFER(JSectHeader& h,
+                           AlignedBuilder& bb,
+                           ClockSource* cs,
+                           int64_t serverStartMs) {
     // Add the JSectHeader
 
     // Invalidate the total length, we will fill it in later.
     h.setSectionLen(0xffffffff);
-    h.seqNumber = generateNextSeqNumber();
+    h.seqNumber = generateNextSeqNumber(cs, serverStartMs);
     h.fileId = j.curFileId();
 
     // Ops other than basic writes (DurOp's) go first
@@ -192,10 +196,13 @@ static void _PREPLOGBUFFER(JSectHeader& h, AlignedBuilder& bb) {
     }
 }
 
-void PREPLOGBUFFER(/*out*/ JSectHeader& outHeader, AlignedBuilder& outBuffer) {
+void PREPLOGBUFFER(/*out*/ JSectHeader& outHeader,
+                   AlignedBuilder& outBuffer,
+                   ClockSource* cs,
+                   int64_t serverStartMs) {
     Timer t;
     j.assureLogFileOpen();  // so fileId is set
-    _PREPLOGBUFFER(outHeader, outBuffer);
+    _PREPLOGBUFFER(outHeader, outBuffer, cs, serverStartMs);
     stats.curr()->_prepLogBufferMicros += t.micros();
 }
 }
diff --git a/src/mongo/db/storage/mmap_v1/mmap_v1_engine.cpp b/src/mongo/db/storage/mmap_v1/mmap_v1_engine.cpp
index 9aa115969b8..b83e72dc580 100644
--- a/src/mongo/db/storage/mmap_v1/mmap_v1_engine.cpp
+++ b/src/mongo/db/storage/mmap_v1/mmap_v1_engine.cpp
@@ -223,7 +223,9 @@ void clearTmpFiles() {
 
 MMAPV1Engine::MMAPV1Engine(const StorageEngineLockFile* lockFile, ClockSource* cs)
     : _recordAccessTracker(cs),
-      _extentManagerFactory(stdx::make_unique<MmapV1ExtentManager::Factory>()) {
+      _extentManagerFactory(stdx::make_unique<MmapV1ExtentManager::Factory>()),
+      _clock(cs),
+      _startMs(_clock->now().toMillisSinceEpoch()) {
     // TODO check non-journal subdirs if using directory-per-db
     checkReadAhead(storageGlobalParams.dbpath);
 
@@ -242,7 +244,7 @@ void MMAPV1Engine::finishInit() {
 
     // Replays the journal (if needed) and starts the background thread. This requires the
     // ability to create OperationContexts.
-    dur::startup();
+    dur::startup(_clock, _startMs);
 }
 
 MMAPV1Engine::~MMAPV1Engine() {
diff --git a/src/mongo/db/storage/mmap_v1/mmap_v1_engine.h b/src/mongo/db/storage/mmap_v1/mmap_v1_engine.h
index c36d06389f2..347b6e02d17 100644
--- a/src/mongo/db/storage/mmap_v1/mmap_v1_engine.h
+++ b/src/mongo/db/storage/mmap_v1/mmap_v1_engine.h
@@ -113,6 +113,9 @@ private:
     RecordAccessTracker _recordAccessTracker;
 
     std::unique_ptr<ExtentManager::Factory> _extentManagerFactory;
+
+    ClockSource* _clock;
+    int64_t _startMs;
 };
 
 void _deleteDataFiles(const std::string& database);