<commit_before>// Copyright 2010-2013 RethinkDB, all rights reserved.
#define __STDC_FORMAT_MACROS
#include "serializer/log/data_block_manager.hpp"

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

#include "arch/runtime/coroutines.hpp"
#include "concurrency/mutex.hpp"
#include "perfmon/perfmon.hpp"
#include "serializer/log/log_serializer.hpp"

// TODO: Right now we perform garbage collection via the do_write() interface on the
// log_serializer_t. This leads to bugs in a couple of ways:
//
// 1. We have to be sure to get the metadata (repli timestamp, delete bit) right. The
//    data block manager shouldn't have to know about that stuff.
//
// 2. We have to special-case the serializer so that it allows us to submit
//    do_write()s during shutdown. If there were an alternative interface, it could
//    ignore or refuse our GC requests when it is shutting down.

// Later, rewrite this so that we have a special interface through which to order
// garbage collection.

data_block_manager_t::data_block_manager_t(const log_serializer_dynamic_config_t *_dynamic_config, extent_manager_t *em, log_serializer_t *_serializer, const log_serializer_on_disk_static_config_t *_static_config, log_serializer_stats_t *_stats)
    : stats(_stats), shutdown_callback(NULL), state(state_unstarted), dynamic_config(_dynamic_config),
      static_config(_static_config), extent_manager(em), serializer(_serializer),
      next_active_extent(0), gc_state(), gc_stats(stats)
{
    rassert(dynamic_config);
    rassert(static_config);
    rassert(extent_manager);
    rassert(serializer);
}

data_block_manager_t::~data_block_manager_t() {
    rassert(state == state_unstarted || state == state_shut_down);
}

void data_block_manager_t::prepare_initial_metablock(metablock_mixin_t *mb) {
    for (int i = 0; i < MAX_ACTIVE_DATA_EXTENTS; i++) {
        mb->active_extents[i] = NULL_OFFSET;
        mb->blocks_in_active_extent[i] = 0;
    }
}

void data_block_manager_t::start_reconstruct() {
    rassert(state == state_unstarted);
    gc_state.set_step(gc_reconstruct);
}

// Marks the block at the given offset as alive, in the appropriate
// gc_entry_t in the entries table.  (This is used when we start up, when
// everything is presumed to be garbage, until we mark it as
// non-garbage.)
void data_block_manager_t::mark_live(int64_t offset, uint32_t ser_block_size) {
    int extent_id = static_config->extent_index(offset);

    if (entries.get(extent_id) == NULL) {
        rassert(gc_state.step() == gc_reconstruct);  // This is called at startup.

        gc_entry_t *entry = new gc_entry_t(this, extent_id * extent_manager->extent_size);
        entry->state = gc_entry_t::state_reconstructing;
        reconstructed_extents.push_back(entry);
    }

    gc_entry_t *entry = entries.get(extent_id);
    entry->mark_live_indexwise_with_offset(offset, ser_block_size);
}

void data_block_manager_t::end_reconstruct() {
    rassert(state == state_unstarted);
    rassert(gc_state.step() == gc_reconstruct);
    gc_state.set_step(gc_ready);
}

void data_block_manager_t::start_existing(file_t *file, metablock_mixin_t *last_metablock) {
    rassert(state == state_unstarted);
    dbfile = file;
    gc_io_account_nice.init(new file_account_t(file, GC_IO_PRIORITY_NICE));
    gc_io_account_high.init(new file_account_t(file, GC_IO_PRIORITY_HIGH));

    /* Reconstruct the active data block extents from the metablock. */
    for (unsigned int i = 0; i < MAX_ACTIVE_DATA_EXTENTS; i++) {
        int64_t offset = last_metablock->active_extents[i];

        if (offset != NULL_OFFSET) {
            /* It is possible to have an active data block extent with no actual data
            blocks in it. In this case we would not have created a gc_entry_t for the
            extent yet. */
            if (entries.get(offset / extent_manager->extent_size) == NULL) {
                gc_entry_t *e = new gc_entry_t(this, offset);
                e->state = gc_entry_t::state_reconstructing;
                reconstructed_extents.push_back(e);
            }

            active_extents[i] = entries.get(offset / extent_manager->extent_size);
            rassert(active_extents[i]);

            /* Turn the extent from a reconstructing extent into an active extent */
            rassert(active_extents[i]->state == gc_entry_t::state_reconstructing);
            active_extents[i]->state = gc_entry_t::state_active;
            reconstructed_extents.remove(active_extents[i]);

            blocks_in_active_extent[i] = last_metablock->blocks_in_active_extent[i];
        } else {
            active_extents[i] = NULL;
        }
    }

    /* Convert any extents that we found live blocks in, but that are not active
    extents, into old extents */
    while (gc_entry_t *entry = reconstructed_extents.head()) {
        reconstructed_extents.remove(entry);

        rassert(entry->state == gc_entry_t::state_reconstructing);
        entry->state = gc_entry_t::state_old;

        entry->our_pq_entry = gc_pq.push(entry);

        gc_stats.old_total_block_bytes += static_config->extent_size();
        gc_stats.old_garbage_block_bytes += entry->garbage_bytes();
    }

    state = state_ready;
}

