page_resolver.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_MEMORY_PAGE_RESOLVER_HPP_
#define FOEDUS_MEMORY_PAGE_RESOLVER_HPP_
#include <cstring>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/thread/thread_id.hpp"

namespace foedus {
namespace memory {
struct LocalPageResolver CXX11_FINAL {
  LocalPageResolver() : begin_(0), end_(0), base_(CXX11_NULLPTR) {}
  LocalPageResolver(storage::Page* base, PagePoolOffset begin, PagePoolOffset end)
    : begin_(begin), end_(end), base_(base) {}

  // default assignment/copy-constructor is enough

  PagePoolOffset resolve_page(storage::Page *page) const ALWAYS_INLINE {
    PagePoolOffset offset = page - base_;
    ASSERT_ND(offset >= begin_);
    ASSERT_ND(offset < end_);
    return offset;
  }
  storage::Page* resolve_offset(PagePoolOffset offset) const ALWAYS_INLINE {
    storage::Page* page = resolve_offset_newpage(offset);
    // this is NOT a new page, so we have sanity checks
    storage::assert_valid_volatile_page(page, offset);
    return page;
  }
  storage::Page* resolve_offset_newpage(PagePoolOffset offset) const ALWAYS_INLINE {
    ASSERT_ND(offset >= begin_);
    ASSERT_ND(offset < end_);
    return base_ + offset;
  }

  PagePoolOffset      begin_;
  PagePoolOffset      end_;
  storage::Page*      base_;
};

struct GlobalVolatilePageResolver CXX11_FINAL {
  typedef storage::Page* Base;
  enum Constants {
    kMaxNumaNode = 256,
  };
  GlobalVolatilePageResolver() : numa_node_count_(0), begin_(0), end_(0) {}
  GlobalVolatilePageResolver(const Base *bases, uint16_t numa_node_count,
             PagePoolOffset begin, PagePoolOffset end)
    : numa_node_count_(numa_node_count), begin_(begin), end_(end) {
    ASSERT_ND(numa_node_count <= kMaxNumaNode);
    std::memcpy(bases_, bases, sizeof(Base) * numa_node_count_);
  }
  // default assignment/copy constructors are a bit wasteful when #numa node is small, thus:
  GlobalVolatilePageResolver(const GlobalVolatilePageResolver &other) {
    operator=(other);
  }
  GlobalVolatilePageResolver& operator=(const GlobalVolatilePageResolver &other) {
    ASSERT_ND(other.numa_node_count_ > 0);
    ASSERT_ND(other.numa_node_count_ <= kMaxNumaNode);
    ASSERT_ND(other.end_ > other.begin_);
    numa_node_count_ = other.numa_node_count_;
    begin_ = other.begin_;
    end_ = other.end_;
    std::memcpy(bases_, other.bases_, sizeof(Base) * numa_node_count_);
    return *this;
  }

  storage::Page* resolve_offset(uint8_t numa_node, PagePoolOffset offset) const ALWAYS_INLINE {
    storage::Page* page = resolve_offset_newpage(numa_node, offset);
    // this is NOT a new page, so we have sanity checks
    storage::assert_valid_volatile_page(page, offset);
    ASSERT_ND(storage::construct_volatile_page_pointer(page->get_header().page_id_).get_numa_node()
      == numa_node);
    return page;
  }
  storage::Page* resolve_offset(storage::VolatilePagePointer page_pointer) const ALWAYS_INLINE {
    return resolve_offset(page_pointer.get_numa_node(), page_pointer.get_offset());
  }

  storage::Page* resolve_offset_newpage(
    uint8_t numa_node,
    PagePoolOffset offset) const ALWAYS_INLINE {
    ASSERT_ND(numa_node < numa_node_count_);
    ASSERT_ND(offset >= begin_);
    ASSERT_ND(offset < end_);
    storage::assert_aligned_page(bases_[numa_node] + offset);
    return bases_[numa_node] + offset;
  }
  storage::Page* resolve_offset_newpage(
    storage::VolatilePagePointer page_pointer) const ALWAYS_INLINE {
    return resolve_offset_newpage(
      page_pointer.get_numa_node(),
      page_pointer.get_offset());
  }

  uint16_t        numa_node_count_;
  PagePoolOffset  begin_;
  PagePoolOffset  end_;

  Base            bases_[kMaxNumaNode];
};

}  // namespace memory
}  // namespace foedus
#endif  // FOEDUS_MEMORY_PAGE_RESOLVER_HPP_