/*
 * 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 "ThinClientPoolDM.hpp"

#include <algorithm>
#include <thread>

#include <ace/INET_Addr.h>
#include <ace/OS_NS_unistd.h>

#include <geode/AuthInitialize.hpp>
#include <geode/PoolManager.hpp>
#include <geode/ResultCollector.hpp>
#include <geode/SystemProperties.hpp>

#include "DistributedSystemImpl.hpp"
#include "ExecutionImpl.hpp"
#include "ExpiryHandler_T.hpp"
#include "ExpiryTaskManager.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrEndpoint.hpp"
#include "ThinClientRegion.hpp"
#include "ThinClientStickyManager.hpp"
#include "UserAttributes.hpp"
#include "statistics/PoolStatsSampler.hpp"
#include "util/exception.hpp"

/** Closes and Deletes connection only if it exists */
#define GF_SAFE_DELETE_CON(x) \
  do {                        \
    x->close();               \
    delete x;                 \
    x = nullptr;              \
  } while (0)

namespace apache {
namespace geode {
namespace client {

class GetAllWork : public PooledWork<GfErrType> {
  ThinClientPoolDM* m_poolDM;
  std::shared_ptr<BucketServerLocation> m_serverLocation;
  TcrMessage* m_request;
  TcrMessageReply* m_reply;
  bool m_attemptFailover;
  bool m_isBGThread;
  bool m_addToLocalCache;
  std::shared_ptr<UserAttributes> m_userAttribute;
  ChunkedGetAllResponse* m_responseHandler;
  std::string m_regionName;
  const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>> m_keys;
  const std::shared_ptr<Region> m_region;
  TcrChunkedResult* m_resultCollector;
  const std::shared_ptr<Serializable>& m_aCallbackArgument;

 public:
  GetAllWork(const GetAllWork&) = delete;
  GetAllWork& operator=(const GetAllWork&) = delete;
  GetAllWork(
      ThinClientPoolDM* poolDM, const std::shared_ptr<Region>& region,
      const std::shared_ptr<BucketServerLocation>& serverLocation,
      const std::shared_ptr<std::vector<std::shared_ptr<CacheableKey>>>& keys,
      bool attemptFailover, bool isBGThread, bool addToLocalCache,
      ChunkedGetAllResponse* responseHandler,
      const std::shared_ptr<Serializable>& aCallbackArgument)
      : m_poolDM(poolDM),
        m_serverLocation(serverLocation),
        m_attemptFailover(attemptFailover),
        m_isBGThread(isBGThread),
        m_addToLocalCache(addToLocalCache),
        m_userAttribute(nullptr),
        m_responseHandler(responseHandler),
        m_regionName(region->getFullPath()),
        m_keys(keys),
        m_region(region),
        m_aCallbackArgument(aCallbackArgument) {
    m_request = new TcrMessageGetAll(
        new DataOutput(region->getCache().createDataOutput()), region.get(),
        m_keys.get(), m_poolDM, m_aCallbackArgument);
    m_reply = new TcrMessageReply(true, m_poolDM);
    if (m_poolDM->isMultiUserMode()) {
      m_userAttribute = UserAttributes::threadLocalUserAttributes;
    }

    m_resultCollector = (new ChunkedGetAllResponse(
        *m_reply, dynamic_cast<ThinClientRegion*>(m_region.get()), m_keys.get(),
        m_responseHandler->getValues(), m_responseHandler->getExceptions(),
        m_responseHandler->getResultKeys(),
        m_responseHandler->getUpdateCounters(), 0, m_addToLocalCache,
        m_responseHandler->getResponseLock()));

    m_reply->setChunkedResultHandler(m_resultCollector);
  }

  ~GetAllWork() override {
    delete m_request;
    delete m_reply;
    delete m_resultCollector;
  }

  TcrMessage* getReply() { return m_reply; }

  void init() {}
  GfErrType execute() override {
    GuardUserAttributes gua;

    if (m_userAttribute != nullptr) {
      gua.setAuthenticatedView(m_userAttribute->getAuthenticatedView());
    }
    m_request->InitializeGetallMsg(
        m_request->getCallbackArgument());  // now init getall msg
    return m_poolDM->sendSyncRequest(*m_request, *m_reply, m_attemptFailover,
                                     m_isBGThread, m_serverLocation);
  }
};

const char* ThinClientPoolDM::NC_Ping_Thread = "NC Ping Thread";
const char* ThinClientPoolDM::NC_MC_Thread = "NC MC Thread";
#define PRIMARY_QUEUE_NOT_AVAILABLE -2

ThinClientPoolDM::ThinClientPoolDM(const char* name,
                                   std::shared_ptr<PoolAttributes> poolAttrs,
                                   TcrConnectionManager& connManager)
    : ThinClientBaseDM(connManager, nullptr),
      Pool(poolAttrs),
      m_poolName(name),
      m_stats(nullptr),
      m_sticky(false),
      m_updateLocatorListSema(0),
      m_pingSema(0),
      m_cliCallbackSema(0),
      m_isDestroyed(false),
      m_destroyPending(false),
      m_destroyPendingHADM(false),
      m_isMultiUserMode(false),
      m_locHelper(nullptr),
      m_poolSize(0),
      m_numRegions(0),
      m_server(0),
      m_connSema(0),
      m_connManageTask(nullptr),
      m_pingTask(nullptr),
      m_updateLocatorListTask(nullptr),
      m_cliCallbackTask(nullptr),
      m_pingTaskId(-1),
      m_updateLocatorListTaskId(-1),
      m_connManageTaskId(-1),
      m_clientOps(0),
      m_PoolStatsSampler(nullptr),
      m_clientMetadataService(nullptr),
      m_primaryServerQueueSize(PRIMARY_QUEUE_NOT_AVAILABLE) {
  static bool firstGuard = false;
  if (firstGuard) {
    ClientProxyMembershipID::increaseSynchCounter();
  }
  firstGuard = true;

  auto cacheImpl = m_connManager.getCacheImpl();
  auto& distributedSystem = cacheImpl->getDistributedSystem();

  auto& sysProp = distributedSystem.getSystemProperties();
  // to set security flag at pool level
  m_isSecurityOn = cacheImpl->getAuthInitialize() != nullptr;

  const auto& durableId = sysProp.durableClientId();

  std::string clientDurableId = durableId;
  if (!m_poolName.empty()) {
    clientDurableId += "_gem_" + m_poolName;
  }

  const auto durableTimeOut = sysProp.durableTimeout();
  m_memId = cacheImpl->getClientProxyMembershipIDFactory().create(
      clientDurableId.c_str(), durableTimeOut);

  if (m_attrs->m_initLocList.empty() && m_attrs->m_initServList.empty()) {
    std::string msg = "No locators or servers provided for pool named ";
    msg += name;
    throw IllegalStateException(msg);
  }
  reset();
  m_locHelper = new ThinClientLocatorHelper(m_attrs->m_initLocList,
                                            m_attrs->m_sniProxyHost,
                                            m_attrs->m_sniProxyPort, this);

  m_stats = new PoolStats(
      cacheImpl->getStatisticsManager().getStatisticsFactory(), m_poolName);
  cacheImpl->getStatisticsManager().forceSample();

  if (!sysProp.isEndpointShufflingDisabled()) {
    if (!m_attrs->m_initServList.empty()) {
      RandGen randgen;
      m_server = randgen(static_cast<uint32_t>(m_attrs->m_initServList.size()));
    }
  }
  if (m_attrs->getPRSingleHopEnabled()) {
    m_clientMetadataService =
        std::unique_ptr<ClientMetadataService>(new ClientMetadataService(this));
  }
  m_manager = new ThinClientStickyManager(this);
}

void ThinClientPoolDM::init() {
  LOGDEBUG("ThinClientPoolDM::init: Starting pool initialization");
  auto cacheImpl = m_connManager.getCacheImpl();
  m_isMultiUserMode = getMultiuserAuthentication();

  if (m_isMultiUserMode) {
    LOGINFO("Multiuser authentication is enabled for pool %s",
            m_poolName.c_str());
  }
  // to set security flag at pool level
  m_isSecurityOn = cacheImpl->getAuthInitialize() != nullptr;

  LOGDEBUG("ThinClientPoolDM::init: security in on/off = %d ", m_isSecurityOn);

  m_connManager.init(true);

  ThinClientPoolDM::startBackgroundThreads();

  LOGDEBUG("ThinClientPoolDM::init: Completed initialization");
}

ThinClientPoolDM::~ThinClientPoolDM() {
  destroy();
  _GEODE_SAFE_DELETE(m_locHelper);
  _GEODE_SAFE_DELETE(m_stats);
  _GEODE_SAFE_DELETE(m_manager);
}

std::shared_ptr<Properties> ThinClientPoolDM::getCredentials(TcrEndpoint* ep) {
  auto cacheImpl = m_connManager.getCacheImpl();
  const auto& distributedSystem = cacheImpl->getDistributedSystem();
  const auto& tmpSecurityProperties =
      distributedSystem.getSystemProperties().getSecurityProperties();

  if (const auto& authInitialize = cacheImpl->getAuthInitialize()) {
    LOGFINER(
        "ThinClientPoolDM::getCredentials: acquired handle to authLoader, "
        "invoking getCredentials %s",
        ep->name().c_str());
    const auto& tmpAuthIniSecurityProperties =
        authInitialize->getCredentials(tmpSecurityProperties, ep->name());
    LOGFINER("Done getting credentials");
    return tmpAuthIniSecurityProperties;
  }

  return nullptr;
}

void ThinClientPoolDM::startBackgroundThreads() {
  LOGDEBUG("ThinClientPoolDM::startBackgroundThreads: Starting ping thread");
  m_pingTask =
      std::unique_ptr<Task<ThinClientPoolDM>>(new Task<ThinClientPoolDM>(
          this, &ThinClientPoolDM::pingServer, NC_Ping_Thread));
  m_pingTask->start();

  auto& props = m_connManager.getCacheImpl()
                    ->getDistributedSystem()
                    .getSystemProperties();

  if (props.onClientDisconnectClearPdxTypeIds()) {
    m_cliCallbackTask =
        std::unique_ptr<Task<ThinClientPoolDM>>(new Task<ThinClientPoolDM>(
            this, &ThinClientPoolDM::cliCallback, "NC_cliCallback"));
    m_cliCallbackTask->start();
  }

  const auto& pingInterval = getPingInterval() / 2;
  if (pingInterval > std::chrono::seconds::zero()) {
    LOGDEBUG(
        "ThinClientPoolDM::startBackgroundThreads: Scheduling ping task at %ld",
        pingInterval.count());
    auto pingHandler =
        new ExpiryHandler_T<ThinClientPoolDM>(this, &ThinClientPoolDM::doPing);
    m_pingTaskId =
        m_connManager.getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
            pingHandler, std::chrono::seconds(1), pingInterval, false);
  } else {
    LOGDEBUG(
        "ThinClientPoolDM::startBackgroundThreads: Not Scheduling ping task as "
        "ping interval %ld",
        getPingInterval().count());
  }

  auto updateLocatorListInterval = getUpdateLocatorListInterval();