class dbm_read_ahead_fsm_t : public iocallback_t {
public:
    data_block_manager_t *parent;
    iocallback_t *callback;
    int64_t extent;
    void *read_ahead_buf;
    int64_t read_ahead_size;
    int64_t read_ahead_offset;
    int64_t off_in;
    void *buf_out;

    dbm_read_ahead_fsm_t(data_block_manager_t *p, int64_t _off_in, void *_buf_out, file_account_t *io_account, iocallback_t *cb)
        : parent(p), callback(cb), read_ahead_buf(NULL), off_in(_off_in), buf_out(_buf_out)
    {
        extent = floor_aligned(off_in, parent->static_config->extent_size());

        // Read up to MAX_READ_AHEAD_BLOCKS blocks
        read_ahead_size = std::min<int64_t>(parent->static_config->extent_size(), MAX_READ_AHEAD_BLOCKS * int64_t(parent->static_config->block_size().ser_value()));
        // We divide the extent into chunks of size read_ahead_size, then select the
        // one which contains off_in
        read_ahead_offset = extent + (off_in - extent) / read_ahead_size * read_ahead_size;
        read_ahead_buf = malloc_aligned(read_ahead_size, DEVICE_BLOCK_SIZE);
        parent->dbfile->read_async(read_ahead_offset, read_ahead_size, read_ahead_buf, io_account, this);
    }

