You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bb...@apache.org on 2021/03/31 15:50:00 UTC
[geode-native] branch develop updated: GEODE-8601: Replace
ExpiryTaskManager ACE implementation by boost::asio one (#678)
This is an automated email from the ASF dual-hosted git repository.
bbender pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode-native.git
The following commit(s) were added to refs/heads/develop by this push:
new db82746 GEODE-8601: Replace ExpiryTaskManager ACE implementation by boost::asio one (#678)
db82746 is described below
commit db82746d2ed72a8d88651e4912a4d3805a2507b3
Author: Mario Salazar de Torres <ma...@est.tech>
AuthorDate: Wed Mar 31 17:49:49 2021 +0200
GEODE-8601: Replace ExpiryTaskManager ACE implementation by boost::asio one (#678)
* GEODE-8601: Replace ExpiryTaskManager impl
- Now ExpiryTaskManager implementation uses boost::asio.
- Handlers have been replaced by ExpiryTasks, which will serve as base
to each custom expiry task.
- Add a generic ExpiryTask implementation which accepts an
std::function as on_expire callback.
- Refactored EntryExpiryHandler code to use new ExpiryTask implementation.
- Class has been renamed to EntryExpiryTask. However file has not being renamed
so git log is not lost. This will be addressed in the next commit.
- As class has been previously renamed, now files have also been renamed to
EntryExpiryTask.{h,c}pp
- Refactored RegionExpiryHandler to use the new ExpiryTask implementation.
- Renamed RegionExpiryHandler class to RegionExpiryTask. Note that files have
not been renamed in order to avoid git log from being lost. This will be
addressed in the following commit.
- Moved TombstoneEntry into a separate file, for clarity's shake.
- Code has been refactored, so only thing stored is:
* Task ID.
* Validity flag. This is a new flag added to avoid race conditions.
This way, tombstone entries can be now invalidated.
* Actual entry's reference.
- Refactored TomstoneExpiryHandler to use new ExpiryTask implementation.
- TombstoneExpiryHandler class files have been renamed to
TombstoneExpiryTask.{h,c}pp.
- Refactored SuspendedTxExpiryHandler to use the new implementation of the
ExpiryTask.
- Renamed SuspendedTxExpiryHandler to SuspendedTxExpiryTask.
- Renamed SuspendedTxExpiryHandler files to SuspendedTxExpiryTask.{h,c}pp
- Used new id_t types from ExpiryTask rather than the ones from
ExpiryTaskManager for the TXState class.
- Chnged setter names to comply with Google style guide.
- Setters have been left the same as they are part of the public API.
Consider if it's possible to change these names.
- Modified CLI's CacheStatistics to convert between system and steady
time_points.
- Changed CacheImpl to integrate with the new implementation of the
ExpiryTaskManager.
- While doing several refactors to adapt to the new ExpiryTaskManager
implementation it has been considered convinient to split MapEntry.hpp
into several files so everything is more readable.
- Minor refactors.
- TombstoneList has been completely refactored to better adapt the new
implementation of ExpiryTaskManager.
- Now ExpiryTask is created and scheduled in the add method, rather
than having two separate methods.
- Refactored TcrConnectionManager to use the new ExpiryTaskManager
implementation.
- Removed unused code in TcrConnectionManager related to expiry tasks.
- Changes made to integrate with the new implementation of the
ExpiryTaskManager.
- Refactored LocalRegion class to integrate with the new ExpiryTaskManager
implementation.
- system_clock has been replaced by steady_clock which is a monotonic one.
- Refactored PreservedDataExpiryHandler to integrate with the new
implementation of ExpiryTaskManager.
- Renamed PreservedDataExpiryHandler to PreservedDataExpiryTask. File renaming
has been left for the following commit in order to avoid git log from being
lost.
- Added the possiblity to internally reset the task, based on the expiration
time point.
- Fixed preserved expiry task as the entry was being removed from a copy,
rather than the actual map.
- Renamed PreservedDataExpiryHandler class files.
- Refactored PdxRemotePreservedData to integrate with the new ExpiryTaskManager
implementation.
- Added expiration time point.
- Refactored PdxTypeRegistry to integrate new ExpiryTaskManager implementation.
- Refactored ThinClientRedundancyManager to integrate with the new
ExpiryTaskmanager implementation.
- Refactored ThinClientPoolHADM to integrate with the new ExpiryTaskManager
implementation.
- Refactored ThinClientPoolDM to integrate with the new ExpiryTaskManager
implementation.
- Replaced 1ms sleep in cleanStaleConnections by thread yield in order
to guarantee thread is re-scheduled as soon as possible.
- Refactored CLI cache MemoryPressureHandler to integrate with the new
ExpiryTaskManager implementation.
- Renamed class to MemoryPressureTask. Files renaming left to the next commit
to avoid git log from being lost.
- Change integrated in DistributedSystem.
- Note that this task is not schduled per-se, however has been left as it was
before, in order not to change the behavior.
- Renamed MemoryPressureHandler class files to MemoryPressureTask.{h,c}pp
- Modified CMakeLists.txt to use new files.
- Integrated change in DistributedSystem
- Due to previous refactor, somne includes were missing both in PdxLocalReader
and PdxRemoteWriter.
- Added missing includes in PdxInstanceImplTest after previous refactor.
- Added a mock for ExpiryTask.
- Added UTs for ExpiryTaskManager.
- Added ExpiryTask UTs.
- Added Google Mock extensions.
- Modified ExpiryTask mock to expose reset method.
- Added Boost linking to testobject and old integrations test to enforce using
BOOST_ALL_NO_LIB preprocessor variable while compiling.
- Some implementations of ExpiryTask were exported. While that's not being
necessary it was causing a compilation error in Windows due to the base class
not being also exported. Exposure is removed.
- Fixed steady/system clock differences. Now CacheStatistics return
system clock timepoint, but uses steady timepoint as internal
representation. Also, added 2 new non-virtual methods to obtain
steady timepoint.
- Used string like logging for EntryExpiryTask.
- Added Boost to legacy integration tests as in order to avoid
clang-tidy reporting boost issues it's needed to include boost
headers as system.
- Removed unused C# MemoryPressureTask
---
clicache/src/CMakeLists.txt | 2 -
clicache/src/CacheStatistics.cpp | 8 +-
clicache/src/DistributedSystem.cpp | 10 -
clicache/src/DistributedSystem.hpp | 17 -
clicache/src/impl/MemoryPressureHandler.cpp | 76 -----
cppcache/include/geode/CacheStatistics.hpp | 24 +-
.../integration-test/testOverflowPutGetSqLite.cpp | 2 +
cppcache/integration/test/PdxInstanceTest.cpp | 1 +
cppcache/src/CacheImpl.cpp | 5 +-
cppcache/src/CacheStatistics.cpp | 38 ++-
cppcache/src/CacheTransactionManagerImpl.cpp | 22 +-
cppcache/src/CacheTransactionManagerImpl.hpp | 3 +-
cppcache/src/ClientMetadataService.cpp | 1 +
cppcache/src/CqEventImpl.cpp | 1 +
cppcache/src/CqQueryImpl.cpp | 4 +-
cppcache/src/CqService.cpp | 1 +
cppcache/src/EntriesMapFactory.cpp | 5 +-
cppcache/src/EntryExpiryHandler.cpp | 150 ---------
cppcache/src/EntryExpiryHandler.hpp | 78 -----
cppcache/src/EntryExpiryTask.cpp | 123 +++++++
...edDataExpiryHandler.hpp => EntryExpiryTask.hpp} | 52 +--
cppcache/src/ExecutionImpl.cpp | 1 +
cppcache/src/ExpEntryProperties.hpp | 95 ++++++
cppcache/src/ExpMapEntry.hpp | 5 +-
cppcache/src/ExpiryHandler_T.hpp | 77 -----
cppcache/src/ExpiryTask.cpp | 69 ++++
cppcache/src/ExpiryTask.hpp | 203 ++++++++++++
cppcache/src/ExpiryTaskManager.cpp | 157 +++++----
cppcache/src/ExpiryTaskManager.hpp | 353 +++++++--------------
cppcache/src/FarSideEntryOp.cpp | 20 +-
cppcache/src/FarSideEntryOp.hpp | 107 +++----
cppcache/src/FunctionExpiryTask.hpp | 69 ++++
.../src/InternalCacheTransactionManager2PCImpl.cpp | 2 +-
cppcache/src/LRUAction.hpp | 2 +-
cppcache/src/LRUExpMapEntry.hpp | 5 +-
cppcache/src/LRUMapEntry.hpp | 2 +-
cppcache/src/LRUQueue.cpp | 2 +-
cppcache/src/LocalRegion.cpp | 102 +++---
cppcache/src/LocalRegion.hpp | 2 +
cppcache/src/MapEntry.hpp | 190 +----------
cppcache/src/{MapEntry.cpp => MapEntryImpl.cpp} | 22 +-
cppcache/src/MapEntryImpl.hpp | 145 +++++++++
cppcache/src/MapSegment.cpp | 323 +++++++++----------
cppcache/src/MapSegment.hpp | 48 ++-
cppcache/src/PdxLocalReader.cpp | 3 +
cppcache/src/PdxRemotePreservedData.hpp | 41 ++-
cppcache/src/PdxRemoteWriter.cpp | 1 +
cppcache/src/PdxTypeRegistry.cpp | 44 +--
cppcache/src/PdxTypeRegistry.hpp | 16 +-
cppcache/src/Pool.cpp | 1 +
...piryHandler.cpp => PreservedDataExpiryTask.cpp} | 49 +--
...piryHandler.hpp => PreservedDataExpiryTask.hpp} | 31 +-
cppcache/src/ProxyRemoteQueryService.cpp | 1 +
cppcache/src/RegionExpiryHandler.cpp | 136 --------
cppcache/src/RegionExpiryTask.cpp | 95 ++++++
...egionExpiryHandler.hpp => RegionExpiryTask.hpp} | 47 ++-
...ExpiryHandler.cpp => SuspendedTxExpiryTask.cpp} | 25 +-
...ExpiryHandler.hpp => SuspendedTxExpiryTask.hpp} | 35 +-
cppcache/src/TXCommitMessage.hpp | 2 +-
cppcache/src/TXState.cpp | 1 +
cppcache/src/TXState.hpp | 7 +-
cppcache/src/TcrConnection.cpp | 1 +
cppcache/src/TcrConnectionManager.cpp | 41 +--
cppcache/src/TcrConnectionManager.hpp | 10 +-
cppcache/src/TcrPoolEndPoint.cpp | 1 +
cppcache/src/ThinClientBaseDM.cpp | 1 +
.../src/ThinClientCacheDistributionManager.cpp | 2 -
cppcache/src/ThinClientDistributionManager.cpp | 1 +
cppcache/src/ThinClientLocatorHelper.cpp | 1 +
cppcache/src/ThinClientPoolDM.cpp | 153 ++++-----
cppcache/src/ThinClientPoolDM.hpp | 12 +-
cppcache/src/ThinClientPoolHADM.cpp | 33 +-
cppcache/src/ThinClientPoolHADM.hpp | 3 +-
cppcache/src/ThinClientRedundancyManager.cpp | 35 +-
cppcache/src/ThinClientRedundancyManager.hpp | 11 +-
.../TombstoneEntry.hpp} | 37 ++-
cppcache/src/TombstoneExpiryHandler.cpp | 92 ------
cppcache/src/TombstoneExpiryHandler.hpp | 73 -----
...CacheStatistics.cpp => TombstoneExpiryTask.cpp} | 34 +-
...taExpiryHandler.hpp => TombstoneExpiryTask.hpp} | 54 ++--
cppcache/src/TombstoneList.cpp | 145 ++++-----
cppcache/src/TombstoneList.hpp | 79 ++---
cppcache/src/TrackedMapEntry.cpp | 2 +-
cppcache/src/VersionedCacheableObjectPartList.cpp | 1 +
cppcache/test/CMakeLists.txt | 4 +
cppcache/test/ExpiryTaskManagerTest.cpp | 148 +++++++++
cppcache/test/ExpiryTaskTest.cpp | 153 +++++++++
cppcache/test/PdxInstanceImplTest.cpp | 10 +-
.../test/gmock_extensions.h | 28 +-
cppcache/test/mock/MapEntryImplMock.hpp | 2 +-
.../MapEntry.cpp => test/mock/MockExpiryTask.hpp} | 24 +-
tests/cpp/testobject/CMakeLists.txt | 1 +
92 files changed, 2198 insertions(+), 2153 deletions(-)
diff --git a/clicache/src/CMakeLists.txt b/clicache/src/CMakeLists.txt
index 7190903..ab5f454 100644
--- a/clicache/src/CMakeLists.txt
+++ b/clicache/src/CMakeLists.txt
@@ -261,8 +261,6 @@ add_library(Apache.Geode SHARED
impl/ManagedTransactionWriter.hpp
impl/ManagedVisitor.cpp
impl/ManagedVisitor.hpp
- impl/MemoryPressureHandler.cpp
- impl/MemoryPressureHandler.hpp
impl/PartitionResolver.hpp
impl/PdxFieldType.cpp
impl/PdxFieldType.hpp
diff --git a/clicache/src/CacheStatistics.cpp b/clicache/src/CacheStatistics.cpp
index 818007a..800c6a2 100644
--- a/clicache/src/CacheStatistics.cpp
+++ b/clicache/src/CacheStatistics.cpp
@@ -32,7 +32,9 @@ namespace Apache
{
try
{
- return TimeUtils::TimePointToDateTime(m_nativeptr->get()->getLastModifiedTime());
+ auto epoch = m_nativeptr->get()->getLastModifiedTime().time_since_epoch();
+ auto duration = std::chrono::duration_cast<std::chrono::system_clock::duration>(epoch);
+ return TimeUtils::TimePointToDateTime(std::chrono::system_clock::time_point(duration));
}
finally
{
@@ -44,7 +46,9 @@ namespace Apache
{
try
{
- return TimeUtils::TimePointToDateTime(m_nativeptr->get()->getLastAccessedTime());
+ auto epoch = m_nativeptr->get()->getLastAccessedTime().time_since_epoch();
+ auto duration = std::chrono::duration_cast<std::chrono::system_clock::duration>(epoch);
+ return TimeUtils::TimePointToDateTime(std::chrono::system_clock::time_point(duration));
}
finally
{
diff --git a/clicache/src/DistributedSystem.cpp b/clicache/src/DistributedSystem.cpp
index 8c297d9..05c8e14 100644
--- a/clicache/src/DistributedSystem.cpp
+++ b/clicache/src/DistributedSystem.cpp
@@ -52,7 +52,6 @@
#include "CacheableBuiltins.hpp"
#include "Log.hpp"
#include "Struct.hpp"
-#include "impl/MemoryPressureHandler.hpp"
#include "impl/SafeConvert.hpp"
#include "impl/PdxType.hpp"
#include "impl/EnumInfo.hpp"
@@ -401,14 +400,6 @@ namespace Apache
}
}
- void DistributedSystem::HandleMemoryPressure(System::Object^ state)
- {
- // TODO global - Need single memory pressue event running?
- ACE_Time_Value dummy(1);
- MemoryPressureHandler handler;
- handler.handle_timeout(dummy, nullptr);
- }
-
DistributedSystem^ DistributedSystem::Create(native::DistributedSystem* nativeptr)
{
auto instance = gcnew DistributedSystem(nativeptr);
@@ -427,7 +418,6 @@ namespace Apache
DistributedSystem::~DistributedSystem()
{
- m_memoryPressureHandler->Dispose(nullptr);
}
void DistributedSystem::registerCliCallback()
diff --git a/clicache/src/DistributedSystem.hpp b/clicache/src/DistributedSystem.hpp
index 8bd0d37..bb856d5 100644
--- a/clicache/src/DistributedSystem.hpp
+++ b/clicache/src/DistributedSystem.hpp
@@ -159,12 +159,6 @@ namespace Apache
private:
- ///// <summary>
- ///// Stores the task ID of the task that adjusts unmanaged memory
- ///// pressure in managed GC.
- ///// </summary>
- //static long s_memoryPressureTaskID = -1;
-
/// <summary>
/// Private constructor to wrap a native object pointer
/// </summary>
@@ -178,17 +172,6 @@ namespace Apache
native_conditional_unique_ptr<native::DistributedSystem>^ m_nativeDistributedSystem;
-
- /// <summary>
- /// Periodically adjust memory pressure of unmanaged heap for GC.
- /// </summary>
- static void HandleMemoryPressure(System::Object^ state);
-
- /// <summary>
- /// Timer task to periodically invoke <c>HandleMemoryPressure</c>.
- /// </summary>
- System::Threading::Timer^ m_memoryPressureHandler;
-
static CliCallbackDelegate^ m_cliCallBackObj;
};
} // namespace Client
diff --git a/clicache/src/impl/MemoryPressureHandler.cpp b/clicache/src/impl/MemoryPressureHandler.cpp
deleted file mode 100644
index 2b01c74..0000000
--- a/clicache/src/impl/MemoryPressureHandler.cpp
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * 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 "MemoryPressureHandler.hpp"
-#include "windows.h"
-#include "psapi.h"
-#include "../Log.hpp"
-
-namespace Apache
-{
- namespace Geode
- {
- namespace Client
- {
-
- System::Int64 g_prevUnmanagedSize = 0;
-
- int MemoryPressureHandler::handle_timeout( const ACE_Time_Value&
- current_time, const void* arg )
- {
- HANDLE hProcess = GetCurrentProcess( );
-
- PROCESS_MEMORY_COUNTERS pmc;
-
- if ( GetProcessMemoryInfo( hProcess, &pmc, sizeof(pmc)) ) {
- System::Int64 totalmem = (System::Int64)pmc.WorkingSetSize;
- System::Int64 curr_managed_size = GC::GetTotalMemory( false );
- System::Int64 curr_unmanagedMemory = totalmem - curr_managed_size;
- Log::Finest( "Current total memory usage: {0}, managed memory: {1}, "
- "unmanaged memory: {2}", totalmem, curr_managed_size,
- curr_unmanagedMemory );
- if ( curr_unmanagedMemory > 0 ) {
- System::Int64 increase = curr_unmanagedMemory - g_prevUnmanagedSize;
- if ( Math::Abs( increase ) > 20*1024*1024 ) {
- if ( increase > 0 ) {
- Log::Fine( "Adding memory pressure information to assist .NET GC: {0} bytes", increase );
- GC::AddMemoryPressure( increase );
- }
- else {
- Log::Fine( "Removing memory pressure information to assist .NET GC: {0} bytes", -increase );
- GC::RemoveMemoryPressure( -increase );
- }
- g_prevUnmanagedSize = curr_unmanagedMemory;
- }
- }
- }
- else {
- return -1;
- }
- return 0;
- }
-
- int MemoryPressureHandler::handle_close(ACE_HANDLE handle,
- ACE_Reactor_Mask close_mask)
- {
- return 0;
- }
- } // namespace Client
- } // namespace Geode
-} // namespace Apache
diff --git a/cppcache/include/geode/CacheStatistics.hpp b/cppcache/include/geode/CacheStatistics.hpp
index 5bf3337..626b835 100644
--- a/cppcache/include/geode/CacheStatistics.hpp
+++ b/cppcache/include/geode/CacheStatistics.hpp
@@ -46,12 +46,13 @@ class LocalRegion;
*@see RegionEntry::getStatistics
*/
class APACHE_GEODE_EXPORT CacheStatistics {
- public:
- typedef std::chrono::system_clock::time_point time_point;
+ private:
+ using time_point = std::chrono::steady_clock::time_point;
- CacheStatistics() : m_lastAccessTime(0), m_lastModifiedTime(0) {}
+ public:
+ CacheStatistics();
CacheStatistics(const CacheStatistics&) = delete;
- virtual ~CacheStatistics() = default;
+ virtual ~CacheStatistics();
/**
* For an entry, returns the time that the entry's value was last modified.
@@ -75,7 +76,7 @@ class APACHE_GEODE_EXPORT CacheStatistics {
* @see Region::create
* @see Region::createSubregion
*/
- virtual time_point getLastModifiedTime() const;
+ virtual std::chrono::system_clock::time_point getLastModifiedTime() const;
/**
* For an entry, returns the last time it was accessed via
@@ -94,14 +95,17 @@ class APACHE_GEODE_EXPORT CacheStatistics {
* @see Region::get
* @see getLastModifiedTime
*/
- virtual time_point getLastAccessedTime() const;
+ virtual std::chrono::system_clock::time_point getLastAccessedTime() const;
+
+ time_point getLastModifiedSteadyTime() const;
+ time_point getLastAccessedSteadyTime() const;
private:
- virtual void setLastAccessedTime(time_point lat);
- virtual void setLastModifiedTime(time_point lmt);
+ virtual void setLastAccessedTime(time_point tp);
+ virtual void setLastModifiedTime(time_point tp);
- std::atomic<time_point::duration::rep> m_lastAccessTime;
- std::atomic<time_point::duration::rep> m_lastModifiedTime;
+ std::atomic<time_point::duration::rep> last_accessed_;
+ std::atomic<time_point::duration::rep> last_modified_;
friend class LocalRegion;
};
diff --git a/cppcache/integration-test/testOverflowPutGetSqLite.cpp b/cppcache/integration-test/testOverflowPutGetSqLite.cpp
index aaffb37..b595423 100644
--- a/cppcache/integration-test/testOverflowPutGetSqLite.cpp
+++ b/cppcache/integration-test/testOverflowPutGetSqLite.cpp
@@ -29,6 +29,8 @@
#include <CacheRegionHelper.hpp>
#include <MapEntry.hpp>
+#include "CacheImpl.hpp"
+
#include "fw_helper.hpp"
using apache::geode::client::Cache;
diff --git a/cppcache/integration/test/PdxInstanceTest.cpp b/cppcache/integration/test/PdxInstanceTest.cpp
index ead7f58..7a4c89f 100644
--- a/cppcache/integration/test/PdxInstanceTest.cpp
+++ b/cppcache/integration/test/PdxInstanceTest.cpp
@@ -33,6 +33,7 @@
#include <geode/RegionShortcut.hpp>
#include <geode/TypeRegistry.hpp>
+#include "CacheImpl.hpp"
#include "LocalRegion.hpp"
#include "NestedPdxObject.hpp"
#include "PdxType.hpp"
diff --git a/cppcache/src/CacheImpl.cpp b/cppcache/src/CacheImpl.cpp
index 97846ef..16539fe 100644
--- a/cppcache/src/CacheImpl.cpp
+++ b/cppcache/src/CacheImpl.cpp
@@ -33,7 +33,6 @@
#include "InternalCacheTransactionManager2PCImpl.hpp"
#include "LocalRegion.hpp"
#include "PdxTypeRegistry.hpp"
-#include "RegionExpiryHandler.hpp"
#include "SerializationRegistry.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrEndpoint.hpp"
@@ -89,7 +88,7 @@ CacheImpl::CacheImpl(Cache* c, const std::shared_ptr<Properties>& dsProps,
LOGINFO("Heap LRU eviction controller thread started");
}
- m_expiryTaskManager->begin();
+ m_expiryTaskManager->start();
m_initialized = true;
m_pdxTypeRegistry = std::make_shared<PdxTypeRegistry>(this);
@@ -322,7 +321,7 @@ void CacheImpl::close(bool keepalive) {
_GEODE_SAFE_DELETE(m_tcrConnectionManager);
m_cacheTXManager = nullptr;
- m_expiryTaskManager->stopExpiryTaskManager();
+ m_expiryTaskManager->stop();
try {
getDistributedSystem().disconnect();
diff --git a/cppcache/src/CacheStatistics.cpp b/cppcache/src/CacheStatistics.cpp
index b6a1dc3..cd055c2 100644
--- a/cppcache/src/CacheStatistics.cpp
+++ b/cppcache/src/CacheStatistics.cpp
@@ -17,26 +17,48 @@
#include <geode/CacheStatistics.hpp>
+namespace {
+std::chrono::system_clock::time_point convert_steady_to_system_tp(
+ std::chrono::steady_clock::time_point tp) {
+ return std::chrono::system_clock::now() +
+ std::chrono::duration_cast<std::chrono::system_clock::duration>(
+ std::chrono::steady_clock::now() - tp);
+}
+} // namespace
+
namespace apache {
namespace geode {
namespace client {
-void CacheStatistics::setLastModifiedTime(time_point lmt) {
- m_lastModifiedTime = lmt.time_since_epoch().count();
+CacheStatistics::CacheStatistics() : last_accessed_{0}, last_modified_{0} {}
+
+CacheStatistics::~CacheStatistics() = default;
+
+void CacheStatistics::setLastModifiedTime(time_point tp) {
+ last_modified_ = tp.time_since_epoch().count();
}
-void CacheStatistics::setLastAccessedTime(time_point lat) {
- m_lastAccessTime = lat.time_since_epoch().count();
+void CacheStatistics::setLastAccessedTime(time_point tp) {
+ last_accessed_ = tp.time_since_epoch().count();
}
-CacheStatistics::time_point CacheStatistics::getLastModifiedTime() const {
- return time_point(std::chrono::system_clock::duration(m_lastModifiedTime));
+std::chrono::system_clock::time_point CacheStatistics::getLastModifiedTime()
+ const {
+ return convert_steady_to_system_tp(getLastModifiedSteadyTime());
}
-CacheStatistics::time_point CacheStatistics::getLastAccessedTime() const {
- return time_point(std::chrono::system_clock::duration(m_lastAccessTime));
+std::chrono::system_clock::time_point CacheStatistics::getLastAccessedTime()
+ const {
+ return convert_steady_to_system_tp(getLastAccessedSteadyTime());
}
+CacheStatistics::time_point CacheStatistics::getLastModifiedSteadyTime() const {
+ return time_point{time_point::duration{last_modified_}};
+}
+
+CacheStatistics::time_point CacheStatistics::getLastAccessedSteadyTime() const {
+ return time_point{time_point::duration{last_accessed_}};
+}
} // namespace client
} // namespace geode
} // namespace apache
diff --git a/cppcache/src/CacheTransactionManagerImpl.cpp b/cppcache/src/CacheTransactionManagerImpl.cpp
index 9baf25d..d0d432d 100644
--- a/cppcache/src/CacheTransactionManagerImpl.cpp
+++ b/cppcache/src/CacheTransactionManagerImpl.cpp
@@ -18,11 +18,11 @@
#include "CacheTransactionManagerImpl.hpp"
#include <geode/ExceptionTypes.hpp>
-#include <geode/PoolManager.hpp>
#include <geode/TransactionId.hpp>
#include "CacheImpl.hpp"
#include "CacheRegionHelper.hpp"
+#include "SuspendedTxExpiryTask.hpp"
#include "TSSTXStateWrapper.hpp"
#include "TXCleaner.hpp"
#include "TcrMessage.hpp"
@@ -217,13 +217,13 @@ TransactionId& CacheTransactionManagerImpl::suspend() {
txState->releaseStickyConnection();
// set the expiry handler for the suspended transaction
- auto suspendedTxTimeout = m_cache->getDistributedSystem()
- .getSystemProperties()
- .suspendedTxTimeout();
- auto handler =
- new SuspendedTxExpiryHandler(this, txState->getTransactionId());
- auto id = m_cache->getExpiryTaskManager().scheduleExpiryTask(
- handler, suspendedTxTimeout, std::chrono::seconds::zero(), false);
+ auto timeout = m_cache->getDistributedSystem()
+ .getSystemProperties()
+ .suspendedTxTimeout();
+ auto& manager = m_cache->getExpiryTaskManager();
+ auto task = std::make_shared<SuspendedTxExpiryTask>(
+ manager, *this, txState->getTransactionId());
+ auto id = manager.schedule(std::move(task), timeout);
txState->setSuspendedExpiryTaskId(id);
// add the transaction state to the list of suspended transactions
@@ -308,11 +308,7 @@ void CacheTransactionManagerImpl::resumeTxUsingTxState(TXState* txState,
if (cancelExpiryTask) {
// cancel the expiry task for the transaction
- m_cache->getExpiryTaskManager().cancelTask(
- txState->getSuspendedExpiryTaskId());
- } else {
- m_cache->getExpiryTaskManager().resetTask(
- txState->getSuspendedExpiryTaskId(), std::chrono::seconds::zero());
+ m_cache->getExpiryTaskManager().cancel(txState->getSuspendedExpiryTaskId());
}
// set the current state as the state of the suspended transaction
diff --git a/cppcache/src/CacheTransactionManagerImpl.hpp b/cppcache/src/CacheTransactionManagerImpl.hpp
index e82edd1..18be867 100644
--- a/cppcache/src/CacheTransactionManagerImpl.hpp
+++ b/cppcache/src/CacheTransactionManagerImpl.hpp
@@ -25,7 +25,6 @@
#include <geode/CacheTransactionManager.hpp>
-#include "SuspendedTxExpiryHandler.hpp"
#include "TXCommitMessage.hpp"
#include "util/Log.hpp"
@@ -36,6 +35,8 @@ namespace client {
enum status { STATUS_COMMITTED = 3, STATUS_ROLLEDBACK = 4 };
enum commitOp { BEFORE_COMMIT, AFTER_COMMIT };
+class ThinClientPoolDM;
+
class CacheTransactionManagerImpl : public virtual CacheTransactionManager {
public:
explicit CacheTransactionManagerImpl(CacheImpl* cache);
diff --git a/cppcache/src/ClientMetadataService.cpp b/cppcache/src/ClientMetadataService.cpp
index d844883..3b46e35 100644
--- a/cppcache/src/ClientMetadataService.cpp
+++ b/cppcache/src/ClientMetadataService.cpp
@@ -24,6 +24,7 @@
#include <geode/FixedPartitionResolver.hpp>
+#include "CacheImpl.hpp"
#include "ClientMetadata.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrMessage.hpp"
diff --git a/cppcache/src/CqEventImpl.cpp b/cppcache/src/CqEventImpl.cpp
index ada5619..a824fa6 100644
--- a/cppcache/src/CqEventImpl.cpp
+++ b/cppcache/src/CqEventImpl.cpp
@@ -19,6 +19,7 @@
#include <geode/CacheableString.hpp>
+#include "CacheImpl.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrMessage.hpp"
#include "ThinClientCacheDistributionManager.hpp"
diff --git a/cppcache/src/CqQueryImpl.cpp b/cppcache/src/CqQueryImpl.cpp
index 2a5ddc0..89e1e9e 100644
--- a/cppcache/src/CqQueryImpl.cpp
+++ b/cppcache/src/CqQueryImpl.cpp
@@ -21,9 +21,11 @@
#include <geode/CqAttributesMutator.hpp>
#include <geode/ExceptionTypes.hpp>
+#include "CacheImpl.hpp"
#include "ResultSetImpl.hpp"
#include "StructSetImpl.hpp"
#include "TcrConnectionManager.hpp"
+#include "ThinClientBaseDM.hpp"
#include "ThinClientRegion.hpp"
#include "UserAttributes.hpp"
#include "util/Log.hpp"
@@ -38,7 +40,7 @@ CqQueryImpl::CqQueryImpl(
const std::shared_ptr<CqService>& cqService, const std::string& cqName,
const std::string& queryString,
const std::shared_ptr<CqAttributes>& cqAttributes,
- StatisticsFactory* factory, const bool isDurable,
+ statistics::StatisticsFactory* factory, const bool isDurable,
const std::shared_ptr<UserAttributes>& userAttributesPtr)
: m_cqName(cqName),
m_queryString(queryString),
diff --git a/cppcache/src/CqService.cpp b/cppcache/src/CqService.cpp
index faad396..ccbd6c6 100644
--- a/cppcache/src/CqService.cpp
+++ b/cppcache/src/CqService.cpp
@@ -24,6 +24,7 @@
#include <geode/ExceptionTypes.hpp>
#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "CqEventImpl.hpp"
#include "CqQueryImpl.hpp"
#include "TcrConnectionManager.hpp"
diff --git a/cppcache/src/EntriesMapFactory.cpp b/cppcache/src/EntriesMapFactory.cpp
index bd9edbe..6ff8165 100644
--- a/cppcache/src/EntriesMapFactory.cpp
+++ b/cppcache/src/EntriesMapFactory.cpp
@@ -17,14 +17,13 @@
#include "EntriesMapFactory.hpp"
-#include <geode/Cache.hpp>
#include <geode/DiskPolicyType.hpp>
+#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "ExpMapEntry.hpp"
#include "LRUEntriesMap.hpp"
#include "LRUExpMapEntry.hpp"
-//#include <geode/ExpirationAction.hpp>
-#include <geode/SystemProperties.hpp>
namespace apache {
namespace geode {
diff --git a/cppcache/src/EntryExpiryHandler.cpp b/cppcache/src/EntryExpiryHandler.cpp
deleted file mode 100644
index 7b82dc6..0000000
--- a/cppcache/src/EntryExpiryHandler.cpp
+++ /dev/null
@@ -1,150 +0,0 @@
-/*
- * 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 "EntryExpiryHandler.hpp"
-
-#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
-#include "RegionInternal.hpp"
-
-namespace apache {
-namespace geode {
-namespace client {
-
-EntryExpiryHandler::EntryExpiryHandler(std::shared_ptr<RegionInternal>& rptr,
- std::shared_ptr<MapEntryImpl>& entryPtr,
- ExpirationAction action,
- std::chrono::seconds duration)
- : m_regionPtr(rptr),
- m_entryPtr(entryPtr),
- m_action(action),
- m_duration(duration) {}
-
-int EntryExpiryHandler::handle_timeout(const ACE_Time_Value& current_time,
- const void*) {
- std::shared_ptr<CacheableKey> key;
- m_entryPtr->getKeyI(key);
- ExpEntryProperties& expProps = m_entryPtr->getExpProperties();
- try {
- auto curr_time = std::chrono::system_clock::from_time_t(current_time.sec());
-
- auto lastTimeForExp = expProps.getLastAccessTime();
- if (m_regionPtr->getAttributes().getEntryTimeToLive() >
- std::chrono::seconds::zero()) {
- lastTimeForExp = expProps.getLastModifiedTime();
- }
-
- auto elapsed = curr_time - lastTimeForExp;
-
- LOGDEBUG("Entered entry expiry task handler for key [%s] of region [%s]",
- Utils::nullSafeToString(key).c_str(),
- m_regionPtr->getFullPath().c_str());
- if (elapsed >= m_duration) {
- DoTheExpirationAction(key);
- } else {
- // reset the task after
- // (lastAccessTime + entryExpiryDuration - curr_time) in seconds
- auto remaining = m_duration - elapsed;
- LOGDEBUG("Resetting expiry task for key [%s] of region [%s]",
- Utils::nullSafeToString(key).c_str(),
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->getCacheImpl()->getExpiryTaskManager().resetTask(
- expProps.getExpiryTaskId(), remaining);
- return 0;
- }
- } catch (...) {
- // Ignore whatever exception comes
- }
- LOGDEBUG("Removing expiry task for key [%s] of region [%s]",
- Utils::nullSafeToString(key).c_str(),
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->getCacheImpl()->getExpiryTaskManager().resetTask(
- expProps.getExpiryTaskId(), std::chrono::seconds::zero());
- // we now delete the handler in GF_Timer_Heap_ImmediateReset_T
- // and always return success.
-
- // set the invalid taskid as we have removed the expiry task
- expProps.setExpiryTaskId(-1);
- return 0;
-}
-
-int EntryExpiryHandler::handle_close(ACE_HANDLE, ACE_Reactor_Mask) {
- delete this;
- return 0;
-}
-
-inline void EntryExpiryHandler::DoTheExpirationAction(
- const std::shared_ptr<CacheableKey>& key) {
- // Pass a blank version tag.
- std::shared_ptr<VersionTag> versionTag;
- switch (m_action) {
- case ExpirationAction::INVALIDATE: {
- LOGDEBUG(
- "EntryExpiryHandler::DoTheExpirationAction INVALIDATE "
- "for region %s entry with key %s",
- m_regionPtr->getFullPath().c_str(),
- Utils::nullSafeToString(key).c_str());
- m_regionPtr->invalidateNoThrow(key, nullptr, -1,
- CacheEventFlags::EXPIRATION, versionTag);
- break;
- }
- case ExpirationAction::LOCAL_INVALIDATE: {
- LOGDEBUG(
- "EntryExpiryHandler::DoTheExpirationAction LOCAL_INVALIDATE "
- "for region %s entry with key %s",
- m_regionPtr->getFullPath().c_str(),
- Utils::nullSafeToString(key).c_str());
- m_regionPtr->invalidateNoThrow(
- key, nullptr, -1,
- CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL, versionTag);
- break;
- }
- case ExpirationAction::DESTROY: {
- LOGDEBUG(
- "EntryExpiryHandler::DoTheExpirationAction DESTROY "
- "for region %s entry with key %s",
- m_regionPtr->getFullPath().c_str(),
- Utils::nullSafeToString(key).c_str());
- m_regionPtr->destroyNoThrow(key, nullptr, -1, CacheEventFlags::EXPIRATION,
- versionTag);
- break;
- }
- case ExpirationAction::LOCAL_DESTROY: {
- LOGDEBUG(
- "EntryExpiryHandler::DoTheExpirationAction LOCAL_DESTROY "
- "for region %s entry with key %s",
- m_regionPtr->getFullPath().c_str(),
- Utils::nullSafeToString(key).c_str());
- m_regionPtr->destroyNoThrow(
- key, nullptr, -1,
- CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL, versionTag);
- break;
- }
- case ExpirationAction::INVALID_ACTION: {
- LOGERROR(
- "Unknown expiration action "
- "%d for region %s for key %s",
- m_action, m_regionPtr->getFullPath().c_str(),
- Utils::nullSafeToString(key).c_str());
- break;
- }
- }
-}
-
-} // namespace client
-} // namespace geode
-} // namespace apache
diff --git a/cppcache/src/EntryExpiryHandler.hpp b/cppcache/src/EntryExpiryHandler.hpp
deleted file mode 100644
index 2cb9c19..0000000
--- a/cppcache/src/EntryExpiryHandler.hpp
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.
- */
-
-#pragma once
-
-#ifndef GEODE_ENTRYEXPIRYHANDLER_H_
-#define GEODE_ENTRYEXPIRYHANDLER_H_
-
-#include <geode/Cache.hpp>
-#include <geode/ExpirationAction.hpp>
-#include <geode/Region.hpp>
-#include <geode/internal/geode_globals.hpp>
-
-#include "ExpMapEntry.hpp"
-#include "RegionInternal.hpp"
-
-namespace apache {
-namespace geode {
-namespace client {
-
-/**
- * @class EntryExpiryTask EntryExpiryTask.hpp
- *
- * The task object which contains the handler which gets triggered
- * when an entry expires.
- *
- * TODO: TODO: cleanup region entry nodes and handlers from expiry task
- * manager when region is destroyed
- */
-class EntryExpiryHandler : public ACE_Event_Handler {
- public:
- EntryExpiryHandler(std::shared_ptr<RegionInternal>& rptr,
- std::shared_ptr<MapEntryImpl>& entryPtr,
- ExpirationAction action, std::chrono::seconds duration);
-
- /** This task object will be registered with the Timer Queue.
- * When the timer expires the handle_timeout is invoked.
- */
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
- /**
- * This is called when the task object needs to be cleaned up..
- */
- int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) override;
-
- private:
- // The region which contains the entry
- std::shared_ptr<RegionInternal> m_regionPtr;
- // The ExpMapEntry contained in the ConcurrentMap against the key.
- std::shared_ptr<MapEntryImpl> m_entryPtr;
- // Action to be taken on expiry
- ExpirationAction m_action;
- // Duration after which the task should be reset in case of
- // modification.
- std::chrono::seconds m_duration;
- // perform the actual expiration action
- void DoTheExpirationAction(const std::shared_ptr<CacheableKey>& key);
-};
-
-} // namespace client
-} // namespace geode
-} // namespace apache
-
-#endif // GEODE_ENTRYEXPIRYHANDLER_H_
diff --git a/cppcache/src/EntryExpiryTask.cpp b/cppcache/src/EntryExpiryTask.cpp
new file mode 100644
index 0000000..5e39067
--- /dev/null
+++ b/cppcache/src/EntryExpiryTask.cpp
@@ -0,0 +1,123 @@
+/*
+ * 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 "EntryExpiryTask.hpp"
+
+#include "CacheImpl.hpp"
+#include "RegionInternal.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+
+EntryExpiryTask::EntryExpiryTask(ExpiryTaskManager& manager,
+ std::shared_ptr<RegionInternal> region,
+ std::shared_ptr<MapEntryImpl> entry,
+ ExpirationAction action,
+ const duration_t& duration)
+ : ExpiryTask(manager),
+ duration_(duration),
+ action_(action),
+ entry_(entry),
+ region_(region) {}
+
+EntryExpiryTask::time_point_t EntryExpiryTask::expire_at() const {
+ auto& properties = entry_->getExpProperties();
+ auto last_time = properties.last_accessed();
+ if (region_->getAttributes().getEntryTimeToLive() >
+ std::chrono::seconds::zero()) {
+ last_time = properties.last_modified();
+ }
+
+ return last_time + duration_;
+}
+
+bool EntryExpiryTask::on_expire() {
+ auto tp = expire_at();
+ if (tp > ExpiryTask::clock_t::now()) {
+ // Entry expiration needs to be re-scheduled as it was accessed/modified
+ // since the last time the expiration task was (re-)scheduled.
+ // This is the best approach, rather than re-scheduling the task each time
+ // the entry is accessed/modified, as access/modify is a more frequent
+ // event than expiration.
+ reset(tp);
+ return false;
+ }
+
+ // Pass a blank version tag.
+ std::shared_ptr<CacheableKey> key;
+ std::shared_ptr<VersionTag> versionTag;
+
+ entry_->getKeyI(key);
+
+ const auto& full_path = region_->getFullPath();
+ auto key_str = Utils::nullSafeToString(key);
+
+ switch (action_) {
+ case ExpirationAction::INVALIDATE: {
+ LOGDEBUG(
+ "EntryExpiryTask::DoTheExpirationAction INVALIDATE "
+ "for region " +
+ full_path + " entry with key " + key_str);
+ region_->invalidateNoThrow(key, nullptr, -1, CacheEventFlags::EXPIRATION,
+ versionTag);
+ break;
+ }
+ case ExpirationAction::LOCAL_INVALIDATE: {
+ LOGDEBUG(
+ "EntryExpiryTask::DoTheExpirationAction LOCAL_INVALIDATE "
+ "for region " +
+ full_path + " entry with key " + key_str);
+ region_->invalidateNoThrow(
+ key, nullptr, -1,
+ CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL, versionTag);
+ break;
+ }
+ case ExpirationAction::DESTROY: {
+ LOGDEBUG(
+ "EntryExpiryTask::DoTheExpirationAction DESTROY "
+ "for region " +
+ full_path + " entry with key " + key_str);
+ region_->destroyNoThrow(key, nullptr, -1, CacheEventFlags::EXPIRATION,
+ versionTag);
+ break;
+ }
+ case ExpirationAction::LOCAL_DESTROY: {
+ LOGDEBUG(
+ "EntryExpiryTask::DoTheExpirationAction LOCAL_DESTROY "
+ "for region " +
+ full_path + " entry with key " + key_str);
+ region_->destroyNoThrow(
+ key, nullptr, -1,
+ CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL, versionTag);
+ break;
+ }
+ case ExpirationAction::INVALID_ACTION: {
+ LOGERROR(
+ "Unknown expiration action "
+ "%d for region %s for key %s",
+ action_, full_path.c_str(), key_str.c_str());
+ break;
+ }
+ }
+
+ return true;
+}
+
+} // namespace client
+} // namespace geode
+} // namespace apache
diff --git a/cppcache/src/PreservedDataExpiryHandler.hpp b/cppcache/src/EntryExpiryTask.hpp
similarity index 58%
copy from cppcache/src/PreservedDataExpiryHandler.hpp
copy to cppcache/src/EntryExpiryTask.hpp
index 1563a45..8d2bbc4 100644
--- a/cppcache/src/PreservedDataExpiryHandler.hpp
+++ b/cppcache/src/EntryExpiryTask.hpp
@@ -1,3 +1,8 @@
+#pragma once
+
+#ifndef GEODE_ENTRYEXPIRYTASK_H_
+#define GEODE_ENTRYEXPIRYTASK_H_
+
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -15,49 +20,50 @@
* limitations under the License.
*/
-#pragma once
-
-#ifndef GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
-#define GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
-
#include <geode/Cache.hpp>
-#include <geode/PdxSerializable.hpp>
+#include <geode/ExpirationAction.hpp>
+#include <geode/Region.hpp>
#include <geode/internal/geode_globals.hpp>
-#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
+#include "ExpMapEntry.hpp"
+#include "ExpiryTask.hpp"
+#include "RegionInternal.hpp"
+
+/**
+ * @file
+ */
namespace apache {
namespace geode {
namespace client {
-
/**
- * @class PreservedDataExpiryHandler
+ * @class EntryExpiryTask EntryExpiryTask.hpp
*
* The task object which contains the handler which gets triggered
- * when a preserved data expires.
- *
+ * when an entry expires.
*/
-class PreservedDataExpiryHandler : public ACE_Event_Handler {
+class EntryExpiryTask : public ExpiryTask {
public:
/**
* Constructor
*/
- PreservedDataExpiryHandler(
- const std::shared_ptr<PdxTypeRegistry>& pdxTypeRegistry,
- const std::shared_ptr<PdxSerializable>& pdxObjectPtr);
-
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
+ EntryExpiryTask(ExpiryTaskManager& manager,
+ std::shared_ptr<RegionInternal> region,
+ std::shared_ptr<MapEntryImpl> entry, ExpirationAction action,
+ const duration_t& duration);
- int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) override;
+ protected:
+ bool on_expire() override;
+ time_point_t expire_at() const;
private:
- std::shared_ptr<PdxTypeRegistry> m_pdxTypeRegistry;
- std::shared_ptr<PdxSerializable> m_pdxObjectPtr;
+ duration_t duration_;
+ ExpirationAction action_;
+ std::shared_ptr<MapEntryImpl> entry_;
+ std::shared_ptr<RegionInternal> region_;
};
} // namespace client
} // namespace geode
} // namespace apache
-#endif // GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
+#endif // GEODE_ENTRYEXPIRYTASK_H_
diff --git a/cppcache/src/ExecutionImpl.cpp b/cppcache/src/ExecutionImpl.cpp
index cda2658..c28b961 100644
--- a/cppcache/src/ExecutionImpl.cpp
+++ b/cppcache/src/ExecutionImpl.cpp
@@ -23,6 +23,7 @@
#include <geode/ExceptionTypes.hpp>
#include <geode/internal/geode_globals.hpp>
+#include "CacheImpl.hpp"
#include "NoResult.hpp"
#include "TcrConnectionManager.hpp"
#include "ThinClientPoolDM.hpp"
diff --git a/cppcache/src/ExpEntryProperties.hpp b/cppcache/src/ExpEntryProperties.hpp
new file mode 100644
index 0000000..5270593
--- /dev/null
+++ b/cppcache/src/ExpEntryProperties.hpp
@@ -0,0 +1,95 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#ifndef GEODE_EXPENTRYPROPERTIES_H_
+#define GEODE_EXPENTRYPROPERTIES_H_
+
+#include <atomic>
+#include <memory>
+
+#include <geode/CacheableKey.hpp>
+#include <geode/internal/geode_globals.hpp>
+
+#include "ExpiryTaskManager.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+/**
+ * @brief This class encapsulates expiration specific properties for
+ * a MapEntry.
+ */
+class APACHE_GEODE_EXPORT ExpEntryProperties {
+ public:
+ using time_point = std::chrono::steady_clock::time_point;
+
+ explicit ExpEntryProperties(ExpiryTaskManager* manager) : manager_{manager} {}
+
+ time_point last_accessed() const {
+ return time_point{time_point::duration{last_accessed_}};
+ }
+
+ time_point last_modified() const {
+ return time_point{time_point::duration{last_modified_}};
+ }
+
+ void last_accessed(const time_point& tp) {
+ last_accessed_ = tp.time_since_epoch().count();
+ }
+
+ void last_modified(const time_point& tp) {
+ last_modified_ = tp.time_since_epoch().count();
+ }
+
+ void task_id(ExpiryTask::id_t task_id) { task_id_ = task_id; }
+
+ bool task_scheduled() const { return task_id_ != ExpiryTask::invalid(); }
+
+ void cancel_task() const { manager_->cancel(task_id_); }
+
+ protected:
+ // this constructor deliberately skips initializing any fields
+ inline explicit ExpEntryProperties(bool) {}
+
+ private:
+ /**
+ * Las time the entry was accessed
+ */
+ std::atomic<time_point::duration::rep> last_accessed_{0};
+
+ /**
+ * Last time the entry was modified
+ */
+ std::atomic<time_point::duration::rep> last_modified_{0};
+
+ /**
+ * Pointer to the ExpiryTaskManager
+ */
+ ExpiryTaskManager* manager_{nullptr};
+
+ /**
+ * ID of the expiry task
+ */
+ ExpiryTask::id_t task_id_{ExpiryTask::invalid()};
+};
+} // namespace client
+} // namespace geode
+} // namespace apache
+
+#endif // GEODE_EXPENTRYPROPERTIES_H_
diff --git a/cppcache/src/ExpMapEntry.hpp b/cppcache/src/ExpMapEntry.hpp
index d532089..3dda03e 100644
--- a/cppcache/src/ExpMapEntry.hpp
+++ b/cppcache/src/ExpMapEntry.hpp
@@ -22,7 +22,8 @@
#include <geode/internal/geode_globals.hpp>
-#include "MapEntry.hpp"
+#include "ExpEntryProperties.hpp"
+#include "MapEntryImpl.hpp"
#include "VersionStamp.hpp"
namespace apache {
@@ -43,7 +44,7 @@ class ExpMapEntry : public MapEntryImpl, public ExpEntryProperties {
virtual void cleanup(const CacheEventFlags eventFlags) override {
if (!eventFlags.isExpiration()) {
- cancelExpiryTaskId(m_key);
+ cancel_task();
}
}
diff --git a/cppcache/src/ExpiryHandler_T.hpp b/cppcache/src/ExpiryHandler_T.hpp
deleted file mode 100644
index 8c3c34a..0000000
--- a/cppcache/src/ExpiryHandler_T.hpp
+++ /dev/null
@@ -1,77 +0,0 @@
-#pragma once
-
-#ifndef GEODE_EXPIRYHANDLER_T_H_
-#define GEODE_EXPIRYHANDLER_T_H_
-
-/*
- * 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.
- */
-/**
- *@file ExpiryHandler_T.hpp
- *@since 1.0
- *@version 1.0
- */
-
-#include <ace/Event_Handler.h>
-
-#include "StackTrace.hpp"
-#include "util/Log.hpp"
-namespace apache {
-namespace geode {
-namespace client {
-/**
- * This class manages deletion of the handler so this class should only be
- * used for cases that are cancelled explicitly and never expire otherwise,
- * since GF_Timer_Heap_ImmediateReset_T also deletes handlers that expire
- */
-template <class T>
-class ExpiryHandler_T : public ACE_Event_Handler {
- public:
- /// Handle timeout events.
- typedef int (T::*TO_HANDLER)(const ACE_Time_Value &, const void *);
-
- // op_handler is the receiver of the timeout event. timeout is the method
- // to be executed by op_handler_
- ExpiryHandler_T(T *op_handler, TO_HANDLER timeout)
- : op_handler_(op_handler), to_handler_(timeout) {}
-
- ~ExpiryHandler_T() override = default;
-
- int handle_timeout(const ACE_Time_Value &tv, const void *arg) override {
- return (this->to_handler_ == nullptr
- ? 0
- : (this->op_handler_->*to_handler_)(tv, arg));
- }
-
- int handle_close(ACE_HANDLE, ACE_Reactor_Mask) override {
- // delete the handler so this class can only be used for cases
- // that do not handle deletion themselves and those that never expire
- // otherwise, since GF_Timer_Heap_ImmediateReset_T also deletes handlers
- // that fall out of the queue
- delete this;
- return 0;
- }
-
- private:
- T *op_handler_;
- /// Handle timeout events.
- TO_HANDLER to_handler_;
-};
-} // namespace client
-} // namespace geode
-} // namespace apache
-
-#endif // GEODE_EXPIRYHANDLER_T_H_
diff --git a/cppcache/src/ExpiryTask.cpp b/cppcache/src/ExpiryTask.cpp
new file mode 100644
index 0000000..0ef562e
--- /dev/null
+++ b/cppcache/src/ExpiryTask.cpp
@@ -0,0 +1,69 @@
+/*
+ * 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 "ExpiryTask.hpp"
+
+#include "ExpiryTaskManager.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+
+ExpiryTask::ExpiryTask(ExpiryTaskManager& manager)
+ : id_{invalid()}, timer_{manager.io_context()}, manager_{manager} {}
+
+int32_t ExpiryTask::reset(const std::chrono::nanoseconds& ns) {
+ return reset(timer_t::clock_type::now() + ns);
+}
+
+int32_t ExpiryTask::reset(const time_point_t& at) {
+ std::unique_lock<decltype(mutex_)> lock{mutex_};
+ if (cancelled_) {
+ return -1;
+ }
+
+ auto self = shared_from_this();
+ auto n = timer_.expires_at(at);
+ timer_.async_wait(
+ [self](const boost::system::error_code& e) { self->on_callback(e); });
+ return static_cast<int32_t>(n);
+}
+
+void ExpiryTask::on_callback(const boost::system::error_code& err) {
+ if (cancelled_ || err) {
+ return;
+ }
+
+ if (on_expire()) {
+ if (periodic()) {
+ reset(timer_.expiry() + interval_);
+ } else {
+ manager_.remove(id_);
+ }
+ }
+}
+
+int32_t ExpiryTask::cancel() {
+ std::unique_lock<decltype(mutex_)> lock{mutex_};
+
+ cancelled_ = true;
+ return static_cast<int32_t>(timer_.cancel());
+}
+
+} // namespace client
+} // namespace geode
+} // namespace apache
diff --git a/cppcache/src/ExpiryTask.hpp b/cppcache/src/ExpiryTask.hpp
new file mode 100644
index 0000000..0b893ea
--- /dev/null
+++ b/cppcache/src/ExpiryTask.hpp
@@ -0,0 +1,203 @@
+
+/*
+ * 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.
+ */
+
+#pragma once
+
+#ifndef GEODE_EXPIRYTASK_H_
+#define GEODE_EXPIRYTASK_H_
+
+#include <chrono>
+#include <functional>
+#include <limits>
+#include <memory>
+#include <mutex>
+
+#include <boost/asio/io_context.hpp>
+#include <boost/asio/steady_timer.hpp>
+
+namespace apache {
+namespace geode {
+namespace client {
+
+class ExpiryTaskManager;
+
+/**
+ * @class ExpiryTask ExpiryTask.hpp
+ *
+ * This is the base class for all expiry tasks.
+ */
+class ExpiryTask : public std::enable_shared_from_this<ExpiryTask> {
+ public:
+ using id_t = uint64_t;
+
+ public:
+ /// Class constructors
+
+ /**
+ * Class constructor
+ * @param manager A reference to the ExpiryTaskManager
+ */
+ explicit ExpiryTask(ExpiryTaskManager& manager);
+
+ /**
+ * Class destructor
+ */
+ virtual ~ExpiryTask() = default;
+
+ /**
+ * Returns the task ID
+ */
+ id_t id() const { return id_; }
+
+ /**
+ * Returns whether the task is periodic
+ */
+ bool periodic() const {
+ return interval_ != std::chrono::nanoseconds::zero();
+ }
+
+ /**
+ * Returns an ID which represents an invalid task
+ */
+ static constexpr id_t invalid() { return std::numeric_limits<id_t>::max(); }
+
+ protected:
+ using timer_t = boost::asio::steady_timer;
+ using clock_t = timer_t::clock_type;
+ using time_point_t = timer_t::time_point;
+ using duration_t = time_point_t::duration;
+
+ protected:
+ friend class ExpiryTaskManager;
+
+ /**
+ * Callback called upon task expiration
+ * @return Returns true if the task can normally complete and false if it has
+ * been reset. Specifically if false is returned it will mean:<br>
+ * - For periodic tasks. They won't re-trigger, as the task was
+ * re-scheduled.
+ * - For non-periodic tasks. They won't be removed from the manager,
+ * as the task was re-scheduled.
+ * @note Note that as this is an abstract method, each implementer of
+ * ExpiryTask should write its own version
+ */
+ virtual bool on_expire() = 0;
+
+ /**
+ * Sets task ID
+ * @param id Task ID
+ * @return A reference to itself
+ */
+ ExpiryTask& id(id_t id) {
+ id_ = id;
+ return *this;
+ }
+
+ /**
+ * Sets the task execution interval
+ * @param interval Task execution interval
+ * @return A reference to itself
+ */
+ ExpiryTask& interval(const duration_t& interval) {
+ interval_ = interval;
+ return *this;
+ }
+
+ /**
+ * Cancels the task
+ * @return Returns the number of pending executions. Take into account
+ * that if 0 is returned it means the task was being executed
+ * while cancel was called.
+ */
+ int32_t cancel();
+
+ /**
+ * Resets the task timer
+ * @param at Time point at which the task is to be re-triggered.
+ * @return Returns -1 if the task was cancelled, and otherwise it
+ * returns the number of pending executions. Take into account
+ * that if 0 is returned it means the task was being executed
+ * while reset was called.
+ */
+ int32_t reset(const time_point_t& at);
+
+ /**
+ * Resets the task timer
+ * @param delay Amount of nano-seconds until the task is triggered.
+ * @return Returns -1 if the task was cancelled, and otherwise it
+ * returns the number of pending executions. Take into account
+ * that if 0 is returned it means the task was being executed
+ * while reset was called.
+ */
+ int32_t reset(const duration_t& delay);
+
+ /**
+ * Function triggered by the timer implementation.
+ * @param err Boost error passed by the timer.
+ */
+ void on_callback(const boost::system::error_code& err);
+
+ protected:
+ /// Member attributes
+
+ /**
+ * Unique identifier of the task
+ */
+ id_t id_;
+
+ /**
+ * Specific timer implementation instance
+ */
+ timer_t timer_;
+
+ /**
+ * Reference to the expiry manager
+ */
+ ExpiryTaskManager& manager_;
+
+ /**
+ * Re-triggering interval
+ * @note This is used in order to define periodic tasks
+ */
+ duration_t interval_;
+
+ /**
+ * Exclusive mutex used to avoid race conditions for the following member
+ * functions:<br>
+ * - cancel
+ * - reset
+ */
+ std::mutex mutex_;
+
+ /**
+ * Flag indicating whether or not the task has been cancelled
+ * @note This is necessary in order to guarantee that whenever the task is
+ * reset the task won't be removed from the ExpiryTaskManager.
+ * The reason why is because reset cancels the execution of all the
+ * functions to be called by the timer and given they all share the
+ * same ExpiryTask instance there was no way to tell whether the
+ * task was cancelled or reset.
+ */
+ bool cancelled_{false};
+};
+
+} // namespace client
+} // namespace geode
+} // namespace apache
+
+#endif // GEODE_EXPIRYTASKMANAGER_H_
diff --git a/cppcache/src/ExpiryTaskManager.cpp b/cppcache/src/ExpiryTaskManager.cpp
index bced26e..04dc994 100644
--- a/cppcache/src/ExpiryTaskManager.cpp
+++ b/cppcache/src/ExpiryTaskManager.cpp
@@ -17,92 +17,131 @@
#include "ExpiryTaskManager.hpp"
-#include "DistributedSystem.hpp"
+#include <future>
+
#include "DistributedSystemImpl.hpp"
-#include "config.h"
#include "util/Log.hpp"
-#if defined(_WIN32)
-#include <ace/WFMO_Reactor.h>
-#endif
-#if defined(WITH_ACE_Select_Reactor)
-#include <ace/Select_Reactor.h>
-#else
-#include <ace/Dev_Poll_Reactor.h>
-#endif
+namespace {
+const char *NC_ETM_Thread = "NC ETM Thread";
+}
namespace apache {
namespace geode {
namespace client {
-const char* ExpiryTaskManager::NC_ETM_Thread = "NC ETM Thread";
-
-ExpiryTaskManager::ExpiryTaskManager() : m_reactorEventLoopRunning(false) {
- // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall): ACE
- auto timer = new GF_Timer_Heap_ImmediateReset();
- m_timer = std::unique_ptr<GF_Timer_Heap_ImmediateReset>(timer);
-#if defined(_WIN32)
- m_reactor = new ACE_Reactor(new ACE_WFMO_Reactor(nullptr, m_timer.get()), 1);
-#elif defined(WITH_ACE_Select_Reactor)
- // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall): ACE
- auto aceSelectReactor = new ACE_Select_Reactor(nullptr, m_timer.get());
- // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall): ACE
- m_reactor = new ACE_Reactor(aceSelectReactor, 1);
-#else
- m_reactor =
- new ACE_Reactor(new ACE_Dev_Poll_Reactor(nullptr, m_timer.get()) 1);
-#endif
-}
+ExpiryTaskManager::ExpiryTaskManager()
+ : running_(false),
+ work_guard_(boost::asio::make_work_guard(io_context_)),
+ last_task_id_(0) {}
-int ExpiryTaskManager::resetTask(ExpiryTaskManager::id_type id, uint32_t sec) {
- ACE_Time_Value interval(sec);
- return m_reactor->reset_timer_interval(id, interval);
+ExpiryTaskManager::~ExpiryTaskManager() noexcept {
+ if (running_) {
+ stop();
+ }
}
-int ExpiryTaskManager::cancelTask(ExpiryTaskManager::id_type id) {
- return m_reactor->cancel_timer(id, nullptr, 0);
+void ExpiryTaskManager::start() {
+ if (running_) {
+ throw IllegalStateException(
+ "Tried to start ExpiryTaskManager when it was already running");
+ }
+
+ std::promise<bool> start_promise;
+ auto start_future = start_promise.get_future();
+ runner_ = std::thread{[this, &start_promise] {
+ start_promise.set_value(true);
+ DistributedSystemImpl::setThreadName(NC_ETM_Thread);
+
+ LOGFINE("ExpiryTaskManager thread is running.");
+ io_context_.run();
+ LOGFINE("ExpiryTaskManager thread has stopped.");
+ }};
+
+ running_ = start_future.get();
}
-int ExpiryTaskManager::svc() {
- DistributedSystemImpl::setThreadName(NC_ETM_Thread);
- LOGFINE("ExpiryTaskManager thread is running.");
+void ExpiryTaskManager::stop() {
{
- std::unique_lock<std::mutex> lock(m_mutex);
- m_reactorEventLoopRunning = true;
- m_condition.notify_all();
+ std::unique_lock<std::mutex> lock(mutex_);
+
+ if (!running_) {
+ throw IllegalStateException(
+ "Tried to stop ExpiryTaskManager when it was not running");
+ }
+
+ LOGDEBUG("Stopping ExpiryTaskManager...");
+
+ work_guard_.reset();
+ running_ = false;
+
+ cancel_all();
}
- m_reactor->owner(ACE_OS::thr_self());
- m_reactor->run_reactor_event_loop();
- LOGFINE("ExpiryTaskManager thread has stopped.");
- return 0;
+
+ runner_.join();
}
-void ExpiryTaskManager::stopExpiryTaskManager() {
- std::unique_lock<std::mutex> lock(m_mutex);
+ExpiryTask::id_t ExpiryTaskManager::schedule(
+ std::shared_ptr<ExpiryTask> task, const ExpiryTask::duration_t &delay,
+ const ExpiryTask::duration_t &interval) {
+ std::unique_lock<std::mutex> lock(mutex_);
+
+ if (!running_) {
+ LOGDEBUG("Tried to add a task while ExpiryTaskManager is not running");
+ return ExpiryTask::invalid();
+ }
- if (m_reactorEventLoopRunning) {
- m_reactor->end_reactor_event_loop();
- this->wait();
- m_reactorEventLoopRunning = false;
- m_condition.notify_all();
+ auto task_id = last_task_id_++;
+ if (task_id == ExpiryTask::invalid()) {
+ last_task_id_ = 0;
+ task_id = 0;
}
+
+ task->id(task_id).interval(interval);
+ task_map_.emplace(task_id, std::move(task)).first->second->reset(delay);
+
+ using apache::geode::internal::chrono::duration::to_string;
+ LOGDEBUG("Task %zu has been scheduled in %s with an interval of %s", task_id,
+ to_string(delay).c_str(), to_string(interval).c_str());
+ return task_id;
}
-void ExpiryTaskManager::begin() {
- this->activate();
- std::unique_lock<std::mutex> lock(m_mutex);
- m_condition.wait(lock, [this] { return m_reactorEventLoopRunning; });
+int32_t ExpiryTaskManager::reset(ExpiryTask::id_t task_id,
+ const ExpiryTask::duration_t &delay) {
+ std::unique_lock<std::mutex> lock(mutex_);
+ auto &&iter = task_map_.find(task_id);
+ if (iter == task_map_.end()) {
+ return -1;
+ }
+
+ auto n = iter->second->reset(delay);
+ return static_cast<int32_t>(n);
}
-ExpiryTaskManager::~ExpiryTaskManager() {
- stopExpiryTaskManager();
+int32_t ExpiryTaskManager::cancel(ExpiryTask::id_t task_id) {
+ std::unique_lock<std::mutex> lock(mutex_);
+ auto &&iter = task_map_.find(task_id);
+ if (iter == task_map_.end()) {
+ return -1;
+ }
+
+ auto n = iter->second->cancel();
+ task_map_.erase(iter);
+ return static_cast<int32_t>(n);
+}
- delete m_reactor;
- m_reactor = nullptr;
+void ExpiryTaskManager::cancel_all() {
+ for (auto &&iter : task_map_) {
+ iter.second->cancel();
+ }
- // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall): ACE
+ task_map_.clear();
}
+void ExpiryTaskManager::remove(ExpiryTask::id_t task_id) {
+ std::unique_lock<std::mutex> lock(mutex_);
+ task_map_.erase(task_id);
+}
} // namespace client
} // namespace geode
} // namespace apache
diff --git a/cppcache/src/ExpiryTaskManager.hpp b/cppcache/src/ExpiryTaskManager.hpp
index bbce064..a00ea6b 100644
--- a/cppcache/src/ExpiryTaskManager.hpp
+++ b/cppcache/src/ExpiryTaskManager.hpp
@@ -22,19 +22,13 @@
#include <chrono>
#include <condition_variable>
-#include <memory>
#include <mutex>
-#include <type_traits>
+#include <thread>
-#include <ace/Reactor.h>
-#include <ace/Task.h>
-#include <ace/Timer_Heap.h>
+#include <boost/asio/executor_work_guard.hpp>
+#include <boost/asio/io_context.hpp>
-#include <geode/internal/chrono/duration.hpp>
-#include <geode/internal/geode_globals.hpp>
-
-#include "ReadWriteLock.hpp"
-#include "util/Log.hpp"
+#include "ExpiryTask.hpp"
namespace apache {
namespace geode {
@@ -43,269 +37,164 @@ namespace client {
/**
* @class ExpiryTaskManager ExpiryTaskManager.hpp
*
- * This class starts a reactor's event loop for taking care of expiry
- * tasks. The scheduling of event also happens through this manager.
+ * This class manages all the ExpiryTaskManagers
+ * It uses Boost.Asio and Boost timers to schedule ExpiryTasks
*/
-class ExpiryTaskManager : public ACE_Task_Base {
+class ExpiryTaskManager {
public:
- typedef decltype(std::declval<ACE_Reactor>().schedule_timer(
- nullptr, nullptr, std::declval<ACE_Time_Value>())) id_type;
-
/**
- * This class allows resetting of the timer to take immediate effect when
- * done from inside ACE_Event_Handler::handle_timeout(). With the default
- * implementation in ACE_Timer_Heap, the reset takes affect after the next
- * timeout so expiry does not work correctly with it where we need to
- * reset timer everytime an entry/region is accessed/modified. It also
- * avoids alternative methods like rescheduling eveytime which is
- * expensive in terms of object (ASCE_Timer_Node_T) creation,deletion.
- *
- * Here is the sequence as in ACE_Timer_Heap:
- *
- * 1) expire() is invoked by reactor to remove all expired entries
- *
- * 2) expire() invokes dispatch_info_i() that checks the first element
- * in the heap and removes it, if required.
- *
- * 3) If the element is a recurring timer then the elemant is reinserted
- * into the heap bu dispatch_info_i()
- *
- * 4) expire() then invokes ACE_Event_Handler::handle_timeout() on the
- * element returned (this->upcall calls handle_timeout) by
- * dispatch_info_t(), if any
- *
- * So when reset_timer_interval() is called from inside handle_timeout()
- * the it does not take effect immediately. This implementation reverses
- * the order of steps 3 and 4, so that resetting timer interval from
- * within handle_timeout() works as expected.
- *
- *
+ * Default class constructor
*/
- template <typename TYPE, typename FUNCTOR, typename ACE_LOCK>
- class GF_Timer_Heap_ImmediateReset_T
- : public ACE_Timer_Heap_T<TYPE, FUNCTOR, ACE_LOCK> {
- public:
- int expire_single(ACE_Command_Base& pre_dispatch_command) override {
- ACE_TRACE("GF_Timer_Heap_ImmediateReset_T::expire_single");
- ACE_Timer_Node_Dispatch_Info_T<TYPE> info;
- ACE_Time_Value cur_time;
- ACE_Timer_Node_T<TYPE>* expired = nullptr;
-
- // Create a scope for the lock ...
- {
- ACE_MT(ACE_GUARD_RETURN(ACE_LOCK, ace_mon, this->mutex_, -1))
-
- if (this->is_empty()) return 0;
-
- // Get the current time
- cur_time = this->gettimeofday_static() + this->timer_skew();
-
- expired = this->getFirstNode(cur_time, info);
-
- if (expired == nullptr) return 0;
- }
-
- const void* upcall_act = nullptr;
-
- // Preinvoke (handles refcount if needed, etc.)
- this->preinvoke(info, cur_time, upcall_act);
-
- // Release the token before expiration upcall.
- pre_dispatch_command.execute();
-
- // call the functor
- this->upcall(info, cur_time);
-
- // Postinvoke (undo refcount if needed, etc.)
- this->postinvoke(info, cur_time, upcall_act);
-
- // Create a scope for the lock ...
- {
- ACE_MT(ACE_GUARD_RETURN(ACE_LOCK, ace_mon, this->mutex_, -1))
-
- // Reschedule after doing the upcall in expire method
- // to let updated expiry interval, if any, take effect correctly
- expired = this->remove_first();
+ ExpiryTaskManager();
- // Check if this is an interval timer.
- if (expired->get_interval() > ACE_Time_Value::zero) {
- // Make sure that we skip past values that have already
- // "expired".
- this->recompute_next_abs_interval_time(expired, cur_time);
- // Since this is an interval timer, we need to reschedule
- // it.
- this->reschedule(expired);
- } else {
- // Delete the underlying object
- delete expired->get_type();
- // Call the factory method to free up the node.
- this->free_node(expired);
- }
- }
+ /**
+ * Class destructor
+ */
+ virtual ~ExpiryTaskManager() noexcept;
- // We have dispatched a timer
- return 1;
- }
+ /**
+ * Starts the manager
+ * @throw IllegalStateException An exception is thrown if the manager is
+ * already started.
+ */
+ void start();
- int expire() override {
- return ACE_Timer_Queue_T<TYPE, FUNCTOR, ACE_LOCK>::expire();
- }
+ /**
+ * Stops the manager
+ * @throw IllegalStateException An exception is thrown if the manager is
+ * already stopped.
+ */
+ void stop();
- int expire(const ACE_Time_Value& cur_time) override {
- ACE_TRACE("GF_Timer_Heap_ImmediateReset_T::expire");
- ACE_MT(ACE_GUARD_RETURN(ACE_LOCK, ace_mon, this->mutex_, -1))
+ /**
+ * Returns whether or not the manager is running.
+ */
+ bool running() const { return running_; }
- // Keep looping while there are timers remaining and the earliest
- // timer is <= the <cur_time> passed in to the method.
- if (this->is_empty()) {
- return 0;
- }
- int number_of_timers_expired = 0;
- ACE_Timer_Node_T<TYPE>* expired = nullptr;
- ACE_Timer_Node_Dispatch_Info_T<TYPE> info;
- while ((expired = this->getFirstNode(cur_time, info)) != nullptr) {
- const void* upcall_act = nullptr;
- this->preinvoke(info, cur_time, upcall_act);
+ /**
+ * Returns the number of scheduled tasks.
+ */
+ std::size_t count() const { return task_map_.size(); }
- this->upcall(info, cur_time);
+ /**
+ * Schedules a new expiry task.
+ * @param task Reference to the task.
+ * @param delay Amount of nano-seconds in which the task expires.
+ * @param interval Amount of nano-seconds in which the task is executed again.
+ * If this parameter is set to zero, the task will only
+ * execute once and finish, otherwise it will execute
+ * continuously every interval nano-seconds until cancelled.
+ * @return ID of the created task. If there was any error during the
+ * task creation ExpiryTask::invalid() is returned as ID.
+ * @note Possible errors are:<br>
+ * - Manager is not running
+ */
+ ExpiryTask::id_t schedule(std::shared_ptr<ExpiryTask> task,
+ const ExpiryTask::duration_t &delay,
+ const ExpiryTask::duration_t &interval);
- this->postinvoke(info, cur_time, upcall_act);
+ /**
+ * Schedules a new expiry task.
+ * @param task Task reference.
+ * @param delay Amount of nano-seconds in which the task expires.
+ * @return ID of the created task. If there was any error during the
+ * task creation ExpiryTask::invalid() is returned as ID.
+ * @note Possible errors are:<br>
+ * - Manager is not running
+ */
+ ExpiryTask::id_t schedule(std::shared_ptr<ExpiryTask> task,
+ const ExpiryTask::duration_t &delay) {
+ return schedule(std::move(task), delay, ExpiryTask::duration_t::zero());
+ }
- // reschedule after doing the upcall in expire method
- // to let updated expiry interval, if any, take affect correctly
- expired = this->remove_first();
- // Check if this is an interval timer.
- if (expired->get_interval() > ACE_Time_Value::zero) {
- // Make sure that we skip past values that have already
- // "expired".
- do {
- expired->set_timer_value(expired->get_timer_value() +
- expired->get_interval());
- } while (expired->get_timer_value() <= cur_time);
- // Since this is an interval timer, we need to reschedule
- // it.
- this->reschedule(expired);
- } else {
- // Delete the underlying object
- delete expired->get_type();
- // Call the factory method to free up the node.
- this->free_node(expired);
- }
+ /**
+ * Re-triggers a task to run in the given delay.
+ * @param task_id ID of the task to be re-triggered.
+ * @param delay Amount of nano-seconds in which the task expires.
+ * @return Returns -1 if the task did not existed, and otherwise it
+ * returns the number of pending executions. Take into account
+ * that if 0 is returned it means the task was being executed
+ * while reset was called.
+ */
+ int32_t reset(ExpiryTask::id_t task_id, const ExpiryTask::duration_t &delay);
- ++number_of_timers_expired;
- }
- return number_of_timers_expired;
- }
+ /**
+ * Cancels an already scheduled expiry task
+ * @param task_id ID of the task to be cancelled.
+ * @return Returns -1 if the task did not existed, and otherwise it
+ * returns the number of pending executions. Take into account
+ * that if 0 is returned it means the task was being executed
+ * while cancel was called.
+ */
+ int32_t cancel(ExpiryTask::id_t task_id);
- ACE_Timer_Node_T<TYPE>* getFirstNode(
- const ACE_Time_Value& cur_time,
- ACE_Timer_Node_Dispatch_Info_T<TYPE>& info) {
- ACE_TRACE("GF_Timer_Heap_ImmediateReset_T::getFirstNode");
+ protected:
+ /// Internal types
- if (this->is_empty()) {
- return nullptr;
- }
- ACE_Timer_Node_T<TYPE>* expired = nullptr;
- if (this->earliest_time() <= cur_time) {
- expired = this->get_first();
- // Get the dispatch info
- expired->get_dispatch_info(info);
- return expired;
- }
- return nullptr;
- }
- };
+ using duration_t = std::chrono::nanoseconds;
+ using task_map_t = std::map<ExpiryTask::id_t, std::shared_ptr<ExpiryTask>>;
- typedef GF_Timer_Heap_ImmediateReset_T<
- ACE_Event_Handler*, ACE_Event_Handler_Handle_Timeout_Upcall,
- ACE_SYNCH_RECURSIVE_MUTEX>
- GF_Timer_Heap_ImmediateReset;
+ protected:
+ friend class ExpiryTask;
/**
- * Constructor
+ * Cancels all scheduled tasks
*/
- ExpiryTaskManager();
+ void cancel_all();
/**
- * Destructor. Stops the reactors event loop if it is not running
- * and then exits.
+ * Removes the reference to the task matching the given ID
+ * @param task_id ID of the task
*/
- ~ExpiryTaskManager() override;
+ void remove(ExpiryTask::id_t task_id);
/**
- * For scheduling a task for expiration.
+ * Returns Boost IO context
*/
- template <class ExpRep, class ExpPeriod, class IntRep, class IntPeriod>
- ExpiryTaskManager::id_type scheduleExpiryTask(
- ACE_Event_Handler* handler,
- std::chrono::duration<ExpRep, ExpPeriod> expTime,
- std::chrono::duration<IntRep, IntPeriod> interval,
- bool cancelExistingTask = false) {
- using ::apache::geode::internal::chrono::duration::to_string;
+ boost::asio::io_context &io_context() { return io_context_; }
- LOGFINER(
- "ExpiryTaskManager: expTime %s, interval %s, cancelExistingTask %d",
- to_string(expTime).c_str(), to_string(interval).c_str(),
- cancelExistingTask);
- if (cancelExistingTask) {
- m_reactor->cancel_timer(handler, 1);
- }
+ protected:
+ /// Class member attributes
- ACE_Time_Value expTimeValue(expTime);
- ACE_Time_Value intervalValue(interval);
- LOGFINER("Scheduled expiration ... in " + to_string(expTime));
- return m_reactor->schedule_timer(handler, nullptr, expTimeValue,
- intervalValue);
- }
+ /**
+ * Flag indicating whether or not the manager is running.
+ */
+ bool running_;
/**
- * for resetting the interval an already registered task.
- * returns '0' if successful '-1' on failure.
- * id - the id assigned to the expiry task initially.
- * millisec - The time after which you would like it to get
- * invoked from the current time.
+ * Thread running the io_context.
*/
- int resetTask(id_type id, uint32_t sec);
+ std::thread runner_;
- template <class Rep, class Period>
- int resetTask(id_type id, std::chrono::duration<Rep, Period> duration) {
- ACE_Time_Value interval(duration);
- return m_reactor->reset_timer_interval(id, interval);
- }
+ /*
+ * Boost IO context processing expiry tasks events.
+ */
+ boost::asio::io_context io_context_;
/**
- * for cancelling an already registered task.
- * returns '0' if successful '-1' on failure.
- * id - the id assigned to the expiry task initially.
+ * Executor guard. It keeps the IO context running even when there are
+ * no task scheduled.
*/
- int cancelTask(id_type id);
+ boost::asio::executor_work_guard<boost::asio::io_context::executor_type>
+ work_guard_;
/**
- * A separate thread is started in which the reactor event loop
- * is kept running unless explicitly stopped or when this object
- * goes out of scope.
+ * Manager mutex. It has the following uses:<br>
+ * - Avoids race conditions involving the task container.
+ * - Avoids creation of new task while the expiry task is being stopped.
+ * - Avoids concurrent modification of the task counter.
*/
- int svc() override;
+ std::mutex mutex_;
/**
- * For explicitly stopping the reactor's event loop.
+ * Task container
*/
- void stopExpiryTaskManager();
-
- /** activate the thread and wait for it to be running. */
- void begin();
+ task_map_t task_map_;
- private:
- ACE_Reactor* m_reactor;
-
- bool m_reactorEventLoopRunning; // flag to indicate if the reactor event
- // loop is running or not.
- static const char* NC_ETM_Thread;
-
- std::mutex m_mutex;
- std::condition_variable m_condition;
-
- std::unique_ptr<GF_Timer_Heap_ImmediateReset> m_timer;
+ /**
+ * Task counter. It's used to assign tasks an UID.
+ */
+ ExpiryTask::id_t last_task_id_;
};
} // namespace client
} // namespace geode
diff --git a/cppcache/src/FarSideEntryOp.cpp b/cppcache/src/FarSideEntryOp.cpp
index 794e79f..d2cdf73 100644
--- a/cppcache/src/FarSideEntryOp.cpp
+++ b/cppcache/src/FarSideEntryOp.cpp
@@ -19,7 +19,7 @@
#include "ClientProxyMembershipID.hpp"
#include "DiskVersionTag.hpp"
-#include "RegionCommit.hpp"
+#include "TcrMessage.hpp"
#include "util/exception.hpp"
namespace apache {
@@ -28,7 +28,7 @@ namespace client {
FarSideEntryOp::FarSideEntryOp(
MemberListForVersionStamp& memberListForVersionStamp)
- : m_op(0),
+ : m_op(Operation::MARKER),
m_modSerialNum(0),
m_eventOffset(0),
m_didDestroy(false),
@@ -36,20 +36,22 @@ FarSideEntryOp::FarSideEntryOp(
{}
-bool FarSideEntryOp::isDestroy(int8_t op) {
- return op == DESTROY || op == LOCAL_DESTROY || op == EVICT_DESTROY ||
- op == EXPIRE_DESTROY || op == EXPIRE_LOCAL_DESTROY || op == REMOVE;
+bool FarSideEntryOp::isDestroy(Operation op) {
+ return op == Operation::DESTROY || op == Operation::LOCAL_DESTROY ||
+ op == Operation::EVICT_DESTROY || op == Operation::EXPIRE_DESTROY ||
+ op == Operation::EXPIRE_LOCAL_DESTROY || op == Operation::REMOVE;
}
-bool FarSideEntryOp::isInvalidate(int8_t op) {
- return op == INVALIDATE || op == LOCAL_INVALIDATE ||
- op == EXPIRE_INVALIDATE || op == EXPIRE_LOCAL_INVALIDATE;
+bool FarSideEntryOp::isInvalidate(Operation op) {
+ return op == Operation::INVALIDATE || op == Operation::LOCAL_INVALIDATE ||
+ op == Operation::EXPIRE_INVALIDATE ||
+ op == Operation::EXPIRE_LOCAL_INVALIDATE;
}
void FarSideEntryOp::fromData(DataInput& input, bool largeModCount,
uint16_t memId) {
m_key = std::dynamic_pointer_cast<CacheableKey>(input.readObject());
- m_op = input.read();
+ m_op = static_cast<Operation>(input.read());
if (largeModCount) {
m_modSerialNum = input.readInt32();
} else {
diff --git a/cppcache/src/FarSideEntryOp.hpp b/cppcache/src/FarSideEntryOp.hpp
index a7316a3..4d7426c 100644
--- a/cppcache/src/FarSideEntryOp.hpp
+++ b/cppcache/src/FarSideEntryOp.hpp
@@ -31,60 +31,61 @@ namespace apache {
namespace geode {
namespace client {
-enum OPERATION {
- MARKER = 0,
- CREATE,
- PUTALL_CREATE,
- GET,
- GET_ENTRY,
- CONTAINS_KEY,
- CONTAINS_VALUE,
- CONTAINS_VALUE_FOR_KEY,
- FUNCTION_EXECUTION,
- SEARCH_CREATE,
- LOCAL_LOAD_CREATE,
- NET_LOAD_CREATE,
- UPDATE,
- PUTALL_UPDATE,
- SEARCH_UPDATE,
- LOCAL_LOAD_UPDATE,
- NET_LOAD_UPDATE,
- INVALIDATE,
- LOCAL_INVALIDATE,
- DESTROY,
- LOCAL_DESTROY,
- EVICT_DESTROY,
- REGION_LOAD_SNAPSHOT,
- REGION_LOCAL_DESTROY,
- REGION_CREATE,
- REGION_CLOSE,
- REGION_DESTROY,
- EXPIRE_DESTROY,
- EXPIRE_LOCAL_DESTROY,
- EXPIRE_INVALIDATE,
- EXPIRE_LOCAL_INVALIDATE,
- REGION_EXPIRE_DESTROY,
- REGION_EXPIRE_LOCAL_DESTROY,
- REGION_EXPIRE_INVALIDATE,
- REGION_EXPIRE_LOCAL_INVALIDATE,
- REGION_LOCAL_INVALIDATE,
- REGION_INVALIDATE,
- REGION_CLEAR,
- REGION_LOCAL_CLEAR,
- CACHE_CREATE,
- CACHE_CLOSE,
- FORCED_DISCONNECT,
- REGION_REINITIALIZE,
- CACHE_RECONNECT,
- PUT_IF_ABSENT,
- REPLACE,
- REMOVE
-};
-
class RegionCommit;
class FarSideEntryOp {
public:
+ enum class Operation : int8_t {
+ MARKER = 0,
+ CREATE,
+ PUTALL_CREATE,
+ GET,
+ GET_ENTRY,
+ CONTAINS_KEY,
+ CONTAINS_VALUE,
+ CONTAINS_VALUE_FOR_KEY,
+ FUNCTION_EXECUTION,
+ SEARCH_CREATE,
+ LOCAL_LOAD_CREATE,
+ NET_LOAD_CREATE,
+ UPDATE,
+ PUTALL_UPDATE,
+ SEARCH_UPDATE,
+ LOCAL_LOAD_UPDATE,
+ NET_LOAD_UPDATE,
+ INVALIDATE,
+ LOCAL_INVALIDATE,
+ DESTROY,
+ LOCAL_DESTROY,
+ EVICT_DESTROY,
+ REGION_LOAD_SNAPSHOT,
+ REGION_LOCAL_DESTROY,
+ REGION_CREATE,
+ REGION_CLOSE,
+ REGION_DESTROY,
+ EXPIRE_DESTROY,
+ EXPIRE_LOCAL_DESTROY,
+ EXPIRE_INVALIDATE,
+ EXPIRE_LOCAL_INVALIDATE,
+ REGION_EXPIRE_DESTROY,
+ REGION_EXPIRE_LOCAL_DESTROY,
+ REGION_EXPIRE_INVALIDATE,
+ REGION_EXPIRE_LOCAL_INVALIDATE,
+ REGION_LOCAL_INVALIDATE,
+ REGION_INVALIDATE,
+ REGION_CLEAR,
+ REGION_LOCAL_CLEAR,
+ CACHE_CREATE,
+ CACHE_CLOSE,
+ FORCED_DISCONNECT,
+ REGION_REINITIALIZE,
+ CACHE_RECONNECT,
+ PUT_IF_ABSENT,
+ REPLACE,
+ REMOVE
+ };
+
+ public:
explicit FarSideEntryOp(MemberListForVersionStamp& memberListForVersionStamp);
virtual ~FarSideEntryOp() = default;
@@ -96,7 +97,7 @@ class FarSideEntryOp {
}
private:
- int8_t m_op;
+ Operation m_op;
int32_t m_modSerialNum;
int32_t m_eventOffset;
std::shared_ptr<CacheableKey> m_key;
@@ -105,8 +106,8 @@ class FarSideEntryOp {
std::shared_ptr<Serializable> m_callbackArg;
std::shared_ptr<VersionTag> m_versionTag;
MemberListForVersionStamp& m_memberListForVersionStamp;
- bool isDestroy(int8_t op);
- bool isInvalidate(int8_t op);
+ bool isDestroy(Operation op);
+ bool isInvalidate(Operation op);
void skipFilterRoutingInfo(DataInput& input);
};
diff --git a/cppcache/src/FunctionExpiryTask.hpp b/cppcache/src/FunctionExpiryTask.hpp
new file mode 100644
index 0000000..cff467d
--- /dev/null
+++ b/cppcache/src/FunctionExpiryTask.hpp
@@ -0,0 +1,69 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#ifndef GEODE_FUNCTIONEXPIRYTASK_H_
+#define GEODE_FUNCTIONEXPIRYTASK_H_
+
+#include "ExpiryTask.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+
+/**
+ * @class FunctionExpiryTask FunctionExpiryTask.hpp
+ *
+ * This class is a placeholder for generic expiry tasks
+ * It runs the given callback upon expiration.
+ * Given that the callback is a std::function, it can take anything ranging
+ * from a classical function, a bound function, a lambda, or a instance of a
+ * class implementing operator()
+ */
+class FunctionExpiryTask : public ExpiryTask {
+ public:
+ using callback_t = std::function<void()>;
+
+ public:
+ /// Class constructors
+
+ /**
+ * Class constructor
+ * @param manager A reference to the ExpiryTaskManager
+ * @param callback Callback to be executed by the task
+ * @see ExpiryTask::on_expire for more info about the return value of the
+ * callback.
+ */
+ FunctionExpiryTask(ExpiryTaskManager& manager, callback_t callback)
+ : ExpiryTask(manager), on_expire_(callback) {}
+
+ protected:
+ bool on_expire() override {
+ on_expire_();
+ return true;
+ }
+
+ protected:
+ callback_t on_expire_;
+};
+
+} // namespace client
+} // namespace geode
+} // namespace apache
+
+#endif // GEODE_FUNCTIONEXPIRYTASK_H_
diff --git a/cppcache/src/InternalCacheTransactionManager2PCImpl.cpp b/cppcache/src/InternalCacheTransactionManager2PCImpl.cpp
index 2ac0d9c..0d54b4c 100644
--- a/cppcache/src/InternalCacheTransactionManager2PCImpl.cpp
+++ b/cppcache/src/InternalCacheTransactionManager2PCImpl.cpp
@@ -18,9 +18,9 @@
#include "InternalCacheTransactionManager2PCImpl.hpp"
#include <geode/ExceptionTypes.hpp>
-#include <geode/PoolManager.hpp>
#include <geode/TransactionId.hpp>
+#include "CacheImpl.hpp"
#include "CacheRegionHelper.hpp"
#include "CacheTransactionManagerImpl.hpp"
#include "TXCleaner.hpp"
diff --git a/cppcache/src/LRUAction.hpp b/cppcache/src/LRUAction.hpp
index 58434b2..24d4f02 100644
--- a/cppcache/src/LRUAction.hpp
+++ b/cppcache/src/LRUAction.hpp
@@ -25,7 +25,7 @@
#include <geode/internal/geode_globals.hpp>
#include "CacheableToken.hpp"
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
#include "RegionInternal.hpp"
#include "Utils.hpp"
diff --git a/cppcache/src/LRUExpMapEntry.hpp b/cppcache/src/LRUExpMapEntry.hpp
index ebf5685..6d25d68 100644
--- a/cppcache/src/LRUExpMapEntry.hpp
+++ b/cppcache/src/LRUExpMapEntry.hpp
@@ -22,8 +22,9 @@
#include <geode/internal/geode_globals.hpp>
+#include "ExpEntryProperties.hpp"
#include "LRUEntryProperties.hpp"
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
#include "VersionStamp.hpp"
namespace apache {
@@ -47,7 +48,7 @@ class LRUExpMapEntry : public MapEntryImpl,
void cleanup(const CacheEventFlags eventFlags) override {
if (!eventFlags.isExpiration()) {
- cancelExpiryTaskId(m_key);
+ cancel_task();
}
}
diff --git a/cppcache/src/LRUMapEntry.hpp b/cppcache/src/LRUMapEntry.hpp
index a9bc1f6..46758d6 100644
--- a/cppcache/src/LRUMapEntry.hpp
+++ b/cppcache/src/LRUMapEntry.hpp
@@ -24,7 +24,7 @@
#include <geode/internal/geode_globals.hpp>
#include "LRUEntryProperties.hpp"
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
#include "VersionStamp.hpp"
namespace apache {
diff --git a/cppcache/src/LRUQueue.cpp b/cppcache/src/LRUQueue.cpp
index e9237ee..3b5d386 100644
--- a/cppcache/src/LRUQueue.cpp
+++ b/cppcache/src/LRUQueue.cpp
@@ -20,7 +20,7 @@
#include <mutex>
#include "LRUEntryProperties.hpp"
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
namespace apache {
namespace geode {
diff --git a/cppcache/src/LocalRegion.cpp b/cppcache/src/LocalRegion.cpp
index f61ce71..05e3342 100644
--- a/cppcache/src/LocalRegion.cpp
+++ b/cppcache/src/LocalRegion.cpp
@@ -26,10 +26,10 @@
#include "CacheRegionHelper.hpp"
#include "CacheableToken.hpp"
#include "EntriesMapFactory.hpp"
-#include "EntryExpiryHandler.hpp"
+#include "EntryExpiryTask.hpp"
#include "ExpiryTaskManager.hpp"
#include "LRUEntriesMap.hpp"
-#include "RegionExpiryHandler.hpp"
+#include "RegionExpiryTask.hpp"
#include "RegionGlobalLocks.hpp"
#include "SerializableHelper.hpp"
#include "TXState.hpp"
@@ -53,6 +53,7 @@ LocalRegion::LocalRegion(const std::string& name, CacheImpl* cacheImpl,
m_name(name),
m_parentRegion(rPtr),
m_destroyPending(false),
+ expiry_task_id_(ExpiryTask::invalid()),
m_listener(nullptr),
m_writer(nullptr),
m_loader(nullptr),
@@ -108,24 +109,27 @@ std::shared_ptr<Region> LocalRegion::getParentRegion() const {
void LocalRegion::updateAccessAndModifiedTime(bool modified) {
// locking not required since setters use atomic operations
- if (regionExpiryEnabled()) {
- auto now = std::chrono::system_clock::now();
- auto timeStr = to_string(now.time_since_epoch());
- LOGDEBUG("Setting last accessed time for region %s to %s",
+ if (!regionExpiryEnabled()) {
+ return;
+ }
+
+ auto now = std::chrono::steady_clock::now();
+ auto timeStr = to_string(now.time_since_epoch());
+ LOGDEBUG("Setting last accessed time for region %s to %s",
+ getFullPath().c_str(), timeStr.c_str());
+ m_cacheStatistics->setLastAccessedTime(now);
+ if (modified) {
+ LOGDEBUG("Setting last modified time for region %s to %s",
getFullPath().c_str(), timeStr.c_str());
- m_cacheStatistics->setLastAccessedTime(now);
- if (modified) {
- LOGDEBUG("Setting last modified time for region %s to %s",
- getFullPath().c_str(), timeStr.c_str());
- m_cacheStatistics->setLastModifiedTime(now);
- }
- // TODO: should we really touch the parent region??
- RegionInternal* ri = dynamic_cast<RegionInternal*>(m_parentRegion.get());
- if (ri != nullptr) {
- ri->updateAccessAndModifiedTime(modified);
- }
+ m_cacheStatistics->setLastModifiedTime(now);
+ }
+ // TODO: should we really touch the parent region??
+ RegionInternal* ri = dynamic_cast<RegionInternal*>(m_parentRegion.get());
+ if (ri != nullptr) {
+ ri->updateAccessAndModifiedTime(modified);
}
}
+
std::shared_ptr<CacheStatistics> LocalRegion::getStatistics() const {
CHECK_DESTROY_PENDING(TryReadGuard, LocalRegion::getStatistics);
@@ -686,48 +690,46 @@ void LocalRegion::setPersistenceManager(
}
void LocalRegion::setRegionExpiryTask() {
- if (regionExpiryEnabled()) {
- auto rptr = std::static_pointer_cast<RegionInternal>(shared_from_this());
- const auto& duration = getRegionExpiryDuration();
- auto handler =
- new RegionExpiryHandler(rptr, getRegionExpiryAction(), duration);
- auto expiryTaskId =
- rptr->getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
- handler, duration, std::chrono::seconds::zero());
- handler->setExpiryTaskId(expiryTaskId);
- auto durationStr = to_string(duration);
- auto expiryTaskIdStr = std::to_string(expiryTaskId);
- LOGFINE(
- "expiry for region [%s], expiry task id = %s, duration = %s, "
- "action = %d",
- m_fullPath.c_str(), expiryTaskIdStr.c_str(), durationStr.c_str(),
- getRegionExpiryAction());
+ if (!regionExpiryEnabled()) {
+ return;
}
+
+ auto& manager = getCacheImpl()->getExpiryTaskManager();
+ auto rptr = std::static_pointer_cast<RegionInternal>(shared_from_this());
+ const auto& duration = getRegionExpiryDuration();
+ auto&& task = std::make_shared<RegionExpiryTask>(
+ manager, rptr, getRegionExpiryAction(), duration);
+ expiry_task_id_ = manager.schedule(task, duration);
+ LOGFINE(
+ "expiry for region [%s], expiry task id = %zu, duration = %s, "
+ "action = %d",
+ m_fullPath.c_str(), expiry_task_id_, to_string(duration).c_str(),
+ getRegionExpiryAction());
}
void LocalRegion::registerEntryExpiryTask(
std::shared_ptr<MapEntryImpl>& entry) {
// locking is not required here since only the thread that creates
// the entry will register the expiry task for that entry
+
ExpEntryProperties& expProps = entry->getExpProperties();
- expProps.initStartTime();
- auto rptr = std::static_pointer_cast<RegionInternal>(shared_from_this());
const auto& duration = getEntryExpiryDuration();
- auto handler =
- new EntryExpiryHandler(rptr, entry, getEntryExpirationAction(), duration);
- auto id = rptr->getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
- handler, duration, std::chrono::seconds::zero());
+ auto& manager = getCacheImpl()->getExpiryTaskManager();
+ auto region = std::static_pointer_cast<RegionInternal>(shared_from_this());
+ auto task = std::make_shared<EntryExpiryTask>(
+ manager, region, entry, getEntryExpirationAction(), duration);
+ auto id = manager.schedule(std::move(task), duration);
+ expProps.task_id(id);
+
if (Log::enabled(LogLevel::Finest)) {
std::shared_ptr<CacheableKey> key;
entry->getKeyI(key);
LOGFINEST(
- "entry expiry in region [%s], key [%s], task id = %d, "
+ "entry expiry in region [%s], key [%s], task id = %zu, "
"duration = %s, action = %d",
- m_fullPath.c_str(), Utils::nullSafeToString(key).c_str(),
- static_cast<int32_t>(id), to_string(duration).c_str(),
- getEntryExpirationAction());
+ m_fullPath.c_str(), Utils::nullSafeToString(key).c_str(), id,
+ to_string(duration).c_str(), getEntryExpirationAction());
}
- expProps.setExpiryTaskId(id);
}
LocalRegion::~LocalRegion() noexcept {
@@ -2493,7 +2495,7 @@ GfErrType LocalRegion::putLocal(const std::string& name, bool isCreate,
Utils::nullSafeToString(value).c_str());
// entry/region expiration
if (entryExpiryEnabled()) {
- if (isUpdate && entry->getExpProperties().getExpiryTaskId() != -1) {
+ if (isUpdate && entry->getExpProperties().task_scheduled()) {
updateAccessAndModifiedTimeForEntry(entry, true);
} else {
registerEntryExpiryTask(entry);
@@ -2812,7 +2814,7 @@ void LocalRegion::updateAccessAndModifiedTimeForEntry(
// locking is not required since setters use atomic operations
if (ptr != nullptr && entryExpiryEnabled()) {
ExpEntryProperties& expProps = ptr->getExpProperties();
- auto currTime = std::chrono::system_clock::now();
+ auto now = std::chrono::steady_clock::now();
std::string keyStr;
if (Log::enabled(LogLevel::Debug)) {
std::shared_ptr<CacheableKey> key;
@@ -2821,13 +2823,13 @@ void LocalRegion::updateAccessAndModifiedTimeForEntry(
}
LOGDEBUG("Setting last accessed time for key [%s] in region %s to %s",
keyStr.c_str(), getFullPath().c_str(),
- to_string(currTime.time_since_epoch()).c_str());
- expProps.updateLastAccessTime(currTime);
+ to_string(now.time_since_epoch()).c_str());
+ expProps.last_accessed(now);
if (modified) {
LOGDEBUG("Setting last modified time for key [%s] in region %s to %s",
keyStr.c_str(), getFullPath().c_str(),
- to_string(currTime.time_since_epoch()).c_str());
- expProps.updateLastModifiedTime(currTime);
+ to_string(now.time_since_epoch()).c_str());
+ expProps.last_modified(now);
}
}
}
diff --git a/cppcache/src/LocalRegion.hpp b/cppcache/src/LocalRegion.hpp
index 057276b..9f37af8 100644
--- a/cppcache/src/LocalRegion.hpp
+++ b/cppcache/src/LocalRegion.hpp
@@ -78,6 +78,7 @@ class CreateActions;
class DestroyActions;
class RemoveActions;
class InvalidateActions;
+class VersionedCacheableObjectPartList;
typedef std::unordered_map<std::shared_ptr<CacheableKey>,
std::pair<std::shared_ptr<Cacheable>, int>>
@@ -509,6 +510,7 @@ class APACHE_GEODE_EXPORT LocalRegion : public RegionInternal {
m_subRegions;
std::string m_fullPath;
volatile bool m_destroyPending;
+ ExpiryTask::id_t expiry_task_id_;
std::shared_ptr<CacheListener> m_listener;
std::shared_ptr<CacheWriter> m_writer;
std::shared_ptr<CacheLoader> m_loader;
diff --git a/cppcache/src/MapEntry.hpp b/cppcache/src/MapEntry.hpp
index 256a7d2..6389d2a 100644
--- a/cppcache/src/MapEntry.hpp
+++ b/cppcache/src/MapEntry.hpp
@@ -29,95 +29,18 @@
#include <geode/Serializable.hpp>
#include <geode/internal/geode_globals.hpp>
-#include "CacheImpl.hpp"
-#include "CacheableToken.hpp"
-#include "ExpiryTaskManager.hpp"
#include "RegionInternal.hpp"
-#include "VersionStamp.hpp"
namespace apache {
namespace geode {
namespace client {
+class ExpEntryProperties;
+class LRUEntryProperties;
class MapEntry;
class MapEntryImpl;
+class VersionStamp;
class LRUEntryProperties;
-class CacheImpl;
-
-/**
- * @brief This class encapsulates expiration specific properties for
- * a MapEntry.
- */
-class ExpEntryProperties {
- public:
- ~ExpEntryProperties() noexcept = default;
-
- typedef std::chrono::system_clock::time_point time_point;
-
- inline explicit ExpEntryProperties(ExpiryTaskManager* expiryTaskManager)
- : m_lastAccessTime(0),
- m_lastModifiedTime(0),
- m_expiryTaskId(-1),
- m_expiryTaskManager(expiryTaskManager) {
- // The reactor always gives +ve id while scheduling.
- // -1 will indicate that an expiry task has not been scheduled
- // for this entry. // TODO confirm
- }
-
- inline time_point getLastAccessTime() const {
- return time_point(std::chrono::system_clock::duration(m_lastAccessTime));
- }
-
- inline time_point getLastModifiedTime() const {
- return time_point(std::chrono::system_clock::duration(m_lastModifiedTime));
- }
-
- // moved time initialization outside of constructor to avoid
- // the costly gettimeofday call in MapSegment spinlock
- inline void initStartTime() {
- time_point currentTime = std::chrono::system_clock::now();
- m_lastAccessTime = currentTime.time_since_epoch().count();
- m_lastModifiedTime = currentTime.time_since_epoch().count();
- }
-
- inline void updateLastAccessTime(time_point currTime) {
- m_lastAccessTime = currTime.time_since_epoch().count();
- }
-
- inline void updateLastModifiedTime(time_point currTime) {
- m_lastModifiedTime = currTime.time_since_epoch().count();
- }
-
- inline void setExpiryTaskId(ExpiryTaskManager::id_type id) {
- m_expiryTaskId = id;
- }
-
- inline ExpiryTaskManager::id_type getExpiryTaskId() const {
- return m_expiryTaskId;
- }
-
- inline void cancelExpiryTaskId(
- const std::shared_ptr<CacheableKey>& key) const {
- auto taskIdStr = std::to_string(m_expiryTaskId);
- LOGDEBUG("Cancelling expiration task for key [%s] with id [%s]",
- Utils::nullSafeToString(key).c_str(), taskIdStr.c_str());
- m_expiryTaskManager->cancelTask(m_expiryTaskId);
- }
-
- protected:
- // this constructor deliberately skips initializing any fields
- inline explicit ExpEntryProperties(bool)
- : m_lastAccessTime(0), m_lastModifiedTime(0) {}
-
- private:
- /** last access time in secs, 32bit.. */
- std::atomic<time_point::duration::rep> m_lastAccessTime;
- /** last modified time in secs, 32bit.. */
- std::atomic<time_point::duration::rep> m_lastModifiedTime;
- /** The expiry task id for this particular entry.. **/
- ExpiryTaskManager::id_type m_expiryTaskId;
- ExpiryTaskManager* m_expiryTaskManager;
-};
/**
* @brief Interface class for region mapped entry value.
@@ -197,113 +120,6 @@ class MapEntry {
inline explicit MapEntry(bool) {}
};
-
-/**
- * @brief Hold region mapped entry value. subclass will hold lru flags.
- * Another holds expiration timestamps.
- */
-class MapEntryImpl : public MapEntry,
- public std::enable_shared_from_this<MapEntryImpl> {
- public:
- ~MapEntryImpl() noexcept override = default;
- MapEntryImpl(const MapEntryImpl&) = delete;
- MapEntryImpl& operator=(const MapEntryImpl&) = delete;
-
- inline void getKeyI(std::shared_ptr<CacheableKey>& result) const {
- result = m_key;
- }
-
- inline void getValueI(std::shared_ptr<Cacheable>& result) const {
- // If value is destroyed, then this returns nullptr
- if (CacheableToken::isDestroyed(m_value)) {
- result = nullptr;
- } else {
- result = m_value;
- }
- }
-
- inline void setValueI(const std::shared_ptr<Cacheable>& value) {
- m_value = value;
- }
-
- void getKey(std::shared_ptr<CacheableKey>& result) const override {
- getKeyI(result);
- }
-
- void getValue(std::shared_ptr<Cacheable>& result) const override {
- getValueI(result);
- }
-
- void setValue(const std::shared_ptr<Cacheable>& value) override {
- setValueI(value);
- }
-
- std::shared_ptr<MapEntryImpl> getImplPtr() override {
- return shared_from_this();
- }
-
- LRUEntryProperties& getLRUProperties() override {
- throw FatalInternalException(
- "MapEntry::getLRUProperties called for "
- "non-LRU MapEntry");
- }
-
- ExpEntryProperties& getExpProperties() override {
- throw FatalInternalException(
- "MapEntry::getExpProperties called for "
- "non-expiration MapEntry");
- }
-
- VersionStamp& getVersionStamp() override {
- throw FatalInternalException(
- "MapEntry::getVersionStamp called for "
- "non-versioned MapEntry");
- }
-
- void cleanup(const CacheEventFlags) override {}
-
- protected:
- inline explicit MapEntryImpl(bool) : MapEntry(true) {}
-
- inline explicit MapEntryImpl(const std::shared_ptr<CacheableKey>& key)
- : MapEntry(), m_key(key) {}
-
- std::shared_ptr<Cacheable> m_value;
- std::shared_ptr<CacheableKey> m_key;
-};
-
-class VersionedMapEntryImpl : public MapEntryImpl, public VersionStamp {
- public:
- ~VersionedMapEntryImpl() noexcept override = default;
-
- VersionedMapEntryImpl(const VersionedMapEntryImpl&) = delete;
- VersionedMapEntryImpl& operator=(const VersionedMapEntryImpl&) = delete;
-
- VersionStamp& getVersionStamp() override { return *this; }
-
- protected:
- inline explicit VersionedMapEntryImpl(bool) : MapEntryImpl(true) {}
-
- inline explicit VersionedMapEntryImpl(
- const std::shared_ptr<CacheableKey>& key)
- : MapEntryImpl(key) {}
-};
-
-class EntryFactory {
- public:
- explicit EntryFactory(const bool concurrencyChecksEnabled)
- : m_concurrencyChecksEnabled(concurrencyChecksEnabled) {}
-
- virtual ~EntryFactory() noexcept = default;
-
- virtual void newMapEntry(ExpiryTaskManager* expiryTaskManager,
- const std::shared_ptr<CacheableKey>& key,
- std::shared_ptr<MapEntryImpl>& result) const;
-
- protected:
- bool m_concurrencyChecksEnabled;
-};
-
} // namespace client
} // namespace geode
} // namespace apache
diff --git a/cppcache/src/MapEntry.cpp b/cppcache/src/MapEntryImpl.cpp
similarity index 70%
copy from cppcache/src/MapEntry.cpp
copy to cppcache/src/MapEntryImpl.cpp
index a6667c5..cddf117 100644
--- a/cppcache/src/MapEntry.cpp
+++ b/cppcache/src/MapEntryImpl.cpp
@@ -15,7 +15,9 @@
* limitations under the License.
*/
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
+
+#include <geode/ExceptionTypes.hpp>
#include "MapEntryT.hpp"
@@ -23,6 +25,24 @@ namespace apache {
namespace geode {
namespace client {
+LRUEntryProperties& MapEntryImpl::getLRUProperties() {
+ throw FatalInternalException(
+ "MapEntry::getLRUProperties called for "
+ "non-LRU MapEntry");
+}
+
+ExpEntryProperties& MapEntryImpl::getExpProperties() {
+ throw FatalInternalException(
+ "MapEntry::getExpProperties called for "
+ "non-expiration MapEntry");
+}
+
+VersionStamp& MapEntryImpl::getVersionStamp() {
+ throw FatalInternalException(
+ "MapEntry::getVersionStamp called for "
+ "non-versioned MapEntry");
+}
+
void EntryFactory::newMapEntry(ExpiryTaskManager*,
const std::shared_ptr<CacheableKey>& key,
std::shared_ptr<MapEntryImpl>& result) const {
diff --git a/cppcache/src/MapEntryImpl.hpp b/cppcache/src/MapEntryImpl.hpp
new file mode 100644
index 0000000..23b00a2
--- /dev/null
+++ b/cppcache/src/MapEntryImpl.hpp
@@ -0,0 +1,145 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#ifndef GEODE_MAPENTRYIMPL_H_
+#define GEODE_MAPENTRYIMPL_H_
+
+#include <atomic>
+#include <memory>
+#include <utility>
+
+#include <geode/CacheableKey.hpp>
+#include <geode/Serializable.hpp>
+#include <geode/internal/geode_globals.hpp>
+
+#include "CacheableToken.hpp"
+#include "MapEntry.hpp"
+#include "RegionInternal.hpp"
+#include "VersionStamp.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+
+class CacheImpl;
+class ExpEntryProperties;
+class ExpiryTaskManager;
+class LRUEntryProperties;
+
+/**
+ * @brief Hold region mapped entry value. subclass will hold lru flags.
+ * Another holds expiration timestamps.
+ */
+class MapEntryImpl : public MapEntry,
+ public std::enable_shared_from_this<MapEntryImpl> {
+ public:
+ ~MapEntryImpl() override = default;
+ MapEntryImpl(const MapEntryImpl&) = delete;
+ MapEntryImpl& operator=(const MapEntryImpl&) = delete;
+
+ inline void getKeyI(std::shared_ptr<CacheableKey>& result) const {
+ result = m_key;
+ }
+
+ inline void getValueI(std::shared_ptr<Cacheable>& result) const {
+ // If value is destroyed, then this returns nullptr
+ if (CacheableToken::isDestroyed(m_value)) {
+ result = nullptr;
+ } else {
+ result = m_value;
+ }
+ }
+
+ inline void setValueI(const std::shared_ptr<Cacheable>& value) {
+ m_value = value;
+ }
+
+ void getKey(std::shared_ptr<CacheableKey>& result) const override {
+ getKeyI(result);
+ }
+
+ void getValue(std::shared_ptr<Cacheable>& result) const override {
+ getValueI(result);
+ }
+
+ void setValue(const std::shared_ptr<Cacheable>& value) override {
+ setValueI(value);
+ }
+
+ std::shared_ptr<MapEntryImpl> getImplPtr() override {
+ return shared_from_this();
+ }
+
+ LRUEntryProperties& getLRUProperties() override;
+
+ ExpEntryProperties& getExpProperties() override;
+
+ VersionStamp& getVersionStamp() override;
+
+ void cleanup(const CacheEventFlags) override {}
+
+ protected:
+ inline explicit MapEntryImpl(bool) : MapEntry(true) {}
+
+ inline explicit MapEntryImpl(const std::shared_ptr<CacheableKey>& key)
+ : MapEntry(), m_key(key) {}
+
+ std::shared_ptr<Cacheable> m_value;
+ std::shared_ptr<CacheableKey> m_key;
+};
+
+class APACHE_GEODE_EXPORT VersionedMapEntryImpl : public MapEntryImpl,
+ public VersionStamp {
+ public:
+ ~VersionedMapEntryImpl() override = default;
+
+ VersionStamp& getVersionStamp() override { return *this; }
+
+ protected:
+ inline explicit VersionedMapEntryImpl(bool) : MapEntryImpl(true) {}
+
+ inline explicit VersionedMapEntryImpl(
+ const std::shared_ptr<CacheableKey>& key)
+ : MapEntryImpl(key) {}
+
+ private:
+ // disabled
+ VersionedMapEntryImpl(const VersionedMapEntryImpl&);
+ VersionedMapEntryImpl& operator=(const VersionedMapEntryImpl&);
+};
+
+class APACHE_GEODE_EXPORT EntryFactory {
+ public:
+ explicit EntryFactory(const bool concurrencyChecksEnabled)
+ : m_concurrencyChecksEnabled(concurrencyChecksEnabled) {}
+
+ virtual ~EntryFactory() {}
+
+ virtual void newMapEntry(ExpiryTaskManager* expiryTaskManager,
+ const std::shared_ptr<CacheableKey>& key,
+ std::shared_ptr<MapEntryImpl>& result) const;
+
+ protected:
+ bool m_concurrencyChecksEnabled;
+};
+} // namespace client
+} // namespace geode
+} // namespace apache
+
+#endif // GEODE_MAPENTRYIMPL_H_
diff --git a/cppcache/src/MapSegment.cpp b/cppcache/src/MapSegment.cpp
index 36ef62f..190e005 100644
--- a/cppcache/src/MapSegment.cpp
+++ b/cppcache/src/MapSegment.cpp
@@ -24,8 +24,8 @@
#include "TableOfPrimes.hpp"
#include "ThinClientPoolDM.hpp"
#include "ThinClientRegion.hpp"
-#include "TombstoneExpiryHandler.hpp"
-#include "TrackedMapEntry.hpp"
+#include "TombstoneEntry.hpp"
+#include "TombstoneExpiryTask.hpp"
#include "Utils.hpp"
#include "util/concurrent/spinlock_mutex.hpp"
@@ -34,25 +34,20 @@ namespace geode {
namespace client {
bool MapSegment::boolVal = false;
-MapSegment::~MapSegment() {
- delete m_map;
- // m_entryFactory will be disposed by the containing EntriesMap impl.
-}
void MapSegment::open(RegionInternal* region, const EntryFactory* entryFactory,
ExpiryTaskManager* expiryTaskManager, uint32_t size,
std::atomic<int32_t>* destroyTrackers,
bool concurrencyChecksEnabled) {
- m_map = new CacheableKeyHashMap();
uint32_t mapSize = TableOfPrimes::nextLargerPrime(size, m_primeIndex);
LOGFINER("Initializing MapSegment with size %d (given size %d).", mapSize,
size);
- m_map->reserve(mapSize);
+ m_map.reserve(mapSize);
m_entryFactory = entryFactory;
m_region = region;
m_tombstoneList =
- std::make_shared<TombstoneList>(this, m_region->getCacheImpl());
- m_expiryTaskManager = expiryTaskManager;
+ std::make_shared<TombstoneList>(*this, *m_region->getCacheImpl());
+ expiry_manager_ = expiryTaskManager;
m_numDestroyTrackers = destroyTrackers;
m_concurrencyChecksEnabled = concurrencyChecksEnabled;
}
@@ -61,7 +56,7 @@ void MapSegment::close() {}
void MapSegment::clear() {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- m_map->clear();
+ m_map.clear();
}
void MapSegment::lock() { m_segmentMutex.lock(); }
@@ -74,63 +69,56 @@ GfErrType MapSegment::create(const std::shared_ptr<CacheableKey>& key,
std::shared_ptr<Cacheable>& oldValue,
int updateCount, int destroyTracker,
std::shared_ptr<VersionTag> versionTag) {
- ExpiryTaskManager::id_type taskid = -1;
- TombstoneExpiryHandler* handler = nullptr;
GfErrType err = GF_NOERR;
- {
- std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- // if size is greater than 75 percent of prime, rehash
- auto mapSize = TableOfPrimes::getPrime(m_primeIndex);
- if (((m_map->size() * 75) / 100) > mapSize) {
- rehash();
- }
+ std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
+ // if size is greater than 75 percent of prime, rehash
+ auto mapSize = TableOfPrimes::getPrime(m_primeIndex);
+ if (((m_map.size() * 75) / 100) > mapSize) {
+ rehash();
+ }
- const auto& find = m_map->find(key);
- if (find == m_map->end()) {
- if ((err = putNoEntry(key, newValue, me, updateCount, destroyTracker,
- versionTag)) != GF_NOERR) {
- return err;
- }
- } else {
- auto& entry = find->second;
- auto entryImpl = entry->getImplPtr();
- entryImpl->getValueI(oldValue);
- if (oldValue == nullptr || CacheableToken::isTombstone(oldValue)) {
- // pass the version stamp
- VersionStamp versionStamp;
- if (m_concurrencyChecksEnabled) {
- versionStamp = entry->getVersionStamp();
- if (versionTag) {
- err = versionStamp.processVersionTag(m_region, key, versionTag,
- false);
- if (err != GF_NOERR) return err;
- versionStamp.setVersions(versionTag);
- }
- }
- // good case; go ahead with the create
- if (oldValue == nullptr) {
- err = putForTrackedEntry(key, newValue, entry, entryImpl, updateCount,
- versionStamp);
- } else {
- unguardedRemoveActualEntryWithoutCancelTask(key, handler, taskid);
- err = putNoEntry(key, newValue, me, updateCount, destroyTracker,
- versionTag, &versionStamp);
+ const auto& find = m_map.find(key);
+ if (find == m_map.end()) {
+ if ((err = putNoEntry(key, newValue, me, updateCount, destroyTracker,
+ versionTag)) != GF_NOERR) {
+ return err;
+ }
+ } else {
+ auto& entry = find->second;
+ auto entryImpl = entry->getImplPtr();
+ entryImpl->getValueI(oldValue);
+ if (oldValue == nullptr || CacheableToken::isTombstone(oldValue)) {
+ // pass the version stamp
+ VersionStamp versionStamp;
+ if (m_concurrencyChecksEnabled) {
+ versionStamp = entry->getVersionStamp();
+ if (versionTag) {
+ err =
+ versionStamp.processVersionTag(m_region, key, versionTag, false);
+ if (err != GF_NOERR) return err;
+ versionStamp.setVersions(versionTag);
}
-
- oldValue = nullptr;
-
- } else {
- err = GF_CACHE_ENTRY_EXISTS;
}
- if (err == GF_NOERR) {
- me = entryImpl;
+ // good case; go ahead with the create
+ if (oldValue == nullptr) {
+ err = putForTrackedEntry(key, newValue, entry, entryImpl, updateCount,
+ versionStamp);
+ } else {
+ remove_entry(key);
+ err = putNoEntry(key, newValue, me, updateCount, destroyTracker,
+ versionTag, &versionStamp);
}
+
+ oldValue = nullptr;
+
+ } else {
+ err = GF_CACHE_ENTRY_EXISTS;
+ }
+ if (err == GF_NOERR) {
+ me = entryImpl;
}
}
- if (taskid != -1) {
- m_expiryTaskManager->cancelTask(taskid);
- if (handler != nullptr) delete handler;
- }
+
return err;
}
@@ -144,67 +132,59 @@ GfErrType MapSegment::put(const std::shared_ptr<CacheableKey>& key,
int destroyTracker, bool& isUpdate,
std::shared_ptr<VersionTag> versionTag,
DataInput* delta) {
- ExpiryTaskManager::id_type taskid = -1;
- TombstoneExpiryHandler* handler = nullptr;
GfErrType err = GF_NOERR;
- {
- std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- // if size is greater than 75 percent of prime, rehash
- uint32_t mapSize = TableOfPrimes::getPrime(m_primeIndex);
- if (((m_map->size() * 75) / 100) > mapSize) {
- rehash();
+ std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
+ // if size is greater than 75 percent of prime, rehash
+ uint32_t mapSize = TableOfPrimes::getPrime(m_primeIndex);
+ if (((m_map.size() * 75) / 100) > mapSize) {
+ rehash();
+ }
+
+ const auto& find = m_map.find(key);
+ if (find == m_map.end()) {
+ if (delta != nullptr) {
+ return GF_INVALID_DELTA; // You can not apply delta when there is no
}
+ // entry hence ask for full object
+ isUpdate = false;
+ err =
+ putNoEntry(key, newValue, me, updateCount, destroyTracker, versionTag);
+ } else {
+ auto& entry = find->second;
+ auto entryImpl = entry->getImplPtr();
+ std::shared_ptr<Cacheable> meOldValue;
+ entryImpl->getValueI(meOldValue);
+ // pass the version stamp
+ VersionStamp versionStamp;
+ if (m_concurrencyChecksEnabled) {
+ versionStamp = entry->getVersionStamp();
+ if (versionTag) {
+ if (delta == nullptr) {
+ err =
+ versionStamp.processVersionTag(m_region, key, versionTag, false);
+ } else {
+ err = versionStamp.processVersionTag(m_region, key, versionTag, true);
+ }
- const auto& find = m_map->find(key);
- if (find == m_map->end()) {
- if (delta != nullptr) {
- return GF_INVALID_DELTA; // You can not apply delta when there is no
+ if (err != GF_NOERR) return err;
+ versionStamp.setVersions(versionTag);
}
- // entry hence ask for full object
- isUpdate = false;
+ }
+ if (CacheableToken::isTombstone(meOldValue)) {
+ remove_entry(key);
err = putNoEntry(key, newValue, me, updateCount, destroyTracker,
- versionTag);
- } else {
- auto& entry = find->second;
- auto entryImpl = entry->getImplPtr();
- std::shared_ptr<Cacheable> meOldValue;
- entryImpl->getValueI(meOldValue);
- // pass the version stamp
- VersionStamp versionStamp;
- if (m_concurrencyChecksEnabled) {
- versionStamp = entry->getVersionStamp();
- if (versionTag) {
- if (delta == nullptr) {
- err = versionStamp.processVersionTag(m_region, key, versionTag,
- false);
- } else {
- err =
- versionStamp.processVersionTag(m_region, key, versionTag, true);
- }
-
- if (err != GF_NOERR) return err;
- versionStamp.setVersions(versionTag);
- }
- }
- if (CacheableToken::isTombstone(meOldValue)) {
- unguardedRemoveActualEntryWithoutCancelTask(key, handler, taskid);
- err = putNoEntry(key, newValue, me, updateCount, destroyTracker,
- versionTag, &versionStamp);
- meOldValue = nullptr;
- isUpdate = false;
- } else if ((err = putForTrackedEntry(key, newValue, entry, entryImpl,
- updateCount, versionStamp, delta)) ==
- GF_NOERR) {
- me = entryImpl;
- oldValue = meOldValue;
- isUpdate = (meOldValue != nullptr);
- }
+ versionTag, &versionStamp);
+ meOldValue = nullptr;
+ isUpdate = false;
+ } else if ((err = putForTrackedEntry(key, newValue, entry, entryImpl,
+ updateCount, versionStamp, delta)) ==
+ GF_NOERR) {
+ me = entryImpl;
+ oldValue = meOldValue;
+ isUpdate = (meOldValue != nullptr);
}
}
- if (taskid != -1) {
- m_expiryTaskManager->cancelTask(taskid);
- if (handler != nullptr) delete handler;
- }
+
return err;
}
@@ -217,8 +197,8 @@ GfErrType MapSegment::invalidate(const std::shared_ptr<CacheableKey>& key,
isTokenAdded = false;
GfErrType err = GF_NOERR;
- const auto& find = m_map->find(key);
- if (find != m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find != m_map.end()) {
auto entry = find->second;
VersionStamp versionStamp;
if (m_concurrencyChecksEnabled) {
@@ -261,13 +241,12 @@ GfErrType MapSegment::removeWhenConcurrencyEnabled(
const std::shared_ptr<CacheableKey>& key,
std::shared_ptr<Cacheable>& oldValue, std::shared_ptr<MapEntryImpl>& me,
int updateCount, std::shared_ptr<VersionTag> versionTag, bool afterRemote,
- bool& isEntryFound, ExpiryTaskManager::id_type expiryTaskID,
- TombstoneExpiryHandler* handler, bool& expTaskSet) {
+ bool& isEntryFound) {
GfErrType err = GF_NOERR;
VersionStamp versionStamp;
// If entry found, else return no entry
- const auto& find = m_map->find(key);
- if (find != m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find != m_map.end()) {
auto entry = find->second;
isEntryFound = true;
// If the version tag is null, use the version tag of
@@ -291,8 +270,7 @@ GfErrType MapSegment::removeWhenConcurrencyEnabled(
if ((err = putForTrackedEntry(key, CacheableToken::tombstone(), entry,
entryImpl, updateCount, versionStamp)) ==
GF_NOERR) {
- m_tombstoneList->add(entryImpl, handler, expiryTaskID);
- expTaskSet = true;
+ m_tombstoneList->add(entryImpl);
}
if (CacheableToken::isTombstone(oldValue)) {
oldValue = nullptr;
@@ -310,8 +288,7 @@ GfErrType MapSegment::removeWhenConcurrencyEnabled(
if (versionTag) {
std::shared_ptr<MapEntryImpl> mapEntry;
putNoEntry(key, CacheableToken::tombstone(), mapEntry, -1, 0, versionTag);
- m_tombstoneList->add(mapEntry->getImplPtr(), handler, expiryTaskID);
- expTaskSet = true;
+ m_tombstoneList->add(mapEntry->getImplPtr());
}
oldValue = nullptr;
isEntryFound = false;
@@ -333,28 +310,20 @@ GfErrType MapSegment::remove(const std::shared_ptr<CacheableKey>& key,
std::shared_ptr<VersionTag> versionTag,
bool afterRemote, bool& isEntryFound) {
if (m_concurrencyChecksEnabled) {
- TombstoneExpiryHandler* handler;
- auto id = m_tombstoneList->getExpiryTask(&handler);
- bool expTaskSet = false;
GfErrType err;
{
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
err = removeWhenConcurrencyEnabled(key, oldValue, me, updateCount,
- versionTag, afterRemote, isEntryFound,
- id, handler, expTaskSet);
+ versionTag, afterRemote, isEntryFound);
}
- if (!expTaskSet) {
- m_expiryTaskManager->cancelTask(id);
- delete handler;
- }
return err;
}
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- auto&& iter = m_map->find(key);
- if (iter == m_map->end()) {
+ auto&& iter = m_map.find(key);
+ if (iter == m_map.end()) {
// didn't unbind, probably no entry...
oldValue = nullptr;
volatile int destroyTrackers = *m_numDestroyTrackers;
@@ -365,7 +334,7 @@ GfErrType MapSegment::remove(const std::shared_ptr<CacheableKey>& key,
}
auto entry = iter->second;
- m_map->erase(iter);
+ m_map.erase(iter);
if (updateCount >= 0 && updateCount != entry->getUpdateCount()) {
// this is the case when entry has been updated while being tracked
@@ -382,31 +351,27 @@ GfErrType MapSegment::remove(const std::shared_ptr<CacheableKey>& key,
return GF_NOERR;
}
-bool MapSegment::unguardedRemoveActualEntry(
- const std::shared_ptr<CacheableKey>& key, bool cancelTask) {
- m_tombstoneList->eraseEntryFromTombstoneList(key, cancelTask);
- if (m_map->erase(key) == 0) {
+void MapSegment::remove_entry(const std::shared_ptr<CacheableKey>& key) {
+ m_tombstoneList->erase(key);
+ m_map.erase(key);
+}
+
+bool MapSegment::remove_tomb_entry(
+ const std::shared_ptr<TombstoneEntry>& entry) {
+ std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
+ if (!entry->valid()) {
return false;
}
- return true;
-}
-bool MapSegment::unguardedRemoveActualEntryWithoutCancelTask(
- const std::shared_ptr<CacheableKey>& key, TombstoneExpiryHandler*& handler,
- ExpiryTaskManager::id_type& taskid) {
- std::shared_ptr<MapEntry> entry;
- taskid = m_tombstoneList->eraseEntryFromTombstoneListWithoutCancelTask(
- key, handler);
- if (m_map->erase(key) == 0) {
+ std::shared_ptr<CacheableKey> key;
+ entry->entry()->getKeyI(key);
+
+ if (!m_tombstoneList->erase(key, false)) {
return false;
}
- return true;
-}
-bool MapSegment::removeActualEntry(const std::shared_ptr<CacheableKey>& key,
- bool cancelTask) {
- std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- return unguardedRemoveActualEntry(key, cancelTask);
+ m_map.erase(key);
+ return true;
}
/**
* @brief get MapEntry for key. throws NoEntryException if absent.
@@ -416,8 +381,8 @@ bool MapSegment::getEntry(const std::shared_ptr<CacheableKey>& key,
std::shared_ptr<Cacheable>& value) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- const auto& find = m_map->find(key);
- if (find == m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find == m_map.end()) {
result = nullptr;
value = nullptr;
return false;
@@ -442,8 +407,8 @@ bool MapSegment::getEntry(const std::shared_ptr<CacheableKey>& key,
bool MapSegment::containsKey(const std::shared_ptr<CacheableKey>& key) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- const auto& find = m_map->find(key);
- if (find == m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find == m_map.end()) {
return false;
}
auto mePtr = find->second;
@@ -463,7 +428,7 @@ bool MapSegment::containsKey(const std::shared_ptr<CacheableKey>& key) {
void MapSegment::getKeys(std::vector<std::shared_ptr<CacheableKey>>& result) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- for (const auto& kv : *m_map) {
+ for (const auto& kv : m_map) {
std::shared_ptr<Cacheable> valuePtr;
kv.second->getImplPtr()->getValueI(valuePtr);
if (!CacheableToken::isTombstone(valuePtr)) {
@@ -478,7 +443,7 @@ void MapSegment::getKeys(std::vector<std::shared_ptr<CacheableKey>>& result) {
void MapSegment::getEntries(std::vector<std::shared_ptr<RegionEntry>>& result) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- for (const auto& kv : *m_map) {
+ for (const auto& kv : m_map) {
std::shared_ptr<CacheableKey> keyPtr;
std::shared_ptr<Cacheable> valuePtr;
auto me = kv.second->getImplPtr();
@@ -499,7 +464,7 @@ void MapSegment::getEntries(std::vector<std::shared_ptr<RegionEntry>>& result) {
*/
void MapSegment::getValues(std::vector<std::shared_ptr<Cacheable>>& result) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- for (const auto& kv : *m_map) {
+ for (const auto& kv : m_map) {
auto& entry = kv.second;
std::shared_ptr<Cacheable> value;
entry->getValue(value);
@@ -529,13 +494,13 @@ int MapSegment::addTrackerForEntry(const std::shared_ptr<CacheableKey>& key,
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
std::shared_ptr<MapEntry> entry;
std::shared_ptr<MapEntry> newEntry;
- const auto& find = m_map->find(key);
- if (find == m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find == m_map.end()) {
oldValue = nullptr;
if (addIfAbsent) {
std::shared_ptr<MapEntryImpl> entryImpl;
// add a new entry with value as destroyed
- m_entryFactory->newMapEntry(m_expiryTaskManager, key, entryImpl);
+ m_entryFactory->newMapEntry(expiry_manager_, key, entryImpl);
entryImpl->setValueI(CacheableToken::destroyed());
entry = entryImpl;
newEntry = entryImpl;
@@ -560,8 +525,8 @@ int MapSegment::addTrackerForEntry(const std::shared_ptr<CacheableKey>& key,
updateCount = entry->addTracker(newEntry);
}
if (newEntry) {
- if (find == m_map->end()) {
- m_map->emplace(key, newEntry);
+ if (find == m_map.end()) {
+ m_map.emplace(key, newEntry);
} else {
find->second = newEntry;
}
@@ -577,8 +542,8 @@ void MapSegment::removeTrackerForEntry(
if (m_concurrencyChecksEnabled) return;
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- const auto& find = m_map->find(key);
- if (find != m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find != m_map.end()) {
auto& entry = find->second;
auto impl = entry->getImplPtr();
removeTrackerForEntry(key, entry, impl);
@@ -595,7 +560,7 @@ void MapSegment::addTrackerForAllEntries(
std::shared_ptr<MapEntry> newEntry;
std::shared_ptr<CacheableKey> key;
- for (auto& kv : *m_map) {
+ for (auto& kv : m_map) {
kv.second->getKey(key);
int updateCount = kv.second->addTracker(newEntry);
if (newEntry != nullptr) {
@@ -622,7 +587,7 @@ void MapSegment::rehash() {
// Only called from put, segment must already be locked...
auto newMapSize = TableOfPrimes::getPrime(++m_primeIndex);
LOGFINER("Rehashing MapSegment to size %d.", newMapSize);
- m_map->reserve(newMapSize);
+ m_map.reserve(newMapSize);
m_rehashCount++;
}
std::shared_ptr<Cacheable> MapSegment::getFromDisc(
@@ -706,7 +671,7 @@ GfErrType MapSegment::putForTrackedEntry(
}
if (m_concurrencyChecksEnabled) {
// erase if the entry is in tombstone
- m_tombstoneList->eraseEntryFromTombstoneList(key, true);
+ m_tombstoneList->erase(key, true);
entryImpl->getVersionStamp().setVersions(versionStamp);
}
(void)incrementUpdateCount(key, entry);
@@ -725,11 +690,11 @@ GfErrType MapSegment::putForTrackedEntry(
}
void MapSegment::reapTombstones(std::map<uint16_t, int64_t>& gcVersions) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- m_tombstoneList->reapTombstones(gcVersions);
+ m_tombstoneList->reap_tombstones(gcVersions);
}
void MapSegment::reapTombstones(std::shared_ptr<CacheableHashSet> removedKeys) {
std::lock_guard<decltype(m_spinlock)> lk(m_spinlock);
- m_tombstoneList->reapTombstones(removedKeys);
+ m_tombstoneList->reap_tombstones(removedKeys);
}
GfErrType MapSegment::isTombstone(std::shared_ptr<CacheableKey> key,
@@ -737,8 +702,8 @@ GfErrType MapSegment::isTombstone(std::shared_ptr<CacheableKey> key,
bool& result) {
std::shared_ptr<Cacheable> value;
std::shared_ptr<MapEntryImpl> mePtr;
- const auto& find = m_map->find(key);
- if (find == m_map->end()) {
+ const auto& find = m_map.find(key);
+ if (find == m_map.end()) {
result = false;
return GF_NOERR;
}
@@ -758,8 +723,8 @@ GfErrType MapSegment::isTombstone(std::shared_ptr<CacheableKey> key,
if (CacheableToken::isTombstone(value)) {
if (m_tombstoneList->exists(key)) {
- const auto& findInTombstoneList = m_map->find(key);
- if (findInTombstoneList != m_map->end()) {
+ const auto& findInTombstoneList = m_map.find(key);
+ if (findInTombstoneList != m_map.end()) {
mePtr = findInTombstoneList->second->getImplPtr();
me = mePtr;
}
diff --git a/cppcache/src/MapSegment.hpp b/cppcache/src/MapSegment.hpp
index 29e993d..750ff92 100644
--- a/cppcache/src/MapSegment.hpp
+++ b/cppcache/src/MapSegment.hpp
@@ -31,7 +31,7 @@
#include <geode/internal/geode_globals.hpp>
#include "CacheableToken.hpp"
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
#include "MapWithLock.hpp"
#include "TombstoneList.hpp"
#include "util/concurrent/spinlock_mutex.hpp"
@@ -40,23 +40,26 @@ namespace apache {
namespace geode {
namespace client {
+class EntryFactory;
class RegionInternal;
-typedef std::unordered_map<std::shared_ptr<CacheableKey>,
- std::shared_ptr<MapEntry>,
- dereference_hash<std::shared_ptr<CacheableKey>>,
- dereference_equal_to<std::shared_ptr<CacheableKey>>>
- CacheableKeyHashMap;
/** @brief type wrapper around the std::unordered_map implementation. */
class MapSegment {
private:
+ using CacheableKeyHashMap =
+ std::unordered_map<std::shared_ptr<CacheableKey>,
+ std::shared_ptr<MapEntry>,
+ dereference_hash<std::shared_ptr<CacheableKey>>,
+ dereference_equal_to<std::shared_ptr<CacheableKey>>>;
+
+ private:
// contain
- CacheableKeyHashMap* m_map;
+ CacheableKeyHashMap m_map;
// refers to object managed by the entries map...
// does not need deletion here.
const EntryFactory* m_entryFactory;
RegionInternal* m_region;
- ExpiryTaskManager* m_expiryTaskManager;
+ ExpiryTaskManager* expiry_manager_;
// index of the current prime in the primes table
uint32_t m_primeIndex;
@@ -85,7 +88,7 @@ class MapSegment {
std::shared_ptr<MapEntry> newEntry;
entry->incrementUpdateCount(newEntry);
if (newEntry != nullptr) {
- m_map->emplace(key, newEntry);
+ m_map.emplace(key, newEntry);
entry = newEntry;
return true;
}
@@ -109,13 +112,13 @@ class MapSegment {
entryImpl->getValueI(value);
if (value == nullptr) {
// get rid of an entry marked as destroyed
- m_map->erase(key);
+ m_map.erase(key);
return;
}
}
if (trackerPair.first) {
entry = entryImpl ? entryImpl : entry->getImplPtr();
- (*m_map)[key] = entry;
+ m_map[key] = entry;
}
}
@@ -139,7 +142,7 @@ class MapSegment {
}
}
}
- m_entryFactory->newMapEntry(m_expiryTaskManager, key, newEntry);
+ m_entryFactory->newMapEntry(expiry_manager_, key, newEntry);
newEntry->setValueI(newValue);
if (m_concurrencyChecksEnabled) {
if (versionTag) {
@@ -148,7 +151,7 @@ class MapSegment {
newEntry->getVersionStamp().setVersions(*versionStamp);
}
}
- m_map->emplace(key, newEntry);
+ m_map.emplace(key, newEntry);
return GF_NOERR;
}
@@ -167,15 +170,14 @@ class MapSegment {
const std::shared_ptr<CacheableKey>& key,
std::shared_ptr<Cacheable>& oldValue, std::shared_ptr<MapEntryImpl>& me,
int updateCount, std::shared_ptr<VersionTag> versionTag, bool afterRemote,
- bool& isEntryFound, ExpiryTaskManager::id_type expiryTaskID,
- TombstoneExpiryHandler* handler, bool& expTaskSet);
+ bool& isEntryFound);
public:
MapSegment()
- : m_map(nullptr),
+ : m_map(),
m_entryFactory(nullptr),
m_region(nullptr),
- m_expiryTaskManager(nullptr),
+ expiry_manager_(nullptr),
m_primeIndex(0),
m_spinlock(),
m_segmentMutex(),
@@ -184,8 +186,6 @@ class MapSegment {
m_rehashCount(0),
m_tombstoneList(nullptr) {}
- ~MapSegment();
-
// methods for BasicLockable
void lock();
void unlock();
@@ -283,15 +283,9 @@ class MapSegment {
void reapTombstones(std::shared_ptr<CacheableHashSet> removedKeys);
- bool removeActualEntry(const std::shared_ptr<CacheableKey>& key,
- bool cancelTask = true);
-
- bool unguardedRemoveActualEntryWithoutCancelTask(
- const std::shared_ptr<CacheableKey>& key,
- TombstoneExpiryHandler*& handler, ExpiryTaskManager::id_type& taskid);
+ bool remove_tomb_entry(const std::shared_ptr<TombstoneEntry>& entry);
- bool unguardedRemoveActualEntry(const std::shared_ptr<CacheableKey>& key,
- bool cancelTask = true);
+ void remove_entry(const std::shared_ptr<CacheableKey>& key);
GfErrType isTombstone(std::shared_ptr<CacheableKey> key,
std::shared_ptr<MapEntryImpl>& me, bool& result);
diff --git a/cppcache/src/PdxLocalReader.cpp b/cppcache/src/PdxLocalReader.cpp
index 6229f9d..e391d46 100644
--- a/cppcache/src/PdxLocalReader.cpp
+++ b/cppcache/src/PdxLocalReader.cpp
@@ -22,7 +22,10 @@
#include "PdxLocalReader.hpp"
+#include <geode/TypeRegistry.hpp>
+
#include "PdxTypeRegistry.hpp"
+#include "util/Log.hpp"
namespace apache {
namespace geode {
diff --git a/cppcache/src/PdxRemotePreservedData.hpp b/cppcache/src/PdxRemotePreservedData.hpp
index 711b694..0415a3f 100644
--- a/cppcache/src/PdxRemotePreservedData.hpp
+++ b/cppcache/src/PdxRemotePreservedData.hpp
@@ -31,31 +31,31 @@ namespace apache {
namespace geode {
namespace client {
-class PdxRemotePreservedData;
-
class PdxRemotePreservedData : public PdxUnreadFields {
private:
+ using time_point_t = std::chrono::steady_clock::time_point;
+
+ private:
std::vector<std::vector<int8_t> > m_preservedData;
int32_t m_typeId;
int32_t m_mergedTypeId;
int32_t m_currentIndex;
std::shared_ptr<Serializable> m_owner;
- ExpiryTaskManager::id_type m_expiryTakId;
+ ExpiryTask::id_t expiry_task_id_;
+ time_point_t expires_at_;
public:
- PdxRemotePreservedData()
- : m_typeId(0), m_mergedTypeId(0), m_currentIndex(0), m_expiryTakId(0) {}
-
- ~PdxRemotePreservedData() noexcept override = default;
+ PdxRemotePreservedData() : PdxRemotePreservedData(0, 0, nullptr) {}
PdxRemotePreservedData(int32_t typeId, int32_t mergedTypeId,
- std::shared_ptr<Serializable> owner) {
- m_typeId = typeId;
- m_mergedTypeId = mergedTypeId;
- m_currentIndex = 0;
- m_owner = owner;
- m_expiryTakId = 0;
- }
+ std::shared_ptr<Serializable> owner)
+ : m_typeId(typeId),
+ m_mergedTypeId(mergedTypeId),
+ m_currentIndex(0),
+ m_owner(owner),
+ expiry_task_id_(ExpiryTask::invalid()) {}
+
+ ~PdxRemotePreservedData() noexcept override = default;
void initialize(int32_t typeId, int32_t mergedTypeId,
std::shared_ptr<Serializable> owner) {
@@ -63,18 +63,14 @@ class PdxRemotePreservedData : public PdxUnreadFields {
m_mergedTypeId = mergedTypeId;
m_currentIndex = 0;
m_owner = owner;
- m_expiryTakId = 0;
+ expiry_task_id_ = 0;
}
inline int32_t getMergedTypeId() { return m_mergedTypeId; }
- inline void setPreservedDataExpiryTaskId(ExpiryTaskManager::id_type expId) {
- m_expiryTakId = expId;
- }
+ inline void task_id(ExpiryTask::id_t task_id) { expiry_task_id_ = task_id; }
- inline ExpiryTaskManager::id_type getPreservedDataExpiryTaskId() {
- return m_expiryTakId;
- }
+ inline ExpiryTask::id_t task_id() { return expiry_task_id_; }
std::shared_ptr<Serializable> getOwner() { return m_owner; }
@@ -103,6 +99,9 @@ class PdxRemotePreservedData : public PdxUnreadFields {
}
return 0;
}
+
+ time_point_t expires_at() const { return expires_at_; }
+ void expires_at(const time_point_t& tp) { expires_at_ = tp; }
};
} // namespace client
} // namespace geode
diff --git a/cppcache/src/PdxRemoteWriter.cpp b/cppcache/src/PdxRemoteWriter.cpp
index 301183c..ee5c7ea 100644
--- a/cppcache/src/PdxRemoteWriter.cpp
+++ b/cppcache/src/PdxRemoteWriter.cpp
@@ -24,6 +24,7 @@
#include "PdxRemoteWriter.hpp"
#include "PdxTypeRegistry.hpp"
+#include "util/Log.hpp"
namespace apache {
namespace geode {
diff --git a/cppcache/src/PdxTypeRegistry.cpp b/cppcache/src/PdxTypeRegistry.cpp
index 2d305fa..602a9ef 100644
--- a/cppcache/src/PdxTypeRegistry.cpp
+++ b/cppcache/src/PdxTypeRegistry.cpp
@@ -21,6 +21,7 @@
#include "CacheImpl.hpp"
#include "CacheRegionHelper.hpp"
+#include "PreservedDataExpiryTask.hpp"
#include "ThinClientPoolDM.hpp"
namespace apache {
@@ -39,7 +40,7 @@ PdxTypeRegistry::PdxTypeRegistry(CacheImpl* cache)
PdxTypeRegistry::~PdxTypeRegistry() {}
size_t PdxTypeRegistry::testNumberOfPreservedData() const {
- return preserveData_.size();
+ return preserved_data_.size();
}
int32_t PdxTypeRegistry::getPDXIdForType(const std::string& type, Pool* pool,
@@ -111,7 +112,7 @@ void PdxTypeRegistry::clear() {
}
{
WriteGuard guard(getPreservedDataLock());
- preserveData_.clear();
+ preserved_data_.clear();
}
}
@@ -163,28 +164,29 @@ std::shared_ptr<PdxType> PdxTypeRegistry::getMergedType(
void PdxTypeRegistry::setPreserveData(
std::shared_ptr<PdxSerializable> obj,
- std::shared_ptr<PdxRemotePreservedData> pData,
+ std::shared_ptr<PdxRemotePreservedData> data,
ExpiryTaskManager& expiryTaskManager) {
WriteGuard guard(getPreservedDataLock());
- pData->setOwner(obj);
- if (preserveData_.find(obj) != preserveData_.end()) {
- // reset expiry task
- // TODO: check value for nullptr
- auto expTaskId = preserveData_[obj]->getPreservedDataExpiryTaskId();
- expiryTaskManager.resetTask(expTaskId, 5);
- LOGDEBUG("PdxTypeRegistry::setPreserveData Reset expiry task Done");
- pData->setPreservedDataExpiryTaskId(expTaskId);
- preserveData_[obj] = pData;
+ data->setOwner(obj);
+
+ auto&& iter = preserved_data_.find(obj);
+ if (iter != preserved_data_.end()) {
+ auto expires_at =
+ std::chrono::steady_clock::now() + std::chrono::seconds(5);
+ data->task_id(iter->second->task_id());
+ data->expires_at(expires_at);
+ iter->second = data;
} else {
- // schedule new expiry task
- auto handler = new PreservedDataExpiryHandler(shared_from_this(), obj);
- auto id = expiryTaskManager.scheduleExpiryTask(
- handler, std::chrono::seconds(20), std::chrono::seconds::zero(), false);
- pData->setPreservedDataExpiryTaskId(id);
+ auto task = std::make_shared<PreservedDataExpiryTask>(
+ expiryTaskManager, shared_from_this(), obj);
+ auto id =
+ expiryTaskManager.schedule(std::move(task), std::chrono::seconds(20));
+ data->task_id(id);
+
LOGDEBUG(
- "PdxTypeRegistry::setPreserveData Schedule new expirt task with id=%ld",
+ "PdxTypeRegistry::setPreserveData Schedule new expiry task with id=%zu",
id);
- preserveData_.emplace(obj, pData);
+ preserved_data_.emplace_hint(iter, std::move(obj), std::move(data));
}
LOGDEBUG(
@@ -194,8 +196,8 @@ void PdxTypeRegistry::setPreserveData(
std::shared_ptr<PdxRemotePreservedData> PdxTypeRegistry::getPreserveData(
std::shared_ptr<PdxSerializable> pdxobj) const {
ReadGuard guard(getPreservedDataLock());
- const auto& iter = preserveData_.find((pdxobj));
- if (iter != preserveData_.end()) {
+ const auto& iter = preserved_data_.find((pdxobj));
+ if (iter != preserved_data_.end()) {
return iter->second;
}
return nullptr;
diff --git a/cppcache/src/PdxTypeRegistry.hpp b/cppcache/src/PdxTypeRegistry.hpp
index af69575..f278ce6 100644
--- a/cppcache/src/PdxTypeRegistry.hpp
+++ b/cppcache/src/PdxTypeRegistry.hpp
@@ -31,7 +31,6 @@
#include "ExpiryTaskManager.hpp"
#include "PdxRemotePreservedData.hpp"
#include "PdxType.hpp"
-#include "PreservedDataExpiryHandler.hpp"
#include "ReadWriteLock.hpp"
namespace apache {
@@ -51,6 +50,8 @@ typedef std::unordered_map<std::shared_ptr<PdxType>, int32_t,
dereference_equal_to<std::shared_ptr<PdxType>>>
PdxTypeToTypeIdMap;
+class PreservedDataExpiryTask;
+
class APACHE_GEODE_EXPORT PdxTypeRegistry
: public std::enable_shared_from_this<PdxTypeRegistry> {
private:
@@ -65,7 +66,7 @@ class APACHE_GEODE_EXPORT PdxTypeRegistry
PdxTypeToTypeIdMap pdxTypeToTypeIdMap_;
// TODO:: preserveData need to be of type WeakHashMap
- PreservedHashMap preserveData_;
+ PreservedHashMap preserved_data_;
mutable ACE_RW_Thread_Mutex g_readerWriterLock_;
@@ -103,7 +104,7 @@ class APACHE_GEODE_EXPORT PdxTypeRegistry
std::shared_ptr<PdxType> getMergedType(int32_t remoteTypeId) const;
void setPreserveData(std::shared_ptr<PdxSerializable> obj,
- std::shared_ptr<PdxRemotePreservedData> preserveDataPtr,
+ std::shared_ptr<PdxRemotePreservedData> data,
ExpiryTaskManager& expiryTaskManager);
std::shared_ptr<PdxRemotePreservedData> getPreserveData(
@@ -122,9 +123,7 @@ class APACHE_GEODE_EXPORT PdxTypeRegistry
bool getPdxReadSerialized() const { return pdxReadSerialized_; }
- inline const PreservedHashMap& getPreserveDataMap() const {
- return preserveData_;
- }
+ const PreservedHashMap& getPreserveDataMap() const { return preserved_data_; }
int32_t getEnumValue(std::shared_ptr<EnumInfo> ei);
@@ -135,6 +134,11 @@ class APACHE_GEODE_EXPORT PdxTypeRegistry
ACE_RW_Thread_Mutex& getPreservedDataLock() const {
return g_preservedDataLock_;
}
+
+ protected:
+ friend class PreservedDataExpiryTask;
+
+ PreservedHashMap& preserved_data_map() { return preserved_data_; }
};
} // namespace client
diff --git a/cppcache/src/Pool.cpp b/cppcache/src/Pool.cpp
index 9983071..795cc23 100644
--- a/cppcache/src/Pool.cpp
+++ b/cppcache/src/Pool.cpp
@@ -20,6 +20,7 @@
#include <geode/CacheFactory.hpp>
#include <geode/Pool.hpp>
+#include "CacheImpl.hpp"
#include "PoolAttributes.hpp"
#include "TcrConnectionManager.hpp"
#include "ThinClientPoolHADM.hpp"
diff --git a/cppcache/src/PreservedDataExpiryHandler.cpp b/cppcache/src/PreservedDataExpiryTask.cpp
similarity index 52%
rename from cppcache/src/PreservedDataExpiryHandler.cpp
rename to cppcache/src/PreservedDataExpiryTask.cpp
index 32fd8c5..96d4486 100644
--- a/cppcache/src/PreservedDataExpiryHandler.cpp
+++ b/cppcache/src/PreservedDataExpiryTask.cpp
@@ -15,48 +15,49 @@
* limitations under the License.
*/
/*
- * PreservedDataExpiryHandler.cpp
+ * PreservedDataExpiryTask.cpp
*
* Created on: Apr 5, 2012
* Author: npatel
*/
-#include "PreservedDataExpiryHandler.hpp"
+#include "PreservedDataExpiryTask.hpp"
+#include "ExpiryTaskManager.hpp"
#include "PdxTypeRegistry.hpp"
namespace apache {
namespace geode {
namespace client {
-PreservedDataExpiryHandler::PreservedDataExpiryHandler(
- const std::shared_ptr<PdxTypeRegistry>& pdxTypeRegistry,
- const std::shared_ptr<PdxSerializable>& pdxObjectPtr)
- : m_pdxTypeRegistry(pdxTypeRegistry), m_pdxObjectPtr(pdxObjectPtr) {}
+PreservedDataExpiryTask::PreservedDataExpiryTask(
+ ExpiryTaskManager& manager, decltype(type_registry_) type_registry,
+ decltype(object_) object)
+ : ExpiryTask(manager), type_registry_(type_registry), object_(object) {}
+
+bool PreservedDataExpiryTask::on_expire() {
+ WriteGuard guard(type_registry_->getPreservedDataLock());
+ auto& map = type_registry_->preserved_data_map();
-int PreservedDataExpiryHandler::handle_timeout(const ACE_Time_Value&,
- const void*) {
- WriteGuard guard(m_pdxTypeRegistry->getPreservedDataLock());
- auto map = m_pdxTypeRegistry->getPreserveDataMap();
LOGDEBUG(
- "Entered PreservedDataExpiryHandler "
+ "Entered PreservedDataExpiryTask "
"PdxTypeRegistry::getPreserveDataMap().size() = %zu",
map.size());
- try {
- // remove the entry from the map
- map.erase(m_pdxObjectPtr);
- } catch (...) {
- // Ignore whatever exception comes
- LOGDEBUG(
- "PreservedDataExpiry:: Error while Clearing PdxObject and its "
- "preserved data. Ignoring the error");
+ auto&& iter = map.find(object_);
+ if (iter == map.end()) {
+ return true;
+ }
+
+ auto expires_at = iter->second->expires_at();
+ if (expires_at < ExpiryTask::clock_t::now()) {
+ LOGDEBUG("Re-scheduling PreservedDataExpiryTask with ID %zu", id());
+
+ reset(expires_at);
+ return false;
}
- return 0;
-}
-int PreservedDataExpiryHandler::handle_close(ACE_HANDLE, ACE_Reactor_Mask) {
- delete this;
- return 0;
+ map.erase(iter);
+ return true;
}
} // namespace client
diff --git a/cppcache/src/PreservedDataExpiryHandler.hpp b/cppcache/src/PreservedDataExpiryTask.hpp
similarity index 58%
copy from cppcache/src/PreservedDataExpiryHandler.hpp
copy to cppcache/src/PreservedDataExpiryTask.hpp
index 1563a45..c14efd1 100644
--- a/cppcache/src/PreservedDataExpiryHandler.hpp
+++ b/cppcache/src/PreservedDataExpiryTask.hpp
@@ -17,47 +17,42 @@
#pragma once
-#ifndef GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
-#define GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
+#ifndef GEODE_PRESERVEDDATAEXPIRYTASK_H_
+#define GEODE_PRESERVEDDATAEXPIRYTASK_H_
#include <geode/Cache.hpp>
#include <geode/PdxSerializable.hpp>
#include <geode/internal/geode_globals.hpp>
#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
+#include "ExpiryTask.hpp"
namespace apache {
namespace geode {
namespace client {
/**
- * @class PreservedDataExpiryHandler
- *
- * The task object which contains the handler which gets triggered
- * when a preserved data expires.
+ * @class PreservedDataExpiryTask
*
+ * The expiry task which gets triggered when a preserved data expires.
*/
-class PreservedDataExpiryHandler : public ACE_Event_Handler {
+class PreservedDataExpiryTask : public ExpiryTask {
public:
/**
* Constructor
*/
- PreservedDataExpiryHandler(
- const std::shared_ptr<PdxTypeRegistry>& pdxTypeRegistry,
- const std::shared_ptr<PdxSerializable>& pdxObjectPtr);
-
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
+ PreservedDataExpiryTask(ExpiryTaskManager& manager,
+ std::shared_ptr<PdxTypeRegistry> type_registry,
+ std::shared_ptr<PdxSerializable> object);
- int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) override;
+ bool on_expire() override;
private:
- std::shared_ptr<PdxTypeRegistry> m_pdxTypeRegistry;
- std::shared_ptr<PdxSerializable> m_pdxObjectPtr;
+ std::shared_ptr<PdxTypeRegistry> type_registry_;
+ std::shared_ptr<PdxSerializable> object_;
};
} // namespace client
} // namespace geode
} // namespace apache
-#endif // GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
+#endif // GEODE_PRESERVEDDATAEXPIRYTASK_H_
diff --git a/cppcache/src/ProxyRemoteQueryService.cpp b/cppcache/src/ProxyRemoteQueryService.cpp
index 2174a73..37c5ac0 100644
--- a/cppcache/src/ProxyRemoteQueryService.cpp
+++ b/cppcache/src/ProxyRemoteQueryService.cpp
@@ -19,6 +19,7 @@
#include <geode/PoolManager.hpp>
+#include "CacheImpl.hpp"
#include "CqQueryImpl.hpp"
#include "ThinClientPoolDM.hpp"
diff --git a/cppcache/src/RegionExpiryHandler.cpp b/cppcache/src/RegionExpiryHandler.cpp
deleted file mode 100644
index 8187874..0000000
--- a/cppcache/src/RegionExpiryHandler.cpp
+++ /dev/null
@@ -1,136 +0,0 @@
-/*
- * 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 "RegionExpiryHandler.hpp"
-
-#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
-#include "RegionInternal.hpp"
-
-namespace apache {
-namespace geode {
-namespace client {
-
-RegionExpiryHandler::RegionExpiryHandler(std::shared_ptr<RegionInternal>& rptr,
- ExpirationAction action,
- std::chrono::seconds duration)
- : m_regionPtr(rptr),
- m_action(action),
- m_duration(duration),
- /* adongre
- * CID 28941: Uninitialized scalar field (UNINIT_CTOR)
- */
- m_expiryTaskId(0) {}
-
-int RegionExpiryHandler::handle_timeout(const ACE_Time_Value& current_time,
- const void*) {
- auto curr_time = std::chrono::system_clock::from_time_t(current_time.sec());
- try {
- auto statistics = m_regionPtr->getStatistics();
- auto lastTimeForExp = statistics->getLastAccessedTime();
- if (m_regionPtr->getAttributes().getRegionTimeToLive() >
- std::chrono::seconds::zero()) {
- lastTimeForExp = statistics->getLastModifiedTime();
- }
-
- auto elapsed = curr_time - lastTimeForExp;
- LOGDEBUG("Entered region expiry task handler for region [%s]: %s,%s,%s,%s",
- m_regionPtr->getFullPath().c_str(),
- to_string(curr_time.time_since_epoch()).c_str(),
- to_string(lastTimeForExp.time_since_epoch()).c_str(),
- to_string(m_duration).c_str(), to_string(elapsed).c_str());
- if (elapsed >= m_duration) {
- DoTheExpirationAction();
- } else {
- auto remaining = m_duration - elapsed;
- // reset the task after
- // (lastAccessTime + entryExpiryDuration - curr_time) in seconds
- LOGDEBUG("Resetting expiry task for region [%s] after %s sec",
- m_regionPtr->getFullPath().c_str(),
- to_string(remaining).c_str());
- m_regionPtr->getCacheImpl()->getExpiryTaskManager().resetTask(
- m_expiryTaskId, remaining);
- return 0;
- }
- LOGDEBUG("Removing expiry task for region [%s]",
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->getCacheImpl()->getExpiryTaskManager().resetTask(
- m_expiryTaskId, std::chrono::seconds::zero());
- } catch (...) {
- // Ignore whatever exception comes
- }
- // we now delete the handler in GF_Timer_Heap_ImmediateReset_T
- // and always return success.
- return 0;
-}
-
-int RegionExpiryHandler::handle_close(ACE_HANDLE, ACE_Reactor_Mask) {
- delete this;
- return 0;
-}
-
-void RegionExpiryHandler::DoTheExpirationAction() {
- switch (m_action) {
- case ExpirationAction::INVALIDATE: {
- LOGDEBUG(
- "RegionExpiryHandler::DoTheExpirationAction INVALIDATE "
- "region [%s]",
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->invalidateRegionNoThrow(nullptr,
- CacheEventFlags::EXPIRATION);
- break;
- }
- case ExpirationAction::LOCAL_INVALIDATE: {
- LOGDEBUG(
- "RegionExpiryHandler::DoTheExpirationAction LOCAL_INVALIDATE "
- "region [%s]",
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->invalidateRegionNoThrow(
- nullptr, CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL);
- break;
- }
- case ExpirationAction::DESTROY: {
- LOGDEBUG(
- "RegionExpiryHandler::DoTheExpirationAction DESTROY "
- "region [%s]",
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->destroyRegionNoThrow(nullptr, true,
- CacheEventFlags::EXPIRATION);
- break;
- }
- case ExpirationAction::LOCAL_DESTROY: {
- LOGDEBUG(
- "RegionExpiryHandler::DoTheExpirationAction LOCAL_DESTROY "
- "region [%s]",
- m_regionPtr->getFullPath().c_str());
- m_regionPtr->destroyRegionNoThrow(
- nullptr, true, CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL);
- break;
- }
- case ExpirationAction::INVALID_ACTION: {
- LOGERROR(
- "Unknown expiration action "
- "%d for region [%s]",
- m_action, m_regionPtr->getFullPath().c_str());
- break;
- }
- }
-}
-
-} // namespace client
-} // namespace geode
-} // namespace apache
diff --git a/cppcache/src/RegionExpiryTask.cpp b/cppcache/src/RegionExpiryTask.cpp
new file mode 100644
index 0000000..198278e
--- /dev/null
+++ b/cppcache/src/RegionExpiryTask.cpp
@@ -0,0 +1,95 @@
+/*
+ * 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 "RegionExpiryTask.hpp"
+
+#include "CacheImpl.hpp"
+#include "RegionInternal.hpp"
+
+namespace apache {
+namespace geode {
+namespace client {
+
+RegionExpiryTask::RegionExpiryTask(ExpiryTaskManager& manager,
+ std::shared_ptr<RegionInternal> region,
+ ExpirationAction action,
+ const duration_t& duration)
+ : ExpiryTask(manager),
+ region_(region),
+ duration_(duration),
+ action_(action) {}
+
+ExpiryTask::time_point_t RegionExpiryTask::expire_at() const {
+ auto statistics = region_->getStatistics();
+ auto last_time = statistics->getLastAccessedSteadyTime();
+ if (region_->getAttributes().getRegionTimeToLive() >
+ std::chrono::seconds::zero()) {
+ last_time = statistics->getLastModifiedSteadyTime();
+ }
+
+ return last_time + duration_;
+}
+
+bool RegionExpiryTask::on_expire() {
+ auto tp = expire_at();
+ if (tp > ExpiryTask::clock_t::now()) {
+ // Region expiration needs to be re-scheduled as it was accessed/modified
+ // since the last time the expiration task was (re-)scheduled.
+ // This is the best approach, rather than re-scheduling the task each time
+ // the region is accessed/modified, as access/modify is a more frequent
+ // event than expiration.
+ reset(tp);
+ return false;
+ }
+
+ const auto full_path = region_->getFullPath().c_str();
+ switch (action_) {
+ case ExpirationAction::INVALIDATE: {
+ LOGDEBUG("RegionExpiryTask INVALIDATE region [%s]", full_path);
+ region_->invalidateRegionNoThrow(nullptr, CacheEventFlags::EXPIRATION);
+ break;
+ }
+ case ExpirationAction::LOCAL_INVALIDATE: {
+ LOGDEBUG("RegionExpiryTask LOCAL_INVALIDATE region [%s]", full_path);
+ region_->invalidateRegionNoThrow(
+ nullptr, CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL);
+ break;
+ }
+ case ExpirationAction::DESTROY: {
+ LOGDEBUG("RegionExpiryTask DESTROY region [%s]", full_path);
+ region_->destroyRegionNoThrow(nullptr, true, CacheEventFlags::EXPIRATION);
+ break;
+ }
+ case ExpirationAction::LOCAL_DESTROY: {
+ LOGDEBUG("RegionExpiryTask LOCAL_DESTROY region [%s]", full_path);
+ region_->destroyRegionNoThrow(
+ nullptr, true, CacheEventFlags::EXPIRATION | CacheEventFlags::LOCAL);
+ break;
+ }
+ case ExpirationAction::INVALID_ACTION: {
+ LOGERROR("Unknown expiration action %d for region [%s]", action_,
+ full_path);
+ break;
+ }
+ }
+
+ return true;
+}
+
+} // namespace client
+} // namespace geode
+} // namespace apache
diff --git a/cppcache/src/RegionExpiryHandler.hpp b/cppcache/src/RegionExpiryTask.hpp
similarity index 53%
rename from cppcache/src/RegionExpiryHandler.hpp
rename to cppcache/src/RegionExpiryTask.hpp
index 5d8e7e0..86c3f7f 100644
--- a/cppcache/src/RegionExpiryHandler.hpp
+++ b/cppcache/src/RegionExpiryTask.hpp
@@ -17,14 +17,14 @@
#pragma once
-#ifndef GEODE_REGIONEXPIRYHANDLER_H_
-#define GEODE_REGIONEXPIRYHANDLER_H_
+#ifndef GEODE_REGIONEXPIRYTASK_H_
+#define GEODE_REGIONEXPIRYTASK_H_
#include <geode/ExpirationAction.hpp>
#include <geode/Region.hpp>
#include <geode/internal/geode_globals.hpp>
-#include "ExpiryTaskManager.hpp"
+#include "ExpiryTask.hpp"
#include "RegionInternal.hpp"
namespace apache {
@@ -32,39 +32,28 @@ namespace geode {
namespace client {
/**
- * @class RegionExpiryHandler RegionExpiryHandler.hpp
+ * @class RegionExpiryTask RegionExpiryTask.hpp
*
* The task object which contains the handler which gets triggered
* when a region expires.
- *
- * TODO: cleanup region entry node and handler from expiry task
- * manager when region is destroyed
- *
*/
-class RegionExpiryHandler : public ACE_Event_Handler {
+class RegionExpiryTask : public ExpiryTask {
public:
- RegionExpiryHandler(std::shared_ptr<RegionInternal>& rptr,
- ExpirationAction action, std::chrono::seconds duration);
-
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
-
- int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) override;
-
- void setExpiryTaskId(ExpiryTaskManager::id_type expiryTaskId) {
- m_expiryTaskId = expiryTaskId;
- }
-
- private:
- std::shared_ptr<RegionInternal> m_regionPtr;
- ExpirationAction m_action;
- std::chrono::seconds m_duration;
- ExpiryTaskManager::id_type m_expiryTaskId;
- // perform the actual expiration action
- void DoTheExpirationAction();
+ RegionExpiryTask(ExpiryTaskManager& manager,
+ std::shared_ptr<RegionInternal> region,
+ ExpirationAction action, const duration_t& duration);
+
+ protected:
+ bool on_expire() override;
+ time_point_t expire_at() const;
+
+ protected:
+ std::shared_ptr<RegionInternal> region_;
+ ExpiryTask::duration_t duration_;
+ ExpirationAction action_;
};
} // namespace client
} // namespace geode
} // namespace apache
-#endif // GEODE_REGIONEXPIRYHANDLER_H_
+#endif // GEODE_REGIONEXPIRYTASK_H_
diff --git a/cppcache/src/SuspendedTxExpiryHandler.cpp b/cppcache/src/SuspendedTxExpiryTask.cpp
similarity index 66%
rename from cppcache/src/SuspendedTxExpiryHandler.cpp
rename to cppcache/src/SuspendedTxExpiryTask.cpp
index ca69364..17f6e57 100644
--- a/cppcache/src/SuspendedTxExpiryHandler.cpp
+++ b/cppcache/src/SuspendedTxExpiryTask.cpp
@@ -14,25 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "SuspendedTxExpiryHandler.hpp"
+#include "SuspendedTxExpiryTask.hpp"
#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
+#include "CacheTransactionManagerImpl.hpp"
namespace apache {
namespace geode {
namespace client {
-SuspendedTxExpiryHandler::SuspendedTxExpiryHandler(
- CacheTransactionManagerImpl* cacheTxMgr, TransactionId& tid)
- : m_cacheTxMgr(cacheTxMgr), m_txid(tid) {}
+SuspendedTxExpiryTask::SuspendedTxExpiryTask(
+ ExpiryTaskManager& expiry_manager, CacheTransactionManagerImpl& tx_manager,
+ TransactionId& tx_id)
+ : ExpiryTask(expiry_manager), tx_manager_(tx_manager), tx_id_(tx_id) {}
-int SuspendedTxExpiryHandler::handle_timeout(const ACE_Time_Value&,
- const void*) {
- LOGDEBUG("Entered SuspendedTxExpiryHandler");
+bool SuspendedTxExpiryTask::on_expire() {
+ LOGDEBUG("Entered SuspendedTxExpiryTask");
try {
// resume the transaction and rollback it
- if (m_cacheTxMgr->tryResume(m_txid, false)) m_cacheTxMgr->rollback();
+ if (tx_manager_.tryResume(tx_id_, false)) {
+ tx_manager_.rollback();
+ }
} catch (...) {
// Ignore whatever exception comes
LOGFINE(
@@ -42,11 +44,6 @@ int SuspendedTxExpiryHandler::handle_timeout(const ACE_Time_Value&,
return 0;
}
-int SuspendedTxExpiryHandler::handle_close(ACE_HANDLE, ACE_Reactor_Mask) {
- delete this;
- return 0;
-}
-
} // namespace client
} // namespace geode
} // namespace apache
diff --git a/cppcache/src/SuspendedTxExpiryHandler.hpp b/cppcache/src/SuspendedTxExpiryTask.hpp
similarity index 58%
rename from cppcache/src/SuspendedTxExpiryHandler.hpp
rename to cppcache/src/SuspendedTxExpiryTask.hpp
index e759221..619b251 100644
--- a/cppcache/src/SuspendedTxExpiryHandler.hpp
+++ b/cppcache/src/SuspendedTxExpiryTask.hpp
@@ -17,46 +17,43 @@
#pragma once
-#ifndef GEODE_SUSPENDEDTXEXPIRYHANDLER_H_
-#define GEODE_SUSPENDEDTXEXPIRYHANDLER_H_
-
-#include <ace/Event_Handler.h>
+#ifndef GEODE_SUSPENDEDTXEXPIRYTASK_H_
+#define GEODE_SUSPENDEDTXEXPIRYTASK_H_
#include <geode/Cache.hpp>
#include <geode/internal/geode_globals.hpp>
-#include "CacheTransactionManagerImpl.hpp"
+#include "ExpiryTask.hpp"
namespace apache {
namespace geode {
namespace client {
+class TransactionId;
class CacheTransactionManagerImpl;
/**
- * @class SuspendedTxExpiryHandler
+ * @class SuspendedTxExpiryTask
*
- * The task object which contains the handler which gets triggered
- * when a suspended transaction expires.
+ * The task gets triggered whenever a suspended transaction expires.
*
*/
-class SuspendedTxExpiryHandler : public ACE_Event_Handler {
+class SuspendedTxExpiryTask : public ExpiryTask {
public:
- SuspendedTxExpiryHandler(CacheTransactionManagerImpl* cacheTxMgr,
- TransactionId& txid);
-
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
+ SuspendedTxExpiryTask(ExpiryTaskManager& expiry_manager,
+ CacheTransactionManagerImpl& tx_manager,
+ TransactionId& tx_id);
- int handle_close(ACE_HANDLE, ACE_Reactor_Mask) override;
+ protected:
+ bool on_expire() override;
- private:
- CacheTransactionManagerImpl* m_cacheTxMgr;
- TransactionId& m_txid;
+ protected:
+ CacheTransactionManagerImpl& tx_manager_;
+ TransactionId& tx_id_;
};
} // namespace client
} // namespace geode
} // namespace apache
-#endif // GEODE_SUSPENDEDTXEXPIRYHANDLER_H_
+#endif // GEODE_SUSPENDEDTXEXPIRYTASK_H_
diff --git a/cppcache/src/TXCommitMessage.hpp b/cppcache/src/TXCommitMessage.hpp
index d8dbecc..a78892f 100644
--- a/cppcache/src/TXCommitMessage.hpp
+++ b/cppcache/src/TXCommitMessage.hpp
@@ -30,7 +30,7 @@ namespace geode {
namespace client {
class TXCommitMessage
- : public DataSerializableFixedId_t<DSFid::TXCommitMessage> {
+ : public internal::DataSerializableFixedId_t<DSFid::TXCommitMessage> {
public:
explicit TXCommitMessage(
MemberListForVersionStamp& memberListForVersionStamp);
diff --git a/cppcache/src/TXState.cpp b/cppcache/src/TXState.cpp
index 4ad6358..8e53e25 100644
--- a/cppcache/src/TXState.cpp
+++ b/cppcache/src/TXState.cpp
@@ -20,6 +20,7 @@
#include <geode/Cache.hpp>
#include <geode/CacheTransactionManager.hpp>
+#include "CacheImpl.hpp"
#include "ThinClientPoolDM.hpp"
#include "TransactionalOperation.hpp"
#include "TssConnectionWrapper.hpp"
diff --git a/cppcache/src/TXState.hpp b/cppcache/src/TXState.hpp
index 1e51ae7..69015f2 100644
--- a/cppcache/src/TXState.hpp
+++ b/cppcache/src/TXState.hpp
@@ -53,11 +53,10 @@ class TXState {
ThinClientPoolDM* getPoolDM() { return m_pooldm; }
void setPoolDM(ThinClientPoolDM* dm) { m_pooldm = dm; }
- void setSuspendedExpiryTaskId(
- ExpiryTaskManager::id_type suspendedExpiryTaskId) {
+ void setSuspendedExpiryTaskId(ExpiryTask::id_t suspendedExpiryTaskId) {
m_suspendedExpiryTaskId = suspendedExpiryTaskId;
}
- ExpiryTaskManager::id_type getSuspendedExpiryTaskId() {
+ ExpiryTask::id_t getSuspendedExpiryTaskId() {
return m_suspendedExpiryTaskId;
}
@@ -84,7 +83,7 @@ class TXState {
std::vector<std::shared_ptr<TransactionalOperation>> m_operations;
CacheImpl* m_cache;
ThinClientPoolDM* m_pooldm;
- ExpiryTaskManager::id_type m_suspendedExpiryTaskId;
+ ExpiryTask::id_t m_suspendedExpiryTaskId;
class ReplayControl {
public:
explicit ReplayControl(TXState* txState) : m_txState(txState) {
diff --git a/cppcache/src/TcrConnection.cpp b/cppcache/src/TcrConnection.cpp
index 8dd173e..5035e6d 100644
--- a/cppcache/src/TcrConnection.cpp
+++ b/cppcache/src/TcrConnection.cpp
@@ -22,6 +22,7 @@
#include <geode/AuthInitialize.hpp>
#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "ClientProxyMembershipID.hpp"
#include "Connector.hpp"
#include "DistributedSystemImpl.hpp"
diff --git a/cppcache/src/TcrConnectionManager.cpp b/cppcache/src/TcrConnectionManager.cpp
index 7f8e9a0..9157129 100644
--- a/cppcache/src/TcrConnectionManager.cpp
+++ b/cppcache/src/TcrConnectionManager.cpp
@@ -24,8 +24,7 @@
#include <geode/SystemProperties.hpp>
#include "CacheImpl.hpp"
-#include "ExpiryHandler_T.hpp"
-#include "ExpiryTaskManager.hpp"
+#include "FunctionExpiryTask.hpp"
#include "TcrConnection.hpp"
#include "TcrEndpoint.hpp"
#include "TcrHADistributionManager.hpp"
@@ -52,8 +51,7 @@ TcrConnectionManager::TcrConnectionManager(CacheImpl *cache)
m_failoverTask(nullptr),
cleanup_semaphore_(0),
m_cleanupTask(nullptr),
- m_pingTaskId(-1),
- m_servermonitorTaskId(-1),
+ ping_task_id_(ExpiryTask::invalid()),
// Create the queues with flag to not delete the objects
notify_cleanup_semaphore_list_(false),
redundancy_semaphore_(0),
@@ -72,17 +70,19 @@ void TcrConnectionManager::init(bool isPool) {
}
auto &props = m_cache->getDistributedSystem().getSystemProperties();
m_isDurable = !props.durableClientId().empty();
- auto pingInterval = props.pingInterval();
+
+ auto interval = props.pingInterval();
if (!isPool) {
- ACE_Event_Handler *connectionChecker =
- new ExpiryHandler_T<TcrConnectionManager>(
- this, &TcrConnectionManager::checkConnection);
- m_pingTaskId = m_cache->getExpiryTaskManager().scheduleExpiryTask(
- connectionChecker, std::chrono::seconds(10), pingInterval, false);
+ auto &expiry_manager = m_cache->getExpiryTaskManager();
+ auto task = std::make_shared<FunctionExpiryTask>(
+ expiry_manager, [this]() { ping_endpoints(); });
+
+ ping_task_id_ = expiry_manager.schedule(std::move(task),
+ std::chrono::seconds(10), interval);
LOGFINE(
"TcrConnectionManager::TcrConnectionManager Registered ping "
"task with id = %ld, interval = %ld",
- m_pingTaskId, pingInterval.count());
+ ping_task_id_, interval.count());
}
m_redundancyManager->m_HAenabled = false;
@@ -114,9 +114,7 @@ void TcrConnectionManager::startFailoverAndCleanupThreads(bool isPool) {
void TcrConnectionManager::close() {
LOGFINE("TcrConnectionManager is closing");
- if (m_pingTaskId > 0) {
- m_cache->getExpiryTaskManager().cancelTask(m_pingTaskId);
- }
+ m_cache->getExpiryTaskManager().cancel(ping_task_id_);
if (m_failoverTask != nullptr) {
m_failoverTask->stopNoblock();
@@ -258,26 +256,13 @@ bool TcrConnectionManager::removeRefToEndpoint(TcrEndpoint *ep,
return hasRemovedEndpoint;
}
-int TcrConnectionManager::processEventIdMap(const ACE_Time_Value &currTime,
- const void *) {
- return m_redundancyManager->processEventIdMap(currTime, nullptr);
-}
-
-int TcrConnectionManager::checkConnection(const ACE_Time_Value &,
- const void *) {
+void TcrConnectionManager::ping_endpoints() {
auto &&guard = m_endpoints.make_lock();
for (const auto &currItr : m_endpoints) {
if (currItr.second->connected() && !m_isNetDown) {
currItr.second->pingServer();
}
}
- return 0;
-}
-
-int TcrConnectionManager::checkRedundancy(const ACE_Time_Value &,
- const void *) {
- redundancy_semaphore_.release();
- return 0;
}
void TcrConnectionManager::failover(std::atomic<bool> &isRunning) {
diff --git a/cppcache/src/TcrConnectionManager.hpp b/cppcache/src/TcrConnectionManager.hpp
index 2bbfe89..2386db7 100644
--- a/cppcache/src/TcrConnectionManager.hpp
+++ b/cppcache/src/TcrConnectionManager.hpp
@@ -28,7 +28,7 @@
#include <geode/internal/geode_globals.hpp>
-#include "ExpiryTaskManager.hpp"
+#include "ExpiryTask.hpp"
#include "Queue.hpp"
#include "Task.hpp"
#include "ThinClientRedundancyManager.hpp"
@@ -63,9 +63,8 @@ class TcrConnectionManager {
void disconnect(ThinClientBaseDM* distMng,
std::vector<TcrEndpoint*>& endpoints,
bool keepEndpoints = false);
- int checkConnection(const ACE_Time_Value&, const void*);
- int checkRedundancy(const ACE_Time_Value&, const void*);
- int processEventIdMap(const ACE_Time_Value&, const void*);
+
+ void ping_endpoints();
void close();
void readyForEvents();
@@ -157,8 +156,7 @@ class TcrConnectionManager {
binary_semaphore cleanup_semaphore_;
std::unique_ptr<Task<TcrConnectionManager>> m_cleanupTask;
- ExpiryTaskManager::id_type m_pingTaskId;
- ExpiryTaskManager::id_type m_servermonitorTaskId;
+ ExpiryTask::id_t ping_task_id_;
Queue<Task<TcrEndpoint>*> m_receiverReleaseList;
Queue<TcrConnection*> m_connectionReleaseList;
Queue<binary_semaphore*> notify_cleanup_semaphore_list_;
diff --git a/cppcache/src/TcrPoolEndPoint.cpp b/cppcache/src/TcrPoolEndPoint.cpp
index 628ff1b..79908aa 100644
--- a/cppcache/src/TcrPoolEndPoint.cpp
+++ b/cppcache/src/TcrPoolEndPoint.cpp
@@ -19,6 +19,7 @@
#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "ThinClientPoolDM.hpp"
namespace apache {
diff --git a/cppcache/src/ThinClientBaseDM.cpp b/cppcache/src/ThinClientBaseDM.cpp
index 4d9123a..44e2349 100644
--- a/cppcache/src/ThinClientBaseDM.cpp
+++ b/cppcache/src/ThinClientBaseDM.cpp
@@ -21,6 +21,7 @@
#include <geode/AuthenticatedView.hpp>
+#include "CacheImpl.hpp"
#include "TcrConnectionManager.hpp"
#include "ThinClientRegion.hpp"
#include "UserAttributes.hpp"
diff --git a/cppcache/src/ThinClientCacheDistributionManager.cpp b/cppcache/src/ThinClientCacheDistributionManager.cpp
index e33e0d7..6f75644 100644
--- a/cppcache/src/ThinClientCacheDistributionManager.cpp
+++ b/cppcache/src/ThinClientCacheDistributionManager.cpp
@@ -20,10 +20,8 @@
#include <algorithm>
#include <geode/ExceptionTypes.hpp>
-#include <geode/internal/geode_globals.hpp>
#include "CacheImpl.hpp"
-#include "ReadWriteLock.hpp"
#include "RemoteQueryService.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrEndpoint.hpp"
diff --git a/cppcache/src/ThinClientDistributionManager.cpp b/cppcache/src/ThinClientDistributionManager.cpp
index 3facc2c..12f34cb 100644
--- a/cppcache/src/ThinClientDistributionManager.cpp
+++ b/cppcache/src/ThinClientDistributionManager.cpp
@@ -22,6 +22,7 @@
#include <geode/AuthInitialize.hpp>
#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "DistributedSystemImpl.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrEndpoint.hpp"
diff --git a/cppcache/src/ThinClientLocatorHelper.cpp b/cppcache/src/ThinClientLocatorHelper.cpp
index 1dc5699..d6ed74e 100644
--- a/cppcache/src/ThinClientLocatorHelper.cpp
+++ b/cppcache/src/ThinClientLocatorHelper.cpp
@@ -26,6 +26,7 @@
#include <geode/DataOutput.hpp>
#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "ClientConnectionRequest.hpp"
#include "ClientConnectionResponse.hpp"
#include "ClientReplacementRequest.hpp"
diff --git a/cppcache/src/ThinClientPoolDM.cpp b/cppcache/src/ThinClientPoolDM.cpp
index 8ae524e..2abd274 100644
--- a/cppcache/src/ThinClientPoolDM.cpp
+++ b/cppcache/src/ThinClientPoolDM.cpp
@@ -20,17 +20,15 @@
#include <algorithm>
#include <thread>
-#include <ace/OS_NS_unistd.h>
-
#include <geode/AuthInitialize.hpp>
#include <geode/PoolManager.hpp>
#include <geode/ResultCollector.hpp>
#include <geode/SystemProperties.hpp>
+#include "CacheImpl.hpp"
#include "DistributedSystemImpl.hpp"
#include "ExecutionImpl.hpp"
-#include "ExpiryHandler_T.hpp"
-#include "ExpiryTaskManager.hpp"
+#include "FunctionExpiryTask.hpp"
#include "TcrConnectionManager.hpp"
#include "TcrEndpoint.hpp"
#include "ThinClientRegion.hpp"
@@ -154,9 +152,6 @@ ThinClientPoolDM::ThinClientPoolDM(const char* name,
m_connManageTask(nullptr),
m_pingTask(nullptr),
m_updateLocatorListTask(nullptr),
- m_pingTaskId(-1),
- m_updateLocatorListTaskId(-1),
- m_connManageTaskId(-1),
m_clientOps(0),
connected_endpoints_(0),
m_PoolStatsSampler(nullptr),
@@ -276,26 +271,25 @@ void ThinClientPoolDM::startBackgroundThreads() {
this, &ThinClientPoolDM::pingServer, NC_Ping_Thread));
m_pingTask->start();
- const auto& pingInterval = getPingInterval();
- if (pingInterval > std::chrono::seconds::zero()) {
+ auto interval = getPingInterval();
+ if (interval > 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);
+ "ThinClientPoolDM::startBackgroundThreads: Scheduling ping task at %zu",
+ interval.count());
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ auto task = std::make_shared<FunctionExpiryTask>(
+ manager, [this] { ping_semaphore_.release(); });
+ ping_task_id_ =
+ manager.schedule(std::move(task), std::chrono::seconds(1), interval);
} else {
LOGDEBUG(
"ThinClientPoolDM::startBackgroundThreads: Not Scheduling ping task as "
"ping interval %ld",
- getPingInterval().count());
+ interval.count());
}
- auto updateLocatorListInterval = getUpdateLocatorListInterval();
-
- if (updateLocatorListInterval > std::chrono::seconds::zero()) {
+ interval = getUpdateLocatorListInterval();
+ if (interval > std::chrono::seconds::zero()) {
m_updateLocatorListTask =
std::unique_ptr<Task<ThinClientPoolDM>>(new Task<ThinClientPoolDM>(
this, &ThinClientPoolDM::updateLocatorList, "NC_LocatorList"));
@@ -304,17 +298,16 @@ void ThinClientPoolDM::startBackgroundThreads() {
LOGDEBUG(
"ThinClientPoolDM::startBackgroundThreads: Creating updateLocatorList "
"task");
- auto updateLocatorListHandler = new ExpiryHandler_T<ThinClientPoolDM>(
- this, &ThinClientPoolDM::doUpdateLocatorList);
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ auto task = std::make_shared<FunctionExpiryTask>(
+ manager, [this] { update_locators_semaphore_.release(); });
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);
+ interval.count());
+ update_locators_task_id_ =
+ manager.schedule(std::move(task), std::chrono::seconds(1), interval);
}
LOGDEBUG(
@@ -329,25 +322,24 @@ void ThinClientPoolDM::startBackgroundThreads() {
auto idle = getIdleTimeout();
auto load = getLoadConditioningInterval();
- if (load > std::chrono::milliseconds::zero()) {
- if (load < idle || idle <= std::chrono::milliseconds::zero()) {
- idle = load;
- }
+ if (load > std::chrono::milliseconds::zero() &&
+ (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);
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ auto task = std::make_shared<FunctionExpiryTask>(
+ manager, [this] { conn_semaphore_.release(); });
LOGDEBUG(
"ThinClientPoolDM::startBackgroundThreads: Scheduling "
"manageConnections task");
- m_connManageTaskId =
- m_connManager.getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
- connHandler, std::chrono::seconds(1), idle, false);
+ conns_mgmt_task_id_ =
+ manager.schedule(std::move(task), std::chrono::seconds(1), idle);
}
LOGDEBUG(
@@ -379,27 +371,28 @@ void ThinClientPoolDM::startBackgroundThreads() {
void ThinClientPoolDM::manageConnections(std::atomic<bool>& isRunning) {
LOGFINE("ThinClientPoolDM: starting manageConnections thread");
+ conn_semaphore_.acquire();
+
while (isRunning) {
- conn_semaphore_.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");
- }
+ 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");
}
+
+ conn_semaphore_.acquire();
}
LOGFINE("ThinClientPoolDM: ending manageConnections thread");
@@ -507,17 +500,18 @@ void ThinClientPoolDM::cleanStaleConnections(std::atomic<bool>& isRunning) {
count++;
if (count % 10 == 0) {
- std::this_thread::sleep_for(std::chrono::milliseconds(1));
+ std::this_thread::yield();
}
}
- 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());
+ if (isRunning) {
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ if (manager.reset(conns_mgmt_task_id_, _nextIdle) < 0) {
+ 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());
@@ -765,9 +759,9 @@ void ThinClientPoolDM::stopPingThread() {
ping_semaphore_.release();
m_pingTask->wait();
m_pingTask = nullptr;
- if (m_pingTaskId >= 0) {
- m_connManager.getCacheImpl()->getExpiryTaskManager().cancelTask(
- m_pingTaskId);
+ if (ping_task_id_ != ExpiryTask::invalid()) {
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ manager.cancel(ping_task_id_);
}
}
}
@@ -779,9 +773,9 @@ void ThinClientPoolDM::stopUpdateLocatorListThread() {
update_locators_semaphore_.release();
m_updateLocatorListTask->wait();
m_updateLocatorListTask = nullptr;
- if (m_updateLocatorListTaskId >= 0) {
- m_connManager.getCacheImpl()->getExpiryTaskManager().cancelTask(
- m_updateLocatorListTaskId);
+ if (update_locators_task_id_ != ExpiryTask::invalid()) {
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ manager.cancel(update_locators_task_id_);
}
}
}
@@ -809,9 +803,7 @@ void ThinClientPoolDM::destroy(bool keepAlive) {
conn_semaphore_.release();
m_connManageTask->wait();
m_connManageTask = nullptr;
- if (m_connManageTaskId >= 0) {
- cacheImpl->getExpiryTaskManager().cancelTask(m_connManageTaskId);
- }
+ cacheImpl->getExpiryTaskManager().cancel(conns_mgmt_task_id_);
}
LOGDEBUG("Closing PoolStatsSampler thread.");
@@ -1716,6 +1708,7 @@ GfErrType ThinClientPoolDM::createPoolConnectionToAEndPoint(
getStats().incPoolConnects();
getStats().setCurPoolConnections(m_poolSize);
}
+
conn_semaphore_.release();
return error;
@@ -1804,6 +1797,7 @@ GfErrType ThinClientPoolDM::createPoolConnection(
break;
}
}
+
conn_semaphore_.release();
// if a fatal error occurred earlier and we don't have
// a connection then return this saved error
@@ -2096,21 +2090,6 @@ void ThinClientPoolDM::clearPdxTypeRegistry() {
cache_impl->getPdxTypeRegistry()->clear();
}
-int ThinClientPoolDM::doPing(const ACE_Time_Value&, const void*) {
- ping_semaphore_.release();
- return 0;
-}
-
-int ThinClientPoolDM::doUpdateLocatorList(const ACE_Time_Value&, const void*) {
- update_locators_semaphore_.release();
- return 0;
-}
-
-int ThinClientPoolDM::doManageConnections(const ACE_Time_Value&, const void*) {
- conn_semaphore_.release();
- return 0;
-}
-
void ThinClientPoolDM::releaseThreadLocalConnection() {
m_manager->releaseThreadLocalConnection();
}
diff --git a/cppcache/src/ThinClientPoolDM.hpp b/cppcache/src/ThinClientPoolDM.hpp
index 460e1df..6a13076 100644
--- a/cppcache/src/ThinClientPoolDM.hpp
+++ b/cppcache/src/ThinClientPoolDM.hpp
@@ -201,6 +201,7 @@ class ThinClientPoolDM
PoolStats* m_stats;
bool m_sticky;
void netDown();
+
binary_semaphore update_locators_semaphore_;
binary_semaphore ping_semaphore_;
volatile bool m_isDestroyed;
@@ -294,17 +295,16 @@ class ThinClientPoolDM
unsigned m_server;
// Manage Connection thread
+
binary_semaphore conn_semaphore_;
std::unique_ptr<Task<ThinClientPoolDM>> m_connManageTask;
std::unique_ptr<Task<ThinClientPoolDM>> m_pingTask;
std::unique_ptr<Task<ThinClientPoolDM>> m_updateLocatorListTask;
- ExpiryTaskManager::id_type m_pingTaskId;
- ExpiryTaskManager::id_type m_updateLocatorListTaskId;
- ExpiryTaskManager::id_type m_connManageTaskId;
+ ExpiryTask::id_t ping_task_id_{ExpiryTask::invalid()};
+ ExpiryTask::id_t update_locators_task_id_{ExpiryTask::invalid()};
+ ExpiryTask::id_t conns_mgmt_task_id_{ExpiryTask::invalid()};
+
void manageConnections(std::atomic<bool>& isRunning);
- int doPing(const ACE_Time_Value&, const void*);
- int doUpdateLocatorList(const ACE_Time_Value&, const void*);
- int doManageConnections(const ACE_Time_Value&, const void*);
void manageConnectionsInternal(std::atomic<bool>& isRunning);
void cleanStaleConnections(std::atomic<bool>& isRunning);
void restoreMinConnections(std::atomic<bool>& isRunning);
diff --git a/cppcache/src/ThinClientPoolHADM.cpp b/cppcache/src/ThinClientPoolHADM.cpp
index f89db1e..bdb6df2 100644
--- a/cppcache/src/ThinClientPoolHADM.cpp
+++ b/cppcache/src/ThinClientPoolHADM.cpp
@@ -19,7 +19,8 @@
#include <geode/SystemProperties.hpp>
-#include "ExpiryHandler_T.hpp"
+#include "CacheImpl.hpp"
+#include "FunctionExpiryTask.hpp"
#include "TcrConnectionManager.hpp"
#include "util/exception.hpp"
@@ -35,7 +36,7 @@ ThinClientPoolHADM::ThinClientPoolHADM(const char* name,
m_theTcrConnManager(connManager),
redundancy_semaphore_(0),
m_redundancyTask(nullptr),
- m_servermonitorTaskId(-1) {
+ server_monitor_task_id_(ExpiryTask::invalid()) {
m_redundancyManager = std::unique_ptr<ThinClientRedundancyManager>(
new ThinClientRedundancyManager(
&connManager, poolAttr->getSubscriptionRedundancy(), this));
@@ -58,19 +59,18 @@ void ThinClientPoolHADM::startBackgroundThreads() {
// operations.
GfErrType err = m_redundancyManager->maintainRedundancyLevel(true);
- ACE_Event_Handler* redundancyChecker =
- new ExpiryHandler_T<ThinClientPoolHADM>(
- this, &ThinClientPoolHADM::checkRedundancy);
- const auto redundancyMonitorInterval = props.redundancyMonitorInterval();
+ const auto interval = props.redundancyMonitorInterval();
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ auto task = std::make_shared<FunctionExpiryTask>(
+ manager, [this]() { redundancy_semaphore_.release(); });
+
+ server_monitor_task_id_ =
+ manager.schedule(std::move(task), std::chrono::seconds(1), interval);
- m_servermonitorTaskId =
- m_connManager.getCacheImpl()->getExpiryTaskManager().scheduleExpiryTask(
- redundancyChecker, std::chrono::seconds(1), redundancyMonitorInterval,
- false);
LOGFINE(
"ThinClientPoolHADM::ThinClientPoolHADM Registered server "
"monitor task with id = %ld, interval = %ld",
- m_servermonitorTaskId, redundancyMonitorInterval.count());
+ server_monitor_task_id_, interval.count());
if (ThinClientBaseDM::isFatalClientError(err)) {
if (err == GF_CACHE_LOCATOR_EXCEPTION) {
@@ -156,11 +156,6 @@ void ThinClientPoolHADM::redundancy(std::atomic<bool>& isRunning) {
LOGFINE("ThinClientPoolHADM: Ending maintain redundancy thread.");
}
-int ThinClientPoolHADM::checkRedundancy(const ACE_Time_Value&, const void*) {
- redundancy_semaphore_.release();
- return 0;
-}
-
void ThinClientPoolHADM::destroy(bool keepAlive) {
LOGDEBUG("ThinClientPoolHADM::destroy");
if (!m_isDestroyed && !m_destroyPending) {
@@ -184,10 +179,8 @@ void ThinClientPoolHADM::destroy(bool keepAlive) {
void ThinClientPoolHADM::sendNotificationCloseMsgs() {
if (m_redundancyTask) {
- if (m_servermonitorTaskId >= 0) {
- m_connManager.getCacheImpl()->getExpiryTaskManager().cancelTask(
- m_servermonitorTaskId);
- }
+ auto& manager = m_connManager.getCacheImpl()->getExpiryTaskManager();
+ manager.cancel(server_monitor_task_id_);
m_redundancyTask->stopNoblock();
redundancy_semaphore_.release();
m_redundancyTask->wait();
diff --git a/cppcache/src/ThinClientPoolHADM.hpp b/cppcache/src/ThinClientPoolHADM.hpp
index dfdf86d..c817027 100644
--- a/cppcache/src/ThinClientPoolHADM.hpp
+++ b/cppcache/src/ThinClientPoolHADM.hpp
@@ -112,8 +112,7 @@ class ThinClientPoolHADM : public ThinClientPoolDM {
void redundancy(std::atomic<bool>& isRunning);
- ExpiryTaskManager::id_type m_servermonitorTaskId;
- int checkRedundancy(const ACE_Time_Value&, const void*);
+ ExpiryTask::id_t server_monitor_task_id_;
std::shared_ptr<TcrEndpoint> createEP(const char* endpointName) override;
diff --git a/cppcache/src/ThinClientRedundancyManager.cpp b/cppcache/src/ThinClientRedundancyManager.cpp
index 5a5cf63..8a2db40 100644
--- a/cppcache/src/ThinClientRedundancyManager.cpp
+++ b/cppcache/src/ThinClientRedundancyManager.cpp
@@ -25,7 +25,7 @@
#include "CacheImpl.hpp"
#include "ClientProxyMembershipID.hpp"
-#include "ExpiryHandler_T.hpp"
+#include "FunctionExpiryTask.hpp"
#include "RemoteQueryService.hpp"
#include "ServerLocation.hpp"
#include "TcrHADistributionManager.hpp"
@@ -57,8 +57,6 @@ ThinClientRedundancyManager::ThinClientRedundancyManager(
m_servers(nullptr),
m_periodicAckTask(nullptr),
periodic_ack_semaphore_(1),
- m_processEventIdMapTaskId(-1),
- m_nextAckInc(0),
m_HAenabled(false) {}
std::list<ServerLocation> ThinClientRedundancyManager::selectServers(
@@ -629,7 +627,7 @@ void ThinClientRedundancyManager::initialize(int redundancyLevel) {
if (interval < std::chrono::milliseconds(100)) {
interval = std::chrono::milliseconds(100);
}
- m_nextAckInc = interval;
+ next_ack_inc_ = interval;
m_nextAck = clock::now() + interval;
}
@@ -678,10 +676,9 @@ void ThinClientRedundancyManager::close() {
LOGDEBUG("ThinClientRedundancyManager::close(): closing redundancy manager.");
if (m_periodicAckTask) {
- if (m_processEventIdMapTaskId >= 0) {
- m_theTcrConnManager->getCacheImpl()->getExpiryTaskManager().cancelTask(
- m_processEventIdMapTaskId);
- }
+ auto& manager = m_theTcrConnManager->getCacheImpl()->getExpiryTaskManager();
+ manager.cancel(process_event_id_map_task_id_);
+
m_periodicAckTask->stopNoblock();
periodic_ack_semaphore_.release();
m_periodicAckTask->wait();
@@ -1115,12 +1112,6 @@ void ThinClientRedundancyManager::readyForEvents() {
m_sentReadyForEvents = true;
}
-int ThinClientRedundancyManager::processEventIdMap(const ACE_Time_Value&,
- const void*) {
- periodic_ack_semaphore_.release();
- return 0;
-}
-
void ThinClientRedundancyManager::periodicAck(std::atomic<bool>& isRunning) {
periodic_ack_semaphore_.acquire();
@@ -1137,7 +1128,7 @@ void ThinClientRedundancyManager::doPeriodicAck() {
// do periodic ack if HA is enabled and the time has come
if (m_HAenabled && (m_nextAck < clock::now())) {
LOGFINER("Doing periodic ack");
- m_nextAck += m_nextAckInc;
+ m_nextAck += next_ack_inc_;
auto entries = m_eventidmap.getUnAcked();
auto count = entries.size();
@@ -1206,18 +1197,16 @@ void ThinClientRedundancyManager::startPeriodicAck() {
->getDistributedSystem()
.getSystemProperties();
// start the periodic ACK task handler
- auto periodicAckTask = new ExpiryHandler_T<ThinClientRedundancyManager>(
- this, &ThinClientRedundancyManager::processEventIdMap);
- m_processEventIdMapTaskId =
- m_theTcrConnManager->getCacheImpl()
- ->getExpiryTaskManager()
- .scheduleExpiryTask(periodicAckTask, m_nextAckInc, m_nextAckInc,
- false);
+ auto& manager = m_theTcrConnManager->getCacheImpl()->getExpiryTaskManager();
+ auto task = std::make_shared<FunctionExpiryTask>(
+ manager, [this] { periodic_ack_semaphore_.release(); });
+ process_event_id_map_task_id_ =
+ manager.schedule(std::move(task), next_ack_inc_, next_ack_inc_);
LOGFINE(
"Registered subscription event "
"periodic ack task with id = %ld, notify-ack-interval = %ld, "
"notify-dupcheck-life = %ld, periodic ack is %sabled",
- m_processEventIdMapTaskId,
+ process_event_id_map_task_id_,
(m_poolHADM ? m_poolHADM->getSubscriptionAckInterval()
: props.notifyAckInterval())
.count(),
diff --git a/cppcache/src/ThinClientRedundancyManager.hpp b/cppcache/src/ThinClientRedundancyManager.hpp
index 747d40a..3b6e4ce 100644
--- a/cppcache/src/ThinClientRedundancyManager.hpp
+++ b/cppcache/src/ThinClientRedundancyManager.hpp
@@ -135,16 +135,15 @@ class ThinClientRedundancyManager {
TcrMessageReply* reply, bool isPrimary);
inline bool isDurable();
- int processEventIdMap(const ACE_Time_Value&, const void*);
+
std::unique_ptr<Task<ThinClientRedundancyManager>> m_periodicAckTask;
binary_semaphore periodic_ack_semaphore_;
- ExpiryTaskManager::id_type
- m_processEventIdMapTaskId; // periodic check eventid map for notify ack
- // and/or expiry
+ ExpiryTask::id_t process_event_id_map_task_id_{ExpiryTask::invalid()};
+
void periodicAck(std::atomic<bool>& isRunning);
void doPeriodicAck();
- time_point m_nextAck; // next ack time
- std::chrono::milliseconds m_nextAckInc; // next ack time increment
+ time_point m_nextAck; // next ack time
+ std::chrono::milliseconds next_ack_inc_; // next ack time increment
volatile bool m_HAenabled;
EventIdMap m_eventidmap;
diff --git a/cppcache/test/mock/MapEntryImplMock.hpp b/cppcache/src/TombstoneEntry.hpp
similarity index 59%
copy from cppcache/test/mock/MapEntryImplMock.hpp
copy to cppcache/src/TombstoneEntry.hpp
index 68fc6c7..6f284ad 100644
--- a/cppcache/test/mock/MapEntryImplMock.hpp
+++ b/cppcache/src/TombstoneEntry.hpp
@@ -17,30 +17,39 @@
#pragma once
-#ifndef GEODE_MAPENTRYIMPLMOCK_H_
-#define GEODE_MAPENTRYIMPLMOCK_H_
+#ifndef GEODE_TOMBSTONEENTRY_H_
+#define GEODE_TOMBSTONEENTRY_H_
-#include <gmock/gmock.h>
+#include <chrono>
+#include <memory>
+#include "ExpiryTask.hpp"
#include "MapEntry.hpp"
namespace apache {
namespace geode {
namespace client {
-class MapEntryImplMock : public MapEntryImpl {
+
+class TombstoneEntry {
public:
- explicit MapEntryImplMock(const std::shared_ptr<CacheableKey>& key)
- : MapEntryImpl(key) {}
-
- MOCK_METHOD0(getLRUProperties, LRUEntryProperties&());
- MOCK_METHOD1(addTracker, int(std::shared_ptr<MapEntry>&));
- MOCK_METHOD0(removeTracker, std::pair<bool, int>());
- MOCK_CONST_METHOD0(getTrackingNumber, int());
- MOCK_CONST_METHOD0(getUpdateCount, int());
- MOCK_METHOD1(incrementUpdateCount, int(std::shared_ptr<MapEntry>&));
+ explicit TombstoneEntry(std::shared_ptr<MapEntryImpl> entry)
+ : entry_(std::move(entry)), task_id_(ExpiryTask::invalid()) {}
+
+ std::shared_ptr<MapEntryImpl> entry() { return entry_; }
+ ExpiryTask::id_t task_id() { return task_id_; }
+ void task_id(ExpiryTask::id_t task_id) { task_id_ = task_id; }
+
+ void invalidate() { valid_ = false; }
+ bool valid() const { return valid_; }
+
+ protected:
+ std::shared_ptr<MapEntryImpl> entry_;
+ ExpiryTask::id_t task_id_;
+ bool valid_{true};
};
+
} // namespace client
} // namespace geode
} // namespace apache
-#endif // GEODE_MAPENTRYIMPLMOCK_H_
+#endif // GEODE_TOMBSTONEENTRY_H_
diff --git a/cppcache/src/TombstoneExpiryHandler.cpp b/cppcache/src/TombstoneExpiryHandler.cpp
deleted file mode 100644
index dc488c9..0000000
--- a/cppcache/src/TombstoneExpiryHandler.cpp
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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 "TombstoneExpiryHandler.hpp"
-
-#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
-#include "MapEntry.hpp"
-#include "RegionInternal.hpp"
-
-namespace apache {
-namespace geode {
-namespace client {
-
-TombstoneExpiryHandler::TombstoneExpiryHandler(
- std::shared_ptr<TombstoneEntry> entryPtr, TombstoneList* tombstoneList,
- std::chrono::milliseconds duration, CacheImpl* cacheImpl)
- : m_entryPtr(entryPtr),
- m_duration(duration),
- m_cacheImpl(cacheImpl),
- m_tombstoneList(tombstoneList) {}
-
-int TombstoneExpiryHandler::handle_timeout(const ACE_Time_Value&, const void*) {
- std::shared_ptr<CacheableKey> key;
- m_entryPtr->getEntry()->getKeyI(key);
- auto creationTime = m_entryPtr->getTombstoneCreationTime();
- auto curr_time = TombstoneEntry::clock::now();
- auto expiryTaskId = m_entryPtr->getExpiryTaskId();
- auto sec = curr_time - creationTime - m_duration;
- try {
- using apache::geode::internal::chrono::duration::to_string;
- LOGDEBUG(
- "Entered entry expiry task handler for tombstone of key [%s]: "
- "%s,%s,%s,%s",
- Utils::nullSafeToString(key).c_str(),
- to_string(curr_time.time_since_epoch()).c_str(),
- to_string(creationTime.time_since_epoch()).c_str(),
- to_string(m_duration).c_str(), to_string(sec).c_str());
- if (sec >= std::chrono::seconds::zero()) {
- DoTheExpirationAction(key);
- } else {
- // reset the task after
- // (lastAccessTime + entryExpiryDuration - curr_time) in seconds
- LOGDEBUG(
- "Resetting expiry task %s later for key "
- "[%s]",
- to_string(-sec).c_str(), Utils::nullSafeToString(key).c_str());
- m_cacheImpl->getExpiryTaskManager().resetTask(
- m_entryPtr->getExpiryTaskId(), -sec);
- return 0;
- }
- } catch (...) {
- // Ignore whatever exception comes
- }
- LOGDEBUG("Removing expiry task for key [%s]",
- Utils::nullSafeToString(key).c_str());
- // we now delete the handler in GF_Timer_Heap_ImmediateReset_T
- // and always return success.
- m_cacheImpl->getExpiryTaskManager().resetTask(expiryTaskId, 0);
- return 0;
-}
-
-int TombstoneExpiryHandler::handle_close(ACE_HANDLE, ACE_Reactor_Mask) {
- return 0;
-}
-
-inline void TombstoneExpiryHandler::DoTheExpirationAction(
- const std::shared_ptr<CacheableKey>& key) {
- LOGDEBUG(
- "EntryExpiryHandler::DoTheExpirationAction LOCAL_DESTROY "
- "for region entry with key %s",
- Utils::nullSafeToString(key).c_str());
- m_tombstoneList->removeEntryFromMapSegment(key);
-}
-
-} // namespace client
-} // namespace geode
-} // namespace apache
diff --git a/cppcache/src/TombstoneExpiryHandler.hpp b/cppcache/src/TombstoneExpiryHandler.hpp
deleted file mode 100644
index 6540ce3..0000000
--- a/cppcache/src/TombstoneExpiryHandler.hpp
+++ /dev/null
@@ -1,73 +0,0 @@
-/*
- * 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.
- */
-
-#pragma once
-
-#ifndef GEODE_TOMBSTONEEXPIRYHANDLER_H_
-#define GEODE_TOMBSTONEEXPIRYHANDLER_H_
-
-#include <geode/ExpirationAction.hpp>
-#include <geode/Region.hpp>
-#include <geode/internal/geode_globals.hpp>
-
-#include "RegionInternal.hpp"
-#include "TombstoneList.hpp"
-
-namespace apache {
-namespace geode {
-namespace client {
-
-/**
- * @class TombstoneExpiryHandler TombstoneExpiryHandler.hpp
- *
- * The task object which contains the handler which gets triggered
- * when a tombstone expires.
- *
- */
-class TombstoneExpiryHandler : public ACE_Event_Handler {
- public:
- TombstoneExpiryHandler(std::shared_ptr<TombstoneEntry> entryPtr,
- TombstoneList* tombstoneList,
- std::chrono::milliseconds duration,
- CacheImpl* cacheImpl);
-
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
-
- int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) override;
-
- void setTombstoneEntry(std::shared_ptr<TombstoneEntry> entryPtr) {
- m_entryPtr = entryPtr;
- }
-
- private:
- // The entry contained in the tombstone list
- std::shared_ptr<TombstoneEntry> m_entryPtr;
- // Duration after which the task should be reset in case of
- // modification.
- std::chrono::milliseconds m_duration;
- CacheImpl* m_cacheImpl;
- // perform the actual expiration action
- void DoTheExpirationAction(const std::shared_ptr<CacheableKey>& key);
-
- TombstoneList* m_tombstoneList;
-};
-} // namespace client
-} // namespace geode
-} // namespace apache
-
-#endif // GEODE_TOMBSTONEEXPIRYHANDLER_H_
diff --git a/cppcache/src/CacheStatistics.cpp b/cppcache/src/TombstoneExpiryTask.cpp
similarity index 57%
copy from cppcache/src/CacheStatistics.cpp
copy to cppcache/src/TombstoneExpiryTask.cpp
index b6a1dc3..c91ed4d 100644
--- a/cppcache/src/CacheStatistics.cpp
+++ b/cppcache/src/TombstoneExpiryTask.cpp
@@ -15,26 +15,34 @@
* limitations under the License.
*/
-#include <geode/CacheStatistics.hpp>
+#include "TombstoneExpiryTask.hpp"
+
+#include "CacheImpl.hpp"
+#include "MapSegment.hpp"
+#include "RegionInternal.hpp"
+#include "TombstoneEntry.hpp"
namespace apache {
namespace geode {
namespace client {
-void CacheStatistics::setLastModifiedTime(time_point lmt) {
- m_lastModifiedTime = lmt.time_since_epoch().count();
-}
+TombstoneExpiryTask::TombstoneExpiryTask(
+ ExpiryTaskManager& manager, MapSegment& segment,
+ std::shared_ptr<TombstoneEntry> tombstone)
+ : ExpiryTask(manager),
+ segment_(segment),
+ tombstone_(std::move(tombstone)) {}
-void CacheStatistics::setLastAccessedTime(time_point lat) {
- m_lastAccessTime = lat.time_since_epoch().count();
-}
-
-CacheStatistics::time_point CacheStatistics::getLastModifiedTime() const {
- return time_point(std::chrono::system_clock::duration(m_lastModifiedTime));
-}
+bool TombstoneExpiryTask::on_expire() {
+ std::shared_ptr<CacheableKey> key;
+ tombstone_->entry()->getKeyI(key);
-CacheStatistics::time_point CacheStatistics::getLastAccessedTime() const {
- return time_point(std::chrono::system_clock::duration(m_lastAccessTime));
+ LOGDEBUG(
+ "TombstoneExpiryTask::on_expire LOCAL_DESTROY "
+ "for region entry with key %s",
+ Utils::nullSafeToString(key).c_str());
+ segment_.remove_tomb_entry(tombstone_);
+ return true;
}
} // namespace client
diff --git a/cppcache/src/PreservedDataExpiryHandler.hpp b/cppcache/src/TombstoneExpiryTask.hpp
similarity index 51%
rename from cppcache/src/PreservedDataExpiryHandler.hpp
rename to cppcache/src/TombstoneExpiryTask.hpp
index 1563a45..ec6e2b5 100644
--- a/cppcache/src/PreservedDataExpiryHandler.hpp
+++ b/cppcache/src/TombstoneExpiryTask.hpp
@@ -17,47 +17,57 @@
#pragma once
-#ifndef GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
-#define GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
+#ifndef GEODE_TOMBSTONEEXPIRYTASK_H_
+#define GEODE_TOMBSTONEEXPIRYTASK_H_
-#include <geode/Cache.hpp>
-#include <geode/PdxSerializable.hpp>
+#include <geode/ExpirationAction.hpp>
+#include <geode/Region.hpp>
#include <geode/internal/geode_globals.hpp>
-#include "CacheImpl.hpp"
-#include "ExpiryTaskManager.hpp"
+#include "ExpiryTask.hpp"
+#include "RegionInternal.hpp"
namespace apache {
namespace geode {
namespace client {
+class MapSegment;
+class TombstoneEntry;
+
/**
- * @class PreservedDataExpiryHandler
- *
- * The task object which contains the handler which gets triggered
- * when a preserved data expires.
+ * @class TombstoneExpiryTask TombstoneExpiryTask.hpp
*
+ * The task which gets triggered when a tombstone expires.
*/
-class PreservedDataExpiryHandler : public ACE_Event_Handler {
+class TombstoneExpiryTask : public ExpiryTask {
public:
/**
- * Constructor
+ * Class constructor
+ * @param manager A reference to the expiry manager
+ * @param segment A reference to the MapSegment the entry sits in
+ * @param tombstone Tombstone entry
*/
- PreservedDataExpiryHandler(
- const std::shared_ptr<PdxTypeRegistry>& pdxTypeRegistry,
- const std::shared_ptr<PdxSerializable>& pdxObjectPtr);
-
- int handle_timeout(const ACE_Time_Value& current_time,
- const void* arg) override;
+ TombstoneExpiryTask(ExpiryTaskManager& manager, MapSegment& segment,
+ std::shared_ptr<TombstoneEntry> tombstone);
- int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask) override;
+ protected:
+ bool on_expire() override;
private:
- std::shared_ptr<PdxTypeRegistry> m_pdxTypeRegistry;
- std::shared_ptr<PdxSerializable> m_pdxObjectPtr;
+ /// Member attributes
+
+ /**
+ * Reference to the map segment in which the entry is
+ */
+ MapSegment& segment_;
+
+ /**
+ * Tombstone entry
+ */
+ std::shared_ptr<TombstoneEntry> tombstone_;
};
} // namespace client
} // namespace geode
} // namespace apache
-#endif // GEODE_PRESERVEDDATAEXPIRYHANDLER_H_
+#endif // GEODE_TOMBSTONEEXPIRYTASK_H_
diff --git a/cppcache/src/TombstoneList.cpp b/cppcache/src/TombstoneList.cpp
index 211d95d..22ee9b3 100644
--- a/cppcache/src/TombstoneList.cpp
+++ b/cppcache/src/TombstoneList.cpp
@@ -17,15 +17,18 @@
#include "TombstoneList.hpp"
-#include <unordered_map>
+#include <unordered_set>
+#include "CacheImpl.hpp"
#include "MapSegment.hpp"
-#include "TombstoneExpiryHandler.hpp"
+#include "TombstoneEntry.hpp"
+#include "TombstoneExpiryTask.hpp"
namespace apache {
namespace geode {
namespace client {
+// TODO. Review this overhead is OK
#define SIZEOF_PTR (sizeof(void*))
#define SIZEOF_SHAREDPTR (SIZEOF_PTR + 4)
// 3 variables in expiry handler, two variables for ace_reactor expiry, one
@@ -40,55 +43,45 @@ namespace client {
#define SIZEOF_TOMBSTONEOVERHEAD \
(SIZEOF_EXPIRYHANDLER + SIZEOF_TOMBSTONELISTENTRY)
-ExpiryTaskManager::id_type TombstoneList::getExpiryTask(
- TombstoneExpiryHandler** handler) {
- // This function is not guarded as all functions of this class are called from
- // MapSegment
- auto duration = m_cacheImpl->getDistributedSystem()
- .getSystemProperties()
- .tombstoneTimeout();
-
- auto tombstoneEntryPtr = std::make_shared<TombstoneEntry>(nullptr);
- *handler = new TombstoneExpiryHandler(tombstoneEntryPtr, this, duration,
- m_cacheImpl);
- tombstoneEntryPtr->setHandler(*handler);
- auto id = m_cacheImpl->getExpiryTaskManager().scheduleExpiryTask(
- *handler, duration, std::chrono::seconds::zero());
- return id;
-}
-
-void TombstoneList::add(const std::shared_ptr<MapEntryImpl>& entry,
- TombstoneExpiryHandler* handler,
- ExpiryTaskManager::id_type taskid) {
+ExpiryTask::id_t TombstoneList::add(
+ const std::shared_ptr<MapEntryImpl>& entry) {
// This function is not guarded as all functions of this class are called from
// MapSegment read TombstoneTImeout from systemProperties.
- auto tombstoneEntryPtr = std::make_shared<TombstoneEntry>(entry);
- handler->setTombstoneEntry(tombstoneEntryPtr);
- tombstoneEntryPtr->setHandler(handler);
+ auto tombstone = std::make_shared<TombstoneEntry>(entry);
std::shared_ptr<CacheableKey> key;
entry->getKeyI(key);
- tombstoneEntryPtr->setExpiryTaskId(taskid);
- m_tombstoneMap[key] = tombstoneEntryPtr;
- m_cacheImpl->getCachePerfStats().incTombstoneCount();
- auto tombstonesize = key->objectSize() + SIZEOF_TOMBSTONEOVERHEAD;
- m_cacheImpl->getCachePerfStats().incTombstoneSize(tombstonesize);
+ auto duration =
+ cache_.getDistributedSystem().getSystemProperties().tombstoneTimeout();
+ auto& manager = cache_.getExpiryTaskManager();
+ auto task =
+ std::make_shared<TombstoneExpiryTask>(manager, segment_, tombstone);
+ auto task_id = manager.schedule(std::move(task), duration);
+ tombstone->task_id(task_id);
+ tombstones_[key] = tombstone;
+
+ auto& perf_stats = cache_.getCachePerfStats();
+
+ perf_stats.incTombstoneCount();
+ perf_stats.incTombstoneSize(key->objectSize() + SIZEOF_TOMBSTONEOVERHEAD);
+
+ return task_id;
}
// Reaps the tombstones which have been gc'ed on server.
// A map that has identifier for ClientProxyMembershipID as key
// and server version of the tombstone with highest version as the
// value is passed as paramter
-void TombstoneList::reapTombstones(std::map<uint16_t, int64_t>& gcVersions) {
+void TombstoneList::reap_tombstones(std::map<uint16_t, int64_t>& gcVersions) {
// This function is not guarded as all functions of this class are called from
// MapSegment
std::unordered_set<std::shared_ptr<CacheableKey>,
dereference_hash<std::shared_ptr<CacheableKey>>,
dereference_equal_to<std::shared_ptr<CacheableKey>>>
tobeDeleted;
- for (const auto& queIter : m_tombstoneMap) {
+ for (const auto& queIter : tombstones_) {
auto const& mapIter = gcVersions.find(
- queIter.second->getEntry()->getVersionStamp().getMemberId());
+ queIter.second->entry()->getVersionStamp().getMemberId());
if (mapIter == gcVersions.end()) {
continue;
@@ -96,87 +89,61 @@ void TombstoneList::reapTombstones(std::map<uint16_t, int64_t>& gcVersions) {
auto const& version = (*mapIter).second;
if (version >=
- queIter.second->getEntry()->getVersionStamp().getRegionVersion()) {
+ queIter.second->entry()->getVersionStamp().getRegionVersion()) {
tobeDeleted.insert(queIter.first);
}
}
- for (const auto& itr : tobeDeleted) {
- unguardedRemoveEntryFromMapSegment(itr);
+
+ for (const auto& key : tobeDeleted) {
+ segment_.remove_entry(key);
}
}
// Reaps the tombstones whose keys are specified in the hash set .
-void TombstoneList::reapTombstones(
- std::shared_ptr<CacheableHashSet> removedKeys) {
+void TombstoneList::reap_tombstones(
+ const std::shared_ptr<CacheableHashSet>& keys) {
// This function is not guarded as all functions of this class are called from
// MapSegment
- for (auto queIter = removedKeys->begin(); queIter != removedKeys->end();
- ++queIter) {
- unguardedRemoveEntryFromMapSegment(*queIter);
+ for (const auto& key : *keys) {
+ segment_.remove_entry(key);
}
}
-// Call this when the lock of MapSegment has not been taken
-void TombstoneList::removeEntryFromMapSegment(
- std::shared_ptr<CacheableKey> key) {
- m_mapSegment->removeActualEntry(key, false);
-}
-
-// Call this when the lock of MapSegment has already been taken
-void TombstoneList::unguardedRemoveEntryFromMapSegment(
- std::shared_ptr<CacheableKey> key) {
- m_mapSegment->unguardedRemoveActualEntry(key);
-}
bool TombstoneList::exists(const std::shared_ptr<CacheableKey>& key) const {
- if (key) {
- return m_tombstoneMap.find(key) != m_tombstoneMap.end();
- }
- return false;
+ return key != nullptr && tombstones_.find(key) != tombstones_.end();
}
-void TombstoneList::eraseEntryFromTombstoneList(
- const std::shared_ptr<CacheableKey>& key, bool cancelTask) {
+bool TombstoneList::erase(const std::shared_ptr<CacheableKey>& key,
+ bool cancel_task) {
// This function is not guarded as all functions of this class are called from
// MapSegment
- if (exists(key)) {
- if (cancelTask) {
- m_cacheImpl->getExpiryTaskManager().cancelTask(
- m_tombstoneMap[key]->getExpiryTaskId());
- delete m_tombstoneMap[key]->getHandler();
- }
- m_cacheImpl->getCachePerfStats().decTombstoneCount();
- auto tombstonesize = key->objectSize() + SIZEOF_TOMBSTONEOVERHEAD;
- m_cacheImpl->getCachePerfStats().decTombstoneSize(tombstonesize);
- m_tombstoneMap.erase(key);
+ auto&& iter = tombstones_.find(key);
+ if (iter == tombstones_.end()) {
+ return false;
}
-}
-ExpiryTaskManager::id_type
-TombstoneList::eraseEntryFromTombstoneListWithoutCancelTask(
- const std::shared_ptr<CacheableKey>& key,
- TombstoneExpiryHandler*& handler) {
- // This function is not guarded as all functions of this class are called from
- // MapSegment
- ExpiryTaskManager::id_type taskid = -1;
- if (exists(key)) {
- taskid = m_tombstoneMap[key]->getExpiryTaskId();
- handler = m_tombstoneMap[key]->getHandler();
- m_cacheImpl->getCachePerfStats().decTombstoneCount();
- auto tombstonesize = key->objectSize() + SIZEOF_TOMBSTONEOVERHEAD;
- m_cacheImpl->getCachePerfStats().decTombstoneSize(tombstonesize);
- m_tombstoneMap.erase(key);
+ iter->second->invalidate();
+
+ if (cancel_task) {
+ cache_.getExpiryTaskManager().cancel(iter->second->task_id());
}
- return taskid;
+
+ auto& perf_stats = cache_.getCachePerfStats();
+
+ perf_stats.decTombstoneCount();
+ perf_stats.decTombstoneSize(key->objectSize() + SIZEOF_TOMBSTONEOVERHEAD);
+
+ tombstones_.erase(iter);
+ return true;
}
-void TombstoneList::cleanUp() {
+void TombstoneList::cleanup() {
// This function is not guarded as all functions of this class are called from
// MapSegment
- auto& expiryTaskManager = m_cacheImpl->getExpiryTaskManager();
- for (const auto& queIter : m_tombstoneMap) {
- expiryTaskManager.cancelTask(queIter.second->getExpiryTaskId());
- delete queIter.second->getHandler();
+ auto& manager = cache_.getExpiryTaskManager();
+ for (const auto& iter : tombstones_) {
+ manager.cancel(iter.second->task_id());
}
}
diff --git a/cppcache/src/TombstoneList.hpp b/cppcache/src/TombstoneList.hpp
index 2e81eeb..951159c 100644
--- a/cppcache/src/TombstoneList.hpp
+++ b/cppcache/src/TombstoneList.hpp
@@ -21,13 +21,13 @@
#define GEODE_TOMBSTONELIST_H_
#include <chrono>
-#include <list>
#include <memory>
#include <unordered_map>
#include <geode/CacheableBuiltins.hpp>
#include <geode/internal/functional.hpp>
+#include "ExpiryTask.hpp"
#include "MapEntry.hpp"
namespace apache {
@@ -35,70 +35,37 @@ namespace geode {
namespace client {
class MapSegment;
-class TombstoneExpiryHandler;
-
-class TombstoneEntry {
- public:
- using clock = std::chrono::steady_clock;
- using time_point = clock::time_point;
-
- explicit TombstoneEntry(const std::shared_ptr<MapEntryImpl>& entry)
- : m_entry(entry),
- m_tombstoneCreationTime(TombstoneEntry::clock::now()),
- m_expiryTaskId(0),
- m_handler(nullptr) {}
- virtual ~TombstoneEntry() {}
- std::shared_ptr<MapEntryImpl> getEntry() { return m_entry; }
- time_point getTombstoneCreationTime() { return m_tombstoneCreationTime; }
- ExpiryTaskManager::id_type getExpiryTaskId() { return m_expiryTaskId; }
- void setExpiryTaskId(ExpiryTaskManager::id_type expiryTaskId) {
- m_expiryTaskId = expiryTaskId;
- }
- TombstoneExpiryHandler* getHandler() { return m_handler; }
- void setHandler(TombstoneExpiryHandler* handler) { m_handler = handler; }
-
- private:
- std::shared_ptr<MapEntryImpl> m_entry;
- time_point m_tombstoneCreationTime;
- ExpiryTaskManager::id_type m_expiryTaskId;
- TombstoneExpiryHandler* m_handler;
-};
+class TombstoneEntry;
class TombstoneList {
public:
- TombstoneList(MapSegment* mapSegment, CacheImpl* cacheImpl)
- : m_mapSegment(mapSegment), m_cacheImpl(cacheImpl) {}
- virtual ~TombstoneList() { cleanUp(); }
- void add(const std::shared_ptr<MapEntryImpl>& entry,
- TombstoneExpiryHandler* handler, ExpiryTaskManager::id_type taskID);
+ TombstoneList(MapSegment& segment, CacheImpl& cache)
+ : segment_(segment), cache_(cache) {}
+ virtual ~TombstoneList() { cleanup(); }
+
+ ExpiryTask::id_t add(const std::shared_ptr<MapEntryImpl>& entry);
+ bool erase(const std::shared_ptr<CacheableKey>& key, bool cancel_task = true);
+ bool exists(const std::shared_ptr<CacheableKey>& key) const;
+ void cleanup();
// Reaps the tombstones which have been gc'ed on server.
// A map that has identifier for ClientProxyMembershipID as key
// and server version of the tombstone with highest version as the
// value is passed as paramter
- void reapTombstones(std::map<uint16_t, int64_t>& gcVersions);
- void reapTombstones(std::shared_ptr<CacheableHashSet> removedKeys);
- void eraseEntryFromTombstoneList(const std::shared_ptr<CacheableKey>& key,
- bool cancelTask = true);
- ExpiryTaskManager::id_type eraseEntryFromTombstoneListWithoutCancelTask(
- const std::shared_ptr<CacheableKey>& key,
- TombstoneExpiryHandler*& handler);
- void cleanUp();
- ExpiryTaskManager::id_type getExpiryTask(TombstoneExpiryHandler** handler);
- bool exists(const std::shared_ptr<CacheableKey>& key) const;
+ void reap_tombstones(std::map<uint16_t, int64_t>& gcVersions);
+ void reap_tombstones(const std::shared_ptr<CacheableHashSet>& keys);
+
+ protected:
+ using tombstone_map_t =
+ std::unordered_map<std::shared_ptr<CacheableKey>,
+ std::shared_ptr<TombstoneEntry>,
+ dereference_hash<std::shared_ptr<CacheableKey>>,
+ dereference_equal_to<std::shared_ptr<CacheableKey>>>;
- private:
- void removeEntryFromMapSegment(std::shared_ptr<CacheableKey> key);
- void unguardedRemoveEntryFromMapSegment(std::shared_ptr<CacheableKey> key);
- typedef std::unordered_map<
- std::shared_ptr<CacheableKey>, std::shared_ptr<TombstoneEntry>,
- dereference_hash<std::shared_ptr<CacheableKey>>,
- dereference_equal_to<std::shared_ptr<CacheableKey>>>
- TombstoneMap;
- TombstoneMap m_tombstoneMap;
- MapSegment* m_mapSegment;
- CacheImpl* m_cacheImpl;
- friend class TombstoneExpiryHandler;
+ protected:
+ tombstone_map_t tombstones_;
+ MapSegment& segment_;
+ CacheImpl& cache_;
};
} // namespace client
diff --git a/cppcache/src/TrackedMapEntry.cpp b/cppcache/src/TrackedMapEntry.cpp
index a413731..0bc21de 100644
--- a/cppcache/src/TrackedMapEntry.cpp
+++ b/cppcache/src/TrackedMapEntry.cpp
@@ -17,7 +17,7 @@
#include "TrackedMapEntry.hpp"
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
namespace apache {
namespace geode {
diff --git a/cppcache/src/VersionedCacheableObjectPartList.cpp b/cppcache/src/VersionedCacheableObjectPartList.cpp
index 47215d1..4260748 100644
--- a/cppcache/src/VersionedCacheableObjectPartList.cpp
+++ b/cppcache/src/VersionedCacheableObjectPartList.cpp
@@ -20,6 +20,7 @@
#include <geode/CacheableString.hpp>
#include <geode/ExceptionTypes.hpp>
+#include "CacheImpl.hpp"
#include "CacheableToken.hpp"
#include "DiskStoreId.hpp"
#include "DiskVersionTag.hpp"
diff --git a/cppcache/test/CMakeLists.txt b/cppcache/test/CMakeLists.txt
index 2386c23..3f22faf 100644
--- a/cppcache/test/CMakeLists.txt
+++ b/cppcache/test/CMakeLists.txt
@@ -37,9 +37,12 @@ add_executable(apache-geode_unittests
DataInputTest.cpp
DataOutputTest.cpp
ExceptionTypesTest.cpp
+ ExpiryTaskTest.cpp
+ ExpiryTaskManagerTest.cpp
GatewaySenderEventCallbackArgumentTest.cpp
geodeBannerTest.cpp
gtest_extensions.h
+ gmock_extensions.h
InterestResultPolicyTest.cpp
LocalRegionTest.cpp
LoggingTest.cpp
@@ -55,6 +58,7 @@ add_executable(apache-geode_unittests
TcrMessageTest.cpp
ThreadPoolTest.cpp
TXIdTest.cpp
+ mock/MockExpiryTask.hpp
mock/MapEntryImplMock.hpp
statistics/HostStatSamplerTest.cpp
util/functionalTests.cpp
diff --git a/cppcache/test/ExpiryTaskManagerTest.cpp b/cppcache/test/ExpiryTaskManagerTest.cpp
new file mode 100644
index 0000000..50cdad7
--- /dev/null
+++ b/cppcache/test/ExpiryTaskManagerTest.cpp
@@ -0,0 +1,148 @@
+/*
+ * 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 <gtest/gtest.h>
+
+#include <geode/ExceptionTypes.hpp>
+
+#include "ExpiryTaskManager.hpp"
+#include "mock/MockExpiryTask.hpp"
+
+using apache::geode::client::ExpiryTask;
+using apache::geode::client::ExpiryTaskManager;
+using apache::geode::client::IllegalStateException;
+using apache::geode::client::MockExpiryTask;
+
+TEST(ExpiryTaskManagerTest, startStop) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+ EXPECT_TRUE(manager.running());
+ EXPECT_NO_THROW(manager.stop());
+}
+
+TEST(ExpiryTaskManagerTest, destroyWithoutStart) { ExpiryTaskManager manager; }
+
+TEST(ExpiryTaskManagerTest, destroyWithoutStop) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+}
+
+TEST(ExpiryTaskManagerTest, startTwice) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+ EXPECT_THROW(manager.start(), IllegalStateException);
+}
+
+TEST(ExpiryTaskManagerTest, stopTwice) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+ EXPECT_NO_THROW(manager.stop());
+ EXPECT_THROW(manager.stop(), IllegalStateException);
+}
+
+TEST(ExpiryTaskManagerTest, schedule) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 1U);
+}
+
+TEST(ExpiryTaskManagerTest, scheduleStop) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto id_first = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id_first, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 1U);
+
+ task = std::make_shared<MockExpiryTask>(manager);
+ auto id_second = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id_second, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 2U);
+
+ EXPECT_NO_THROW(manager.stop());
+ EXPECT_EQ(manager.count(), 0U);
+}
+
+TEST(ExpiryTaskManagerTest, scheduleCancelStop) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto id_first = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id_first, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 1U);
+
+ task = std::make_shared<MockExpiryTask>(manager);
+ auto id_second = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id_second, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 2U);
+
+ EXPECT_EQ(manager.cancel(id_first), 1U);
+ EXPECT_EQ(manager.cancel(id_second), 1U);
+ EXPECT_NO_THROW(manager.stop());
+ EXPECT_EQ(manager.count(), 0U);
+}
+
+TEST(ExpiryTaskManagerTest, scheduleWithoutStart) {
+ ExpiryTaskManager manager;
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(1));
+ EXPECT_EQ(id, ExpiryTask::invalid());
+}
+
+TEST(ExpiryTaskManagerTest, reset) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 1U);
+ EXPECT_EQ(manager.reset(id, std::chrono::seconds(1)), 1U);
+}
+
+TEST(ExpiryTaskManagerTest, resetInvalid) {
+ ExpiryTaskManager manager;
+ manager.start();
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ (void)manager.schedule(std::move(task), std::chrono::seconds(1));
+ EXPECT_EQ(manager.reset(10, std::chrono::seconds(1)), -1);
+}
+
+TEST(ExpiryTaskManagerTest, cancel) {
+ ExpiryTaskManager manager;
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(10));
+ EXPECT_NE(id, ExpiryTask::invalid());
+ EXPECT_EQ(manager.count(), 1U);
+ EXPECT_EQ(manager.cancel(id), 1U);
+}
+
+TEST(ExpiryTaskManagerTest, cancelInvalid) {
+ ExpiryTaskManager manager;
+ manager.start();
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ (void)manager.schedule(std::move(task), std::chrono::seconds(1));
+ EXPECT_EQ(manager.cancel(10), -1);
+}
diff --git a/cppcache/test/ExpiryTaskTest.cpp b/cppcache/test/ExpiryTaskTest.cpp
new file mode 100644
index 0000000..9f0cc37
--- /dev/null
+++ b/cppcache/test/ExpiryTaskTest.cpp
@@ -0,0 +1,153 @@
+/*
+ * 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 <gtest/gtest.h>
+
+#include <geode/ExceptionTypes.hpp>
+
+#include "ExpiryTaskManager.hpp"
+#include "gmock_extensions.h"
+#include "mock/MockExpiryTask.hpp"
+
+using ::testing::DoAll;
+using ::testing::InvokeWithoutArgs;
+using ::testing::Return;
+using ::testing::Sequence;
+
+using apache::geode::client::ExpiryTask;
+using apache::geode::client::ExpiryTaskManager;
+using apache::geode::client::IllegalStateException;
+using apache::geode::client::MockExpiryTask;
+
+constexpr std::chrono::milliseconds DEFAULT_TIMEOUT{100};
+
+TEST(ExpiryTaskTest, scheduleSingleshot) {
+ std::mutex cv_mutex;
+ ExpiryTaskManager manager;
+ std::condition_variable cv;
+ std::unique_lock<std::mutex> lock(cv_mutex);
+
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ EXPECT_CALL(*task, on_expire())
+ .Times(1)
+ .WillOnce(DoAll(CvNotifyOne(&cv), Return(true)));
+
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(0));
+ EXPECT_NE(id, ExpiryTask::invalid());
+
+ EXPECT_EQ(cv.wait_for(lock, DEFAULT_TIMEOUT), std::cv_status::no_timeout);
+
+ EXPECT_NO_THROW(manager.stop());
+}
+
+TEST(ExpiryTaskTest, schedulePeriodic) {
+ std::mutex cv_mutex;
+ ExpiryTaskManager manager;
+ std::condition_variable cv;
+ std::unique_lock<std::mutex> lock(cv_mutex);
+
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ {
+ Sequence s;
+ EXPECT_CALL(*task, on_expire())
+ .Times(1)
+ .InSequence(s)
+ .WillOnce(Return(true));
+ EXPECT_CALL(*task, on_expire())
+ .Times(1)
+ .InSequence(s)
+ .WillOnce(DoAll(CvNotifyOne(&cv), Return(true)));
+ EXPECT_CALL(*task, on_expire()).InSequence(s).WillRepeatedly(Return(true));
+ }
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(0),
+ std::chrono::nanoseconds(1));
+ EXPECT_NE(id, ExpiryTask::invalid());
+
+ EXPECT_EQ(cv.wait_for(lock, DEFAULT_TIMEOUT), std::cv_status::no_timeout);
+
+ EXPECT_NO_THROW(manager.stop());
+}
+
+TEST(ExpiryTaskTest, resetSingleshot) {
+ std::mutex cv_mutex;
+ ExpiryTaskManager manager;
+ std::condition_variable cv;
+ std::unique_lock<std::mutex> lock(cv_mutex);
+
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto& task_ref = *task;
+
+ {
+ Sequence s;
+ EXPECT_CALL(task_ref, on_expire())
+ .Times(1)
+ .InSequence(s)
+ .WillOnce(InvokeWithoutArgs([&task_ref] {
+ task_ref.reset(std::chrono::nanoseconds(1));
+ return false;
+ }));
+ EXPECT_CALL(task_ref, on_expire())
+ .Times(1)
+ .InSequence(s)
+ .WillOnce(DoAll(CvNotifyOne(&cv), Return(true)));
+ }
+
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(0));
+ EXPECT_NE(id, ExpiryTask::invalid());
+
+ EXPECT_EQ(cv.wait_for(lock, DEFAULT_TIMEOUT), std::cv_status::no_timeout);
+
+ EXPECT_NO_THROW(manager.stop());
+}
+
+TEST(ExpiryTaskTest, resetPeriodic) {
+ std::mutex cv_mutex;
+ ExpiryTaskManager manager;
+ std::condition_variable cv;
+ std::unique_lock<std::mutex> lock(cv_mutex);
+
+ EXPECT_NO_THROW(manager.start());
+
+ auto task = std::make_shared<MockExpiryTask>(manager);
+ auto& task_ref = *task;
+
+ {
+ Sequence s;
+ EXPECT_CALL(task_ref, on_expire())
+ .Times(1)
+ .InSequence(s)
+ .WillOnce(InvokeWithoutArgs([&task_ref] {
+ task_ref.reset(std::chrono::nanoseconds(1));
+ return false;
+ }));
+ EXPECT_CALL(task_ref, on_expire())
+ .InSequence(s)
+ .WillRepeatedly(DoAll(CvNotifyOne(&cv), Return(true)));
+ }
+ auto id = manager.schedule(std::move(task), std::chrono::seconds(0));
+ EXPECT_NE(id, ExpiryTask::invalid());
+
+ EXPECT_EQ(cv.wait_for(lock, DEFAULT_TIMEOUT), std::cv_status::no_timeout);
+
+ EXPECT_NO_THROW(manager.stop());
+}
diff --git a/cppcache/test/PdxInstanceImplTest.cpp b/cppcache/test/PdxInstanceImplTest.cpp
index d54f3e7..d93f9c7 100644
--- a/cppcache/test/PdxInstanceImplTest.cpp
+++ b/cppcache/test/PdxInstanceImplTest.cpp
@@ -15,19 +15,17 @@
* limitations under the License.
*/
-#include <CachePerfStats.hpp>
-#include <PdxInstanceImpl.hpp>
-#include <statistics/StatisticsFactory.hpp>
-
#include <gtest/gtest.h>
#include <geode/AuthenticatedView.hpp>
-#include <geode/Cache.hpp>
-#include <geode/PoolManager.hpp>
#include <geode/Properties.hpp>
#include <geode/RegionFactory.hpp>
#include <geode/RegionShortcut.hpp>
+#include "CacheImpl.hpp"
+#include "PdxInstanceImpl.hpp"
+#include "statistics/StatisticsFactory.hpp"
+
using apache::geode::client::Cache;
using apache::geode::client::CacheFactory;
using apache::geode::client::CacheImpl;
diff --git a/clicache/src/impl/MemoryPressureHandler.hpp b/cppcache/test/gmock_extensions.h
similarity index 59%
rename from clicache/src/impl/MemoryPressureHandler.hpp
rename to cppcache/test/gmock_extensions.h
index 9ac0a6d..34b688f 100644
--- a/clicache/src/impl/MemoryPressureHandler.hpp
+++ b/cppcache/test/gmock_extensions.h
@@ -17,29 +17,11 @@
#pragma once
+#ifndef GEODE_GMOCK_EXTENSIONS_H_
+#define GEODE_GMOCK_EXTENSIONS_H_
-#include "../begin_native.hpp"
-#include <geode/internal/geode_globals.hpp>
-#include <ExpiryTaskManager.hpp>
-#include "../end_native.hpp"
+#include <gmock/gmock.h>
-namespace Apache
-{
- namespace Geode
- {
- namespace Client
- {
-
- class MemoryPressureHandler
- : public ACE_Event_Handler
- {
- public:
- int handle_timeout( const ACE_Time_Value& current_time,
- const void* arg );
-
- int handle_close( ACE_HANDLE handle, ACE_Reactor_Mask close_mask );
- };
- } // namespace Client
- } // namespace Geode
-} // namespace Apache
+ACTION_P(CvNotifyOne, cv) { cv->notify_one(); }
+#endif // GEODE_GMOCK_EXTENSIONS_H_
diff --git a/cppcache/test/mock/MapEntryImplMock.hpp b/cppcache/test/mock/MapEntryImplMock.hpp
index 68fc6c7..ab234d0 100644
--- a/cppcache/test/mock/MapEntryImplMock.hpp
+++ b/cppcache/test/mock/MapEntryImplMock.hpp
@@ -22,7 +22,7 @@
#include <gmock/gmock.h>
-#include "MapEntry.hpp"
+#include "MapEntryImpl.hpp"
namespace apache {
namespace geode {
diff --git a/cppcache/src/MapEntry.cpp b/cppcache/test/mock/MockExpiryTask.hpp
similarity index 68%
rename from cppcache/src/MapEntry.cpp
rename to cppcache/test/mock/MockExpiryTask.hpp
index a6667c5..c7363b5 100644
--- a/cppcache/src/MapEntry.cpp
+++ b/cppcache/test/mock/MockExpiryTask.hpp
@@ -15,24 +15,26 @@
* limitations under the License.
*/
-#include "MapEntry.hpp"
+#ifndef GEODE_MOCKEXPIRYTASK_H_
+#define GEODE_MOCKEXPIRYTASK_H_
-#include "MapEntryT.hpp"
+#include <gmock/gmock.h>
+
+#include "ExpiryTask.hpp"
namespace apache {
namespace geode {
namespace client {
+class MockExpiryTask : public ExpiryTask {
+ public:
+ explicit MockExpiryTask(ExpiryTaskManager& manager) : ExpiryTask(manager) {}
-void EntryFactory::newMapEntry(ExpiryTaskManager*,
- const std::shared_ptr<CacheableKey>& key,
- std::shared_ptr<MapEntryImpl>& result) const {
- if (m_concurrencyChecksEnabled) {
- result = MapEntryT<VersionedMapEntryImpl, 0, 0>::create(key);
- } else {
- result = MapEntryT<MapEntryImpl, 0, 0>::create(key);
- }
-}
+ using ExpiryTask::reset;
+ MOCK_METHOD0(on_expire, bool());
+};
} // namespace client
} // namespace geode
} // namespace apache
+
+#endif // GEODE_MOCKEXPIRYTASK_H_
diff --git a/tests/cpp/testobject/CMakeLists.txt b/tests/cpp/testobject/CMakeLists.txt
index fee4ee1..538afa2 100644
--- a/tests/cpp/testobject/CMakeLists.txt
+++ b/tests/cpp/testobject/CMakeLists.txt
@@ -95,6 +95,7 @@ target_link_libraries(testobject
framework
PRIVATE
ACE::ACE
+ Boost::boost
_WarningsAsError
)