database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-29 08:21:15 +03:00
Code Activity

Deep build refactoring phase 2 (#3564)

Browse Source 
* configcpp refactored

* chore(build): massive removals, auto add files to debian install file

* chore(build): configure before autobake

* chore(build): use custom cmake commands for components, mariadb-plugin-columnstore.install generated

* chore(build): install deps as separate step for build-packages

* more deps

* chore(codemanagement, build): build refactoring stage2

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

 Please enter the commit message for your changes. Lines starting

* chore(codemanagement, ci): better coredumps handling, deps fixed

* Delete build/bootstrap_mcs.py

* Update charset.cpp (add license)
This commit is contained in:
Leonid Fedorov
2025-07-17 16:14:10 +04:00
committed by GitHub
parent d0ee5dae32
commit 449029a827
107 changed files with 354 additions and 3327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
utils/messageqcpp/CMakeLists.txt
View File

@ -8,7 +8,6 @@ set(messageqcpp_LIB_SRCS
iosocket.cpp
messagequeue.cpp
messagequeuepool.cpp
samenodepseudosocket.cpp
socketparms.cpp
)

92
utils/messageqcpp/messagequeuepool.cpp
View File

@ -25,40 +25,57 @@
#include <new>
#include <type_traits>
namespace messageqcpp
{
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;
static typename std::aligned_storage<sizeof(LockedClientMap), alignof(LockedClientMap)>::type clientMapBuf;
auto& lockedMap = reinterpret_cast<LockedClientMap&>(clientMapBuf);
LockedClientMapInitilizer::LockedClientMapInitilizer ()
class LockedClientMap
{
if (clientMapNiftyCounter++ == 0) new (&lockedMap) LockedClientMap (); // placement new
}
LockedClientMapInitilizer::~LockedClientMapInitilizer ()
{
if (--clientMapNiftyCounter == 0) (&lockedMap)->~LockedClientMap();
}
struct KeyToUsePrivateCtor
{
explicit KeyToUsePrivateCtor() = default;
};
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
#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)
{
auto lock = std::scoped_lock(lockedMap.queueMutex);
auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
std::ostringstream oss;
oss << dnOrIp << "_" << port;
@ -93,14 +110,13 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrI
newClientObject->client.reset(new MessageQueueClient(dnOrIp, port));
newClientObject->inUse = true;
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();
}
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);
@ -116,13 +132,11 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& modul
clock_gettime(CLOCK_MONOTONIC, &now);
uint64_t nowSeconds = TimeSpecToSeconds(&now);
newClientObject->client.reset(new MessageQueueClient(module));
newClientObject->inUse = true;
newClientObject->lastUsed = nowSeconds;
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;
}
@ -133,11 +147,10 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
uint64_t nowSeconds = TimeSpecToSeconds(&now);
MessageQueueClient* returnClient = NULL;
auto it = lockedMap.clientMap.begin();
auto it = LockedClientMap::getInstance().clientMap.begin();
// Scan pool
while (it != lockedMap.clientMap.end())
while (it != LockedClientMap::getInstance().clientMap.end())
{
ClientObject* clientObject = it->second.get();
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
auto toDelete = it;
it++;
lockedMap.clientMap.erase(toDelete);
LockedClientMap::getInstance().clientMap.erase(toDelete);
continue;
}
@ -163,7 +176,7 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
// Do this so we don't invalidate current interator
auto toDelete = it;
it++;
lockedMap.clientMap.erase(toDelete);
LockedClientMap::getInstance().clientMap.erase(toDelete);
continue;
}
}
@ -193,10 +206,10 @@ void MessageQueueClientPool::releaseInstance(MessageQueueClient* client)
if (client == NULL)
return;
auto lock = std::scoped_lock(lockedMap.queueMutex);
auto it = lockedMap.clientMap.begin();
auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
auto it = LockedClientMap::getInstance().clientMap.begin();
while (it != lockedMap.clientMap.end())
while (it != LockedClientMap::getInstance().clientMap.end())
{
if (it->second->client.get() == client)
{
@ -221,15 +234,14 @@ void MessageQueueClientPool::deleteInstance(MessageQueueClient* client)
if (client == NULL)
return;
auto lock = std::scoped_lock(LockedClientMap::getInstance().queueMutex);
auto it = LockedClientMap::getInstance().clientMap.begin();
auto lock = std::scoped_lock(lockedMap.queueMutex);
auto it = lockedMap.clientMap.begin();
while (it != lockedMap.clientMap.end())
while (it != LockedClientMap::getInstance().clientMap.end())
{
if (it->second->client.get() == client)
{
lockedMap.clientMap.erase(it);
LockedClientMap::getInstance().clientMap.erase(it);
return;
}

10
utils/messageqcpp/messagequeuepool.h
View File

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

127
utils/messageqcpp/samenodepseudosocket.cpp
View File

@ -1,127 +0,0 @@
/* Copyright (C) 2024 MariaDB Corp.
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; version 2 of
the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
MA 02110-1301, USA. */
#include <string>
#include "samenodepseudosocket.h"
#include "iosocket.h"
namespace messageqcpp
{
SameNodePseudoSocket::SameNodePseudoSocket(joblist::DistributedEngineComm* exeMgrDecPtr) : dec_(exeMgrDecPtr)
{
assert(dec_);
}
SameNodePseudoSocket::~SameNodePseudoSocket()
{
}
void SameNodePseudoSocket::open()
{
}
void SameNodePseudoSocket::close()
{
}
Socket* SameNodePseudoSocket::clone() const
{
return nullptr;
}
SameNodePseudoSocket::SameNodePseudoSocket(const SameNodePseudoSocket& /*rhs*/)
{
}
SameNodePseudoSocket& SameNodePseudoSocket::operator=(const SameNodePseudoSocket& /*rhs*/)
{
return *this;
}
const SBS SameNodePseudoSocket::read(const struct ::timespec* /*timeout*/, bool* /*isTimeOut*/, Stats* /*stats*/) const
{
return nullptr;
}
// This is the only working method of this class. It puts SBS directly into DEC queue.
void SameNodePseudoSocket::write(SBS msg, Stats* /*stats*/)
{
dec_->addDataToOutput(msg);
}
void SameNodePseudoSocket::write(const ByteStream& /*msg*/, Stats* /*stats*/)
{
}
void SameNodePseudoSocket::write_raw(const ByteStream& /*msg*/, Stats* /*stats*/) const
{
}
void SameNodePseudoSocket::connect(const sockaddr* /*serv_addr*/)
{
}
void SameNodePseudoSocket::bind(const sockaddr* /*serv_addr*/)
{
}
const IOSocket SameNodePseudoSocket::accept(const struct timespec* /*timeout*/)
{
return IOSocket();
}
void SameNodePseudoSocket::listen(int /*backlog*/)
{
}
const std::string SameNodePseudoSocket::toString() const
{
return "";
}
const std::string SameNodePseudoSocket::addr2String() const
{
return "";
}
bool SameNodePseudoSocket::isSameAddr(const Socket* /*rhs*/) const
{
return false;
}
bool SameNodePseudoSocket::isSameAddr(const struct in_addr& /*ipv4Addr*/) const
{
return false;
}
int SameNodePseudoSocket::ping(const std::string& /*ipaddr*/, const struct timespec* /*timeout*/)
{
return 0;
}
bool SameNodePseudoSocket::isConnected() const
{
return true;
}
bool SameNodePseudoSocket::hasData() const
{
return false;
}
} // namespace messageqcpp

99
utils/messageqcpp/samenodepseudosocket.h
View File

@ -1,99 +0,0 @@
/* Copyright (C) 2024 MariaDB Corp.
This program is free software; you can redistribute it and/or
modify it under the terms of the GNU General Public License
as published by the Free Software Foundation; version 2 of
the License.
This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
GNU General Public License for more details.
You should have received a copy of the GNU General Public License
along with this program; if not, write to the Free Software
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
MA 02110-1301, USA. */
#pragma once
#include "../../dbcon/joblist/distributedenginecomm.h"
#include "socket.h"
#include "socketparms.h"
#include "bytestream.h"
namespace messageqcpp
{
class IOSocket;
// This class is a dummy replacement for a TCP socket
// wrapper to communicate with the same node.
class SameNodePseudoSocket : public Socket
{
public:
explicit SameNodePseudoSocket(joblist::DistributedEngineComm* exeMgrDecPtr);
~SameNodePseudoSocket() override;
void write(SBS msg, Stats* stats = nullptr) override;
private:
void bind(const sockaddr* serv_addr) override;
SameNodePseudoSocket(const SameNodePseudoSocket& rhs);
virtual SameNodePseudoSocket& operator=(const SameNodePseudoSocket& rhs);
void connectionTimeout(const struct ::timespec* /*timeout*/) override
{
}
void syncProto(bool /*use*/) override
{
}
int getConnectionNum() const override
{
return 1;
}
inline void socketParms(const SocketParms& /*socket*/) override
{
}
inline const SocketParms socketParms() const override
{
return SocketParms();
}
// all these virtual methods are to stay inaccessable.
inline void sa(const sockaddr* sa) override;
void open() override;
void close() override;
inline bool isOpen() const override;
const SBS read(const struct timespec* timeout = nullptr, bool* isTimeOut = nullptr,
Stats* stats = nullptr) const override;
void write(const ByteStream& msg, Stats* stats = nullptr) override;
void write_raw(const ByteStream& msg, Stats* stats = nullptr) const override;
void listen(int backlog = 5) override;
const IOSocket accept(const struct timespec* timeout = nullptr) override;
void connect(const sockaddr* serv_addr) override;
Socket* clone() const override;
virtual const std::string toString() const;
const std::string addr2String() const override;
bool isSameAddr(const Socket* rhs) const override;
bool isSameAddr(const struct in_addr& ipv4Addr) const override;
static int ping(const std::string& ipaddr, const struct timespec* timeout = nullptr);
bool isConnected() const override;
bool hasData() const override;
joblist::DistributedEngineComm* dec_ = nullptr;
};
inline bool SameNodePseudoSocket::isOpen() const
{
return true;
}
inline void SameNodePseudoSocket::sa(const sockaddr* /*sa*/)
{
}
} // namespace messageqcpp