thread.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_THREAD_THREAD_HPP_
#define FOEDUS_THREAD_THREAD_HPP_
#include <iosfwd>

#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/assorted/uniform_random.hpp"
#include "foedus/log/fwd.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/memory/page_resolver.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/thread_id.hpp"
#include "foedus/xct/fwd.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace thread {
class Thread CXX11_FINAL : public virtual Initializable {
 public:
  enum Constants {
    kMaxFindPagesBatch = 32,
  };
  Thread() CXX11_FUNC_DELETE;
  Thread(Engine* engine, ThreadId id, ThreadGlobalOrdinal global_ordinal);
  ~Thread();
  ErrorStack  initialize() CXX11_OVERRIDE;
  bool        is_initialized() const CXX11_OVERRIDE;
  ErrorStack  uninitialize() CXX11_OVERRIDE;

  Engine*     get_engine() const;
  ThreadId    get_thread_id() const;
  ThreadGroupId get_numa_node() const { return decompose_numa_node(get_thread_id()); }
  ThreadGlobalOrdinal get_thread_global_ordinal() const;

  xct::Xct&   get_current_xct();
  bool        is_running_xct() const;

  memory::NumaCoreMemory* get_thread_memory() const;
  memory::NumaNodeMemory* get_node_memory() const;

  log::ThreadLogBuffer&   get_thread_log_buffer();

  const memory::GlobalVolatilePageResolver& get_global_volatile_page_resolver() const;
  const memory::LocalPageResolver& get_local_volatile_page_resolver() const;

  uint64_t      get_snapshot_cache_hits() const;
  uint64_t      get_snapshot_cache_misses() const;
  void          reset_snapshot_cache_counts() const;

  storage::Page* resolve(storage::VolatilePagePointer ptr) const;
  storage::Page* resolve_newpage(storage::VolatilePagePointer ptr) const;
  storage::Page* resolve(memory::PagePoolOffset offset) const;
  storage::Page* resolve_newpage(memory::PagePoolOffset offset) const;
  template <typename P> P* resolve_cast(storage::VolatilePagePointer ptr) const {
    return reinterpret_cast<P*>(resolve(ptr));
  }
  template <typename P> P* resolve_newpage_cast(storage::VolatilePagePointer ptr) const {
    return reinterpret_cast<P*>(resolve_newpage(ptr));
  }
  template <typename P> P* resolve_cast(memory::PagePoolOffset offset) const {
    return reinterpret_cast<P*>(resolve(offset));
  }
  template <typename P> P* resolve_newpage_cast(memory::PagePoolOffset offset) const {
    return reinterpret_cast<P*>(resolve_newpage(offset));
  }


  ErrorCode     find_or_read_a_snapshot_page(
    storage::SnapshotPagePointer page_id,
    storage::Page** out);
  ErrorCode     find_or_read_snapshot_pages_batch(
    uint16_t batch_size,
    const storage::SnapshotPagePointer* page_ids,
    storage::Page** out);

  ErrorCode     read_a_snapshot_page(storage::SnapshotPagePointer page_id, storage::Page* buffer);

  ErrorCode     read_snapshot_pages(
    storage::SnapshotPagePointer page_id_begin,
    uint32_t page_count,
    storage::Page* buffer);

  ErrorCode     install_a_volatile_page(
    storage::DualPagePointer* pointer,
    storage::Page** installed_page);

  ErrorCode     follow_page_pointer(
    storage::VolatilePageInit page_initializer,
    bool tolerate_null_pointer,
    bool will_modify,
    bool take_ptr_set_snapshot,
    storage::DualPagePointer* pointer,
    storage::Page** page,
    const storage::Page* parent,
    uint16_t index_in_parent);