  if (updateLocatorListInterval > std::chrono::seconds::zero()) {
    m_updateLocatorListTask =
        std::unique_ptr<Task<ThinClientPoolDM>>(new Task<ThinClientPoolDM>(
            this, &ThinClientPoolDM::updateLocatorList, "NC_LocatorList"));
    m_updateLocatorListTask->start();

    LOGDEBUG(
        "ThinClientPoolDM::startBackgroundThreads: Creating updateLocatorList "
        "task");
    auto updateLocatorListHandler = new ExpiryHandler_T<ThinClientPoolDM>(
        this, &ThinClientPoolDM::doUpdateLocatorList);

    LOGDEBUG(
        "ThinClientPoolDM::startBackgroundThreads: Scheduling updater Locator "
        "task at %ld",
        updateLocatorListInterval.count());
    m_updateLocatorListTaskId =
        m_connManager.getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
            updateLocatorListHandler, std::chrono::seconds(1),
            updateLocatorListInterval, false);
  }

  LOGDEBUG(
      "ThinClientPoolDM::startBackgroundThreads: Starting manageConnections "
      "thread");
  // Manage Connection Thread
  m_connManageTask =
      std::unique_ptr<Task<ThinClientPoolDM>>(new Task<ThinClientPoolDM>(
          this, &ThinClientPoolDM::manageConnections, NC_MC_Thread));
  m_connManageTask->start();

  auto idle = getIdleTimeout();
  auto load = getLoadConditioningInterval();

  if (load > std::chrono::milliseconds::zero()) {
    if (load < idle || idle <= std::chrono::milliseconds::zero()) {
      idle = load;
    }
  }

  if (idle > std::chrono::milliseconds::zero()) {
    LOGDEBUG(
        "ThinClientPoolDM::startBackgroundThreads: Starting manageConnections "
        "task");
    ACE_Event_Handler* connHandler = new ExpiryHandler_T<ThinClientPoolDM>(
        this, &ThinClientPoolDM::doManageConnections);

    LOGDEBUG(
        "ThinClientPoolDM::startBackgroundThreads: Scheduling "
        "manageConnections task");
    m_connManageTaskId =
        m_connManager.getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
            connHandler, std::chrono::seconds(1), idle, false);
  }

  LOGDEBUG(
      "ThinClientPoolDM::startBackgroundThreads: Starting remote query "
      "service");
  // Init Query Service
  m_remoteQueryServicePtr =
      std::make_shared<RemoteQueryService>(m_connManager.getCacheImpl(), this);
  m_remoteQueryServicePtr->init();

  LOGDEBUG(
      "ThinClientPoolDM::startBackgroundThreads: Starting pool stat sampler");
  if (!m_PoolStatsSampler &&
      getStatisticInterval() > std::chrono::milliseconds::zero() &&
      props.statisticsEnabled()) {
    m_PoolStatsSampler = std::unique_ptr<statistics::PoolStatsSampler>(
        new statistics::PoolStatsSampler(getStatisticInterval(),
                                         m_connManager.getCacheImpl(), this));
    m_PoolStatsSampler->start();
  }

  // starting chunk processing helper thread
  ThinClientBaseDM::init();

  if (m_clientMetadataService) {
    m_clientMetadataService->start();
  }
}
void ThinClientPoolDM::manageConnections(std::atomic<bool>& isRunning) {
  LOGFINE("ThinClientPoolDM: starting manageConnections thread");

  while (isRunning) {
    m_connSema.acquire();
    if (isRunning) {
      try {
        LOGFINE(
            "ThinClientPoolDM::manageConnections: checking connections in "
            "pool");

        manageConnectionsInternal(isRunning);
      } catch (const Exception& e) {
        LOGERROR("ThinClientPoolDM::manageConnections: Geode Exception: \"%s\"",
                 e.what());
        LOGERROR(e.getStackTrace());
      } catch (const std::exception& e) {
        LOGERROR(
            "ThinClientPoolDM::manageConnections: Standard exception: \"%s\"",
            e.what());
      } catch (...) {
        LOGERROR("ThinClientPoolDM::manageConnections: Unexpected exception");
      }
    }
  }

  LOGFINE("ThinClientPoolDM: ending manageConnections thread");
}

