<commit_before>#include "buffer_cache/mirrored/mirrored.hpp"

#include "errors.hpp"
#include <boost/bind.hpp>

#include "arch/arch.hpp"
#include "buffer_cache/stats.hpp"
#include "do_on_thread.hpp"
#include "stats/persist.hpp"
#include "serializer/serializer.hpp"

/**
 * Buffer implementation.
 */

perfmon_counter_t pm_registered_snapshots("registered_snapshots"),
                  pm_registered_snapshot_blocks("registered_snapshot_blocks");
perfmon_sampler_t pm_snapshots_per_transaction("snapshots_per_transaction", secs_to_ticks(1));

perfmon_persistent_counter_t pm_cache_hits("cache_hits"), pm_cache_misses("cache_misses");

// Types of snapshots

// In order for snapshots to get deleted properly, they must obey the invariant that, after
// get_data() returns or get_data_if_available() returns non-NULL, get_data_if_available() will
// return the same (non-NULL) value.

// TODO (sam): What is a buf_snapshot_t, really?

// TODO (rntz): it should be possible for us to cause snapshots which were not cow-referenced to be
// flushed to disk during writeback, sans block id, to allow them to be unloaded if necessary.
class mc_inner_buf_t::buf_snapshot_t : private evictable_t, public intrusive_list_node_t<mc_inner_buf_t::buf_snapshot_t> {
public:
    buf_snapshot_t(mc_inner_buf_t *buf, version_id_t version,
                   size_t _snapshot_refcount, size_t _active_refcount,
                   void *_data, const boost::intrusive_ptr<standard_block_token_t>& _token)
        : evictable_t(buf->cache, _data ? true : false),
          parent(buf), snapshotted_version(version),
          data(_data), token(_token),
          snapshot_refcount(_snapshot_refcount), active_refcount(_active_refcount) {
        rassert(data || token, "creating buf snapshot without data or block token");
        rassert(snapshot_refcount + active_refcount, "creating buf snapshot with 0 refcount");
    }

private:
    ~buf_snapshot_t() {
        rassert(!snapshot_refcount && !active_refcount);
        parent->snapshots.remove(this);
        if (data) {
            cache->serializer->free(data);
        }
    }

public:
    void *acquire_data(file_account_t *io_account) {
        cache->assert_thread();
        ++active_refcount;

        mutex_acquisition_t m(&data_mutex);
        // The buffer might have already been loaded.
        if (data) {
            return data;
        }
        rassert(token, "buffer snapshot lacks both token and data");

        // Use a temporary to avoid putting our data member in an allocated-but-uninitialized state.
        void *tmp = cache->serializer->malloc();
        cache->serializer->block_read(token, tmp, io_account);
        rassert(!data, "data changed while holding mutex");
        data = tmp;

        return data;
    }

    void release_data() {
        rassert(active_refcount, "releasing snapshot data with no active references");
        --active_refcount;
        if (0 == active_refcount + snapshot_refcount) {
            delete this;
        }
    }

    void release() {
        rassert(snapshot_refcount, "releasing snapshot with no references");
        --snapshot_refcount;
        if (0 == snapshot_refcount + active_refcount) {
            delete this;
        }
    }

    // We are safe to unload if we are saved to disk and have no mc_buf_ts actively referencing us.
    bool safe_to_unload() {
        return bool(token) && !active_refcount;
    }

    void unload() {
        // Acquiring the mutex isn't necessary here because active_refcount == 0.
        rassert(safe_to_unload());
        rassert(data);
        cache->serializer->free(data);
        data = NULL;
    }

private:
    friend class mc_inner_buf_t;

    // I guess we know what this means.
    mc_inner_buf_t *parent;

    // Some kind of snapshotted version.  :/
    version_id_t snapshotted_version;

    mutable mutex_t data_mutex;

    // The buffer of the snapshot we hold.  TODO (sam): We own this
    // buffer but the mc_inner_buf_t does too?  Uh...  Also add
    // testing for this problem.
    void *data;

    // Our block token to the serializer.
    boost::intrusive_ptr<standard_block_token_t> token;

