block_index.cpp

#include "block_index.hpp"
#include "logic_error.hpp"
#include "block.hpp"
#include "drive.hpp"
#include <algorithm>

namespace AppenDB {

  void BlockIndex::initialize(uint32_t* p_data, size_t p_data_length, uint8_t filter, const BlockIndexImport &block_import) {
    data = p_data;
    data_length = p_data_length;
    shard_thread_filter = filter;
    num_available_blocks = 0;
    next_block_position = 0;
    ::std::fill(first_free_data_region_by_size, first_free_data_region_by_size + 31, 0xFFFFFFFF);
    for (size_t i = 0; i < 1024; ++i) {
      if (block_import.tables_used[i]) {
        ::std::fill(p_data + (i << RESPONSIBLE_BLOCK_BITS), p_data + ((i + 1) << RESPONSIBLE_BLOCK_BITS), 0xFFFFFFFD);
      }
    }
    ::std::fill(p_data + p_data_length - 4, p_data + p_data_length, 0xFFFFFFFD);
    _free_block(0x4000000 >> INSIGNIFICANT_BITS, p_data_length - 4);
    for (const auto& i : block_import.data) {
      uint32_t block_info = i.first;
      const TableIndex table_index = static_cast<TableIndex>(block_info >> 6) & 0x3ff;
      const uint16_t shard = static_cast<uint16_t>(block_info >> 16);
      if ((shard & Constants::SHARD_THREAD_MASK) == shard_thread_filter) {
        for (auto& j : i.second) {
          add_block(table_index, shard, j.second);
        }
      }
    }
    for (uint32_t index : block_import.free_blocks) {
      uint32_t block_page = index / Constants::BLOCKS_PER_PAGE;
      auto owner = block_import.owners.find(block_page);
      if (owner != block_import.owners.end() && (*owner).first == shard_thread_filter) {
        available_blocks[num_available_blocks] = index;
        num_available_blocks++;
        if (num_available_blocks == 256) break;
      }
    }
  }

  void BlockIndex::_free_block_high_aligned(uint32_t start, uint32_t end, size_t alignment){
    const uint32_t length = end - start;
    for(size_t i = 2; i <= alignment; ++i){
      const uint32_t single_bit_mask = 1 << i;
      if(length & single_bit_mask) _return_data_block((length & (0xFFFFFFFF << i)) + start - single_bit_mask, i);
    }
  }

  void BlockIndex::_free_block_low_aligned(uint32_t start, uint32_t end, size_t alignment){
    const uint32_t length = end - start;
    for(size_t i = 2; i <= alignment; ++i){
      const uint32_t single_bit_mask = 1 << i;
      if(length & single_bit_mask) _return_data_block(end - (length & (0xFFFFFFFF << i)), i);
    }
  }

  void BlockIndex::_free_block(uint32_t start, uint32_t end){
    const uint32_t length = end - start;
    if(length == 0) return;
    const uint32_t most_significant_bit = num_significant_bits(length) - 1;
    const uint32_t split_point = end & (0xFFFFFFFF << most_significant_bit);
    _free_block_low_aligned(start, split_point, most_significant_bit);
    _free_block_high_aligned(split_point, end, most_significant_bit);
  }