    void on_io_complete() {
        rassert(off_in >= read_ahead_offset);
        rassert(off_in < read_ahead_offset + read_ahead_size);
        rassert(divides(parent->static_config->block_size().ser_value(), off_in - read_ahead_offset));

        // Walk over the read ahead buffer and copy stuff...
        for (int64_t current_block = 0; current_block * parent->static_config->block_size().ser_value() < read_ahead_size; ++current_block) {

            // RSI: Probably should not use block_size() here...
            const char *current_buf = reinterpret_cast<char *>(read_ahead_buf) + (current_block * parent->static_config->block_size().ser_value());

            // RSI: Probably should not use block_size() here...
            const int64_t current_offset = read_ahead_offset + (current_block * parent->static_config->block_size().ser_value());

            // Copy either into buf_out or create a new buffer for read ahead<commit_msg>Made all the fields of dbm_read_ahead_fsm_t be const.<commit_after>// Copyright 2010-2013 RethinkDB, all rights reserved.
#define __STDC_FORMAT_MACROS
#include "serializer/log/data_block_manager.hpp"

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

#include "arch/runtime/coroutines.hpp"
#include "concurrency/mutex.hpp"
#include "perfmon/perfmon.hpp"
#include "serializer/log/log_serializer.hpp"

// TODO: Right now we perform garbage collection via the do_write() interface on the
// log_serializer_t. This leads to bugs in a couple of ways:
//
// 1. We have to be sure to get the metadata (repli timestamp, delete bit) right. The
//    data block manager shouldn't have to know about that stuff.
//
// 2. We have to special-case the serializer so that it allows us to submit
//    do_write()s during shutdown. If there were an alternative interface, it could
//    ignore or refuse our GC requests when it is shutting down.

// Later, rewrite this so that we have a special interface through which to order
// garbage collection.

data_block_manager_t::data_block_manager_t(const log_serializer_dynamic_config_t *_dynamic_config, extent_manager_t *em, log_serializer_t *_serializer, const log_serializer_on_disk_static_config_t *_static_config, log_serializer_stats_t *_stats)
    : stats(_stats), shutdown_callback(NULL), state(state_unstarted), dynamic_config(_dynamic_config),
      static_config(_static_config), extent_manager(em), serializer(_serializer),
      next_active_extent(0), gc_state(), gc_stats(stats)
{
    rassert(dynamic_config);
    rassert(static_config);
    rassert(extent_manager);
    rassert(serializer);
}

data_block_manager_t::~data_block_manager_t() {
    rassert(state == state_unstarted || state == state_shut_down);
}

void data_block_manager_t::prepare_initial_metablock(metablock_mixin_t *mb) {
    for (int i = 0; i < MAX_ACTIVE_DATA_EXTENTS; i++) {
        mb->active_extents[i] = NULL_OFFSET;
        mb->blocks_in_active_extent[i] = 0;
    }
}

void data_block_manager_t::start_reconstruct() {
    rassert(state == state_unstarted);
    gc_state.set_step(gc_reconstruct);
}

// Marks the block at the given offset as alive, in the appropriate
// gc_entry_t in the entries table.  (This is used when we start up, when
// everything is presumed to be garbage, until we mark it as
// non-garbage.)
void data_block_manager_t::mark_live(int64_t offset, uint32_t ser_block_size) {
    int extent_id = static_config->extent_index(offset);

    if (entries.get(extent_id) == NULL) {
        rassert(gc_state.step() == gc_reconstruct);  // This is called at startup.

        gc_entry_t *entry = new gc_entry_t(this, extent_id * extent_manager->extent_size);
        entry->state = gc_entry_t::state_reconstructing;
        reconstructed_extents.push_back(entry);
    }

    gc_entry_t *entry = entries.get(extent_id);
    entry->mark_live_indexwise_with_offset(offset, ser_block_size);
}

void data_block_manager_t::end_reconstruct() {
    rassert(state == state_unstarted);
    rassert(gc_state.step() == gc_reconstruct);
    gc_state.set_step(gc_ready);
}

void data_block_manager_t::start_existing(file_t *file, metablock_mixin_t *last_metablock) {
    rassert(state == state_unstarted);
    dbfile = file;
    gc_io_account_nice.init(new file_account_t(file, GC_IO_PRIORITY_NICE));
    gc_io_account_high.init(new file_account_t(file, GC_IO_PRIORITY_HIGH));

    /* Reconstruct the active data block extents from the metablock. */
    for (unsigned int i = 0; i < MAX_ACTIVE_DATA_EXTENTS; i++) {
        int64_t offset = last_metablock->active_extents[i];

        if (offset != NULL_OFFSET) {
            /* It is possible to have an active data block extent with no actual data
            blocks in it. In this case we would not have created a gc_entry_t for the
            extent yet. */
            if (entries.get(offset / extent_manager->extent_size) == NULL) {
                gc_entry_t *e = new gc_entry_t(this, offset);
                e->state = gc_entry_t::state_reconstructing;
                reconstructed_extents.push_back(e);
            }

            active_extents[i] = entries.get(offset / extent_manager->extent_size);
            rassert(active_extents[i]);

            /* Turn the extent from a reconstructing extent into an active extent */
            rassert(active_extents[i]->state == gc_entry_t::state_reconstructing);
            active_extents[i]->state = gc_entry_t::state_active;
            reconstructed_extents.remove(active_extents[i]);

            blocks_in_active_extent[i] = last_metablock->blocks_in_active_extent[i];
        } else {
            active_extents[i] = NULL;
        }
    }

    /* Convert any extents that we found live blocks in, but that are not active
    extents, into old extents */
    while (gc_entry_t *entry = reconstructed_extents.head()) {
        reconstructed_extents.remove(entry);

        rassert(entry->state == gc_entry_t::state_reconstructing);
        entry->state = gc_entry_t::state_old;

        entry->our_pq_entry = gc_pq.push(entry);

        gc_stats.old_total_block_bytes += static_config->extent_size();
        gc_stats.old_garbage_block_bytes += entry->garbage_bytes();
    }

    state = state_ready;
}

class dbm_read_ahead_fsm_t : public iocallback_t {
public:
    data_block_manager_t *const parent;
    iocallback_t *const callback;
    void *const buf_out;
    const int64_t off_in;
    const int64_t extent;
    const int64_t read_ahead_size;
    const int64_t read_ahead_offset;
    void *const read_ahead_buf;

    // RSI: Basically all block_size() calls here.
    // RSI: MAX_READ_AHEAD_BLOCKS is an obsoletely named constant.

    dbm_read_ahead_fsm_t(data_block_manager_t *p, int64_t _off_in, void *_buf_out,
                         file_account_t *io_account, iocallback_t *cb)
        : parent(p),
          callback(cb),
          buf_out(_buf_out),
          off_in(_off_in),
          extent(floor_aligned(off_in, parent->static_config->extent_size())),
          read_ahead_size(std::min<int64_t>(parent->static_config->extent_size(), MAX_READ_AHEAD_BLOCKS * static_cast<int64_t>(parent->static_config->block_size().ser_value()))),
          // We divide the extent into chunks of size read_ahead_size, then select the
          // one which contains off_in
          read_ahead_offset(extent + (off_in - extent) / read_ahead_size * read_ahead_size),
          read_ahead_buf(malloc_aligned(read_ahead_size, DEVICE_BLOCK_SIZE)) {
        parent->dbfile->read_async(read_ahead_offset, read_ahead_size, read_ahead_buf, io_account, this);
    }

    void on_io_complete() {
        rassert(off_in >= read_ahead_offset);
        rassert(off_in < read_ahead_offset + read_ahead_size);
        rassert(divides(parent->static_config->block_size().ser_value(), off_in - read_ahead_offset));

        // Walk over the read ahead buffer and copy stuff...
        for (int64_t current_block = 0; current_block * parent->static_config->block_size().ser_value() < read_ahead_size; ++current_block) {

            // RSI: Probably should not use block_size() here...
            const char *current_buf = reinterpret_cast<char *>(read_ahead_buf) + (current_block * parent->static_config->block_size().ser_value());

            // RSI: Probably should not use block_size() here...
            const int64_t current_offset = read_ahead_offset + (current_block * parent->static_config->block_size().ser_value());

            // Copy either into buf_out or create a new buffer for read ahead
{'fraction_non_alphanumeric': 0.09980470065324264, 'fraction_numerical': 0.005252878981749613, 'mean_word_length': 3.6610169491525424, 'pattern_counts': {'":': 0, '<': 18, '<?xml version=': 0, '>': 80, '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': '3093198', 'n_tokens_mistral': 4970, 'n_tokens_neox': 4767, 'n_words': 1087}