    // snapshot_refcount is the number of snapshots that could potentially use this buf_snapshot_t.
    size_t snapshot_refcount;

    // active_refcount is the number of mc_buf_ts currently using this buf_snapshot_t. As long as
    // >0, we cannot be unloaded.
    size_t active_refcount;

    DISABLE_COPYING(buf_snapshot_t);
};

// This loads a block from the serializer and stores it into buf.
void mc_inner_buf_t::load_inner_buf(bool should_lock, file_account_t *io_account) {
    if (should_lock) {
        bool locked UNUSED = lock.lock(rwi_write, NULL);
        rassert(locked);
    } else {
        // We should have at least *some* kind of lock on the buffer, shouldn't we?
        rassert(lock.locked());
    }

    // Read the block...
    {
        on_thread_t thread(cache->serializer->home_thread());
        subtree_recency = cache->serializer->get_recency(block_id);
        // TODO: Merge this initialization with the read itself eventually
        data_token = cache->serializer->index_read(block_id);
        cache->serializer->block_read(data_token, data, io_account);
    }

    // Read the block sequence id
    block_sequence_id = cache->serializer->get_block_sequence_id(block_id, data);

    replay_patches();

    if (should_lock) {
        lock.unlock();
    }
}

// This form of the buf constructor is used when the block exists on disk and needs to be loaded
mc_inner_buf_t::mc_inner_buf_t(cache_t *_cache, block_id_t _block_id, bool _should_load, file_account_t *_io_account)
    : evictable_t(_cache),
      writeback_t::local_buf_t(),
      block_id(_block_id),
      subtree_recency(repli_timestamp_t::invalid),  // Gets initialized by load_inner_buf
      data(_should_load ? _cache->serializer->malloc() : NULL),
      version_id(_cache->get_min_snapshot_version(_cache->get_current_version_id())),
      next_patch_counter(1),
      refcount(0),
      do_delete(false),
      cow_refcount(0),
      snap_refcount(0),
      block_sequence_id(NULL_BLOCK_SEQUENCE_ID) {

    rassert(version_id != faux_version_id);

    array_map_t::constructing_inner_buf(this);

    if (_should_load) {
        // Some things expect us to return immediately (as of 5/12/2011), so we do the loading in a
        // separate coro. We have to make sure that load_inner_buf() acquires the lock first
        // however, so we use spawn_now().
        coro_t::spawn_now(boost::bind(&mc_inner_buf_t::load_inner_buf, this, true, _io_account));
    }

    // TODO (sam): Figure out wtf this todo is talking about.
    // TODO: only increment pm_n_blocks_in_memory when we actually load the block into memory.
    pm_n_blocks_in_memory++;
    refcount++; // Make the refcount nonzero so this block won't be considered safe to unload.

    _cache->page_repl.make_space();
    _cache->maybe_unregister_read_ahead_callback();

    refcount--;
}

// This form of the buf constructor is used when the block exists on disks but has been loaded into buf already
mc_inner_buf_t::mc_inner_buf_t(cache_t *_cache, block_id_t _block_id, void *_buf, repli_timestamp_t _recency_timestamp)
    : evictable_t(_cache),
      writeback_t::local_buf_t(),
      block_id(_block_id),
      subtree_recency(_recency_timestamp),
      data(_buf),
      version_id(_cache->get_min_snapshot_version(_cache->get_current_version_id())),
      refcount(0),
      do_delete(false),
      cow_refcount(0),
      snap_refcount(0),
      block_sequence_id(NULL_BLOCK_SEQUENCE_ID) {

    rassert(version_id != faux_version_id);

    array_map_t::constructing_inner_buf(this);

    pm_n_blocks_in_memory++;
    refcount++; // Make the refcount nonzero so this block won't be considered safe to unload.
    _cache->page_repl.make_space();
    _cache->maybe_unregister_read_ahead_callback();
    refcount--;

    // Read the block sequence id
    block_sequence_id = _cache->serializer->get_block_sequence_id(_block_id, data);

    replay_patches();

    // TODO (rntz): This should initialize data_token at some point. That however requires switching
    // to the serializer thread and we cannot afford that here, except if we lock. Maybe read ahead
    // should pass the token through to here.
}

