sequential_cursor.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_SEQUENTIAL_SEQUENTIAL_CURSOR_HPP_
#define FOEDUS_STORAGE_SEQUENTIAL_SEQUENTIAL_CURSOR_HPP_

#include <stdint.h>

#include <algorithm>
#include <cstring>
#include <iosfwd>
#include <vector>

#include "foedus/assert_nd.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/epoch.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/sequential/fwd.hpp"
#include "foedus/storage/sequential/sequential_id.hpp"
#include "foedus/storage/sequential/sequential_storage.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace storage {
namespace sequential {
class SequentialCursor {
 public:
  enum OrderMode {
    kNodeFirstMode,
    kLooseEpochSortMode,
  };

  SequentialCursor(
    thread::Thread* context,
    const sequential::SequentialStorage& storage,
    void* buffer,
    uint64_t buffer_size,
    OrderMode order_mode = kNodeFirstMode,
    Epoch from_epoch = INVALID_EPOCH,
    Epoch to_epoch = INVALID_EPOCH,
    int32_t node_filter = -1);

  ~SequentialCursor();

  thread::Thread*                       get_context() const { return context_;}
  const sequential::SequentialStorage&  get_storage() const { return storage_; }

  Epoch     get_from_epoch() const { return from_epoch_; }
  Epoch     get_to_epoch() const { return to_epoch_; }

  ErrorCode next_batch(SequentialRecordIterator* out);

  bool      is_valid() const {
    return !(finished_snapshots_ && finished_safe_volatiles_ && finished_unsafe_volatiles_);
  }

  bool      is_finished_snapshots() const { return finished_snapshots_; }
  bool      is_finished_safe_volatiles() const { return finished_safe_volatiles_; }
  bool      is_finished_unsafe_volatiles() const { return finished_unsafe_volatiles_; }

  friend std::ostream& operator<<(std::ostream& o, const SequentialCursor& v);

 private:
  struct NodeState {
    explicit NodeState(uint16_t node_id);
    ~NodeState();

    const HeadPagePointer& get_cur_head() const {
      return snapshot_heads_[snapshot_cur_head_];
    }

    const uint16_t                node_id_;
    uint16_t                      volatile_cur_core_;
    uint32_t                      snapshot_cur_head_;
    uint32_t                      snapshot_cur_buffer_;
    uint32_t                      snapshot_buffered_pages_;
    uint64_t                      snapshot_buffer_begin_;
    std::vector<HeadPagePointer>  snapshot_heads_;

    std::vector<SequentialPage*>  volatile_cur_pages_;
  };

  ErrorCode init_states();

  ErrorCode next_batch_snapshot(SequentialRecordIterator* out, bool* found);
  ErrorCode next_batch_safe_volatiles(SequentialRecordIterator* out, bool* found);
  ErrorCode next_batch_unsafe_volatiles(SequentialRecordIterator* out, bool* found);

  ErrorCode buffer_snapshot_pages(uint16_t node);

  SequentialPage* resolve_volatile(VolatilePagePointer pointer) const;

  void refresh_grace_epoch();


  enum VolatileCheckPageResult {
    kValidPage,
    kNextPage,
    kNextCore,
  };

  VolatileCheckPageResult next_batch_safe_volatiles_check_page(const SequentialPage* page) const;
  VolatileCheckPageResult next_batch_unsafe_volatiles_check_page(const SequentialPage* page) const;

  thread::Thread* const               context_;
  xct::Xct* const                     xct_;
  Engine* const                       engine_;
  const memory::GlobalVolatilePageResolver& resolver_;
  sequential::SequentialStorage const storage_;
  Epoch                         from_epoch_;
  const Epoch                   to_epoch_;

  const Epoch                   latest_snapshot_epoch_;

  const Epoch                   from_epoch_volatile_;

  Epoch                         truncate_epoch_;

  const int32_t                 node_filter_;
  const uint16_t                node_count_;
  const OrderMode               order_mode_;
  bool                          snapshot_only_;
  bool                          safe_epoch_only_;

  SequentialRecordBatch* const  buffer_;
  const uint64_t                buffer_size_;
  const uint32_t                buffer_pages_;


  uint16_t                      current_node_;

  bool                          finished_snapshots_;
  bool                          finished_safe_volatiles_;
  bool                          finished_unsafe_volatiles_;

  Epoch                         grace_epoch_;

  std::vector<NodeState>        states_;
};


struct SequentialRecordBatch CXX11_FINAL {
  PageHeader            header_;            // +32 -> 32