void ThinClientPoolDM::cleanStaleConnections(std::atomic<bool>& isRunning) {
  if (!isRunning) {
    return;
  }

  LOGDEBUG("Cleaning stale connections");

  auto _idle = getIdleTimeout();
  auto _nextIdle = _idle;

  std::vector<TcrConnection*> removelist;
  std::set<ServerLocation> excludeServers;

  auto availableConns = size();
  auto savedConns = 0;

  for (unsigned int i = 0; (i < availableConns) && isRunning; i++) {
    auto* conn = getNoWait();
    if (conn == nullptr) {
      break;
    }
    if (canItBeDeleted(conn)) {
      removelist.push_back(conn);
    } else if (conn) {
      auto nextIdle =
          _idle -
          std::chrono::duration_cast<std::chrono::milliseconds>(
              std::chrono::steady_clock::now() - conn->getLastAccessed());
      if (nextIdle > std::chrono::seconds::zero() && nextIdle < _nextIdle) {
        _nextIdle = nextIdle;
      }
      put(conn, false);
      savedConns++;
    }
  }

  auto replaceCount = m_attrs->getMinConnections() - savedConns;

  LOGDEBUG("Preserving %d connections", savedConns);

  int count = 0;

  for (std::vector<TcrConnection*>::const_iterator iter = removelist.begin();
       iter != removelist.end(); ++iter) {
    TcrConnection* conn = *iter;
    if (replaceCount <= 0) {
      try {
        GF_SAFE_DELETE_CON(conn);
      } catch (...) {
      }
      removeEPConnections(1, false);
      getStats().incLoadCondDisconnects();
      LOGDEBUG("Removed a connection");
    } else {
      TcrConnection* newConn = nullptr;
      bool maxConnLimit = false;
      createPoolConnection(newConn, excludeServers, maxConnLimit,
                           /*hasExpired(conn) ? nullptr :*/ conn);
      if (newConn) {
        auto nextIdle =
            _idle -
            std::chrono::duration_cast<std::chrono::milliseconds>(
                std::chrono::steady_clock::now() - conn->getLastAccessed());
        if (nextIdle > std::chrono::seconds::zero() && nextIdle < _nextIdle) {
          _nextIdle = nextIdle;
        }
        put(newConn, false);
        if (newConn != conn) {
          try {
            GF_SAFE_DELETE_CON(conn);
          } catch (...) {
          }
          removeEPConnections(1, false);
          getStats().incLoadCondDisconnects();
          LOGDEBUG("Removed a connection");
        }
      } else {
        if (hasExpired(conn)) {
          try {
            GF_SAFE_DELETE_CON(conn);
          } catch (...) {
          }
          removeEPConnections(1, false);
          getStats().incLoadCondDisconnects();
          LOGDEBUG("Removed a connection");
        } else {
          conn->updateCreationTime();
          auto nextIdle =
              _idle -
              std::chrono::duration_cast<std::chrono::milliseconds>(
                  std::chrono::steady_clock::now() - conn->getLastAccessed());
          if (nextIdle > std::chrono::seconds::zero() && nextIdle < _nextIdle) {
            _nextIdle = nextIdle;
          }
          put(conn, false);
        }
      }
    }
    replaceCount--;
    count++;

    if (count % 10 == 0) {
      std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
  }

  if (m_connManageTaskId >= 0 && isRunning &&
      m_connManager.getCacheImpl()->getExpiryTaskManager().resetTask(
          m_connManageTaskId, _nextIdle)) {
    LOGERROR("Failed to reschedule connection manager");
  } else {
    LOGFINEST("Rescheduled next connection manager run after %s",
              to_string(_nextIdle).c_str());
  }

  LOGDEBUG("Pool size is %zu, pool counter is %d", size(), m_poolSize.load());
}

void ThinClientPoolDM::cleanStickyConnections(std::atomic<bool>&) {}

void ThinClientPoolDM::restoreMinConnections(std::atomic<bool>& isRunning) {
  if (!isRunning) {
    return;
  }

  LOGDEBUG("Restoring minimum connection level");

  int min = m_attrs->getMinConnections();
  int limit = 2 * min;

  std::set<ServerLocation> excludeServers;

  int restored = 0;

  if (m_poolSize < min) {
    while (m_poolSize < min && limit-- && isRunning) {
      TcrConnection* conn = nullptr;
      bool maxConnLimit = false;
      createPoolConnection(conn, excludeServers, maxConnLimit);
      if (conn) {
        put(conn, false);
        restored++;
        getStats().incMinPoolSizeConnects();
      }
    }
  }

  LOGDEBUG("Restored %d connections", restored);
  LOGDEBUG("Pool size is %zu, pool counter is %d", size(), m_poolSize.load());
}

void ThinClientPoolDM::manageConnectionsInternal(std::atomic<bool>& isRunning) {
  try {
    LOGFINE(
        "ThinClientPoolDM::manageConnections(): checking connections in pool "
        "queue %zu",
        size());

    cleanStaleConnections(isRunning);

    cleanStickyConnections(isRunning);

    restoreMinConnections(isRunning);

    getStats().setCurPoolConnections(m_poolSize);
  } catch (const Exception& e) {
    LOGERROR(e.what());
  } catch (const std::exception& e) {
    LOGERROR(e.what());
  } catch (...) {
    LOGERROR("Unexpected exception during manage connections");
  }
}

std::string ThinClientPoolDM::selectEndpoint(
    std::set<ServerLocation>& excludeServers,
    const TcrConnection* currentServer) {
  if (!m_attrs->m_initLocList.empty()) {  // query locators
    ServerLocation outEndpoint;
    std::string additionalLoc;
    LOGFINE("Asking locator for server from group [%s]",
            m_attrs->m_serverGrp.c_str());

    // Update Locator Request Stats
    getStats().incLoctorRequests();

    if (GF_NOERR != (m_locHelper)
                        ->getEndpointForNewFwdConn(
                            outEndpoint, additionalLoc, excludeServers,
                            m_attrs->m_serverGrp, currentServer)) {
      throw IllegalStateException("Locator query failed selecting an endpoint");
    }
    // Update Locator stats
    getStats().setLocators(
        static_cast<int32_t>(m_locHelper->getCurLocatorsNum()));
    getStats().incLoctorResposes();

    std::string epNameStr = outEndpoint.getServerName() + ":" +
                            std::to_string(outEndpoint.getPort());
    LOGFINE("ThinClientPoolDM: Locator returned endpoint [%s]",
            epNameStr.c_str());
    return epNameStr;
  } else if (!m_attrs->m_initServList.empty()) {
    // use specified server endpoints
    // highly complex round-robin algorithm
    std::lock_guard<decltype(m_endpointSelectionLock)> _guard(
        m_endpointSelectionLock);
    if (m_server >= m_attrs->m_initServList.size()) {
      m_server = 0;
    }

    unsigned int epCount = 0;
    do {
      if (!excludeServer(m_attrs->m_initServList[m_server], excludeServers)) {
        LOGFINE("ThinClientPoolDM: Selecting endpoint [%s] from position %d",
                m_attrs->m_initServList[m_server].c_str(), m_server);
        return m_attrs->m_initServList[m_server++];
      } else {
        if (++m_server >= m_attrs->m_initServList.size()) {
          m_server = 0;
        }
      }
    } while (++epCount < m_attrs->m_initServList.size());

    throw NotConnectedException("No server endpoints are available.");
  } else {
    LOGERROR("No locators or servers provided");
    throw IllegalStateException("No locators or servers provided");
  }
}

void ThinClientPoolDM::addConnection(TcrConnection* conn) {
  std::lock_guard<decltype(mutex_)> lock(mutex_);
  put(conn, false);
  ++m_poolSize;
}

GfErrType ThinClientPoolDM::sendRequestToAllServers(
    const char* func, uint8_t getResult, std::chrono::milliseconds timeout,
    std::shared_ptr<Cacheable> args, std::shared_ptr<ResultCollector>& rs,
    std::shared_ptr<CacheableString>& exceptionPtr) {
  getStats().setCurClientOps(++m_clientOps);

  auto resultCollectorLock = std::make_shared<std::recursive_mutex>();

  auto csArray = getServers();

  if (csArray != nullptr && csArray->length() == 0) {
    LOGWARN("No server found to execute the function");
    return GF_NOSERVER_FOUND;
  }

  std::vector<std::shared_ptr<FunctionExecution>> fePtrList;
  fePtrList.reserve(csArray->length());
  auto& threadPool = m_connManager.getCacheImpl()->getThreadPool();
  auto userAttr = UserAttributes::threadLocalUserAttributes;
  for (int i = 0; i < csArray->length(); i++) {
    auto cs = (*csArray)[i];
    auto endpointStr = cs->value();
    auto ep = getEndpoint(endpointStr);
    if (!ep) {
      ep = addEP(cs->value());
    } else if (!ep->connected()) {
      LOGFINE(
          "ThinClientPoolDM::sendRequestToAllServers server not connected "
          "%s ",
          cs->value().c_str());
    }
    auto funcExe = std::make_shared<FunctionExecution>();
    funcExe->setParameters(func, getResult, timeout, args, ep.get(), this,
                           resultCollectorLock, &rs, userAttr);
    fePtrList.push_back(funcExe);
    threadPool.perform(funcExe);
  }
  GfErrType finalErrorReturn = GF_NOERR;

  for (auto& funcExe : fePtrList) {
    auto err = funcExe->getResult();
    if (err != GF_NOERR) {
      if (funcExe->getException() == nullptr) {
        if (err == GF_TIMEOUT) {
          getStats().incTimeoutClientOps();
        } else {
          getStats().incFailedClientOps();
        }
        if (err == GF_IOERR) {
          err = GF_NOTCON;
        }
      } else {
        exceptionPtr = funcExe->getException();
      }
    }

    if (err == GF_AUTHENTICATION_FAILED_EXCEPTION ||
        err == GF_NOT_AUTHORIZED_EXCEPTION ||
        err == GF_AUTHENTICATION_REQUIRED_EXCEPTION) {
      finalErrorReturn = err;
    } else if ((err != GF_NOERR) &&
               (!(finalErrorReturn == GF_AUTHENTICATION_FAILED_EXCEPTION ||
                  finalErrorReturn == GF_NOT_AUTHORIZED_EXCEPTION ||
                  finalErrorReturn ==
                      GF_AUTHENTICATION_REQUIRED_EXCEPTION)))  // returning auth
                                                               // errors
    // to client..preference
    // over other errors..
    {
      finalErrorReturn = err;
    }
  }

  if (static_cast<uint8_t>(getResult & 2) == static_cast<uint8_t>(2)) {
    rs->endResults();
  }

  getStats().setCurClientOps(--m_clientOps);
  getStats().incSucceedClientOps();

  return finalErrorReturn;
}

const std::shared_ptr<CacheableStringArray> ThinClientPoolDM::getLocators()
    const {
  std::vector<std::shared_ptr<CacheableString>> locators;
  locators.reserve(m_attrs->m_initLocList.size());
  for (auto&& locator : m_attrs->m_initLocList) {
    locators.emplace_back(CacheableString::create(locator));
  }
  return CacheableStringArray::create(std::move(locators));
}

const std::shared_ptr<CacheableStringArray> ThinClientPoolDM::getServers() {
  std::vector<std::shared_ptr<CacheableString>> servers;
  if (!m_attrs->m_initServList.empty()) {
    servers.reserve(m_attrs->m_initServList.size());

    for (auto&& server : m_attrs->m_initServList) {
      servers.emplace_back(CacheableString::create(server));
    }
    return CacheableStringArray::create(std::move(servers));
  } else if (!m_attrs->m_initLocList.empty()) {
    std::vector<std::shared_ptr<ServerLocation>> vec;
    m_locHelper->getAllServers(vec, m_attrs->m_serverGrp);

    servers.reserve(vec.size());
    for (auto&& serLoc : vec) {
      servers.emplace_back(CacheableString::create(
          serLoc->getServerName() + ":" + std::to_string(serLoc->getPort())));
    }
    return CacheableStringArray::create(std::move(servers));
  }
  return CacheableStringArray::create(std::move(servers));
}

void ThinClientPoolDM::stopPingThread() {
  if (m_pingTask) {
    LOGFINE("ThinClientPoolDM::destroy(): Closing ping thread.");
    m_pingTask->stopNoblock();
    m_pingSema.release();
    m_pingTask->wait();
    m_pingTask = nullptr;
    if (m_pingTaskId >= 0) {
      m_connManager.getCacheImpl()->getExpiryTaskManager().cancelTask(
          m_pingTaskId);
    }
  }
}

void ThinClientPoolDM::stopUpdateLocatorListThread() {
  if (m_updateLocatorListTask) {
    LOGFINE("ThinClientPoolDM::destroy(): Closing updateLocatorList thread.");
    m_updateLocatorListTask->stopNoblock();
    m_updateLocatorListSema.release();
    m_updateLocatorListTask->wait();
    m_updateLocatorListTask = nullptr;
    if (m_updateLocatorListTaskId >= 0) {
      m_connManager.getCacheImpl()->getExpiryTaskManager().cancelTask(
          m_updateLocatorListTaskId);
    }
  }
}

void ThinClientPoolDM::stopCliCallbackThread() {
  if (m_cliCallbackTask) {
    LOGFINE("ThinClientPoolDM::destroy(): Closing cliCallback thread.");
    m_cliCallbackTask->stopNoblock();
    m_cliCallbackSema.release();
    m_cliCallbackTask->wait();
    m_cliCallbackTask = nullptr;
  }
}

void ThinClientPoolDM::destroy(bool keepAlive) {
  LOGDEBUG("ThinClientPoolDM::destroy...");
  if (!m_isDestroyed && (!m_destroyPending || m_destroyPendingHADM)) {
    checkRegions();
    TcrMessage::setKeepAlive(keepAlive);
    if (m_remoteQueryServicePtr != nullptr) {
      m_remoteQueryServicePtr->close();
      m_remoteQueryServicePtr = nullptr;
    }

    LOGDEBUG("Closing PoolStatsSampler thread.");
    if (m_PoolStatsSampler) {
      m_PoolStatsSampler->stop();
      m_PoolStatsSampler = nullptr;
    }
    LOGDEBUG("PoolStatsSampler thread closed .");
    stopCliCallbackThread();
    LOGDEBUG("ThinClientPoolDM::destroy( ): Closing connection manager.");
    auto cacheImpl = m_connManager.getCacheImpl();
    if (m_connManageTask) {
      m_connManageTask->stopNoblock();
      m_connSema.release();
      m_connManageTask->wait();
      m_connManageTask = nullptr;
      if (m_connManageTaskId >= 0) {
        cacheImpl->getExpiryTaskManager().cancelTask(m_connManageTaskId);
      }
    }

    LOGDEBUG("Closing PoolStatsSampler thread.");
    stopPingThread();
    stopUpdateLocatorListThread();

    if (m_clientMetadataService) {
      m_clientMetadataService->stop();
      // m_clientMetadataService = nullptr;
    }
    // closing all the thread local connections ( sticky).
    LOGDEBUG(
        "ThinClientPoolDM::destroy( ): closing ConnectionQueue, pool size = "
        "%d",
        m_poolSize.load());
    close();
    LOGDEBUG("ThinClientPoolDM::destroy( ): after close ");

    // Close Stats
    getStats().close();
    cacheImpl->getStatisticsManager().forceSample();

    cacheImpl->getPoolManager().removePool(m_poolName);

    stopChunkProcessor();
    m_manager->closeAllStickyConnections();
    m_isDestroyed = true;
    LOGDEBUG("ThinClientPoolDM::destroy( ): after close m_isDestroyed = %d ",
             m_isDestroyed);
  }
  if (m_poolSize != 0) {
    LOGFINE("Pool connection size is not zero %d", m_poolSize.load());
  }
}

bool ThinClientPoolDM::isDestroyed() const {
  // TODO: dummy implementation
  return m_isDestroyed;
}
std::shared_ptr<QueryService> ThinClientPoolDM::getQueryService() {
  // TODO:
  if (m_isMultiUserMode) {
    LOGERROR(
        "Pool is in multiuser authentication mode. Get query service using "
        "RegionService.getQueryService()");
    throw UnsupportedOperationException(
        "Pool is in multiuser authentication mode. Get QueryService() using "
        "RegionService.getQueryService()");
  }

  return getQueryServiceWithoutCheck();
}
std::shared_ptr<QueryService> ThinClientPoolDM::getQueryServiceWithoutCheck() {
  if (!(m_remoteQueryServicePtr == nullptr)) {
    return m_remoteQueryServicePtr;
  }

  LOGWARN("Remote query service is not initialized.");

  return m_remoteQueryServicePtr;
}
void ThinClientPoolDM::sendUserCacheCloseMessage(bool keepAlive) {
  LOGDEBUG("ThinClientPoolDM::sendUserCacheCloseMessage");
  auto userAttribute = UserAttributes::threadLocalUserAttributes;

  auto& ucs = userAttribute->getUserConnectionServers();

  for (const auto& it : ucs) {
    auto uca = it.second;
    if (uca->isAuthenticated() && uca->getEndpoint()->connected()) {
      TcrMessageRemoveUserAuth request(
          new DataOutput(m_connManager.getCacheImpl()->createDataOutput()),
          keepAlive, this);
      TcrMessageReply reply(true, this);

      sendRequestToEP(request, reply, uca->getEndpoint());

      uca->setUnAuthenticated();
    } else {
      uca->setUnAuthenticated();
    }
  }
}

int32_t ThinClientPoolDM::GetPDXIdForType(
    std::shared_ptr<Serializable> pdxType) {
  LOGDEBUG("ThinClientPoolDM::GetPDXIdForType:");

  TcrMessageGetPdxIdForType request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()), pdxType,
      this);

  TcrMessageReply reply(true, this);

  throwExceptionIfError("Operation Failed", sendSyncRequest(request, reply));

  if (reply.getMessageType() == TcrMessage::EXCEPTION) {
    LOGDEBUG("ThinClientPoolDM::GetPDXTypeById: Exception = " +
             reply.getException());
    throw IllegalStateException("Failed to register PdxSerializable Type");
  }

  auto pdxTypeId =
      std::dynamic_pointer_cast<CacheableInt32>(reply.getValue())->value();

  // need to broadcast this id to all other pool
  {
    auto& poolManager = m_connManager.getCacheImpl()->getPoolManager();
    for (const auto& iter : poolManager.getAll()) {
      auto currPool = static_cast<ThinClientPoolDM*>(iter.second.get());

      if (currPool != this) {
        currPool->AddPdxType(pdxType, pdxTypeId);
      }
    }
  }

  return pdxTypeId;
}