mc_inner_buf_t *mc_inner_buf_t::allocate(cache_t *cache, version_id_t snapshot_version, repli_timestamp_t recency_timestamp) {
    cache->assert_thread();

    if (snapshot_version == faux_version_id)
        snapshot_version = cache->get_current_version_id();

    block_id_t block_id = cache->free_list.gen_block_id();
    mc_inner_buf_t *inner_buf = cache->find_buf(block_id);
    if (!inner_buf) {
        return new mc_inner_buf_t(cache, block_id, snapshot_version, recency_timestamp);
    } else {
        // Block with block_id was logically deleted, but its inner_buf survived.
        // That can happen when there are active snapshots that holding older versions
        // of the block. It's safe to update the top version of the block though.
        rassert(inner_buf->do_delete);
        rassert(inner_buf->data == NULL);

        inner_buf->subtree_recency = recency_timestamp;
        inner_buf->data = cache->serializer->malloc();
        #if !defined(NDEBUG) || defined(VALGRIND)
            // The memory allocator already filled this with 0xBD, but it's nice to be able to distinguish
            // between problems with uninitialized memory and problems with uninitialized blocks
            memset(inner_buf->data, 0xCD, cache->serializer->get_block_size().value());
        #endif
        inner_buf->version_id = snapshot_version;
        inner_buf->do_delete = false;
        inner_buf->next_patch_counter = 1;
        inner_buf->cow_refcount = 0;
        inner_buf->snap_refcount = 0;
        inner_buf->block_sequence_id = NULL_BLOCK_SEQUENCE_ID;
        inner_buf->data_token.reset();

        return inner_buf;
    }
}

// This form of the buf constructor is used when a completely new block is being created.
// Used by mc_inner_buf_t::allocate() and by the patch log.
// If you update this constructor, please don't forget to update mc_inner_buf_t::allocate
// accordingly.
mc_inner_buf_t::mc_inner_buf_t(cache_t *_cache, block_id_t _block_id, version_id_t _snapshot_version, repli_timestamp_t _recency_timestamp)
    : evictable_t(_cache),
      writeback_t::local_buf_t(),
      block_id(_block_id),
      subtree_recency(_recency_timestamp),
      data(_cache->serializer->malloc()),
      version_id(_snapshot_version),
      next_patch_counter(1),
      refcount(0),
      do_delete(false),
      cow_refcount(0),
      snap_refcount(0),
      block_sequence_id(NULL_BLOCK_SEQUENCE_ID) {

    rassert(version_id != faux_version_id);
    _cache->assert_thread();

#if !defined(NDEBUG) || defined(VALGRIND)
    // The memory allocator already filled this with 0xBD, but it's nice to be able to distinguish
    // between problems with uninitialized memory and problems with uninitialized blocks
    memset(data, 0xCD, _cache->serializer->get_block_size().value());
#endif

    array_map_t::constructing_inner_buf(this);

    pm_n_blocks_in_memory++;
    refcount++; // Make the refcount nonzero so this block won't be considered safe to unload.

    _cache->page_repl.make_space();
    _cache->maybe_unregister_read_ahead_callback();

    refcount--;
}

void mc_inner_buf_t::unload() {
    delete this;
}

mc_inner_buf_t::~mc_inner_buf_t() {
    cache->assert_thread();

#ifndef NDEBUG
    // We're about to free the data, let's set it to a recognizable
    // value to make sure we don't depend on accessing things that may
    // be flushed out of the cache.
    if (data)
        memset(data, 0xDD, cache->serializer->get_block_size().value());
#endif

    array_map_t::destroying_inner_buf(this);

    rassert(safe_to_unload());
    if (data)
        cache->serializer->free(data);

    pm_n_blocks_in_memory--;
}

