disk_write_ops.cpp

#include "constants.hpp"
#include "drive.hpp"
#include "decoded_block.hpp"
#include "block.hpp"
#include "fast_hash.hpp"
#include "data.hpp"
#include "logic_error.hpp"
#include "optimized_rows.hpp"
#include "cache.hpp"
#include "block_index.hpp"
#include "shard_accumulator.hpp"

namespace AppenDB {
  // TODO: avoid blocking the main thread on a page fault

  void Drive::raw_add_data(TableIndex type, const char *data, size_t length, uint32_t time, uint32_t hash, uint16_t shard, uint32_t misc) {
    bool is_row_already_present = false;
    bool found_spot_to_write = false;
    uint32_t best_block_candidate_index = 0;
    uint32_t best_block_candidate_min_time = 0;
    uint32_t best_block_candidate_max_time = 0;
    uint32_t best_block_candidate_distance = 0xFFFFFFFF;
    uint16_t cache_index =  shard & Constants::SHARD_THREAD_MASK;
    BlockIndex &block_index = this->block_index(shard & Constants::SHARD_THREAD_MASK);

    block_index.for_each_time_match(type, shard, time, [this, &block_index, cache_index, time, hash, data, length, &is_row_already_present, &found_spot_to_write, &best_block_candidate_index, &best_block_candidate_min_time, &best_block_candidate_max_time, &best_block_candidate_distance](uint32_t matching_block_index, bool &terminate){
      const uint32_t ablock_start_time = block_start_time(matching_block_index);
      const uint32_t ablock_end_time = block_end_time(matching_block_index);

      // NOTE: so that we don't forget that we're assuming this in the function dealing with deduplicating cases where we have the same shard
      //const uint16_t ablock_shard_filter = block_shard_filter(matching_block_index);
      //assert((DEFAULT_SHARD_MASK & ablock_shard_filter) + (DEFAULT_SHARD_MASK + 1) == ablock_shard_filter);

      if((time <= ablock_end_time) && (time >= ablock_start_time)){
        DecodedBlock* decoded_block = Cache::CachePool::get(cache_index).prefetch_cacheable<DecodedBlock>(matching_block_index);
        assert(decoded_block);

        // FIXME: this assumes that we wait for the cache entry synchronously
        assert(decoded_block->is_ready());

        if(decoded_block->has_row(time, hash, data, length)){
          terminate = is_row_already_present = true;
          return;
        }
      }

      // TODO: don't put data from way later into a block (keep the various blocks more or less non-overlapping)
      if(best_block_candidate_distance) {
        uint32_t distance = time <= ablock_start_time ? ablock_start_time - time
          : time >= ablock_end_time ? time - ablock_end_time
          : 0;
        if (distance < best_block_candidate_distance && block_index.block_has_room(matching_block_index, block_data_size(matching_block_index) + length)){
          found_spot_to_write = true;
          best_block_candidate_index = matching_block_index;
          best_block_candidate_max_time = ablock_end_time;
          best_block_candidate_min_time = ablock_start_time;
          best_block_candidate_distance = distance;
        }
      }
    });

    if (is_row_already_present) return;

    uint32_t shard_mask = (shard & Constants::DEFAULT_SHARD_MASK) | Constants::DEFAULT_NUM_SHARDS;
    if(!found_spot_to_write){
      block_index.allocate_block(type, shard_mask, data, length, time, time);
    } else {
      // we can write to an existing block
      uint32_t row_start_offset = this->block_data_end(best_block_candidate_index) - this->block_data_start(best_block_candidate_index);
      block_index.write_rows(type, shard_mask, best_block_candidate_index, data, length, ::std::min(time, best_block_candidate_min_time), ::std::max(time, best_block_candidate_max_time));

      // update the cache entry for this
      DecodedBlock::update_cache_for_new_row(cache_index, best_block_candidate_index, time, hash, length, row_start_offset | misc);
    }
  }