void ThinClientPoolDM::AddPdxType(std::shared_ptr<Serializable> pdxType,
                                  int32_t pdxTypeId) {
  LOGDEBUG("ThinClientPoolDM::GetPDXIdForType:");

  TcrMessageAddPdxType request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()), pdxType,
      this, pdxTypeId);

  TcrMessageReply reply(true, this);

  throwExceptionIfError("Operation Failed", sendSyncRequest(request, reply));

  if (reply.getMessageType() == TcrMessage::EXCEPTION) {
    LOGDEBUG("ThinClientPoolDM::GetPDXTypeById: Exception = " +
             reply.getException());
    throw IllegalStateException("Failed to register PdxSerializable Type");
  }
}
std::shared_ptr<Serializable> ThinClientPoolDM::GetPDXTypeById(int32_t typeId) {
  LOGDEBUG("ThinClientPoolDM::GetPDXTypeById:");

  TcrMessageGetPdxTypeById request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()), typeId,
      this);

  TcrMessageReply reply(true, this);

  throwExceptionIfError("Operation Failed", sendSyncRequest(request, reply));

  if (reply.getMessageType() == TcrMessage::EXCEPTION) {
    LOGDEBUG("ThinClientPoolDM::GetPDXTypeById: Exception = " +
             reply.getException());
    throw IllegalStateException("Failed to understand PdxSerializable Type");
  }

  return reply.getValue();
}

int32_t ThinClientPoolDM::GetEnumValue(std::shared_ptr<Serializable> enumInfo) {
  LOGDEBUG("ThinClientPoolDM::GetEnumValue:");

  TcrMessageGetPdxIdForEnum request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()),
      enumInfo, this);

  TcrMessageReply reply(true, this);

  throwExceptionIfError("Operation Failed", sendSyncRequest(request, reply));

  if (reply.getMessageType() == TcrMessage::EXCEPTION) {
    LOGDEBUG("ThinClientPoolDM::GetEnumValue: Exception = " +
             reply.getException());
    throw IllegalStateException("Failed to register Pdx enum Type");
  }

  auto enumVal =
      std::dynamic_pointer_cast<CacheableInt32>(reply.getValue())->value();

  // need to broadcast this id to all other pool
  {
    auto& poolManager = m_connManager.getCacheImpl()->getPoolManager();
    for (const auto& iter : poolManager.getAll()) {
      const auto& currPool =
          std::dynamic_pointer_cast<ThinClientPoolDM>(iter.second);

      if (currPool && this != currPool.get()) {
        currPool->AddEnum(enumInfo, enumVal);
      }
    }
  }

  return enumVal;
}
std::shared_ptr<Serializable> ThinClientPoolDM::GetEnum(int32_t val) {
  LOGDEBUG("ThinClientPoolDM::GetEnum:");

  TcrMessageGetPdxEnumById request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()), val,
      this);

  TcrMessageReply reply(true, this);

  throwExceptionIfError("Operation Failed", sendSyncRequest(request, reply));

  if (reply.getMessageType() == TcrMessage::EXCEPTION) {
    LOGDEBUG("ThinClientPoolDM::GetEnum: Exception = " + reply.getException());
    throw IllegalStateException("Failed to understand enum Type");
  }

  return reply.getValue();
}

void ThinClientPoolDM::AddEnum(std::shared_ptr<Serializable> enumInfo,
                               int enumVal) {
  LOGDEBUG("ThinClientPoolDM::AddEnum:");

  TcrMessageAddPdxEnum request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()),
      enumInfo, this, enumVal);

  TcrMessageReply reply(true, this);

  throwExceptionIfError("Operation Failed", sendSyncRequest(request, reply));

  if (reply.getMessageType() == TcrMessage::EXCEPTION) {
    LOGDEBUG("ThinClientPoolDM::AddEnum: Exception = " + reply.getException());
    throw IllegalStateException("Failed to register enum Type");
  }
}

GfErrType ThinClientPoolDM::sendUserCredentials(
    std::shared_ptr<Properties> credentials, TcrConnection*& conn,
    bool isBGThread, bool& isServerException) {
  LOGDEBUG("ThinClientPoolDM::sendUserCredentials:");

  TcrMessageUserCredential request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()),
      credentials, this);

  TcrMessageReply reply(true, this);

  auto err =
      conn->getEndpointObject()->sendRequestConnWithRetry(request, reply, conn);

  if (conn) {
    err = handleEPError(conn->getEndpointObject(), reply, err);
  }

  LOGDEBUG(
      "ThinClientPoolDM::sendUserCredentials: Error after sending cred "
      "request "
      "= %d ",
      err);

  if (err == GF_NOERR) {
    switch (reply.getMessageType()) {
      case TcrMessage::RESPONSE: {
        // nothing to be done;
        break;
      }
      case TcrMessage::EXCEPTION: {
        if (err == GF_NOERR && conn) {
          putInQueue(
              conn,
              isBGThread);  // connFound is only relevant for Sticky conn.
        }
        // this will set error type if there is some server exception
        err = ThinClientRegion::handleServerException(
            "ThinClientPoolDM::sendUserCredentials AuthException",
            reply.getException());
        isServerException = true;
        break;
      }
      default: {
        if (err == GF_NOERR && conn) {
          putInQueue(
              conn,
              isBGThread);  // connFound is only relevant for Sticky conn.
        }
        LOGERROR(
            "Unknown message type %d during secure response, possible "
            "serialization mismatch",
            reply.getMessageType());
        err = GF_MSG;

        break;
      }
    }
  }
  return err;
}

TcrEndpoint* ThinClientPoolDM::getSingleHopServer(
    TcrMessage& request, int8_t& version,
    std::shared_ptr<BucketServerLocation>& serverlocation,
    std::set<ServerLocation>& excludeServers) {
  const std::shared_ptr<CacheableKey>& key = request.getKeyRef();
  if (!m_clientMetadataService || key == nullptr) return nullptr;
  auto r = request.getRegion();
  auto region = nullptr == r ? nullptr : r->shared_from_this();
  TcrEndpoint* ep = nullptr;
  if (region == nullptr) {
    region = m_connManager.getCacheImpl()->getRegion(request.getRegionName());
  }
  if (region != nullptr) {
    m_clientMetadataService->getBucketServerLocation(
        region, key, request.getValueRef(), request.getCallbackArgumentRef(),
        request.forPrimary(), serverlocation, version);

    if (serverlocation != nullptr && serverlocation->isValid()) {
      LOGFINE("Server host and port are %s:%d",
              serverlocation->getServerName().c_str(),
              serverlocation->getPort());
      ep = getEndPoint(serverlocation, version, excludeServers);
    }
  }
  return ep;
}

TcrEndpoint* ThinClientPoolDM::getEndPoint(
    const std::shared_ptr<BucketServerLocation>& serverLocation, int8_t&,
    std::set<ServerLocation>& excludeServers) {
  std::shared_ptr<TcrEndpoint> ep = nullptr;
  if (serverLocation->isValid()) {
    if (excludeServer(serverLocation->getEpString(), excludeServers)) {
      LOGFINE("ThinClientPoolDM::getEndPoint Exclude Server true for %s ",
              serverLocation->getEpString().c_str());
      return ep.get();
    }

    ep = getEndpoint(serverLocation->getEpString());
    if (ep) {
      return ep.get();
    }

    // do for pool with endpoints. Add endpoint into m_endpoints only when we
    // did not find it above and it is in the pool's m_initServList.
    for (const auto& itr : m_attrs->m_initServList) {
      if (serverLocation->getEpString() == itr) {
        ep = addEP(*serverLocation);  // see if this is new endpoint
        break;
      }
    }

    // do only for locator
    // if servergroup is there, then verify otherwise you may reach to another
    // group
    if (!m_attrs->m_initLocList.empty()) {
      auto&& servGrp = getServerGroup();
      if (servGrp.length() > 0) {
        auto groups = serverLocation->getServerGroups();
        if ((groups != nullptr) && (groups->length() > 0)) {
          for (int i = 0; i < groups->length(); i++) {
            auto cs = (*groups)[i];
            if (cs->length() > 0) {
              auto&& str = cs->toString();
              if (str == servGrp) {
                // see if this is new endpoint
                ep = addEP(*serverLocation);
                break;
              }
            }
          }
        }
      } else  // just add it
      {
        ep = addEP(*(serverLocation.get()));  // see if this is new endpoint
      }
    }
  }

  return ep.get();
}

std::shared_ptr<TcrEndpoint> ThinClientPoolDM::getEndpoint(
    const std::string& endpointName) {
  m_endpoints.make_lock();
  const auto& find = m_endpoints.find(endpointName);
  if (find == m_endpoints.end()) {
    return {};
  }
  return find->second;
}

GfErrType ThinClientPoolDM::sendSyncRequest(TcrMessage& request,
                                            TcrMessageReply& reply,
                                            bool attemptFailover,
                                            bool isBGThread) {
  int32_t type = request.getMessageType();

  if (!request.forTransaction() && m_attrs->getPRSingleHopEnabled() &&
      (type == TcrMessage::GET_ALL_70 ||
       type == TcrMessage::GET_ALL_WITH_CALLBACK) &&
      m_clientMetadataService) {
    GfErrType error = GF_NOERR;

    auto region =
        m_connManager.getCacheImpl()->getRegion(request.getRegionName());
    auto locationMap = m_clientMetadataService->getServerToFilterMap(
        *(request.getKeys()), region, request.forPrimary());

    if (!locationMap) {
      request.InitializeGetallMsg(
          request.getCallbackArgument());  // now initialize getall msg
      return sendSyncRequest(request, reply, attemptFailover, isBGThread,
                             nullptr);
    }
    std::vector<std::shared_ptr<GetAllWork>> getAllWorkers;
    auto& threadPool = m_connManager.getCacheImpl()->getThreadPool();
    auto responseHandler =
        static_cast<ChunkedGetAllResponse*>(reply.getChunkedResultHandler());

    for (const auto& locationIter : *locationMap) {
      const auto& serverLocation = locationIter.first;
      const auto& keys = locationIter.second;
      auto worker = std::make_shared<GetAllWork>(
          this, region, serverLocation, keys, attemptFailover, isBGThread,
          responseHandler->getAddToLocalCache(), responseHandler,
          request.getCallbackArgument());
      threadPool.perform(worker);
      getAllWorkers.push_back(worker);
    }
    reply.setMessageType(TcrMessage::RESPONSE);

    for (auto& worker : getAllWorkers) {
      auto err = worker->getResult();

      if (err != GF_NOERR) {
        error = err;
      }

      TcrMessage* currentReply = worker->getReply();
      if (currentReply->getMessageType() != TcrMessage::RESPONSE) {
        reply.setMessageType(currentReply->getMessageType());
      }
    }
    return error;
  } else {
    if (type == TcrMessage::GET_ALL_70 ||
        type == TcrMessage::GET_ALL_WITH_CALLBACK) {
      request.InitializeGetallMsg(
          request.getCallbackArgument());  // now initialize getall msg
    }
    return sendSyncRequest(request, reply, attemptFailover, isBGThread,
                           nullptr);
  }
}

