wiredtiger_oplog_manager.h

Go to the documentation of this file.

#pragma once

#include "mongo/base/disallow_copying.h"
#include "mongo/db/storage/wiredtiger/wiredtiger_record_store.h"
#include "mongo/stdx/condition_variable.h"
#include "mongo/stdx/mutex.h"
#include "mongo/stdx/thread.h"
#include "mongo/util/concurrency/with_lock.h"

namespace mongo {

class WiredTigerRecordStore;
class WiredTigerSessionCache;


// Manages oplog visibility, by periodically querying WiredTiger's all_committed timestamp value and
// then using that timestamp for all transactions that read the oplog collection.
class WiredTigerOplogManager {
    MONGO_DISALLOW_COPYING(WiredTigerOplogManager);

public:
    WiredTigerOplogManager() {}
    ~WiredTigerOplogManager() {}

    // This method will initialize the oplog read timestamp and start the background thread that
    // refreshes the value.
    void start(OperationContext* opCtx,
               const std::string& uri,
               WiredTigerRecordStore* oplogRecordStore);

    void halt();

    bool isRunning() {
        stdx::lock_guard<stdx::mutex> lk(_oplogVisibilityStateMutex);
        return _isRunning && !_shuttingDown;
    }

    // The oplogReadTimestamp is the timestamp used for oplog reads, to prevent readers from
    // reading past uncommitted transactions (which may create "holes" in the oplog after an
    // unclean shutdown).
    std::uint64_t getOplogReadTimestamp() const;
    void setOplogReadTimestamp(Timestamp ts);

    // Triggers the oplogJournal thread to update its oplog read timestamp, by flushing the journal.
    void triggerJournalFlush();

    // Waits until all committed writes at this point to become visible (that is, no holes exist in
    // the oplog.)
    void waitForAllEarlierOplogWritesToBeVisible(const WiredTigerRecordStore* oplogRecordStore,
                                                 OperationContext* opCtx) const;

    // Returns the all committed timestamp. All transactions with timestamps earlier than the
    // all committed timestamp are committed.
    uint64_t fetchAllCommittedValue(WT_CONNECTION* conn);

private:
    void _oplogJournalThreadLoop(WiredTigerSessionCache* sessionCache,
                                 WiredTigerRecordStore* oplogRecordStore) noexcept;

    void _setOplogReadTimestamp(WithLock, uint64_t newTimestamp);

    stdx::thread _oplogJournalThread;
    mutable stdx::mutex _oplogVisibilityStateMutex;
    mutable stdx::condition_variable
        _opsWaitingForJournalCV;  // Signaled to trigger a journal flush.
    mutable stdx::condition_variable
        _opsBecameVisibleCV;  // Signaled when a journal flush is complete.

    bool _isRunning = false;     // Guarded by the oplogVisibilityStateMutex.
    bool _shuttingDown = false;  // Guarded by oplogVisibilityStateMutex.

    // This is the RecordId of the newest oplog document in the oplog on startup.  It is used as a
    // floor in waitForAllEarlierOplogWritesToBeVisible().
    RecordId _oplogMaxAtStartup = RecordId(0);  // Guarded by oplogVisibilityStateMutex.
    bool _opsWaitingForJournal = false;         // Guarded by oplogVisibilityStateMutex.

    AtomicUInt64 _oplogReadTimestamp;
};
}  // namespace mongo