  void BlockIndex::_return_data_block(uint32_t start, size_t power_of_two_size){
    const size_t size = 1 << power_of_two_size;
    const uint32_t bookends = 0xFFFFFF00 | power_of_two_size;

    // check to see if we can merge with the block next to us
    const uint32_t lowest_bit = start & size;

    if(data[lowest_bit ? (start - 1) : (start + size)] == bookends){
      // allocate the neighbor block
      const uint32_t neighbor_start = start ^ size;
      uint32_t* neighbor_block = data +neighbor_start;
      const uint32_t neighbor_previous = neighbor_block[1];
      const uint32_t neighbor_next = neighbor_block[2];

      // if this is the first free block, set the size entry for this size data block to the next free chunk
      if(neighbor_previous == 0xFFFFFFFF) first_free_data_region_by_size[power_of_two_size] = neighbor_next;

      // otherwise, set the previous free block in the Linked List of free entries to point to the next one
      else data[neighbor_previous + 2] = neighbor_next;

      // if there is a next free block, point that to this block's previous
      if(neighbor_next != 0xFFFFFFFF) data[neighbor_next + 1] = neighbor_previous;

      // mark the larger block as free
      return _return_data_block(start ^ lowest_bit, power_of_two_size + 1);
    }

    const uint32_t next = first_free_data_region_by_size[power_of_two_size];
    const uint32_t previous = 0xFFFFFFFF;
    uint32_t *data_start = data + start;
    data_start[0] = bookends;
    data_start[1] = next;
    data_start[2] = previous;
    data_start[size - 1] = bookends;
    first_free_data_region_by_size[power_of_two_size] = start;
  }

  uint32_t BlockIndex::_allocate_data_block(size_t power_of_two_size){
    uint32_t data_index = first_free_data_region_by_size[power_of_two_size];
    if(data_index == 0xFFFFFFFF){

      // out of memory
      assert(power_of_two_size <= 30);

      uint32_t rv = _allocate_data_block(power_of_two_size + 1);
      _return_data_block(rv + (1 << power_of_two_size), power_of_two_size);
      return rv;
    }

    // point the next entry back to the beginning
    uint32_t *data_start = data + data_index;
    uint32_t next_free_block = data_start[1];
    if(next_free_block != 0xFFFFFFFF) data[next_free_block + 2] = 0xFFFFFFFF;
    first_free_data_region_by_size[power_of_two_size] = next_free_block;

    return data_index;
  }

  uint32_t BlockIndex::_allocate_string(size_t min_size, const uint32_t* __restrict source, size_t num_to_copy){
    // allocated size = (min_size + 2) rounded up to the nearest power of 2
    const size_t power_of_two_size = num_significant_bits(min_size + 1);
    const size_t allocated_size = 1 << power_of_two_size;
    const uint32_t data_index = _allocate_data_block(power_of_two_size);
    uint32_t * __restrict data_start = data + data_index;
    *data_start = num_to_copy;
    ::std::copy(source, source + num_to_copy, data_start + 1);
    ::std::fill(data_start + num_to_copy + 1, data_start + allocated_size - 1, 0xFFFFFFFF);
    *(data_start + allocated_size - 1) = 0xFFFFFFFE;
    return data_index;
  }

  void BlockIndex::write_rows(TableIndex type, uint16_t key, uint32_t block_index, const char *row_data, size_t row_data_length, uint32_t min_time, uint32_t max_time) {
    drive.block_append_data(block_index, row_data, row_data_length, min_time, max_time);
    uint32_t* head = get_head_location(type, key);
    
    // Single element inlined in the head. No need to reorder.
    if (!(*head & 0x80000000)) return;

    uint32_t *string_start = data + (*head & 0x7FFFFFFF);
    uint32_t num_items = *string_start;
    uint32_t *begin = string_start + 1;
    uint32_t *end = begin + num_items;

    uint32_t *matched_block = ::std::find(begin, end, block_index);

    // Block not found
    if (matched_block == end) return;

    uint32_t *next = matched_block + 1;
    // TODO: upper/lower bound to find the new index and then:
    //   ::std::move(match_block + 1, new_index, match_block); new_index = block_index;
    while (next != end && max_time > drive.get_block(*next).end_time_hi_32()) {
      *matched_block = *next;
      ++next;
      ++matched_block;
    }
    *matched_block = block_index;
  }

