database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-07-04 04:42:30 +03:00
Code Activity

Fixes of bugs from ASAN warnings, part one (#2796)

Browse Source
This commit is contained in:
Leonid Fedorov
2023-03-30 18:29:04 +03:00
committed by GitHub
parent a1d20d82d5
commit 2f153184c3
71 changed files with 591 additions and 2038 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
utils/messageqcpp/messagequeuepool.cpp
View File

@ -15,18 +15,25 @@
Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
MA 02110-1301, USA. */
#include <memory>
#include <sstream>
#include <map>
#include <boost/thread/mutex.hpp>
#include <time.h>
#include <mutex>
#include "messagequeuepool.h"
#include "messagequeue.h"
namespace messageqcpp
{
boost::mutex queueMutex;
std::mutex& getQueueMutex()
{
static std::mutex queueMutex;
return queueMutex;
}
// Make linker happy
std::multimap<std::string, ClientObject*> MessageQueueClientPool::clientMap;
MessageQueueClientPool::ClientMapType MessageQueueClientPool::clientMap;
// 300 seconds idle until cleanup
#define MAX_IDLE_TIME 300
@ -38,7 +45,7 @@ static uint64_t TimeSpecToSeconds(struct timespec* ts)
MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrIp, uint64_t port)
{
boost::mutex::scoped_lock lock(queueMutex);
std::scoped_lock lock(getQueueMutex());
std::ostringstream oss;
oss << dnOrIp << "_" << port;
@ -58,16 +65,16 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& dnOrI
clock_gettime(CLOCK_MONOTONIC, &now);
uint64_t nowSeconds = TimeSpecToSeconds(&now);
newClientObject->client = new MessageQueueClient(dnOrIp, port);
newClientObject->client.reset(new MessageQueueClient(dnOrIp, port));
newClientObject->inUse = true;
newClientObject->lastUsed = nowSeconds;
clientMap.insert(std::pair<std::string, ClientObject*>(searchString, newClientObject));
return newClientObject->client;
clientMap.emplace(std::move(searchString), std::move(newClientObject));
return newClientObject->client.get();
}
MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& module)
{
boost::mutex::scoped_lock lock(queueMutex);
std::scoped_lock lock(getQueueMutex());
MessageQueueClient* returnClient = MessageQueueClientPool::findInPool(module);
@ -78,16 +85,19 @@ MessageQueueClient* MessageQueueClientPool::getInstance(const std::string& modul
}
// We didn't find one, create new one
ClientObject* newClientObject = new ClientObject();
auto newClientObject = std::make_unique<ClientObject>();
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
uint64_t nowSeconds = TimeSpecToSeconds(&now);
newClientObject->client = new MessageQueueClient(module);
newClientObject->client.reset(new MessageQueueClient(module));
newClientObject->inUse = true;
newClientObject->lastUsed = nowSeconds;
clientMap.insert(std::pair<std::string, ClientObject*>(module, newClientObject));
return newClientObject->client;
auto result = newClientObject->client.get();
clientMap.emplace(std::move(module), std::move(newClientObject));
return result;
}
MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search)
@ -97,22 +107,21 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
uint64_t nowSeconds = TimeSpecToSeconds(&now);
MessageQueueClient* returnClient = NULL;
std::multimap<std::string, ClientObject*>::iterator it = clientMap.begin();
auto it = clientMap.begin();
// Scan pool
while (it != clientMap.end())
{
ClientObject* clientObject = it->second;
ClientObject* clientObject = it->second.get();
uint64_t elapsedTime = nowSeconds - clientObject->lastUsed;
// If connection hasn't been used for MAX_IDLE_TIME we probably don't need it so drop it
// Don't drop in use connections that have been in use a long time
if ((elapsedTime >= MAX_IDLE_TIME) && (!clientObject->inUse))
{
delete clientObject->client;
delete clientObject;
// Do this so we don't invalidate current interator
std::multimap<std::string, ClientObject*>::iterator toDelete = it;
auto toDelete = it;
it++;
clientMap.erase(toDelete);
continue;
@ -120,15 +129,13 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
if (!clientObject->inUse)
{
MessageQueueClient* client = clientObject->client;
MessageQueueClient* client = clientObject->client.get();
// If the unused socket isn't connected or has data pending read, destroy it
if (!client->isConnected() || client->hasData())
{
delete client;
delete clientObject;
// Do this so we don't invalidate current interator
std::multimap<std::string, ClientObject*>::iterator toDelete = it;
auto toDelete = it;
it++;
clientMap.erase(toDelete);
continue;
@ -140,7 +147,7 @@ MessageQueueClient* MessageQueueClientPool::findInPool(const std::string& search
{
if ((returnClient == NULL) && (!clientObject->inUse))
{
returnClient = clientObject->client;
returnClient = clientObject->client.get();
clientObject->inUse = true;
return returnClient;
}
@ -160,12 +167,13 @@ void MessageQueueClientPool::releaseInstance(MessageQueueClient* client)
if (client == NULL)
return;
boost::mutex::scoped_lock lock(queueMutex);
std::multimap<std::string, ClientObject*>::iterator it = clientMap.begin();
std::scoped_lock lock(getQueueMutex());
auto it = clientMap.begin();
while (it != clientMap.end())
{
if (it->second->client == client)
if (it->second->client.get() == client)
{
struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now);
@ -188,15 +196,14 @@ void MessageQueueClientPool::deleteInstance(MessageQueueClient* client)
if (client == NULL)
return;
boost::mutex::scoped_lock lock(queueMutex);
std::multimap<std::string, ClientObject*>::iterator it = clientMap.begin();
std::scoped_lock lock(getQueueMutex());
auto it = clientMap.begin();
while (it != clientMap.end())
{
if (it->second->client == client)
if (it->second->client.get() == client)
{
delete it->second->client;
delete it->second;
clientMap.erase(it);
return;
}