foedus::storage::hash::HashTmpBin Class Reference

An in-memory single-threaded data structure to compose tuples in a hash bin. More...

Detailed Description

An in-memory single-threaded data structure to compose tuples in a hash bin.

This object is used in the hash composer to organize data pages for each hash bin, applying log entries to existing tuples. In essense, it is another hash-table where bucket is low-bits (while bin-bits use high-bits). For testability and modularity, we separated this to its own class.

Access pattern to optimize for

Reducer sorts hash log entries by their hash bins. In each bin, logs are sorted by timestamp. So, the inputs are not sorted by full-hash values, which would have simplified how this object indexes each tuple (just linked list is enough). Instead, we can assume that the next access on the same key surely has a larger timestamp, so this object doesn't have to worry about the order of applying modifications.

Definition at line 54 of file hash_tmpbin.hpp.

#include <hash_tmpbin.hpp>

Classes
struct	Record
	Represents a record with a unique key. More...

Public Types
enum	Constants { kBucketCount = 1 << 16, kDefaultInitialSize = 1 << 24 }
typedef uint32_t	RecordIndex
	Pointer to Record. More...

Public Member Functions
	HashTmpBin ()
	Constructor doesn't do any initialization. More...
ErrorCode	create_memory (uint16_t numa_node, uint64_t initial_size=kDefaultInitialSize)
	Allocates the memory to use by this object. More...
void	release_memory ()
	Destructor automatically releases everything, but you can use this to do it earlier. More...
void	steal_memory (memory::AlignedMemory *provider)
	This is a special version of create_memory() where this object steals the memory ownership from the recipient. More...
void	give_memory (memory::AlignedMemory *recipient)
	This is a special version of release_memory() where this object moves the memory ownership to the recipient. More...
void	clean ()
	Removes all tuple data for the current bin. More...
void	clean_quick ()
	This version selectively clears buckets_ by seeing individual records. More...
Record *	get_record (RecordIndex index) const
RecordIndex	get_records_capacity () const
RecordIndex	get_records_consumed () const
RecordIndex	get_first_record () const
uint32_t	get_physical_record_count () const
RecordIndex	get_bucket_head (uint32_t bucket) const
ErrorCode	insert_record (xct::XctId xct_id, const void *key, uint16_t key_length, HashValue hash, const void *payload, uint16_t payload_length)
	Inserts a new record of the given key and payload. More...
ErrorCode	delete_record (xct::XctId xct_id, const void *key, uint16_t key_length, HashValue hash)
	Logically deletes a record of the given key. More...
ErrorCode	overwrite_record (xct::XctId xct_id, const void *key, uint16_t key_length, HashValue hash, const void *payload, uint16_t payload_offset, uint16_t payload_count)
	Overwrites a part of the record of the given key. More...
ErrorCode	update_record (xct::XctId xct_id, const void *key, uint16_t key_length, HashValue hash, const void *payload, uint16_t payload_length)
	Updates a record of the given key with the given payload, which might change length. More...

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

Member Typedef Documentation

typedef uint32_t foedus::storage::hash::HashTmpBin::RecordIndex

Pointer to Record.

0 means null.

Definition at line 69 of file hash_tmpbin.hpp.

Member Enumeration Documentation

enum foedus::storage::hash::HashTmpBin::Constants

Enumerator
kBucketCount	This is a per-bin data structure. This should be large enough. If not, the bin has many data pages, causing low performance anyways.
kDefaultInitialSize	Let's be super generous, assuming that we are reusing memories.

Definition at line 56 of file hash_tmpbin.hpp.

                 {
    kBucketCount = 1 << 16,
    kDefaultInitialSize = 1 << 24,
  };

Constructor & Destructor Documentation

foedus::storage::hash::HashTmpBin::HashTmpBin

(

)

Constructor doesn't do any initialization.

You must call create_memory() before using.

Definition at line 32 of file hash_tmpbin.cpp.

  : memory_(), records_capacity_(0), records_consumed_(0), buckets_(nullptr), records_(nullptr) {
}

Member Function Documentation

void foedus::storage::hash::HashTmpBin::clean

(

)

Removes all tuple data for the current bin.

The memory is kept so that we can efficiently reuse resources for next bins to process.

This is a full cleaning method that can be used anytime, but costs a bit more. When there were few records, it is a bit expensive to mem-zero the entire buckets_. In that case, use clean_quick()

Definition at line 74 of file hash_tmpbin.cpp.

References get_first_record().