void mc_inner_buf_t::replay_patches() {
    // Remove obsolete patches from diff storage<commit_msg>Added a personal comment, made a memset of a buffer happen only outside of valgrind.<commit_after>#include "buffer_cache/mirrored/mirrored.hpp"

#include "errors.hpp"
#include <boost/bind.hpp>

#include "arch/arch.hpp"
#include "buffer_cache/stats.hpp"
#include "do_on_thread.hpp"
#include "stats/persist.hpp"
#include "serializer/serializer.hpp"

/**
 * Buffer implementation.
 */

perfmon_counter_t pm_registered_snapshots("registered_snapshots"),
                  pm_registered_snapshot_blocks("registered_snapshot_blocks");
perfmon_sampler_t pm_snapshots_per_transaction("snapshots_per_transaction", secs_to_ticks(1));

perfmon_persistent_counter_t pm_cache_hits("cache_hits"), pm_cache_misses("cache_misses");

// Types of snapshots

// In order for snapshots to get deleted properly, they must obey the invariant that, after
// get_data() returns or get_data_if_available() returns non-NULL, get_data_if_available() will
// return the same (non-NULL) value.

// TODO (sam): What is a buf_snapshot_t, really?

// TODO (rntz): it should be possible for us to cause snapshots which were not cow-referenced to be
// flushed to disk during writeback, sans block id, to allow them to be unloaded if necessary.
class mc_inner_buf_t::buf_snapshot_t : private evictable_t, public intrusive_list_node_t<mc_inner_buf_t::buf_snapshot_t> {
public:
    buf_snapshot_t(mc_inner_buf_t *buf, version_id_t version,
                   size_t _snapshot_refcount, size_t _active_refcount,
                   void *_data, const boost::intrusive_ptr<standard_block_token_t>& _token)
        : evictable_t(buf->cache, _data ? true : false),
          parent(buf), snapshotted_version(version),
          data(_data), token(_token),
          snapshot_refcount(_snapshot_refcount), active_refcount(_active_refcount) {
        rassert(data || token, "creating buf snapshot without data or block token");
        rassert(snapshot_refcount + active_refcount, "creating buf snapshot with 0 refcount");
    }

private:
    ~buf_snapshot_t() {
        rassert(!snapshot_refcount && !active_refcount);
        parent->snapshots.remove(this);
        if (data) {
            cache->serializer->free(data);
        }
    }

public:
    void *acquire_data(file_account_t *io_account) {
        cache->assert_thread();
        ++active_refcount;

        mutex_acquisition_t m(&data_mutex);
        // The buffer might have already been loaded.
        if (data) {
            return data;
        }
        rassert(token, "buffer snapshot lacks both token and data");

        // Use a temporary to avoid putting our data member in an allocated-but-uninitialized state.
        void *tmp = cache->serializer->malloc();
        cache->serializer->block_read(token, tmp, io_account);
        rassert(!data, "data changed while holding mutex");
        data = tmp;

        return data;
    }

    void release_data() {
        rassert(active_refcount, "releasing snapshot data with no active references");
        --active_refcount;
        if (0 == active_refcount + snapshot_refcount) {
            delete this;
        }
    }

    void release() {
        rassert(snapshot_refcount, "releasing snapshot with no references");
        --snapshot_refcount;
        if (0 == snapshot_refcount + active_refcount) {
            delete this;
        }
    }

    // We are safe to unload if we are saved to disk and have no mc_buf_ts actively referencing us.
    bool safe_to_unload() {
        return bool(token) && !active_refcount;
    }

    void unload() {
        // Acquiring the mutex isn't necessary here because active_refcount == 0.
        rassert(safe_to_unload());
        rassert(data);
        cache->serializer->free(data);
        data = NULL;
    }

private:
    friend class mc_inner_buf_t;

    // I guess we know what this means.
    mc_inner_buf_t *parent;

    // Some kind of snapshotted version.  :/
    version_id_t snapshotted_version;

    mutable mutex_t data_mutex;

    // The buffer of the snapshot we hold.  TODO (sam): We own this
    // buffer but the mc_inner_buf_t does too?  Uh...  Also add
    // testing for this problem.
    void *data;

    // Our block token to the serializer.
    boost::intrusive_ptr<standard_block_token_t> token;

    // snapshot_refcount is the number of snapshots that could potentially use this buf_snapshot_t.
    size_t snapshot_refcount;

    // active_refcount is the number of mc_buf_ts currently using this buf_snapshot_t. As long as
    // >0, we cannot be unloaded.
    size_t active_refcount;

    DISABLE_COPYING(buf_snapshot_t);
};

// This loads a block from the serializer and stores it into buf.
void mc_inner_buf_t::load_inner_buf(bool should_lock, file_account_t *io_account) {
    if (should_lock) {
        bool locked UNUSED = lock.lock(rwi_write, NULL);
        rassert(locked);
    } else {
        // We should have at least *some* kind of lock on the buffer, shouldn't we?
        rassert(lock.locked());
    }

    // Read the block...
    {
        on_thread_t thread(cache->serializer->home_thread());
        subtree_recency = cache->serializer->get_recency(block_id);
        // TODO: Merge this initialization with the read itself eventually
        data_token = cache->serializer->index_read(block_id);
        cache->serializer->block_read(data_token, data, io_account);
    }

    // Read the block sequence id
    block_sequence_id = cache->serializer->get_block_sequence_id(block_id, data);

    replay_patches();

    if (should_lock) {
        lock.unlock();
    }
}

// This form of the buf constructor is used when the block exists on disk and needs to be loaded
mc_inner_buf_t::mc_inner_buf_t(cache_t *_cache, block_id_t _block_id, bool _should_load, file_account_t *_io_account)
    : evictable_t(_cache),
      writeback_t::local_buf_t(),
      block_id(_block_id),
      subtree_recency(repli_timestamp_t::invalid),  // Gets initialized by load_inner_buf
      data(_should_load ? _cache->serializer->malloc() : NULL),
      version_id(_cache->get_min_snapshot_version(_cache->get_current_version_id())),
      next_patch_counter(1),
      refcount(0),
      do_delete(false),
      cow_refcount(0),
      snap_refcount(0),
      block_sequence_id(NULL_BLOCK_SEQUENCE_ID) {

    rassert(version_id != faux_version_id);

    array_map_t::constructing_inner_buf(this);

    if (_should_load) {
        // Some things expect us to return immediately (as of 5/12/2011), so we do the loading in a
        // separate coro. We have to make sure that load_inner_buf() acquires the lock first
        // however, so we use spawn_now().
        coro_t::spawn_now(boost::bind(&mc_inner_buf_t::load_inner_buf, this, true, _io_account));
    }

    // TODO (sam): Figure out wtf this todo is talking about.
    // TODO: only increment pm_n_blocks_in_memory when we actually load the block into memory.
    pm_n_blocks_in_memory++;
    refcount++; // Make the refcount nonzero so this block won't be considered safe to unload.

    _cache->page_repl.make_space();
    _cache->maybe_unregister_read_ahead_callback();

    refcount--;
}

// This form of the buf constructor is used when the block exists on disks but has been loaded into buf already
mc_inner_buf_t::mc_inner_buf_t(cache_t *_cache, block_id_t _block_id, void *_buf, repli_timestamp_t _recency_timestamp)
    : evictable_t(_cache),
      writeback_t::local_buf_t(),
      block_id(_block_id),
      subtree_recency(_recency_timestamp),
      data(_buf),
      version_id(_cache->get_min_snapshot_version(_cache->get_current_version_id())),
      refcount(0),
      do_delete(false),
      cow_refcount(0),
      snap_refcount(0),
      block_sequence_id(NULL_BLOCK_SEQUENCE_ID) {

    rassert(version_id != faux_version_id);

    array_map_t::constructing_inner_buf(this);

    pm_n_blocks_in_memory++;
    refcount++; // Make the refcount nonzero so this block won't be considered safe to unload.
    _cache->page_repl.make_space();
    _cache->maybe_unregister_read_ahead_callback();
    refcount--;

    // Read the block sequence id
    block_sequence_id = _cache->serializer->get_block_sequence_id(_block_id, data);

    replay_patches();

    // TODO (sam): Look at this.
    // TODO (rntz): This should initialize data_token at some point. That however requires switching
    // to the serializer thread and we cannot afford that here, except if we lock. Maybe read ahead
    // should pass the token through to here.
}

mc_inner_buf_t *mc_inner_buf_t::allocate(cache_t *cache, version_id_t snapshot_version, repli_timestamp_t recency_timestamp) {
    cache->assert_thread();

    if (snapshot_version == faux_version_id)
        snapshot_version = cache->get_current_version_id();

    block_id_t block_id = cache->free_list.gen_block_id();
    mc_inner_buf_t *inner_buf = cache->find_buf(block_id);
    if (!inner_buf) {
        return new mc_inner_buf_t(cache, block_id, snapshot_version, recency_timestamp);
    } else {
        // Block with block_id was logically deleted, but its inner_buf survived.
        // That can happen when there are active snapshots that holding older versions
        // of the block. It's safe to update the top version of the block though.
        rassert(inner_buf->do_delete);
        rassert(inner_buf->data == NULL);

        inner_buf->subtree_recency = recency_timestamp;
        inner_buf->data = cache->serializer->malloc();
        #if !defined(NDEBUG) || defined(VALGRIND)
            // The memory allocator already filled this with 0xBD, but it's nice to be able to distinguish
            // between problems with uninitialized memory and problems with uninitialized blocks
            memset(inner_buf->data, 0xCD, cache->serializer->get_block_size().value());
        #endif
        inner_buf->version_id = snapshot_version;
        inner_buf->do_delete = false;
        inner_buf->next_patch_counter = 1;
        inner_buf->cow_refcount = 0;
        inner_buf->snap_refcount = 0;
        inner_buf->block_sequence_id = NULL_BLOCK_SEQUENCE_ID;
        inner_buf->data_token.reset();

        return inner_buf;
    }
}

// This form of the buf constructor is used when a completely new block is being created.
// Used by mc_inner_buf_t::allocate() and by the patch log.
// If you update this constructor, please don't forget to update mc_inner_buf_t::allocate
// accordingly.
mc_inner_buf_t::mc_inner_buf_t(cache_t *_cache, block_id_t _block_id, version_id_t _snapshot_version, repli_timestamp_t _recency_timestamp)
    : evictable_t(_cache),
      writeback_t::local_buf_t(),
      block_id(_block_id),
      subtree_recency(_recency_timestamp),
      data(_cache->serializer->malloc()),
      version_id(_snapshot_version),
      next_patch_counter(1),
      refcount(0),
      do_delete(false),
      cow_refcount(0),
      snap_refcount(0),
      block_sequence_id(NULL_BLOCK_SEQUENCE_ID) {

    rassert(version_id != faux_version_id);
    _cache->assert_thread();

#if !defined(NDEBUG) || defined(VALGRIND)
    // The memory allocator already filled this with 0xBD, but it's nice to be able to distinguish
    // between problems with uninitialized memory and problems with uninitialized blocks
    memset(data, 0xCD, _cache->serializer->get_block_size().value());
#endif

    array_map_t::constructing_inner_buf(this);

    pm_n_blocks_in_memory++;
    refcount++; // Make the refcount nonzero so this block won't be considered safe to unload.

    _cache->page_repl.make_space();
    _cache->maybe_unregister_read_ahead_callback();

    refcount--;
}

void mc_inner_buf_t::unload() {
    delete this;
}

mc_inner_buf_t::~mc_inner_buf_t() {
    cache->assert_thread();

#ifndef NDEBUG
#ifndef VALGRIND
    // We're about to free the data, let's set it to a recognizable
    // value to make sure we don't depend on accessing things that may
    // be flushed out of the cache.
    if (data)
        memset(data, 0xDD, cache->serializer->get_block_size().value());
#endif
#endif

    array_map_t::destroying_inner_buf(this);

    rassert(safe_to_unload());
    if (data)
        cache->serializer->free(data);

    pm_n_blocks_in_memory--;
}

void mc_inner_buf_t::replay_patches() {
    // Remove obsolete patches from diff storage
{'fraction_non_alphanumeric': 0.0960019743336624, 'fraction_numerical': 0.0026324448831852583, 'mean_word_length': 3.354047277936963, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5270565', 'n_tokens_mistral': 7679, 'n_tokens_neox': 7530, 'n_words': 1969}