You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bb...@apache.org on 2020/11/17 21:01:37 UTC
[geode-native] branch develop updated: GEODE-8553: Fix inter-locks
in ThinClientLocator (#660)
This is an automated email from the ASF dual-hosted git repository.
bbender pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode-native.git
The following commit(s) were added to refs/heads/develop by this push:
new 72455cd GEODE-8553: Fix inter-locks in ThinClientLocator (#660)
72455cd is described below
commit 72455cd0811bf793a8680b839bbc3ed957ad1830
Author: Mario Salazar de Torres <ma...@est.tech>
AuthorDate: Tue Nov 17 22:00:57 2020 +0100
GEODE-8553: Fix inter-locks in ThinClientLocator (#660)
- ThinClientLocatorHelper uses a mutex called m_locatorLock, which is
used in the entire scope of all the public member functions for this
class.
- Problem with that is all of those functions perform some network
communication and in the event of networking issues, this will result
in having all the application threads inter-locked.
- Also, the only purpose of this mutex is to avoid that the locators
list is read while it's being updated.
- Given all of the above, the class has been refactored, so every time
the locators list has to be accessed, a copy of it is created, being
that copy created only while owning the mutex.
- And for the case of the update, a new locators list is composed and
its only at the end of the update function where the mutex is locked
and the locators list swapped by the new one.
- This whole change ensures that the time the mutex is in use is
minimal while the functionality remains intact.
- Refactored ConnectionWrapper so is move constructible.
- Locators list is now initialized using iterator range constructor.
- Added getConnRetries utility function.
- Changed mutex for a shared mutex.
- getLocators function now returns a shuffled list each time.
- Cleaned-up repeated code. As a result of this createConnection has been
refactored and a new function called sendRequest has been created.
- Implemented proposal for locator connections retries, treating the locators
list as a circular buffer.
- Added new integration test to verify that threads don't inter-lock anymore.
- The test works as follow:
* Set 5s connect-timeout.
* Deploy 1 locator.
* Add 1 non-existent locator to the pool.
* Send 64 concurrent requests to get the server list.
* By probability if there are no interlocks ~50%+-E% of the requests should
have completed in a brief period of time.
* Note that E% is the error allowance percentage, which is set to 10%.
* Therefore a 10 seconds timeout is set in order to reach the request
completed treshold.
* Test passes if the number of completed requests before the timeout was
greater or equal to the thresholds, otherwise it fails.
---
cppcache/integration/test/CMakeLists.txt | 1 +
cppcache/integration/test/LocatorRequestsTest.cpp | 122 +++++
cppcache/src/ThinClientLocatorHelper.cpp | 533 +++++++++-------------
cppcache/src/ThinClientLocatorHelper.hpp | 91 +++-
cppcache/src/ThinClientPoolDM.cpp | 3 +-
5 files changed, 403 insertions(+), 347 deletions(-)
diff --git a/cppcache/integration/test/CMakeLists.txt b/cppcache/integration/test/CMakeLists.txt
index fd6ebc5..65d2ddc 100644
--- a/cppcache/integration/test/CMakeLists.txt
+++ b/cppcache/integration/test/CMakeLists.txt
@@ -27,6 +27,7 @@ add_executable(cpp-integration-test
ExampleTest.cpp
ExpirationTest.cpp
FunctionExecutionTest.cpp
+ LocatorRequestsTest.cpp
PartitionRegionOpsTest.cpp
PdxInstanceTest.cpp
PdxJsonTypeTest.cpp
diff --git a/cppcache/integration/test/LocatorRequestsTest.cpp b/cppcache/integration/test/LocatorRequestsTest.cpp
new file mode 100644
index 0000000..4b1a547
--- /dev/null
+++ b/cppcache/integration/test/LocatorRequestsTest.cpp
@@ -0,0 +1,122 @@
+/* Licensed to the Apache Software Foundation (ASF) under one or more *
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to You under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <chrono>
+#include <condition_variable>
+#include <iostream>
+#include <mutex>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include <geode/Cache.hpp>
+#include <geode/CacheFactory.hpp>
+#include <geode/PoolManager.hpp>
+#include <geode/RegionFactory.hpp>
+#include <geode/RegionShortcut.hpp>
+
+#include "framework/Cluster.h"
+#include "framework/Gfsh.h"
+
+using apache::geode::client::Cache;
+using apache::geode::client::CacheableInt16;
+using apache::geode::client::CacheableKey;
+using apache::geode::client::CacheableString;
+using apache::geode::client::CacheFactory;
+using apache::geode::client::IllegalStateException;
+using apache::geode::client::Region;
+using apache::geode::client::RegionShortcut;
+
+Cache createTestCache() {
+ CacheFactory cacheFactory;
+ return cacheFactory.set("log-level", "none")
+ .set("connect-timeout", "5s")
+ .set("statistic-sampling-enabled", "false")
+ .create();
+}
+
+std::shared_ptr<Region> setupCachingProxyRegion(Cache& cache) {
+ auto region = cache.createRegionFactory(RegionShortcut::CACHING_PROXY)
+ .setPoolName("default")
+ .create("region");
+
+ return region;
+}
+
+std::shared_ptr<Region> setupProxyRegion(Cache& cache) {
+ auto region = cache.createRegionFactory(RegionShortcut::PROXY)
+ .setPoolName("default")
+ .create("region");
+
+ return region;
+}
+
+TEST(LocatorRequestsTest, verifyNoInterlock) {
+ const auto N = 64U;
+ const auto ERR_MARGIN = 0.10;
+ const std::chrono::seconds TIMEOUT{10};
+
+ std::mutex mutex;
+ std::thread runners[N];
+ std::condition_variable cv;
+ std::atomic_size_t completed_requests;
+ const auto threshold = std::size_t(N * (0.5 - ERR_MARGIN));
+
+ Cluster cluster{LocatorCount{1}, ServerCount{1}};
+ cluster.start();
+
+ const auto& locator = cluster.getLocators().front();
+ auto cache = createTestCache();
+ auto poolFactory = cache.getPoolManager()
+ .createFactory()
+ .setSubscriptionEnabled(true)
+ .setUpdateLocatorListInterval(std::chrono::seconds(0))
+ .addLocator("0.1.2.3", 10334);
+ cluster.applyLocators(poolFactory);
+ auto pool = poolFactory.create("default");
+
+ completed_requests = 0;
+ std::unique_lock<std::mutex> lk(mutex);
+
+ for (auto i = 0U; i < N;) {
+ runners[i++] = std::thread{[&completed_requests, &cv, &pool] {
+ auto servers = pool->getServers();
+ EXPECT_TRUE(servers);
+ ++completed_requests;
+ cv.notify_one();
+ }};
+ }
+
+ auto before = std::chrono::steady_clock::now();
+ auto res = cv.wait_for(lk, TIMEOUT, [&completed_requests, &threshold] {
+ return completed_requests >= threshold;
+ });
+
+ auto after = std::chrono::steady_clock::now();
+ auto elapsed =
+ std::chrono::duration_cast<std::chrono::milliseconds>(
+ std::chrono::steady_clock::duration{(after - before).count()})
+ .count();
+
+ std::clog << completed_requests.load() << " out of " << N << " in " << elapsed
+ << " ms." << std::endl;
+
+ for (auto i = 0U; i < N;) {
+ runners[i++].join();
+ }
+
+ EXPECT_TRUE(res);
+}
diff --git a/cppcache/src/ThinClientLocatorHelper.cpp b/cppcache/src/ThinClientLocatorHelper.cpp
index a8c58c5..89cc97d 100644
--- a/cppcache/src/ThinClientLocatorHelper.cpp
+++ b/cppcache/src/ThinClientLocatorHelper.cpp
@@ -20,6 +20,8 @@
#include <algorithm>
#include <set>
+#include <boost/thread/lock_types.hpp>
+
#include <geode/DataInput.hpp>
#include <geode/DataOutput.hpp>
#include <geode/SystemProperties.hpp>
@@ -39,313 +41,235 @@ namespace apache {
namespace geode {
namespace client {
-const int BUFF_SIZE = 3000;
-
-class ConnectionWrapper {
- private:
- Connector*& m_conn;
+const size_t BUFF_SIZE = 3000;
+const size_t DEFAULT_CONNECTION_RETRIES = 3;
- public:
- explicit ConnectionWrapper(Connector*& conn) : m_conn(conn) {}
- ~ConnectionWrapper() {
- LOGDEBUG("closing the connection locator1");
- if (m_conn != nullptr) {
- LOGDEBUG("closing the connection locator");
- m_conn->close();
- delete m_conn;
- }
+ThinClientLocatorHelper::ConnectionWrapper::~ConnectionWrapper() {
+ if (conn_ != nullptr) {
+ LOGDEBUG("Closing the locator connection");
+ conn_->close();
+ delete conn_;
}
-};
+}
ThinClientLocatorHelper::ThinClientLocatorHelper(
- const std::vector<std::string>& locatorAddresses,
- const ThinClientPoolDM* poolDM)
- : m_poolDM(poolDM) {
- for (auto&& locatorAddress : locatorAddresses) {
- m_locHostPort.emplace_back(locatorAddress);
- }
-}
+ const std::vector<std::string>& locators, const ThinClientPoolDM* poolDM)
+ : locators_(locators.begin(), locators.end()), m_poolDM(poolDM) {}
ThinClientLocatorHelper::ThinClientLocatorHelper(
- const std::vector<std::string>& locatorAddresses,
- const std::string& sniProxyHost, int sniProxyPort,
- const ThinClientPoolDM* poolDM)
- : m_poolDM(poolDM),
+ const std::vector<std::string>& locators, const std::string& sniProxyHost,
+ int sniProxyPort, const ThinClientPoolDM* poolDM)
+ : locators_(locators.begin(), locators.end()),
+ m_poolDM(poolDM),
m_sniProxyHost(sniProxyHost),
- m_sniProxyPort(sniProxyPort) {
- for (auto&& locatorAddress : locatorAddresses) {
- m_locHostPort.emplace_back(locatorAddress);
+ m_sniProxyPort(sniProxyPort) {}
+
+size_t ThinClientLocatorHelper::getConnRetries() const {
+ auto retries = m_poolDM->getRetryAttempts();
+ return retries <= 0 ? DEFAULT_CONNECTION_RETRIES : retries;
+}
+
+std::vector<ServerLocation> ThinClientLocatorHelper::getLocators() const {
+ decltype(locators_) locators;
+ {
+ boost::shared_lock<decltype(mutex_)> guard{mutex_};
+ if (locators_.empty()) {
+ return {};
+ }
+
+ locators = locators_;
}
+
+ RandGen randGen;
+ std::random_shuffle(locators.begin(), locators.end(), randGen);
+ return locators;
}
-Connector* ThinClientLocatorHelper::createConnection(
- Connector*& conn, const char* hostname, int32_t port,
- std::chrono::microseconds waitSeconds, int32_t maxBuffSizePool) {
- Connector* socket = nullptr;
- auto& systemProperties = m_poolDM->getConnectionManager()
- .getCacheImpl()
- ->getDistributedSystem()
- .getSystemProperties();
- if (systemProperties.sslEnabled()) {
+ThinClientLocatorHelper::ConnectionWrapper
+ThinClientLocatorHelper::createConnection(
+ const ServerLocation& location) const {
+ auto& sys_prop = m_poolDM->getConnectionManager()
+ .getCacheImpl()
+ ->getDistributedSystem()
+ .getSystemProperties();
+
+ Connector* conn;
+ const auto port = location.getPort();
+ auto timeout = sys_prop.connectTimeout();
+ const auto& hostname = location.getServerName();
+ auto buffer_size = m_poolDM->getSocketBufferSize();
+
+ if (sys_prop.sslEnabled()) {
if (m_sniProxyHost.empty()) {
- socket = new TcpSslConn(
- hostname, static_cast<uint16_t>(port), waitSeconds, maxBuffSizePool,
- systemProperties.sslTrustStore(), systemProperties.sslKeyStore(),
- systemProperties.sslKeystorePassword());
+ conn = new TcpSslConn(hostname, static_cast<uint16_t>(port), timeout,
+ buffer_size, sys_prop.sslTrustStore(),
+ sys_prop.sslKeyStore(),
+ sys_prop.sslKeystorePassword());
} else {
- socket =
- new TcpSslConn(hostname, waitSeconds, maxBuffSizePool, m_sniProxyHost,
- m_sniProxyPort, systemProperties.sslTrustStore(),
- systemProperties.sslKeyStore(),
- systemProperties.sslKeystorePassword());
+ conn = new TcpSslConn(hostname, timeout, buffer_size, m_sniProxyHost,
+ m_sniProxyPort, sys_prop.sslTrustStore(),
+ sys_prop.sslKeyStore(),
+ sys_prop.sslKeystorePassword());
}
} else {
- socket = new TcpConn(hostname, port, waitSeconds, maxBuffSizePool);
+ conn = new TcpConn(hostname, port, timeout, buffer_size);
}
- conn = socket;
- socket->init();
- return socket;
-}
-GfErrType ThinClientLocatorHelper::getAllServers(
- std::vector<std::shared_ptr<ServerLocation> >& servers,
- const std::string& serverGrp) {
- std::lock_guard<decltype(m_locatorLock)> guard(m_locatorLock);
+ ConnectionWrapper cw{conn};
+ cw->init();
+ return cw;
+}
- auto& sysProps = m_poolDM->getConnectionManager()
+std::shared_ptr<Serializable> ThinClientLocatorHelper::sendRequest(
+ const ServerLocation& location,
+ const std::shared_ptr<Serializable>& request) const {
+ auto& sys_prop = m_poolDM->getConnectionManager()
.getCacheImpl()
->getDistributedSystem()
.getSystemProperties();
- for (size_t i = 0; i < m_locHostPort.size(); i++) {
- auto& loc = m_locHostPort[i];
- try {
- LOGDEBUG("getAllServers getting servers from server = %s ",
- loc.getServerName().c_str());
- auto buffSize = m_poolDM->getSocketBufferSize();
- Connector* conn = nullptr;
- ConnectionWrapper cw(conn);
- createConnection(conn, loc.getServerName().c_str(), loc.getPort(),
- sysProps.connectTimeout(), buffSize);
- auto request = std::make_shared<GetAllServersRequest>(serverGrp);
- auto data =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
- data.writeInt(static_cast<int32_t>(1001)); // GOSSIPVERSION
- data.writeObject(request);
- auto sentLength = conn->send(
- reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
- data.getBufferLength(), m_poolDM->getReadTimeout());
- if (sentLength <= 0) {
- continue;
- }
- char buff[BUFF_SIZE];
- auto receivedLength =
- conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
- if (receivedLength <= 0) {
- continue;
- }
- auto di =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
- reinterpret_cast<uint8_t*>(buff), receivedLength);
+ try {
+ auto conn = createConnection(location);
+ auto data =
+ m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
+ data.writeInt(static_cast<int32_t>(1001)); // GOSSIPVERSION
+ data.writeObject(request);
+ auto sentLength = conn->send(
+ reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
+ data.getBufferLength(), m_poolDM->getReadTimeout());
+ if (sentLength <= 0) {
+ return nullptr;
+ }
+ char buff[BUFF_SIZE];
+ auto receivedLength =
+ conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
+ if (receivedLength <= 0) {
+ return nullptr;
+ }
- if (di.read() == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
- LOGERROR("SSL is enabled on locator, enable SSL in client as well");
- throw AuthenticationRequiredException(
- "SSL is enabled on locator, enable SSL in client as well");
- }
- di.rewindCursor(1);
+ auto di = m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
+ reinterpret_cast<uint8_t*>(buff), receivedLength);
- auto response =
- std::dynamic_pointer_cast<GetAllServersResponse>(di.readObject());
- servers = response->getServers();
- return GF_NOERR;
- } catch (const AuthenticationRequiredException&) {
- continue;
- } catch (const Exception& excp) {
- LOGFINE("Exception while querying locator: %s: %s",
- excp.getName().c_str(), excp.what());
+ if (di.read() == REPLY_SSL_ENABLED && !sys_prop.sslEnabled()) {
+ LOGERROR("SSL is enabled on locator, enable SSL in client as well");
+ throw AuthenticationRequiredException(
+ "SSL is enabled on locator, enable SSL in client as well");
+ }
+
+ di.rewindCursor(1);
+ return di.readObject();
+ } catch (const AuthenticationRequiredException& excp) {
+ throw excp;
+ } catch (const Exception& excp) {
+ LOGFINE("Exception while querying locator: %s: %s", excp.getName().c_str(),
+ excp.what());
+ }
+
+ return nullptr;
+}
+
+GfErrType ThinClientLocatorHelper::getAllServers(
+ std::vector<std::shared_ptr<ServerLocation> >& servers,
+ const std::string& serverGrp) const {
+ for (const auto& loc : getLocators()) {
+ LOGDEBUG("getAllServers getting servers from server = %s ",
+ loc.getServerName().c_str());
+
+ auto request = std::make_shared<GetAllServersRequest>(serverGrp);
+ auto response = std::dynamic_pointer_cast<GetAllServersResponse>(
+ sendRequest(loc, request));
+ if (response == nullptr) {
continue;
}
+
+ servers = response->getServers();
+ return GF_NOERR;
}
+
return GF_NOERR;
}
GfErrType ThinClientLocatorHelper::getEndpointForNewCallBackConn(
ClientProxyMembershipID& memId, std::list<ServerLocation>& outEndpoint,
std::string&, int redundancy, const std::set<ServerLocation>& exclEndPts,
- const std::string& serverGrp) {
- std::lock_guard<decltype(m_locatorLock)> guard(m_locatorLock);
- auto& sysProps = m_poolDM->getConnectionManager()
- .getCacheImpl()
- ->getDistributedSystem()
- .getSystemProperties();
- int locatorsRetry = 3;
- if (m_poolDM) {
- int poolRetry = m_poolDM->getRetryAttempts();
- locatorsRetry = poolRetry <= 0 ? locatorsRetry : poolRetry;
- }
+ const std::string& serverGrp) const {
+ auto locators = getLocators();
+ auto locatorsSize = locators.size();
+ auto maxAttempts = getConnRetries();
+
LOGFINER(
- "ThinClientLocatorHelper::getEndpointForNewCallBackConn locatorsRetry = "
- "%d ",
- locatorsRetry);
- for (unsigned attempts = 0;
- attempts <
- (m_locHostPort.size() == 1 ? locatorsRetry : m_locHostPort.size());
- attempts++) {
- ServerLocation loc;
- if (m_locHostPort.size() == 1) {
- loc = m_locHostPort[0];
- } else {
- loc = m_locHostPort[attempts];
- }
+ "ThinClientLocatorHelper::getEndpointForNewCallBackConn maxAttempts = "
+ "%zu",
+ maxAttempts);
- try {
- LOGFINER("Querying locator at [%s:%d] for queue server from group [%s]",
- loc.getServerName().c_str(), loc.getPort(), serverGrp.c_str());
- auto buffSize = m_poolDM->getSocketBufferSize();
- Connector* conn = nullptr;
- ConnectionWrapper cw(conn);
- createConnection(conn, loc.getServerName().c_str(), loc.getPort(),
- sysProps.connectTimeout(), buffSize);
- auto request = std::make_shared<QueueConnectionRequest>(
- memId, exclEndPts, redundancy, false, serverGrp);
- auto data =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
- data.writeInt(static_cast<int32_t>(1001)); // GOSSIPVERSION
- data.writeObject(request);
- auto sentLength = conn->send(
- reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
- data.getBufferLength(), m_poolDM->getReadTimeout());
- if (sentLength <= 0) {
- continue;
- }
- char buff[BUFF_SIZE];
- auto receivedLength =
- conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
- if (receivedLength <= 0) {
- continue;
- }
- auto di =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
- reinterpret_cast<uint8_t*>(buff), receivedLength);
-
- const auto acceptanceCode = di.read();
- if (acceptanceCode == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
- LOGERROR("SSL is enabled on locator, enable SSL in client as well");
- throw AuthenticationRequiredException(
- "SSL is enabled on locator, enable SSL in client as well");
- }
- di.rewindCursor(1);
- auto response =
- std::dynamic_pointer_cast<QueueConnectionResponse>(di.readObject());
- outEndpoint = response->getServers();
- return GF_NOERR;
- } catch (const AuthenticationRequiredException& excp) {
- throw excp;
- } catch (const Exception& excp) {
- LOGFINE("Exception while querying locator: %s: %s",
- excp.getName().c_str(), excp.what());
+ for (auto attempt = 0ULL; attempt < maxAttempts;) {
+ const auto& loc = locators[attempt++ % locatorsSize];
+ LOGFINER("Querying locator at [%s:%d] for queue server from group [%s]",
+ loc.getServerName().c_str(), loc.getPort(), serverGrp.c_str());
+
+ auto request = std::make_shared<QueueConnectionRequest>(
+ memId, exclEndPts, redundancy, false, serverGrp);
+ auto response = std::dynamic_pointer_cast<QueueConnectionResponse>(
+ sendRequest(loc, request));
+ if (response == nullptr) {
continue;
}
+
+ outEndpoint = response->getServers();
+ return GF_NOERR;
}
+
throw NoAvailableLocatorsException("Unable to query any locators");
}
GfErrType ThinClientLocatorHelper::getEndpointForNewFwdConn(
ServerLocation& outEndpoint, std::string&,
const std::set<ServerLocation>& exclEndPts, const std::string& serverGrp,
- const TcrConnection* currentServer) {
+ const TcrConnection* currentServer) const {
bool locatorFound = false;
- std::lock_guard<decltype(m_locatorLock)> guard(m_locatorLock);
- auto& sysProps = m_poolDM->getConnectionManager()
- .getCacheImpl()
- ->getDistributedSystem()
- .getSystemProperties();
+ auto locators = getLocators();
+ auto locatorsSize = locators.size();
+ auto maxAttempts = getConnRetries();
- auto poolRetry = m_poolDM->getRetryAttempts();
- auto locatorsRetry = poolRetry <= 0 ? 3 : poolRetry;
LOGFINER(
- "ThinClientLocatorHelper::getEndpointForNewFwdConn locatorsRetry = %d ",
- locatorsRetry);
- for (unsigned attempts = 0;
- attempts <
- (m_locHostPort.size() == 1 ? locatorsRetry : m_locHostPort.size());
- attempts++) {
- ServerLocation serLoc;
- if (m_locHostPort.size() == 1) {
- serLoc = m_locHostPort[0];
+ "ThinClientLocatorHelper::getEndpointForNewFwdConn maxAttempts = %zu",
+ maxAttempts);
+
+ for (auto attempt = 0ULL; attempt < maxAttempts;) {
+ const auto& loc = locators[attempt++ % locatorsSize];
+ LOGFINE("Querying locator at [%s:%d] for server from group [%s]",
+ loc.getServerName().c_str(), loc.getPort(), serverGrp.c_str());
+
+ std::shared_ptr<Serializable> request;
+ if (currentServer == nullptr) {
+ LOGDEBUG("Creating ClientConnectionRequest");
+ request =
+ std::make_shared<ClientConnectionRequest>(exclEndPts, serverGrp);
} else {
- serLoc = m_locHostPort[attempts];
+ LOGDEBUG("Creating ClientReplacementRequest for connection: %s",
+ currentServer->getEndpointObject()->name().c_str());
+ request = std::make_shared<ClientReplacementRequest>(
+ currentServer->getEndpointObject()->name(), exclEndPts, serverGrp);
}
- try {
- LOGFINE("Querying locator at [%s:%d] for server from group [%s]",
- serLoc.getServerName().c_str(), serLoc.getPort(),
- serverGrp.c_str());
- auto buffSize = m_poolDM->getSocketBufferSize();
- Connector* conn = nullptr;
- ConnectionWrapper cw(conn);
- createConnection(conn, serLoc.getServerName().c_str(), serLoc.getPort(),
- sysProps.connectTimeout(), buffSize);
- auto data =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
- data.writeInt(1001); // GOSSIPVERSION
- if (currentServer == nullptr) {
- LOGDEBUG("Creating ClientConnectionRequest");
- std::shared_ptr<ClientConnectionRequest> request =
- std::make_shared<ClientConnectionRequest>(exclEndPts, serverGrp);
- data.writeObject(request);
- } else {
- LOGDEBUG("Creating ClientReplacementRequest for connection: ",
- currentServer->getEndpointObject()->name().c_str());
- std::shared_ptr<ClientReplacementRequest> request =
- std::make_shared<ClientReplacementRequest>(
- currentServer->getEndpointObject()->name(), exclEndPts,
- serverGrp);
- data.writeObject(request);
- }
- auto sentLength = conn->send(
- reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
- data.getBufferLength(), m_poolDM->getReadTimeout());
- if (sentLength <= 0) {
- continue;
- }
- char buff[BUFF_SIZE];
- auto receivedLength =
- conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
- if (receivedLength <= 0) {
- continue; // return GF_EUNDEF;
- }
- auto di =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
- reinterpret_cast<uint8_t*>(buff), receivedLength);
-
- const auto acceptanceCode = di.read();
- if (acceptanceCode == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
- LOGERROR("SSL is enabled on locator, enable SSL in client as well");
- throw AuthenticationRequiredException(
- "SSL is enabled on locator, enable SSL in client as well");
- }
- di.rewindCursor(1);
-
- auto response =
- std::dynamic_pointer_cast<ClientConnectionResponse>(di.readObject());
- response->printInfo();
- if (!response->serverFound()) {
- LOGFINE("Server not found");
- locatorFound = true;
- continue;
- }
- outEndpoint = response->getServerLocation();
- LOGFINE("Server found at [%s:%d]", outEndpoint.getServerName().c_str(),
- outEndpoint.getPort());
- return GF_NOERR;
- } catch (const AuthenticationRequiredException& excp) {
- throw excp;
- } catch (const Exception& excp) {
- LOGFINE("Exception while querying locator: %s: %s",
- excp.getName().c_str(), excp.what());
+
+ auto response = std::dynamic_pointer_cast<ClientConnectionResponse>(
+ sendRequest(loc, request));
+ if (response == nullptr) {
continue;
}
+
+ response->printInfo();
+ if (!response->serverFound()) {
+ LOGFINE("Server not found");
+ locatorFound = true;
+ continue;
+ }
+
+ outEndpoint = response->getServerLocation();
+ LOGFINE("Server found at [%s:%d]", outEndpoint.getServerName().c_str(),
+ outEndpoint.getPort());
+
+ return GF_NOERR;
}
if (locatorFound) {
@@ -357,81 +281,32 @@ GfErrType ThinClientLocatorHelper::getEndpointForNewFwdConn(
GfErrType ThinClientLocatorHelper::updateLocators(
const std::string& serverGrp) {
- std::lock_guard<decltype(m_locatorLock)> guard(m_locatorLock);
- auto& sysProps = m_poolDM->getConnectionManager()
- .getCacheImpl()
- ->getDistributedSystem()
- .getSystemProperties();
+ auto locators = getLocators();
+ for (const auto& loc : locators) {
+ LOGFINER("Querying locator list at: [%s:%d] for update from group [%s]",
+ loc.getServerName().c_str(), loc.getPort(), serverGrp.c_str());
- for (size_t attempts = 0; attempts < m_locHostPort.size(); attempts++) {
- auto&& serLoc = m_locHostPort[attempts];
- Connector* conn = nullptr;
- try {
- auto buffSize = m_poolDM->getSocketBufferSize();
- LOGFINER("Querying locator list at: [%s:%d] for update from group [%s]",
- serLoc.getServerName().c_str(), serLoc.getPort(),
- serverGrp.c_str());
- ConnectionWrapper cw(conn);
- createConnection(conn, serLoc.getServerName().c_str(), serLoc.getPort(),
- sysProps.connectTimeout(), buffSize);
- auto request = std::make_shared<LocatorListRequest>(serverGrp);
- auto data =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput();
- data.writeInt(static_cast<int32_t>(1001)); // GOSSIPVERSION
- data.writeObject(request);
- auto sentLength = conn->send(
- reinterpret_cast<char*>(const_cast<uint8_t*>(data.getBuffer())),
- data.getBufferLength(), m_poolDM->getReadTimeout());
- if (sentLength <= 0) {
- conn = nullptr;
- continue;
- }
- char buff[BUFF_SIZE];
- auto receivedLength =
- conn->receive(buff, BUFF_SIZE, m_poolDM->getReadTimeout());
- if (receivedLength <= 0) {
- continue;
- }
- auto di =
- m_poolDM->getConnectionManager().getCacheImpl()->createDataInput(
- reinterpret_cast<uint8_t*>(buff), receivedLength);
-
- const auto acceptanceCode = di.read();
- if (acceptanceCode == REPLY_SSL_ENABLED && !sysProps.sslEnabled()) {
- LOGERROR("SSL is enabled on locator, enable SSL in client as well");
- throw AuthenticationRequiredException(
- "SSL is enabled on locator, enable SSL in client as well");
- }
- di.rewindCursor(1);
-
- auto response =
- std::dynamic_pointer_cast<LocatorListResponse>(di.readObject());
- auto locators = response->getLocators();
- if (locators.size() > 0) {
- RandGen randGen;
- std::random_shuffle(locators.begin(), locators.end(), randGen);
- }
- std::vector<ServerLocation> temp(m_locHostPort.begin(),
- m_locHostPort.end());
- m_locHostPort.clear();
- m_locHostPort.insert(m_locHostPort.end(), locators.begin(),
- locators.end());
- for (std::vector<ServerLocation>::iterator it = temp.begin();
- it != temp.end(); it++) {
- std::vector<ServerLocation>::iterator it1 =
- std::find(m_locHostPort.begin(), m_locHostPort.end(), *it);
- if (it1 == m_locHostPort.end()) {
- m_locHostPort.push_back(*it);
- }
- }
- return GF_NOERR;
- } catch (const AuthenticationRequiredException& excp) {
- throw excp;
- } catch (const Exception& excp) {
- LOGFINE("Exception while querying locator: %s: %s",
- excp.getName().c_str(), excp.what());
+ auto request = std::make_shared<LocatorListRequest>(serverGrp);
+ auto response = std::dynamic_pointer_cast<LocatorListResponse>(
+ sendRequest(loc, request));
+ if (response == nullptr) {
continue;
}
+
+ auto new_locators = response->getLocators();
+ for (const auto& old_loc : locators) {
+ auto iter = std::find(new_locators.begin(), new_locators.end(), old_loc);
+ if (iter == new_locators.end()) {
+ new_locators.push_back(old_loc);
+ }
+ }
+
+ {
+ boost::unique_lock<decltype(mutex_)> lock(mutex_);
+ locators_.swap(new_locators);
+ }
+
+ return GF_NOERR;
}
return GF_NOTCON;
}
diff --git a/cppcache/src/ThinClientLocatorHelper.hpp b/cppcache/src/ThinClientLocatorHelper.hpp
index df522a3..d59fab9 100644
--- a/cppcache/src/ThinClientLocatorHelper.hpp
+++ b/cppcache/src/ThinClientLocatorHelper.hpp
@@ -21,10 +21,11 @@
#define GEODE_THINCLIENTLOCATORHELPER_H_
#include <list>
-#include <mutex>
#include <set>
#include <string>
+#include <boost/thread/shared_mutex.hpp>
+
#include <geode/internal/geode_globals.hpp>
#include "ClientProxyMembershipID.hpp"
@@ -42,38 +43,94 @@ class Connector;
class ThinClientLocatorHelper {
public:
- ThinClientLocatorHelper(const std::vector<std::string>& locatorAddresses,
+ ThinClientLocatorHelper(const ThinClientLocatorHelper&) = delete;
+ ThinClientLocatorHelper& operator=(const ThinClientLocatorHelper&) = delete;
+
+ ThinClientLocatorHelper(const std::vector<std::string>& locators,
const ThinClientPoolDM* poolDM);
- ThinClientLocatorHelper(const std::vector<std::string>& locatorAddresses,
+ ThinClientLocatorHelper(const std::vector<std::string>& locators,
const std::string& sniProxyHost, int sniProxyPort,
const ThinClientPoolDM* poolDM);
GfErrType getEndpointForNewFwdConn(
ServerLocation& outEndpoint, std::string& additionalLoc,
const std::set<ServerLocation>& exclEndPts,
const std::string& serverGrp = "",
- const TcrConnection* currentServer = nullptr);
+ const TcrConnection* currentServer = nullptr) const;
GfErrType getEndpointForNewCallBackConn(
ClientProxyMembershipID& memId, std::list<ServerLocation>& outEndpoint,
std::string& additionalLoc, int redundancy,
- const std::set<ServerLocation>& exclEndPts, const std::string& serverGrp);
+ const std::set<ServerLocation>& exclEndPts,
+ const std::string& serverGrp) const;
GfErrType getAllServers(
std::vector<std::shared_ptr<ServerLocation> >& servers,
- const std::string& serverGrp);
- int32_t getCurLocatorsNum() {
- return static_cast<int32_t>(m_locHostPort.size());
- }
+ const std::string& serverGrp) const;
+ size_t getCurLocatorsNum() const { return locators_.size(); }
GfErrType updateLocators(const std::string& serverGrp = "");
private:
- Connector* createConnection(Connector*& conn, const char* hostname,
- int32_t port,
- std::chrono::microseconds waitSeconds,
- int32_t maxBuffSizePool = 0);
- std::mutex m_locatorLock;
- std::vector<ServerLocation> m_locHostPort;
+ /**
+ * Auxiliary types
+ */
+
+ class ConnectionWrapper {
+ private:
+ Connector* conn_;
+
+ public:
+ ConnectionWrapper(const ConnectionWrapper&) = delete;
+ ConnectionWrapper& operator=(const ConnectionWrapper&) = delete;
+
+ explicit ConnectionWrapper(Connector* conn) : conn_(conn) {}
+ ConnectionWrapper(ConnectionWrapper&& other) : conn_(other.conn_) {
+ other.conn_ = nullptr;
+ }
+
+ ~ConnectionWrapper();
+
+ Connector* operator->() { return conn_; }
+ };
+
+ /**
+ * Returns the number of connections retries per request
+ * @return Number of connection retries towards locators
+ */
+ size_t getConnRetries() const;
+
+ /**
+ * Returns a shuffled copy of the current locators list
+ * @return Locators list
+ * @note The original list of locators is copied under the mutex scope
+ * @note This method is used instead of directly using the original locators
+ * list in order to avoid having to lock the mutex while establishing
+ * the connection to any of the locators, which is a bad practice in
+ * general.
+ */
+ std::vector<ServerLocation> getLocators() const;
+
+ /**
+ * Creates a connection to the given locator
+ * @param location Locator ServerLocation
+ * @return A connection wrapper for the locator
+ */
+ ConnectionWrapper createConnection(const ServerLocation& location) const;
+
+ /**
+ * Sends a request to the given locator
+ * @param location Locator ServerLocation
+ * @param request Serializable object of the request
+ * @return Serializable object of the response if no error happened,
+ * otherwise it returns a nullptr
+ */
+ std::shared_ptr<Serializable> sendRequest(
+ const ServerLocation& location,
+ const std::shared_ptr<Serializable>& request) const;
+
+ /**
+ * Data members
+ */
+ mutable boost::shared_mutex mutex_;
+ std::vector<ServerLocation> locators_;
const ThinClientPoolDM* m_poolDM;
- ThinClientLocatorHelper(const ThinClientLocatorHelper&);
- ThinClientLocatorHelper& operator=(const ThinClientLocatorHelper&);
std::string m_sniProxyHost;
int m_sniProxyPort;
};
diff --git a/cppcache/src/ThinClientPoolDM.cpp b/cppcache/src/ThinClientPoolDM.cpp
index 95c9867..a828a5b 100644
--- a/cppcache/src/ThinClientPoolDM.cpp
+++ b/cppcache/src/ThinClientPoolDM.cpp
@@ -568,7 +568,8 @@ std::string ThinClientPoolDM::selectEndpoint(
throw IllegalStateException("Locator query failed");
}
// Update Locator stats
- getStats().setLocators((m_locHelper)->getCurLocatorsNum());
+ getStats().setLocators(
+ static_cast<int32_t>(m_locHelper->getCurLocatorsNum()));
getStats().incLoctorResposes();
std::string epNameStr = outEndpoint.getServerName() + ":" +