Google Git
Sign in
apache / geode-native / 5d012199055a9a7657563727f6e26a406b287fc3 / . / cppcache / src / ThinClientRegion.cpp
blob: 7444d195c96d6d8e915247ee3d1081f10d282abf [file] [log] [blame]
/*
* 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 "ThinClientRegion.hpp"
#include <algorithm>
#include <limits>
#include <regex>
#include <geode/PoolManager.hpp>
#include <geode/Struct.hpp>
#include <geode/SystemProperties.hpp>
#include <geode/UserFunctionExecutionException.hpp>
#include "AutoDelete.hpp"
#include "CacheImpl.hpp"
#include "CacheRegionHelper.hpp"
#include "DataInputInternal.hpp"
#include "PutAllPartialResultServerException.hpp"
#include "ReadWriteLock.hpp"
#include "RegionGlobalLocks.hpp"
#include "RemoteQuery.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrDistributionManager.hpp"
#include "TcrEndpoint.hpp"
#include "ThinClientBaseDM.hpp"
#include "ThinClientPoolDM.hpp"
#include "UserAttributes.hpp"
#include "Utils.hpp"
#include "VersionedCacheableObjectPartList.hpp"
#include "util/bounds.hpp"
#include "util/exception.hpp"
namespace apache {
namespace geode {
namespace client {
static const std::regex PREDICATE_IS_FULL_QUERY_REGEX(
"^\\s*(?:select|import)\\b", std::regex::icase);
void setThreadLocalExceptionMessage(const char* exMsg);
class PutAllWork : public PooledWork<GfErrType>,
private NonCopyable,
private NonAssignable {
ThinClientPoolDM* m_poolDM;
std::shared_ptr<BucketServerLocation> m_serverLocation;
TcrMessage* m_request;
TcrMessageReply* m_reply;
MapOfUpdateCounters m_mapOfUpdateCounters;
bool m_attemptFailover;
bool m_isBGThread;
std::shared_ptr<UserAttributes> m_userAttribute;
const std::shared_ptr<Region> m_region;
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> m_keys;
std::shared_ptr<HashMapOfCacheable> m_map;
std::shared_ptr<VersionedCacheableObjectPartList> m_verObjPartListPtr;
std::chrono::milliseconds m_timeout;
std::shared_ptr<PutAllPartialResultServerException> m_papException;
bool m_isPapeReceived;
ChunkedPutAllResponse* m_resultCollector;
// UNUSED const std::shared_ptr<Serializable>& m_aCallbackArgument;
public:
PutAllWork(
ThinClientPoolDM* poolDM,
const std::shared_ptr<BucketServerLocation>& serverLocation,
const std::shared_ptr<Region>& region, bool attemptFailover,
bool isBGThread, const std::shared_ptr<HashMapOfCacheable> map,
const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> keys,
std::chrono::milliseconds timeout,
const std::shared_ptr<Serializable>& aCallbackArgument)
: m_poolDM(poolDM),
m_serverLocation(serverLocation),
m_attemptFailover(attemptFailover),
m_isBGThread(isBGThread),
m_userAttribute(nullptr),
m_region(region),
m_keys(keys),
m_map(map),
m_timeout(timeout),
m_papException(nullptr),
m_isPapeReceived(false)
// UNUSED , m_aCallbackArgument(aCallbackArgument)
{
m_request = new TcrMessagePutAll(
new DataOutput(m_region->getCache().createDataOutput()), m_region.get(),
*m_map, m_timeout, m_poolDM, aCallbackArgument);
m_reply = new TcrMessageReply(true, m_poolDM);
// create new instanceof VCOPL
std::recursive_mutex responseLock;
m_verObjPartListPtr =
std::make_shared<VersionedCacheableObjectPartList>(keys, responseLock);
if (m_poolDM->isMultiUserMode()) {
m_userAttribute = UserAttributes::threadLocalUserAttributes;
}
m_request->setTimeout(m_timeout);
m_reply->setTimeout(m_timeout);
m_resultCollector = new ChunkedPutAllResponse(
m_region, *m_reply, responseLock, m_verObjPartListPtr);
m_reply->setChunkedResultHandler(m_resultCollector);
}
~PutAllWork() {
delete m_request;
delete m_reply;
delete m_resultCollector;
}
TcrMessage* getReply() { return m_reply; }
std::shared_ptr<HashMapOfCacheable> getPutAllMap() { return m_map; }
std::shared_ptr<VersionedCacheableObjectPartList> getVerObjPartList() {
return m_verObjPartListPtr;
}
ChunkedPutAllResponse* getResultCollector() { return m_resultCollector; }
std::shared_ptr<BucketServerLocation> getServerLocation() {
return m_serverLocation;
}
std::shared_ptr<PutAllPartialResultServerException> getPaPResultException() {
return m_papException;
}
void init() {}
GfErrType execute(void) {
GuardUserAttributes gua;
if (m_userAttribute != nullptr) {
gua.setAuthenticatedView(m_userAttribute->getAuthenticatedView());
}
GfErrType err = GF_NOERR;
err = m_poolDM->sendSyncRequest(*m_request, *m_reply, m_attemptFailover,
m_isBGThread, m_serverLocation);
// Set Version Tags
LOGDEBUG(" m_verObjPartListPtr size = %d err = %d ",
m_resultCollector->getList()->size(), err);
m_verObjPartListPtr->setVersionedTagptr(
m_resultCollector->getList()->getVersionedTagptr());
if (err != GF_NOERR) {
return err;
} /*This can be GF_NOTCON, counterpart to java
ServerConnectivityException*/
switch (m_reply->getMessageType()) {
case TcrMessage::REPLY:
break;
case TcrMessage::RESPONSE:
LOGDEBUG("PutAllwork execute err = %d ", err);
break;
case TcrMessage::EXCEPTION:
// TODO::Check for the PAPException and READ
// PutAllPartialResultServerException and set its member for later use.
// set m_papException and m_isPapeReceived
m_isPapeReceived = true;
if (m_poolDM->isNotAuthorizedException(m_reply->getException())) {
LOGDEBUG("received NotAuthorizedException");
err = GF_AUTHENTICATION_FAILED_EXCEPTION;
} else if (m_poolDM->isPutAllPartialResultException(
m_reply->getException())) {
LOGDEBUG("received PutAllPartialResultException");
err = GF_PUTALL_PARTIAL_RESULT_EXCEPTION;
} else {
LOGDEBUG("received unknown exception:%s", m_reply->getException());
err = GF_PUTALL_PARTIAL_RESULT_EXCEPTION;
// TODO should assign a new err code
}
break;
case TcrMessage::PUT_DATA_ERROR:
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type %d during region put-all",
m_reply->getMessageType());
err = GF_NOTOBJ;
break;
}
return err;
}
};
class RemoveAllWork : public PooledWork<GfErrType>,
private NonCopyable,
private NonAssignable {
ThinClientPoolDM* m_poolDM;
std::shared_ptr<BucketServerLocation> m_serverLocation;
TcrMessage* m_request;
TcrMessageReply* m_reply;
MapOfUpdateCounters m_mapOfUpdateCounters;
bool m_attemptFailover;
bool m_isBGThread;
std::shared_ptr<UserAttributes> m_userAttribute;
const std::shared_ptr<Region> m_region;
const std::shared_ptr<Serializable>& m_aCallbackArgument;
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> m_keys;
std::shared_ptr<VersionedCacheableObjectPartList> m_verObjPartListPtr;
std::shared_ptr<PutAllPartialResultServerException> m_papException;
bool m_isPapeReceived;
ChunkedRemoveAllResponse* m_resultCollector;
public:
RemoveAllWork(
ThinClientPoolDM* poolDM,
const std::shared_ptr<BucketServerLocation>& serverLocation,
const std::shared_ptr<Region>& region, bool attemptFailover,
bool isBGThread,
const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> keys,
const std::shared_ptr<Serializable>& aCallbackArgument)
: m_poolDM(poolDM),
m_serverLocation(serverLocation),
m_attemptFailover(attemptFailover),
m_isBGThread(isBGThread),
m_userAttribute(nullptr),
m_region(region),
m_aCallbackArgument(aCallbackArgument),
m_keys(keys),
m_papException(nullptr),
m_isPapeReceived(false) {
m_request = new TcrMessageRemoveAll(
new DataOutput(m_region->getCache().createDataOutput()), m_region.get(),
*keys, m_aCallbackArgument, m_poolDM);
m_reply = new TcrMessageReply(true, m_poolDM);
// create new instanceof VCOPL
std::recursive_mutex responseLock;
m_verObjPartListPtr =
std::make_shared<VersionedCacheableObjectPartList>(keys, responseLock);
if (m_poolDM->isMultiUserMode()) {
m_userAttribute = UserAttributes::threadLocalUserAttributes;
}
m_resultCollector = new ChunkedRemoveAllResponse(
m_region, *m_reply, responseLock, m_verObjPartListPtr);
m_reply->setChunkedResultHandler(m_resultCollector);
}
~RemoveAllWork() {
delete m_request;
delete m_reply;
delete m_resultCollector;
}
TcrMessage* getReply() { return m_reply; }
std::shared_ptr<VersionedCacheableObjectPartList> getVerObjPartList() {
return m_verObjPartListPtr;
}
ChunkedRemoveAllResponse* getResultCollector() { return m_resultCollector; }
std::shared_ptr<BucketServerLocation> getServerLocation() {
return m_serverLocation;
}
std::shared_ptr<PutAllPartialResultServerException> getPaPResultException() {
return m_papException;
}
void init() {}
GfErrType execute(void) {
GuardUserAttributes gua;
if (m_userAttribute != nullptr) {
gua.setAuthenticatedView(m_userAttribute->getAuthenticatedView());
}
GfErrType err = GF_NOERR;
err = m_poolDM->sendSyncRequest(*m_request, *m_reply, m_attemptFailover,
m_isBGThread, m_serverLocation);
// Set Version Tags
LOGDEBUG(" m_verObjPartListPtr size = %d err = %d ",
m_resultCollector->getList()->size(), err);
m_verObjPartListPtr->setVersionedTagptr(
m_resultCollector->getList()->getVersionedTagptr());
if (err != GF_NOERR) {
return err;
} /*This can be GF_NOTCON, counterpart to java
ServerConnectivityException*/
switch (m_reply->getMessageType()) {
case TcrMessage::REPLY:
break;
case TcrMessage::RESPONSE:
LOGDEBUG("RemoveAllWork execute err = %d ", err);
break;
case TcrMessage::EXCEPTION:
// TODO::Check for the PAPException and READ
// PutAllPartialResultServerException and set its member for later use.
// set m_papException and m_isPapeReceived
m_isPapeReceived = true;
if (m_poolDM->isNotAuthorizedException(m_reply->getException())) {
LOGDEBUG("received NotAuthorizedException");
err = GF_AUTHENTICATION_FAILED_EXCEPTION;
} else if (m_poolDM->isPutAllPartialResultException(
m_reply->getException())) {
LOGDEBUG("received PutAllPartialResultException");
err = GF_PUTALL_PARTIAL_RESULT_EXCEPTION;
} else {
LOGDEBUG("received unknown exception:%s", m_reply->getException());
err = GF_PUTALL_PARTIAL_RESULT_EXCEPTION;
// TODO should assign a new err code
}
break;
case TcrMessage::PUT_DATA_ERROR:
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type %d during region remove-all",
m_reply->getMessageType());
err = GF_NOTOBJ;
break;
}
return err;
}
};
ThinClientRegion::ThinClientRegion(
const std::string& name, CacheImpl* cacheImpl,
const std::shared_ptr<RegionInternal>& rPtr, RegionAttributes attributes,
const std::shared_ptr<CacheStatistics>& stats, bool shared)
: LocalRegion(name, cacheImpl, rPtr, attributes, stats, shared),
m_tcrdm(nullptr),
m_notifyRelease(false),
m_isMetaDataRefreshed(false) {
m_transactionEnabled = true;
m_isDurableClnt = !cacheImpl->getDistributedSystem()
.getSystemProperties()
.durableClientId()
.empty();
}
void ThinClientRegion::initTCR() {
try {
m_tcrdm = std::make_shared<TcrDistributionManager>(
this, m_cacheImpl->tcrConnectionManager());
m_tcrdm->init();
} catch (const Exception& ex) {
LOGERROR("Exception while initializing region: %s: %s",
ex.getName().c_str(), ex.what());
throw;
}
}
void ThinClientRegion::registerKeys(
const std::vector<std::shared_ptr<CacheableKey>>& keys, bool isDurable,
bool getInitialValues, bool receiveValues) {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGERROR(
"Registering keys is supported "
"only if subscription-enabled attribute is true for pool " +
pool->getName());
throw UnsupportedOperationException(
"Registering keys is supported "
"only if pool subscription-enabled attribute is true.");
}
}
if (keys.empty()) {
LOGERROR("Register keys list is empty");
throw IllegalArgumentException(
"Register keys "
"keys vector is empty");
}
if (isDurable && !isDurableClient()) {
LOGERROR(
"Register keys durable flag is only applicable for durable clients");
throw IllegalStateException(
"Durable flag only applicable for "
"durable clients");
}
if (getInitialValues && !m_regionAttributes.getCachingEnabled()) {
LOGERROR(
"Register keys getInitialValues flag is only applicable for caching"
"clients");
throw IllegalStateException(
"getInitialValues flag only applicable for caching clients");
}
InterestResultPolicy interestPolicy = InterestResultPolicy::NONE;
if (getInitialValues) {
interestPolicy = InterestResultPolicy::KEYS_VALUES;
}
LOGDEBUG("ThinClientRegion::registerKeys : interestpolicy is %d",
interestPolicy.ordinal);
GfErrType err = registerKeysNoThrow(keys, true, nullptr, isDurable,
interestPolicy, receiveValues);
if (m_tcrdm->isFatalError(err)) {
throwExceptionIfError("Region::registerKeys", err);
}
throwExceptionIfError("Region::registerKeys", err);
}
void ThinClientRegion::unregisterKeys(
const std::vector<std::shared_ptr<CacheableKey>>& keys) {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGERROR(
"Unregister keys is supported "
"only if subscription-enabled attribute is true for pool " +
pool->getName());
throw UnsupportedOperationException(
"Unregister keys is supported "
"only if pool subscription-enabled attribute is true.");
}
} else {
if (!getAttributes().getClientNotificationEnabled()) {
LOGERROR(
"Unregister keys is supported "
"only if region client-notification attribute is true.");
throw UnsupportedOperationException(
"Unregister keys is supported "
"only if region client-notification attribute is true.");
}
}
if (keys.empty()) {
LOGERROR("Unregister keys list is empty");
throw IllegalArgumentException(
"Unregister keys "
"keys vector is empty");
}
GfErrType err = unregisterKeysNoThrow(keys);
throwExceptionIfError("Region::unregisterKeys", err);
}
void ThinClientRegion::registerAllKeys(bool isDurable, bool getInitialValues,
bool receiveValues) {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGERROR(
"Register all keys is supported only "
"if subscription-enabled attribute is true for pool " +
pool->getName());
throw UnsupportedOperationException(
"Register all keys is supported only "
"if pool subscription-enabled attribute is true.");
}
}
if (isDurable && !isDurableClient()) {
LOGERROR(
"Register all keys durable flag is only applicable for durable "
"clients");
throw IllegalStateException(
"Durable flag only applicable for durable clients");
}
if (getInitialValues && !m_regionAttributes.getCachingEnabled()) {
LOGERROR(
"Register all keys getInitialValues flag is only applicable for caching"
"clients");
throw IllegalStateException(
"getInitialValues flag only applicable for caching clients");
}
InterestResultPolicy interestPolicy = InterestResultPolicy::NONE;
if (getInitialValues) {
interestPolicy = InterestResultPolicy::KEYS_VALUES;
} else {
interestPolicy = InterestResultPolicy::KEYS;
}
LOGDEBUG("ThinClientRegion::registerAllKeys : interestpolicy is %d",
interestPolicy.ordinal);
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> resultKeys;
// if we need to fetch initial data, then we get the keys in
// that call itself using the special GET_ALL message and do not need
// to get the keys in the initial register interest call
GfErrType err =
registerRegexNoThrow(".*", true, nullptr, isDurable, resultKeys,
interestPolicy, receiveValues);
if (m_tcrdm->isFatalError(err)) {
throwExceptionIfError("Region::registerAllKeys", err);
}
// Get the entries from the server using a special GET_ALL message
throwExceptionIfError("Region::registerAllKeys", err);
}
void ThinClientRegion::registerRegex(const std::string& regex, bool isDurable,
bool getInitialValues,
bool receiveValues) {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGERROR(
"Register regex is supported only if "
"subscription-enabled attribute is true for pool " +
pool->getName());
throw UnsupportedOperationException(
"Register regex is supported only if "
"pool subscription-enabled attribute is true.");
}
}
if (isDurable && !isDurableClient()) {
LOGERROR("Register regex durable flag only applicable for durable clients");
throw IllegalStateException(
"Durable flag only applicable for durable clients");
}
if (regex.empty()) {
throw IllegalArgumentException(
"Region::registerRegex: Regex string is empty");
}
auto interestPolicy = InterestResultPolicy::NONE;
if (getInitialValues) {
interestPolicy = InterestResultPolicy::KEYS_VALUES;
} else {
interestPolicy = InterestResultPolicy::KEYS;
}
LOGDEBUG("ThinClientRegion::registerRegex : interestpolicy is %d",
interestPolicy.ordinal);
auto resultKeys2 =
std::make_shared<std::vector<std::shared_ptr<CacheableKey>>>();
// if we need to fetch initial data for "allKeys" case, then we
// get the keys in that call itself using the special GET_ALL message and
// do not need to get the keys in the initial register interest call
GfErrType err =
registerRegexNoThrow(regex, true, nullptr, isDurable, resultKeys2,
interestPolicy, receiveValues);
if (m_tcrdm->isFatalError(err)) {
throwExceptionIfError("Region::registerRegex", err);
}
throwExceptionIfError("Region::registerRegex", err);
}
void ThinClientRegion::unregisterRegex(const std::string& regex) {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGERROR(
"Unregister regex is supported only if "
"subscription-enabled attribute is true for pool " +
pool->getName());
throw UnsupportedOperationException(
"Unregister regex is supported only if "
"pool subscription-enabled attribute is true.");
}
}
if (regex.empty()) {
LOGERROR("Unregister regex string is empty");
throw IllegalArgumentException("Unregister regex string is empty");
}
GfErrType err = unregisterRegexNoThrow(regex);
throwExceptionIfError("Region::unregisterRegex", err);
}
void ThinClientRegion::unregisterAllKeys() {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGERROR(
"Unregister all keys is supported only if "
"subscription-enabled attribute is true for pool " +
pool->getName());
throw UnsupportedOperationException(
"Unregister all keys is supported only if "
"pool subscription-enabled attribute is true.");
}
}
GfErrType err = unregisterRegexNoThrow(".*");
throwExceptionIfError("Region::unregisterAllKeys", err);
}
std::shared_ptr<SelectResults> ThinClientRegion::query(
const std::string& predicate, std::chrono::milliseconds timeout) {
util::PROTOCOL_OPERATION_TIMEOUT_BOUNDS(timeout);
CHECK_DESTROY_PENDING(TryReadGuard, Region::query);
if (predicate.empty()) {
LOGERROR("Region query predicate string is empty");
throw IllegalArgumentException("Region query predicate string is empty");
}
std::string squery;
if (std::regex_search(predicate, PREDICATE_IS_FULL_QUERY_REGEX)) {
squery = predicate;
} else {
squery = "select distinct * from ";
squery += getFullPath();
squery += " this where ";
squery += predicate;
}
std::shared_ptr<RemoteQuery> queryPtr;
if (auto poolDM = std::dynamic_pointer_cast<ThinClientPoolDM>(m_tcrdm)) {
queryPtr = std::dynamic_pointer_cast<RemoteQuery>(
poolDM->getQueryServiceWithoutCheck()->newQuery(squery.c_str()));
} else {
queryPtr = std::dynamic_pointer_cast<RemoteQuery>(
m_cacheImpl->getQueryService()->newQuery(squery.c_str()));
}
return queryPtr->execute(timeout, "Region::query", m_tcrdm.get(), nullptr);
}
bool ThinClientRegion::existsValue(const std::string& predicate,
std::chrono::milliseconds timeout) {
util::PROTOCOL_OPERATION_TIMEOUT_BOUNDS(timeout);
auto results = query(predicate, timeout);
if (results == nullptr) {
return false;
}
return results->size() > 0;
}
GfErrType ThinClientRegion::unregisterKeysBeforeDestroyRegion() {
auto pool = m_cacheImpl->getPoolManager().find(getAttributes().getPoolName());
if (pool != nullptr) {
if (!pool->getSubscriptionEnabled()) {
LOGDEBUG(
"pool subscription-enabled attribute is false, No need to Unregister "
"keys");
return GF_NOERR;
}
}
GfErrType err = GF_NOERR;
GfErrType opErr = GF_NOERR;
opErr = unregisterStoredRegexLocalDestroy(m_interestListRegex);
err = opErr != GF_NOERR ? opErr : err;
opErr = unregisterStoredRegexLocalDestroy(m_durableInterestListRegex);
err = opErr != GF_NOERR ? opErr : err;
opErr = unregisterStoredRegexLocalDestroy(
m_interestListRegexForUpdatesAsInvalidates);
err = opErr != GF_NOERR ? opErr : err;
opErr = unregisterStoredRegexLocalDestroy(
m_durableInterestListRegexForUpdatesAsInvalidates);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVec;
copyInterestList(keysVec, m_interestList);
opErr = unregisterKeysNoThrowLocalDestroy(keysVec, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVecDurable;
copyInterestList(keysVecDurable, m_durableInterestList);
opErr = unregisterKeysNoThrowLocalDestroy(keysVecDurable, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVecForUpdatesAsInvalidates;
copyInterestList(keysVecForUpdatesAsInvalidates,
m_interestListForUpdatesAsInvalidates);
opErr =
unregisterKeysNoThrowLocalDestroy(keysVecForUpdatesAsInvalidates, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>>
keysVecDurableForUpdatesAsInvalidates;
copyInterestList(keysVecDurableForUpdatesAsInvalidates,
m_durableInterestListForUpdatesAsInvalidates);
opErr = unregisterKeysNoThrowLocalDestroy(
keysVecDurableForUpdatesAsInvalidates, false);
err = opErr != GF_NOERR ? opErr : err;
return err;
}
std::shared_ptr<Serializable> ThinClientRegion::selectValue(
const std::string& predicate, std::chrono::milliseconds timeout) {
auto results = query(predicate, timeout);
if (results == nullptr || results->size() == 0) {
return nullptr;
}
if (results->size() > 1) {
throw QueryException("selectValue has more than one result");
}
return results->operator[](0);
}
std::vector<std::shared_ptr<CacheableKey>> ThinClientRegion::serverKeys() {
CHECK_DESTROY_PENDING(TryReadGuard, Region::serverKeys);
TcrMessageReply reply(true, m_tcrdm.get());
TcrMessageKeySet request(new DataOutput(m_cacheImpl->createDataOutput()),
m_fullPath, m_tcrdm.get());
reply.setMessageTypeRequest(TcrMessage::KEY_SET);
std::vector<std::shared_ptr<CacheableKey>> serverKeys;
ChunkedKeySetResponse resultCollector(request, serverKeys, reply);
reply.setChunkedResultHandler(&resultCollector);
GfErrType err = GF_NOERR;
err = m_tcrdm->sendSyncRequest(request, reply);
throwExceptionIfError("Region::serverKeys", err);
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE: {
// keyset result is handled by ChunkedKeySetResponse
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region:serverKeys", reply.getException());
break;
}
case TcrMessage::KEY_SET_DATA_ERROR: {
LOGERROR("Region serverKeys: an error occurred on the server");
err = GF_CACHESERVER_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d during region serverKeys",
reply.getMessageType());
err = GF_MSG;
break;
}
}
throwExceptionIfError("Region::serverKeys", err);
return serverKeys;
}
bool ThinClientRegion::containsKeyOnServer(
const std::shared_ptr<CacheableKey>& keyPtr) const {
GfErrType err = GF_NOERR;
bool ret = false;
/** @brief Create message and send to bridge server */
TcrMessageContainsKey request(
new DataOutput(m_cacheImpl->createDataOutput()), this, keyPtr,
static_cast<std::shared_ptr<Serializable>>(nullptr), true, m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
reply.setMessageTypeRequest(TcrMessage::CONTAINS_KEY);
err = m_tcrdm->sendSyncRequest(request, reply);
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE:
ret = reply.getBoolValue();
break;
case TcrMessage::EXCEPTION:
err = handleServerException("Region::containsKeyOnServer:",
reply.getException());
break;
case TcrMessage::REQUEST_DATA_ERROR:
LOGERROR(
"Region::containsKeyOnServer: read error occurred on the endpoint %s",
m_tcrdm->getActiveEndpoint()->name().c_str());
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type in Region::containsKeyOnServer %d",
reply.getMessageType());
err = GF_MSG;
break;
}
auto rptr = CacheableBoolean::create(ret);
rptr = std::dynamic_pointer_cast<CacheableBoolean>(handleReplay(err, rptr));
throwExceptionIfError("Region::containsKeyOnServer ", err);
return rptr->value();
}
bool ThinClientRegion::containsValueForKey_remote(
const std::shared_ptr<CacheableKey>& keyPtr) const {
GfErrType err = GF_NOERR;
bool ret = false;
/** @brief Create message and send to bridge server */
TcrMessageContainsKey request(
new DataOutput(m_cacheImpl->createDataOutput()), this, keyPtr,
static_cast<std::shared_ptr<Serializable>>(nullptr), false,
m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
reply.setMessageTypeRequest(TcrMessage::CONTAINS_KEY);
err = m_tcrdm->sendSyncRequest(request, reply);
// if ( err != GF_NOERR ) return ret;
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE:
ret = reply.getBoolValue();
break;
case TcrMessage::EXCEPTION:
err = handleServerException("Region::containsValueForKey:",
reply.getException());
break;
case TcrMessage::REQUEST_DATA_ERROR:
LOGERROR(
"Region::containsValueForKey: read error occurred on the endpoint %s",
m_tcrdm->getActiveEndpoint()->name().c_str());
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type in Region::containsValueForKey %d",
reply.getMessageType());
err = GF_MSG;
break;
}
auto rptr = CacheableBoolean::create(ret);
rptr = std::dynamic_pointer_cast<CacheableBoolean>(handleReplay(err, rptr));
throwExceptionIfError("Region::containsValueForKey ", err);
return rptr->value();
}
void ThinClientRegion::clear(
const std::shared_ptr<Serializable>& aCallbackArgument) {
GfErrType err = GF_NOERR;
err = localClearNoThrow(aCallbackArgument, CacheEventFlags::NORMAL);
if (err != GF_NOERR) throwExceptionIfError("Region::clear", err);
/** @brief Create message and send to bridge server */
TcrMessageClearRegion request(new DataOutput(m_cacheImpl->createDataOutput()),
this, aCallbackArgument,
std::chrono::milliseconds(-1), m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) throwExceptionIfError("Region::clear", err);
switch (reply.getMessageType()) {
case TcrMessage::REPLY:
LOGFINE("Region %s clear message sent to server successfully",
m_fullPath.c_str());
break;
case TcrMessage::EXCEPTION:
err = handleServerException("Region::clear:", reply.getException());
break;
case TcrMessage::CLEAR_REGION_DATA_ERROR:
LOGERROR("Region clear read error occurred on the endpoint %s",
m_tcrdm->getActiveEndpoint()->name().c_str());
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type %d during region clear",
reply.getMessageType());
err = GF_MSG;
break;
}
if (err == GF_NOERR) {
err = invokeCacheListenerForRegionEvent(
aCallbackArgument, CacheEventFlags::NORMAL, AFTER_REGION_CLEAR);
}
throwExceptionIfError("Region::clear", err);
}
GfErrType ThinClientRegion::getNoThrow_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
std::shared_ptr<Cacheable>& valPtr,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag) {
GfErrType err = GF_NOERR;
/** @brief Create message and send to bridge server */
TcrMessageRequest request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keyPtr, aCallbackArgument, m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) return err;
// put the object into local region
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE: {
valPtr = reply.getValue();
versionTag = reply.getVersionTag();
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::get", reply.getException());
break;
}
case TcrMessage::REQUEST_DATA_ERROR: {
// LOGERROR("A read error occurred on the endpoint %s",
// m_tcrdm->getActiveEndpoint()->name().c_str());
err = GF_CACHESERVER_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d while getting entry from region",
reply.getMessageType());
err = GF_MSG;
break;
}
}
return err;
}
GfErrType ThinClientRegion::invalidateNoThrow_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag) {
GfErrType err = GF_NOERR;
TcrMessageInvalidate request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keyPtr, aCallbackArgument, m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) return err;
// put the object into local region
switch (reply.getMessageType()) {
case TcrMessage::REPLY: {
versionTag = reply.getVersionTag();
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::get", reply.getException());
break;
}
case TcrMessage::INVALIDATE_ERROR: {
err = GF_CACHESERVER_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d while getting entry from region",
reply.getMessageType());
err = GF_MSG;
break;
}
}
return err;
}
GfErrType ThinClientRegion::putNoThrow_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
const std::shared_ptr<Cacheable>& valuePtr,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag, bool checkDelta) {
GfErrType err = GF_NOERR;
// do TCR put
// bool delta = valuePtr->hasDelta();
bool delta = false;
auto&& conFlationValue = getCacheImpl()
->getDistributedSystem()
.getSystemProperties()
.conflateEvents();
if (checkDelta && valuePtr && conFlationValue != "true" &&
ThinClientBaseDM::isDeltaEnabledOnServer()) {
auto&& temp = std::dynamic_pointer_cast<Delta>(valuePtr);
delta = temp && temp->hasDelta();
}
TcrMessagePut request(new DataOutput(m_cacheImpl->createDataOutput()), this,
keyPtr, valuePtr, aCallbackArgument, delta,
m_tcrdm.get());
TcrMessageReply* reply = new TcrMessageReply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, *reply);
if (delta) {
// Does not check whether success of failure..
m_cacheImpl->getCachePerfStats().incDeltaPut();
if (reply->getMessageType() == TcrMessage::PUT_DELTA_ERROR) {
// Try without delta
TcrMessagePut request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keyPtr, valuePtr, aCallbackArgument, false,
m_tcrdm.get(), false, true);
delete reply;
reply = new TcrMessageReply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, *reply);
}
}
if (err != GF_NOERR) return err;
// put the object into local region
switch (reply->getMessageType()) {
case TcrMessage::REPLY: {
versionTag = reply->getVersionTag();
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::put", reply->getException());
break;
}
case TcrMessage::PUT_DATA_ERROR: {
err = GF_CACHESERVER_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d during region put reply",
reply->getMessageType());
err = GF_MSG;
}
}
delete reply;
reply = nullptr;
return err;
}
GfErrType ThinClientRegion::createNoThrow_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
const std::shared_ptr<Cacheable>& valuePtr,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag) {
return putNoThrow_remote(keyPtr, valuePtr, aCallbackArgument, versionTag,
false);
}
GfErrType ThinClientRegion::destroyNoThrow_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag) {
GfErrType err = GF_NOERR;
// do TCR destroy
TcrMessageDestroy request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keyPtr, nullptr, aCallbackArgument,
m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) return err;
switch (reply.getMessageType()) {
case TcrMessage::REPLY: {
if (reply.getEntryNotFound() == 1) {
err = GF_CACHE_ENTRY_NOT_FOUND;
} else {
LOGDEBUG("Remote key [%s] is destroyed successfully in region %s",
Utils::nullSafeToString(keyPtr).c_str(), m_fullPath.c_str());
}
versionTag = reply.getVersionTag();
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::destroy", reply.getException());
break;
}
case TcrMessage::DESTROY_DATA_ERROR: {
err = GF_CACHE_ENTRY_DESTROYED_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d while destroying region entry",
reply.getMessageType());
err = GF_MSG;
}
}
return err;
} // destroyNoThrow_remote()
GfErrType ThinClientRegion::removeNoThrow_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
const std::shared_ptr<Cacheable>& cvalue,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag) {
GfErrType err = GF_NOERR;
// do TCR remove
TcrMessageDestroy request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keyPtr, cvalue, aCallbackArgument,
m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) {
return err;
}
switch (reply.getMessageType()) {
case TcrMessage::REPLY: {
if (reply.getEntryNotFound() == 1) {
err = GF_ENOENT;
} else {
LOGDEBUG("Remote key [%s] is removed successfully in region %s",
Utils::nullSafeToString(keyPtr).c_str(), m_fullPath.c_str());
}
versionTag = reply.getVersionTag();
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::remove", reply.getException());
break;
}
case TcrMessage::DESTROY_DATA_ERROR: {
err = GF_CACHE_ENTRY_DESTROYED_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d while removing region entry",
reply.getMessageType());
err = GF_MSG;
}
}
return err;
}
GfErrType ThinClientRegion::removeNoThrowEX_remote(
const std::shared_ptr<CacheableKey>& keyPtr,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag>& versionTag) {
GfErrType err = GF_NOERR;
// do TCR remove
TcrMessageDestroy request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keyPtr, nullptr, aCallbackArgument,
m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) {
return err;
}
switch (reply.getMessageType()) {
case TcrMessage::REPLY: {
versionTag = reply.getVersionTag();
if (reply.getEntryNotFound() == 1) {
err = GF_ENOENT;
} else {
LOGDEBUG("Remote key [%s] is removed successfully in region %s",
Utils::nullSafeToString(keyPtr).c_str(), m_fullPath.c_str());
}
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::removeEx", reply.getException());
break;
}
case TcrMessage::DESTROY_DATA_ERROR: {
err = GF_CACHE_ENTRY_DESTROYED_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d while removing region entry",
reply.getMessageType());
err = GF_MSG;
}
}
return err;
}
GfErrType ThinClientRegion::getAllNoThrow_remote(
const std::vector<std::shared_ptr<CacheableKey>>* keys,
const std::shared_ptr<HashMapOfCacheable>& values,
const std::shared_ptr<HashMapOfException>& exceptions,
const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>>&
resultKeys,
bool addToLocalCache,
const std::shared_ptr<Serializable>& aCallbackArgument) {
GfErrType err = GF_NOERR;
MapOfUpdateCounters updateCountMap;
int32_t destroyTracker = 0;
addToLocalCache = addToLocalCache && m_regionAttributes.getCachingEnabled();
if (addToLocalCache && !m_regionAttributes.getConcurrencyChecksEnabled()) {
// start tracking the entries
if (keys == nullptr) {
// track all entries with destroy tracking for non-existent entries
destroyTracker = m_entries->addTrackerForAllEntries(updateCountMap, true);
} else {
for (const auto& key : *keys) {
std::shared_ptr<Cacheable> oldValue;
int updateCount =
m_entries->addTrackerForEntry(key, oldValue, true, false, false);
updateCountMap.emplace(key, updateCount);
}
}
}
// create the GET_ALL request
TcrMessageGetAll request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keys, m_tcrdm.get(), aCallbackArgument);
TcrMessageReply reply(true, m_tcrdm.get());
std::recursive_mutex responseLock;
// need to check
TcrChunkedResult* resultCollector(new ChunkedGetAllResponse(
reply, this, keys, values, exceptions, resultKeys, updateCountMap,
destroyTracker, addToLocalCache, responseLock));
reply.setChunkedResultHandler(resultCollector);
err = m_tcrdm->sendSyncRequest(request, reply);
if (addToLocalCache && !m_regionAttributes.getConcurrencyChecksEnabled()) {
// remove the tracking for remaining keys in case some keys do not have
// values from server in GII
for (MapOfUpdateCounters::const_iterator iter = updateCountMap.begin();
iter != updateCountMap.end(); ++iter) {
if (iter->second >= 0) {
m_entries->removeTrackerForEntry(iter->first);
}
}
// remove tracking for destroys
if (destroyTracker > 0) {
m_entries->removeDestroyTracking();
}
}
delete resultCollector;
if (err != GF_NOERR) {
return err;
}
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE: {
// nothing to be done; put in local region, if required,
// is handled by ChunkedGetAllResponse
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region:getAll", reply.getException());
break;
}
case TcrMessage::GET_ALL_DATA_ERROR: {
LOGERROR("Region get-all: a read error occurred on the endpoint %s",
m_tcrdm->getActiveEndpoint()->name().c_str());
err = GF_CACHESERVER_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d during region get-all",
reply.getMessageType());
err = GF_MSG;
break;
}
}
return err;
}
GfErrType ThinClientRegion::singleHopPutAllNoThrow_remote(
ThinClientPoolDM* tcrdm, const HashMapOfCacheable& map,
std::shared_ptr<VersionedCacheableObjectPartList>& versionedObjPartList,
std::chrono::milliseconds timeout,
const std::shared_ptr<Serializable>& aCallbackArgument) {
LOGDEBUG(" ThinClientRegion::singleHopPutAllNoThrow_remote map size = %zu",
map.size());
auto region = shared_from_this();
auto error = GF_NOERR;
/*Step-1::
* populate the keys vector from the user Map and pass it to the
* getServerToFilterMap to generate locationMap
* If locationMap is nullptr try the old, existing putAll impl that may take
* multiple n/w hops
*/
auto userKeys = std::vector<std::shared_ptr<CacheableKey>>();
for (const auto& iter : map) {
userKeys.push_back(iter.first);
}
// last param in getServerToFilterMap() is false for putAll
// LOGDEBUG("ThinClientRegion::singleHopPutAllNoThrow_remote keys.size() = %d
// ", userKeys->size());
auto locationMap = tcrdm->getClientMetaDataService()->getServerToFilterMap(
userKeys, region, true);
if (!locationMap) {
// putAll with multiple hop implementation
LOGDEBUG("locationMap is Null or Empty");
return multiHopPutAllNoThrow_remote(map, versionedObjPartList, timeout,
aCallbackArgument);
}
// set this flag that indicates putAll on PR is invoked with singlehop
// enabled.
m_isPRSingleHopEnabled = true;
// LOGDEBUG("locationMap.size() = %d ", locationMap->size());
/*Step-2
* a. create vector of PutAllWork
* b. locationMap<std::shared_ptr<BucketServerLocation>,
* std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>> >>. Create
* server specific filteredMap/subMap by populating all keys
* (locationIter.second()) and its corr. values from the user Map.
* c. create new instance of PutAllWork, i.e worker with required params.
* //TODO:: Add details of each parameter later
* d. enqueue the worker for thread from threadPool to perform/run execute
* method.
* e. insert the worker into the vector.
*/
std::vector<std::shared_ptr<PutAllWork>> putAllWorkers;
auto& threadPool = m_cacheImpl->getThreadPool();
int locationMapIndex = 0;
for (const auto& locationIter : *locationMap) {
const auto& serverLocation = locationIter.first;
if (serverLocation == nullptr) {
LOGDEBUG("serverLocation is nullptr");
}
const auto& keys = locationIter.second;
// Create server specific Sub-Map by iterating over keys.
auto filteredMap = std::make_shared<HashMapOfCacheable>();
if (keys != nullptr && keys->size() > 0) {
for (const auto& key : *keys) {
const auto& iter = map.find(key);
if (iter != map.end()) {
filteredMap->emplace(iter->first, iter->second);
}
}
}
auto worker = std::make_shared<PutAllWork>(
tcrdm, serverLocation, region, true /*attemptFailover*/,
false /*isBGThread*/, filteredMap, keys, timeout, aCallbackArgument);
threadPool.perform(worker);
putAllWorkers.push_back(worker);
locationMapIndex++;
}
// TODO::CHECK, do we need to set following ..??
// reply.setMessageType(TcrMessage::RESPONSE);
int cnt = 1;
/**
* Step::3
* a. Iterate over all vector of putAllWorkers and populate worker specific
* information into the HashMap
* resultMap<std::shared_ptr<BucketServerLocation>,
* std::shared_ptr<Serializable>>, 2nd part, Value can be a
* std::shared_ptr<VersionedCacheableObjectPartList> or
* std::shared_ptr<PutAllPartialResultServerException>.
* failedServers<std::shared_ptr<BucketServerLocation>,
* std::shared_ptr<CacheableInt32>>, 2nd part, Value is a ErrorCode. b. delete
* the worker
*/
auto resultMap = ResultMap();
auto failedServers = FailedServersMap();
for (const auto& worker : putAllWorkers) {
auto err =
worker->getResult(); // wait() or blocking call for worker thread.
LOGDEBUG("Error code :: %s:%d err = %d ", __FILE__, __LINE__, err);
if (GF_NOERR == err) {
// No Exception from server
resultMap.emplace(worker->getServerLocation(),
worker->getResultCollector()->getList());
} else {
error = err;
if (error == GF_PUTALL_PARTIAL_RESULT_EXCEPTION) {
resultMap.emplace(worker->getServerLocation(),
worker->getPaPResultException());
} else if (error == GF_NOTCON) {
// Refresh the metadata in case of GF_NOTCON.
tcrdm->getClientMetaDataService()->enqueueForMetadataRefresh(
region->getFullPath(), 0);
}
failedServers.emplace(worker->getServerLocation(),
CacheableInt32::create(error));
}
LOGDEBUG("worker->getPutAllMap()->size() = %zu ",
worker->getPutAllMap()->size());
LOGDEBUG(
"worker->getResultCollector()->getList()->getVersionedTagsize() = %d ",
worker->getResultCollector()->getList()->getVersionedTagsize());
// TODO::CHECK, why do we need following code... ??
// TcrMessage* currentReply = worker->getReply();
/*
if(currentReply->getMessageType() != TcrMessage::REPLY)
{
reply.setMessageType(currentReply->getMessageType());
}
*/
cnt++;
}
/**
* Step:4
* a. create instance of std::shared_ptr<PutAllPartialResult> with total size=
* map.size() b. Iterate over the resultMap and value for the particular
* serverlocation is of type VersionedCacheableObjectPartList add keys and
* versions. C. ToDO:: what if the value in the resultMap is of type
* PutAllPartialResultServerException
*/
std::recursive_mutex responseLock;
auto result = std::make_shared<PutAllPartialResult>(
static_cast<int>(map.size()), responseLock);
LOGDEBUG(
" TCRegion:: %s:%d "
"result->getSucceededKeysAndVersions()->getVersionedTagsize() = %d ",
__FILE__, __LINE__,
result->getSucceededKeysAndVersions()->getVersionedTagsize());
LOGDEBUG(" TCRegion:: %s:%d resultMap->size() ", __FILE__, __LINE__,
resultMap.size());
for (const auto& resultMapIter : resultMap) {
const auto& value = resultMapIter.second;
if (const auto papException =
std::dynamic_pointer_cast<PutAllPartialResultServerException>(
value)) {
// PutAllPartialResultServerException CASE:: value in resultMap is of type
// PutAllPartialResultServerException.
// TODO:: Add failedservers.keySet= all fialed servers, i.e list out all
// keys in map failedServers,
// that is set view of the keys contained in failedservers map.
// TODO:: need to read papException and populate PutAllPartialResult.
result->consolidate(papException->getResult());
} else if (const auto list =
std::dynamic_pointer_cast<VersionedCacheableObjectPartList>(
value)) {
// value in resultMap is of type VCOPL.
result->addKeysAndVersions(list);
} else {
// ERROR CASE
if (value) {
LOGERROR(
"ERROR:: ThinClientRegion::singleHopPutAllNoThrow_remote value "
"could not Cast to either VCOPL or "
"PutAllPartialResultServerException:%s",
value->toString().c_str());
} else {
LOGERROR(
"ERROR:: ThinClientRegion::singleHopPutAllNoThrow_remote value is "
"nullptr");
}
}
}
/**
* a. if PutAllPartialResult result does not contains any entry, Iterate over
* locationMap.
* b. Create std::vector<std::shared_ptr<CacheableKey>> succeedKeySet, and
* keep adding set of keys (locationIter.second()) in locationMap for which
* failedServers->contains(locationIter.first()is false.
*/
LOGDEBUG("ThinClientRegion:: %s:%d failedServers->size() = %zu", __FILE__,
__LINE__, failedServers.size());
// if the partial result set doesn't already have keys (for tracking version
// tags)
// then we need to gather up the keys that we know have succeeded so far and
// add them to the partial result set (See bug Id #955)
if (!failedServers.empty()) {
auto succeedKeySet =
std::make_shared<std::vector<std::shared_ptr<CacheableKey>>>();
if (result->getSucceededKeysAndVersions()->size() == 0) {
for (const auto& locationIter : *locationMap) {
if (failedServers.find(locationIter.first) != failedServers.end()) {
for (const auto& i : *(locationIter.second)) {
succeedKeySet->push_back(i);
}
}
}
result->addKeys(succeedKeySet);
}
}
/**
* a. Iterate over the failedServers map
* c. if failedServer map contains "GF_PUTALL_PARTIAL_RESULT_EXCEPTION" then
* continue, Do not retry putAll for corr. keys.
* b. Retry for all the failed server.
* Generate a newSubMap by finding Keys specific to failedServers from
* locationMap and finding their respective values from the usermap.
*/
error = GF_NOERR;
bool oneSubMapRetryFailed = false;
for (const auto& failedServerIter : failedServers) {
if (failedServerIter.second->value() ==
GF_PUTALL_PARTIAL_RESULT_EXCEPTION) { // serverLocation
// will not retry for PutAllPartialResultException
// but it means at least one sub map ever failed
oneSubMapRetryFailed = true;
error = GF_PUTALL_PARTIAL_RESULT_EXCEPTION;
continue;
}
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> failedKeys =
nullptr;
const auto& failedSerInLocMapIter =
locationMap->find(failedServerIter.first);
if (failedSerInLocMapIter != locationMap->end()) {
failedKeys = failedSerInLocMapIter->second;
}
if (failedKeys == nullptr) {
LOGERROR(
"TCRegion::singleHopPutAllNoThrow_remote :: failedKeys are nullptr "
"that is not valid");
}
auto newSubMap = std::make_shared<HashMapOfCacheable>();
if (failedKeys && !failedKeys->empty()) {
for (const auto& key : *failedKeys) {
const auto& iter = map.find(key);
if (iter != map.end()) {
newSubMap->emplace(iter->first, iter->second);
} else {
LOGERROR(
"DEBUG:: TCRegion.cpp singleHopPutAllNoThrow_remote KEY not "
"found in user failedSubMap");
}
}
}
std::shared_ptr<VersionedCacheableObjectPartList> vcopListPtr;
GfErrType errCode = multiHopPutAllNoThrow_remote(
*newSubMap, vcopListPtr, timeout, aCallbackArgument);
if (errCode == GF_NOERR) {
result->addKeysAndVersions(vcopListPtr);
} else if (errCode == GF_PUTALL_PARTIAL_RESULT_EXCEPTION) {
oneSubMapRetryFailed = true;
// TODO:: Commented it as papResultServerExc is nullptr this time
// UnComment it once you read papResultServerExc.
// result->consolidate(papResultServerExc->getResult());
error = errCode;
} else /*if(errCode != GF_NOERR)*/ {
oneSubMapRetryFailed = true;
const auto& firstKey = newSubMap->begin()->first;
std::shared_ptr<Exception> excptPtr = nullptr;
// TODO:: formulat excptPtr from the errCode
result->saveFailedKey(firstKey, excptPtr);
error = errCode;
}
}
if (!oneSubMapRetryFailed) {
error = GF_NOERR;
}
versionedObjPartList = result->getSucceededKeysAndVersions();
LOGDEBUG("singlehop versionedObjPartList = %d error=%d",
versionedObjPartList->size(), error);
return error;
}
GfErrType ThinClientRegion::multiHopPutAllNoThrow_remote(
const HashMapOfCacheable& map,
std::shared_ptr<VersionedCacheableObjectPartList>& versionedObjPartList,
std::chrono::milliseconds timeout,
const std::shared_ptr<Serializable>& aCallbackArgument) {
// Multiple hop implementation
LOGDEBUG("ThinClientRegion::multiHopPutAllNoThrow_remote ");
auto err = GF_NOERR;
// Construct request/reply for putAll
TcrMessagePutAll request(new DataOutput(m_cacheImpl->createDataOutput()),
this, map, timeout, m_tcrdm.get(),
aCallbackArgument);
TcrMessageReply reply(true, m_tcrdm.get());
request.setTimeout(timeout);
reply.setTimeout(timeout);
std::recursive_mutex responseLock;
versionedObjPartList =
std::make_shared<VersionedCacheableObjectPartList>(this, responseLock);
// need to check
ChunkedPutAllResponse* resultCollector(new ChunkedPutAllResponse(
shared_from_this(), reply, responseLock, versionedObjPartList));
reply.setChunkedResultHandler(resultCollector);
err = m_tcrdm->sendSyncRequest(request, reply);
versionedObjPartList = resultCollector->getList();
LOGDEBUG("multiple hop versionedObjPartList size = %d , err = %d ",
versionedObjPartList->size(), err);
delete resultCollector;
if (err != GF_NOERR) return err;
LOGDEBUG(
"ThinClientRegion::multiHopPutAllNoThrow_remote reply.getMessageType() = "
"%d ",
reply.getMessageType());
switch (reply.getMessageType()) {
case TcrMessage::REPLY:
// LOGDEBUG("Map is written into remote server at region %s",
// m_fullPath.c_str());
break;
case TcrMessage::RESPONSE:
LOGDEBUG(
"multiHopPutAllNoThrow_remote TcrMessage::RESPONSE %s, err = %d ",
m_fullPath.c_str(), err);
break;
case TcrMessage::EXCEPTION:
err = handleServerException("ThinClientRegion::putAllNoThrow",
reply.getException());
// TODO:: Do we need to read PutAllPartialServerException for multiple
// hop.
break;
case TcrMessage::PUT_DATA_ERROR:
// LOGERROR( "A write error occurred on the endpoint %s",
// m_tcrdm->getActiveEndpoint( )->name( ).c_str( ) );
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type %d during region put-all",
reply.getMessageType());
err = GF_NOTOBJ;
break;
}
return err;
}
GfErrType ThinClientRegion::putAllNoThrow_remote(
const HashMapOfCacheable& map,
std::shared_ptr<VersionedCacheableObjectPartList>& versionedObjPartList,
std::chrono::milliseconds timeout,
const std::shared_ptr<Serializable>& aCallbackArgument) {
LOGDEBUG("ThinClientRegion::putAllNoThrow_remote");
if (auto poolDM = std::dynamic_pointer_cast<ThinClientPoolDM>(m_tcrdm)) {
if (poolDM->getPRSingleHopEnabled() && poolDM->getClientMetaDataService() &&
!TSSTXStateWrapper::get().getTXState()) {
return singleHopPutAllNoThrow_remote(
poolDM.get(), map, versionedObjPartList, timeout, aCallbackArgument);
} else {
return multiHopPutAllNoThrow_remote(map, versionedObjPartList, timeout,
aCallbackArgument);
}
} else {
LOGERROR("ThinClientRegion::putAllNoThrow_remote :: Pool Not Specified ");
return GF_NOTSUP;
}
}
GfErrType ThinClientRegion::singleHopRemoveAllNoThrow_remote(
ThinClientPoolDM* tcrdm,
const std::vector<std::shared_ptr<CacheableKey>>& keys,
std::shared_ptr<VersionedCacheableObjectPartList>& versionedObjPartList,
const std::shared_ptr<Serializable>& aCallbackArgument) {
LOGDEBUG(
" ThinClientRegion::singleHopRemoveAllNoThrow_remote keys size = %zu",
keys.size());
auto region = shared_from_this();
GfErrType error = GF_NOERR;
auto locationMap = tcrdm->getClientMetaDataService()->getServerToFilterMap(
keys, region, true);
if (!locationMap) {
// removeAll with multiple hop implementation
LOGDEBUG("locationMap is Null or Empty");
return multiHopRemoveAllNoThrow_remote(keys, versionedObjPartList,
aCallbackArgument);
}
// set this flag that indicates putAll on PR is invoked with singlehop
// enabled.
m_isPRSingleHopEnabled = true;
LOGDEBUG("locationMap.size() = %zu ", locationMap->size());
/*Step-2
* a. create vector of RemoveAllWork
* b. locationMap<std::shared_ptr<BucketServerLocation>,
* std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>> >>. Create
* server specific filteredMap/subMap by populating all keys
* (locationIter.second()) and its corr. values from the user Map.
* c. create new instance of RemoveAllWork, i.e worker with required params.
* //TODO:: Add details of each parameter later
* d. enqueue the worker for thread from threadPool to perform/run execute
* method.
* e. insert the worker into the vector.
*/
std::vector<std::shared_ptr<RemoveAllWork>> removeAllWorkers;
auto& threadPool = m_cacheImpl->getThreadPool();
int locationMapIndex = 0;
for (const auto& locationIter : *locationMap) {
const auto& serverLocation = locationIter.first;
if (serverLocation == nullptr) {
LOGDEBUG("serverLocation is nullptr");
}
const auto& mappedkeys = locationIter.second;
auto worker = std::make_shared<RemoveAllWork>(
tcrdm, serverLocation, region, true /*attemptFailover*/,
false /*isBGThread*/, mappedkeys, aCallbackArgument);
threadPool.perform(worker);
removeAllWorkers.push_back(worker);
locationMapIndex++;
}
// TODO::CHECK, do we need to set following ..??
// reply.setMessageType(TcrMessage::RESPONSE);
int cnt = 1;
/**
* Step::3
* a. Iterate over all vector of putAllWorkers and populate worker specific
* information into the HashMap
* resultMap<std::shared_ptr<BucketServerLocation>,
* std::shared_ptr<Serializable>>, 2nd part, Value can be a
* std::shared_ptr<VersionedCacheableObjectPartList> or
* std::shared_ptr<PutAllPartialResultServerException>.
* failedServers<std::shared_ptr<BucketServerLocation>,
* std::shared_ptr<CacheableInt32>>, 2nd part, Value is a ErrorCode. b. delete
* the worker
*/
auto resultMap = ResultMap();
auto failedServers = FailedServersMap();
for (const auto& worker : removeAllWorkers) {
auto err =
worker->getResult(); // wait() or blocking call for worker thread.
LOGDEBUG("Error code :: %s:%d err = %d ", __FILE__, __LINE__, err);
if (GF_NOERR == err) {
// No Exception from server
resultMap.emplace(worker->getServerLocation(),
worker->getResultCollector()->getList());
} else {
error = err;
if (error == GF_PUTALL_PARTIAL_RESULT_EXCEPTION) {
resultMap.emplace(worker->getServerLocation(),
worker->getPaPResultException());
} else if (error == GF_NOTCON) {
// Refresh the metadata in case of GF_NOTCON.
tcrdm->getClientMetaDataService()->enqueueForMetadataRefresh(
region->getFullPath(), 0);
}
failedServers.emplace(worker->getServerLocation(),
CacheableInt32::create(error));
}
LOGDEBUG(
"worker->getResultCollector()->getList()->getVersionedTagsize() = %d ",
worker->getResultCollector()->getList()->getVersionedTagsize());
cnt++;
}
/**
* Step:4
* a. create instance of std::shared_ptr<PutAllPartialResult> with total size=
* map.size() b. Iterate over the resultMap and value for the particular
* serverlocation is of type VersionedCacheableObjectPartList add keys and
* versions. C. ToDO:: what if the value in the resultMap is of type
* PutAllPartialResultServerException
*/
std::recursive_mutex responseLock;
auto result = std::make_shared<PutAllPartialResult>(
static_cast<int>(keys.size()), responseLock);
LOGDEBUG(
" TCRegion:: %s:%d "
"result->getSucceededKeysAndVersions()->getVersionedTagsize() = %d ",
__FILE__, __LINE__,
result->getSucceededKeysAndVersions()->getVersionedTagsize());
LOGDEBUG(" TCRegion:: %s:%d resultMap->size() ", __FILE__, __LINE__,
resultMap.size());
for (const auto& resultMapIter : resultMap) {
const auto& value = resultMapIter.second;
if (const auto papException =
std::dynamic_pointer_cast<PutAllPartialResultServerException>(
value)) {
// PutAllPartialResultServerException CASE:: value in resultMap is of type
// PutAllPartialResultServerException.
// TODO:: Add failedservers.keySet= all fialed servers, i.e list out all
// keys in map failedServers,
// that is set view of the keys contained in failedservers map.
// TODO:: need to read papException and populate PutAllPartialResult.
result->consolidate(papException->getResult());
} else if (const auto list =
std::dynamic_pointer_cast<VersionedCacheableObjectPartList>(
value)) {
// value in resultMap is of type VCOPL.
result->addKeysAndVersions(list);
} else {
// ERROR CASE
if (value) {
LOGERROR(
"ERROR:: ThinClientRegion::singleHopRemoveAllNoThrow_remote value "
"could not Cast to either VCOPL or "
"PutAllPartialResultServerException:%s",
value->toString().c_str());
} else {
LOGERROR(
"ERROR:: ThinClientRegion::singleHopRemoveAllNoThrow_remote value "
"is nullptr");
}
}
}
/**
* a. if PutAllPartialResult result does not contains any entry, Iterate over
* locationMap.
* b. Create std::vector<std::shared_ptr<CacheableKey>> succeedKeySet, and
* keep adding set of keys (locationIter.second()) in locationMap for which
* failedServers->contains(locationIter.first()is false.
*/
LOGDEBUG("ThinClientRegion:: %s:%d failedServers->size() = %zu", __FILE__,
__LINE__, failedServers.size());
// if the partial result set doesn't already have keys (for tracking version
// tags)
// then we need to gather up the keys that we know have succeeded so far and
// add them to the partial result set (See bug Id #955)
if (!failedServers.empty()) {
auto succeedKeySet =
std::make_shared<std::vector<std::shared_ptr<CacheableKey>>>();
if (result->getSucceededKeysAndVersions()->size() == 0) {
for (const auto& locationIter : *locationMap) {
if (failedServers.find(locationIter.first) != failedServers.end()) {
for (const auto& i : *(locationIter.second)) {
succeedKeySet->push_back(i);
}
}
}
result->addKeys(succeedKeySet);
}
}
/**
* a. Iterate over the failedServers map
* c. if failedServer map contains "GF_PUTALL_PARTIAL_RESULT_EXCEPTION" then
* continue, Do not retry putAll for corr. keys.
* b. Retry for all the failed server.
* Generate a newSubMap by finding Keys specific to failedServers from
* locationMap and finding their respective values from the usermap.
*/
error = GF_NOERR;
bool oneSubMapRetryFailed = false;
for (const auto& failedServerIter : failedServers) {
if (failedServerIter.second->value() ==
GF_PUTALL_PARTIAL_RESULT_EXCEPTION) { // serverLocation
// will not retry for PutAllPartialResultException
// but it means at least one sub map ever failed
oneSubMapRetryFailed = true;
error = GF_PUTALL_PARTIAL_RESULT_EXCEPTION;
continue;
}
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> failedKeys =
nullptr;
const auto& failedSerInLocMapIter =
locationMap->find(failedServerIter.first);
if (failedSerInLocMapIter != locationMap->end()) {
failedKeys = failedSerInLocMapIter->second;
}
if (failedKeys == nullptr) {
LOGERROR(
"TCRegion::singleHopRemoveAllNoThrow_remote :: failedKeys are "
"nullptr "
"that is not valid");
}
std::shared_ptr<VersionedCacheableObjectPartList> vcopListPtr;
std::shared_ptr<PutAllPartialResultServerException> papResultServerExc =
nullptr;
GfErrType errCode = multiHopRemoveAllNoThrow_remote(
*failedKeys, vcopListPtr, aCallbackArgument);
if (errCode == GF_NOERR) {
result->addKeysAndVersions(vcopListPtr);
} else if (errCode == GF_PUTALL_PARTIAL_RESULT_EXCEPTION) {
oneSubMapRetryFailed = true;
error = errCode;
} else /*if(errCode != GF_NOERR)*/ {
oneSubMapRetryFailed = true;
std::shared_ptr<Exception> excptPtr = nullptr;
result->saveFailedKey(failedKeys->at(0), excptPtr);
error = errCode;
}
}
if (!oneSubMapRetryFailed) {
error = GF_NOERR;
}
versionedObjPartList = result->getSucceededKeysAndVersions();
LOGDEBUG("singlehop versionedObjPartList = %d error=%d",
versionedObjPartList->size(), error);
return error;
}
GfErrType ThinClientRegion::multiHopRemoveAllNoThrow_remote(
const std::vector<std::shared_ptr<CacheableKey>>& keys,
std::shared_ptr<VersionedCacheableObjectPartList>& versionedObjPartList,
const std::shared_ptr<Serializable>& aCallbackArgument) {
// Multiple hop implementation
LOGDEBUG("ThinClientRegion::multiHopRemoveAllNoThrow_remote ");
GfErrType err = GF_NOERR;
// Construct request/reply for putAll
TcrMessageRemoveAll request(new DataOutput(m_cacheImpl->createDataOutput()),
this, keys, aCallbackArgument, m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
std::recursive_mutex responseLock;
versionedObjPartList =
std::make_shared<VersionedCacheableObjectPartList>(this, responseLock);
// need to check
ChunkedRemoveAllResponse* resultCollector(new ChunkedRemoveAllResponse(
shared_from_this(), reply, responseLock, versionedObjPartList));
reply.setChunkedResultHandler(resultCollector);
err = m_tcrdm->sendSyncRequest(request, reply);
versionedObjPartList = resultCollector->getList();
LOGDEBUG("multiple hop versionedObjPartList size = %d , err = %d ",
versionedObjPartList->size(), err);
delete resultCollector;
if (err != GF_NOERR) return err;
LOGDEBUG(
"ThinClientRegion::multiHopRemoveAllNoThrow_remote "
"reply.getMessageType() = %d ",
reply.getMessageType());
switch (reply.getMessageType()) {
case TcrMessage::REPLY:
// LOGDEBUG("Map is written into remote server at region %s",
// m_fullPath.c_str());
break;
case TcrMessage::RESPONSE:
LOGDEBUG(
"multiHopRemoveAllNoThrow_remote TcrMessage::RESPONSE %s, err = %d ",
m_fullPath.c_str(), err);
break;
case TcrMessage::EXCEPTION:
err = handleServerException("ThinClientRegion::putAllNoThrow",
reply.getException());
break;
default:
LOGERROR("Unknown message type %d during region remove-all",
reply.getMessageType());
err = GF_NOTOBJ;
break;
}
return err;
}
GfErrType ThinClientRegion::removeAllNoThrow_remote(
const std::vector<std::shared_ptr<CacheableKey>>& keys,
std::shared_ptr<VersionedCacheableObjectPartList>& versionedObjPartList,
const std::shared_ptr<Serializable>& aCallbackArgument) {
LOGDEBUG("ThinClientRegion::removeAllNoThrow_remote");
if (auto poolDM = std::dynamic_pointer_cast<ThinClientPoolDM>(m_tcrdm)) {
if (poolDM->getPRSingleHopEnabled() && poolDM->getClientMetaDataService() &&
!TSSTXStateWrapper::get().getTXState()) {
return singleHopRemoveAllNoThrow_remote(
poolDM.get(), keys, versionedObjPartList, aCallbackArgument);
} else {
return multiHopRemoveAllNoThrow_remote(keys, versionedObjPartList,
aCallbackArgument);
}
} else {
LOGERROR(
"ThinClientRegion::removeAllNoThrow_remote :: Pool Not Specified ");
return GF_NOTSUP;
}
}
uint32_t ThinClientRegion::size_remote() {
LOGDEBUG("ThinClientRegion::size_remote");
GfErrType err = GF_NOERR;
// do TCR size
TcrMessageSize request(new DataOutput(m_cacheImpl->createDataOutput()),
m_fullPath.c_str());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) {
throwExceptionIfError("Region::size", err);
}
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE: {
auto size = std::dynamic_pointer_cast<CacheableInt32>(reply.getValue());
return size->value();
}
case TcrMessage::EXCEPTION:
err =
handleServerException("ThinClientRegion::size", reply.getException());
break;
case TcrMessage::SIZE_ERROR:
err = GF_CACHESERVER_EXCEPTION;
break;
default:
LOGERROR("Unknown message type %d during region size",
reply.getMessageType());
err = GF_NOTOBJ;
}
throwExceptionIfError("Region::size", err);
return 0;
}
GfErrType ThinClientRegion::registerStoredRegex(
TcrEndpoint* endpoint,
std::unordered_map<std::string, InterestResultPolicy>& interestListRegex,
bool isDurable, bool receiveValues) {
GfErrType opErr = GF_NOERR;
GfErrType retVal = GF_NOERR;
for (std::unordered_map<std::string, InterestResultPolicy>::iterator it =
interestListRegex.begin();
it != interestListRegex.end(); ++it) {
opErr = registerRegexNoThrow(it->first, false, endpoint, isDurable, nullptr,
it->second, receiveValues);
if (opErr != GF_NOERR) {
retVal = opErr;
}
}
return retVal;
}
GfErrType ThinClientRegion::registerKeys(TcrEndpoint* endpoint,
const TcrMessage* request,
TcrMessageReply* reply) {
GfErrType err = GF_NOERR;
GfErrType opErr = GF_NOERR;
// called when failover to a different server
opErr = registerStoredRegex(endpoint, m_interestListRegex);
err = opErr != GF_NOERR ? opErr : err;
opErr = registerStoredRegex(
endpoint, m_interestListRegexForUpdatesAsInvalidates, false, false);
err = opErr != GF_NOERR ? opErr : err;
opErr = registerStoredRegex(endpoint, m_durableInterestListRegex, true);
err = opErr != GF_NOERR ? opErr : err;
opErr = registerStoredRegex(
endpoint, m_durableInterestListRegexForUpdatesAsInvalidates, true, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVec;
InterestResultPolicy interestPolicy =
copyInterestList(keysVec, m_interestList);
opErr = registerKeysNoThrow(keysVec, false, endpoint, false, interestPolicy);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVecForUpdatesAsInvalidates;
interestPolicy = copyInterestList(keysVecForUpdatesAsInvalidates,
m_interestListForUpdatesAsInvalidates);
opErr = registerKeysNoThrow(keysVecForUpdatesAsInvalidates, false, endpoint,
false, interestPolicy, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVecDurable;
interestPolicy = copyInterestList(keysVecDurable, m_durableInterestList);
opErr = registerKeysNoThrow(keysVecDurable, false, endpoint, true,
interestPolicy);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>>
keysVecDurableForUpdatesAsInvalidates;
interestPolicy =
copyInterestList(keysVecDurableForUpdatesAsInvalidates,
m_durableInterestListForUpdatesAsInvalidates);
opErr = registerKeysNoThrow(keysVecDurableForUpdatesAsInvalidates, false,
endpoint, true, interestPolicy, false);
err = opErr != GF_NOERR ? opErr : err;
if (request != nullptr && request->getRegionName() == m_fullPath &&
(request->getMessageType() == TcrMessage::REGISTER_INTEREST ||
request->getMessageType() == TcrMessage::REGISTER_INTEREST_LIST)) {
const std::vector<std::shared_ptr<CacheableKey>>* newKeysVec =
request->getKeys();
bool isDurable = request->isDurable();
bool receiveValues = request->receiveValues();
if (newKeysVec == nullptr || newKeysVec->empty()) {
const std::string& newRegex = request->getRegex();
if (!newRegex.empty()) {
if (request->getRegionName() != m_fullPath) {
reply = nullptr;
}
opErr = registerRegexNoThrow(
newRegex, false, endpoint, isDurable, nullptr,
request->getInterestResultPolicy(), receiveValues, reply);
err = opErr != GF_NOERR ? opErr : err;
}
} else {
opErr = registerKeysNoThrow(*newKeysVec, false, endpoint, isDurable,
request->getInterestResultPolicy(),
receiveValues, reply);
err = opErr != GF_NOERR ? opErr : err;
}
}
return err;
}
GfErrType ThinClientRegion::unregisterStoredRegex(
std::unordered_map<std::string, InterestResultPolicy>& interestListRegex) {
GfErrType opErr = GF_NOERR;
GfErrType retVal = GF_NOERR;
for (std::unordered_map<std::string, InterestResultPolicy>::iterator it =
interestListRegex.begin();
it != interestListRegex.end(); ++it) {
opErr = unregisterRegexNoThrow(it->first, false);
if (opErr != GF_NOERR) {
retVal = opErr;
}
}
return retVal;
}
GfErrType ThinClientRegion::unregisterStoredRegexLocalDestroy(
std::unordered_map<std::string, InterestResultPolicy>& interestListRegex) {
GfErrType opErr = GF_NOERR;
GfErrType retVal = GF_NOERR;
for (std::unordered_map<std::string, InterestResultPolicy>::iterator it =
interestListRegex.begin();
it != interestListRegex.end(); ++it) {
opErr = unregisterRegexNoThrowLocalDestroy(it->first, false);
if (opErr != GF_NOERR) {
retVal = opErr;
}
}
return retVal;
}
GfErrType ThinClientRegion::unregisterKeys() {
GfErrType err = GF_NOERR;
GfErrType opErr = GF_NOERR;
// called when disconnect from a server
opErr = unregisterStoredRegex(m_interestListRegex);
err = opErr != GF_NOERR ? opErr : err;
opErr = unregisterStoredRegex(m_durableInterestListRegex);
err = opErr != GF_NOERR ? opErr : err;
opErr = unregisterStoredRegex(m_interestListRegexForUpdatesAsInvalidates);
err = opErr != GF_NOERR ? opErr : err;
opErr =
unregisterStoredRegex(m_durableInterestListRegexForUpdatesAsInvalidates);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVec;
copyInterestList(keysVec, m_interestList);
opErr = unregisterKeysNoThrow(keysVec, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVecDurable;
copyInterestList(keysVecDurable, m_durableInterestList);
opErr = unregisterKeysNoThrow(keysVecDurable, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>> keysVecForUpdatesAsInvalidates;
copyInterestList(keysVecForUpdatesAsInvalidates,
m_interestListForUpdatesAsInvalidates);
opErr = unregisterKeysNoThrow(keysVecForUpdatesAsInvalidates, false);
err = opErr != GF_NOERR ? opErr : err;
std::vector<std::shared_ptr<CacheableKey>>
keysVecDurableForUpdatesAsInvalidates;
copyInterestList(keysVecDurableForUpdatesAsInvalidates,
m_durableInterestListForUpdatesAsInvalidates);
opErr = unregisterKeysNoThrow(keysVecDurableForUpdatesAsInvalidates, false);
err = opErr != GF_NOERR ? opErr : err;
return err;
}
GfErrType ThinClientRegion::destroyRegionNoThrow_remote(
const std::shared_ptr<Serializable>& aCallbackArgument) {
GfErrType err = GF_NOERR;
// do TCR destroyRegion
TcrMessageDestroyRegion request(
new DataOutput(m_cacheImpl->createDataOutput()), this, aCallbackArgument,
std::chrono::milliseconds(-1), m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) return err;
switch (reply.getMessageType()) {
case TcrMessage::REPLY: {
// LOGINFO("Region %s at remote is destroyed successfully",
// m_fullPath.c_str());
break;
}
case TcrMessage::EXCEPTION: {
err =
handleServerException("Region::destroyRegion", reply.getException());
break;
}
case TcrMessage::DESTROY_REGION_DATA_ERROR: {
err = GF_CACHE_REGION_DESTROYED_EXCEPTION;
break;
}
default: {
LOGERROR("Unknown message type %d during destroy region",
reply.getMessageType());
err = GF_MSG;
}
}
return err;
}
GfErrType ThinClientRegion::registerKeysNoThrow(
const std::vector<std::shared_ptr<CacheableKey>>& keys,
bool attemptFailover, TcrEndpoint* endpoint, bool isDurable,
InterestResultPolicy interestPolicy, bool receiveValues,
TcrMessageReply* reply) {
RegionGlobalLocks acquireLocksRedundancy(this, false);
RegionGlobalLocks acquireLocksFailover(this);
CHECK_DESTROY_PENDING_NOTHROW(TryReadGuard);
GfErrType err = GF_NOERR;
std::lock_guard<decltype(m_keysLock)> keysGuard(m_keysLock);
if (keys.empty()) {
return err;
}
TcrMessageReply replyLocal(true, m_tcrdm.get());
bool needToCreateRC = true;
if (reply == nullptr) {
reply = &replyLocal;
} else {
needToCreateRC = false;
}
LOGDEBUG("ThinClientRegion::registerKeysNoThrow : interestpolicy is %d",
interestPolicy.ordinal);
TcrMessageRegisterInterestList request(
new DataOutput(m_cacheImpl->createDataOutput()), this, keys, isDurable,
getAttributes().getCachingEnabled(), receiveValues, interestPolicy,
m_tcrdm.get());
std::recursive_mutex responseLock;
TcrChunkedResult* resultCollector = nullptr;
if (interestPolicy.ordinal == InterestResultPolicy::KEYS_VALUES.ordinal) {
auto values = std::make_shared<HashMapOfCacheable>();
auto exceptions = std::make_shared<HashMapOfException>();
MapOfUpdateCounters trackers;
int32_t destroyTracker = 1;
if (needToCreateRC) {
resultCollector = (new ChunkedGetAllResponse(
request, this, &keys, values, exceptions, nullptr, trackers,
destroyTracker, true, responseLock));
reply->setChunkedResultHandler(resultCollector);
}
} else {
if (needToCreateRC) {
resultCollector = (new ChunkedInterestResponse(request, nullptr, *reply));
reply->setChunkedResultHandler(resultCollector);
}
}
err = m_tcrdm->sendSyncRequestRegisterInterest(
request, *reply, attemptFailover, this, endpoint);
if (err == GF_NOERR /*|| err == GF_CACHE_REDUNDANCY_FAILURE*/) {
if (reply->getMessageType() == TcrMessage::RESPONSE_FROM_SECONDARY &&
endpoint) {
LOGFINER(
"registerKeysNoThrow - got response from secondary for "
"endpoint %s, ignoring.",
endpoint->name().c_str());
} else if (attemptFailover) {
addKeys(keys, isDurable, receiveValues, interestPolicy);
if (!(interestPolicy.ordinal ==
InterestResultPolicy::KEYS_VALUES.ordinal)) {
localInvalidateForRegisterInterest(keys);
}
}
}
if (needToCreateRC) {
delete resultCollector;
}
return err;
}
GfErrType ThinClientRegion::unregisterKeysNoThrow(
const std::vector<std::shared_ptr<CacheableKey>>& keys,
bool attemptFailover) {
RegionGlobalLocks acquireLocksRedundancy(this, false);
RegionGlobalLocks acquireLocksFailover(this);
CHECK_DESTROY_PENDING_NOTHROW(TryReadGuard);
GfErrType err = GF_NOERR;
std::lock_guard<decltype(m_keysLock)> keysGuard(m_keysLock);
TcrMessageReply reply(true, m_tcrdm.get());
if (keys.empty()) {
return err;
}
if (m_interestList.empty() && m_durableInterestList.empty() &&
m_interestListForUpdatesAsInvalidates.empty() &&
m_durableInterestListForUpdatesAsInvalidates.empty()) {
// did not register any keys before.
return GF_CACHE_ILLEGAL_STATE_EXCEPTION;
}
TcrMessageUnregisterInterestList request(
new DataOutput(m_cacheImpl->createDataOutput()), this, keys, false, true,
InterestResultPolicy::NONE, m_tcrdm.get());
err = m_tcrdm->sendSyncRequestRegisterInterest(request, reply);
if (err == GF_NOERR /*|| err == GF_CACHE_REDUNDANCY_FAILURE*/) {
if (attemptFailover) {
for (const auto& key : keys) {
m_interestList.erase(key);
m_durableInterestList.erase(key);
m_interestListForUpdatesAsInvalidates.erase(key);
m_durableInterestListForUpdatesAsInvalidates.erase(key);
}
}
}
return err;
}
GfErrType ThinClientRegion::unregisterKeysNoThrowLocalDestroy(
const std::vector<std::shared_ptr<CacheableKey>>& keys,
bool attemptFailover) {
RegionGlobalLocks acquireLocksRedundancy(this, false);
RegionGlobalLocks acquireLocksFailover(this);
GfErrType err = GF_NOERR;
std::lock_guard<decltype(m_keysLock)> keysGuard(m_keysLock);
TcrMessageReply reply(true, m_tcrdm.get());
if (keys.empty()) {
return err;
}
if (m_interestList.empty() && m_durableInterestList.empty() &&
m_interestListForUpdatesAsInvalidates.empty() &&
m_durableInterestListForUpdatesAsInvalidates.empty()) {
// did not register any keys before.
return GF_CACHE_ILLEGAL_STATE_EXCEPTION;
}
TcrMessageUnregisterInterestList request(
new DataOutput(m_cacheImpl->createDataOutput()), this, keys, false, true,
InterestResultPolicy::NONE, m_tcrdm.get());
err = m_tcrdm->sendSyncRequestRegisterInterest(request, reply);
if (err == GF_NOERR) {
if (attemptFailover) {
for (const auto& key : keys) {
m_interestList.erase(key);
m_durableInterestList.erase(key);
m_interestListForUpdatesAsInvalidates.erase(key);
m_durableInterestListForUpdatesAsInvalidates.erase(key);
}
}
}
return err;
}
bool ThinClientRegion::isRegexRegistered(
std::unordered_map<std::string, InterestResultPolicy>& interestListRegex,
const std::string& regex, bool allKeys) {
if (interestListRegex.find(".*") != interestListRegex.end() ||
(!allKeys && interestListRegex.find(regex) != interestListRegex.end())) {
return true;
}
return false;
}
GfErrType ThinClientRegion::registerRegexNoThrow(
const std::string& regex, bool attemptFailover, TcrEndpoint* endpoint,
bool isDurable,
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> resultKeys,
InterestResultPolicy interestPolicy, bool receiveValues,
TcrMessageReply* reply) {
RegionGlobalLocks acquireLocksRedundancy(this, false);
RegionGlobalLocks acquireLocksFailover(this);
CHECK_DESTROY_PENDING_NOTHROW(TryReadGuard);
GfErrType err = GF_NOERR;
bool allKeys = (regex == ".*");
std::lock_guard<decltype(m_keysLock)> keysGuard(m_keysLock);
if (attemptFailover) {
if ((isDurable &&
(isRegexRegistered(m_durableInterestListRegex, regex, allKeys) ||
isRegexRegistered(m_durableInterestListRegexForUpdatesAsInvalidates,
regex, allKeys))) ||
(!isDurable &&
(isRegexRegistered(m_interestListRegex, regex, allKeys) ||
isRegexRegistered(m_interestListRegexForUpdatesAsInvalidates, regex,
allKeys)))) {
return err;
}
}
ChunkedInterestResponse* resultCollector = nullptr;
ChunkedGetAllResponse* getAllResultCollector = nullptr;
if (reply != nullptr) {
// need to check
resultCollector = dynamic_cast<ChunkedInterestResponse*>(
reply->getChunkedResultHandler());
if (resultCollector != nullptr) {
resultKeys = resultCollector->getResultKeys();
} else {
getAllResultCollector = dynamic_cast<ChunkedGetAllResponse*>(
reply->getChunkedResultHandler());
resultKeys = getAllResultCollector->getResultKeys();
}
}
bool isRCCreatedLocally = false;
LOGDEBUG("ThinClientRegion::registerRegexNoThrow : interestpolicy is %d",
interestPolicy.ordinal);
// TODO:
TcrMessageRegisterInterest request(
new DataOutput(m_cacheImpl->createDataOutput()), m_fullPath,
regex.c_str(), interestPolicy, isDurable,
getAttributes().getCachingEnabled(), receiveValues, m_tcrdm.get());
std::recursive_mutex responseLock;
if (reply == nullptr) {
TcrMessageReply replyLocal(true, m_tcrdm.get());
auto values = std::make_shared<HashMapOfCacheable>();
auto exceptions = std::make_shared<HashMapOfException>();
reply = &replyLocal;
if (interestPolicy.ordinal == InterestResultPolicy::KEYS_VALUES.ordinal) {
MapOfUpdateCounters trackers;
int32_t destroyTracker = 1;
if (resultKeys == nullptr) {
resultKeys =
std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>>(
new std::vector<std::shared_ptr<CacheableKey>>());
}
// need to check
getAllResultCollector = (new ChunkedGetAllResponse(
request, this, nullptr, values, exceptions, resultKeys, trackers,
destroyTracker, true, responseLock));
reply->setChunkedResultHandler(getAllResultCollector);
isRCCreatedLocally = true;
} else {
isRCCreatedLocally = true;
// need to check
resultCollector =
new ChunkedInterestResponse(request, resultKeys, replyLocal);
reply->setChunkedResultHandler(resultCollector);
}
err = m_tcrdm->sendSyncRequestRegisterInterest(
request, replyLocal, attemptFailover, this, endpoint);
} else {
err = m_tcrdm->sendSyncRequestRegisterInterest(
request, *reply, attemptFailover, this, endpoint);
}
if (err == GF_NOERR /*|| err == GF_CACHE_REDUNDANCY_FAILURE*/) {
if (reply->getMessageType() == TcrMessage::RESPONSE_FROM_SECONDARY &&
endpoint) {
LOGFINER(
"registerRegexNoThrow - got response from secondary for "
"endpoint %s, ignoring.",
endpoint->name().c_str());
} else if (attemptFailover) {
addRegex(regex, isDurable, receiveValues, interestPolicy);
if (interestPolicy.ordinal != InterestResultPolicy::KEYS_VALUES.ordinal) {
if (allKeys) {
localInvalidateRegion_internal();
} else {
const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>>&
keys = resultCollector != nullptr
? resultCollector->getResultKeys()
: getAllResultCollector->getResultKeys();
if (keys != nullptr) {
localInvalidateForRegisterInterest(*keys);
}
}
}
}
}
if (isRCCreatedLocally == true) {
if (resultCollector != nullptr) delete resultCollector;
if (getAllResultCollector != nullptr) delete getAllResultCollector;
}
return err;
}
GfErrType ThinClientRegion::unregisterRegexNoThrow(const std::string& regex,
bool attemptFailover) {
RegionGlobalLocks acquireLocksRedundancy(this, false);
RegionGlobalLocks acquireLocksFailover(this);
CHECK_DESTROY_PENDING_NOTHROW(TryReadGuard);
GfErrType err = GF_NOERR;
err = findRegex(regex);
if (err == GF_NOERR) {
TcrMessageReply reply(false, m_tcrdm.get());
TcrMessageUnregisterInterest request(
new DataOutput(m_cacheImpl->createDataOutput()), m_fullPath, regex,
InterestResultPolicy::NONE, false, true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequestRegisterInterest(request, reply);
if (err == GF_NOERR /*|| err == GF_CACHE_REDUNDANCY_FAILURE*/) {
if (attemptFailover) {
clearRegex(regex);
}
}
}
return err;
}
GfErrType ThinClientRegion::findRegex(const std::string& regex) {
GfErrType err = GF_NOERR;
std::lock_guard<decltype(m_keysLock)> keysGuard(m_keysLock);
if (m_interestListRegex.find(regex) == m_interestListRegex.end() &&
m_durableInterestListRegex.find(regex) ==
m_durableInterestListRegex.end() &&
m_interestListRegexForUpdatesAsInvalidates.find(regex) ==
m_interestListRegexForUpdatesAsInvalidates.end() &&
m_durableInterestListRegexForUpdatesAsInvalidates.find(regex) ==
m_durableInterestListRegexForUpdatesAsInvalidates.end()) {
return GF_CACHE_ILLEGAL_STATE_EXCEPTION;
} else {
return err;
}
}
void ThinClientRegion::clearRegex(const std::string& regex) {
std::lock_guard<decltype(m_keysLock)> keysGuard(m_keysLock);
m_interestListRegex.erase(regex);
m_durableInterestListRegex.erase(regex);
m_interestListRegexForUpdatesAsInvalidates.erase(regex);
m_durableInterestListRegexForUpdatesAsInvalidates.erase(regex);
}
GfErrType ThinClientRegion::unregisterRegexNoThrowLocalDestroy(
const std::string& regex, bool attemptFailover) {
GfErrType err = GF_NOERR;
err = findRegex(regex);
if (err == GF_NOERR) {
TcrMessageReply reply(false, m_tcrdm.get());
TcrMessageUnregisterInterest request(
new DataOutput(m_cacheImpl->createDataOutput()), m_fullPath, regex,
InterestResultPolicy::NONE, false, true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequestRegisterInterest(request, reply);
if (err == GF_NOERR) {
if (attemptFailover) {
clearRegex(regex);
}
}
}
return err;
}
void ThinClientRegion::addKeys(
const std::vector<std::shared_ptr<CacheableKey>>& keys, bool isDurable,
bool receiveValues, InterestResultPolicy interestpolicy) {
std::unordered_map<std::shared_ptr<CacheableKey>, InterestResultPolicy>&
interestList =
isDurable
? (receiveValues ? m_durableInterestList
: m_durableInterestListForUpdatesAsInvalidates)
: (receiveValues ? m_interestList
: m_interestListForUpdatesAsInvalidates);
for (const auto& key : keys) {
interestList.insert(
std::pair<std::shared_ptr<CacheableKey>, InterestResultPolicy>(
key, interestpolicy));
}
}
void ThinClientRegion::addRegex(const std::string& regex, bool isDurable,
bool receiveValues,
InterestResultPolicy interestpolicy) {
std::unordered_map<std::shared_ptr<CacheableKey>, InterestResultPolicy>&
interestList =
isDurable
? (receiveValues ? m_durableInterestList
: m_durableInterestListForUpdatesAsInvalidates)
: (receiveValues ? m_interestList
: m_interestListForUpdatesAsInvalidates);
std::unordered_map<std::string, InterestResultPolicy>& interestListRegex =
isDurable
? (receiveValues ? m_durableInterestListRegex
: m_durableInterestListRegexForUpdatesAsInvalidates)
: (receiveValues ? m_interestListRegex
: m_interestListRegexForUpdatesAsInvalidates);
if (regex == ".*") {
interestListRegex.clear();
interestList.clear();
}
interestListRegex.insert(
std::pair<std::string, InterestResultPolicy>(regex, interestpolicy));
}
std::vector<std::shared_ptr<CacheableKey>> ThinClientRegion::getInterestList()
const {
auto nthis = const_cast<ThinClientRegion*>(this);
RegionGlobalLocks acquireLocksRedundancy(nthis, false);
RegionGlobalLocks acquireLocksFailover(nthis);
CHECK_DESTROY_PENDING(TryReadGuard, getInterestList);
std::lock_guard<decltype(m_keysLock)> keysGuard(nthis->m_keysLock);
std::vector<std::shared_ptr<CacheableKey>> vlist;
std::transform(std::begin(m_durableInterestList),
std::end(m_durableInterestList), std::back_inserter(vlist),
[](const decltype(m_durableInterestList)::value_type& e) {
return e.first;
});
std::transform(
std::begin(m_interestList), std::end(m_interestList),
std::back_inserter(vlist),
[](const decltype(m_interestList)::value_type& e) { return e.first; });
return vlist;
}
std::vector<std::shared_ptr<CacheableString>>
ThinClientRegion::getInterestListRegex() const {
auto nthis = const_cast<ThinClientRegion*>(this);
RegionGlobalLocks acquireLocksRedundancy(nthis, false);
RegionGlobalLocks acquireLocksFailover(nthis);
CHECK_DESTROY_PENDING(TryReadGuard, getInterestListRegex);
std::lock_guard<decltype(m_keysLock)> keysGuard(nthis->m_keysLock);
std::vector<std::shared_ptr<CacheableString>> vlist;
std::transform(std::begin(m_durableInterestListRegex),
std::end(m_durableInterestListRegex),
std::back_inserter(vlist),
[](const decltype(m_durableInterestListRegex)::value_type& e) {
return CacheableString::create(e.first.c_str());
});
std::transform(std::begin(m_interestListRegex), std::end(m_interestListRegex),
std::back_inserter(vlist),
[](const decltype(m_interestListRegex)::value_type& e) {
return CacheableString::create(e.first.c_str());
});
return vlist;
}
GfErrType ThinClientRegion::clientNotificationHandler(TcrMessage& msg) {
GfErrType err = GF_NOERR;
std::shared_ptr<Cacheable> oldValue;
switch (msg.getMessageType()) {
case TcrMessage::LOCAL_INVALIDATE: {
LocalRegion::invalidateNoThrow(
msg.getKey(), msg.getCallbackArgument(), -1,
CacheEventFlags::NOTIFICATION | CacheEventFlags::LOCAL,
msg.getVersionTag());
break;
}
case TcrMessage::LOCAL_DESTROY: {
err = LocalRegion::destroyNoThrow(
msg.getKey(), msg.getCallbackArgument(), -1,
CacheEventFlags::NOTIFICATION | CacheEventFlags::LOCAL,
msg.getVersionTag());
break;
}
case TcrMessage::CLEAR_REGION: {
LOGDEBUG("remote clear region event for reigon[%s]",
msg.getRegionName().c_str());
err = localClearNoThrow(
nullptr, CacheEventFlags::NOTIFICATION | CacheEventFlags::LOCAL);
break;
}
case TcrMessage::LOCAL_DESTROY_REGION: {
m_notifyRelease = true;
err = LocalRegion::destroyRegionNoThrow(
msg.getCallbackArgument(), true,
CacheEventFlags::NOTIFICATION | CacheEventFlags::LOCAL);
break;
}
case TcrMessage::LOCAL_CREATE:
err = LocalRegion::putNoThrow(
msg.getKey(), msg.getValue(), msg.getCallbackArgument(), oldValue, -1,
CacheEventFlags::NOTIFICATION | CacheEventFlags::LOCAL,
msg.getVersionTag());
break;
case TcrMessage::LOCAL_UPDATE: {
// for update set the NOTIFICATION_UPDATE to trigger the
// afterUpdate event even if the key is not present in local cache
err = LocalRegion::putNoThrow(
msg.getKey(), msg.getValue(), msg.getCallbackArgument(), oldValue, -1,
CacheEventFlags::NOTIFICATION | CacheEventFlags::NOTIFICATION_UPDATE |
CacheEventFlags::LOCAL,
msg.getVersionTag(), msg.getDelta(), msg.getEventId());
break;
}
case TcrMessage::TOMBSTONE_OPERATION:
LocalRegion::tombstoneOperationNoThrow(msg.getTombstoneVersions(),
msg.getTombstoneKeys());
break;
default: {
if (TcrMessage::getAllEPDisMess() == &msg) {
setProcessedMarker(false);
LocalRegion::invokeAfterAllEndPointDisconnected();
} else {
LOGERROR(
"Unknown message type %d in subscription event handler; possible "
"serialization mismatch",
msg.getMessageType());
err = GF_MSG;
}
break;
}
}
// Update EventIdMap to mark event processed, Only for durable client.
// In case of closing, don't send it as listener might not be invoked.
if (!m_destroyPending && (m_isDurableClnt || msg.hasDelta()) &&
TcrMessage::getAllEPDisMess() != &msg) {
m_tcrdm->checkDupAndAdd(msg.getEventId());
}
return err;
}
GfErrType ThinClientRegion::handleServerException(const char* func,
const char* exceptionMsg) {
GfErrType error = GF_NOERR;
setThreadLocalExceptionMessage(exceptionMsg);
if (strstr(exceptionMsg,
"org.apache.geode.security.NotAuthorizedException") != nullptr) {
error = GF_NOT_AUTHORIZED_EXCEPTION;
} else if (strstr(exceptionMsg,
"org.apache.geode.cache.CacheWriterException") != nullptr) {
error = GF_CACHE_WRITER_EXCEPTION;
} else if (strstr(
exceptionMsg,
"org.apache.geode.security.AuthenticationFailedException") !=
nullptr) {
error = GF_AUTHENTICATION_FAILED_EXCEPTION;
} else if (strstr(exceptionMsg,
"org.apache.geode.internal.cache.execute."
"InternalFunctionInvocationTargetException") != nullptr) {
error = GF_FUNCTION_EXCEPTION;
} else if (strstr(exceptionMsg,
"org.apache.geode.cache.CommitConflictException") !=
nullptr) {
error = GF_COMMIT_CONFLICT_EXCEPTION;
} else if (strstr(exceptionMsg,
"org.apache.geode.cache."
"TransactionDataNodeHasDepartedException") != nullptr) {
error = GF_TRANSACTION_DATA_NODE_HAS_DEPARTED_EXCEPTION;
} else if (strstr(
exceptionMsg,
"org.apache.geode.cache.TransactionDataRebalancedException") !=
nullptr) {
error = GF_TRANSACTION_DATA_REBALANCED_EXCEPTION;
} else if (strstr(
exceptionMsg,
"org.apache.geode.security.AuthenticationRequiredException") !=
nullptr) {
error = GF_AUTHENTICATION_REQUIRED_EXCEPTION;
} else {
error = GF_CACHESERVER_EXCEPTION;
}
if (error != GF_AUTHENTICATION_REQUIRED_EXCEPTION) {
LOGERROR("%s: An exception (%s) happened at remote server.", func,
exceptionMsg);
} else {
LOGFINER("%s: An exception (%s) happened at remote server.", func,
exceptionMsg);
}
return error;
}
void ThinClientRegion::receiveNotification(TcrMessage* msg) {
std::unique_lock<std::mutex> lock(m_notificationMutex, std::defer_lock);
{
TryReadGuard guard(m_rwLock, m_destroyPending);
if (m_destroyPending) {
if (msg != TcrMessage::getAllEPDisMess()) {
_GEODE_SAFE_DELETE(msg);
}
return;
}
lock.lock();
}
if (msg->getMessageType() == TcrMessage::CLIENT_MARKER) {
handleMarker();
} else {
clientNotificationHandler(*msg);
}
lock.unlock();
if (TcrMessage::getAllEPDisMess() != msg) _GEODE_SAFE_DELETE(msg);
}
void ThinClientRegion::localInvalidateRegion_internal() {
std::shared_ptr<MapEntryImpl> me;
std::shared_ptr<Cacheable> oldValue;
std::vector<std::shared_ptr<CacheableKey>> keysVec = keys_internal();
for (const auto& key : keysVec) {
std::shared_ptr<VersionTag> versionTag;
m_entries->invalidate(key, me, oldValue, versionTag);
}
}
void ThinClientRegion::invalidateInterestList(
std::unordered_map<std::shared_ptr<CacheableKey>, InterestResultPolicy>&
interestList) {
std::shared_ptr<MapEntryImpl> me;
std::shared_ptr<Cacheable> oldValue;
if (!m_regionAttributes.getCachingEnabled()) {
return;
}
for (const auto& iter : interestList) {
std::shared_ptr<VersionTag> versionTag;
m_entries->invalidate(iter.first, me, oldValue, versionTag);
}
}
void ThinClientRegion::localInvalidateFailover() {
CHECK_DESTROY_PENDING(TryReadGuard,
ThinClientRegion::localInvalidateFailover);
// No need to invalidate from the "m_xxxForUpdatesAsInvalidates" lists?
if (m_interestListRegex.empty() && m_durableInterestListRegex.empty()) {
invalidateInterestList(m_interestList);
invalidateInterestList(m_durableInterestList);
} else {
localInvalidateRegion_internal();
}
}
void ThinClientRegion::localInvalidateForRegisterInterest(
const std::vector<std::shared_ptr<CacheableKey>>& keys) {
CHECK_DESTROY_PENDING(TryReadGuard,
ThinClientRegion::localInvalidateForRegisterInterest);
if (!m_regionAttributes.getCachingEnabled()) {
return;
}
std::shared_ptr<Cacheable> oldValue;
std::shared_ptr<MapEntryImpl> me;
for (const auto& key : keys) {
std::shared_ptr<VersionTag> versionTag;
m_entries->invalidate(key, me, oldValue, versionTag);
updateAccessAndModifiedTimeForEntry(me, true);
}
}
InterestResultPolicy ThinClientRegion::copyInterestList(
std::vector<std::shared_ptr<CacheableKey>>& keysVector,
std::unordered_map<std::shared_ptr<CacheableKey>, InterestResultPolicy>&
interestList) const {
InterestResultPolicy interestPolicy = InterestResultPolicy::NONE;
for (std::unordered_map<std::shared_ptr<CacheableKey>,
InterestResultPolicy>::const_iterator iter =
interestList.begin();
iter != interestList.end(); ++iter) {
keysVector.push_back(iter->first);
interestPolicy = iter->second;
}
return interestPolicy;
}
void ThinClientRegion::registerInterestGetValues(
const char* method, const std::vector<std::shared_ptr<CacheableKey>>* keys,
const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>>&
resultKeys) {
auto exceptions = std::make_shared<HashMapOfException>();
auto err = getAllNoThrow_remote(keys, nullptr, exceptions, resultKeys, true,
nullptr);
throwExceptionIfError(method, err);
// log any exceptions here
for (const auto& iter : *exceptions) {
LOGWARN("%s Exception for key %s:: %s: %s", method,
Utils::nullSafeToString(iter.first).c_str(),
iter.second->getName().c_str(), iter.second->what());
}
}
void ThinClientRegion::destroyDM(bool keepEndpoints) {
if (m_tcrdm != nullptr) {
m_tcrdm->destroy(keepEndpoints);
}
}
void ThinClientRegion::release(bool invokeCallbacks) {
if (m_released) {
return;
}
std::unique_lock<std::mutex> lock(m_notificationMutex, std::defer_lock);
if (!m_notifyRelease) {
lock.lock();
}
destroyDM(invokeCallbacks);
m_interestList.clear();
m_interestListRegex.clear();
m_durableInterestList.clear();
m_durableInterestListRegex.clear();
m_interestListForUpdatesAsInvalidates.clear();
m_interestListRegexForUpdatesAsInvalidates.clear();
m_durableInterestListForUpdatesAsInvalidates.clear();
m_durableInterestListRegexForUpdatesAsInvalidates.clear();
LocalRegion::release(invokeCallbacks);
}
ThinClientRegion::~ThinClientRegion() noexcept {
TryWriteGuard guard(m_rwLock, m_destroyPending);
if (!m_destroyPending) {
release(false);
}
}
void ThinClientRegion::acquireGlobals(bool isFailover) {
if (isFailover) {
m_tcrdm->acquireFailoverLock();
}
}
void ThinClientRegion::releaseGlobals(bool isFailover) {
if (isFailover) {
m_tcrdm->releaseFailoverLock();
}
}
void ThinClientRegion::executeFunction(
const std::string& func, const std::shared_ptr<Cacheable>& args,
std::shared_ptr<CacheableVector> routingObj, uint8_t getResult,
std::shared_ptr<ResultCollector> rc, int32_t retryAttempts,
std::chrono::milliseconds timeout) {
int32_t attempt = 0;
auto failedNodes = CacheableHashSet::create();
// if pools retry attempts are not set then retry once on all available
// endpoints
if (retryAttempts == -1) {
retryAttempts = static_cast<int32_t>(m_tcrdm->getNumberOfEndPoints());
}
bool reExecute = false;
bool reExecuteForServ = false;
do {
TcrMessage* msg;
if (reExecuteForServ) {
msg = new TcrMessageExecuteRegionFunction(
new DataOutput(m_cacheImpl->createDataOutput()), func, this, args,
routingObj, getResult, failedNodes, timeout, m_tcrdm.get(),
static_cast<int8_t>(1));
} else {
msg = new TcrMessageExecuteRegionFunction(
new DataOutput(m_cacheImpl->createDataOutput()), func, this, args,
routingObj, getResult, failedNodes, timeout, m_tcrdm.get(),
static_cast<int8_t>(0));
}
TcrMessageReply reply(true, m_tcrdm.get());
// need to check
ChunkedFunctionExecutionResponse* resultCollector(
new ChunkedFunctionExecutionResponse(reply, (getResult & 2) == 2, rc));
reply.setChunkedResultHandler(resultCollector);
reply.setTimeout(timeout);
GfErrType err = GF_NOERR;
err = m_tcrdm->sendSyncRequest(*msg, reply, !(getResult & 1));
resultCollector->reset();
delete msg;
delete resultCollector;
if (err == GF_NOERR &&
(reply.getMessageType() == TcrMessage::EXCEPTION ||
reply.getMessageType() == TcrMessage::EXECUTE_REGION_FUNCTION_ERROR)) {
err = ThinClientRegion::handleServerException("Execute",
reply.getException());
}
if (ThinClientBaseDM::isFatalClientError(err)) {
throwExceptionIfError("ExecuteOnRegion:", err);
} else if (err != GF_NOERR) {
if (err == GF_FUNCTION_EXCEPTION) {
reExecute = true;
rc->clearResults();
std::shared_ptr<CacheableHashSet> failedNodesIds(reply.getFailedNode());
failedNodes->clear();
if (failedNodesIds) {
LOGDEBUG(
"ThinClientRegion::executeFunction with GF_FUNCTION_EXCEPTION "
"failedNodesIds size = %zu ",
failedNodesIds->size());
failedNodes->insert(failedNodesIds->begin(), failedNodesIds->end());
}
} else if (err == GF_NOTCON) {
attempt++;
LOGDEBUG(
"ThinClientRegion::executeFunction with GF_NOTCON retry attempt = "
"%d ",
attempt);
if (attempt > retryAttempts) {
throwExceptionIfError("ExecuteOnRegion:", err);
}
reExecuteForServ = true;
rc->clearResults();
failedNodes->clear();
} else if (err == GF_TIMEOUT) {
LOGINFO("function timeout. Name: %s, timeout: %z, params: %" PRIu8
", "
"retryAttempts: %d ",
func.c_str(), timeout.count(), getResult, retryAttempts);
throwExceptionIfError("ExecuteOnRegion", GF_TIMEOUT);
} else if (err == GF_CLIENT_WAIT_TIMEOUT ||
err == GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA) {
LOGINFO(
"function timeout, possibly bucket is not available or bucket "
"blacklisted. Name: %s, timeout: %z, params: %" PRIu8
", retryAttempts: "
"%d ",
func.c_str(), timeout.count(), getResult, retryAttempts);
throwExceptionIfError("ExecuteOnRegion", GF_CLIENT_WAIT_TIMEOUT);
} else {
LOGDEBUG("executeFunction err = %d ", err);
throwExceptionIfError("ExecuteOnRegion:", err);
}
} else {
reExecute = false;
reExecuteForServ = false;
}
} while (reExecuteForServ);
if (reExecute && (getResult & 1)) {
reExecuteFunction(func, args, routingObj, getResult, rc, retryAttempts,
failedNodes, timeout);
}
}
std::shared_ptr<CacheableVector> ThinClientRegion::reExecuteFunction(
const std::string& func, const std::shared_ptr<Cacheable>& args,
std::shared_ptr<CacheableVector> routingObj, uint8_t getResult,
std::shared_ptr<ResultCollector> rc, int32_t retryAttempts,
std::shared_ptr<CacheableHashSet>& failedNodes,
std::chrono::milliseconds timeout) {
int32_t attempt = 0;
bool reExecute = true;
// if pools retry attempts are not set then retry once on all available
// endpoints
if (retryAttempts == -1) {
retryAttempts = static_cast<int32_t>(m_tcrdm->getNumberOfEndPoints());
}
do {
reExecute = false;
TcrMessageExecuteRegionFunction msg(
new DataOutput(m_cacheImpl->createDataOutput()), func, this, args,
routingObj, getResult, failedNodes, timeout, m_tcrdm.get(),
static_cast<int8_t>(1));
TcrMessageReply reply(true, m_tcrdm.get());
// need to check
ChunkedFunctionExecutionResponse* resultCollector(
new ChunkedFunctionExecutionResponse(reply, (getResult & 2) == 2, rc));
reply.setChunkedResultHandler(resultCollector);
reply.setTimeout(timeout);
GfErrType err = GF_NOERR;
err = m_tcrdm->sendSyncRequest(msg, reply, !(getResult & 1));
delete resultCollector;
if (err == GF_NOERR &&
(reply.getMessageType() == TcrMessage::EXCEPTION ||
reply.getMessageType() == TcrMessage::EXECUTE_REGION_FUNCTION_ERROR)) {
err = ThinClientRegion::handleServerException("Execute",
reply.getException());
}
if (ThinClientBaseDM::isFatalClientError(err)) {
throwExceptionIfError("ExecuteOnRegion:", err);
} else if (err != GF_NOERR) {
if (err == GF_FUNCTION_EXCEPTION) {
reExecute = true;
rc->clearResults();
std::shared_ptr<CacheableHashSet> failedNodesIds(reply.getFailedNode());
failedNodes->clear();
if (failedNodesIds) {
LOGDEBUG(
"ThinClientRegion::reExecuteFunction with GF_FUNCTION_EXCEPTION "
"failedNodesIds size = %zu ",
failedNodesIds->size());
failedNodes->insert(failedNodesIds->begin(), failedNodesIds->end());
}
} else if ((err == GF_NOTCON) || (err == GF_CLIENT_WAIT_TIMEOUT) ||
(err == GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA)) {
attempt++;
LOGDEBUG(
"ThinClientRegion::reExecuteFunction with GF_NOTCON OR TIMEOUT "
"retry attempt "
"= %d ",
attempt);
if (attempt > retryAttempts) {
throwExceptionIfError("ExecuteOnRegion:", err);
}
reExecute = true;
rc->clearResults();
failedNodes->clear();
} else if (err == GF_TIMEOUT) {
LOGINFO("function timeout");
throwExceptionIfError("ExecuteOnRegion", GF_CACHE_TIMEOUT_EXCEPTION);
} else {
LOGDEBUG("reExecuteFunction err = %d ", err);
throwExceptionIfError("ExecuteOnRegion:", err);
}
}
} while (reExecute);
return nullptr;
}
bool ThinClientRegion::executeFunctionSH(
const std::string& func, const std::shared_ptr<Cacheable>& args,
uint8_t getResult, std::shared_ptr<ResultCollector> rc,
const std::shared_ptr<ClientMetadataService::ServerToKeysMap>& locationMap,
std::shared_ptr<CacheableHashSet>& failedNodes,
std::chrono::milliseconds timeout, bool allBuckets) {
bool reExecute = false;
auto resultCollectorLock = std::make_shared<std::recursive_mutex>();
const auto& userAttr = UserAttributes::threadLocalUserAttributes;
std::vector<std::shared_ptr<OnRegionFunctionExecution>> feWorkers;
auto& threadPool =
CacheRegionHelper::getCacheImpl(&getCache())->getThreadPool();
for (const auto& locationIter : *locationMap) {
const auto& serverLocation = locationIter.first;
const auto& routingObj = locationIter.second;
auto worker = std::make_shared<OnRegionFunctionExecution>(
func, this, args, routingObj, getResult, timeout,
dynamic_cast<ThinClientPoolDM*>(m_tcrdm.get()), resultCollectorLock, rc,
userAttr, false, serverLocation, allBuckets);
threadPool.perform(worker);
feWorkers.push_back(worker);
}
GfErrType abortError = GF_NOERR;
for (auto worker : feWorkers) {
auto err = worker->getResult();
auto currentReply = worker->getReply();
if (err == GF_NOERR &&
(currentReply->getMessageType() == TcrMessage::EXCEPTION ||
currentReply->getMessageType() ==
TcrMessage::EXECUTE_REGION_FUNCTION_ERROR)) {
err = ThinClientRegion::handleServerException(
"Execute", currentReply->getException());
}
if (err != GF_NOERR) {
if (err == GF_FUNCTION_EXCEPTION) {
reExecute = true;
if (auto poolDM =
std::dynamic_pointer_cast<ThinClientPoolDM>(m_tcrdm)) {
if (poolDM->getClientMetaDataService()) {
poolDM->getClientMetaDataService()->enqueueForMetadataRefresh(
this->getFullPath(), 0);
}
}
worker->getResultCollector()->reset();
{
std::lock_guard<decltype(*resultCollectorLock)> guard(
*resultCollectorLock);
rc->clearResults();
}
std::shared_ptr<CacheableHashSet> failedNodeIds(
currentReply->getFailedNode());
if (failedNodeIds) {
LOGDEBUG(
"ThinClientRegion::executeFunctionSH with GF_FUNCTION_EXCEPTION "
"failedNodeIds size = %zu ",
failedNodeIds->size());
failedNodes->insert(failedNodeIds->begin(), failedNodeIds->end());
}
} else if ((err == GF_NOTCON) || (err == GF_CLIENT_WAIT_TIMEOUT) ||
(err == GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA)) {
reExecute = true;
LOGINFO(
"ThinClientRegion::executeFunctionSH with GF_NOTCON or "
"GF_CLIENT_WAIT_TIMEOUT ");
if (auto poolDM =
std::dynamic_pointer_cast<ThinClientPoolDM>(m_tcrdm)) {
if (poolDM->getClientMetaDataService()) {
poolDM->getClientMetaDataService()->enqueueForMetadataRefresh(
this->getFullPath(), 0);
}
}
worker->getResultCollector()->reset();
{
std::lock_guard<decltype(*resultCollectorLock)> guard(
*resultCollectorLock);
rc->clearResults();
}
} else {
if (ThinClientBaseDM::isFatalClientError(err)) {
LOGERROR("ThinClientRegion::executeFunctionSH: Fatal Exception");
} else {
LOGWARN("ThinClientRegion::executeFunctionSH: Unexpected Exception");
}
if (abortError == GF_NOERR) {
abortError = err;
}
}
}
}
if (abortError != GF_NOERR) {
throwExceptionIfError("ExecuteOnRegion:", abortError);
}
return reExecute;
}
GfErrType ThinClientRegion::getFuncAttributes(const std::string& func,
std::vector<int8_t>** attr) {
GfErrType err = GF_NOERR;
// do TCR GET_FUNCTION_ATTRIBUTES
LOGDEBUG("Tcrmessage request GET_FUNCTION_ATTRIBUTES ");
TcrMessageGetFunctionAttributes request(
new DataOutput(m_cacheImpl->createDataOutput()), func, m_tcrdm.get());
TcrMessageReply reply(true, m_tcrdm.get());
err = m_tcrdm->sendSyncRequest(request, reply);
if (err != GF_NOERR) {
return err;
}
switch (reply.getMessageType()) {
case TcrMessage::RESPONSE: {
*attr = reply.getFunctionAttributes();
break;
}
case TcrMessage::EXCEPTION: {
err = handleServerException("Region::GET_FUNCTION_ATTRIBUTES",
reply.getException());
break;
}
case TcrMessage::REQUEST_DATA_ERROR: {
LOGERROR("Error message from server: " + reply.getValue()->toString());
throw FunctionExecutionException(reply.getValue()->toString());
}
default: {
LOGERROR("Unknown message type %d while getting function attributes.",
reply.getMessageType());
err = GF_MSG;
break;
}
}
return err;
}
GfErrType ThinClientRegion::getNoThrow_FullObject(
std::shared_ptr<EventId> eventId, std::shared_ptr<Cacheable>& fullObject,
std::shared_ptr<VersionTag>& versionTag) {
TcrMessageRequestEventValue fullObjectMsg(
new DataOutput(m_cacheImpl->createDataOutput()), eventId);
TcrMessageReply reply(true, nullptr);
GfErrType err = GF_NOTCON;
err = m_tcrdm->sendSyncRequest(fullObjectMsg, reply, false, true);
if (err == GF_NOERR) {
fullObject = reply.getValue();
}
versionTag = reply.getVersionTag();
return err;
}
void ThinClientRegion::txDestroy(
const std::shared_ptr<CacheableKey>& key,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag> versionTag) {
GfErrType err = destroyNoThrowTX(key, aCallbackArgument, -1,
CacheEventFlags::NORMAL, versionTag);
throwExceptionIfError("Region::destroyTX", err);
}
void ThinClientRegion::txInvalidate(
const std::shared_ptr<CacheableKey>& key,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag> versionTag) {
GfErrType err = invalidateNoThrowTX(key, aCallbackArgument, -1,
CacheEventFlags::NORMAL, versionTag);
throwExceptionIfError("Region::invalidateTX", err);
}
void ThinClientRegion::txPut(
const std::shared_ptr<CacheableKey>& key,
const std::shared_ptr<Cacheable>& value,
const std::shared_ptr<Serializable>& aCallbackArgument,
std::shared_ptr<VersionTag> versionTag) {
std::shared_ptr<Cacheable> oldValue;
int64_t sampleStartNanos = startStatOpTime();
GfErrType err = putNoThrowTX(key, value, aCallbackArgument, oldValue, -1,
CacheEventFlags::NORMAL, versionTag);
updateStatOpTime(m_regionStats->getStat(), m_regionStats->getPutTimeId(),
sampleStartNanos);
throwExceptionIfError("Region::putTX", err);
}
void ThinClientRegion::setProcessedMarker(bool) {}
void ChunkedInterestResponse::reset() {
if (m_resultKeys != nullptr && m_resultKeys->size() > 0) {
m_resultKeys->clear();
}
}
void ChunkedInterestResponse::handleChunk(const uint8_t* chunk,
int32_t chunkLen,
uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
auto input =
cacheImpl->createDataInput(chunk, chunkLen, m_replyMsg.getPool());
uint32_t partLen;
if (TcrMessageHelper::readChunkPartHeader(
m_msg, input, DSCode::FixedIDDefault,
static_cast<int32_t>(DSCode::CacheableArrayList),
"ChunkedInterestResponse", partLen, isLastChunkWithSecurity) !=
TcrMessageHelper::ChunkObjectType::OBJECT) {
// encountered an exception part, so return without reading more
m_replyMsg.readSecureObjectPart(input, false, true,
isLastChunkWithSecurity);
return;
}
if (m_resultKeys == nullptr) {
m_resultKeys =
std::make_shared<std::vector<std::shared_ptr<CacheableKey>>>();
}
serializer::readObject(input, *m_resultKeys);
m_replyMsg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
}
void ChunkedKeySetResponse::reset() {
if (m_resultKeys.size() > 0) {
m_resultKeys.clear();
}
}
void ChunkedKeySetResponse::handleChunk(const uint8_t* chunk, int32_t chunkLen,
uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
auto input =
cacheImpl->createDataInput(chunk, chunkLen, m_replyMsg.getPool());
uint32_t partLen;
if (TcrMessageHelper::readChunkPartHeader(
m_msg, input, DSCode::FixedIDDefault,
static_cast<int32_t>(DSCode::CacheableArrayList),
"ChunkedKeySetResponse", partLen, isLastChunkWithSecurity) !=
TcrMessageHelper::ChunkObjectType::OBJECT) {
// encountered an exception part, so return without reading more
m_replyMsg.readSecureObjectPart(input, false, true,
isLastChunkWithSecurity);
return;
}
serializer::readObject(input, m_resultKeys);
m_replyMsg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
}
void ChunkedQueryResponse::reset() {
m_queryResults->clear();
m_structFieldNames.clear();
}
void ChunkedQueryResponse::readObjectPartList(DataInput& input,
bool isResultSet) {
if (input.readBoolean()) {
LOGERROR("Query response has keys which is unexpected.");
throw IllegalStateException("Query response has keys which is unexpected.");
}
int32_t len = input.readInt32();
for (int32_t index = 0; index < len; ++index) {
if (input.read() == 2 /* for exception*/) {
input.advanceCursor(input.readArrayLength()); // skipLen
auto exMsgPtr = input.readString();
throw IllegalStateException(exMsgPtr);
} else {
if (isResultSet) {
std::shared_ptr<Cacheable> value;
input.readObject(value);
m_queryResults->push_back(value);
} else {
auto code = static_cast<DSCode>(input.read());
if (code == DSCode::FixedIDByte) {
auto arrayType = static_cast<DSFid>(input.read());
if (arrayType != DSFid::CacheableObjectPartList) {
LOGERROR(
"Query response got unhandled message format %d while "
"expecting struct set object part list; possible serialization "
"mismatch",
arrayType);
throw MessageException(
"Query response got unhandled message format while expecting "
"struct set object part list; possible serialization mismatch");
}
readObjectPartList(input, true);
} else {
LOGERROR(
"Query response got unhandled message format %" PRId8
"while expecting "
"struct set object part list; possible serialization mismatch",
code);
throw MessageException(
"Query response got unhandled message format while expecting "
"struct set object part list; possible serialization mismatch");
}
}
}
}
}
void ChunkedQueryResponse::handleChunk(const uint8_t* chunk, int32_t chunkLen,
uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
LOGDEBUG("ChunkedQueryResponse::handleChunk..");
auto input = cacheImpl->createDataInput(chunk, chunkLen, m_msg.getPool());
uint32_t partLen;
auto objType = TcrMessageHelper::readChunkPartHeader(
m_msg, input, DSCode::FixedIDByte,
static_cast<int32_t>(DSFid::CollectionTypeImpl), "ChunkedQueryResponse",
partLen, isLastChunkWithSecurity);
if (objType == TcrMessageHelper::ChunkObjectType::EXCEPTION) {
// encountered an exception part, so return without reading more
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
} else if (objType == TcrMessageHelper::ChunkObjectType::NULL_OBJECT) {
// special case for scalar result
input.readInt32(); // ignored part length
input.read(); // ignored is object
auto intVal = std::dynamic_pointer_cast<CacheableInt32>(input.readObject());
m_queryResults->push_back(intVal);
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
}
// ignoring parent classes for now
// we will require to look at it once CQ is to be implemented.
// skipping HashSet/StructSet
// qhe: It was agreed upon that we'll use set for all kinds of results.
// to avoid dealing with compare operator for user objets.
// If the results on server are in a bag, or the user need to manipulate
// the elements, then we have to revisit this issue.
// For now, we'll live with duplicate records, hoping they do not cost much.
skipClass(input);
// skipping CollectionTypeImpl
// skipClass(input); // no longer, since GFE 5.7
input.read(); // this is Fixed ID byte (1)
input.read(); // this is DataSerializable (45)
input.read(); // this is Class 43
const auto isStructTypeImpl = input.readString();
if (isStructTypeImpl == "org.apache.geode.cache.query.Struct") {
int32_t numOfFldNames = input.readArrayLength();
bool skip = false;
if (m_structFieldNames.size() != 0) {
skip = true;
}
for (int i = 0; i < numOfFldNames; i++) {
auto sptr = input.readString();
if (!skip) {
m_structFieldNames.push_back(sptr);
}
}
}
// skip the remaining part
input.reset();
// skip the whole part including partLen and isObj (4+1)
input.advanceCursor(partLen + 5);
input.readInt32(); // skip part length
if (!input.read()) {
LOGERROR(
"Query response part is not an object; possible serialization "
"mismatch");
throw MessageException(
"Query response part is not an object; possible serialization "
"mismatch");
}
bool isResultSet = (m_structFieldNames.size() == 0);
auto arrayType = static_cast<DSCode>(input.read());
if (arrayType == DSCode::CacheableObjectArray) {
int32_t arraySize = input.readArrayLength();
skipClass(input);
for (int32_t arrayItem = 0; arrayItem < arraySize; ++arrayItem) {
std::shared_ptr<Serializable> value;
if (isResultSet) {
input.readObject(value);
m_queryResults->push_back(value);
} else {
input.read();
int32_t arraySize2 = input.readArrayLength();
skipClass(input);
for (int32_t index = 0; index < arraySize2; ++index) {
input.readObject(value);
m_queryResults->push_back(value);
}
}
}
} else if (arrayType == DSCode::FixedIDByte) {
arrayType = static_cast<DSCode>(input.read());
if (static_cast<int32_t>(arrayType) !=
static_cast<int32_t>(DSFid::CacheableObjectPartList)) {
LOGERROR(
"Query response got unhandled message format %d while expecting "
"object part list; possible serialization mismatch",
arrayType);
throw MessageException(
"Query response got unhandled message format while expecting object "
"part list; possible serialization mismatch");
}
readObjectPartList(input, isResultSet);
} else {
LOGERROR(
"Query response got unhandled message format %d; possible "
"serialization mismatch",
arrayType);
throw MessageException(
"Query response got unhandled message format; possible serialization "
"mismatch");
}
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
}
void ChunkedQueryResponse::skipClass(DataInput& input) {
auto classByte = static_cast<DSCode>(input.read());
if (classByte == DSCode::Class) {
// ignore string type id - assuming its a normal (under 64k) string.
input.read();
uint16_t classLen = input.readInt16();
input.advanceCursor(classLen);
} else {
throw IllegalStateException(
"ChunkedQueryResponse::skipClass: "
"Did not get expected class header byte");
}
}
void ChunkedFunctionExecutionResponse::reset() {
// m_functionExecutionResults->clear();
}
void ChunkedFunctionExecutionResponse::handleChunk(
const uint8_t* chunk, int32_t chunkLen, uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
LOGDEBUG("ChunkedFunctionExecutionResponse::handleChunk");
auto input = cacheImpl->createDataInput(chunk, chunkLen, m_msg.getPool());
uint32_t partLen;
TcrMessageHelper::ChunkObjectType arrayType;
if ((arrayType = TcrMessageHelper::readChunkPartHeader(
m_msg, input, "ChunkedFunctionExecutionResponse", partLen,
isLastChunkWithSecurity)) ==
TcrMessageHelper::ChunkObjectType::EXCEPTION) {
// encountered an exception part, so return without reading more
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
}
if (m_getResult == false) {
return;
}
if (static_cast<TcrMessageHelper::ChunkObjectType>(arrayType) ==
TcrMessageHelper::ChunkObjectType::NULL_OBJECT) {
LOGDEBUG("ChunkedFunctionExecutionResponse::handleChunk nullptr object");
// m_functionExecutionResults->push_back(nullptr);
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
}
auto startLen =
input.getBytesRead() -
1; // from here need to look value part + memberid AND -1 for array type
// iread adn gnore array length
input.readArrayLength();
// read a byte to determine whether to read exception part for sendException
// or read objects.
auto partType = static_cast<DSCode>(input.read());
bool isExceptionPart = false;
// See If partType is JavaSerializable
const int CHUNK_HDR_LEN = 5;
const int SECURE_PART_LEN = 5 + 8;
bool readPart = true;
LOGDEBUG(
"ChunkedFunctionExecutionResponse::handleChunk chunkLen = %d & partLen = "
"%d ",
chunkLen, partLen);
if (partType == DSCode::JavaSerializable) {
isExceptionPart = true;
// reset the input.
input.reset();
if (((isLastChunkWithSecurity & 0x02) &&
(chunkLen - static_cast<int32_t>(partLen) <=
CHUNK_HDR_LEN + SECURE_PART_LEN)) ||
(((isLastChunkWithSecurity & 0x02) == 0) &&
(chunkLen - static_cast<int32_t>(partLen) <= CHUNK_HDR_LEN))) {
readPart = false;
partLen = input.readInt32();
input.advanceCursor(1); // skip isObject byte
input.advanceCursor(partLen);
} else {
// skip first part i.e JavaSerializable.
TcrMessageHelper::skipParts(m_msg, input, 1);
// read the second part which is string in usual manner, first its length.
partLen = input.readInt32();
// then isObject byte
input.read(); // ignore iSobject
startLen = input.getBytesRead(); // reset from here need to look value
// part + memberid AND -1 for array type
// Since it is contained as a part of other results, read arrayType which
// is arrayList = 65.
input.read();
// read and ignore its len which is 2
input.readArrayLength();
}
} else {
// rewind cursor by 1 to what we had read a byte to determine whether to
// read exception part or read objects.
input.rewindCursor(1);
}
// Read either object or exception string from sendException.
std::shared_ptr<Serializable> value;
// std::shared_ptr<Cacheable> memberId;
if (readPart) {
input.readObject(value);
// TODO: track this memberId for PrFxHa
// input.readObject(memberId);
auto objectlen = input.getBytesRead() - startLen;
auto memberIdLen = partLen - objectlen;
input.advanceCursor(memberIdLen);
LOGDEBUG("function partlen = %d , objectlen = %z, memberidlen = %z ",
partLen, objectlen, memberIdLen);
LOGDEBUG("function input.getBytesRemaining() = %z ",
input.getBytesRemaining());
} else {
value = CacheableString::create("Function exception result.");
}
if (m_rc != nullptr) {
std::shared_ptr<Cacheable> result = nullptr;
if (isExceptionPart) {
result = std::make_shared<UserFunctionExecutionException>(
std::dynamic_pointer_cast<CacheableString>(value)->value());
} else {
result = value;
}
if (m_resultCollectorLock) {
std::lock_guard<decltype(*m_resultCollectorLock)> guard(
*m_resultCollectorLock);
m_rc->addResult(result);
} else {
m_rc->addResult(result);
}
}
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
// m_functionExecutionResults->push_back(value);
}
void ChunkedGetAllResponse::reset() {
m_keysOffset = 0;
if (m_resultKeys != nullptr && m_resultKeys->size() > 0) {
m_resultKeys->clear();
}
}
// process a GET_ALL response chunk
void ChunkedGetAllResponse::handleChunk(const uint8_t* chunk, int32_t chunkLen,
uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
auto input = cacheImpl->createDataInput(chunk, chunkLen, m_msg.getPool());
uint32_t partLen;
if (TcrMessageHelper::readChunkPartHeader(
m_msg, input, DSCode::FixedIDByte,
static_cast<int32_t>(DSFid::VersionedObjectPartList),
"ChunkedGetAllResponse", partLen, isLastChunkWithSecurity) !=
TcrMessageHelper::ChunkObjectType::OBJECT) {
// encountered an exception part, so return without reading more
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
}
VersionedCacheableObjectPartList objectList(
m_keys, &m_keysOffset, m_values, m_exceptions, m_resultKeys, m_region,
&m_trackerMap, m_destroyTracker, m_addToLocalCache, m_dsmemId,
m_responseLock);
objectList.fromData(input);
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
}
void ChunkedGetAllResponse::add(const ChunkedGetAllResponse* other) {
if (m_values) {
for (const auto& iter : *m_values) {
m_values->emplace(iter.first, iter.second);
}
}
if (m_exceptions) {
m_exceptions->insert(other->m_exceptions->begin(),
other->m_exceptions->end());
}
for (const auto& iter : other->m_trackerMap) {
m_trackerMap[iter.first] = iter.second;
}
if (m_resultKeys) {
m_resultKeys->insert(m_resultKeys->end(), other->m_resultKeys->begin(),
other->m_resultKeys->end());
}
}
void ChunkedPutAllResponse::reset() {
if (m_list != nullptr && m_list->size() > 0) {
m_list->getVersionedTagptr().clear();
}
}
// process a PUT_ALL response chunk
void ChunkedPutAllResponse::handleChunk(const uint8_t* chunk, int32_t chunkLen,
uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
auto input = cacheImpl->createDataInput(chunk, chunkLen, m_msg.getPool());
uint32_t partLen;
TcrMessageHelper::ChunkObjectType chunkType;
if ((chunkType = TcrMessageHelper::readChunkPartHeader(
m_msg, input, DSCode::FixedIDByte,
static_cast<int32_t>(DSFid::VersionedObjectPartList),
"ChunkedPutAllResponse", partLen, isLastChunkWithSecurity)) ==
TcrMessageHelper::ChunkObjectType::NULL_OBJECT) {
LOGDEBUG("ChunkedPutAllResponse::handleChunk nullptr object");
// No issues it will be empty in case of disabled caching.
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
}
if (chunkType == TcrMessageHelper::ChunkObjectType::OBJECT) {
LOGDEBUG("ChunkedPutAllResponse::handleChunk object");
std::recursive_mutex responseLock;
auto vcObjPart = std::make_shared<VersionedCacheableObjectPartList>(
dynamic_cast<ThinClientRegion*>(m_region.get()),
m_msg.getChunkedResultHandler()->getEndpointMemId(), responseLock);
vcObjPart->fromData(input);
m_list->addAll(vcObjPart);
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
} else {
LOGDEBUG("ChunkedPutAllResponse::handleChunk BYTES PART");
const auto byte0 = input.read();
LOGDEBUG("ChunkedPutAllResponse::handleChunk single-hop bytes byte0 = %d ",
byte0);
const auto byte1 = input.read();
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
auto pool = m_msg.getPool();
if (pool != nullptr && !pool->isDestroyed() &&
pool->getPRSingleHopEnabled()) {
auto poolDM = dynamic_cast<ThinClientPoolDM*>(pool);
if ((poolDM != nullptr) &&
(poolDM->getClientMetaDataService() != nullptr) && (byte0 != 0)) {
LOGFINE("enqueued region " + m_region->getFullPath() +
" for metadata refresh for singlehop for PUTALL operation.");
poolDM->getClientMetaDataService()->enqueueForMetadataRefresh(
m_region->getFullPath(), byte1);
}
}
}
}
void ChunkedRemoveAllResponse::reset() {
if (m_list != nullptr && m_list->size() > 0) {
m_list->getVersionedTagptr().clear();
}
}
// process a REMOVE_ALL response chunk
void ChunkedRemoveAllResponse::handleChunk(const uint8_t* chunk,
int32_t chunkLen,
uint8_t isLastChunkWithSecurity,
const CacheImpl* cacheImpl) {
auto input = cacheImpl->createDataInput(chunk, chunkLen, m_msg.getPool());
uint32_t partLen;
TcrMessageHelper::ChunkObjectType chunkType;
if ((chunkType = TcrMessageHelper::readChunkPartHeader(
m_msg, input, DSCode::FixedIDByte,
static_cast<int32_t>(DSFid::VersionedObjectPartList),
"ChunkedRemoveAllResponse", partLen, isLastChunkWithSecurity)) ==
TcrMessageHelper::ChunkObjectType::NULL_OBJECT) {
LOGDEBUG("ChunkedRemoveAllResponse::handleChunk nullptr object");
// No issues it will be empty in case of disabled caching.
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
return;
}
if (chunkType == TcrMessageHelper::ChunkObjectType::OBJECT) {
LOGDEBUG("ChunkedRemoveAllResponse::handleChunk object");
std::recursive_mutex responseLock;
auto vcObjPart = std::make_shared<VersionedCacheableObjectPartList>(
dynamic_cast<ThinClientRegion*>(m_region.get()),
m_msg.getChunkedResultHandler()->getEndpointMemId(), responseLock);
vcObjPart->fromData(input);
m_list->addAll(vcObjPart);
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
} else {
LOGDEBUG("ChunkedRemoveAllResponse::handleChunk BYTES PART");
const auto byte0 = input.read();
LOGDEBUG(
"ChunkedRemoveAllResponse::handleChunk single-hop bytes byte0 = %d ",
byte0);
const auto byte1 = input.read();
m_msg.readSecureObjectPart(input, false, true, isLastChunkWithSecurity);
auto pool = m_msg.getPool();
if (pool != nullptr && !pool->isDestroyed() &&
pool->getPRSingleHopEnabled()) {
auto poolDM = dynamic_cast<ThinClientPoolDM*>(pool);
if ((poolDM != nullptr) &&
(poolDM->getClientMetaDataService() != nullptr) && (byte0 != 0)) {
LOGFINE("enqueued region " + m_region->getFullPath() +
" for metadata refresh for singlehop for REMOVEALL operation.");
poolDM->getClientMetaDataService()->enqueueForMetadataRefresh(
m_region->getFullPath(), byte1);
}
}
}
}
void ChunkedDurableCQListResponse::reset() {
if (m_resultList) {
m_resultList->clear();
}
}
// handles the chunk response for GETDURABLECQS_MSG_TYPE
void ChunkedDurableCQListResponse::handleChunk(const uint8_t* chunk,
int32_t chunkLen, uint8_t,
const CacheImpl* cacheImpl) {
auto input = cacheImpl->createDataInput(chunk, chunkLen, m_msg.getPool());
// read and ignore part length
input.readInt32();
if (!input.readBoolean()) {
// we're currently always expecting an object
char exMsg[256];
std::snprintf(
exMsg, 255,
"ChunkedDurableCQListResponse::handleChunk: part is not object");
throw MessageException(exMsg);
}
input.advanceCursor(1); // skip the CacheableArrayList type ID byte
const auto stringParts = input.read(); // read the number of strings in the
// message this is one byte
for (int i = 0; i < stringParts; i++) {
m_resultList->push_back(
std::dynamic_pointer_cast<CacheableString>(input.readObject()));
}
}
} // namespace client
} // namespace geode
} // namespace apache