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

MCOL-1551 CS now supports hostnames in Columnstore.xml.

Browse Source
This commit is contained in:
drrtuy
2018-07-24 23:05:09 +03:00
committed by Roman Nozdrin
parent f9f6dc43dd
commit 0e856ce9b0
4 changed files with 65 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
utils/messageqcpp/messagequeue.cpp
View File

@@ -154,24 +154,42 @@ void MessageQueueClient::setup(bool syncProto)
{ {
string otherEndIPStr; string otherEndIPStr;
string otherEndPortStr; string otherEndPortStr;
uint16_t port; struct addrinfo hints, *servinfo;
int rc = 0;
otherEndIPStr = fConfig->getConfig(fOtherEnd, "IPAddr"); otherEndIPStr = fConfig->getConfig(fOtherEnd, "IPAddr");
otherEndPortStr = fConfig->getConfig(fOtherEnd, "Port"); otherEndPortStr = fConfig->getConfig(fOtherEnd, "Port");
if (otherEndIPStr.length() == 0) otherEndIPStr = "127.0.0.1"; if (otherEndIPStr.length() == 0) otherEndIPStr = "127.0.0.1";
if (otherEndPortStr.length() == 0 || (port = static_cast<uint16_t>(strtol(otherEndPortStr.c_str(), 0, 0))) == 0) if (otherEndPortStr.length() == 0 || static_cast<uint16_t>(strtol(otherEndPortStr.c_str(), 0, 0)) == 0)
{ {
string msg = "MessageQueueClient::MessageQueueClient: config error: Invalid/Missing Port attribute"; string msg = "MessageQueueClient::setup(): config error: Invalid/Missing Port attribute";
throw runtime_error(msg); throw runtime_error(msg);
} }
memset(&hints, 0, sizeof hints);
// ATM We support IPv4 only.
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
if( !(rc = getaddrinfo(otherEndIPStr.c_str(), otherEndPortStr.c_str(), &hints, &servinfo)) )
{
memset(&fServ_addr, 0, sizeof(fServ_addr)); memset(&fServ_addr, 0, sizeof(fServ_addr));
sockaddr_in* sinp = reinterpret_cast<sockaddr_in*>(&fServ_addr); sockaddr_in* sinp = reinterpret_cast<sockaddr_in*>(&fServ_addr);
sinp->sin_family = AF_INET; *sinp = *reinterpret_cast<sockaddr_in*>(servinfo->ai_addr);
sinp->sin_port = htons(port); freeaddrinfo(servinfo);
sinp->sin_addr.s_addr = inet_addr(otherEndIPStr.c_str()); }
else
{
string msg = "MessageQueueClient::setup(): ";
msg.append(gai_strerror(rc));
logging::Message::Args args;
logging::LoggingID li(31);
args.add(msg);
fLogger.logMessage(logging::LOG_TYPE_ERROR, logging::M0000, args, li);
}
#ifdef SKIP_IDB_COMPRESSION #ifdef SKIP_IDB_COMPRESSION
fClientSock.setSocketImpl(new InetStreamSocket()); fClientSock.setSocketImpl(new InetStreamSocket());
@@ -197,15 +215,34 @@ MessageQueueClient::MessageQueueClient(const string& otherEnd, Config* config, b
setup(syncProto); setup(syncProto);
} }
MessageQueueClient::MessageQueueClient(const string& ip, uint16_t port, bool syncProto) : MessageQueueClient::MessageQueueClient(const string& dnOrIp, uint16_t port, bool syncProto) :
fLogger(31), fIsAvailable(true) fLogger(31), fIsAvailable(true)
{ {
struct addrinfo hints, *servinfo;
int rc = 0;
memset(&hints, 0, sizeof hints);
// ATM We support IPv4 only.
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
if( !(rc = getaddrinfo(dnOrIp.c_str(), NULL, &hints, &servinfo)) )
{
memset(&fServ_addr, 0, sizeof(fServ_addr)); memset(&fServ_addr, 0, sizeof(fServ_addr));
sockaddr_in* sinp = reinterpret_cast<sockaddr_in*>(&fServ_addr); sockaddr_in* sinp = reinterpret_cast<sockaddr_in*>(&fServ_addr);
sinp->sin_family = AF_INET; *sinp = *reinterpret_cast<sockaddr_in*>(servinfo->ai_addr);
sinp->sin_port = htons(port); sinp->sin_port = htons(port);
sinp->sin_addr.s_addr = inet_addr(ip.c_str()); freeaddrinfo(servinfo);
}
else
{
string msg = "MessageQueueClient::MessageQueueClient(): ";
msg.append(gai_strerror(rc));
logging::Message::Args args;
logging::LoggingID li(31);
args.add(msg);
fLogger.logMessage(logging::LOG_TYPE_ERROR, logging::M0000, args, li);
}
#ifdef SKIP_IDB_COMPRESSION #ifdef SKIP_IDB_COMPRESSION
fClientSock.setSocketImpl(new InetStreamSocket()); fClientSock.setSocketImpl(new InetStreamSocket());
#else #else

3
utils/messageqcpp/messagequeue.h
View File

@@ -33,6 +33,7 @@
#include <stdio.h> #include <stdio.h>
#else #else
#include <netinet/in.h> #include <netinet/in.h>
#include <netdb.h>
#endif #endif
#include "serversocket.h" #include "serversocket.h"
@@ -182,7 +183,7 @@ public:
* *
* construct a queue from this process to otherEnd on the given IP and Port. * construct a queue from this process to otherEnd on the given IP and Port.
*/ */
EXPORT explicit MessageQueueClient(const std::string& ip, uint16_t port, bool syncProto=true); EXPORT explicit MessageQueueClient(const std::string& dnOrIp, uint16_t port, bool syncProto=true);
/** /**

6
utils/messageqcpp/messagequeuepool.cpp
View File

@@ -36,12 +36,12 @@ static uint64_t TimeSpecToSeconds(struct timespec* ts)
return (uint64_t)ts->tv_sec + (uint64_t)ts->tv_nsec / 1000000000; return (uint64_t)ts->tv_sec + (uint64_t)ts->tv_nsec / 1000000000;
} }
MessageQueueClient *MessageQueueClientPool::getInstance(const std::string &ip, uint64_t port) MessageQueueClient *MessageQueueClientPool::getInstance(const std::string &dnOrIp, uint64_t port)
{ {
boost::mutex::scoped_lock lock(queueMutex); boost::mutex::scoped_lock lock(queueMutex);
std::ostringstream oss; std::ostringstream oss;
oss << ip << "_" << port; oss << dnOrIp << "_" << port;
std::string searchString = oss.str(); std::string searchString = oss.str();
MessageQueueClient *returnClient = MessageQueueClientPool::findInPool(searchString); MessageQueueClient *returnClient = MessageQueueClientPool::findInPool(searchString);
@@ -58,7 +58,7 @@ MessageQueueClient *MessageQueueClientPool::getInstance(const std::string &ip, u
clock_gettime(CLOCK_MONOTONIC, &now); clock_gettime(CLOCK_MONOTONIC, &now);
uint64_t nowSeconds = TimeSpecToSeconds(&now); uint64_t nowSeconds = TimeSpecToSeconds(&now);
newClientObject->client = new MessageQueueClient(ip, port); newClientObject->client = new MessageQueueClient(dnOrIp, port);
newClientObject->inUse = true; newClientObject->inUse = true;
newClientObject->lastUsed = nowSeconds; newClientObject->lastUsed = nowSeconds;
clientMap.insert(std::pair<std::string, ClientObject*>(searchString, newClientObject)); clientMap.insert(std::pair<std::string, ClientObject*>(searchString, newClientObject));

2
utils/messageqcpp/messagequeuepool.h
View File

@@ -41,7 +41,7 @@ class MessageQueueClientPool
{ {
public: public:
static MessageQueueClient *getInstance(const std::string &module); static MessageQueueClient *getInstance(const std::string &module);
static MessageQueueClient *getInstance(const std::string &ip, uint64_t port); static MessageQueueClient *getInstance(const std::string &dnOrIp, uint64_t port);
static void releaseInstance(MessageQueueClient * client); static void releaseInstance(MessageQueueClient * client);
static void deleteInstance(MessageQueueClient * client); static void deleteInstance(MessageQueueClient * client);
static MessageQueueClient *findInPool(const std::string &search); static MessageQueueClient *findInPool(const std::string &search);