  void BlockIndex::allocate_block(TableIndex type, uint16_t key, const char *data, size_t data_length, uint32_t min_time, uint32_t max_time) {
    uint32_t new_block_index;
    do {
      if (num_available_blocks == next_block_position) {
        next_block_position = 0;
        drive.allocate_block_page(available_blocks, num_available_blocks);
        assert(num_available_blocks);
        // The current code in block_has_room relies on the following fact.
        // This should always be the case and we should try to keep it the case once we add compaction.
        assert((available_blocks[0] / Constants::BLOCKS_PER_PAGE) == (available_blocks[num_available_blocks - 1] / Constants::BLOCKS_PER_PAGE));
      }
      if (num_available_blocks == next_block_position) fatal_error("Unable to allocate block page.");
      new_block_index = available_blocks[next_block_position];
      next_block_position++;
    } while(!drive.get_block(new_block_index).is_free());
    drive.block_allocate(new_block_index, type, key, data, data_length, min_time, max_time);
    add_block(type, key, new_block_index);
  }

  bool BlockIndex::block_has_room(uint32_t block_index, size_t max_size_to_scan) {
    if(DRIVE_BLOCK_SIZE_MAX < max_size_to_scan) return false;
    size_t total_size = DRIVE_BLOCK_SIZE;
    ++block_index;
    for(;;){
      if(total_size >= max_size_to_scan) return true;

      Block block = drive.get_block(block_index);
      if(block.is_continuation() || (num_available_blocks != next_block_position && block_index >= available_blocks[next_block_position] && block_index <= available_blocks[num_available_blocks - 1] && block.is_free())) {
        ++block_index;
        total_size += DRIVE_BLOCK_SIZE;
      }

      // no more free space
      else return false;
    }
  }

  void BlockIndex::add_block(TableIndex type, uint16_t key, uint32_t block_index){
    uint32_t* head = get_head_location(type, key);
    // TODO: make resorting faster
    if(*head & 0x80000000){

      // this entry has nothing in it yet, so we can just put the entry inline
      if(*head >= 0xFFFFFF00) *head = block_index;

      // there exists a string of blocks in the data section
      else {
        const uint32_t existing_data_offset = *head & 0x7FFFFFFF;
        uint32_t* const data_string_start = data + existing_data_offset;
        uint32_t* data_string = data_string_start + *data_string_start + 1;
        uint32_t new_end_time = drive.get_block(block_index).end_time_hi_32();

        // if there is room to add another block here, add it
        if(*data_string == 0xFFFFFFFF) {
          (*data_string_start)++;
          uint32_t *iter = data_string;
          uint32_t *prev = iter - 1;
          while (prev != data_string_start && drive.get_block(*prev).end_time_hi_32() > new_end_time) {
            *iter = *prev;
            --iter;
            --prev;
          }
          *iter = block_index;
        }

        // TODO: check to see if we can extend into the next block rather than reallocating

        // otherwise, create a new block and copy over existing data plus the new entry
        else {
          size_t old_length = *data_string_start;
          const uint32_t new_data_index = _allocate_string(old_length + 1, data_string_start + 1, old_length);
          data[new_data_index]++;
          uint32_t *begin = data + new_data_index;
          uint32_t *iter = data + new_data_index + old_length + 1;
          uint32_t *prev = iter - 1;
          while (prev != begin && drive.get_block(*prev).end_time_hi_32() > new_end_time) {
            *iter = *prev;
            --iter;
            --prev;
          }
          *iter = block_index;
          _free_block(existing_data_offset, existing_data_offset + old_length + 1);
          *head = 0x80000000 | new_data_index;
        }
      }
    }

    // a single entry is present and inlined in the head
    else {
      uint32_t new_index = _allocate_string(2, head, 1);
      *head = 0x80000000 | new_index;
      data[new_index]++;
      if (drive.get_block(data[new_index + 1]).end_time_hi_32() > drive.get_block(block_index).end_time_hi_32()) {
        data[new_index + 2] = data[new_index + 1];
        data[new_index + 1] = block_index;
      } else {
        data[new_index + 2] = block_index;
      }
    }
  }

