array_composer_impl.hpp

Go to the documentation of this file.

/*
 * Copyright (c) 2014-2015, Hewlett-Packard Development Company, LP.
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option)
 * any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details. You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 * HP designates this particular file as subject to the "Classpath" exception
 * as provided by HP in the LICENSE.txt file that accompanied this code.
 */
#ifndef FOEDUS_STORAGE_ARRAY_ARRAY_COMPOSER_IMPL_HPP_
#define FOEDUS_STORAGE_ARRAY_ARRAY_COMPOSER_IMPL_HPP_

#include <stdint.h>

#include <iosfwd>
#include <string>

#include "foedus/compiler.hpp"
#include "foedus/fwd.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/storage/composer.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/array/array_id.hpp"
#include "foedus/storage/array/array_route.hpp"
#include "foedus/storage/array/array_storage.hpp"
#include "foedus/storage/array/fwd.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace storage {
namespace array {
class ArrayComposer final {
 public:
  explicit ArrayComposer(Composer *parent);

  std::string to_string() const;

  ErrorStack compose(const Composer::ComposeArguments& args);
  ErrorStack construct_root(const Composer::ConstructRootArguments& args);
  Composer::DropResult  drop_volatiles(const Composer::DropVolatilesArguments& args);
  void                  drop_root_volatile(const Composer::DropVolatilesArguments& args);

 private:
  Engine* const             engine_;
  const StorageId           storage_id_;
  const ArrayStorage        storage_;

  ArrayPage*  resolve_volatile(VolatilePagePointer pointer);
  Composer::DropResult drop_volatiles_recurse(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer);
  Composer::DropResult drop_volatiles_intermediate(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer,
    ArrayPage* volatile_page);
  Composer::DropResult drop_volatiles_leaf(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer,
    ArrayPage* volatile_page);
  bool is_to_keep_volatile(uint16_t level);
  void drop_all_recurse(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer);
};

struct ArrayRootInfoPage final {
  PageHeader          header_;

  SnapshotPagePointer pointers_[kInteriorFanout];

  char                filler_[
    kPageSize
    - sizeof(PageHeader)
    - kInteriorFanout * sizeof(SnapshotPagePointer)];
};

class ArrayComposeContext {
 public:
  ArrayComposeContext(
    Engine*                           engine,
    snapshot::MergeSort*              merge_sort,
    snapshot::SnapshotWriter*         snapshot_writer,
    cache::SnapshotFileSet*           previous_snapshot_files,
    Page*                             root_info_page);

  ErrorStack execute();

 private:
  ErrorStack initialize(ArrayOffset initial_offset);
  ErrorStack init_root_page();

  void apply_batch(uint64_t cur, uint64_t next);

  ErrorStack execute_single_level_array();
  ErrorStack finalize();

  ErrorCode update_cur_path(ArrayOffset next_offset);

  ErrorCode read_or_init_page(
    SnapshotPagePointer old_page_id,
    SnapshotPagePointer new_page_id,
    uint8_t level,
    ArrayRange range,
    ArrayPage* page) ALWAYS_INLINE;

  ErrorCode create_empty_pages(ArrayOffset from, ArrayOffset to);
  ErrorCode create_empty_pages_recurse(ArrayOffset from, ArrayOffset to, ArrayPage* page);
  ErrorCode create_empty_intermediate_page(ArrayPage* parent, uint16_t index, ArrayRange range);
  ErrorCode create_empty_leaf_page(ArrayPage* parent, uint16_t index, ArrayRange range);

  ErrorCode expand_intermediate_pool_if_needed() ALWAYS_INLINE;

  ErrorStack install_snapshot_pointers(
    SnapshotPagePointer snapshot_base,
    uint64_t* installed_count) const;
  ErrorCode install_snapshot_pointers_recurse(
    SnapshotPagePointer snapshot_base,
    const memory::GlobalVolatilePageResolver& resolver,
    const ArrayPage* snapshot_page,
    ArrayPage* volatile_page,
    uint64_t* installed_count) const;

  ErrorCode dump_leaf_pages();
  bool verify_cur_path() const;
  bool verify_snapshot_pointer(storage::SnapshotPagePointer pointer);

  ArrayRange to_leaf_range(ArrayOffset offset) const {
    ArrayOffset begin = offset / offset_intervals_[0] * offset_intervals_[0];
    return ArrayRange(begin, begin + offset_intervals_[0], storage_.get_array_size());
  }
  bool is_initial_snapshot() const { return previous_root_page_pointer_ == 0; }

  uint16_t get_root_children() const;

  // these properties are initialized in constructor and never changed afterwards
  Engine* const                   engine_;
  snapshot::MergeSort* const      merge_sort_;
  const Epoch                     system_initial_epoch_;
  const StorageId                 storage_id_;
  const snapshot::SnapshotId      snapshot_id_;
  const ArrayStorage              storage_;
  snapshot::SnapshotWriter* const snapshot_writer_;
  cache::SnapshotFileSet*  const  previous_snapshot_files_;
  ArrayRootInfoPage* const        root_info_page_;

  const uint16_t                  payload_size_;
  const uint8_t                   levels_;
  const SnapshotPagePointer       previous_root_page_pointer_;

  uint64_t                        offset_intervals_[kMaxLevels];

  ArrayPage*                cur_path_[kMaxLevels];

  uint32_t                  allocated_pages_;
  uint32_t                  allocated_intermediates_;
  uint32_t                  max_pages_;
  uint32_t                  max_intermediates_;
  ArrayPage*                page_base_;
  ArrayPage*                intermediate_base_;

  const ArrayPartitionerData* partitioning_data_;
};

static_assert(sizeof(ArrayRootInfoPage) == kPageSize, "incorrect sizeof(RootInfoPage)");
}  // namespace array
}  // namespace storage
}  // namespace foedus
#endif  // FOEDUS_STORAGE_ARRAY_ARRAY_COMPOSER_IMPL_HPP_