  ErrorCode follow_page_pointers_for_read_batch(
    uint16_t batch_size,
    storage::VolatilePageInit page_initializer,
    bool tolerate_null_pointer,
    bool take_ptr_set_snapshot,
    storage::DualPagePointer** pointers,
    storage::Page** parents,
    const uint16_t* index_in_parents,
    bool* followed_snapshots,
    storage::Page** out);

  ErrorCode follow_page_pointers_for_write_batch(
    uint16_t batch_size,
    storage::VolatilePageInit page_initializer,
    storage::DualPagePointer** pointers,
    storage::Page** parents,
    const uint16_t* index_in_parents,
    storage::Page** out);

  void          collect_retired_volatile_page(storage::VolatilePagePointer ptr);

  xct::McsRwSimpleBlock* get_mcs_rw_simple_blocks();
  xct::McsRwExtendedBlock* get_mcs_rw_extended_blocks();

  ErrorCode cll_try_or_acquire_single_lock(xct::LockListPosition pos);
  ErrorCode cll_try_or_acquire_multiple_locks(xct::LockListPosition upto_pos);
  void      cll_giveup_all_locks_after(xct::UniversalLockId address);
  void      cll_giveup_all_locks_at_and_after(xct::UniversalLockId address) {
    if (address == xct::kNullUniversalLockId) {
      cll_giveup_all_locks_after(xct::kNullUniversalLockId);
    } else {
      cll_giveup_all_locks_after(address - 1U);
    }
  }
  void      cll_release_all_locks_after(xct::UniversalLockId address);
  void      cll_release_all_locks_at_and_after(xct::UniversalLockId address) {
    if (address == xct::kNullUniversalLockId) {
      cll_release_all_locks_after(xct::kNullUniversalLockId);
    } else {
      cll_release_all_locks_after(address - 1U);
    }
  }
  void      cll_release_all_locks();


  ErrorCode run_nested_sysxct(xct::SysxctFunctor* functor, uint32_t max_retries = 0);
  ErrorCode sysxct_record_lock(
    xct::SysxctWorkspace* sysxct_workspace,
    storage::VolatilePagePointer page_id,
    xct::RwLockableXctId* lock);
  ErrorCode sysxct_batch_record_locks(
    xct::SysxctWorkspace* sysxct_workspace,
    storage::VolatilePagePointer page_id,
    uint32_t lock_count,
    xct::RwLockableXctId** locks);
  ErrorCode sysxct_page_lock(xct::SysxctWorkspace* sysxct_workspace, storage::Page* page);
  ErrorCode sysxct_batch_page_locks(
    xct::SysxctWorkspace* sysxct_workspace,
    uint32_t lock_count,
    storage::Page** pages);

  Epoch*        get_in_commit_epoch_address();

  ThreadPimpl*  get_pimpl() const { return pimpl_; }

  inline assorted::UniformRandom& get_lock_rnd() { return lock_rnd_; }

  bool          is_hot_page(const storage::Page* page) const;

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

 private:
  ThreadPimpl*    pimpl_;

  // The following should be in pimpl. later.
  assorted::UniformRandom lock_rnd_;
};

class GrabFreeVolatilePagesScope {
 public:
  GrabFreeVolatilePagesScope(Thread* context, memory::PagePoolOffset* offsets)
    : context_(context), offsets_(offsets), count_(0) {}
  ~GrabFreeVolatilePagesScope() {
    release();
  }

  ErrorCode grab(uint32_t count);
  void      release();
  uint32_t  get_count() const { return count_; }

  void                    dispatch(uint32_t index) {
    ASSERT_ND(index < count_);
    offsets_[index] = 0;  // This entry will be skipped in release()
  }
  memory::PagePoolOffset  get(uint32_t index) const {
    ASSERT_ND(index < count_);
    return offsets_[index];
  }

 private:
  Thread* const                 context_;
  memory::PagePoolOffset* const offsets_;
  uint32_t                      count_;
};

}  // namespace thread
}  // namespace foedus
#endif  // FOEDUS_THREAD_THREAD_HPP_