Skip to content

Commit a15a723

Browse files
cameron314cameron314
committed
Made implicit producers free dynamically allocated blocks all the way back to the heap by default (see issue #279)
1 parent 9c44f36 commit a15a723

3 files changed

Lines changed: 100 additions & 39 deletions

File tree

‎benchmarks/benchmarks.cpp‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -260,6 +260,9 @@ struct Traits : public moodycamel::ConcurrentQueueDefaultTraits
260260
// block size will improve throughput (which is mostly what
261261
// we're after with these benchmarks).
262262
static const size_t BLOCK_SIZE = 64;
263+
264+
// Reuse blocks once allocated.
265+
static const bool RECYCLE_ALLOCATED_BLOCKS = true;
263266
};
264267

265268

‎concurrentqueue.h‎

Lines changed: 17 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -378,7 +378,14 @@ struct ConcurrentQueueDefaultTraits
378378
// consumer threads exceeds the number of idle cores (in which case try 0-100).
379379
// Only affects instances of the BlockingConcurrentQueue.
380380
static const int MAX_SEMA_SPINS = 10000;
381-
381+
382+
// Whether to recycle dynamically-allocated blocks into an internal free list or
383+
// not. If false, only pre-allocated blocks (controlled by the constructor
384+
// arguments) will be recycled, and all others will be `free`d back to the heap.
385+
// Note that blocks consumed by explicit producers are only freed on destruction
386+
// of the queue (not following destruction of the token) regardless of this trait.
387+
static const bool RECYCLE_ALLOCATED_BLOCKS = false;
388+
382389

383390
#ifndef MCDBGQ_USE_RELACY
384391
// Memory allocation can be customized if needed.
@@ -789,7 +796,7 @@ class ConcurrentQueue
789796
// queue is fully constructed before it starts being used by other threads (this
790797
// includes making the memory effects of construction visible, possibly with a
791798
// memory barrier).
792-
explicit ConcurrentQueue(size_t capacity = 6 * BLOCK_SIZE)
799+
explicit ConcurrentQueue(size_t capacity = 32 * BLOCK_SIZE)
793800
: producerListTail(nullptr),
794801
producerCount(0),
795802
initialBlockPoolIndex(0),
@@ -1538,7 +1545,7 @@ class ConcurrentQueue
15381545
struct Block
15391546
{
15401547
Block()
1541-
: next(nullptr), elementsCompletelyDequeued(0), freeListRefs(0), freeListNext(nullptr), shouldBeOnFreeList(false), dynamicallyAllocated(true)
1548+
: next(nullptr), elementsCompletelyDequeued(0), freeListRefs(0), freeListNext(nullptr), dynamicallyAllocated(true)
15421549
{
15431550
#ifdef MCDBGQ_TRACKMEM
15441551
owner = nullptr;
@@ -1655,7 +1662,6 @@ class ConcurrentQueue
16551662
public:
16561663
std::atomic<std::uint32_t> freeListRefs;
16571664
std::atomic<Block*> freeListNext;
1658-
std::atomic<bool> shouldBeOnFreeList;
16591665
bool dynamicallyAllocated; // Perhaps a better name for this would be 'isNotPartOfInitialBlockPool'
16601666

16611667
#ifdef MCDBGQ_TRACKMEM
@@ -1810,12 +1816,7 @@ class ConcurrentQueue
18101816
auto block = this->tailBlock;
18111817
do {
18121818
auto nextBlock = block->next;
1813-
if (block->dynamicallyAllocated) {
1814-
destroy(block);
1815-
}
1816-
else {
1817-
this->parent->add_block_to_free_list(block);
1818-
}
1819+
this->parent->add_block_to_free_list(block);
18191820
block = nextBlock;
18201821
} while (block != this->tailBlock);
18211822
}
@@ -3054,7 +3055,12 @@ class ConcurrentQueue
30543055
#ifdef MCDBGQ_TRACKMEM
30553056
block->owner = nullptr;
30563057
#endif
3057-
freeList.add(block);
3058+
if (!Traits::RECYCLE_ALLOCATED_BLOCKS && block->dynamicallyAllocated) {
3059+
destroy(block);
3060+
}
3061+
else {
3062+
freeList.add(block);
3063+
}
30583064
}
30593065

30603066
inline void add_blocks_to_free_list(Block* block)

‎tests/unittests/unittests.cpp‎

Lines changed: 80 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -96,7 +96,7 @@ struct MallocTrackingTraits : public ConcurrentQueueDefaultTraits
9696
static inline void free(void* ptr) { tracking_allocator::free(ptr); }
9797
};
9898