GfErrType ThinClientPoolDM::sendSyncRequest(
    TcrMessage& request, TcrMessageReply& reply, bool attemptFailover,
    bool isBGThread,
    const std::shared_ptr<BucketServerLocation>& serverLocation) {
  LOGDEBUG("ThinClientPoolDM::sendSyncRequest: ....%d %s",
           request.getMessageType(), m_poolName.c_str());
  // Increment clientOps
  getStats().setCurClientOps(++m_clientOps);

  GfErrType error = GF_NOTCON;

  std::shared_ptr<UserAttributes> userAttr = nullptr;
  reply.setDM(this);

  int32_t type = request.getMessageType();

  if (!(type == TcrMessage::QUERY ||
        type == TcrMessage::QUERY_WITH_PARAMETERS ||
        type == TcrMessage::PUTALL ||
        type == TcrMessage::PUT_ALL_WITH_CALLBACK ||
        type == TcrMessage::EXECUTE_FUNCTION ||
        type == TcrMessage::EXECUTE_REGION_FUNCTION ||
        type == TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP ||
        type == TcrMessage::EXECUTECQ_WITH_IR_MSG_TYPE)) {
    // set only when message is not query, putall and executeCQ
    reply.setTimeout(getReadTimeout());
    request.setTimeout(getReadTimeout());
  }

  bool retryAllEPsOnce = false;
  if (m_attrs->getRetryAttempts() == -1) {
    retryAllEPsOnce = true;
  }
  auto retry = m_attrs->getRetryAttempts() + 1;
  TcrConnection* conn = nullptr;
  std::set<ServerLocation> excludeServers;
  type = request.getMessageType();
  bool isAuthRequireExcep = false;
  int isAuthRequireExcepMaxTry = 2;
  bool firstTry = true;
  LOGFINE("sendSyncRequest:: retry = %d", retry);
  while (retryAllEPsOnce || retry-- ||
         (isAuthRequireExcep && isAuthRequireExcepMaxTry >= 0)) {
    isAuthRequireExcep = false;
    if (!firstTry) request.updateHeaderForRetry();
    // if it's a query or putall and we had a timeout, just return with the
    // newly selected endpoint without failover-retry
    if ((type == TcrMessage::QUERY ||
         type == TcrMessage::QUERY_WITH_PARAMETERS ||
         type == TcrMessage::PUTALL ||
         type == TcrMessage::PUT_ALL_WITH_CALLBACK ||
         type == TcrMessage::EXECUTE_FUNCTION ||
         type == TcrMessage::EXECUTE_REGION_FUNCTION ||
         type == TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP ||
         type == TcrMessage::EXECUTECQ_WITH_IR_MSG_TYPE) &&
        error == GF_TIMEOUT) {
      return error;
    }

    GfErrType queueErr = GF_NOERR;
    auto lastExcludeSize = static_cast<uint32_t>(excludeServers.size());
    int8_t version = 0;

    bool isUserNeedToReAuthenticate = false;
    bool singleHopConnFound = false;
    bool connFound = false;
    if (!m_isMultiUserMode || (!TcrMessage::isUserInitiativeOps(request))) {
      conn = getConnectionFromQueueW(&queueErr, excludeServers, isBGThread,
                                     request, version, singleHopConnFound,
                                     connFound, serverLocation);
    } else {
      userAttr = UserAttributes::threadLocalUserAttributes;
      if (userAttr == nullptr) {
        LOGWARN("Attempted operation type %d without credentials",
                request.getMessageType());
        return GF_NOT_AUTHORIZED_EXCEPTION;
      }
      // Can i assume here that we will always get connection here
      conn = getConnectionFromQueueW(&queueErr, excludeServers, isBGThread,
                                     request, version, singleHopConnFound,
                                     connFound, serverLocation);

      if (conn != nullptr) {  // need to chk whether user is already
                              // authenticated to this endpoint or not.
        isUserNeedToReAuthenticate =
            !(userAttr->isEndpointAuthenticated(conn->getEndpointObject()));
      }
    }

    if (queueErr == GF_CLIENT_WAIT_TIMEOUT) {
      LOGFINE("Request timeout at client only");
      return GF_CLIENT_WAIT_TIMEOUT;
    } else if (queueErr == GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA) {
      // need to refresh meta data
      auto region =
          m_connManager.getCacheImpl()->getRegion(request.getRegionName());

      if (region != nullptr) {
        LOGFINE(
            "Need to refresh pr-meta-data timeout in client only  with "
            "refresh "
            "metadata");
        auto* tcrRegion = dynamic_cast<ThinClientRegion*>(region.get());
        tcrRegion->setMetaDataRefreshed(false);
        m_clientMetadataService->enqueueForMetadataRefresh(
            region->getFullPath(), reply.getserverGroupVersion());
      }
      return GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA;
    }

    LOGDEBUG(
        "ThinClientPoolDM::sendSyncRequest: isUserNeedToReAuthenticate = %d ",
        isUserNeedToReAuthenticate);
    LOGDEBUG(
        "ThinClientPoolDM::sendSyncRequest: m_isMultiUserMode = %d  conn = "
        "%p  "
        "type = %d",
        m_isMultiUserMode, conn, type);

    if (!conn) {
      // lets assume all connection are in use will happen
      if (queueErr == GF_NOERR) {
        queueErr = GF_ALL_CONNECTIONS_IN_USE_EXCEPTION;
        getStats().setCurClientOps(--m_clientOps);
        getStats().incFailedClientOps();
        return queueErr;
      } else if (queueErr == GF_IOERR) {
        error = GF_NOTCON;
      } else {
        error = queueErr;
      }
    }
    if (conn) {
      TcrEndpoint* ep = conn->getEndpointObject();
      LOGDEBUG(
          "ThinClientPoolDM::sendSyncRequest: sendSyncReq "
          "ep->isAuthenticated() = %d ",
          ep->isAuthenticated());
      GfErrType userCredMsgErr = GF_NOERR;
      bool isServerException = false;
      if (TcrMessage::isUserInitiativeOps(request) &&
          (m_isSecurityOn || m_isMultiUserMode)) {
        if (!m_isMultiUserMode && !ep->isAuthenticated()) {
          // first authenticate him on this endpoint
          userCredMsgErr = sendUserCredentials(getCredentials(ep), conn,
                                               isBGThread, isServerException);
        } else if (isUserNeedToReAuthenticate) {
          userCredMsgErr = sendUserCredentials(userAttr->getCredentials(), conn,
                                               isBGThread, isServerException);
        }
      }

      if (userCredMsgErr == GF_NOERR) {
        error = ep->sendRequestConnWithRetry(request, reply, conn);
        error = handleEPError(ep, reply, error);
      } else {
        error = userCredMsgErr;
      }

      if (!isServerException) {
        if (error == GF_NOERR) {
          LOGDEBUG("putting connection back in queue");
          putInQueue(conn,
                     isBGThread ||
                         request.getMessageType() == TcrMessage::GET_ALL_70 ||
                         request.getMessageType() ==
                             TcrMessage::GET_ALL_WITH_CALLBACK ||
                         request.getMessageType() ==
                             TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP,
                     request.forTransaction());  // connFound is only relevant
                                                 // for Sticky conn.
          LOGDEBUG("putting connection back in queue DONE");
        } else {
          if (error != GF_TIMEOUT) removeEPConnections(ep);
          // Update stats for the connection that failed.
          removeEPConnections(1, false);
          setStickyNull(isBGThread ||
                        request.getMessageType() == TcrMessage::GET_ALL_70 ||
                        request.getMessageType() ==
                            TcrMessage::GET_ALL_WITH_CALLBACK ||
                        request.getMessageType() ==
                            TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP);
          if (conn) {
            try {
              GF_SAFE_DELETE_CON(conn);
            } catch (...) {
            }
          }
          excludeServers.insert(ServerLocation(ep->name()));
          removeEPFromMetadataIfError(error, ep);
        }
      } else {
        return error;  // server exception while sending credential message to
      }
      // server...
    }

    if (error == GF_NOERR) {
      if ((m_isSecurityOn || m_isMultiUserMode)) {
        if (reply.getMessageType() == TcrMessage::EXCEPTION) {
          if (isAuthRequireException(reply.getException())) {
            TcrEndpoint* ep = conn->getEndpointObject();
            if (!m_isMultiUserMode) {
              ep->setAuthenticated(false);
            } else if (userAttr != nullptr) {
              userAttr->unAuthenticateEP(ep);
            }
            LOGFINEST(
                "After getting AuthenticationRequiredException trying "
                "again.");
            isAuthRequireExcepMaxTry--;
            isAuthRequireExcep = true;
            continue;
          } else if (isNotAuthorizedException(reply.getException())) {
            LOGDEBUG("received NotAuthorizedException");
            // TODO should we try again?
          }
        }
      }
      LOGFINER(
          "reply Metadata version is %d & bsl version is %d "
          "reply.isFEAnotherHop()=%d",
          reply.getMetaDataVersion(), version, reply.isFEAnotherHop());
      if (m_clientMetadataService && request.forSingleHop() &&
          (reply.getMetaDataVersion() != 0 ||
           (request.getMessageType() == TcrMessage::EXECUTE_REGION_FUNCTION &&
            request.getKeyRef() != nullptr && reply.isFEAnotherHop()))) {
        // Need to get direct access to Region's name to avoid referencing
        // temp data and causing crashes
        auto region =
            m_connManager.getCacheImpl()->getRegion(request.getRegionName());

        if (region != nullptr) {
          if (!connFound)  // max limit case then don't refresh otherwise
                           // always refresh
          {
            LOGFINE("Need to refresh pr-meta-data");
            auto* tcrRegion = dynamic_cast<ThinClientRegion*>(region.get());
            tcrRegion->setMetaDataRefreshed(false);
          }
          m_clientMetadataService->enqueueForMetadataRefresh(
              region->getFullPath(), reply.getserverGroupVersion());
        }
      }
    }

    if (excludeServers.size() == lastExcludeSize) {
      excludeServers.clear();
      if (retryAllEPsOnce) {
        break;
      }
    }

    if (!attemptFailover || error == GF_NOERR) {
      getStats().setCurClientOps(--m_clientOps);
      if (error == GF_NOERR) {
        getStats().incSucceedClientOps(); /*inc Id for clientOs stat*/
      } else if (error == GF_TIMEOUT) {
        getStats().incTimeoutClientOps();
      } else {
        getStats().incFailedClientOps();
      }
      // Top-level only sees NotConnectedException
      if (error == GF_IOERR) {
        error = GF_NOTCON;
      }
      return error;
    }

    conn = nullptr;
    firstTry = false;
  }  // While

  getStats().setCurClientOps(--m_clientOps);

  if (error == GF_NOERR) {
    getStats().incSucceedClientOps();
  } else if (error == GF_TIMEOUT) {
    getStats().incTimeoutClientOps();
  } else {
    getStats().incFailedClientOps();
  }

  // Top-level only sees NotConnectedException
  if (error == GF_IOERR) {
    error = GF_NOTCON;
  }
  return error;
}

void ThinClientPoolDM::removeEPFromMetadataIfError(const GfErrType& error,
                                                   const TcrEndpoint* ep) {
  if ((error == GF_IOERR || error == GF_TIMEOUT) && (m_clientMetadataService)) {
    auto sl = std::make_shared<BucketServerLocation>(ep->name());
    LOGINFO("Removing bucketServerLocation %s due to %s",
            sl->toString().c_str(),
            (error == GF_IOERR ? "GF_IOERR" : "GF_TIMEOUT"));
    m_clientMetadataService->removeBucketServerLocation(sl);
  }
}