  uint16_t              record_count_;      // +2 -> 34
  uint16_t              used_data_bytes_;   // +2 -> 36
  uint32_t              filler_;            // +4 -> 40

  DualPagePointer       next_page_;         // +16 -> 56

  char                  data_[kDataSize];


  uint16_t       get_record_count() const { return record_count_; }
  uint16_t       get_record_length(uint16_t index) const {
    ASSERT_ND(index < record_count_);
    return reinterpret_cast<const uint16_t*>(data_ + sizeof(data_))[-index - 1];
  }
  const xct::RwLockableXctId* get_owner_id_from_offset(uint16_t offset) const {
    ASSERT_ND(offset + record_count_ * sizeof(uint16_t) <= kDataSize);
    return reinterpret_cast<const xct::RwLockableXctId*>(data_ + offset);
  }
  const char*    get_payload_from_offset(uint16_t offset) const {
    ASSERT_ND(offset + record_count_ * sizeof(uint16_t) <= kDataSize);
    return data_ + offset + sizeof(xct::RwLockableXctId);
  }
  Epoch          get_epoch_from_offset(uint16_t offset) const {
    return get_owner_id_from_offset(offset)->xct_id_.get_epoch();
  }
};

class SequentialRecordIterator CXX11_FINAL {
 public:
  SequentialRecordIterator();
  SequentialRecordIterator(const SequentialRecordBatch* batch, Epoch from_epoch, Epoch to_epoch);

  void        reset() {
    std::memset(this, 0, sizeof(SequentialRecordIterator));
  }

  bool        is_valid() const ALWAYS_INLINE { return cur_record_ < record_count_; }
  void        next() ALWAYS_INLINE {
    while (true) {
      if (UNLIKELY(cur_record_ + 1U >= record_count_)) {
        cur_record_ = record_count_;
        break;
      }

      ++cur_record_;
      cur_offset_ += assorted::align8(cur_record_length_) + sizeof(xct::RwLockableXctId);
      cur_record_length_ = batch_->get_record_length(cur_record_);
      cur_record_epoch_ = batch_->get_epoch_from_offset(cur_offset_);
      ASSERT_ND(cur_record_epoch_.is_valid());
      if (LIKELY(in_epoch_range(cur_record_epoch_))) {
        break;
      }
      // we have to skip this record
      ++stat_skipped_records_;
    }
  }
  uint16_t    get_cur_record_length() const ALWAYS_INLINE {
    ASSERT_ND(is_valid());
    return cur_record_length_;
  }
  Epoch       get_cur_record_epoch() const ALWAYS_INLINE {
    ASSERT_ND(is_valid());
    return cur_record_epoch_;
  }
  void        copy_cur_record(char* out, uint16_t out_size) const ALWAYS_INLINE {
    ASSERT_ND(is_valid());
    const char* raw = get_cur_record_raw();
    uint16_t copy_size = std::min<uint16_t>(out_size, cur_record_length_);
    std::memcpy(out, raw, copy_size);
  }
  const char* get_cur_record_raw() const ALWAYS_INLINE {
    ASSERT_ND(is_valid());
    return batch_->get_payload_from_offset(cur_offset_);
  }
  const xct::RwLockableXctId* get_cur_record_owner_id() const ALWAYS_INLINE {
    ASSERT_ND(is_valid());
    return batch_->get_owner_id_from_offset(cur_offset_);
  }
  bool        in_epoch_range(Epoch epoch) const ALWAYS_INLINE {
    return epoch >= from_epoch_ && epoch < to_epoch_;
  }

  uint16_t    get_stat_skipped_records() const { return stat_skipped_records_; }
  uint16_t    get_record_count() const { return record_count_; }

  const SequentialRecordBatch* get_raw_batch() const { return batch_; }

 private:
  const SequentialRecordBatch* batch_;  // +8 -> 8
  Epoch     from_epoch_;                // +4 -> 12
  Epoch     to_epoch_;                  // +4 -> 16
  Epoch     cur_record_epoch_;          // +4 -> 20
  uint16_t  record_count_;              // +2 -> 22
  uint16_t  cur_record_;                // +2 -> 24
  uint16_t  cur_record_length_;         // +2 -> 26
  uint16_t  cur_offset_;                // +2 -> 28
  uint16_t  stat_skipped_records_;      // +2 -> 30
  uint16_t  filler_;                    // +2 -> 32
};

STATIC_SIZE_CHECK(sizeof(SequentialRecordBatch), kPageSize)

}  // namespace sequential
}  // namespace storage
}  // namespace foedus
#endif  // FOEDUS_STORAGE_SEQUENTIAL_SEQUENTIAL_CURSOR_HPP_