  // TODO: verify this is correct
  void BlockIndex::replace_block(TableIndex type, uint16_t key, uint32_t old_block_index, uint32_t new_block_index) {
    uint32_t* head = get_head_location(type, key);

    // the a single entry is in the head, we know that it is the old block index, but testing like this is fastest anyway
    if(*head == old_block_index){
      *head = new_block_index;
    }

    // the old block index must be in a string
    else {
      const uint32_t existing_data_offset = *head & 0x7FFFFFFF;
      uint32_t* existing_data_base = data + existing_data_offset + 1;
      for(;;){
        if(*existing_data_base == old_block_index){
          *existing_data_base = new_block_index;
          break;
        }
        else ++existing_data_base;
      }
    }
  }

  // FIXME: this code should probably handle the block_index not being present
  // TODO: verify this is correct
  void BlockIndex::remove_block(TableIndex type, uint16_t key, uint32_t block_index){
    uint32_t* head = get_head_location(type, key);

    // the entry is in the head
    if(*head == block_index){
      *head = 0xFFFFFFFD;
    }

    // the entry is in the data string
    else {
      const uint32_t existing_data_offset = *head & 0x7FFFFFFF;
      uint32_t* string_start = data + existing_data_offset;
      uint32_t* existing_data_base = string_start + 1;

      // find the item and length of used data
      size_t num_items;
      size_t match_location;
      for(num_items = 0;; ++num_items){
        if(existing_data_base[num_items] == block_index){
          match_location = num_items;
          for(;existing_data_base[num_items] <= 0xFFFFFF00;++num_items);
          break;
        }
      }

      // find the length of the data block
      size_t block_size;
      for(block_size = num_items; existing_data_base[block_size] != 0xFFFFFFFE; ++block_size);
      ++block_size;

      // this is the number of items minus one for the removed item, plus one for the needed termination entry
      const size_t needed_string_length_after_remove = num_items;

      // if we are overallocated by a factor of more than 2, 
      if(needed_string_length_after_remove < (block_size >> 1)){

        // we can cram everything in the header, so we're able to free the data string
        if(needed_string_length_after_remove == 2){
          const uint32_t remaining_block_index = existing_data_base[match_location ^ 1];
          _free_block(existing_data_offset, existing_data_offset + block_size);
          *head = remaining_block_index;
        }

        // we can free part of the block we're on
        else {
          // figure out the new size
          const size_t new_power_of_two_size = num_significant_bits(needed_string_length_after_remove);
          const size_t new_size = 1 << new_power_of_two_size;

          ::std::move(existing_data_base + match_location + 1, existing_data_base + (*string_start), existing_data_base + match_location);
          (*string_start)--;
          existing_data_base[num_items - 1] = 0xFFFFFFFF;
          existing_data_base[new_size - 1] = 0xFFFFFFFE;

          // return the unused portion to the pool of available memory
          _free_block(existing_data_offset + new_size, existing_data_offset + block_size);
        }
      }

      // we can just remove the item from the string
      else {
        ::std::move(existing_data_base + match_location + 1, existing_data_base + (*string_start), existing_data_base + match_location);
        (*string_start)--;
        existing_data_base[num_items - 1] = 0xFFFFFFFF;
      }
    }
  }

  void BlockIndex::change_block(TableIndex type, uint32_t block_index, const char* old_data, const char* new_data){
    Block old_block(old_data);
    Block new_block(new_data);

    if(old_block.is_root_block() && old_block.is_used() && ((old_block.type() != 0x3FF) || (old_block.shard_filter() == 0))){
      remove_block(type, old_block.shard_filter(), block_index);
    }

    if(new_block.is_root_block() && new_block.is_used() && ((new_block.type() != 0x3FF) || (new_block.shard_filter() == 0))){
      add_block(type, new_block.shard_filter(), block_index);
    }
  }

}