dur_journal_writer.h

Go to the documentation of this file.

#pragma once

#include "mongo/base/disallow_copying.h"
#include "mongo/db/storage/journal_listener.h"
#include "mongo/db/storage/mmap_v1/aligned_builder.h"
#include "mongo/db/storage/mmap_v1/commit_notifier.h"
#include "mongo/db/storage/mmap_v1/dur_journalformat.h"
#include "mongo/stdx/thread.h"
#include "mongo/util/queue.h"

namespace mongo {
namespace dur {

class JournalWriter {
    MONGO_DISALLOW_COPYING(JournalWriter);

public:
    class Buffer {
    public:
        Buffer(size_t initialSize);
        ~Buffer();

        JSectHeader& getHeader() {
            return _header;
        }
        AlignedBuilder& getBuilder() {
            return _builder;
        }

        void setNoop() {
            _isNoop = true;
        }

        JournalListener::Token journalListenerToken;

    private:
        friend class BufferGuard;
        friend class JournalWriter;


        void _assertEmpty();
        void _reset();
        void _setShutdown() {
            _isShutdown = true;
        }

        // Specifies the commit number which flushing this buffer would notify. This value is
        // zero, if there is no data to be flushed or if the buffer is noop/shutdown.
        CommitNotifier::When _commitNumber;

        // Special buffer that's posted when there is nothing to be written to the journal,
        // but we want to order a notification so it happens after all other writes have
        // completed.
        bool _isNoop;

        // Special buffer that's posted when the receiving thread must terminate. This should
        // be the last entry posted to the queue and the commit number should be zero.
        bool _isShutdown;

        JSectHeader _header;
        AlignedBuilder _builder;
    };


    JournalWriter(CommitNotifier* commitNotify,
                  CommitNotifier* applyToDataFilesNotify,
                  size_t numBuffers);
    ~JournalWriter();

    void start();

    void shutdown();

    void assertIdle();

    Buffer* newBuffer();

    void writeBuffer(Buffer* buffer, CommitNotifier::When commitNumber);

    void flush();

private:
    friend class BufferGuard;

    typedef BlockingQueue<Buffer*> BufferQueue;

    // Start all buffers with 4MB of size
    enum { InitialBufferSizeBytes = 4 * 1024 * 1024 };


    void _journalWriterThread();


    // This gets notified as journal buffers are written. It is not owned and needs to outlive
    // the journal writer object.
    CommitNotifier* const _commitNotify;

    // This gets notified as journal buffers are done being applied to the shared view
    CommitNotifier* const _applyToDataFilesNotify;

    // Wraps and controls the journal writer thread
    stdx::thread _journalWriterThreadHandle;

    // Indicates that shutdown has been requested. Used for idempotency of the shutdown call.
    bool _shutdownRequested;

    // Queue of buffers, which need to be written by the journal writer thread
    BufferQueue _journalQueue;
    CommitNotifier::When _lastCommitNumber;

    // Queue of buffers, whose write has been completed by the journal writer thread.
    BufferQueue _readyQueue;
};

}  // namespace dur
}  // namespace mongo