23 CacheMap
CacheMap<Key,Value,Calculator>
implements a cache that maps from Key to Value.
It is assumed the Value can be calculated from the Key with an externally
provided "calculator" function.
Saying that the value is "calculated" from the key should not be taken too literally. For example, it may correspond to loading an object from disk where the key represents some kind of object identifier.
Clients retrieve values for given key values with calls to GetValue(). This method is threadsafe.
The implementation ensures that at most one thread calculates a value for a given key. I.e. when a
key is not found in the map the first calling thread is given responsibility for calculating the
value and subsequent threads block in an efficient wait state until the value is calculated.
Note that the calculator must be threadsafe - i.e. it must itself allow for concurrent calculation of values for different keys.
The calculator must implement operator() that takes the key as a single argument (either by value
or reference) and returns the calculated value.
struct MyCalculator
{
Value operator()(const Key& key);
};
It is permissible for operator() to throw an exception! This exception will be passed out to the
thread that called GetValue() on the CacheMap. Queued threads will subsequently get their own
chance to try to calculate the value (and of course they may throw exceptions as well).
The calculator is stored as a member of the CacheMap, and may be provided in the constructor.
The CacheMap doesn't implement any policy for removing entries. Clear() and RemoveCachedValue()
are functions provided to support removal of map entries.
Usage in CEDA
CacheMap was used to implement the Resident Object Table (ROT) of a PSpace.
However that is no longer done because the ROT requirements became more complex in order to support asynchronous
binding to objects from the LSS, which represent independent nodes of the DGS.
CacheMap.h
// CacheMap.h
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2006
#pragma once
#ifndef Ceda_cxThread_CacheMap_H
#define Ceda_cxThread_CacheMap_H
#include "cxThread.h"
#include "ManualResetEvent.h"
#include "Ceda/cxUtils/CedaAssert.h"
#include <map>
#include <mutex>
namespace ceda
{
template <typename Key, typename Value, typename Calculator>
class CacheMap
{
public:
CacheMap(Calculator c = Calculator()) : calculator_(c) {}
void Clear()
{
std::lock_guard<std::mutex> lock(_mutex);
map_.clear();
}
bool RemoveCachedValue(const Key& key)
{
std::lock_guard<std::mutex> lock(_mutex);
ssize_t numRemoved = map_.erase(key);
cxAssert(numRemoved == 0 || numRemoved == 1);
return numRemoved == 1;
}
// Used to poll for whether an entry for the given key is currently present in the map.
bool IsCached(const Key& key) const
{
std::lock_guard<std::mutex> lock(_mutex);
return map_.find(key) != map_.end();
}
// Allows for polling whether an entry in the map is already present. This function always
// returns quickly without blocking on a calculation - either directly or indirectly.
bool GetCachedValue(const Key& key, Value& v) const
{
std::lock_guard<std::mutex> lock(_mutex);
typename MAP::const_iterator i = map_.find(key);
if (i != map_.end())
{
v = i->second;
return true;
}
else
{
return false;
}
}
// Try to insert a (key,value) pair. Returns false if insertion failed because that key
// is already present.
bool InsertCachedValue(const Key& key, const Value& value)
{
std::lock_guard<std::mutex> lock(_mutex);
std::pair<typename MAP::iterator,bool> r = map_.insert(MAP::value_type(key,value));
return r.second;
}
void SetCachedValue(const Key& key, const Value& value)
{
std::lock_guard<std::mutex> lock(_mutex);
map_[key] = value;
}
Value GetValue(const Key& key)
{
retry:
EventEntry* e = nullptr;
bool needToCalc = false;
{
std::lock_guard<std::mutex> lock(_mutex);
typename MAP::const_iterator i = map_.find(key);
if (i != map_.end())
{
// Value is cached in the map
return i->second;
}
else
{
typename EVENTMAP::iterator j = eventMap_.find(key);
if (j == eventMap_.end())
{
// This thread is responsible for calculating the value
needToCalc = true;
e = &eventMap_[key];
cxAssert(e->status_ == NOT_CALCULATED);
e->status_ = CALCULATING;
}
else
{
// Another thread has already taken responsibility for calculating the value
e = &j->second;
if (e->status_ == NOT_CALCULATED)
{
// The thread that had tried to calculate the value must have thrown an
// exception. Therefore make this thread responsible for calculating the
// value
needToCalc = true;
e->status_ = CALCULATING;
cxAssert(e->count_ > 0);
cxAssert(e->event_);
e->event_->Reset();
}
else if (e->status_ == CALCULATED)
{
// This is only possible if entries have been removed from the main map
// or AllowInsertionInMap() can return false.
return e->value_;
}
else
{
cxAssert(e->status_ == CALCULATING);
if (e->count_++ == 0)
{
cxAssert( e->event_ == nullptr );
e->event_ = new ManualResetEvent;
}
}
cxAssert( e->event_ );
}
}
}
cxAssert(e);
if (needToCalc)
{
Value v;
try
{
cxAssert(e->status_ == CALCULATING);
v = calculator_(key);
}
catch(...) // todo: find a better solution, catching with ... and rethrowing stops the visual studio debugger stopping at the point where an access violation occurred etc
{
std::lock_guard<std::mutex> lock(_mutex);
if (e->count_ == 0)
{
cxAssert( e->event_ == nullptr );
eventMap_.erase(key);
}
else
{
// Calculation failed. Need to indicate failure to other threads and signal
// them
cxAssert( e->event_ );
e->status_ = NOT_CALCULATED;
e->event_->Signal();
}
throw;
}
std::lock_guard<std::mutex> lock(_mutex);
if (calculator_.AllowInsertionInMap(v))
{
map_[key] = v;
}
if (e->count_ == 0)
{
cxAssert( e->event_ == nullptr );
eventMap_.erase(key);
}
else
{
cxAssert( e->event_ );
e->value_ = v;
e->status_ = CALCULATED;
e->event_->Signal();
}
return v;
}
else
{
cxAssert( e->event_ );
e->event_->Wait();
std::lock_guard<std::mutex> lock(_mutex);
if (e->status_ == CALCULATED)
{
Value v = e->value_;
if (--e->count_ == 0)
{
delete e->event_;
eventMap_.erase(key);
}
return v;
}
else
{
if (--e->count_ == 0)
{
delete e->event_;
e->event_ = nullptr;
if (e->status_ == NOT_CALCULATED)
{
eventMap_.erase(key);
}
}
goto retry;
}
}
}
void Transfer(CacheMap& src)
{
std::lock_guard<std::mutex> lock1(_mutex);
std::lock_guard<std::mutex> lock2(src._mutex);
cxAssert(src.eventMap_.empty());
for (typename MAP::iterator i = src.map_.begin() ; i != src.map_.end() ; ++i)
{
cxAssert(map_.find(i->first) == map_.end());
map_[i->first] = i->second;
}
src.map_.clear();
}
Calculator& GetCalculator() { return calculator_; }
protected:
mutable std::mutex _mutex;
Calculator calculator_;
typedef std::map<Key,Value> MAP;
MAP map_;
enum EStatus
{
NOT_CALCULATED,
CALCULATING,
CALCULATED
};
struct EventEntry
{
EventEntry() : status_(NOT_CALCULATED), count_(0), event_(nullptr) {}
EStatus status_;
int count_;
ManualResetEvent* event_;
Value value_;
};
friend struct EventEntry;
typedef std::map<Key, EventEntry> EVENTMAP;
EVENTMAP eventMap_;
};
} // namespace ceda
#endif // include guardCacheMapTests.cpp in txThread
// CacheMapTests.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2006
#ifdef _MSC_VER
// Identifier was truncated to 'number' characters in the debug information
#pragma warning(disable: 4786)
#endif
#include "Ceda/cxThread/CacheMap.h"
#include "Ceda/cxUtils/Tracer.h"
#include "Ceda/cxUtils/IsaacRandom.h"
#include "Ceda/cxUtils/TestTimer.h"
#include <thread>
#include <mutex>
#include <algorithm>
///////////////////////////////////////////////////////////////////////////////////////////////////
// Timing test
struct IncrementCalculator
{
IncrementCalculator() : numCalcs_(0) {}
int operator()(int x)
{
++numCalcs_;
return x+1;
}
static bool AllowInsertionInMap(int) { return true; }
int numCalcs_;
};
typedef ceda::CacheMap<int,int,IncrementCalculator> IncrementCacheMap;
/*
Performance on Pentium IV 2GHz
numIterations numKeys Time (ns)
1000000 1 200
1000000 10 250
1000000 100 280
1000000 1000 340
1000000 10000 480
1000000 100000 2150
1000000 1000000 6300
1000000 10000000 9000
1000000 100000000 9500
Performance on Intel Core 2 Duo E8400 3GHz
numIterations numKeys Time (ns)
1000000 1 57
1000000 10 72
1000000 100 96
1000000 1000 112
1000000 10000 172
1000000 100000 280
1000000 1000000 890
1000000 10000000 1270
1000000 100000000 1440
*/
void CacheMapTimingTest(double timeForTestInSecs)
{
Tracer() << "CacheMap timing tests\n";
ceda::TraceIndenter indent(4);
int NK = 9;
double timePerTest = timeForTestInSecs/NK;
// Tests with numKeys = 1,10,100,...,100M
int numKeys = 1;
for (int i=0 ; i < NK ; ++i)
{
bool seedFromClock = true;
ceda::IsaacRandomNumberGen isaac(seedFromClock);
double rate = 300000.0;
if (numKeys > 500) rate = 50000;
ceda::TimeCode( cxMakeString2("GetValue() with random keys on IncrementCacheMap with " << numKeys << " keys"), timePerTest, rate, 10000000,
[numKeys, &isaac](int n)
{
IncrementCacheMap cm;
for (int i = 0 ; i < n ; ++i)
{
int r = isaac.GetUniformDistInteger(0,numKeys);
cxAlwaysAssert(cm.GetValue(r) == r+1);
}
Tracer() << " Num calc = " << cm.GetCalculator().numCalcs_ << '\n';
});
numKeys *= 10;
}
}
///////////////////////////////////////////////////////////////////////////////////////////////////
// Unit test
class AbortedException {};
int volatile g_v = 0;
int g_counter = 0;
struct TestCalculator
{
TestCalculator() {}
TestCalculator(const TestCalculator& rhs) : isaac_(rhs.isaac_) {}
int operator()(int x)
{
for (int i=0 ; i < 100 ; ++i)
{
++g_v;
}
//Sleep(1);
{
std::lock_guard<std::mutex> lock(mutex_);
if (isaac_.GetUniformDistInteger(0,10) == 0)
{
//Tracer() << "Throwing exception\n";
throw AbortedException();
}
}
//Sleep(1);
++g_counter;
return x+1;
}
static bool AllowInsertionInMap(int) { return true; }
std::mutex mutex_;
ceda::IsaacRandomNumberGen isaac_;
};
typedef ceda::CacheMap<int,int,TestCalculator> DemoCacheMap;
///////////////////////////////////////////////////////////////////////////////////////////////////
// CacheMapThread
const bool SEED_FROM_CLOCK = true;
class CacheMapThread
{
public:
CacheMapThread(DemoCacheMap& cm, int numKeys, bool trace, int indent) :
cm_(cm),
isaac_(SEED_FROM_CLOCK),
requestShutDown_(false),
numGets_(0),
numFails_(0),
//numKeys_(numKeys),
trace_(trace),
indent_(indent)
{
thread_ = std::thread( &CacheMapThread::WorkerThreadFn, this );
}
~CacheMapThread()
{
if (trace_)
{
Tracer() << "Stopping CacheMapThread : numGets_ = " << numGets_ << " numFails_ = " << numFails_ << '\n';
}
requestShutDown_ = true;
thread_.join(); // Blocks until worker thread exits
}
void WorkerThreadFn();
private:
DemoCacheMap& cm_;
ceda::IsaacRandomNumberGen isaac_;
std::thread thread_;
volatile bool requestShutDown_;
int numGets_;
int numFails_;
//int numKeys_;
bool trace_;
int indent_;
};
void CacheMapThread::WorkerThreadFn()
{
ceda::TraceIndenter id(indent_);
//Tracer() << "CacheMapThread " << GetCurrentThreadId() << '\n';
while(!requestShutDown_)
{
if (isaac_.GetUniformDistInteger(0,10) == 0)
{
Sleep(1);
}
++numGets_;
if (isaac_.GetUniformDistInteger(0,5) == 0)
{
cm_.RemoveCachedValue( g_counter + isaac_.GetUniformDistInteger(-4,4) );
}
int r = g_counter + isaac_.GetUniformDistInteger(-4,4);
try
{
cxAlwaysAssert(cm_.GetValue(r) == r+1);
}
catch(AbortedException&)
{
++numFails_;
//Tracer() << "Caught abort exception\n";
}
}
}
void UnitTestCacheMap2(double timeForTestInSecs,int numKeys,bool trace)
{
ceda::HPTimer timer;
g_counter = 0;
DemoCacheMap cm;
CacheMapThread t1(cm, numKeys, trace, 20);
Sleep(5);
CacheMapThread t2(cm, numKeys, trace, 40);
Sleep(5);
CacheMapThread t3(cm, numKeys, trace, 60);
Sleep(5);
CacheMapThread t4(cm, numKeys, trace, 80);
Sleep(5);
CacheMapThread t5(cm, numKeys, trace, 100);
Sleep(5);
CacheMapThread t6(cm, numKeys, trace, 120);
Sleep(5);
for (int i=0 ; i < (int) (timeForTestInSecs*10) ; ++i)
{
if (i % 10 == 0) ceda::xcout << "Time: " << i/10.0 << '\n' << ceda::flush;
Sleep(100);
}
timer.Done() << "numKeys= " << numKeys << " calls to CalcValue = " << g_counter;
}
///////////////////////////////////////////////////////////////////////////////////////////////////
void UnitTestCacheMap(double timeForTestInSecs)
{
Tracer() << "Unit Test CacheMap\n";
ceda::TraceIndenter indent(4);
int NK = 9;
int numExperiments = 5;
if (timeForTestInSecs < 10)
{
numExperiments = 1;
NK = 4;
}
double timePerTest = timeForTestInSecs/(NK*numExperiments);
Tracer() << "timePerTest = " << timePerTest << '\n';
// Tests with numKeys = 1,10,100,...,100M
int numKeys = 1;
for (int i=0 ; i < NK ; ++i)
{
for (int c = 0; c < numExperiments ; ++c)
{
UnitTestCacheMap2(timePerTest,numKeys, i==0);
}
numKeys *= 10;
}
}