snapshot_manager_pimpl.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_SNAPSHOT_SNAPSHOT_MANAGER_PIMPL_HPP_
#define FOEDUS_SNAPSHOT_SNAPSHOT_MANAGER_PIMPL_HPP_

#include <atomic>
#include <chrono>
#include <map>
#include <string>
#include <thread>
#include <vector>

#include "foedus/epoch.hpp"
#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/fs/path.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/snapshot/log_gleaner_resource.hpp"
#include "foedus/snapshot/snapshot.hpp"
#include "foedus/snapshot/snapshot_id.hpp"
#include "foedus/soc/shared_memory_repo.hpp"
#include "foedus/soc/shared_mutex.hpp"
#include "foedus/soc/shared_polling.hpp"
#include "foedus/thread/condition_variable_impl.hpp"

namespace foedus {
namespace snapshot {

struct LogGleanerControlBlock {
  // this is backed by shared memory. not instantiation. just reinterpret_cast.
  LogGleanerControlBlock() = delete;
  ~LogGleanerControlBlock() = delete;

  void initialize() {
    clear_counts();
    mappers_count_ = 0;
    reducers_count_ = 0;
    all_count_ = 0;
    terminating_ = false;
  }
  void uninitialize() {
  }
  void clear_counts() {
    cur_snapshot_.clear();
    completed_count_ = 0;
    completed_mapper_count_ = 0;
    error_count_ = 0;
    exit_count_ = 0;
    gleaning_ = false;
    cancelled_ = false;
  }

  bool is_error() const { return error_count_ > 0 || cancelled_ || terminating_; }

  std::atomic<bool>               gleaning_;
  std::atomic<bool>               cancelled_;
  std::atomic<bool>               terminating_;

  Snapshot                        cur_snapshot_;

  std::atomic<uint16_t>           completed_count_;

  std::atomic<uint16_t>           completed_mapper_count_;

  std::atomic<uint16_t>           error_count_;

  std::atomic<uint16_t>           exit_count_;

  uint16_t                        mappers_count_;
  uint16_t                        reducers_count_;
  uint16_t                        all_count_;
};

struct SnapshotManagerControlBlock {
  // this is backed by shared memory. not instantiation. just reinterpret_cast.
  SnapshotManagerControlBlock() = delete;
  ~SnapshotManagerControlBlock() = delete;

  void initialize() {
    snapshot_taken_.initialize();
    snapshot_wakeup_.initialize();
    snapshot_children_wakeup_.initialize();
    gleaner_.initialize();
    requested_snapshot_epoch_.store(Epoch::kEpochInvalid);
  }
  void uninitialize() {
    gleaner_.uninitialize();
  }

  Epoch get_snapshot_epoch() const { return Epoch(snapshot_epoch_.load()); }
  Epoch get_snapshot_epoch_weak() const {
    return Epoch(snapshot_epoch_.load(std::memory_order_relaxed));
  }
  SnapshotId get_previous_snapshot_id() const { return previous_snapshot_id_.load(); }
  SnapshotId get_previous_snapshot_id_weak() const {
    return previous_snapshot_id_.load(std::memory_order_relaxed);
  }
  Epoch get_requested_snapshot_epoch() const { return Epoch(requested_snapshot_epoch_.load()); }

  void wakeup_snapshot_children() {
    assorted::memory_fence_release();
    snapshot_children_wakeup_.signal();
  }

  std::atomic< Epoch::EpochInteger >  snapshot_epoch_;

  std::atomic< Epoch::EpochInteger >  requested_snapshot_epoch_;


  std::atomic<SnapshotId>         previous_snapshot_id_;

  soc::SharedPolling              snapshot_taken_;

  soc::SharedPolling              snapshot_wakeup_;

  soc::SharedPolling              snapshot_children_wakeup_;

  LogGleanerControlBlock          gleaner_;
};

class SnapshotManagerPimpl final : public DefaultInitializable {
 public:
  SnapshotManagerPimpl() = delete;
  explicit SnapshotManagerPimpl(Engine* engine)
    : engine_(engine), local_reducer_(nullptr) {}
  ErrorStack  initialize_once() override;
  ErrorStack  uninitialize_once() override;

  const SnapshotOptions& get_option() const;

  Epoch get_snapshot_epoch() const { return control_block_->get_snapshot_epoch(); }
  Epoch get_snapshot_epoch_weak() const  { return control_block_->get_snapshot_epoch_weak(); }

  SnapshotId get_previous_snapshot_id() const { return control_block_->get_previous_snapshot_id(); }
  SnapshotId get_previous_snapshot_id_weak() const  {
    return control_block_->get_previous_snapshot_id_weak();
  }

  ErrorStack read_snapshot_metadata(SnapshotId snapshot_id, SnapshotMetadata* out);

  void    trigger_snapshot_immediate(
    bool wait_completion,
    Epoch suggested_snapshot_epoch);
  void    stop_snapshot_thread();

  SnapshotId issue_next_snapshot_id() {
    if (control_block_->previous_snapshot_id_ == kNullSnapshotId) {
      control_block_->previous_snapshot_id_ = 1;
    } else {
      control_block_->previous_snapshot_id_ = increment(control_block_->previous_snapshot_id_);
    }
    return control_block_->previous_snapshot_id_;
  }

  void wakeup();
  void sleep_a_while();
  bool is_stop_requested() const { return stop_requested_; }
  bool is_gleaning() const { return control_block_->gleaner_.gleaning_; }

  void        handle_snapshot();
  ErrorStack  handle_snapshot_triggered(Snapshot *new_snapshot);

  void        handle_snapshot_child();

  ErrorStack  glean_logs(
    const Snapshot& new_snapshot,
    std::map<storage::StorageId, storage::SnapshotPagePointer>* new_root_page_pointers);

  ErrorStack  snapshot_metadata(
    const Snapshot& new_snapshot,
    const std::map<storage::StorageId, storage::SnapshotPagePointer>& new_root_page_pointers);

  ErrorStack  snapshot_savepoint(const Snapshot& new_snapshot);

  ErrorStack  drop_volatile_pages(
    const Snapshot& new_snapshot,
    const std::map<storage::StorageId, storage::SnapshotPagePointer>& new_root_page_pointers);
  void        drop_volatile_pages_parallel(
    const Snapshot& new_snapshot,
    const std::map<storage::StorageId, storage::SnapshotPagePointer>& new_root_page_pointers,
    void* result_memory,
    uint16_t parallel_id);

  fs::Path    get_snapshot_metadata_file_path(SnapshotId snapshot_id) const;

  Engine* const           engine_;

  SnapshotManagerControlBlock*  control_block_;

  std::vector< Snapshot >   snapshots_;

  std::atomic<bool>         stop_requested_;

  std::thread               snapshot_thread_;

  std::chrono::system_clock::time_point   previous_snapshot_time_;

  std::vector<LogMapper*>     local_mappers_;
  LogReducer*                 local_reducer_;

  LogGleanerResource          gleaner_resource_;
};

static_assert(
  sizeof(SnapshotManagerControlBlock) <= soc::GlobalMemoryAnchors::kSnapshotManagerMemorySize,
  "SnapshotManagerControlBlock is too large.");

}  // namespace snapshot
}  // namespace foedus
#endif  // FOEDUS_SNAPSHOT_SNAPSHOT_MANAGER_PIMPL_HPP_