void ThinClientPoolDM::removeEPConnections(int numConn,
                                           bool triggerManageConn) {
  // TODO: Delete EP

  reducePoolSize(numConn);

  // Raise Semaphore for manage thread
  if (triggerManageConn) {
    m_connSema.release();
  }
}

// Tries to get connection to a endpoint. If no connection is available, it
// tries
// to create one. If it fails to create one,  it gets connection to any other
// server
// and fails over the transaction to that server.
// This function is used when the transaction is to be resumed to a specified
// server.
GfErrType ThinClientPoolDM::getConnectionToAnEndPoint(std::string epNameStr,
                                                      TcrConnection*& conn) {
  conn = nullptr;

  GfErrType error = GF_NOERR;
  auto theEP = getEndpoint(epNameStr);

  LOGFINE(
      "ThinClientPoolDM::getConnectionToAnEndPoint( ): Getting endpoint "
      "object "
      "for %s",
      epNameStr.c_str());
  if (theEP && theEP->connected()) {
    LOGFINE(
        "ThinClientPoolDM::getConnectionToAnEndPoint( ): Getting connection "
        "for endpoint %s",
        epNameStr.c_str());
    conn = getFromEP(theEP.get());
    // if connection is null, possibly because there are no idle connections
    // to this endpoint, create a new pool connection to this endpoint.
    bool maxConnLimit = false;
    if (conn == nullptr) {
      LOGFINE(
          "ThinClientPoolDM::getConnectionToAnEndPoint( ): Create connection "
          "for endpoint %s",
          epNameStr.c_str());
      error = createPoolConnectionToAEndPoint(conn, theEP.get(), maxConnLimit);
    }
  }

  // if connection is null, it has failed to get a connection to the specified
  // endpoint. Get a connection to any other server and failover the
  // transaction to that server.
  if (!conn) {
    std::set<ServerLocation> excludeServers;
    bool maxConnLimit = false;
    LOGFINE(
        "ThinClientPoolDM::getConnectionToAnEndPoint( ): No connection "
        "available for endpoint %s. Create connection to any endpoint.",
        epNameStr.c_str());
    conn = getConnectionFromQueue(true, &error, excludeServers, maxConnLimit);
    if (conn && error == GF_NOERR) {
      if (conn->getEndpointObject()->name() != epNameStr) {
        LOGFINE(
            "ThinClientPoolDM::getConnectionToAnEndPoint( ): Endpoint %s "
            "different than the endpoint %s. New connection created and "
            "failing over.",
            epNameStr.c_str(), conn->getEndpointObject()->name().c_str());
        auto failoverErr = doFailover(conn);
        if (failoverErr != GF_NOERR) {
          LOGFINE(
              "ThinClientPoolDM::getConnectionToAnEndPoint( ):Failed to "
              "failover transaction to another server. From endpoint %s to "
              "%s",
              epNameStr.c_str(), conn->getEndpointObject()->name().c_str());
          putInQueue(conn, false);
          conn = nullptr;
        }
      }
    }
  }

  if (!(conn && error == GF_NOERR)) {
    LOGFINE(
        "ThinClientPoolDM::getConnectionToAEndPoint( ):Failed to connect to "
        "%s",
        epNameStr.c_str());
    if (conn) {
      _GEODE_SAFE_DELETE(conn);
    }
  }

  return error;
}

// Create a pool connection to specified endpoint. First checks if the number
// of connections has exceeded the maximum allowed. If not, create a
// connection to the specified endpoint. Else, throws an error.
GfErrType ThinClientPoolDM::createPoolConnectionToAEndPoint(
    TcrConnection*& conn, TcrEndpoint* theEP, bool& maxConnLimit,
    bool appThreadrequest) {
  GfErrType error = GF_NOERR;
  conn = nullptr;
  int min = 0;
  {
    // Check if the pool size has exceeded maximum allowed.

    int max = m_attrs->getMaxConnections();
    if (max == -1) {
      max = 0x7fffffff;
    }
    min = m_attrs->getMinConnections();
    max = max > min ? max : min;

    if (m_poolSize >= max) {
      maxConnLimit = true;
      LOGFINER(
          "ThinClientPoolDM::createPoolConnectionToAEndPoint( ): current "
          "pool "
          "size has reached limit %d, %d",
          m_poolSize.load(), max);
      return error;
    }
  }

  LOGFINE(
      "ThinClientPoolDM::createPoolConnectionToAEndPoint( ): creating a new "
      "connection to the endpoint %s",
      theEP->name().c_str());
  // if the pool size is within limits, create a new connection.
  error = theEP->createNewConnection(conn, false, false,
                                     m_connManager.getCacheImpl()
                                         ->getDistributedSystem()
                                         .getSystemProperties()
                                         .connectTimeout(),
                                     false, appThreadrequest);
  if (conn == nullptr || error != GF_NOERR) {
    LOGFINE("2Failed to connect to %s", theEP->name().c_str());
    if (conn != nullptr) _GEODE_SAFE_DELETE(conn);
  } else {
    theEP->setConnected();
    ++m_poolSize;
    if (m_poolSize > min) {
      getStats().incLoadCondConnects();
    }
    // Update Stats
    getStats().incPoolConnects();
    getStats().setCurPoolConnections(m_poolSize);
  }
  m_connSema.release();

  return error;
}

void ThinClientPoolDM::reducePoolSize(int num) {
  LOGFINE("removing connection %d ,  pool-size =%d", num, m_poolSize.load());
  m_poolSize -= num;
  if (m_poolSize <= 0) {
    if (m_cliCallbackTask != nullptr) m_cliCallbackSema.release();
  }
}

GfErrType ThinClientPoolDM::createPoolConnection(
    TcrConnection*& conn, std::set<ServerLocation>& excludeServers,
    bool& maxConnLimit, const TcrConnection* currentserver) {
  GfErrType error = GF_NOERR;
  int max = m_attrs->getMaxConnections();
  if (max == -1) {
    max = 0x7fffffff;
  }
  int min = m_attrs->getMinConnections();
  max = max > min ? max : min;

  conn = nullptr;
  if (m_poolSize >= max) {
    maxConnLimit = true;
    return error;
  }

  bool fatal = false;
  GfErrType fatalError = GF_NOERR;

  while (true) {
    std::string epNameStr;
    try {
      epNameStr = selectEndpoint(excludeServers, currentserver);
    } catch (const NoAvailableLocatorsException&) {
      LOGFINE("Locator query failed while creating pool connection");
      return GF_CACHE_LOCATOR_EXCEPTION;
    } catch (const Exception&) {
      LOGFINE("Endpoint selection failed");
      return GF_NOTCON;
    }

    LOGFINE("Connecting to %s", epNameStr.c_str());
    auto ep = addEP(epNameStr);

    if (currentserver != nullptr &&
        epNameStr == currentserver->getEndpointObject()->name()) {
      LOGDEBUG("Updating existing connection: ", epNameStr.c_str());
      conn = const_cast<TcrConnection*>(currentserver);
      conn->updateCreationTime();
      break;
    } else {
      error = ep->createNewConnection(conn, false, false,
                                      m_connManager.getCacheImpl()
                                          ->getDistributedSystem()
                                          .getSystemProperties()
                                          .connectTimeout(),
                                      false);
    }

    if (conn == nullptr || error != GF_NOERR) {
      LOGFINE("1Failed to connect to %s", epNameStr.c_str());
      excludeServers.insert(ServerLocation(ep->name()));
      if (conn != nullptr) {
        _GEODE_SAFE_DELETE(conn);
      }
      if (ThinClientBaseDM::isFatalError(error)) {
        // save this error for later to override the
        // error code to be returned
        fatalError = error;
        fatal = true;
      }
      if (ThinClientBaseDM::isFatalClientError(error)) {
        //  log the error string instead of error number.
        LOGFINE("Connection failed due to fatal client error %d", error);
        return error;
      }
    } else {
      ep->setConnected();
      if (++m_poolSize > min) {
        getStats().incLoadCondConnects();
      }
      // Update Stats
      getStats().incPoolConnects();
      getStats().setCurPoolConnections(m_poolSize);
      break;
    }
  }
  m_connSema.release();
  // if a fatal error occurred earlier and we don't have
  // a connection then return this saved error
  if (fatal && !conn && error != GF_NOERR) {
    return fatalError;
  }

  return error;
}

TcrConnection* ThinClientPoolDM::getConnectionFromQueue(
    bool, GfErrType* error, std::set<ServerLocation>& excludeServers,
    bool& maxConnLimit) {
  std::chrono::microseconds timeoutTime = m_attrs->getFreeConnectionTimeout();

  getStats().incCurWaitingConnections();
  getStats().incWaitingConnections();

  /*get the start time for connectionWaitTime stat*/
  bool enableTimeStatistics = m_connManager.getCacheImpl()
                                  ->getDistributedSystem()
                                  .getSystemProperties()
                                  .getEnableTimeStatistics();
  auto sampleStartNanos = enableTimeStatistics ? Utils::startStatOpTime() : 0;
  auto mp = getUntil(timeoutTime, error, excludeServers, maxConnLimit);
  /*Update the time stat for clientOpsTime */
  if (enableTimeStatistics) {
    Utils::updateStatOpTime(getStats().getStats(),
                            getStats().getTotalWaitingConnTimeId(),
                            sampleStartNanos);
  }
  getStats().decCurWaitingConnections();
  return mp;
}

bool ThinClientPoolDM::isEndpointAttached(TcrEndpoint*) { return true; }

