1
0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-08-05 16:15:50 +03:00

chore(safety): Locked Map for MessageqCpp with a simpler way

Please enter the commit message for your changes. Lines starting
This commit is contained in:
Leonid Fedorov
2025-05-29 02:09:20 +00:00
parent 9fc3cd8c8f
commit 3948d89a01
2 changed files with 54 additions and 48 deletions

View File

@@ -25,40 +25,57 @@
#include <new> #include <new>
#include <type_traits> #include <type_traits>
namespace messageqcpp namespace messageqcpp
{ {
using ClientMapType = std::multimap<std::string, std::unique_ptr<ClientObject>>; using ClientMapType = std::multimap<std::string, std::unique_ptr<ClientObject>>;
struct LockedClientMap template <class T>
struct Immortal
{ {
LockedClientMap() template <class... Args>
Immortal(Args&&... args)
{ {
::new (space) T(std::forward<Args>(args)...);
} }
~LockedClientMap()
operator T&() & noexcept
{ {
return reinterpret_cast<T&>(space);
} }
ClientMapType clientMap;
std::mutex queueMutex; private:
alignas(T) unsigned char space[sizeof(T)];
}; };
static int clientMapNiftyCounter; class LockedClientMap
static typename std::aligned_storage<sizeof(LockedClientMap), alignof(LockedClientMap)>::type clientMapBuf;
auto& lockedMap = reinterpret_cast<LockedClientMap&>(clientMapBuf);
LockedClientMapInitilizer::LockedClientMapInitilizer ()
{ {
if (clientMapNiftyCounter++ == 0) new (&lockedMap) LockedClientMap (); // placement new struct KeyToUsePrivateCtor
} {
LockedClientMapInitilizer::~LockedClientMapInitilizer () explicit KeyToUsePrivateCtor() = default;
{ };
if (--clientMapNiftyCounter == 0) (&lockedMap)->~LockedClientMap();
}
public:
LockedClientMap(const LockedClientMap&) = delete;
LockedClientMap& operator=(const LockedClientMap&) = delete;
~LockedClientMap() = delete;
static LockedClientMap& getInstance()
{
static Immortal<LockedClientMap> instance(KeyToUsePrivateCtor{});
return instance;
}
LockedClientMap(KeyToUsePrivateCtor)
{
}
private:
ClientMapType clientMap;
std::mutex queueMutex;
friend class MessageQueueClientPool;
};
// 300 seconds idle until cleanup // 300 seconds idle until cleanup
#define MAX_IDLE_TIME 300 #define MAX_IDLE_TIME 300
@@ -70,7 +87,7 @@ static uint64_t TimeSpecToSeconds(struct timespec* ts)
MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrIp, uint64_t port) MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrIp, uint64_t port)
{ {
auto lock = std::scoped_lock(lockedMap.queueMutex); auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
std::ostringstream oss; std::ostringstream oss;
oss << dnOrIp << "_" << port; oss << dnOrIp << "_" << port;
@@ -93,14 +110,13 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrI
newClientObject->client.reset(new MessageQueueClient(dnOrIp, port)); newClientObject->client.reset(new MessageQueueClient(dnOrIp, port));
newClientObject->inUse = true; newClientObject->inUse = true;
newClientObject->lastUsed = nowSeconds; newClientObject->lastUsed = nowSeconds;
lockedMap.clientMap.emplace(std::move(searchString), std::move(newClientObject)); LockedClientMap::getInstance().clientMap.emplace(std::move(searchString), std::move(newClientObject));
return newClientObject->client.get(); return newClientObject->client.get();
} }
MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& module) MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& module)
{ {
auto lock = std::scoped_lock(lockedMap.queueMutex); auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
MessageQueueClient* returnClient = MessageQueueClientPool::findInPool(module); MessageQueueClient* returnClient = MessageQueueClientPool::findInPool(module);
@@ -116,13 +132,11 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& modul
clock_gettime(CLOCK_MONOTONIC, &now); clock_gettime(CLOCK_MONOTONIC, &now);
uint64_t nowSeconds = TimeSpecToSeconds(&now); uint64_t nowSeconds = TimeSpecToSeconds(&now);
newClientObject->client.reset(new MessageQueueClient(module)); newClientObject->client.reset(new MessageQueueClient(module));
newClientObject->inUse = true; newClientObject->inUse = true;
newClientObject->lastUsed = nowSeconds; newClientObject->lastUsed = nowSeconds;
auto result = newClientObject->client.get(); auto result = newClientObject->client.get();
lockedMap.clientMap.emplace(std::move(module), std::move(newClientObject)); LockedClientMap::getInstance().clientMap.emplace(std::move(module), std::move(newClientObject));
return result; return result;
} }
@@ -133,11 +147,10 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
uint64_t nowSeconds = TimeSpecToSeconds(&now); uint64_t nowSeconds = TimeSpecToSeconds(&now);
MessageQueueClient* returnClient = NULL; MessageQueueClient* returnClient = NULL;
auto it = lockedMap.clientMap.begin(); auto it = LockedClientMap::getInstance().clientMap.begin();
// Scan pool // Scan pool
while (it != lockedMap.clientMap.end()) while (it != LockedClientMap::getInstance().clientMap.end())
{ {
ClientObject* clientObject = it->second.get(); ClientObject* clientObject = it->second.get();
uint64_t elapsedTime = nowSeconds - clientObject->lastUsed; uint64_t elapsedTime = nowSeconds - clientObject->lastUsed;
@@ -149,7 +162,7 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
// Do this so we don't invalidate current interator // Do this so we don't invalidate current interator
auto toDelete = it; auto toDelete = it;
it++; it++;
lockedMap.clientMap.erase(toDelete); LockedClientMap::getInstance().clientMap.erase(toDelete);
continue; continue;
} }
@@ -163,7 +176,7 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
// Do this so we don't invalidate current interator // Do this so we don't invalidate current interator
auto toDelete = it; auto toDelete = it;
it++; it++;
lockedMap.clientMap.erase(toDelete); LockedClientMap::getInstance().clientMap.erase(toDelete);
continue; continue;
} }
} }
@@ -193,10 +206,10 @@ void MessageQueueClientPool::releaseInstance(MessageQueueClient* client)
if (client == NULL) if (client == NULL)
return; return;
auto lock = std::scoped_lock(lockedMap.queueMutex); auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
auto it = lockedMap.clientMap.begin(); auto it = LockedClientMap::getInstance().clientMap.begin();
while (it != lockedMap.clientMap.end()) while (it != LockedClientMap::getInstance().clientMap.end())
{ {
if (it->second->client.get() == client) if (it->second->client.get() == client)
{ {
@@ -221,15 +234,14 @@ void MessageQueueClientPool::deleteInstance(MessageQueueClient* client)
if (client == NULL) if (client == NULL)
return; return;
auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
auto it = LockedClientMap::getInstance().clientMap.begin();
auto lock = std::scoped_lock(lockedMap.queueMutex); while (it != LockedClientMap::getInstance().clientMap.end())
auto it = lockedMap.clientMap.begin();
while (it != lockedMap.clientMap.end())
{ {
if (it->second->client.get() == client) if (it->second->client.get() == client)
{ {
lockedMap.clientMap.erase(it); LockedClientMap::getInstance().clientMap.erase(it);
return; return;
} }

View File

@@ -26,12 +26,6 @@
namespace messageqcpp namespace messageqcpp
{ {
static struct LockedClientMapInitilizer {
LockedClientMapInitilizer ();
~LockedClientMapInitilizer ();
} clientMapInitilizer; // static initializer for every translation unit
struct ClientObject struct ClientObject
{ {
std::unique_ptr<MessageQueueClient> client; std::unique_ptr<MessageQueueClient> client;
@@ -49,8 +43,8 @@ class MessageQueueClientPool
static MessageQueueClient* findInPool(const std::string& search); static MessageQueueClient* findInPool(const std::string& search);
private: private:
MessageQueueClientPool(){}; MessageQueueClientPool() {};
~MessageQueueClientPool(){}; ~MessageQueueClientPool() {};
}; };
} // namespace messageqcpp } // namespace messageqcpp