From 0e856ce9b0e8925938d1651c87bd571287df37fb Mon Sep 17 00:00:00 2001 From: drrtuy Date: Tue, 24 Jul 2018 23:05:09 +0300 Subject: [PATCH] MCOL-1551 CS now supports hostnames in Columnstore.xml. --- utils/messageqcpp/messagequeue.cpp | 81 +++++++++++++++++++------- utils/messageqcpp/messagequeue.h | 3 +- utils/messageqcpp/messagequeuepool.cpp | 6 +- utils/messageqcpp/messagequeuepool.h | 2 +- 4 files changed, 65 insertions(+), 27 deletions(-) diff --git a/utils/messageqcpp/messagequeue.cpp b/utils/messageqcpp/messagequeue.cpp index 4800faf13..085426d9c 100644 --- a/utils/messageqcpp/messagequeue.cpp +++ b/utils/messageqcpp/messagequeue.cpp @@ -152,26 +152,44 @@ void MessageQueueClient::shutdown() void MessageQueueClient::setup(bool syncProto) { - string otherEndIPStr; - string otherEndPortStr; - uint16_t port; + string otherEndIPStr; + string otherEndPortStr; + struct addrinfo hints, *servinfo; + int rc = 0; - otherEndIPStr = fConfig->getConfig(fOtherEnd, "IPAddr"); - otherEndPortStr = fConfig->getConfig(fOtherEnd, "Port"); + otherEndIPStr = fConfig->getConfig(fOtherEnd, "IPAddr"); + 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(strtol(otherEndPortStr.c_str(), 0, 0))) == 0) - { - string msg = "MessageQueueClient::MessageQueueClient: config error: Invalid/Missing Port attribute"; - throw runtime_error(msg); - } + if (otherEndPortStr.length() == 0 || static_cast(strtol(otherEndPortStr.c_str(), 0, 0)) == 0) + { + string msg = "MessageQueueClient::setup(): config error: Invalid/Missing Port attribute"; + throw runtime_error(msg); + } - memset(&fServ_addr, 0, sizeof(fServ_addr)); - sockaddr_in* sinp = reinterpret_cast(&fServ_addr); - sinp->sin_family = AF_INET; - sinp->sin_port = htons(port); - sinp->sin_addr.s_addr = inet_addr(otherEndIPStr.c_str()); + 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)); + sockaddr_in* sinp = reinterpret_cast(&fServ_addr); + *sinp = *reinterpret_cast(servinfo->ai_addr); + freeaddrinfo(servinfo); + } + 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 fClientSock.setSocketImpl(new InetStreamSocket()); @@ -197,15 +215,34 @@ MessageQueueClient::MessageQueueClient(const string& otherEnd, Config* config, b 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) { - memset(&fServ_addr, 0, sizeof(fServ_addr)); - sockaddr_in* sinp = reinterpret_cast(&fServ_addr); - sinp->sin_family = AF_INET; - sinp->sin_port = htons(port); - sinp->sin_addr.s_addr = inet_addr(ip.c_str()); + 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)); + sockaddr_in* sinp = reinterpret_cast(&fServ_addr); + *sinp = *reinterpret_cast(servinfo->ai_addr); + sinp->sin_port = htons(port); + 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 fClientSock.setSocketImpl(new InetStreamSocket()); #else diff --git a/utils/messageqcpp/messagequeue.h b/utils/messageqcpp/messagequeue.h index 8de4df398..e33e5cd84 100644 --- a/utils/messageqcpp/messagequeue.h +++ b/utils/messageqcpp/messagequeue.h @@ -33,6 +33,7 @@ #include #else #include +#include #endif #include "serversocket.h" @@ -182,7 +183,7 @@ public: * * 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); /** diff --git a/utils/messageqcpp/messagequeuepool.cpp b/utils/messageqcpp/messagequeuepool.cpp index 5b8c9862c..27459991f 100644 --- a/utils/messageqcpp/messagequeuepool.cpp +++ b/utils/messageqcpp/messagequeuepool.cpp @@ -36,12 +36,12 @@ static uint64_t TimeSpecToSeconds(struct timespec* ts) 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); std::ostringstream oss; - oss << ip << "_" << port; + oss << dnOrIp << "_" << port; std::string searchString = oss.str(); MessageQueueClient *returnClient = MessageQueueClientPool::findInPool(searchString); @@ -58,7 +58,7 @@ MessageQueueClient *MessageQueueClientPool::getInstance(const std::string &ip, u clock_gettime(CLOCK_MONOTONIC, &now); uint64_t nowSeconds = TimeSpecToSeconds(&now); - newClientObject->client = new MessageQueueClient(ip, port); + newClientObject->client = new MessageQueueClient(dnOrIp, port); newClientObject->inUse = true; newClientObject->lastUsed = nowSeconds; clientMap.insert(std::pair(searchString, newClientObject)); diff --git a/utils/messageqcpp/messagequeuepool.h b/utils/messageqcpp/messagequeuepool.h index fc5576203..227b13b2c 100644 --- a/utils/messageqcpp/messagequeuepool.h +++ b/utils/messageqcpp/messagequeuepool.h @@ -41,7 +41,7 @@ class MessageQueueClientPool { public: 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 deleteInstance(MessageQueueClient * client); static MessageQueueClient *findInPool(const std::string &search);