diff options
Diffstat (limited to 'src/mongo/db/storage/mmap_v1/dur.cpp')
| -rw-r--r-- | src/mongo/db/storage/mmap_v1/dur.cpp | 1272 |
1 files changed, 620 insertions, 652 deletions
diff --git a/src/mongo/db/storage/mmap_v1/dur.cpp b/src/mongo/db/storage/mmap_v1/dur.cpp index a596bba061f..21c729eea17 100644 --- a/src/mongo/db/storage/mmap_v1/dur.cpp +++ b/src/mongo/db/storage/mmap_v1/dur.cpp @@ -38,15 +38,15 @@ have to handle falling behind which would use too much ram (going back into a read lock would suffice to stop that). for now (1.7.5/1.8.0) we are in read lock which is not ideal. WRITETODATAFILES - actually write to the database data files in this phase. currently done by memcpy'ing the writes back to - the non-private MMF. alternatively one could write to the files the traditional way; however the way our + actually write to the database data files in this phase. currently done by memcpy'ing the writes back to + the non-private MMF. alternatively one could write to the files the traditional way; however the way our storage engine works that isn't any faster (actually measured a tiny bit slower). REMAPPRIVATEVIEW we could in a write lock quickly flip readers back to the main view, then stay in read lock and do our real remapping. with many files (e.g., 1000), remapping could be time consuming (several ms), so we don't want to be too frequent. there could be a slow down immediately after remapping as fresh copy-on-writes for commonly written pages will - be required. so doing these remaps fractionally is helpful. + be required. so doing these remaps fractionally is helpful. mutexes: @@ -99,820 +99,788 @@ namespace mongo { - using std::endl; - using std::fixed; - using std::hex; - using std::set; - using std::setprecision; - using std::setw; - using std::string; - using std::stringstream; +using std::endl; +using std::fixed; +using std::hex; +using std::set; +using std::setprecision; +using std::setw; +using std::string; +using std::stringstream; namespace dur { namespace { - // Used to activate the flush thread - stdx::mutex flushMutex; - stdx::condition_variable flushRequested; +// Used to activate the flush thread +stdx::mutex flushMutex; +stdx::condition_variable flushRequested; - // This is waited on for getlasterror acknowledgements. It means that data has been written to - // the journal, but not necessarily applied to the shared view, so it is all right to - // acknowledge the user operation, but NOT all right to delete the journal files for example. - NotifyAll commitNotify; +// This is waited on for getlasterror acknowledgements. It means that data has been written to +// the journal, but not necessarily applied to the shared view, so it is all right to +// acknowledge the user operation, but NOT all right to delete the journal files for example. +NotifyAll commitNotify; - // This is waited on for complete flush. It means that data has been both written to journal - // and applied to the shared view, so it is allowed to delete the journal files. Used for - // fsync:true, close DB, shutdown acknowledgements. - NotifyAll applyToDataFilesNotify; +// This is waited on for complete flush. It means that data has been both written to journal +// and applied to the shared view, so it is allowed to delete the journal files. Used for +// fsync:true, close DB, shutdown acknowledgements. +NotifyAll applyToDataFilesNotify; - // When set, the flush thread will exit - AtomicUInt32 shutdownRequested(0); +// When set, the flush thread will exit +AtomicUInt32 shutdownRequested(0); - enum { - // How many commit cycles to do before considering doing a remap - NumCommitsBeforeRemap = 10, +enum { + // How many commit cycles to do before considering doing a remap + NumCommitsBeforeRemap = 10, - // How many outstanding journal flushes should be allowed before applying writer back - // pressure. Size of 1 allows two journal blocks to be in the process of being written - - // one on the journal writer's buffer and one blocked waiting to be picked up. - NumAsyncJournalWrites = 1, - }; + // How many outstanding journal flushes should be allowed before applying writer back + // pressure. Size of 1 allows two journal blocks to be in the process of being written - + // one on the journal writer's buffer and one blocked waiting to be picked up. + NumAsyncJournalWrites = 1, +}; - // Remap loop state - unsigned remapFileToStartAt; +// Remap loop state +unsigned remapFileToStartAt; - // How frequently to reset the durability statistics - enum { DurStatsResetIntervalMillis = 3 * 1000 }; +// How frequently to reset the durability statistics +enum { DurStatsResetIntervalMillis = 3 * 1000 }; - // Size sanity checks - BOOST_STATIC_ASSERT(UncommittedBytesLimit > BSONObjMaxInternalSize * 3); - BOOST_STATIC_ASSERT(sizeof(void*) == 4 || UncommittedBytesLimit > BSONObjMaxInternalSize * 6); +// Size sanity checks +BOOST_STATIC_ASSERT(UncommittedBytesLimit > BSONObjMaxInternalSize * 3); +BOOST_STATIC_ASSERT(sizeof(void*) == 4 || UncommittedBytesLimit > BSONObjMaxInternalSize * 6); - /** - * MMAP V1 durability server status section. - */ - class DurSSS : public ServerStatusSection { - public: - DurSSS() : ServerStatusSection("dur") { +/** + * MMAP V1 durability server status section. + */ +class DurSSS : public ServerStatusSection { +public: + DurSSS() : ServerStatusSection("dur") {} - } + virtual bool includeByDefault() const { + return true; + } - virtual bool includeByDefault() const { return true; } + virtual BSONObj generateSection(OperationContext* txn, const BSONElement& configElement) const { + if (!getDur().isDurable()) { + return BSONObj(); + } - virtual BSONObj generateSection(OperationContext* txn, - const BSONElement& configElement) const { + return dur::stats.asObj(); + } - if (!getDur().isDurable()) { - return BSONObj(); - } +} durSSS; - return dur::stats.asObj(); - } - } durSSS; +/** + * A no-op durability interface. Used for the case when journaling is not enabled. + */ +class NonDurableImpl : public DurableInterface { +public: + NonDurableImpl() {} + // DurableInterface virtual methods + virtual void* writingPtr(void* x, unsigned len) { + return x; + } + virtual void declareWriteIntent(void*, unsigned) {} + virtual void declareWriteIntents(const std::vector<std::pair<void*, unsigned>>& intents) {} + virtual void createdFile(const std::string& filename, unsigned long long len) {} + virtual bool waitUntilDurable() { + return false; + } + virtual bool commitNow(OperationContext* txn) { + return false; + } + virtual bool commitIfNeeded() { + return false; + } + virtual void syncDataAndTruncateJournal(OperationContext* txn) {} + virtual bool isDurable() const { + return false; + } + virtual void closingFileNotification() {} + virtual void commitAndStopDurThread() {} +}; - /** - * A no-op durability interface. Used for the case when journaling is not enabled. - */ - class NonDurableImpl : public DurableInterface { - public: - NonDurableImpl() { } - // DurableInterface virtual methods - virtual void* writingPtr(void *x, unsigned len) { return x; } - virtual void declareWriteIntent(void*, unsigned) { } - virtual void declareWriteIntents(const std::vector<std::pair<void*, unsigned> >& intents) { +/** + * The actual durability interface, when journaling is enabled. + */ +class DurableImpl : public DurableInterface { +public: + DurableImpl() {} + + // DurableInterface virtual methods + virtual void declareWriteIntents(const std::vector<std::pair<void*, unsigned>>& intents); + virtual void createdFile(const std::string& filename, unsigned long long len); + virtual bool waitUntilDurable(); + virtual bool commitNow(OperationContext* txn); + virtual bool commitIfNeeded(); + virtual void syncDataAndTruncateJournal(OperationContext* txn); + virtual bool isDurable() const { + return true; + } + virtual void closingFileNotification(); + virtual void commitAndStopDurThread(); - } - virtual void createdFile(const std::string& filename, unsigned long long len) { } - virtual bool waitUntilDurable() { return false; } - virtual bool commitNow(OperationContext* txn) { return false; } - virtual bool commitIfNeeded() { return false; } - virtual void syncDataAndTruncateJournal(OperationContext* txn) {} - virtual bool isDurable() const { return false; } - virtual void closingFileNotification() { } - virtual void commitAndStopDurThread() { } - }; - - - /** - * The actual durability interface, when journaling is enabled. - */ - class DurableImpl : public DurableInterface { - public: - DurableImpl() { } - - // DurableInterface virtual methods - virtual void declareWriteIntents(const std::vector<std::pair<void*, unsigned> >& intents); - virtual void createdFile(const std::string& filename, unsigned long long len); - virtual bool waitUntilDurable(); - virtual bool commitNow(OperationContext* txn); - virtual bool commitIfNeeded(); - virtual void syncDataAndTruncateJournal(OperationContext* txn); - virtual bool isDurable() const { return true; } - virtual void closingFileNotification(); - virtual void commitAndStopDurThread(); - - void start(); - - private: - stdx::thread _durThreadHandle; - }; - - - /** - * Diagnostic to check that the private view and the non-private view are in sync after - * applying the journal changes. This function is very slow and only runs when paranoid checks - * are enabled. - * - * Must be called under at least S flush lock to ensure that there are no concurrent writes - * happening. - */ - void debugValidateFileMapsMatch(const DurableMappedFile* mmf) { - const unsigned char *p = (const unsigned char *)mmf->getView(); - const unsigned char *w = (const unsigned char *)mmf->view_write(); - - // Ignore pre-allocated files that are not fully created yet - if (!p || !w) { - return; - } + void start(); - if (memcmp(p, w, (unsigned)mmf->length()) == 0) { - return; - } +private: + stdx::thread _durThreadHandle; +}; - unsigned low = 0xffffffff; - unsigned high = 0; - log() << "DurParanoid mismatch in " << mmf->filename(); +/** + * Diagnostic to check that the private view and the non-private view are in sync after + * applying the journal changes. This function is very slow and only runs when paranoid checks + * are enabled. + * + * Must be called under at least S flush lock to ensure that there are no concurrent writes + * happening. + */ +void debugValidateFileMapsMatch(const DurableMappedFile* mmf) { + const unsigned char* p = (const unsigned char*)mmf->getView(); + const unsigned char* w = (const unsigned char*)mmf->view_write(); - int logged = 0; - unsigned lastMismatch = 0xffffffff; + // Ignore pre-allocated files that are not fully created yet + if (!p || !w) { + return; + } - for (unsigned i = 0; i < mmf->length(); i++) { - if (p[i] != w[i]) { + if (memcmp(p, w, (unsigned)mmf->length()) == 0) { + return; + } - if (lastMismatch != 0xffffffff && lastMismatch + 1 != i) { - // Separate blocks of mismatches - log() << std::endl; - } + unsigned low = 0xffffffff; + unsigned high = 0; - lastMismatch = i; + log() << "DurParanoid mismatch in " << mmf->filename(); - if (++logged < 60) { - if (logged == 1) { - // For .ns files to find offset in record - log() << "ofs % 628 = 0x" << hex << (i % 628) << endl; - } + int logged = 0; + unsigned lastMismatch = 0xffffffff; - stringstream ss; - ss << "mismatch ofs:" << hex << i - << "\tfilemap:" << setw(2) << (unsigned)w[i] - << "\tprivmap:" << setw(2) << (unsigned)p[i]; + for (unsigned i = 0; i < mmf->length(); i++) { + if (p[i] != w[i]) { + if (lastMismatch != 0xffffffff && lastMismatch + 1 != i) { + // Separate blocks of mismatches + log() << std::endl; + } - if (p[i] > 32 && p[i] <= 126) { - ss << '\t' << p[i]; - } + lastMismatch = i; - log() << ss.str() << endl; + if (++logged < 60) { + if (logged == 1) { + // For .ns files to find offset in record + log() << "ofs % 628 = 0x" << hex << (i % 628) << endl; } - if (logged == 60) { - log() << "..." << endl; + stringstream ss; + ss << "mismatch ofs:" << hex << i << "\tfilemap:" << setw(2) << (unsigned)w[i] + << "\tprivmap:" << setw(2) << (unsigned)p[i]; + + if (p[i] > 32 && p[i] <= 126) { + ss << '\t' << p[i]; } - if (i < low) low = i; - if (i > high) high = i; + log() << ss.str() << endl; } + + if (logged == 60) { + log() << "..." << endl; + } + + if (i < low) + low = i; + if (i > high) + high = i; } + } - if (low != 0xffffffff) { - std::stringstream ss; - ss << "journal error warning views mismatch " << mmf->filename() << ' ' - << hex << low << ".." << high - << " len:" << high - low + 1; + if (low != 0xffffffff) { + std::stringstream ss; + ss << "journal error warning views mismatch " << mmf->filename() << ' ' << hex << low + << ".." << high << " len:" << high - low + 1; - log() << ss.str() << endl; - log() << "priv loc: " << (void*)(p + low) << ' ' << endl; + log() << ss.str() << endl; + log() << "priv loc: " << (void*)(p + low) << ' ' << endl; - severe() << "Written data does not match in-memory view. Missing WriteIntent?"; - invariant(false); - } + severe() << "Written data does not match in-memory view. Missing WriteIntent?"; + invariant(false); } +} - /** - * Main code of the remap private view function. - */ - void remapPrivateViewImpl(double fraction) { - LOG(4) << "journal REMAPPRIVATEVIEW" << endl; - - // There is no way that the set of files can change while we are in this method, because - // we hold the flush lock in X mode. For files to go away, a database needs to be dropped, - // which means acquiring the flush lock in at least IX mode. - // - // However, the record fetcher logic unfortunately operates without any locks and on - // Windows and Solaris remap is not atomic and there is a window where the record fetcher - // might get an access violation. That's why we acquire the mongo files mutex here in X - // mode and the record fetcher takes in in S-mode (see MmapV1RecordFetcher for more - // detail). - // - // See SERVER-5723 for performance improvement. - // See SERVER-5680 to see why this code is necessary on Windows. - // See SERVER-8795 to see why this code is necessary on Solaris. +/** + * Main code of the remap private view function. + */ +void remapPrivateViewImpl(double fraction) { + LOG(4) << "journal REMAPPRIVATEVIEW" << endl; + +// There is no way that the set of files can change while we are in this method, because +// we hold the flush lock in X mode. For files to go away, a database needs to be dropped, +// which means acquiring the flush lock in at least IX mode. +// +// However, the record fetcher logic unfortunately operates without any locks and on +// Windows and Solaris remap is not atomic and there is a window where the record fetcher +// might get an access violation. That's why we acquire the mongo files mutex here in X +// mode and the record fetcher takes in in S-mode (see MmapV1RecordFetcher for more +// detail). +// +// See SERVER-5723 for performance improvement. +// See SERVER-5680 to see why this code is necessary on Windows. +// See SERVER-8795 to see why this code is necessary on Solaris. #if defined(_WIN32) || defined(__sun) - LockMongoFilesExclusive lk; + LockMongoFilesExclusive lk; #else - LockMongoFilesShared lk; + LockMongoFilesShared lk; #endif - std::set<MongoFile*>& files = MongoFile::getAllFiles(); + std::set<MongoFile*>& files = MongoFile::getAllFiles(); - const unsigned sz = files.size(); - if (sz == 0) { - return; - } + const unsigned sz = files.size(); + if (sz == 0) { + return; + } - unsigned ntodo = (unsigned) (sz * fraction); - if( ntodo < 1 ) ntodo = 1; - if( ntodo > sz ) ntodo = sz; + unsigned ntodo = (unsigned)(sz * fraction); + if (ntodo < 1) + ntodo = 1; + if (ntodo > sz) + ntodo = sz; + + const set<MongoFile*>::iterator b = files.begin(); + const set<MongoFile*>::iterator e = files.end(); + set<MongoFile*>::iterator i = b; + + // Skip to our starting position as remembered from the last remap cycle + for (unsigned x = 0; x < remapFileToStartAt; x++) { + i++; + if (i == e) + i = b; + } - const set<MongoFile*>::iterator b = files.begin(); - const set<MongoFile*>::iterator e = files.end(); - set<MongoFile*>::iterator i = b; + // Mark where to start on the next cycle + const unsigned startedAt = remapFileToStartAt; + remapFileToStartAt = (remapFileToStartAt + ntodo) % sz; - // Skip to our starting position as remembered from the last remap cycle - for (unsigned x = 0; x < remapFileToStartAt; x++) { - i++; - if (i == e) i = b; - } + Timer t; - // Mark where to start on the next cycle - const unsigned startedAt = remapFileToStartAt; - remapFileToStartAt = (remapFileToStartAt + ntodo) % sz; + for (unsigned x = 0; x < ntodo; x++) { + if ((*i)->isDurableMappedFile()) { + DurableMappedFile* const mmf = (DurableMappedFile*)*i; - Timer t; + // Sanity check that the contents of the shared and the private view match so we + // don't end up overwriting data. + if (mmapv1GlobalOptions.journalOptions & MMAPV1Options::JournalParanoid) { + debugValidateFileMapsMatch(mmf); + } - for (unsigned x = 0; x < ntodo; x++) { - if ((*i)->isDurableMappedFile()) { - DurableMappedFile* const mmf = (DurableMappedFile*) *i; + if (mmf->willNeedRemap()) { + mmf->remapThePrivateView(); + } - // Sanity check that the contents of the shared and the private view match so we - // don't end up overwriting data. - if (mmapv1GlobalOptions.journalOptions & MMAPV1Options::JournalParanoid) { - debugValidateFileMapsMatch(mmf); - } + i++; - if (mmf->willNeedRemap()) { - mmf->remapThePrivateView(); - } + if (i == e) + i = b; + } + } - i++; + LOG(3) << "journal REMAPPRIVATEVIEW done startedAt: " << startedAt << " n:" << ntodo << ' ' + << t.millis() << "ms"; +} - if (i == e) i = b; - } - } - LOG(3) << "journal REMAPPRIVATEVIEW done startedAt: " << startedAt << " n:" << ntodo - << ' ' << t.millis() << "ms"; - } +// One instance of each durability interface +DurableImpl durableImpl; +NonDurableImpl nonDurableImpl; +} // namespace - // One instance of each durability interface - DurableImpl durableImpl; - NonDurableImpl nonDurableImpl; -} // namespace +// Declared in dur_preplogbuffer.cpp +void PREPLOGBUFFER(JSectHeader& outHeader, AlignedBuilder& outBuffer); +// Declared in dur_journal.cpp +boost::filesystem::path getJournalDir(); +void preallocateFiles(); - // Declared in dur_preplogbuffer.cpp - void PREPLOGBUFFER(JSectHeader& outHeader, AlignedBuilder& outBuffer); +// Forward declaration +static void durThread(); - // Declared in dur_journal.cpp - boost::filesystem::path getJournalDir(); - void preallocateFiles(); +// Durability activity statistics +Stats stats; - // Forward declaration - static void durThread(); +// Reference to the write intents tracking object +CommitJob commitJob; - // Durability activity statistics - Stats stats; +// Reference to the active durability interface +DurableInterface* DurableInterface::_impl(&nonDurableImpl); - // Reference to the write intents tracking object - CommitJob commitJob; - // Reference to the active durability interface - DurableInterface* DurableInterface::_impl(&nonDurableImpl); +// +// Stats +// +Stats::Stats() : _currIdx(0) {} - // - // Stats - // +void Stats::reset() { + // Seal the current metrics + _stats[_currIdx]._durationMillis = _stats[_currIdx].getCurrentDurationMillis(); - Stats::Stats() : _currIdx(0) { + // Use a new metric + const unsigned newCurrIdx = (_currIdx + 1) % (sizeof(_stats) / sizeof(_stats[0])); + _stats[newCurrIdx].reset(); - } + _currIdx = newCurrIdx; +} - void Stats::reset() { - // Seal the current metrics - _stats[_currIdx]._durationMillis = _stats[_currIdx].getCurrentDurationMillis(); +BSONObj Stats::asObj() const { + // Use the previous statistic + const S& stats = _stats[(_currIdx - 1) % (sizeof(_stats) / sizeof(_stats[0]))]; - // Use a new metric - const unsigned newCurrIdx = (_currIdx + 1) % (sizeof(_stats) / sizeof(_stats[0])); - _stats[newCurrIdx].reset(); + BSONObjBuilder builder; + stats._asObj(&builder); - _currIdx = newCurrIdx; - } + return builder.obj(); +} - BSONObj Stats::asObj() const { - // Use the previous statistic - const S& stats = _stats[(_currIdx - 1) % (sizeof(_stats) / sizeof(_stats[0]))]; +void Stats::S::reset() { + memset(this, 0, sizeof(*this)); + _startTimeMicros = curTimeMicros64(); +} - BSONObjBuilder builder; - stats._asObj(&builder); +std::string Stats::S::_CSVHeader() const { + return "cmts\t jrnMB\t wrDFMB\t cIWLk\t early\t prpLgB\t wrToJ\t wrToDF\t rmpPrVw"; +} - return builder.obj(); - } +std::string Stats::S::_asCSV() const { + stringstream ss; + ss << setprecision(2) << _commits << '\t' << _journaledBytes / 1000000.0 << '\t' + << _writeToDataFilesBytes / 1000000.0 << '\t' << _commitsInWriteLock << '\t' << 0 << '\t' + << (unsigned)(_prepLogBufferMicros / 1000) << '\t' + << (unsigned)(_writeToJournalMicros / 1000) << '\t' + << (unsigned)(_writeToDataFilesMicros / 1000) << '\t' + << (unsigned)(_remapPrivateViewMicros / 1000) << '\t' << (unsigned)(_commitsMicros / 1000) + << '\t' << (unsigned)(_commitsInWriteLockMicros / 1000) << '\t'; - void Stats::S::reset() { - memset(this, 0, sizeof(*this)); - _startTimeMicros = curTimeMicros64(); - } + return ss.str(); +} - std::string Stats::S::_CSVHeader() const { - return "cmts\t jrnMB\t wrDFMB\t cIWLk\t early\t prpLgB\t wrToJ\t wrToDF\t rmpPrVw"; - } +void Stats::S::_asObj(BSONObjBuilder* builder) const { + BSONObjBuilder& b = *builder; + b << "commits" << _commits << "journaledMB" << _journaledBytes / 1000000.0 + << "writeToDataFilesMB" << _writeToDataFilesBytes / 1000000.0 << "compression" + << _journaledBytes / (_uncompressedBytes + 1.0) << "commitsInWriteLock" << _commitsInWriteLock + << "earlyCommits" << 0 << "timeMs" + << BSON("dt" << _durationMillis << "prepLogBuffer" << (unsigned)(_prepLogBufferMicros / 1000) + << "writeToJournal" << (unsigned)(_writeToJournalMicros / 1000) + << "writeToDataFiles" << (unsigned)(_writeToDataFilesMicros / 1000) + << "remapPrivateView" << (unsigned)(_remapPrivateViewMicros / 1000) << "commits" + << (unsigned)(_commitsMicros / 1000) << "commitsInWriteLock" + << (unsigned)(_commitsInWriteLockMicros / 1000)); - std::string Stats::S::_asCSV() const { - stringstream ss; - ss << setprecision(2) - << _commits << '\t' - << _journaledBytes / 1000000.0 << '\t' - << _writeToDataFilesBytes / 1000000.0 << '\t' - << _commitsInWriteLock << '\t' - << 0 << '\t' - << (unsigned) (_prepLogBufferMicros / 1000) << '\t' - << (unsigned) (_writeToJournalMicros / 1000) << '\t' - << (unsigned) (_writeToDataFilesMicros / 1000) << '\t' - << (unsigned) (_remapPrivateViewMicros / 1000) << '\t' - << (unsigned) (_commitsMicros / 1000) << '\t' - << (unsigned) (_commitsInWriteLockMicros / 1000) << '\t'; - - return ss.str(); + if (mmapv1GlobalOptions.journalCommitInterval != 0) { + b << "journalCommitIntervalMs" << mmapv1GlobalOptions.journalCommitInterval; } +} - void Stats::S::_asObj(BSONObjBuilder* builder) const { - BSONObjBuilder& b = *builder; - b << "commits" << _commits - << "journaledMB" << _journaledBytes / 1000000.0 - << "writeToDataFilesMB" << _writeToDataFilesBytes / 1000000.0 - << "compression" << _journaledBytes / (_uncompressedBytes + 1.0) - << "commitsInWriteLock" << _commitsInWriteLock - << "earlyCommits" << 0 - << "timeMs" << BSON("dt" << _durationMillis << - "prepLogBuffer" << (unsigned) (_prepLogBufferMicros / 1000) << - "writeToJournal" << (unsigned) (_writeToJournalMicros / 1000) << - "writeToDataFiles" << (unsigned) (_writeToDataFilesMicros / 1000) << - "remapPrivateView" << (unsigned) (_remapPrivateViewMicros / 1000) << - "commits" << (unsigned)(_commitsMicros / 1000) << - "commitsInWriteLock" - << (unsigned)(_commitsInWriteLockMicros / 1000)); - - if (mmapv1GlobalOptions.journalCommitInterval != 0) { - b << "journalCommitIntervalMs" << mmapv1GlobalOptions.journalCommitInterval; - } - } +// +// DurableInterface +// - // - // DurableInterface - // +DurableInterface::DurableInterface() {} - DurableInterface::DurableInterface() { +DurableInterface::~DurableInterface() {} - } - DurableInterface::~DurableInterface() { +// +// DurableImpl +// - } +bool DurableImpl::commitNow(OperationContext* txn) { + NotifyAll::When when = commitNotify.now(); + AutoYieldFlushLockForMMAPV1Commit flushLockYield(txn->lockState()); - // - // DurableImpl - // + // There is always just one waiting anyways + flushRequested.notify_one(); - bool DurableImpl::commitNow(OperationContext* txn) { - NotifyAll::When when = commitNotify.now(); + // commitNotify.waitFor ensures that whatever was scheduled for journaling before this + // call has been persisted to the journal file. This does not mean that this data has been + // applied to the shared view yet though, that's why we wait for applyToDataFilesNotify. + applyToDataFilesNotify.waitFor(when); - AutoYieldFlushLockForMMAPV1Commit flushLockYield(txn->lockState()); + return true; +} - // There is always just one waiting anyways - flushRequested.notify_one(); +bool DurableImpl::waitUntilDurable() { + commitNotify.awaitBeyondNow(); + return true; +} - // commitNotify.waitFor ensures that whatever was scheduled for journaling before this - // call has been persisted to the journal file. This does not mean that this data has been - // applied to the shared view yet though, that's why we wait for applyToDataFilesNotify. - applyToDataFilesNotify.waitFor(when); +void DurableImpl::createdFile(const std::string& filename, unsigned long long len) { + std::shared_ptr<DurOp> op(new FileCreatedOp(filename, len)); + commitJob.noteOp(op); +} - return true; - } - bool DurableImpl::waitUntilDurable() { - commitNotify.awaitBeyondNow(); - return true; +void DurableImpl::declareWriteIntents(const std::vector<std::pair<void*, unsigned>>& intents) { + typedef std::vector<std::pair<void*, unsigned>> Intents; + stdx::lock_guard<SimpleMutex> lk(commitJob.groupCommitMutex); + for (Intents::const_iterator it(intents.begin()), end(intents.end()); it != end; ++it) { + commitJob.note(it->first, it->second); } +} - void DurableImpl::createdFile(const std::string& filename, unsigned long long len) { - std::shared_ptr<DurOp> op(new FileCreatedOp(filename, len)); - commitJob.noteOp(op); +bool DurableImpl::commitIfNeeded() { + if (MONGO_likely(commitJob.bytes() < UncommittedBytesLimit)) { + return false; } + // Just wake up the flush thread + flushRequested.notify_one(); + return true; +} - void DurableImpl::declareWriteIntents( - const std::vector<std::pair<void*, unsigned> >& intents) { - typedef std::vector<std::pair<void*, unsigned> > Intents; - stdx::lock_guard<SimpleMutex> lk(commitJob.groupCommitMutex); - for (Intents::const_iterator it(intents.begin()), end(intents.end()); it != end; ++it) { - commitJob.note(it->first, it->second); - } - } - - bool DurableImpl::commitIfNeeded() { - if (MONGO_likely(commitJob.bytes() < UncommittedBytesLimit)) { - return false; - } +void DurableImpl::syncDataAndTruncateJournal(OperationContext* txn) { + invariant(txn->lockState()->isW()); - // Just wake up the flush thread - flushRequested.notify_one(); - return true; - } + // Once this returns, all the outstanding journal has been applied to the data files and + // so it's safe to do the flushAll/journalCleanup below. + commitNow(txn); - void DurableImpl::syncDataAndTruncateJournal(OperationContext* txn) { - invariant(txn->lockState()->isW()); + // Flush the shared view to disk. + MongoFile::flushAll(true); - // Once this returns, all the outstanding journal has been applied to the data files and - // so it's safe to do the flushAll/journalCleanup below. - commitNow(txn); + // Once the shared view has been flushed, we do not need the journal files anymore. + journalCleanup(true); - // Flush the shared view to disk. - MongoFile::flushAll(true); + // Double check post-conditions + invariant(!haveJournalFiles()); +} - // Once the shared view has been flushed, we do not need the journal files anymore. - journalCleanup(true); +void DurableImpl::closingFileNotification() { + if (commitJob.hasWritten()) { + severe() << "journal warning files are closing outside locks with writes pending"; - // Double check post-conditions - invariant(!haveJournalFiles()); + // File is closing while there are unwritten changes + invariant(false); } +} - void DurableImpl::closingFileNotification() { - if (commitJob.hasWritten()) { - severe() << "journal warning files are closing outside locks with writes pending"; +void DurableImpl::commitAndStopDurThread() { + NotifyAll::When when = commitNotify.now(); - // File is closing while there are unwritten changes - invariant(false); - } - } + // There is always just one waiting anyways + flushRequested.notify_one(); - void DurableImpl::commitAndStopDurThread() { - NotifyAll::When when = commitNotify.now(); + // commitNotify.waitFor ensures that whatever was scheduled for journaling before this + // call has been persisted to the journal file. This does not mean that this data has been + // applied to the shared view yet though, that's why we wait for applyToDataFilesNotify. + applyToDataFilesNotify.waitFor(when); - // There is always just one waiting anyways - flushRequested.notify_one(); + // Flush the shared view to disk. + MongoFile::flushAll(true); - // commitNotify.waitFor ensures that whatever was scheduled for journaling before this - // call has been persisted to the journal file. This does not mean that this data has been - // applied to the shared view yet though, that's why we wait for applyToDataFilesNotify. - applyToDataFilesNotify.waitFor(when); + // Once the shared view has been flushed, we do not need the journal files anymore. + journalCleanup(true); - // Flush the shared view to disk. - MongoFile::flushAll(true); + // Double check post-conditions + invariant(!haveJournalFiles()); - // Once the shared view has been flushed, we do not need the journal files anymore. - journalCleanup(true); + shutdownRequested.store(1); - // Double check post-conditions - invariant(!haveJournalFiles()); + // Wait for the durability thread to terminate + log() << "Terminating durability thread ..."; + _durThreadHandle.join(); +} - shutdownRequested.store(1); +void DurableImpl::start() { + // Start the durability thread + stdx::thread t(durThread); + _durThreadHandle.swap(t); +} - // Wait for the durability thread to terminate - log() << "Terminating durability thread ..."; - _durThreadHandle.join(); - } - void DurableImpl::start() { - // Start the durability thread - stdx::thread t(durThread); - _durThreadHandle.swap(t); +/** + * Remaps the private view from the shared view so that it does not consume too much + * copy-on-write/swap space. Must only be called after the in-memory journal has been flushed + * to disk and applied on top of the shared view. + * + * @param fraction Value between (0, 1] indicating what fraction of the memory to remap. + * Remapping too much or too frequently incurs copy-on-write page fault cost. + */ +static void remapPrivateView(double fraction) { + // Remapping private views must occur after WRITETODATAFILES otherwise we wouldn't see any + // newly written data on reads. + invariant(!commitJob.hasWritten()); + + try { + Timer t; + remapPrivateViewImpl(fraction); + stats.curr()->_remapPrivateViewMicros += t.micros(); + + LOG(4) << "remapPrivateView end"; + return; + } catch (DBException& e) { + severe() << "dbexception in remapPrivateView causing immediate shutdown: " << e.toString(); + } catch (std::ios_base::failure& e) { + severe() << "ios_base exception in remapPrivateView causing immediate shutdown: " + << e.what(); + } catch (std::bad_alloc& e) { + severe() << "bad_alloc exception in remapPrivateView causing immediate shutdown: " + << e.what(); + } catch (std::exception& e) { + severe() << "exception in remapPrivateView causing immediate shutdown: " << e.what(); + } catch (...) { + severe() << "unknown exception in remapPrivateView causing immediate shutdown: "; } + invariant(false); +} - /** - * Remaps the private view from the shared view so that it does not consume too much - * copy-on-write/swap space. Must only be called after the in-memory journal has been flushed - * to disk and applied on top of the shared view. - * - * @param fraction Value between (0, 1] indicating what fraction of the memory to remap. - * Remapping too much or too frequently incurs copy-on-write page fault cost. - */ - static void remapPrivateView(double fraction) { - // Remapping private views must occur after WRITETODATAFILES otherwise we wouldn't see any - // newly written data on reads. - invariant(!commitJob.hasWritten()); - try { - Timer t; - remapPrivateViewImpl(fraction); - stats.curr()->_remapPrivateViewMicros += t.micros(); +/** + * The main durability thread loop. There is a single instance of this function running. + */ +static void durThread() { + Client::initThread("durability"); - LOG(4) << "remapPrivateView end"; - return; - } - catch (DBException& e) { - severe() << "dbexception in remapPrivateView causing immediate shutdown: " - << e.toString(); - } - catch (std::ios_base::failure& e) { - severe() << "ios_base exception in remapPrivateView causing immediate shutdown: " - << e.what(); - } - catch (std::bad_alloc& e) { - severe() << "bad_alloc exception in remapPrivateView causing immediate shutdown: " - << e.what(); - } - catch (std::exception& e) { - severe() << "exception in remapPrivateView causing immediate shutdown: " - << e.what(); - } - catch (...) { - severe() << "unknown exception in remapPrivateView causing immediate shutdown: "; - } + log() << "Durability thread started"; - invariant(false); + bool samePartition = true; + try { + const std::string dbpathDir = boost::filesystem::path(storageGlobalParams.dbpath).string(); + samePartition = onSamePartition(getJournalDir().string(), dbpathDir); + } catch (...) { } + // Spawn the journal writer thread + JournalWriter journalWriter(&commitNotify, &applyToDataFilesNotify, NumAsyncJournalWrites); + journalWriter.start(); - /** - * The main durability thread loop. There is a single instance of this function running. - */ - static void durThread() { - Client::initThread("durability"); + // Used as an estimate of how much / how fast to remap + uint64_t commitCounter(0); + uint64_t estimatedPrivateMapSize(0); + uint64_t remapLastTimestamp(0); - log() << "Durability thread started"; - - bool samePartition = true; - try { - const std::string dbpathDir = - boost::filesystem::path(storageGlobalParams.dbpath).string(); - samePartition = onSamePartition(getJournalDir().string(), dbpathDir); + while (shutdownRequested.loadRelaxed() == 0) { + unsigned ms = mmapv1GlobalOptions.journalCommitInterval; + if (ms == 0) { + ms = samePartition ? 100 : 30; } - catch(...) { - } + // +1 so it never goes down to zero + const unsigned oneThird = (ms / 3) + 1; - // Spawn the journal writer thread - JournalWriter journalWriter(&commitNotify, &applyToDataFilesNotify, NumAsyncJournalWrites); - journalWriter.start(); + // Reset the stats based on the reset interval + if (stats.curr()->getCurrentDurationMillis() > DurStatsResetIntervalMillis) { + stats.reset(); + } - // Used as an estimate of how much / how fast to remap - uint64_t commitCounter(0); - uint64_t estimatedPrivateMapSize(0); - uint64_t remapLastTimestamp(0); + try { + stdx::unique_lock<stdx::mutex> lock(flushMutex); - while (shutdownRequested.loadRelaxed() == 0) { - unsigned ms = mmapv1GlobalOptions.journalCommitInterval; - if (ms == 0) { - ms = samePartition ? 100 : 30; - } + for (unsigned i = 0; i <= 2; i++) { + if (boost::cv_status::no_timeout == + flushRequested.wait_for(lock, Milliseconds(oneThird))) { + // Someone forced a flush + break; + } - // +1 so it never goes down to zero - const unsigned oneThird = (ms / 3) + 1; + if (commitNotify.nWaiting()) { + // One or more getLastError j:true is pending + break; + } - // Reset the stats based on the reset interval - if (stats.curr()->getCurrentDurationMillis() > DurStatsResetIntervalMillis) { - stats.reset(); + if (commitJob.bytes() > UncommittedBytesLimit / 2) { + // The number of written bytes is growing + break; + } } - try { - stdx::unique_lock<stdx::mutex> lock(flushMutex); + // The commit logic itself + LOG(4) << "groupCommit begin"; - for (unsigned i = 0; i <= 2; i++) { - if (boost::cv_status::no_timeout == flushRequested.wait_for( - lock, Milliseconds(oneThird))) { - // Someone forced a flush - break; - } + Timer t; - if (commitNotify.nWaiting()) { - // One or more getLastError j:true is pending - break; + OperationContextImpl txn; + AutoAcquireFlushLockForMMAPV1Commit autoFlushLock(txn.lockState()); + + // We need to snapshot the commitNumber after the flush lock has been obtained, + // because at this point we know that we have a stable snapshot of the data. + const NotifyAll::When commitNumber(commitNotify.now()); + + LOG(4) << "Processing commit number " << commitNumber; + + if (!commitJob.hasWritten()) { + // We do not need the journal lock anymore. Free it here, for the really + // unlikely possibility that the writeBuffer command below blocks. + autoFlushLock.release(); + + // getlasterror request could have came after the data was already committed. + // No need to call committingReset though, because we have not done any + // writes (hasWritten == false). + JournalWriter::Buffer* const buffer = journalWriter.newBuffer(); + buffer->setNoop(); + + journalWriter.writeBuffer(buffer, commitNumber); + } 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()); + + estimatedPrivateMapSize += commitJob.bytes(); + commitCounter++; + + // Now that the write intents have been copied to the buffer, the commit job is + // free to be reused. We need to reset the commit job's contents while under + // the S flush lock, because otherwise someone might have done a write and this + // would wipe out their changes without ever being committed. + commitJob.committingReset(); + + double systemMemoryPressurePercentage = + ProcessInfo::getSystemMemoryPressurePercentage(); + + // Now that the in-memory modifications have been collected, we can potentially + // release the flush lock if remap is not necessary. + // When we remap due to memory pressure, we look at two criteria + // 1. If the amount of 4k pages touched exceeds 512 MB, + // a reasonable estimate of memory pressure on Linux. + // 2. Check if the amount of free memory on the machine is running low, + // since #1 is underestimates the memory pressure on Windows since + // commits in 64MB chunks. + const bool shouldRemap = (estimatedPrivateMapSize >= UncommittedBytesLimit) || + (systemMemoryPressurePercentage > 0.0) || + (commitCounter % NumCommitsBeforeRemap == 0) || + (mmapv1GlobalOptions.journalOptions & MMAPV1Options::JournalAlwaysRemap); + + double remapFraction = 0.0; + + if (shouldRemap) { + // We want to remap all private views about every 2 seconds. There could be + // ~1000 views so we do a little each pass. There will be copy on write + // faults after remapping, so doing a little bit at a time will avoid big + // load spikes when the pages are touched. + // + // TODO: Instead of the time-based logic above, consider using ProcessInfo + // and watching for getResidentSize to drop, which is more precise. + remapFraction = (curTimeMicros64() - remapLastTimestamp) / 2000000.0; + + if (mmapv1GlobalOptions.journalOptions & MMAPV1Options::JournalAlwaysRemap) { + remapFraction = 1; + } else { + // We don't want to get close to the UncommittedBytesLimit + const double remapMemFraction = + estimatedPrivateMapSize / ((double)UncommittedBytesLimit); + + remapFraction = std::max(remapMemFraction, remapFraction); + + remapFraction = std::max(systemMemoryPressurePercentage, remapFraction); } + } else { + LOG(4) << "Early release flush lock"; - if (commitJob.bytes() > UncommittedBytesLimit / 2) { - // The number of written bytes is growing - break; - } + // We will not be doing a remap so drop the flush lock. That way we will be + // doing the journal I/O outside of lock, so other threads can proceed. + invariant(!shouldRemap); + autoFlushLock.release(); } - // The commit logic itself - LOG(4) << "groupCommit begin"; + // Request async I/O to the journal. This may block. + journalWriter.writeBuffer(buffer, commitNumber); + + // Data has now been written to the shared view. If remap was requested, we + // would still be holding the S flush lock here, so just upgrade it and + // perform the remap. + if (shouldRemap) { + // Need to wait for the previously scheduled journal writes to complete + // before any remap is attempted. + journalWriter.flush(); + journalWriter.assertIdle(); + + // Upgrading the journal lock to flush stops all activity on the system, + // because we will be remapping memory and we don't want readers to be + // accessing it. Technically this step could be avoided on systems, which + // support atomic remap. + autoFlushLock.upgradeFlushLockToExclusive(); + remapPrivateView(remapFraction); - Timer t; - - OperationContextImpl txn; - AutoAcquireFlushLockForMMAPV1Commit autoFlushLock(txn.lockState()); - - // We need to snapshot the commitNumber after the flush lock has been obtained, - // because at this point we know that we have a stable snapshot of the data. - const NotifyAll::When commitNumber(commitNotify.now()); - - LOG(4) << "Processing commit number " << commitNumber; - - if (!commitJob.hasWritten()) { - // We do not need the journal lock anymore. Free it here, for the really - // unlikely possibility that the writeBuffer command below blocks. autoFlushLock.release(); - // getlasterror request could have came after the data was already committed. - // No need to call committingReset though, because we have not done any - // writes (hasWritten == false). - JournalWriter::Buffer* const buffer = journalWriter.newBuffer(); - buffer->setNoop(); - - journalWriter.writeBuffer(buffer, commitNumber); - } - 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()); - - estimatedPrivateMapSize += commitJob.bytes(); - commitCounter++; - - // Now that the write intents have been copied to the buffer, the commit job is - // free to be reused. We need to reset the commit job's contents while under - // the S flush lock, because otherwise someone might have done a write and this - // would wipe out their changes without ever being committed. - commitJob.committingReset(); - - double systemMemoryPressurePercentage = - ProcessInfo::getSystemMemoryPressurePercentage(); - - // Now that the in-memory modifications have been collected, we can potentially - // release the flush lock if remap is not necessary. - // When we remap due to memory pressure, we look at two criteria - // 1. If the amount of 4k pages touched exceeds 512 MB, - // a reasonable estimate of memory pressure on Linux. - // 2. Check if the amount of free memory on the machine is running low, - // since #1 is underestimates the memory pressure on Windows since - // commits in 64MB chunks. - const bool shouldRemap = - (estimatedPrivateMapSize >= UncommittedBytesLimit) || - (systemMemoryPressurePercentage > 0.0) || - (commitCounter % NumCommitsBeforeRemap == 0) || - (mmapv1GlobalOptions.journalOptions & MMAPV1Options::JournalAlwaysRemap); - - double remapFraction = 0.0; - - if (shouldRemap) { - // We want to remap all private views about every 2 seconds. There could be - // ~1000 views so we do a little each pass. There will be copy on write - // faults after remapping, so doing a little bit at a time will avoid big - // load spikes when the pages are touched. - // - // TODO: Instead of the time-based logic above, consider using ProcessInfo - // and watching for getResidentSize to drop, which is more precise. - remapFraction = (curTimeMicros64() - remapLastTimestamp) / 2000000.0; - - if (mmapv1GlobalOptions.journalOptions & - MMAPV1Options::JournalAlwaysRemap) { - remapFraction = 1; - } - else { - // We don't want to get close to the UncommittedBytesLimit - const double remapMemFraction = - estimatedPrivateMapSize / ((double)UncommittedBytesLimit); - - remapFraction = std::max(remapMemFraction, remapFraction); - - remapFraction = std::max(systemMemoryPressurePercentage, remapFraction); - } - } - else { - LOG(4) << "Early release flush lock"; - - // We will not be doing a remap so drop the flush lock. That way we will be - // doing the journal I/O outside of lock, so other threads can proceed. - invariant(!shouldRemap); - autoFlushLock.release(); - } + // Reset the private map estimate outside of the lock + estimatedPrivateMapSize = 0; + remapLastTimestamp = curTimeMicros64(); - // Request async I/O to the journal. This may block. - journalWriter.writeBuffer(buffer, commitNumber); - - // Data has now been written to the shared view. If remap was requested, we - // would still be holding the S flush lock here, so just upgrade it and - // perform the remap. - if (shouldRemap) { - // Need to wait for the previously scheduled journal writes to complete - // before any remap is attempted. - journalWriter.flush(); - journalWriter.assertIdle(); - - // Upgrading the journal lock to flush stops all activity on the system, - // because we will be remapping memory and we don't want readers to be - // accessing it. Technically this step could be avoided on systems, which - // support atomic remap. - autoFlushLock.upgradeFlushLockToExclusive(); - remapPrivateView(remapFraction); - - autoFlushLock.release(); - - // Reset the private map estimate outside of the lock - estimatedPrivateMapSize = 0; - remapLastTimestamp = curTimeMicros64(); - - stats.curr()->_commitsInWriteLock++; - stats.curr()->_commitsInWriteLockMicros += t.micros(); - } + stats.curr()->_commitsInWriteLock++; + stats.curr()->_commitsInWriteLockMicros += t.micros(); } - - stats.curr()->_commits++; - stats.curr()->_commitsMicros += t.micros(); - - LOG(4) << "groupCommit end"; - } - catch (DBException& e) { - severe() << "dbexception in durThread causing immediate shutdown: " - << e.toString(); - invariant(false); - } - catch (std::ios_base::failure& e) { - severe() << "ios_base exception in durThread causing immediate shutdown: " - << e.what(); - invariant(false); } - catch (std::bad_alloc& e) { - severe() << "bad_alloc exception in durThread causing immediate shutdown: " - << e.what(); - invariant(false); - } - catch (std::exception& e) { - severe() << "exception in durThread causing immediate shutdown: " - << e.what(); - invariant(false); - } - catch (...) { - severe() << "unhandled exception in durThread causing immediate shutdown"; - invariant(false); - } - } - // Stops the journal thread and ensures everything was written - invariant(!commitJob.hasWritten()); + stats.curr()->_commits++; + stats.curr()->_commitsMicros += t.micros(); - journalWriter.flush(); - journalWriter.shutdown(); - - log() << "Durability thread stopped"; + LOG(4) << "groupCommit end"; + } catch (DBException& e) { + severe() << "dbexception in durThread causing immediate shutdown: " << e.toString(); + invariant(false); + } catch (std::ios_base::failure& e) { + severe() << "ios_base exception in durThread causing immediate shutdown: " << e.what(); + invariant(false); + } catch (std::bad_alloc& e) { + severe() << "bad_alloc exception in durThread causing immediate shutdown: " << e.what(); + invariant(false); + } catch (std::exception& e) { + severe() << "exception in durThread causing immediate shutdown: " << e.what(); + invariant(false); + } catch (...) { + severe() << "unhandled exception in durThread causing immediate shutdown"; + invariant(false); + } } + // Stops the journal thread and ensures everything was written + invariant(!commitJob.hasWritten()); - /** - * Invoked at server startup. Recovers the database by replaying journal files and then - * starts the durability thread. - */ - void startup() { - if (!storageGlobalParams.dur) { - return; - } + journalWriter.flush(); + journalWriter.shutdown(); - journalMakeDir(); + log() << "Durability thread stopped"; +} - try { - replayJournalFilesAtStartup(); - } - catch (DBException& e) { - severe() << "dbexception during recovery: " << e.toString(); - throw; - } - catch (std::exception& e) { - severe() << "std::exception during recovery: " << e.what(); - throw; - } - catch (...) { - severe() << "exception during recovery"; - throw; - } - preallocateFiles(); +/** + * Invoked at server startup. Recovers the database by replaying journal files and then + * starts the durability thread. + */ +void startup() { + if (!storageGlobalParams.dur) { + return; + } - durableImpl.start(); - DurableInterface::_impl = &durableImpl; + journalMakeDir(); + + try { + replayJournalFilesAtStartup(); + } catch (DBException& e) { + severe() << "dbexception during recovery: " << e.toString(); + throw; + } catch (std::exception& e) { + severe() << "std::exception during recovery: " << e.what(); + throw; + } catch (...) { + severe() << "exception during recovery"; + throw; } -} // namespace dur -} // namespace mongo + preallocateFiles(); + + durableImpl.start(); + DurableInterface::_impl = &durableImpl; +} + +} // namespace dur +} // namespace mongo |