MCPcopy Create free account
hub / github.com/Snapchat/KeyDB / enqueue_bulk

Function enqueue_bulk

deps/concurrentqueue/concurrentqueue.h:2611–2749  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

2609#endif
2610 template<AllocationMode allocMode, typename It>
2611 bool enqueue_bulk(It itemFirst, size_t count)
2612 {
2613 // First, we need to make sure we have enough room to enqueue all of the elements;
2614 // this means pre-allocating blocks and putting them in the block index (but only if
2615 // all the allocations succeeded).
2616
2617 // Note that the tailBlock we start off with may not be owned by us any more;
2618 // this happens if it was filled up exactly to the top (setting tailIndex to
2619 // the first index of the next block which is not yet allocated), then dequeued
2620 // completely (putting it on the free list) before we enqueue again.
2621
2622 index_t startTailIndex = this->tailIndex.load(std::memory_order_relaxed);
2623 auto startBlock = this->tailBlock;
2624 Block* firstAllocatedBlock = nullptr;
2625 auto endBlock = this->tailBlock;
2626
2627 // Figure out how many blocks we'll need to allocate, and do so
2628 size_t blockBaseDiff = ((startTailIndex + count - 1) & ~static_cast<index_t>(BLOCK_SIZE - 1)) - ((startTailIndex - 1) & ~static_cast<index_t>(BLOCK_SIZE - 1));
2629 index_t currentTailIndex = (startTailIndex - 1) & ~static_cast<index_t>(BLOCK_SIZE - 1);
2630 if (blockBaseDiff > 0) {
2631#ifdef MCDBGQ_NOLOCKFREE_IMPLICITPRODBLOCKINDEX
2632 debug::DebugLock lock(mutex);
2633#endif
2634 do {
2635 blockBaseDiff -= static_cast<index_t>(BLOCK_SIZE);
2636 currentTailIndex += static_cast<index_t>(BLOCK_SIZE);
2637
2638 // Find out where we'll be inserting this block in the block index
2639 BlockIndexEntry* idxEntry = nullptr; // initialization here unnecessary but compiler can't always tell
2640 Block* newBlock;
2641 bool indexInserted = false;
2642 auto head = this->headIndex.load(std::memory_order_relaxed);
2643 assert(!details::circular_less_than<index_t>(currentTailIndex, head));
2644 bool full = !details::circular_less_than<index_t>(head, currentTailIndex + BLOCK_SIZE) || (MAX_SUBQUEUE_SIZE != details::const_numeric_max<size_t>::value && (MAX_SUBQUEUE_SIZE == 0 || MAX_SUBQUEUE_SIZE - BLOCK_SIZE < currentTailIndex - head));
2645
2646 if (full || !(indexInserted = insert_block_index_entry<allocMode>(idxEntry, currentTailIndex)) || (newBlock = this->parent->ConcurrentQueue::template requisition_block<allocMode>()) == nullptr) {
2647 // Index allocation or block allocation failed; revert any other allocations
2648 // and index insertions done so far for this operation
2649 if (indexInserted) {
2650 rewind_block_index_tail();
2651 idxEntry->value.store(nullptr, std::memory_order_relaxed);
2652 }
2653 currentTailIndex = (startTailIndex - 1) & ~static_cast<index_t>(BLOCK_SIZE - 1);
2654 for (auto block = firstAllocatedBlock; block != nullptr; block = block->next) {
2655 currentTailIndex += static_cast<index_t>(BLOCK_SIZE);
2656 idxEntry = get_block_index_entry_for_index(currentTailIndex);
2657 idxEntry->value.store(nullptr, std::memory_order_relaxed);
2658 rewind_block_index_tail();
2659 }
2660 this->parent->add_blocks_to_free_list(firstAllocatedBlock);
2661 this->tailBlock = startBlock;
2662
2663 return false;
2664 }
2665
2666#ifdef MCDBGQ_TRACKMEM
2667 newBlock->owner = this;
2668#endif

Callers

nothing calls this directly

Calls 3

rewind_block_index_tailFunction · 0.85
deref_noexceptFunction · 0.85

Tested by

no test coverage detected