99-
template<std::size_t BlockSize = ConcurrentQueueDefaultTraits::BLOCK_SIZE, std::size_t InitialIndexSize = ConcurrentQueueDefaultTraits::EXPLICIT_INITIAL_INDEX_SIZE>
99+
template<std::size_t BlockSize = ConcurrentQueueDefaultTraits::BLOCK_SIZE, std::size_t InitialIndexSize = ConcurrentQueueDefaultTraits::EXPLICIT_INITIAL_INDEX_SIZE, bool RecycleBlocks = ConcurrentQueueDefaultTraits::RECYCLE_ALLOCATED_BLOCKS>
100100
struct TestTraits : public MallocTrackingTraits
101101
{
102102
typedef std::size_t size_t;
@@ -105,6 +105,7 @@ struct TestTraits : public MallocTrackingTraits
105105
static const size_t BLOCK_SIZE = BlockSize;
106106
static const size_t EXPLICIT_INITIAL_INDEX_SIZE = InitialIndexSize;
107107
static const size_t IMPLICIT_INITIAL_INDEX_SIZE = InitialIndexSize * 2;
108+
static const bool RECYCLE_ALLOCATED_BLOCKS = RecycleBlocks;
108109

109110
static inline void reset() { _malloc_count() = 0; _free_count() = 0; }
110111
static inline std::atomic<int>& _malloc_count() { static std::atomic<int> c; return c; }
@@ -991,80 +992,132 @@ class ConcurrentQueueTests : public TestClass<ConcurrentQueueTests>
991992
bool block_alloc()
992993
{
993994
typedef TestTraits<2> Traits;
995+
typedef TestTraits<2, 32, true> RecycleTraits;
994996
Traits::reset();
995997

998+
// Explicit
996999
{
9971000
ConcurrentQueue<int, Traits> q(7);
9981001
ASSERT_OR_FAIL(q.initialBlockPoolSize == 4);
9991002

10001003
ASSERT_OR_FAIL(Traits::malloc_count() == 1);
10011004
ASSERT_OR_FAIL(Traits::free_count() == 0);
10021005

1003-
ProducerToken tok(q);
1006+
{
1007+
ProducerToken tok(q);
1008+
ASSERT_OR_FAIL(Traits::malloc_count() == 3); // one for producer, one for its block index
1009+
ASSERT_OR_FAIL(Traits::free_count() == 0);
1010+
1011+
// Enqueue one item too many (force extra block allocation)
1012+
for (int i = 0; i != 9; ++i) {
1013+
ASSERT_OR_FAIL(q.enqueue(tok, i));
1014+
}
1015+
1016+
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1017+
ASSERT_OR_FAIL(Traits::free_count() == 0);
1018+
1019+
// Still room for one more...
1020+
ASSERT_OR_FAIL(q.enqueue(tok, 9));
1021+
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1022+
ASSERT_OR_FAIL(Traits::free_count() == 0);
1023+
1024+
// No more room without further allocations
1025+
ASSERT_OR_FAIL(!q.try_enqueue(tok, 10));
1026+
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1027+
ASSERT_OR_FAIL(Traits::free_count() == 0);
1028+
1029+
// Check items were enqueued properly
1030+
int item;
1031+
for (int i = 0; i != 10; ++i) {
1032+
ASSERT_OR_FAIL(q.try_dequeue_from_producer(tok, item));
1033+
ASSERT_OR_FAIL(item == i);
1034+
}
1035+
1036+
// Queue should be empty, but not freed
1037+
ASSERT_OR_FAIL(!q.try_dequeue_from_producer(tok, item));
1038+
ASSERT_OR_FAIL(Traits::free_count() == 0);
1039+
}
1040+
// Explicit producers are recycled, so block should still be allocated
1041+
ASSERT_OR_FAIL(Traits::free_count() == 0);
1042+
}
1043+
1044+
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1045+
ASSERT_OR_FAIL(Traits::free_count() == 4);
1046+
1047+
// Implicit
1048+
Traits::reset();
1049+
{
1050+
ConcurrentQueue<int, Traits> q(7);
1051+
ASSERT_OR_FAIL(q.initialBlockPoolSize == 4);
1052+
1053+
ASSERT_OR_FAIL(q.enqueue(39));
1054+
10041055
ASSERT_OR_FAIL(Traits::malloc_count() == 3); // one for producer, one for its block index
10051056
ASSERT_OR_FAIL(Traits::free_count() == 0);
10061057

10071058
// Enqueue one item too many (force extra block allocation)
1008-
for (int i = 0; i != 9; ++i) {
1009-
ASSERT_OR_FAIL(q.enqueue(tok, i));
1059+
for (int i = 0; i != 8; ++i) {
1060+
ASSERT_OR_FAIL(q.enqueue(i));
10101061
}
10111062

10121063
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
10131064
ASSERT_OR_FAIL(Traits::free_count() == 0);
10141065

10151066
// Still room for one more...
1016-
ASSERT_OR_FAIL(q.enqueue(tok, 9));
1067+
ASSERT_OR_FAIL(q.enqueue(8));
10171068
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
10181069
ASSERT_OR_FAIL(Traits::free_count() == 0);
10191070

10201071
// No more room without further allocations
1021-
ASSERT_OR_FAIL(!q.try_enqueue(tok, 10));
1072+
ASSERT_OR_FAIL(!q.try_enqueue(9));
10221073
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
10231074
ASSERT_OR_FAIL(Traits::free_count() == 0);
10241075

10251076
// Check items were enqueued properly
10261077
int item;
1027-
for (int i = 0; i != 10; ++i) {
1028-
ASSERT_OR_FAIL(q.try_dequeue_from_producer(tok, item));
1078+
ASSERT_OR_FAIL(q.try_dequeue(item));
1079+
ASSERT_OR_FAIL(item == 39);
1080+
for (int i = 0; i != 9; ++i) {
1081+
ASSERT_OR_FAIL(q.try_dequeue(item));
10291082
ASSERT_OR_FAIL(item == i);
10301083
}
10311084

1032-
// Queue should be empty, but not freed
1033-
ASSERT_OR_FAIL(!q.try_dequeue_from_producer(tok, item));
1034-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1085+
// Queue should be empty, and extra block freed
1086+
ASSERT_OR_FAIL(!q.try_dequeue(item));
1087+
ASSERT_OR_FAIL(Traits::free_count() == 1);
10351088
}
10361089

10371090
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
10381091
ASSERT_OR_FAIL(Traits::free_count() == 4);
10391092

10401093
// Implicit
1041-
Traits::reset();
1094+
RecycleTraits::reset();
10421095
{
1043-
ConcurrentQueue<int, Traits> q(7);
1096+
ConcurrentQueue<int, RecycleTraits> q(7);
10441097
ASSERT_OR_FAIL(q.initialBlockPoolSize == 4);
10451098

10461099
ASSERT_OR_FAIL(q.enqueue(39));
10471100

1048-
ASSERT_OR_FAIL(Traits::malloc_count() == 3); // one for producer, one for its block index
1049-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1101+
ASSERT_OR_FAIL(RecycleTraits::malloc_count() == 3); // one for producer, one for its block index
1102+
ASSERT_OR_FAIL(RecycleTraits::free_count() == 0);
10501103

10511104
// Enqueue one item too many (force extra block allocation)
10521105
for (int i = 0; i != 8; ++i) {
10531106
ASSERT_OR_FAIL(q.enqueue(i));
10541107
}
10551108

1056-
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1057-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1109+
ASSERT_OR_FAIL(RecycleTraits::malloc_count() == 4);
1110+
ASSERT_OR_FAIL(RecycleTraits::free_count() == 0);
10581111

10591112
// Still room for one more...
10601113
ASSERT_OR_FAIL(q.enqueue(8));
1061-
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1062-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1114+
ASSERT_OR_FAIL(RecycleTraits::malloc_count() == 4);
1115+
ASSERT_OR_FAIL(RecycleTraits::free_count() == 0);
10631116

10641117
// No more room without further allocations
10651118
ASSERT_OR_FAIL(!q.try_enqueue(9));
1066-
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1067-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1119+
ASSERT_OR_FAIL(RecycleTraits::malloc_count() == 4);
1120+
ASSERT_OR_FAIL(RecycleTraits::free_count() == 0);
10681121

10691122
// Check items were enqueued properly
10701123
int item;
@@ -1075,13 +1128,12 @@ class ConcurrentQueueTests : public TestClass<ConcurrentQueueTests>
10751128
ASSERT_OR_FAIL(item == i);
10761129
}
10771130

1078-
// Queue should be empty, but not freed
1131+
// Queue should be empty, but extra block not freed
10791132
ASSERT_OR_FAIL(!q.try_dequeue(item));
1080-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1133+
ASSERT_OR_FAIL(RecycleTraits::free_count() == 0);
10811134
}
1082-
1083-
ASSERT_OR_FAIL(Traits::malloc_count() == 4);
1084-
ASSERT_OR_FAIL(Traits::free_count() == 4);
1135+
ASSERT_OR_FAIL(RecycleTraits::malloc_count() == 4);
1136+
ASSERT_OR_FAIL(RecycleTraits::free_count() == 4);
10851137

10861138
// Super-aligned
10871139
Traits::reset();
@@ -1125,9 +1177,9 @@ class ConcurrentQueueTests : public TestClass<ConcurrentQueueTests>
11251177
ASSERT_OR_FAIL(VeryAligned::errors == 0);
11261178
}
11271179

1128-
// Queue should be empty, but not freed
1180+
// Queue should be empty, and extra block freed
11291181
ASSERT_OR_FAIL(!q.try_dequeue(item));
1130-
ASSERT_OR_FAIL(Traits::free_count() == 0);
1182+
ASSERT_OR_FAIL(Traits::free_count() == 1);
11311183
ASSERT_OR_FAIL(VeryAligned::errors == 0);
11321184
}
11331185

0 commit comments

Comments
 (0)