Detailed Description

Partitioner for a hash storage.

Partitioning for hash is much easier than for B-trees, but somewhat trickier than array. Compared to array, there will not be and should not be any locality in terms of ranges of hash bins. If there are, it means that the hash function is bad!

Footprint vs Accuracy: The main challenge here is the tradeoff between the footprint and accuracy of the partitioning information. One extreme, which is our current implementation, is to determine a node to assign for every single hash bin. Assuming there are few records in each hash bin, this should give a good locality so that mappers can avoid sending out too many logs to remote reducers. However, we need sizeof(PartitionId) * hashbins bytes of partitioning information. Another extreme is to assign partition in a super coarse granularity, such as root page pointers. Then the footprint is fixed and tiny, but has basically zero accuracy. All random. So far, we picked the former extreme. We thus issue an error when a user specifies a large bin-bits such that 2^bits bytes >= storage_options.partitioner_data_memory_mb_ MB.

Current policy and its limitations

Based on the above simple design, we just pick the owner of current volatile data page as the partition. We don't even balance out so far. In sum, we have many limitations as follows:

Large footprints, which might prohibit large hash-tables and cause lots of L1 misses while partitioning.
Users must make sure the initial inserts are well balanced between nodes.

Note: This is a private implementation-details of Hashtable Storage, thus file name ends with _impl. Do not include this header from a client program. There is no case client program needs to access this internal class.

Definition at line 67 of file hash_partitioner_impl.hpp.

#include <hash_partitioner_impl.hpp>

Public Member Functions
	HashPartitioner (Partitioner *parent)

ErrorStack	design_partition (const Partitioner::DesignPartitionArguments &args)

void	design_partition_task (uint16_t task)

bool	is_partitionable () const

void	partition_batch (const Partitioner::PartitionBatchArguments &args) const

void	sort_batch (const Partitioner::SortBatchArguments &args) const

const PartitionId *	get_bucket_owners () const

Friends
std::ostream &	operator<< (std::ostream &o, const HashPartitioner &v)

Constructor & Destructor Documentation

foedus::storage::hash::HashPartitioner::HashPartitioner ( Partitioner * parent )

explicit

Definition at line 43 of file hash_partitioner_impl.cpp.

References ASSERT_ND, foedus::soc::SharedMutex::is_initialized(), foedus::storage::PartitionerMetadata::locate_data(), foedus::storage::PartitionerMetadata::mutex_, and foedus::storage::PartitionerMetadata::valid_.

   : engine_(parent->get_engine()),
     id_(parent->get_storage_id()),
     metadata_(PartitionerMetadata::get_metadata(engine_, id_)) {
   ASSERT_ND(metadata_->mutex_.is_initialized());
   if (metadata_->valid_) {
     data_ = reinterpret_cast<HashPartitionerData*>(metadata_->locate_data(engine_));
   } else {
     data_ = nullptr;
   }
 }

Here is the call graph for this function:

Member Function Documentation

ErrorStack foedus::storage::hash::HashPartitioner::design_partition ( const Partitioner::DesignPartitionArguments & args )

Definition at line 68 of file hash_partitioner_impl.cpp.