GfErrType ThinClientPoolDM::sendRequestToEP(const TcrMessage& request,
                                            TcrMessageReply& reply,
                                            TcrEndpoint* currentEndpoint) {
  LOGDEBUG("ThinClientPoolDM::sendRequestToEP()");
  int isAuthRequireExcepMaxTry = 2;
  bool isAuthRequireExcep = true;
  GfErrType error = GF_NOERR;
  while (isAuthRequireExcep && isAuthRequireExcepMaxTry >= 0) {
    isAuthRequireExcep = false;
    TcrConnection* conn = getFromEP(currentEndpoint);

    bool isTmpConnectedStatus = false;
    bool putConnInPool = true;
    if (conn == nullptr) {
      LOGDEBUG(
          "ThinClientPoolDM::sendRequestToEP(): got nullptr connection from "
          "pool, "
          "creating new connection in the pool.");
      bool maxConnLimit = false;
      error =
          createPoolConnectionToAEndPoint(conn, currentEndpoint, maxConnLimit);
      if (conn == nullptr || error != GF_NOERR) {
        LOGDEBUG(
            "ThinClientPoolDM::sendRequestToEP(): couldnt create a pool "
            "connection, creating a temporary connection.");
        error =
            currentEndpoint->createNewConnection(conn, false, false,
                                                 m_connManager.getCacheImpl()
                                                     ->getDistributedSystem()
                                                     .getSystemProperties()
                                                     .connectTimeout(),
                                                 false);
        putConnInPool = false;
        currentEndpoint->setConnectionStatus(true);
      }
    }

    if (conn == nullptr || error != GF_NOERR) {
      LOGFINE("3Failed to connect to %s", currentEndpoint->name().c_str());
      if (conn != nullptr) {
        _GEODE_SAFE_DELETE(conn);
      }
      if (putConnInPool) {
        reducePoolSize(1);
      }
      currentEndpoint->setConnectionStatus(false);
      return error;
    } else if (!putConnInPool && !currentEndpoint->connected()) {
      isTmpConnectedStatus = true;
      currentEndpoint->setConnectionStatus(true);
    }

    int32_t type = request.getMessageType();

    if (!(type == TcrMessage::QUERY || type == TcrMessage::PUTALL ||
          type == TcrMessage::PUT_ALL_WITH_CALLBACK ||
          type == TcrMessage::EXECUTE_FUNCTION ||
          type == TcrMessage::EXECUTE_REGION_FUNCTION ||
          type == TcrMessage::EXECUTE_REGION_FUNCTION_SINGLE_HOP ||
          type == TcrMessage::EXECUTECQ_WITH_IR_MSG_TYPE)) {
      reply.setTimeout(getReadTimeout());
    }

    reply.setDM(this);
    std::shared_ptr<UserAttributes> ua = nullptr;
    // in multi user mode need to chk whether user is authenticated or not
    // and then follow usual process which we did in send syncrequest.
    // need to user initiative ops
    LOGDEBUG("ThinClientPoolDM::sendRequestToEP: m_isMultiUserMode = %d",
             m_isMultiUserMode);
    bool isServerException = false;
    if (TcrMessage::isUserInitiativeOps((request)) &&
        (m_isSecurityOn || m_isMultiUserMode)) {
      if (!m_isMultiUserMode && !currentEndpoint->isAuthenticated()) {
        // first authenticate him on this endpoint
        error = sendUserCredentials(getCredentials(currentEndpoint), conn,
                                    false, isServerException);
      } else if (m_isMultiUserMode) {
        ua = UserAttributes::threadLocalUserAttributes;
        if (ua) {
          UserConnectionAttributes* uca =
              ua->getConnectionAttribute(currentEndpoint);

          if (uca == nullptr) {
            error = sendUserCredentials(ua->getCredentials(), conn, false,
                                        isServerException);
          }
        } else {
          LOGWARN("Attempted operation type %d without credentials",
                  request.getMessageType());
          if (conn) {
            putInQueue(conn, false, request.forTransaction());
          }
          return GF_NOT_AUTHORIZED_EXCEPTION;
        }
      }
    }

    LOGDEBUG("ThinClientPoolDM::sendRequestToEP after getting creds");
    if (error == GF_NOERR && conn != nullptr) {
      error =
          currentEndpoint->sendRequestConnWithRetry(request, reply, conn, true);
    }

    if (isServerException) return error;

    if (error == GF_NOERR) {
      int32_t replyMsgType = reply.getMessageType();
      if (replyMsgType == TcrMessage::EXCEPTION ||
          replyMsgType == TcrMessage::CQ_EXCEPTION_TYPE ||
          replyMsgType == TcrMessage::CQDATAERROR_MSG_TYPE) {
        error = ThinClientRegion::handleServerException(
            "ThinClientPoolDM::sendRequestToEP", reply.getException());
      }

      if (putConnInPool) {
        put(conn, false);
      } else {
        if (isTmpConnectedStatus) currentEndpoint->setConnectionStatus(false);
        conn->close();
        _GEODE_SAFE_DELETE(conn);
      }
    } else if (error != GF_NOERR) {
      currentEndpoint->setConnectionStatus(false);
      if (putConnInPool) {
        removeEPConnections(1);
      }
      removeEPFromMetadataIfError(error, currentEndpoint);
    }

    if (error == GF_NOERR || error == GF_CACHESERVER_EXCEPTION ||
        error == GF_AUTHENTICATION_REQUIRED_EXCEPTION) {
      if ((m_isSecurityOn || m_isMultiUserMode)) {
        if (reply.getMessageType() == TcrMessage::EXCEPTION) {
          if (isAuthRequireException(reply.getException())) {
            if (!m_isMultiUserMode) {
              currentEndpoint->setAuthenticated(false);
            } else if (ua != nullptr) {
              ua->unAuthenticateEP(currentEndpoint);
            }
            LOGFINEST(
                "After getting AuthenticationRequiredException trying "
                "again.");
            isAuthRequireExcepMaxTry--;
            isAuthRequireExcep = true;
            if (isAuthRequireExcepMaxTry >= 0) error = GF_NOERR;
            continue;
          }
        }
      }
    }
  }
  LOGDEBUG("ThinClientPoolDM::sendRequestToEP Done.");
  return error;
}

std::shared_ptr<TcrEndpoint> ThinClientPoolDM::addEP(
    ServerLocation& serverLoc) {
  return addEP(serverLoc.getServerName() + ":" +
               std::to_string(serverLoc.getPort()));
}

std::shared_ptr<TcrEndpoint> ThinClientPoolDM::addEP(
    const std::string& endpointName) {
  std::lock_guard<decltype(m_endpointsLock)> guard(m_endpointsLock);

  auto ep = getEndpoint(endpointName);
  if (!ep) {
    ep = createEP(endpointName.c_str());
    LOGFINE("Created new endpoint %s for pool %s", endpointName.c_str(),
            m_poolName.c_str());
    if (!m_endpoints.emplace(endpointName, ep).second) {
      LOGERROR("Failed to add endpoint %s to pool %s", endpointName.c_str(),
               m_poolName.c_str());
    }
  }

  // Update Server Stats
  getStats().setServers(static_cast<int32_t>(m_endpoints.size()));
  return ep;
}

void ThinClientPoolDM::netDown() {
  close();
  reset();
}

void ThinClientPoolDM::pingServerLocal() {
  std::lock_guard<decltype(m_endpointsLock)> guard(m_endpointsLock);
  for (auto& it : m_endpoints) {
    auto endpoint = it.second;
    if (endpoint->connected()) {
      endpoint->pingServer(this);
      if (!endpoint->connected()) {
        removeEPConnections(endpoint.get());
        removeCallbackConnection(endpoint.get());
      }
    }
  }
}

void ThinClientPoolDM::updateLocatorList(std::atomic<bool>& isRunning) {
  LOGFINE("Starting updateLocatorList thread for pool %s", m_poolName.c_str());

  while (isRunning) {
    m_updateLocatorListSema.acquire();
    if (isRunning && !m_connManager.isNetDown()) {
      (m_locHelper)->updateLocators(getServerGroup());
    }
  }

  LOGFINE("Ending updateLocatorList thread for pool %s", m_poolName.c_str());
}

void ThinClientPoolDM::pingServer(std::atomic<bool>& isRunning) {
  LOGFINE("Starting ping thread for pool %s", m_poolName.c_str());

  while (isRunning) {
    m_pingSema.acquire();
    if (isRunning && !m_connManager.isNetDown()) {
      pingServerLocal();
      m_pingSema.acquire();
    }
  }

  LOGFINE("Ending ping thread for pool %s", m_poolName.c_str());
}

void ThinClientPoolDM::cliCallback(std::atomic<bool>& isRunning) {
  LOGFINE("Starting cliCallback thread for pool %s", m_poolName.c_str());
  while (isRunning) {
    m_cliCallbackSema.acquire();
    if (isRunning) {
      LOGFINE("Clearing Pdx Type Registry");
      // this call for csharp client
      DistributedSystemImpl::CallCliCallBack(
          *(m_connManager.getCacheImpl()->getCache()));
      // this call for cpp client
      m_connManager.getCacheImpl()->getPdxTypeRegistry()->clear();
      m_cliCallbackSema.acquire();
    }
  }
  LOGFINE("Ending cliCallback thread for pool %s", m_poolName.c_str());
}

int ThinClientPoolDM::doPing(const ACE_Time_Value&, const void*) {
  m_pingSema.release();
  return 0;
}

int ThinClientPoolDM::doUpdateLocatorList(const ACE_Time_Value&, const void*) {
  m_updateLocatorListSema.release();
  return 0;
}

int ThinClientPoolDM::doManageConnections(const ACE_Time_Value&, const void*) {
  m_connSema.release();
  return 0;
}

void ThinClientPoolDM::releaseThreadLocalConnection() {
  m_manager->releaseThreadLocalConnection();
}
void ThinClientPoolDM::setThreadLocalConnection(TcrConnection* conn) {
  m_manager->addStickyConnection(conn);
}

inline bool ThinClientPoolDM::hasExpired(TcrConnection* conn) {
  return conn->hasExpired(getLoadConditioningInterval());
}

bool ThinClientPoolDM::canItBeDeleted(TcrConnection* conn) {
  const auto& load = getLoadConditioningInterval();
  auto idle = getIdleTimeout();
  int min = getMinConnections();

  if (load > std::chrono::milliseconds::zero()) {
    if (load < idle || idle <= std::chrono::milliseconds::zero()) {
      idle = load;
    }
  }

  bool hasExpired = conn->hasExpired(load);
  bool isIdle = conn->isIdle(idle);

  bool candidateForDeletion = hasExpired || (isIdle && m_poolSize > min);
  bool canBeDeleted = false;

  if (conn && candidateForDeletion) {
    TcrEndpoint* endPt = conn->getEndpointObject();
    {
      std::lock_guard<decltype(endPt->getQueueHostedMutex())> guardQueue(
          endPt->getQueueHostedMutex());
      bool queue = endPt->isQueueHosted();
      if (queue) {
        TcrConnection* connTemp = getFromEP(endPt);
        if (connTemp) {
          put(connTemp, false);
          canBeDeleted = true;
        }
      } else {
        canBeDeleted = true;
      }
    }
  }

  return canBeDeleted;
}

bool ThinClientPoolDM::excludeServer(std::string endpoint,
                                     std::set<ServerLocation>& excludeServers) {
  return !(excludeServers.empty() || excludeServers.find(ServerLocation(
                                         endpoint)) == excludeServers.end());
}

bool ThinClientPoolDM::excludeConnection(
    TcrConnection* conn, std::set<ServerLocation>& excludeServers) {
  return excludeServer(conn->getEndpointObject()->name(), excludeServers);
}

TcrConnection* ThinClientPoolDM::getFromEP(TcrEndpoint* theEP) {
  std::lock_guard<decltype(mutex_)> lock(mutex_);
  for (auto itr = queue_.begin(); itr != queue_.end(); itr++) {
    if ((*itr)->getEndpointObject() == theEP) {
      LOGDEBUG("ThinClientPoolDM::getFromEP got connection");
      TcrConnection* retVal = *itr;
      queue_.erase(itr);
      return retVal;
    }
  }

  return nullptr;
}

void ThinClientPoolDM::removeEPConnections(TcrEndpoint* theEP) {
  std::lock_guard<decltype(mutex_)> lock(mutex_);
  auto size = static_cast<int32_t>(queue_.size());
  int numConn = 0;

  while (size--) {
    TcrConnection* curConn = queue_.back();
    queue_.pop_back();
    if (curConn->getEndpointObject() != theEP) {
      queue_.push_front(curConn);
    } else {
      curConn->close();
      _GEODE_SAFE_DELETE(curConn);
      numConn++;
    }
  }

  removeEPConnections(numConn);
}