Referenced by create_memory(), foedus::storage::hash::HashComposeContext::execute(), give_memory(), release_memory(), and steal_memory().

                       {
  records_consumed_ = get_first_record();
  if (buckets_) {
    std::memset(buckets_, 0, records_consumed_ * sizeof(Record));
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foedus::storage::hash::HashTmpBin::clean_quick

(

)

This version selectively clears buckets_ by seeing individual records.

Hence, this can be used only after records_consumed_ is populated. Instead, when there are just a few records, this is much faster than clean().

Definition at line 81 of file hash_tmpbin.cpp.

References ASSERT_ND, get_first_record(), get_record(), and foedus::storage::hash::HashTmpBin::Record::hash_.

                             {
  ASSERT_ND(buckets_);
  ASSERT_ND(records_capacity_ > 0);
  ASSERT_ND(records_consumed_ >= get_first_record());
  // at some point, just mem-zero is faster. we could do a switch like below, but it's rare.
  // this also makes unit-testing more tricky.
  // if (records_consumed_ > get_first_record() + 256U) {
  //   clean();
  //   return;
  // }
  for (RecordIndex index = get_first_record(); index < records_consumed_; ++index) {
    Record* record = get_record(index);
    buckets_[extract_bucket(record->hash_)] = 0;
  }
  records_consumed_ = get_first_record();
}

Here is the call graph for this function:

ErrorCode foedus::storage::hash::HashTmpBin::create_memory	(	uint16_t	numa_node,
		uint64_t	initial_size = kDefaultInitialSize
	)

Allocates the memory to use by this object.

Must be called before using this object.

Parameters

[in]	numa_node	the NUMA node to allocate this object's memory on
[in]	initial_size	byte size of the memory to start from.

Precondition

initial_size > sizeof(RecordIndex)*kBucketCount, in reality should be much larger

Postcondition

!memory_->is_null()

Definition at line 36 of file hash_tmpbin.cpp.

References foedus::memory::AlignedMemory::alloc(), ASSERT_ND, clean(), kBucketCount, foedus::kErrorCodeOk, foedus::memory::AlignedMemory::kNumaAllocOnnode, and foedus::storage::kPageSize.

Referenced by foedus::storage::hash::HashComposeContext::execute().

                                                                             {
  ASSERT_ND(initial_size > sizeof(RecordIndex) * kBucketCount);
  ASSERT_ND(sizeof(Record) == kPageSize);
  memory_.alloc(
    initial_size,
    1ULL << 21,
    memory::AlignedMemory::kNumaAllocOnnode,
    numa_node);
  on_memory_set();
  clean();
  return kErrorCodeOk;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ErrorCode foedus::storage::hash::HashTmpBin::delete_record	(	xct::XctId	xct_id,
		const void *	key,
		uint16_t	key_length,
		HashValue	hash
	)

Logically deletes a record of the given key.

If there is no existing record of the key, or such a record is already logically deleted, this method returns an error (kErrorCodeStrKeyNotFound). Mustn't happen either.

Definition at line 167 of file hash_tmpbin.cpp.

References ASSERT_ND, foedus::xct::XctId::compare_epoch_and_orginal(), get_record(), foedus::storage::hash::HashTmpBin::Record::hash_, foedus::storage::hash::hashinate(), foedus::xct::XctId::is_deleted(), foedus::kErrorCodeOk, foedus::kErrorCodeStrKeyNotFound, UNLIKELY, and foedus::storage::hash::HashTmpBin::Record::xct_id_.

                  {
  ASSERT_ND(xct_id.is_deleted());
  ASSERT_ND(hashinate(key, key_length) == hash);
  SearchResult result = search_bucket(key, key_length, hash);
  if (UNLIKELY(result.found_ == 0)) {
    DLOG(WARNING) << "HashTmpBin::delete_record() hit KeyNotFound case 1. This must not"
      << " happen except unit testcases.";
    return kErrorCodeStrKeyNotFound;
  } else {
    Record* record = get_record(result.found_);
    ASSERT_ND(record->hash_ == hash);
    if (UNLIKELY(record->xct_id_.is_deleted())) {
      DLOG(WARNING) << "HashTmpBin::delete_record() hit KeyNotFound case 2. This must not"
        << " happen except unit testcases.";
      return kErrorCodeStrKeyNotFound;
    }
    ASSERT_ND(record->xct_id_.compare_epoch_and_orginal(xct_id) < 0);
    record->xct_id_ = xct_id;
  }

  return kErrorCodeOk;
}

Here is the call graph for this function:

RecordIndex foedus::storage::hash::HashTmpBin::get_bucket_head ( uint32_t  bucket ) const

inline

Definition at line 211 of file hash_tmpbin.hpp.

References ASSERT_ND, and kBucketCount.

                                                     {
    ASSERT_ND(bucket < kBucketCount);
    return buckets_[bucket];
  }

RecordIndex foedus::storage::hash::HashTmpBin::get_first_record ( ) const

inline

Definition at line 204 of file hash_tmpbin.hpp.

References kBucketCount.

Referenced by clean(), clean_quick(), get_physical_record_count(), and foedus::storage::hash::operator<<().

                                       {
    return sizeof(RecordIndex) * kBucketCount / sizeof(Record);
  }

Here is the caller graph for this function:

uint32_t foedus::storage::hash::HashTmpBin::get_physical_record_count ( ) const

inline

Returns

number of physical records, which may or may not be logically deleted

Definition at line 208 of file hash_tmpbin.hpp.

References get_first_record(), and get_records_consumed().

                                                {
    return get_records_consumed() - get_first_record();
  }

Here is the call graph for this function:

Record* foedus::storage::hash::HashTmpBin::get_record ( RecordIndex  index ) const

inline

Definition at line 197 of file hash_tmpbin.hpp.

References ASSERT_ND, and kBucketCount.

Referenced by clean_quick(), delete_record(), insert_record(), foedus::storage::hash::operator<<(), overwrite_record(), and update_record().

                                                  {
    ASSERT_ND(index < records_capacity_);
    ASSERT_ND(index >= sizeof(RecordIndex) * kBucketCount / sizeof(Record));
    return records_ + index;
  }

Here is the caller graph for this function:

RecordIndex foedus::storage::hash::HashTmpBin::get_records_capacity ( ) const

inline

Definition at line 202 of file hash_tmpbin.hpp.

{ return records_capacity_; }

RecordIndex foedus::storage::hash::HashTmpBin::get_records_consumed ( ) const

inline

Definition at line 203 of file hash_tmpbin.hpp.

Referenced by get_physical_record_count(), and foedus::storage::hash::operator<<().

{ return records_consumed_; }

Here is the caller graph for this function:

void foedus::storage::hash::HashTmpBin::give_memory

(

memory::AlignedMemory *

recipient

)

This is a special version of release_memory() where this object moves the memory ownership to the recipient.

Combined with steal_memory(), this allows completely reusing memory.

Parameters

[in,out]

recipient

the memory to receive the ownership of memory this object has

Precondition

!memory_->is_null()

Postcondition

!recipient->is_null()

memory_->is_null()

Definition at line 64 of file hash_tmpbin.cpp.

References ASSERT_ND, clean(), and foedus::memory::AlignedMemory::is_null().

                                                           {
  ASSERT_ND(!memory_.is_null());
  *recipient = std::move(memory_);
  ASSERT_ND(!recipient->is_null());
  ASSERT_ND(memory_.is_null());
  on_memory_set();
  clean();
}

Here is the call graph for this function:

ErrorCode foedus::storage::hash::HashTmpBin::insert_record	(	xct::XctId	xct_id,
		const void *	key,
		uint16_t	key_length,
		HashValue	hash,
		const void *	payload,
		uint16_t	payload_length
	)

Inserts a new record of the given key and payload.

If there is an existing record of the key that is logically deleted, we flip the deletion flag. If such an record exists and also is not logically deleted, this method returns an error (kErrorCodeStrKeyAlreadyExists). Considering how this object is used, it mustn't hapepn.

Definition at line 126 of file hash_tmpbin.cpp.

References ASSERT_ND, CHECK_ERROR_CODE, foedus::xct::XctId::compare_epoch_and_orginal(), get_record(), foedus::storage::hash::HashTmpBin::Record::hash_, foedus::storage::hash::hashinate(), foedus::xct::XctId::is_deleted(), foedus::kErrorCodeOk, foedus::kErrorCodeStrKeyAlreadyExists, foedus::storage::hash::HashTmpBin::Record::next_, foedus::storage::hash::HashTmpBin::Record::set_all(), foedus::storage::hash::HashTmpBin::Record::set_payload(), UNLIKELY, and foedus::storage::hash::HashTmpBin::Record::xct_id_.

                           {
  ASSERT_ND(!xct_id.is_deleted());
  ASSERT_ND(hashinate(key, key_length) == hash);
  SearchResult result = search_bucket(key, key_length, hash);
  if (result.found_ == 0) {
    RecordIndex new_index;
    CHECK_ERROR_CODE(alloc_record(&new_index));
    Record* record = get_record(new_index);
    record->set_all(xct_id, key, key_length, hash, payload, payload_length);

    if (result.tail_ == 0) {
      uint16_t bucket_index = extract_bucket(hash);
      ASSERT_ND(buckets_[bucket_index] == 0);
      buckets_[bucket_index] = new_index;
    } else {
      ASSERT_ND(buckets_[extract_bucket(hash)] != 0);
      Record* tail_record = get_record(result.tail_);
      tail_record->next_ = new_index;
    }
  } else {
    Record* record = get_record(result.found_);
    ASSERT_ND(record->hash_ == hash);
    if (UNLIKELY(!record->xct_id_.is_deleted())) {
      DLOG(WARNING) << "HashTmpBin::insert_record() hit KeyAlreadyExists case. This must not"
        << " happen except unit testcases.";
      return kErrorCodeStrKeyAlreadyExists;
    }
    ASSERT_ND(record->xct_id_.compare_epoch_and_orginal(xct_id) < 0);
    record->xct_id_ = xct_id;
    record->set_payload(payload, payload_length);
  }

  return kErrorCodeOk;
}

Here is the call graph for this function:

ErrorCode foedus::storage::hash::HashTmpBin::overwrite_record	(	xct::XctId	xct_id,
		const void *	key,
		uint16_t	key_length,
		HashValue	hash,
		const void *	payload,
		uint16_t	payload_offset,
		uint16_t	payload_count
	)

Overwrites a part of the record of the given key.

If there is no existing record of the key, or such a record is already logically deleted, this method returns an error (kErrorCodeStrKeyNotFound). Mustn't happen either.

Definition at line 194 of file hash_tmpbin.cpp.

References ASSERT_ND, foedus::xct::XctId::compare_epoch_and_orginal(), get_record(), foedus::storage::hash::HashTmpBin::Record::hash_, foedus::storage::hash::hashinate(), foedus::xct::XctId::is_deleted(), foedus::kErrorCodeOk, foedus::kErrorCodeStrKeyNotFound, foedus::kErrorCodeStrTooShortPayload, foedus::storage::hash::HashTmpBin::Record::overwrite_payload(), foedus::storage::hash::HashTmpBin::Record::payload_length_, UNLIKELY, and foedus::storage::hash::HashTmpBin::Record::xct_id_.

                          {
  ASSERT_ND(!xct_id.is_deleted());
  ASSERT_ND(hashinate(key, key_length) == hash);
  SearchResult result = search_bucket(key, key_length, hash);
  if (UNLIKELY(result.found_ == 0)) {
    DLOG(WARNING) << "HashTmpBin::overwrite_record() hit KeyNotFound case 1. This must not"
      << " happen except unit testcases.";
    return kErrorCodeStrKeyNotFound;
  } else {
    Record* record = get_record(result.found_);
    ASSERT_ND(record->hash_ == hash);
    if (UNLIKELY(record->xct_id_.is_deleted())) {
      DLOG(WARNING) << "HashTmpBin::overwrite_record() hit KeyNotFound case 2. This must not"
        << " happen except unit testcases.";
      return kErrorCodeStrKeyNotFound;
    } else if (UNLIKELY(record->payload_length_ < payload_offset + payload_count)) {
      DLOG(WARNING) << "HashTmpBin::overwrite_record() hit TooShortPayload case. This must not"
        << " happen except unit testcases.";
      return kErrorCodeStrTooShortPayload;
    }
    ASSERT_ND(record->xct_id_.compare_epoch_and_orginal(xct_id) < 0);
    record->xct_id_ = xct_id;
    record->overwrite_payload(payload, payload_offset, payload_count);
  }

  return kErrorCodeOk;
}

Here is the call graph for this function:

void foedus::storage::hash::HashTmpBin::release_memory

(

)

Destructor automatically releases everything, but you can use this to do it earlier.

Definition at line 49 of file hash_tmpbin.cpp.

References clean(), and foedus::memory::AlignedMemory::release_block().

                                {
  memory_.release_block();
  on_memory_set();
  clean();
}

Here is the call graph for this function:

void foedus::storage::hash::HashTmpBin::steal_memory

(

memory::AlignedMemory *

provider

)

This is a special version of create_memory() where this object steals the memory ownership from the recipient.

Combined with give_memory(), this allows completely reusing memory.

Parameters

[in,out]

provider

the memory from which this object steals ownership

Precondition

!provider->is_null()

Postcondition

provider->is_null()

!memory_->is_null()

Definition at line 55 of file hash_tmpbin.cpp.

References ASSERT_ND, clean(), and foedus::memory::AlignedMemory::is_null().

                                                           {
  ASSERT_ND(!provider->is_null());
  memory_ = std::move(*provider);
  ASSERT_ND(provider->is_null());
  ASSERT_ND(!memory_.is_null());
  on_memory_set();
  clean();
}

Here is the call graph for this function:

ErrorCode foedus::storage::hash::HashTmpBin::update_record	(	xct::XctId	xct_id,
		const void *	key,
		uint16_t	key_length,
		HashValue	hash,
		const void *	payload,
		uint16_t	payload_length
	)

Updates a record of the given key with the given payload, which might change length.

This method is same as delete_record()+insert_record(). The difference is that this can be faster because it does only one lookup. If there is no existing record of the key, or such a record is already logically deleted, this method returns an error (kErrorCodeStrKeyNotFound). Mustn't happen either.

Definition at line 229 of file hash_tmpbin.cpp.

References ASSERT_ND, foedus::xct::XctId::compare_epoch_and_orginal(), get_record(), foedus::storage::hash::HashTmpBin::Record::hash_, foedus::storage::hash::hashinate(), foedus::xct::XctId::is_deleted(), foedus::kErrorCodeOk, foedus::kErrorCodeStrKeyNotFound, foedus::storage::hash::HashTmpBin::Record::set_payload(), UNLIKELY, and foedus::storage::hash::HashTmpBin::Record::xct_id_.

                           {
  ASSERT_ND(!xct_id.is_deleted());
  ASSERT_ND(hashinate(key, key_length) == hash);
  SearchResult result = search_bucket(key, key_length, hash);
  if (UNLIKELY(result.found_ == 0)) {
    DLOG(WARNING) << "HashTmpBin::update_record() hit KeyNotFound case 1. This must not"
      << " happen except unit testcases.";
    return kErrorCodeStrKeyNotFound;
  } else {
    Record* record = get_record(result.found_);
    ASSERT_ND(record->hash_ == hash);
    if (UNLIKELY(record->xct_id_.is_deleted())) {
      DLOG(WARNING) << "HashTmpBin::update_record() hit KeyNotFound case 2. This must not"
        << " happen except unit testcases.";
      return kErrorCodeStrKeyNotFound;
    }
    ASSERT_ND(record->xct_id_.compare_epoch_and_orginal(xct_id) < 0);
    record->xct_id_ = xct_id;
    record->set_payload(payload, payload_length);
  }

  return kErrorCodeOk;
}

Here is the call graph for this function:

Friends And Related Function Documentation

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

friend

Definition at line 287 of file hash_tmpbin.cpp.

                                                           {
  // Each bin shouldn't have that many records... so, output everything!
  o << "<HashTmpBin>" << std::endl;
  o << "  " << v.memory_ << std::endl;
  o << "  <records_capacity_>" << v.records_capacity_ << "</records_capacity_>" << std::endl;
  o << "  <records_consumed_>" << v.records_consumed_ << "</records_consumed_>" << std::endl;
  o << "  <buckets_>" << std::endl;
  for (uint32_t i = 0; i < HashTmpBin::kBucketCount; ++i) {
    if (v.buckets_[i] != 0) {
      o << "    <bucket idx=\"" << i << "\" head_rec=\"" << v.buckets_[i] << "\" />" << std::endl;
    }
  }
  o << "  </buckets_>" << std::endl;
  o << "  <records_>" << std::endl;
  uint32_t begin = v.get_first_record();
  for (uint32_t i = begin; i < v.get_records_consumed(); ++i) {
    HashTmpBin::Record* record = v.get_record(i);
    o << "    <record id=\"" << i << "\" hash=\"" << assorted::Hex(record->hash_, 16) << "\"";
    if (record->next_) {
      o << " next_rec=\"" << record->next_ << "\"";
    }
    o << ">";
    o << "<key>" << assorted::HexString(std::string(record->get_key(), record->key_length_))
      << "</key>";
    o << "<payload>"
      << assorted::HexString(std::string(record->get_payload(), record->payload_length_))
      << "</payload>";
    o << record->xct_id_;
    o << "</record>" << std::endl;
  }
  o << "  </records_>" << std::endl;
  o << "</HashTmpBin>";
  return o;
}

---

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

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