Referenced by foedus::storage::Partitioner::design_partition().

                                                 {
   ASSERT_ND(metadata_->mutex_.is_initialized());
   ASSERT_ND(data_ == nullptr);
   HashStorage storage(engine_, id_);
   ASSERT_ND(storage.exists());
   HashStorageControlBlock* control_block = storage.get_control_block();
 
   soc::SharedMutexScope mutex_scope(&metadata_->mutex_);
   ASSERT_ND(!metadata_->valid_);
   HashBin total_bin_count = storage.get_bin_count();
   uint16_t node_count = engine_->get_soc_count();
   uint64_t bytes = HashPartitionerData::object_size(node_count, total_bin_count);
   WRAP_ERROR_CODE(metadata_->allocate_data(engine_, &mutex_scope, bytes));
   data_ = reinterpret_cast<HashPartitionerData*>(metadata_->locate_data(engine_));
 
   data_->levels_ = storage.get_levels();
   ASSERT_ND(storage.get_levels() >= 1U);
   data_->bin_bits_ = storage.get_bin_bits();
   data_->bin_shifts_ = storage.get_bin_shifts();
   data_->partitionable_ = node_count > 1U;
   data_->total_bin_count_ = total_bin_count;
 
   if (!data_->partitionable_) {
     // No partitioning needed. We don't even allocate memory for bin_owners_ in this case
     metadata_->valid_ = true;
     return kRetOk;
   }
 
   ASSERT_ND(!control_block->root_page_pointer_.volatile_pointer_.is_null());
 
   // simply checks the owner of volatile pointers in last-level intermediate pages.
   // though this is an in-memory task, parallelize to make it even faster.
   // each sub-task is a pointer from the root intermediate page.
   if (storage.get_levels() == 1U) {
     VLOG(0) << "Not worth parallelization. just one level. Design on a single thread";
     // Note, this is just about partition-design. Mapper/Reducer still runs on multi-nodes.
     for (uint16_t i = 0; i < total_bin_count; ++i) {
       design_partition_task(i);
     }
   } else {
     HashBin interval = kHashMaxBins[storage.get_levels() - 1U];
     ASSERT_ND(interval < total_bin_count);
     ASSERT_ND(interval * kHashIntermediatePageFanout >= total_bin_count);
     std::vector<std::thread> threads;
     for (uint16_t i = 0; interval * i < total_bin_count; ++i) {
       threads.emplace_back(design_partition_thread, this, i);
     }
     LOG(INFO) << "Lanched " << threads.size() << " threads. joining..";
 
     for (auto& t : threads) {
       t.join();
     }
     LOG(INFO) << "Joined. Designing done";
   }
 
   metadata_->valid_ = true;
   return kRetOk;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::storage::hash::HashPartitioner::design_partition_task ( uint16_t task )

Definition at line 128 of file hash_partitioner_impl.cpp.

References ASSERT_ND, foedus::storage::hash::HashPartitionerData::bin_owners_, foedus::Attachable< CONTROL_BLOCK >::get_control_block(), foedus::memory::EngineMemory::get_global_volatile_page_resolver(), foedus::storage::hash::HashIntermediatePage::get_level(), foedus::storage::hash::HashStorage::get_levels(), foedus::Engine::get_memory_manager(), foedus::storage::VolatilePagePointer::get_numa_node(), foedus::storage::hash::HashIntermediatePage::get_pointer(), foedus::storage::VolatilePagePointer::is_null(), foedus::storage::hash::kHashMaxBins, foedus::storage::hash::HashPartitionerData::partitionable_, foedus::memory::GlobalVolatilePageResolver::resolve_offset(), foedus::storage::hash::HashStorageControlBlock::root_page_pointer_, and foedus::storage::DualPagePointer::volatile_pointer_.

Referenced by design_partition(), and foedus::storage::hash::design_partition_thread().

                                                          {
   VLOG(0) << "Task-" << task << " started.";
   ASSERT_ND(data_->partitionable_);
   HashStorage storage(engine_, id_);
   HashStorageControlBlock* control_block = storage.get_control_block();
   const memory::GlobalVolatilePageResolver& resolver
     = engine_->get_memory_manager()->get_global_volatile_page_resolver();
 
   // root page is guaranteed to have volatile version.
   HashIntermediatePage* root_page = reinterpret_cast<HashIntermediatePage*>(
     resolver.resolve_offset(control_block->root_page_pointer_.volatile_pointer_));
 
   HashBin interval = kHashMaxBins[root_page->get_level()];
   VolatilePagePointer pointer = root_page->get_pointer(task).volatile_pointer_;
   if (pointer.is_null()) {
     VLOG(0) << "Task-" << task << " got an empty task. null volatile pointer";
     std::memset(data_->bin_owners_ + (interval * task), 0, interval);
   } else if (root_page->get_level() == 0) {
     ASSERT_ND(interval = 1ULL);
     ASSERT_ND(storage.get_levels() == 1U);
     VLOG(2) << "Task-" << task << " is trivial.";
     data_->bin_owners_[task] = pointer.get_numa_node();
   } else {
     HashIntermediatePage* child
       = reinterpret_cast<HashIntermediatePage*>(resolver.resolve_offset(pointer));
     design_partition_task_recurse(resolver, child);
   }
 
   VLOG(0) << "Task-" << task << " ended.";
 }

Here is the call graph for this function:

Here is the caller graph for this function:

const PartitionId * foedus::storage::hash::HashPartitioner::get_bucket_owners ( ) const

Definition at line 55 of file hash_partitioner_impl.cpp.

References ASSERT_ND, and foedus::storage::hash::HashPartitionerData::bin_owners_.

                                                             {
   ASSERT_ND(data_);
   return data_->bin_owners_;
 }

bool foedus::storage::hash::HashPartitioner::is_partitionable ( ) const

Definition at line 59 of file hash_partitioner_impl.cpp.

References ASSERT_ND, and foedus::storage::hash::HashPartitionerData::partitionable_.

Referenced by foedus::storage::Partitioner::is_partitionable(), and partition_batch().

                                              {
   ASSERT_ND(data_);
   return data_->partitionable_;
 }

Here is the caller graph for this function:

void foedus::storage::hash::HashPartitioner::partition_batch ( const Partitioner::PartitionBatchArguments & args ) const

Definition at line 184 of file hash_partitioner_impl.cpp.

References ASSERT_ND, foedus::storage::hash::HashCommonLogType::assert_type(), foedus::storage::hash::HashPartitionerData::bin_owners_, foedus::storage::hash::HashStorage::get_bin_count(), foedus::storage::hash::HashStorage::get_bin_shifts(), foedus::storage::hash::HashCommonLogType::hash_, foedus::log::BaseLogType::header_, is_partitionable(), foedus::storage::Partitioner::PartitionBatchArguments::log_buffer_, foedus::storage::Partitioner::PartitionBatchArguments::log_positions_, foedus::storage::Partitioner::PartitionBatchArguments::logs_count_, foedus::snapshot::LogBuffer::resolve(), foedus::storage::Partitioner::PartitionBatchArguments::results_, and foedus::log::LogHeader::storage_id_.

Referenced by foedus::storage::Partitioner::partition_batch().

                                                                                           {
   if (!is_partitionable()) {
     std::memset(args.results_, 0, sizeof(PartitionId) * args.logs_count_);
     return;
   }
 
   HashStorage storage(engine_, id_);
   uint8_t bin_shifts = storage.get_bin_shifts();
   for (uint32_t i = 0; i < args.logs_count_; ++i) {
     const HashCommonLogType *log = reinterpret_cast<const HashCommonLogType*>(
       args.log_buffer_.resolve(args.log_positions_[i]));
     log->assert_type();
     ASSERT_ND(log->header_.storage_id_ == id_);
     HashValue hash = log->hash_;
     HashBin bin = hash >> bin_shifts;
     ASSERT_ND(bin < storage.get_bin_count());
     args.results_[i] = data_->bin_owners_[bin];
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::storage::hash::HashPartitioner::sort_batch ( const Partitioner::SortBatchArguments & args ) const

Definition at line 329 of file hash_partitioner_impl.cpp.

References ASSERT_ND, foedus::memory::AlignedMemory::assure_capacity(), foedus::storage::hash::HashPartitionerData::bin_shifts_, foedus::storage::hash::compact_logs(), foedus::debugging::StopWatch::elapsed_ms(), foedus::memory::AlignedMemory::get_block(), foedus::storage::Partitioner::SortBatchArguments::logs_count_, foedus::storage::hash::prepare_sort_entries(), foedus::debugging::StopWatch::stop(), foedus::storage::Partitioner::SortBatchArguments::work_memory_, and foedus::storage::Partitioner::SortBatchArguments::written_count_.

Referenced by foedus::storage::Partitioner::sort_batch().

                                                                                 {
   if (args.logs_count_ == 0) {
     *args.written_count_ = 0;
     return;
   }
 
   // we so far sort them in one path.
   // to save memory, we could do multi-path merge-sort.
   // however, in reality each log has many bytes, so log_count is not that big.
   args.work_memory_->assure_capacity(sizeof(SortEntry) * args.logs_count_);
 
   debugging::StopWatch stop_watch_entire;
 
   ASSERT_ND(sizeof(SortEntry) == 16U);
   SortEntry* entries = reinterpret_cast<SortEntry*>(args.work_memory_->get_block());
   prepare_sort_entries(data_->bin_shifts_, args, entries);
 
   debugging::StopWatch stop_watch;
   // Gave up non-gcc support because of aarch64 support. yes, we can also assume __uint128_t.
   // CPU profile of partition_hash_perf: ??% (introsort_loop) + ??% (other inlined).
   std::sort(
     reinterpret_cast<__uint128_t*>(entries),
     reinterpret_cast<__uint128_t*>(entries + args.logs_count_));
   stop_watch.stop();
   VLOG(0) << "Sorted " << args.logs_count_ << " log entries in " << stop_watch.elapsed_ms() << "ms";
 
   uint32_t result_count = compact_logs(data_->bin_shifts_, args, entries);
 
   stop_watch_entire.stop();
   VLOG(0) << "Hash-" << id_ << " sort_batch() done in  " << stop_watch_entire.elapsed_ms()
       << "ms  for " << args.logs_count_ << " log entries, compacted them to"
         << result_count << " log entries";
   *args.written_count_ = result_count;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Friends And Related Function Documentation

std::ostream& operator<<	(	std::ostream &	o,
		const HashPartitioner &	v
	)

friend

Definition at line 364 of file hash_partitioner_impl.cpp.

                                                                 {
   o << "<HashPartitioner>";
   if (v.data_) {
     o << "<levels_>" << static_cast<int>(v.data_->levels_) << "</levels_>"
       << "<total_bin_count_>" << v.data_->total_bin_count_ << "</total_bin_count_>";
   } else {
     o << "Not yet designed";
   }
   o << "</HashPartitioner>";
   return o;
 }

The documentation for this class was generated from the following files:

/home/shino/foedus_code/foedus-core/include/foedus/storage/hash/hash_partitioner_impl.hpp
/home/shino/foedus_code/foedus-core/src/foedus/storage/hash/hash_partitioner_impl.cpp

Detailed Description

Public Member Functions

Friends

Constructor & Destructor Documentation

Member Function Documentation

Friends And Related Function Documentation