  void Drive::add_data(TableIndex type, const char *data, size_t data_length){
    // setup the code to read the data
    // TODO: create optimized readers for various types
    Table definition = get_table(type);
    OptimizedRows::template Uncompressed<Table>::RowReader row_reader(definition);

    ShardAccumulator<Table> row(definition);
    const char *iterator = data;
    const char *end = data + data_length;
    const char *row_begin;
    size_t row_length;
    OptimizedRows::any_row(definition, row, iterator, end, [this, type, &row, &row_begin, &row_length](){
      uint32_t const time = row.time;
      const char * const data = row_begin;
      size_t const length = row_length;
      uint64_t hash = hash64(data, length);
      // FIXME: write code that detects cases where a row would be placed in a spot where it is duplicated
      // FIXME: getting this to allow multiple concurrent writes will require a bit of work
      for(size_t i = 0; i != row.num_shards; ++i){
        uint32_t const shard = row.shards[i];
        uint32_t const misc = row.misc_bloom_bits[i];
        raw_add_data(type, data, length, time, hash, shard, misc);
      }
      row.reset();
      return false;
    }, row_begin, row_length);
  }

  void Drive::add_data_to_shard(TableIndex type, const uint32_t shard, const char *data, size_t data_length) {
    // setup the code to read the data
    // TODO: create optimized readers for various types
    Table definition = get_table(type);
    OptimizedRows::template Uncompressed<Table>::RowReader row_reader(definition);
    ShardAccumulatorSingle<Table> row(shard, definition);
    const char *iterator = data;
    const char *end = data + data_length;
    const char *row_begin;
    size_t row_length;
    OptimizedRows::any_row(definition, row, iterator, end, [this, type, &row, &row_begin, &row_length, shard](){
      uint32_t const time = row.time;
      const char * const data = row_begin;
      size_t const length = row_length;
      uint64_t hash = hash64(data, length);
      raw_add_data(type, data, row_length, time, hash, shard, row.misc_bloom_bits);
      row.reset();
      return false;
    }, row_begin, row_length);
  }

  uint32_t Drive::find_compatible_block(TableIndex type, uint16_t shard_filter, uint32_t max_time_end, size_t min_data_capacity){
    bool found_spot = false;
    uint32_t best_block_candidate_index = 0;
    uint32_t best_block_candidate_min_time = 0;
    uint32_t best_block_candidate_max_time = 0;
    const char* block_entries = this->block_entries();

    // TODO: We should use for_each_match variant instead of for_each_exact_match
    block_index(shard_filter & Constants::SHARD_THREAD_MASK).for_each_exact_match(type, shard_filter, [this, block_entries, type, max_time_end, min_data_capacity, &found_spot, &best_block_candidate_index, &best_block_candidate_min_time, &best_block_candidate_max_time](uint32_t block_index, bool &terminate){
      Block block(block_entries + (block_index * 16));
      const uint32_t block_start_time = block.begin_time_hi_32();
      const uint32_t block_end_time = block.end_time_hi_32();
      // TODO: don't put data from way later into a block (keep the various blocks more or less non-overlapping)
      if((block.type() == type) && (block_end_time <= max_time_end) && ((block_end_time > best_block_candidate_max_time) || ((block_end_time == best_block_candidate_max_time) && (block_start_time > best_block_candidate_min_time))) && block_has_room(block_index, min_data_capacity)){
        found_spot = true;
        best_block_candidate_index = block_index;
        best_block_candidate_max_time = block_end_time;
        best_block_candidate_min_time = block_start_time;
      }
    });
    return best_block_candidate_index;
  }

  void Drive::add_processed_data(TableIndex type, uint16_t shard_filter, uint32_t time_begin, uint32_t time_end, const char *data, size_t data_length){
    // try to find an existing block to write to
    uint32_t best_block_candidate_index = find_compatible_block(type, shard_filter, time_begin, data_length);

    // a block was found where we can append this data, so let's do that
    if(best_block_candidate_index){
      // actually write the data to the block
      // write the update to any affected block entries
      block_append_data(best_block_candidate_index, data, data_length, time_begin, time_end);
      return;
    }

    // if none exist, create a new one
    uint32_t new_block_index = find_free_block(data_length);
    block_allocate(new_block_index, type, shard_filter, data, data_length, time_begin, time_end);
  }
}