自学教程：C++ ASSERT_ND函数代码示例

DumpFileSortedBuffer::DumpFileSortedBuffer(  fs::DirectIoFile* file, memory::AlignedMemorySlice io_buffer)  : SortedBuffer(    reinterpret_cast<char*>(io_buffer.get_block()),    io_buffer.get_size(),    fs::file_size(file->get_path())),    file_(file),    io_buffer_(io_buffer) {  ASSERT_ND(buffer_size_ % kAlignment == 0);  ASSERT_ND(total_size_ % kAlignment == 0);}

inline uint32_t ConstDiv::rem64(uint64_t n, uint32_t d, uint64_t q) const {#ifndef NDEBUG  ASSERT_ND(d == d_);#endif  // NDEBUG  ASSERT_ND(n / d == q);  if (flags_ & kFlagPowerOfTwo) {    return n & ((1ULL << d_highest_bits_) - 1ULL);  } else {    return n - d * q;  }}

xct::McsBlockIndex ThreadPimpl::mcs_initial_lock(xct::McsLock* mcs_lock) {  assert_mcs_aligned(mcs_lock);  ASSERT_ND(!mcs_lock->is_locked());  // so far we allow only 2^16 MCS blocks per transaction. we might increase later.  ASSERT_ND(current_xct_.get_mcs_block_current() < 0xFFFFU);  xct::McsBlockIndex block_index = current_xct_.increment_mcs_block_current();  ASSERT_ND(block_index > 0);  mcs_init_block(mcs_lock, block_index, false);  mcs_lock->reset(id_, block_index);  assorted::memory_fence_acq_rel();  return block_index;}

ErrorCode ThreadPimpl::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) {  // REMINDER: Remember that it might be parents == out. It's not an issue in this function, tho.  // this method is not quite batched as it doesn't need to be.  // still, less branches because we can assume all of them need a writable volatile page.  for (uint16_t b = 0; b < batch_size; ++b) {    storage::DualPagePointer* pointer = pointers[b];    if (pointer == nullptr) {      out[b] = nullptr;      continue;    }    ASSERT_ND(!parents[b]->get_header().snapshot_);    storage::Page** page = out + b;    storage::VolatilePagePointer volatile_pointer = pointer->volatile_pointer_;    if (!volatile_pointer.is_null()) {      *page = global_volatile_page_resolver_.resolve_offset(volatile_pointer);    } else if (pointer->snapshot_pointer_ == 0) {      // we need a volatile page. so construct it from snapshot      CHECK_ERROR_CODE(install_a_volatile_page(pointer, page));    } else {      ASSERT_ND(page_initializer);      // we must not install a new volatile page in snapshot page. We must not hit this case.      memory::PagePoolOffset offset = core_memory_->grab_free_volatile_page();      if (UNLIKELY(offset == 0)) {        return kErrorCodeMemoryNoFreePages;      }      storage::Page* new_page = local_volatile_page_resolver_.resolve_offset_newpage(offset);      storage::VolatilePagePointer new_page_id;      new_page_id.components.numa_node = numa_node_;      new_page_id.components.offset = offset;      storage::VolatilePageInitArguments args = {        holder_,        new_page_id,        new_page,        parents[b],        index_in_parents[b]      };      page_initializer(args);      storage::assert_valid_volatile_page(new_page, offset);      ASSERT_ND(new_page->get_header().snapshot_ == false);      *page = place_a_new_volatile_page(offset, pointer);    }    ASSERT_ND(out[b] != nullptr);  }  return kErrorCodeOk;}

uint32_t PagePoolOffsetAndEpochChunk::get_safe_offset_count(const Epoch& threshold) const {  ASSERT_ND(is_sorted());  OffsetAndEpoch dummy;  dummy.safe_epoch_ = threshold.value();  struct CompareEpoch {    bool operator() (const OffsetAndEpoch& left, const OffsetAndEpoch& right) {      return Epoch(left.safe_epoch_) < Epoch(right.safe_epoch_);    }  };  const OffsetAndEpoch* result = std::lower_bound(chunk_, chunk_ + size_, dummy, CompareEpoch());  ASSERT_ND(result);  ASSERT_ND(result - chunk_ <= size_);  return result - chunk_;}

bool LogMapper::add_new_bucket(storage::StorageId storage_id) {  if (buckets_allocated_count_ >= buckets_memory_.get_size() / sizeof(Bucket)) {    // we allocated all buckets_memory_! we have to flush the buckets now.    // this shouldn't happen often.    LOG(WARNING) << to_string() << " ran out of buckets_memory_, so it has to flush buckets before"      " processing one IO buffer. This shouldn't happen often. check your log_mapper_bucket_kb_"      " setting. this=" << *this;    return false;  }  Bucket* base_address = reinterpret_cast<Bucket*>(buckets_memory_.get_block());  Bucket* new_bucket = base_address + buckets_allocated_count_;  ++buckets_allocated_count_;  new_bucket->storage_id_ = storage_id;  new_bucket->counts_ = 0;  new_bucket->next_bucket_ = nullptr;  BucketHashList* hashlist = find_storage_hashlist(storage_id);  if (hashlist) {    // just add this as a new tail    ASSERT_ND(hashlist->storage_id_ == storage_id);    ASSERT_ND(hashlist->tail_->is_full());    hashlist->tail_->next_bucket_ = new_bucket;    hashlist->tail_ = new_bucket;    ++hashlist->bucket_counts_;  } else {    // we don't even have a linked list for this.    // If this happens often, maybe we should have 65536 hash buckets...    if (hashlist_allocated_count_ >= kBucketHashListMaxCount) {      LOG(WARNING) << to_string() << " ran out of hashlist memory, so it has to flush buckets now"        " This shouldn't happen often. We must consider increasing kBucketHashListMaxCount."        " this=" << *this;      return false;    }    // allocate from the pool    BucketHashList* new_hashlist =      reinterpret_cast<BucketHashList*>(tmp_hashlist_buffer_slice_.get_block())      + hashlist_allocated_count_;    ++hashlist_allocated_count_;    new_hashlist->storage_id_ = storage_id;    new_hashlist->head_ = new_bucket;    new_hashlist->tail_ = new_bucket;    new_hashlist->bucket_counts_ = 1;    add_storage_hashlist(new_hashlist);  }  return true;}

void LogMapper::send_bucket_partition_general(  const Bucket* bucket,  storage::StorageType storage_type,  storage::PartitionId partition,  const BufferPosition* positions) {  uint64_t written = 0;  uint32_t log_count = 0;  uint32_t shortest_key_length = 0xFFFF;  uint32_t longest_key_length = 0;  // stitch the log entries in send buffer  char* send_buffer = reinterpret_cast<char*>(tmp_send_buffer_slice_.get_block());  const char* io_base = reinterpret_cast<const char*>(io_buffer_.get_block());  ASSERT_ND(tmp_send_buffer_slice_.get_size() == kSendBufferSize);  for (uint32_t i = 0; i < bucket->counts_; ++i) {    uint64_t pos = from_buffer_position(positions[i]);    const log::LogHeader* header = reinterpret_cast<const log::LogHeader*>(io_base + pos);    ASSERT_ND(header->storage_id_ == bucket->storage_id_);    uint16_t log_length = header->log_length_;    ASSERT_ND(log_length > 0);    ASSERT_ND(log_length % 8 == 0);    if (written + log_length > kSendBufferSize) {      // buffer full. send out.      send_bucket_partition_buffer(        bucket,        partition,        send_buffer,        log_count,        written,        shortest_key_length,        longest_key_length);      log_count = 0;      written = 0;      shortest_key_length = 0xFFFF;      longest_key_length = 0;    }    std::memcpy(send_buffer + written, header, header->log_length_);    written += header->log_length_;    ++log_count;    update_key_lengthes(header, storage_type, &shortest_key_length, &longest_key_length);  }  send_bucket_partition_buffer(    bucket,    partition,    send_buffer,    log_count,    written,    shortest_key_length,    longest_key_length);}

ErrorStack inserts_varlen_task(const proc::ProcArguments& args) {  EXPECT_EQ(sizeof(uint32_t), args.input_len_);  uint32_t id = *reinterpret_cast<const uint32_t*>(args.input_buffer_);  EXPECT_NE(id, 2U);  thread::Thread* context = args.context_;  storage::hash::HashStorage hash(args.engine_, kName);  ASSERT_ND(hash.exists());  xct::XctManager* xct_manager = args.engine_->get_xct_manager();  Epoch commit_epoch;  WRAP_ERROR_CODE(xct_manager->begin_xct(context, xct::kSerializable));  char buffer[16];  std::memset(buffer, 0, sizeof(buffer));  for (uint32_t i = 0; i < kRecords / 2U; ++i) {    uint64_t rec = id * kRecords / 2U + i;    // first 8 bytes, mod 17 to have next layers.    assorted::write_bigendian<uint64_t>(static_cast<uint64_t>(rec % 17U), buffer);    // and 1-4 bytes of decimal representation in text    std::string str = std::to_string(rec);    std::memcpy(buffer + sizeof(uint64_t), str.data(), str.size());    uint16_t len = sizeof(uint64_t) + str.size();    uint64_t data = rec + kDataAddendum;    ErrorCode ret = hash.insert_record(context, buffer, len, &data, sizeof(data));    EXPECT_EQ(kErrorCodeOk, ret) << i;  }  // CHECK_ERROR(hash.debugout_single_thread(args.engine_));  WRAP_ERROR_CODE(xct_manager->precommit_xct(context, &commit_epoch));  // CHECK_ERROR(hash.debugout_single_thread(args.engine_));  WRAP_ERROR_CODE(xct_manager->wait_for_commit(commit_epoch));  return kRetOk;}

ErrorStack verify_varlen_task(const proc::ProcArguments& args) {  thread::Thread* context = args.context_;  storage::hash::HashStorage hash(args.engine_, kName);  ASSERT_ND(hash.exists());  CHECK_ERROR(hash.verify_single_thread(context));  // CHECK_ERROR(hash.debugout_single_thread(args.engine_));  xct::XctManager* xct_manager = args.engine_->get_xct_manager();  WRAP_ERROR_CODE(xct_manager->begin_xct(context, xct::kSerializable));  char buffer[16];  std::memset(buffer, 0, sizeof(buffer));  for (uint32_t i = 0; i < kRecords; ++i) {    uint64_t rec = i;    assorted::write_bigendian<uint64_t>(static_cast<uint64_t>(rec % 17U), buffer);    std::string str = std::to_string(rec);    std::memcpy(buffer + sizeof(uint64_t), str.data(), str.size());    uint16_t len = sizeof(uint64_t) + str.size();    uint64_t data;    uint16_t capacity = sizeof(data);    ErrorCode ret = hash.get_record(context, buffer, len, &data, &capacity);    EXPECT_EQ(kErrorCodeOk, ret) << i;    EXPECT_EQ(i + kDataAddendum, data) << i;    EXPECT_EQ(sizeof(data), capacity) << i;  }  Epoch commit_epoch;  ErrorCode committed = xct_manager->precommit_xct(context, &commit_epoch);  EXPECT_EQ(kErrorCodeOk, committed);  return kRetOk;}

void HashDataPage::release_pages_recursive(  const memory::GlobalVolatilePageResolver& page_resolver,  memory::PageReleaseBatch* batch) {  if (next_page_.volatile_pointer_.components.offset != 0) {    HashDataPage* next = reinterpret_cast<HashDataPage*>(      page_resolver.resolve_offset(next_page_.volatile_pointer_));    ASSERT_ND(next->header().get_in_layer_level() == 0);    ASSERT_ND(next->get_bin() == get_bin());    next->release_pages_recursive(page_resolver, batch);    next_page_.volatile_pointer_.components.offset = 0;  }  VolatilePagePointer volatile_id;  volatile_id.word = header().page_id_;  batch->release(volatile_id);}

ErrorStack verify_task(const proc::ProcArguments& args) {  thread::Thread* context = args.context_;  storage::masstree::MasstreeStorage masstree(args.engine_, kName);  ASSERT_ND(masstree.exists());  CHECK_ERROR(masstree.verify_single_thread(context));  xct::XctManager* xct_manager = args.engine_->get_xct_manager();  WRAP_ERROR_CODE(xct_manager->begin_xct(context, xct::kSerializable));  for (uint32_t i = 0; i < kRecords; ++i) {    uint64_t rec = i;    storage::masstree::KeySlice slice = storage::masstree::normalize_primitive<uint64_t>(rec);    uint64_t data;    uint16_t capacity = sizeof(data);    ErrorCode ret = masstree.get_record_normalized(context, slice, &data, &capacity, true);/*    if (ret != kErrorCodeOk || rec != data) {      CHECK_ERROR(masstree.verify_single_thread(context));      CHECK_ERROR(masstree.debugout_single_thread(args.engine_));      std::cout << "asdasd" << std::endl;    }*/    EXPECT_EQ(kErrorCodeOk, ret) << i;    EXPECT_EQ(rec, data) << i;    EXPECT_EQ(sizeof(data), capacity) << i;  }  Epoch commit_epoch;  ErrorCode committed = xct_manager->precommit_xct(context, &commit_epoch);  EXPECT_EQ(kErrorCodeOk, committed);  return kRetOk;}

ErrorStack TpccLoadTask::load_customers_in_district(Wid wid, Did did) {  LOG(INFO) << "Loading Customer for DID=" << static_cast<int>(did) << ", WID=" << wid    << ": " << engine_->get_memory_manager()->dump_free_memory_stat();  // insert to customers_secondary at the end after sorting  memory::AlignedMemory secondary_keys_buffer;  struct Secondary {    char  last_[17];      // +17 -> 17    char  first_[17];     // +17 -> 34    char  padding_[2];    // +2 -> 36    Cid   cid_;           // +4 -> 40    static bool compare(const Secondary &left, const Secondary& right) ALWAYS_INLINE {      int cmp = std::memcmp(left.last_, right.last_, sizeof(left.last_));      if (cmp < 0) {        return true;      } else if (cmp > 0) {        return false;      }      cmp = std::memcmp(left.first_, right.first_, sizeof(left.first_));      if (cmp < 0) {        return true;      } else if (cmp > 0) {        return false;      }        ASSERT_ND(left.cid_ != right.cid_);      if (left.cid_ < right.cid_) {        return true;      } else {        return false;      }    }  };

TpccLoadTask::TpccLoadTask(char* ctime_buffer) {  // Initialize timestamp (for date columns)  time_t t_clock;  ::time(&t_clock);  timestamp_ = ::ctime_r(&t_clock, ctime_buffer);  ASSERT_ND(timestamp_);}

void ThreadPimpl::flush_retired_volatile_page(  uint16_t node,  Epoch current_epoch,  memory::PagePoolOffsetAndEpochChunk* chunk) {  if (chunk->size() == 0) {    return;  }  uint32_t safe_count = chunk->get_safe_offset_count(current_epoch);  while (safe_count < chunk->size() / 10U) {    LOG(WARNING) << "Thread-" << id_ << " can return only "      << safe_count << " out of " << chunk->size()      << " retired pages to node-" << node  << " in epoch=" << current_epoch      << ". This means the thread received so many retired pages in a short time period."      << " Will adavance an epoch to safely return the retired pages."      << " This should be a rare event.";    engine_->get_xct_manager()->advance_current_global_epoch();    current_epoch = engine_->get_xct_manager()->get_current_global_epoch();    LOG(INFO) << "okay, advanced epoch. now we should be able to return more pages";    safe_count = chunk->get_safe_offset_count(current_epoch);  }  VLOG(0) << "Thread-" << id_ << " batch-returning retired volatile pages to node-" << node    << " safe_count/count=" << safe_count << "/" << chunk->size() << ". epoch=" << current_epoch;  memory::PagePool* volatile_pool    = engine_->get_memory_manager()->get_node_memory(node)->get_volatile_pool();  volatile_pool->release(safe_count, chunk);  ASSERT_ND(!chunk->full());}

DivvyupPageGrabBatch::DivvyupPageGrabBatch(Engine* engine): engine_(engine), node_count_(engine->get_options().thread_.group_count_) {  chunks_ = new PagePoolOffsetChunk[node_count_];  for (uint16_t node = 0; node < node_count_; ++node) {    ASSERT_ND(chunks_[node].empty());  }}

 void get_assignments(uint32_t count, uint32_t* result_from, uint32_t* result_to) const {   if (load_threads_ == 1U) {     ASSERT_ND(thread_ordinal_ == 0);     *result_from = 0;     *result_to = count;   } else {     ASSERT_ND(thread_ordinal_ < load_threads_);     uint32_t div = count / load_threads_;     ASSERT_ND(div > 0);     EXPECT_GT(div, 0);     *result_from = div * thread_ordinal_;     *result_to = div * (thread_ordinal_ + 1U);     if (thread_ordinal_ + 1U == load_threads_) {       *result_to = count;     }   } }

bool ImpersonateSession::is_running() const {  ASSERT_ND(thread_->get_control_block()->current_ticket_ >= ticket_);  if (thread_->get_control_block()->current_ticket_ != ticket_) {    return false;  }  return (thread_->get_control_block()->status_ == kWaitingForExecution ||    thread_->get_control_block()->status_ == kRunningTask);}

void hash_data_volatile_page_init(const VolatilePageInitArguments& args) {  ASSERT_ND(args.parent_);  ASSERT_ND(args.page_);  StorageId storage_id = args.parent_->get_header().storage_id_;  HashDataPage* page = reinterpret_cast<HashDataPage*>(args.page_);  PageType parent_type = args.parent_->get_header().get_page_type();  uint64_t bin;  if (parent_type == kHashIntermediatePageType) {    const HashIntermediatePage* parent      = reinterpret_cast<const HashIntermediatePage*>(args.parent_);    ASSERT_ND(args.index_in_parent_ < kHashIntermediatePageFanout);    ASSERT_ND(parent->get_level() == 0);    ASSERT_ND(parent->get_bin_range().length() == kHashIntermediatePageFanout);    bin = parent->get_bin_range().begin_ + args.index_in_parent_;  } else {    ASSERT_ND(parent_type == kHashDataPageType);    ASSERT_ND(args.index_in_parent_ == 0);    const HashDataPage* parent = reinterpret_cast<const HashDataPage*>(args.parent_);    bin = parent->get_bin();  }  HashStorage storage(args.context_->get_engine(), storage_id);  uint8_t bin_bits = storage.get_bin_bits();  uint8_t bin_shifts = storage.get_bin_shifts();  page->initialize_volatile_page(storage_id, args.page_id, args.parent_, bin, bin_bits, bin_shifts);}

void MetaLogger::meta_logger_main() {  LOG(INFO) << "Meta-logger started";  while (!stop_requested_) {    {      uint64_t demand = control_block_->logger_wakeup_.acquire_ticket();      if (!stop_requested_ && !control_block_->has_waiting_log()) {        VLOG(0) << "Meta-logger going to sleep";        control_block_->logger_wakeup_.timedwait(demand, 100000ULL);      }    }    VLOG(0) << "Meta-logger woke up";    if (stop_requested_) {      break;    } else if (!control_block_->has_waiting_log()) {      continue;    }    // we observed buffer_used_ > 0 BEFORE the fence. Now we can read buffer_ safely.    assorted::memory_fence_acq_rel();    ASSERT_ND(control_block_->buffer_used_ > 0);    uint32_t write_size = sizeof(control_block_->buffer_);    ASSERT_ND(control_block_->buffer_used_ <= write_size);    LOG(INFO) << "Meta-logger got a log (" << control_block_->buffer_used_ << " b) to write out";    FillerLogType* filler = reinterpret_cast<FillerLogType*>(      control_block_->buffer_ + control_block_->buffer_used_);    filler->populate(write_size - control_block_->buffer_used_);    ErrorCode er = current_file_->write_raw(write_size, control_block_->buffer_);    if (er != kErrorCodeOk) {      LOG(FATAL) << "Meta-logger couldn't write a log. error=" << get_error_message(er)        << ", os_error=" << assorted::os_error();    }    // also fsync on the file and parent. every, single, time.    // we don't care performance on metadata logger. just make it simple and robust.    bool synced = fs::fsync(current_file_->get_path(), true);    if (!synced) {      LOG(FATAL) << "Meta-logger couldn't fsync. os_error=" << assorted::os_error();    }    assorted::memory_fence_release();    control_block_->buffer_used_ = 0;    assorted::memory_fence_release();    control_block_->durable_offset_ += write_size;  }  LOG(INFO) << "Meta-logger terminated";}

void hash_intermediate_volatile_page_init(const VolatilePageInitArguments& args) {  ASSERT_ND(args.parent_);  // because this is always called for non-root pages.  ASSERT_ND(args.page_);  ASSERT_ND(args.index_in_parent_ < kHashIntermediatePageFanout);  StorageId storage_id = args.parent_->get_header().storage_id_;  HashIntermediatePage* page = reinterpret_cast<HashIntermediatePage*>(args.page_);  ASSERT_ND(args.parent_->get_header().get_page_type() == kHashIntermediatePageType);  const HashIntermediatePage* parent = reinterpret_cast<const HashIntermediatePage*>(args.parent_);  ASSERT_ND(parent->get_level() > 0);  parent->assert_range();  HashBin interval = kHashMaxBins[parent->get_level()];  HashBin parent_begin = parent->get_bin_range().begin_;  HashBin begin = parent_begin + args.index_in_parent_ * interval;  uint8_t level = parent->get_level() - 1U;  page->initialize_volatile_page(storage_id, args.page_id, parent, level, begin);}

 // Compose a key for read/update/scan YcsbKey& build_rus_key(uint32_t total_thread_count) {   // Choose a high-bits field first. Then take a look at that worker's local counter   auto high = rnd_record_select_.uniform_within(0, total_thread_count - 1);   auto cnt = channel_->peek_local_key_counter(engine_, high);   // The load should have inserted at least one record on behalf of this worker   ASSERT_ND(cnt > 0);   auto low = rnd_record_select_.uniform_within(0, cnt - 1);   return key_arena_.build(high, low); }