database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-08-01 06:46:55 +03:00
Code Activity

fix(perf,allocator): reduce CountingAllocator step size to improve its memory consumption reaction speed.

Browse Source
This commit is contained in:
drrtuy
2025-04-29 18:50:45 +00:00
parent 01cc73d416
commit 4d32a59775
7 changed files with 17 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
dbcon/joblist/tuple-bps.cpp
View File

@ -2221,7 +2221,7 @@ void TupleBPS::processByteStreamVector(vector<boost::shared_ptr<messageqcpp::Byt
// changes made here should also be made there and vice versa.
if (hasUMJoin || !fBPP->pmSendsFinalResult())
{
utils::setThreadName("BSPJoin");
utils::setThreadName("BPSJoin");
data->joinedData = RGData(data->local_outputRG);
data->local_outputRG.setData(&data->joinedData);

3
tests/CMakeLists.txt
View File

@ -94,9 +94,6 @@ if (WITH_UNITTESTS)
target_link_libraries(stlpoolallocator ${ENGINE_LDFLAGS} ${ENGINE_WRITE_LIBS} ${GTEST_LIBRARIES})
gtest_add_tests(TARGET stlpoolallocator TEST_PREFIX columnstore:)
add_executable(oid_server_return oid-server-return.cpp)
target_link_libraries(oid_server_return ${ENGINE_LDFLAGS} ${ENGINE_WRITE_LIBS})
add_executable(comparators_tests comparators-tests.cpp)
target_link_libraries(comparators_tests ${ENGINE_LDFLAGS} ${ENGINE_WRITE_LIBS} ${CPPUNIT_LIBRARIES} cppunit)
add_test(NAME columnstore:comparators_tests COMMAND comparators_tests)

66
tests/stlpoolallocator.cpp
View File

@ -116,94 +116,36 @@ TEST_F(STLPoolAllocatorTest, VectorIntegration)
vec.shrink_to_fit();
}
// /**
// * Test multithreaded allocation
// */
// TEST_F(STLPoolAllocatorTest, MultithreadedAllocation)
// {
// const int THREAD_COUNT = 4;
// const size_t ALLOC_SIZE = 100;
// std::vector<std::thread> threads;
// Allocator alloc;
// uint64_t initialUsage = alloc.getMemUsage();
// // Create threads that will allocate memory
// for (int i = 0; i < THREAD_COUNT; ++i)
// {
// threads.emplace_back(
// [&alloc]()
// {
// std::vector<TestType*, STLPoolAllocator<TestType*>> ptrs(alloc);
// for (size_t j = 0; j < 10; ++j)
// {
// TestType* ptr = alloc.allocate(ALLOC_SIZE);
// for (size_t k = 0; k < ALLOC_SIZE; ++k)
// {
// alloc.construct(ptr + k, static_cast<TestType>(k));
// }
// ptrs.push_back(ptr);
// }
// // Cleanup
// for (auto ptr : ptrs)
// {
// for (size_t k = 0; k < ALLOC_SIZE; ++k)
// {
// alloc.destroy(ptr + k);
// }
// alloc.deallocate(ptr, ALLOC_SIZE);
// }
// });
// }
// // Wait for all threads to complete
// for (auto& th : threads)
// {
// th.join();
// }
// // Memory usage should be greater than initial due to vector allocations
// EXPECT_GT(alloc.getMemUsage(), initialUsage);
// }
/**
* Test ResourceManager integration
*/
TEST_F(STLPoolAllocatorTest, ResourceManagerIntegration)
{
using TestType = int8_t;
using Allocator = STLPoolAllocator<TestType>;
joblist::ResourceManager rm(true, nullptr);
// To set the memory allowance
rm.setMemory(MemoryAllowance);
std::cout << "Memory allowance: " << MemoryAllowance << std::endl;
Allocator alloc(&rm, 1024, 512);
Allocator alloc(&rm);
std::cout << "Memory available 1 : " << rm.availableMemory() << std::endl;
// Basic allocation test with ResourceManager
TestType* ptr = alloc.allocate(1);
ASSERT_NE(ptr, nullptr);
std::cout << "Memory available 2 : " << rm.availableMemory() << std::endl;
alloc.construct(ptr, 42);
EXPECT_EQ(*ptr, 42);
alloc.destroy(ptr);
alloc.deallocate(ptr, 1);
std::cout << "Memory available 3 : " << rm.availableMemory() << std::endl;
TestType* ptr2 = alloc.allocate(65537);
ASSERT_NE(ptr2, nullptr);
alloc.construct(ptr2, 42);
EXPECT_EQ(*ptr2, 42);
std::cout << "Memory available 4 : " << rm.availableMemory() << std::endl;
alloc.destroy(ptr2);
alloc.deallocate(ptr2, 1);
std::cout << "Memory available 5 : " << rm.availableMemory() << std::endl;
}
/**

5
utils/common/countingallocator.h
View File

@ -36,8 +36,9 @@ namespace allocators
// When a sync op hits MemoryLimitLowerBound trying to allocate more memory, it throws.
// SQL operators or TBPS runtime must catch the exception and act acordingly.
const constexpr int64_t MemoryLimitLowerBound = 500 * 1024 * 1024; // WIP
const constexpr int64_t CheckPointStepSize = 100 * 1024 * 1024; // WIP
const constexpr int64_t MemoryLimitLowerBound = 500 * 1024 * 1024;
// Higher values demonstrate slower response to memory limit violations.
const constexpr int64_t CheckPointStepSize = 1024;
// Custom Allocator that tracks allocated memory using an atomic counter
template <typename T>

7
utils/common/poolallocator.cpp
View File

@ -57,19 +57,15 @@ void PoolAllocator::deallocateAll()
void PoolAllocator::newBlock()
{
capacityRemaining = allocSize;
std::cout << "PoolAllocator new block" << std::endl;
if (!tmpSpace || mem.size() == 0)
{
if (alloc)
{
// std::cout << "PoolAllocator new block with counting alloc" << std::endl;
mem.emplace_back(boost::allocate_shared<PoolAllocatorBufType>(*alloc, allocSize));
}
else
{
// std::cout << "PoolAllocator new block w/o counting alloc" << std::endl;
mem.emplace_back(boost::make_shared<PoolAllocatorBufType>(allocSize));
}
nextAlloc = mem.back().get();
@ -81,17 +77,14 @@ void PoolAllocator::newBlock()
void* PoolAllocator::allocOOB(uint64_t size)
{
OOBMemInfo memInfo;
// std::cout << "PoolAllocator allocOOB" << std::endl;
memUsage += size;
if (alloc)
{
// std::cout << "PoolAllocator allocOOB with counting alloc" << std::endl;
memInfo.mem = boost::allocate_shared<PoolAllocatorBufType>(*alloc, size);
}
else
{
// std::cout << "PoolAllocator allocOOB w/o counting alloc" << std::endl;
memInfo.mem = boost::make_shared<PoolAllocatorBufType>(size);
}
memInfo.size = size;

2
utils/common/poolallocator.h
View File

@ -59,7 +59,6 @@ class PoolAllocator
, useLock(_useLock)
, lock(false)
{
std::cout << "PoolAllocator w/o counting allocator created" << std::endl;
}
PoolAllocator(allocators::CountingAllocator<PoolAllocatorBufType> alloc, unsigned windowSize = DEFAULT_WINDOW_SIZE,
bool isTmpSpace = false, bool _useLock = false)
@ -72,7 +71,6 @@ class PoolAllocator
, lock(false)
, alloc(alloc)
{
std::cout << "PoolAllocator with counting allocator created" << std::endl;
}
PoolAllocator(const PoolAllocator& p)
: allocSize(p.allocSize)

14
utils/common/stlpoolallocator.h
View File

@ -38,7 +38,7 @@ namespace utils
as the deleter. */
struct BoostPoolDeallocator
{
inline void operator()(void* ptr){};
inline void operator()(void* ptr) {};
};
/* This is an STL-compliant wrapper for PoolAllocator + an optimization for containers
@ -62,7 +62,9 @@ class STLPoolAllocator
};
STLPoolAllocator() throw();
STLPoolAllocator(joblist::ResourceManager* rm);
STLPoolAllocator(joblist::ResourceManager* rm,
const int64_t checkPointStepSize = allocators::CheckPointStepSize,
const int64_t memoryLimitLowerBound = allocators::MemoryLimitLowerBound);
STLPoolAllocator(const STLPoolAllocator&) throw();
STLPoolAllocator(uint32_t capacity) throw();
template <class U>
@ -97,17 +99,17 @@ STLPoolAllocator<T>::STLPoolAllocator() throw()
}
template <class T>
STLPoolAllocator<T>::STLPoolAllocator(joblist::ResourceManager* rm)
STLPoolAllocator<T>::STLPoolAllocator(joblist::ResourceManager* rm,
const int64_t checkPointStepSize,
const int64_t memoryLimitLowerBound)
{
if (rm)
{
// std::cout << "STLPoolAllocator with RM " << std::endl;
auto alloc = rm->getAllocator<PoolAllocatorBufType>(1024);
auto alloc = rm->getAllocator<PoolAllocatorBufType>(checkPointStepSize, memoryLimitLowerBound);
pa.reset(new PoolAllocator(alloc, DEFAULT_SIZE));
}
else
{
// std::cout << "STLPoolAllocator w/o RM " << std::endl;
pa.reset(new PoolAllocator(DEFAULT_SIZE));
}
}