wiredtiger_session_cache.h

Go to the documentation of this file.

// wiredtiger_session_cache.h

#pragma once

#include <list>
#include <string>

#include <wiredtiger.h>

#include "mongo/db/storage/journal_listener.h"
#include "mongo/db/storage/wiredtiger/wiredtiger_snapshot_manager.h"
#include "mongo/platform/atomic_word.h"
#include "mongo/stdx/mutex.h"
#include "mongo/util/concurrency/spin_lock.h"

namespace mongo {

class WiredTigerKVEngine;
class WiredTigerSessionCache;

class WiredTigerCachedCursor {
public:
    WiredTigerCachedCursor(uint64_t id, uint64_t gen, WT_CURSOR* cursor)
        : _id(id), _gen(gen), _cursor(cursor) {}

    uint64_t _id;   // Source ID, assigned to each URI
    uint64_t _gen;  // Generation, used to age out old cursors
    WT_CURSOR* _cursor;
};

class WiredTigerSession {
public:
    WiredTigerSession(WT_CONNECTION* conn, uint64_t epoch = 0, uint64_t cursorEpoch = 0);

    WiredTigerSession(WT_CONNECTION* conn,
                      WiredTigerSessionCache* cache,
                      uint64_t epoch = 0,
                      uint64_t cursorEpoch = 0);

    ~WiredTigerSession();

    WT_SESSION* getSession() const {
        return _session;
    }

    WT_CURSOR* getCursor(const std::string& uri, uint64_t id, bool forRecordStore);

    void releaseCursor(uint64_t id, WT_CURSOR* cursor);

    void closeCursorsForQueuedDrops(WiredTigerKVEngine* engine);

    void closeAllCursors(const std::string& uri);

    int cursorsOut() const {
        return _cursorsOut;
    }

    bool isDropQueuedIdentsAtSessionEndAllowed() const {
        return _dropQueuedIdentsAtSessionEnd;
    }

    void dropQueuedIdentsAtSessionEndAllowed(bool dropQueuedIdentsAtSessionEnd) {
        _dropQueuedIdentsAtSessionEnd = dropQueuedIdentsAtSessionEnd;
    }

    static uint64_t genTableId();

    static const uint64_t kMetadataTableId = 0;

private:
    friend class WiredTigerSessionCache;

    // The cursor cache is a list of pairs that contain an ID and cursor
    typedef std::list<WiredTigerCachedCursor> CursorCache;

    // Used internally by WiredTigerSessionCache
    uint64_t _getEpoch() const {
        return _epoch;
    }

    // Used internally by WiredTigerSessionCache
    uint64_t _getCursorEpoch() const {
        return _cursorEpoch;
    }

    const uint64_t _epoch;
    uint64_t _cursorEpoch;
    WiredTigerSessionCache* _cache;  // not owned
    WT_SESSION* _session;            // owned
    CursorCache _cursors;            // owned
    uint64_t _cursorGen;
    int _cursorsOut;
    bool _dropQueuedIdentsAtSessionEnd = true;
};

class WiredTigerSessionCache {
public:
    WiredTigerSessionCache(WiredTigerKVEngine* engine);
    WiredTigerSessionCache(WT_CONNECTION* conn);
    ~WiredTigerSessionCache();

    class WiredTigerSessionDeleter {
    public:
        void operator()(WiredTigerSession* session) const;
    };

    static bool isEngineCachingCursors();

    std::unique_ptr<WiredTigerSession, WiredTigerSessionDeleter> getSession();

    void closeAll();

    void closeCursorsForQueuedDrops();

    void closeAllCursors(const std::string& uri);

    void shuttingDown();

    bool isEphemeral();
    void waitUntilDurable(bool forceCheckpoint, bool stableCheckpoint);

    void waitUntilPreparedUnitOfWorkCommitsOrAborts(OperationContext* opCtx);

    void notifyPreparedUnitOfWorkHasCommittedOrAborted();

    WT_CONNECTION* conn() const {
        return _conn;
    }

    WiredTigerSnapshotManager& snapshotManager() {
        return _snapshotManager;
    }
    const WiredTigerSnapshotManager& snapshotManager() const {
        return _snapshotManager;
    }

    void setJournalListener(JournalListener* jl);

    uint64_t getCursorEpoch() const {
        return _cursorEpoch.load();
    }

    WiredTigerKVEngine* getKVEngine() const {
        return _engine;
    }

private:
    WiredTigerKVEngine* _engine;  // not owned, might be NULL
    WT_CONNECTION* _conn;         // not owned
    WiredTigerSnapshotManager _snapshotManager;

    // Used as follows:
    //   The low 31 bits are a count of active calls to releaseSession.
    //   The high bit is a flag that is set if and only if we're shutting down.
    AtomicUInt32 _shuttingDown;
    static const uint32_t kShuttingDownMask = 1 << 31;

    stdx::mutex _cacheLock;
    typedef std::vector<WiredTigerSession*> SessionCache;
    SessionCache _sessions;

    // Bumped when all open sessions need to be closed
    AtomicUInt64 _epoch;  // atomic so we can check it outside of the lock

    // Bumped when all open cursors need to be closed
    AtomicUInt64 _cursorEpoch;  // atomic so we can check it outside of the lock

    // Counter and critical section mutex for waitUntilDurable
    AtomicUInt32 _lastSyncTime;
    stdx::mutex _lastSyncMutex;

    // Mutex and cond var for waiting on prepare commit or abort.
    stdx::mutex _prepareCommittedOrAbortedMutex;
    stdx::condition_variable _prepareCommittedOrAbortedCond;
    std::uint64_t _lastCommitOrAbortCounter;

    // Protects _journalListener.
    stdx::mutex _journalListenerMutex;
    // Notified when we commit to the journal.
    JournalListener* _journalListener = &NoOpJournalListener::instance;

    WT_SESSION* _waitUntilDurableSession = nullptr;  // owned, and never explicitly closed
                                                     // (uses connection close to clean up)

    void releaseSession(WiredTigerSession* session);
};

typedef std::unique_ptr<WiredTigerSession,
                        typename WiredTigerSessionCache::WiredTigerSessionDeleter>
    UniqueWiredTigerSession;
}  // namespace