TcrConnection* ThinClientPoolDM::getNoGetLock(
    bool& isClosed, GfErrType* error, std::set<ServerLocation>& excludeServers,
    bool& maxConnLimit) {
  TcrConnection* returnT = nullptr;
  {
    std::lock_guard<decltype(mutex_)> lock(mutex_);

    do {
      returnT = popNoLock(isClosed);
      if (returnT) {
        if (excludeConnection(returnT, excludeServers)) {
          returnT->close();
          _GEODE_SAFE_DELETE(returnT);
          removeEPConnections(1, false);
        } else {
          break;
        }
      } else {
        break;
      }
    } while (!returnT);
  }

  if (!returnT) {
    *error = createPoolConnection(returnT, excludeServers, maxConnLimit);
  }

  return returnT;
}

void ThinClientPoolDM::incRegionCount() {
  std::lock_guard<decltype(mutex_)> lock(mutex_);

  if (!m_isDestroyed && !m_destroyPending) {
    m_numRegions++;
  } else {
    throw IllegalStateException("Pool has been destroyed.");
  }
}

void ThinClientPoolDM::decRegionCount() {
  std::lock_guard<decltype(mutex_)> lock(mutex_);

  m_numRegions--;
}

void ThinClientPoolDM::checkRegions() {
  std::lock_guard<decltype(mutex_)> lock(mutex_);

  if (m_numRegions > 0) {
    throw IllegalStateException(
        "Failed to destroy pool because regions are connected with it.");
  }

  m_destroyPending = true;
}
void ThinClientPoolDM::updateNotificationStats(
    bool isDeltaSuccess, std::chrono::nanoseconds timeInNanoSecond) {
  if (isDeltaSuccess) {
    getStats().incProcessedDeltaMessages();
    getStats().incProcessedDeltaMessagesTime(timeInNanoSecond.count());
  } else {
    getStats().incDeltaMessageFailures();
  }
}

GfErrType ThinClientPoolDM::doFailover(TcrConnection* conn) {
  m_manager->setStickyConnection(conn, true);
  TcrMessageTxFailover request(
      new DataOutput(m_connManager.getCacheImpl()->createDataOutput()));
  TcrMessageReply reply(true, nullptr);

  auto err = sendSyncRequest(request, reply);

  if (err == GF_NOERR) {
    switch (reply.getMessageType()) {
      case TcrMessage::REPLY: {
        break;
      }
      case TcrMessage::EXCEPTION: {
        const auto& exceptionMsg = reply.getException();
        err = ThinClientRegion::handleServerException(
            "CacheTransactionManager::failover", exceptionMsg);
        break;
      }
      case TcrMessage::REQUEST_DATA_ERROR: {
        err = GF_CACHESERVER_EXCEPTION;
        break;
      }
      default: {
        LOGERROR("Unknown message type in failover reply %d",
                 reply.getMessageType());
        err = GF_MSG;
        break;
      }
    }
  }

  return err;
}

bool ThinClientPoolDM::canItBeDeletedNoImpl(TcrConnection*) { return false; }

void ThinClientPoolDM::putInQueue(TcrConnection* conn, bool,
                                  bool isTransaction) {
  if (isTransaction) {
    m_manager->setStickyConnection(conn, isTransaction);
  } else {
    put(conn, false);
  }
}

TcrConnection* ThinClientPoolDM::getConnectionFromQueueW(
    GfErrType* error, std::set<ServerLocation>& excludeServers, bool,
    TcrMessage& request, int8_t& version, bool& match, bool& connFound,
    const std::shared_ptr<BucketServerLocation>& serverLocation) {
  TcrConnection* conn = nullptr;
  TcrEndpoint* theEP = nullptr;
  LOGDEBUG("prEnabled = %s, forSingleHop = %s %d",
           m_attrs->getPRSingleHopEnabled() ? "true" : "false",
           request.forSingleHop() ? "true" : "false", request.getMessageType());

  match = false;
  std::shared_ptr<BucketServerLocation> slTmp = nullptr;
  if (request.forTransaction()) {
    connFound =
        m_manager->getStickyConnection(conn, error, excludeServers, true);
    auto txState = TSSTXStateWrapper::get().getTXState();
    if (*error == GF_NOERR && !connFound &&
        (txState == nullptr || txState->isDirty())) {
      *error = doFailover(conn);
    }

    if (*error != GF_NOERR) {
      return nullptr;
    }

    if (txState != nullptr) {
      txState->setDirty();
    }
  } else if (serverLocation != nullptr /*&& excludeServers.size() == 0*/) {
    theEP = getEndPoint(serverLocation, version, excludeServers);
  } else if (m_attrs->getPRSingleHopEnabled() /*&& excludeServers.size() == 0*/
             && request.forSingleHop() &&
             (request.getMessageType() != TcrMessage::GET_ALL_70) &&
             (request.getMessageType() != TcrMessage::GET_ALL_WITH_CALLBACK)) {
    theEP = getSingleHopServer(request, version, slTmp, excludeServers);
    if (theEP != nullptr) {
      // if all buckets are not initialized
      //  match = true;
    }
    if (slTmp != nullptr && m_clientMetadataService) {
      if (m_clientMetadataService->isBucketMarkedForTimeout(
              request.getRegionName().c_str(), slTmp->getBucketId())) {
        *error = GF_CLIENT_WAIT_TIMEOUT;
        return nullptr;
      }
    }
    LOGDEBUG("theEP is %p", theEP);
  }
  bool maxConnLimit = false;
  if (theEP != nullptr) {
    conn = getFromEP(theEP);
    if (!conn) {
      LOGFINER("Creating connection to endpoint as not found in pool ");
      *error = createPoolConnectionToAEndPoint(conn, theEP, maxConnLimit, true);
      if (*error == GF_CLIENT_WAIT_TIMEOUT ||
          *error == GF_CLIENT_WAIT_TIMEOUT_REFRESH_PRMETADATA) {
        if (!m_clientMetadataService || request.getKey() == nullptr) {
          return nullptr;
        }

        auto region =
            m_connManager.getCacheImpl()->getRegion(request.getRegionName());
        if (region != nullptr) {
          slTmp = nullptr;
          m_clientMetadataService
              ->markPrimaryBucketForTimeoutButLookSecondaryBucket(
                  region, request.getKey(), request.getValue(),
                  request.getCallbackArgument(), request.forPrimary(), slTmp,
                  version);
        }
        return nullptr;
      } else {
        removeEPFromMetadataIfError(*error, theEP);
      }
    }
  }
  if (conn == nullptr) {
    LOGDEBUG("conn not found");
    match = false;
    LOGDEBUG("looking For connection");
    conn = getConnectionFromQueue(true, error, excludeServers, maxConnLimit);
    LOGDEBUG("Connection Found");
  }

  if (maxConnLimit) {
    // we reach max connection limit, found connection but endpoint is
    // (not)different, no need to refresh pr-meta-data
    connFound = true;
  } else {
    // if server hints pr-meta-data refresh then refresh
    // anything else???
  }

  LOGDEBUG(
      "ThinClientPoolDM::getConnectionFromQueueW return conn = %p match = %d "
      "connFound=%d",
      conn, match, connFound);
  return conn;
}

bool ThinClientPoolDM::checkDupAndAdd(std::shared_ptr<EventId> eventid) {
  return m_connManager.checkDupAndAdd(eventid);
}

std::shared_ptr<TcrEndpoint> ThinClientPoolDM::createEP(
    const char* endpointName) {
  return std::make_shared<TcrPoolEndPoint>(
      endpointName, m_connManager.getCacheImpl(), m_connManager.m_failoverSema,
      m_connManager.m_cleanupSema, m_connManager.m_redundancySema, this);
}

GfErrType FunctionExecution::execute() {
  GuardUserAttributes gua;

  if (m_userAttr) {
    gua.setAuthenticatedView(m_userAttr->getAuthenticatedView());
  }

  std::string funcName(m_func);
  TcrMessageExecuteFunction request(
      new DataOutput(
          m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput()),
      funcName, m_args, m_getResult, m_poolDM, m_timeout);
  TcrMessageReply reply(true, m_poolDM);
  auto resultProcessor = std::unique_ptr<ChunkedFunctionExecutionResponse>(
      new ChunkedFunctionExecutionResponse(reply, (m_getResult & 2) == 2, *m_rc,
                                           m_resultCollectorLock));
  reply.setChunkedResultHandler(resultProcessor.get());
  reply.setTimeout(m_timeout);
  reply.setDM(m_poolDM);

  LOGDEBUG(
      "ThinClientPoolDM::sendRequestToAllServer sendRequest on endpoint[%s]!",
      m_ep->name().c_str());

  m_error = m_poolDM->sendRequestToEP(request, reply, m_ep);
  m_error = m_poolDM->handleEPError(m_ep, reply, m_error);
  if (m_error != GF_NOERR) {
    if (m_error == GF_NOTCON) {
      return GF_NOERR;  // if server is unavailable its not an error for
      // functionexec OnServers() case
    }
    LOGDEBUG("FunctionExecution::execute failed on endpoint[%s]!. Error = %d ",
             m_ep->name().c_str(), m_error);
    if (reply.getMessageType() == TcrMessage::EXCEPTION) {
      exceptionPtr = CacheableString::create(reply.getException());
    }

    return m_error;
  } else if (reply.getMessageType() == TcrMessage::EXCEPTION ||
             reply.getMessageType() == TcrMessage::EXECUTE_FUNCTION_ERROR) {
    m_error = ThinClientRegion::handleServerException("Execute",
                                                      reply.getException());
    exceptionPtr = CacheableString::create(reply.getException());
  }

  return m_error;
}

OnRegionFunctionExecution::OnRegionFunctionExecution(
    std::string func, const Region* region, std::shared_ptr<Cacheable> args,
    std::shared_ptr<CacheableHashSet> routingObj, uint8_t getResult,
    std::chrono::milliseconds timeout, ThinClientPoolDM* poolDM,
    const std::shared_ptr<std::recursive_mutex>& rCL,
    std::shared_ptr<ResultCollector> rs,
    std::shared_ptr<UserAttributes> userAttr, bool isBGThread,
    const std::shared_ptr<BucketServerLocation>& serverLocation,
    bool allBuckets)
    : m_serverLocation(serverLocation),
      m_isBGThread(isBGThread),
      m_poolDM(poolDM),
      m_func(func),
      m_getResult(getResult),
      m_timeout(timeout),
      m_args(args),
      m_routingObj(routingObj),
      m_rc(rs),
      m_resultCollectorLock(rCL),
      m_userAttr(userAttr),
      m_region(region),
      m_allBuckets(allBuckets) {
  m_request = new TcrMessageExecuteRegionFunctionSingleHop(
      new DataOutput(
          m_poolDM->getConnectionManager().getCacheImpl()->createDataOutput()),
      m_func, m_region, m_args, m_routingObj, m_getResult, nullptr,
      m_allBuckets, timeout, m_poolDM);
  m_reply = new TcrMessageReply(true, m_poolDM);
  m_resultCollector = new ChunkedFunctionExecutionResponse(
      *m_reply, (m_getResult & 2) == 2, m_rc, m_resultCollectorLock);
  m_reply->setChunkedResultHandler(m_resultCollector);
  m_reply->setTimeout(m_timeout);
  m_reply->setDM(m_poolDM);
}

}  // namespace client
}  // namespace geode
}  // namespace apache