Open-Harbor/osrm-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Refactor shared memory storage to allow for multiple named datasets

Browse Source
This commit is contained in:
Patrick Niklaus 2018-03-28 12:11:47 +00:00 committed by Patrick Niklaus
parent bc120776f0
commit 666ce46d36
8 changed files with 169 additions and 98 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
include/engine/data_watchdog.hpp
View File

@ -32,20 +32,28 @@ template <typename AlgorithmT, typename FacadeT> class DataWatchdogImpl;
template <typename AlgorithmT>
class DataWatchdogImpl<AlgorithmT, datafacade::ContiguousInternalMemoryDataFacade<AlgorithmT>> final
{
using mutex_type = typename storage::SharedMonitor<storage::SharedDataTimestamp>::mutex_type;
using mutex_type = typename storage::SharedMonitor<storage::SharedRegionRegister>::mutex_type;
using Facade = datafacade::ContiguousInternalMemoryDataFacade<AlgorithmT>;
public:
DataWatchdogImpl() : active(true), timestamp(0)
DataWatchdogImpl() : active(true)
{
// create the initial facade before launching the watchdog thread
{
boost::interprocess::scoped_lock<mutex_type> current_region_lock(barrier.get_mutex());
auto& shared_register = barrier.data();
auto region_id = shared_register.Find("data");
if (region_id == storage::SharedRegionRegister::INVALID_REGION_ID)
{
throw util::exception("Could not find shared memory region. Did you run osrm-datastore?");
}
shared_region = &shared_register.GetRegion(region_id);
region = *shared_region;
facade_factory =
DataFacadeFactory<datafacade::ContiguousInternalMemoryDataFacade, AlgorithmT>(
std::make_shared<datafacade::SharedMemoryAllocator>(barrier.data().region));
timestamp = barrier.data().timestamp;
std::make_shared<datafacade::SharedMemoryAllocator>(region.shm_key));
}
watcher = std::thread(&DataWatchdogImpl::Run, this);
@ -74,30 +82,30 @@ class DataWatchdogImpl<AlgorithmT, datafacade::ContiguousInternalMemoryDataFacad
{
boost::interprocess::scoped_lock<mutex_type> current_region_lock(barrier.get_mutex());
while (active && timestamp == barrier.data().timestamp)
while (active && region.timestamp == shared_region->timestamp)
{
barrier.wait(current_region_lock);
}
if (timestamp != barrier.data().timestamp)
if (region.timestamp != shared_region->timestamp)
{
auto region = barrier.data().region;
region = *shared_region;
facade_factory =
DataFacadeFactory<datafacade::ContiguousInternalMemoryDataFacade, AlgorithmT>(
std::make_shared<datafacade::SharedMemoryAllocator>(region));
timestamp = barrier.data().timestamp;
util::Log() << "updated facade to region " << region << " with timestamp "
<< timestamp;
std::make_shared<datafacade::SharedMemoryAllocator>(region.shm_key));
util::Log() << "updated facade to region " << region.shm_key << " with timestamp "
<< region.timestamp;
}
}
util::Log() << "DataWatchdog thread stopped";
}
storage::SharedMonitor<storage::SharedDataTimestamp> barrier;
storage::SharedMonitor<storage::SharedRegionRegister> barrier;
std::thread watcher;
bool active;
unsigned timestamp;
storage::SharedRegion region;
storage::SharedRegion *shared_region;
DataFacadeFactory<datafacade::ContiguousInternalMemoryDataFacade, AlgorithmT> facade_factory;
};
}

2
include/engine/datafacade/shared_memory_allocator.hpp
View File

@ -23,7 +23,7 @@ namespace datafacade
class SharedMemoryAllocator : public ContiguousBlockAllocator
{
public:
explicit SharedMemoryAllocator(storage::SharedDataType data_region);
explicit SharedMemoryAllocator(storage::SharedRegionRegister::ShmKey data_shm_key);
~SharedMemoryAllocator() override final;
// interface to give access to the datafacades

114
include/storage/shared_datatype.hpp
View File

@ -6,13 +6,13 @@
#include "util/exception.hpp"
#include "util/exception_utils.hpp"
#include "util/log.hpp"
#include <boost/assert.hpp>
#include <array>
#include <cstdint>
#include <map>
#include <numeric>
#include <unordered_set>
namespace osrm
@ -160,40 +160,100 @@ class DataLayout
std::map<std::string, Block> blocks;
};
enum SharedDataType
struct SharedRegion
{
REGION_NONE,
REGION_1,
REGION_2
};
static constexpr const int MAX_NAME_LENGTH = 254;
struct SharedDataTimestamp
{
explicit SharedDataTimestamp(SharedDataType region, unsigned timestamp)
: region(region), timestamp(timestamp)
SharedRegion() : name{0}, timestamp{0} {}
SharedRegion(const std::string &name_, std::uint64_t timestamp, std::uint8_t shm_key)
: name{0}, timestamp{timestamp}, shm_key{shm_key}
{
std::copy_n(name_.begin(), std::min<std::size_t>(MAX_NAME_LENGTH, name_.size()), name);
}
SharedDataType region;
unsigned timestamp;
bool IsEmpty() const { return timestamp == 0; }
char name[MAX_NAME_LENGTH + 1];
std::uint64_t timestamp;
std::uint8_t shm_key;
};
// Keeps a list of all shared regions in a fixed-sized struct
// for fast access and deserialization.
struct SharedRegionRegister
{
using RegionID = std::uint8_t;
static constexpr const RegionID INVALID_REGION_ID = std::numeric_limits<RegionID>::max();
using ShmKey = decltype(SharedRegion::shm_key);
// Returns the key of the region with the given name
RegionID Find(const std::string &name) const
{
auto iter = std::find_if(regions.begin(), regions.end(), [&](const auto &region) {
return std::strncmp(region.name, name.c_str(), SharedRegion::MAX_NAME_LENGTH) == 0;
});
if (iter == regions.end())
{
return INVALID_REGION_ID;
}
else
{
return std::distance(regions.begin(), iter);
}
}
RegionID Register(const std::string &name, ShmKey key)
{
auto iter = std::find_if(
regions.begin(), regions.end(), [&](const auto &region) { return region.IsEmpty(); });
if (iter == regions.end())
{
throw util::exception("No shared memory regions left. Could not register " + name +
".");
}
else
{
constexpr std::uint32_t INITIAL_TIMESTAMP = 1;
*iter = SharedRegion{name, INITIAL_TIMESTAMP, key};
RegionID key = std::distance(regions.begin(), iter);
return key;
}
}
void Deregister(const RegionID key) { regions[key] = SharedRegion{}; }
const auto &GetRegion(const RegionID key) const { return regions[key]; }
auto &GetRegion(const RegionID key) { return regions[key]; }
ShmKey ReserveKey()
{
auto free_key_iter = std::find(shm_key_in_use.begin(), shm_key_in_use.end(), false);
if (free_key_iter == shm_key_in_use.end())
{
throw util::exception("Could not reserve a new SHM key. All keys are in use");
}
*free_key_iter = true;
return std::distance(shm_key_in_use.begin(), free_key_iter);
}
void ReleaseKey(ShmKey key) { shm_key_in_use[key] = false; }
static constexpr const std::uint8_t MAX_SHARED_REGIONS =
std::numeric_limits<RegionID>::max() - 1;
static_assert(MAX_SHARED_REGIONS < std::numeric_limits<RegionID>::max(),
"Number of shared memory regions needs to be less than the region id size.");
static constexpr const std::uint8_t MAX_SHM_KEYS = std::numeric_limits<std::uint8_t>::max() - 1;
static constexpr const char *name = "osrm-region";
};
inline std::string regionToString(const SharedDataType region)
{
switch (region)
{
case REGION_1:
return "REGION_1";
case REGION_2:
return "REGION_2";
case REGION_NONE:
return "REGION_NONE";
default:
return "INVALID_REGION";
}
}
private:
std::array<SharedRegion, MAX_SHARED_REGIONS> regions;
std::array<bool, MAX_SHM_KEYS> shm_key_in_use;
};
}
}

18
include/storage/shared_monitor.hpp
View File

@ -47,7 +47,7 @@ template <typename Data> struct SharedMonitor
bi::offset_t size = 0;
if (shmem.get_size(size) && size == 0)
{
shmem.truncate(internal_size + sizeof(Data));
shmem.truncate(rounded_internal_size + sizeof(Data));
region = bi::mapped_region(shmem, bi::read_write);
new (&internal()) InternalData;
new (&data()) Data(initial_data);
@ -65,11 +65,11 @@ template <typename Data> struct SharedMonitor
shmem = bi::shared_memory_object(bi::open_only, Data::name, bi::read_write);
bi::offset_t size = 0;
if (!shmem.get_size(size) || size != internal_size + sizeof(Data))
if (!shmem.get_size(size) || size != rounded_internal_size + sizeof(Data))
{
auto message =
boost::format("Wrong shared memory block '%1%' size %2%, expected %3% bytes") %
(const char *)Data::name % size % (internal_size + sizeof(Data));
(const char *)Data::name % size % (rounded_internal_size + sizeof(Data));
throw util::exception(message.str() + SOURCE_REF);
}
@ -87,7 +87,7 @@ template <typename Data> struct SharedMonitor
Data &data() const
{
auto region_pointer = reinterpret_cast<char *>(region.get_address());
return *reinterpret_cast<Data *>(region_pointer + internal_size);
return *reinterpret_cast<Data *>(region_pointer + rounded_internal_size);
}
mutex_type &get_mutex() const { return internal().mutex; }
@ -120,15 +120,13 @@ template <typename Data> struct SharedMonitor
private:
#if USE_BOOST_INTERPROCESS_CONDITION
static constexpr int internal_size = 128;
struct InternalData
{
mutex_type mutex;
bi::interprocess_condition condition;
};
#else
// Implement a conditional variable using a queue of semaphores.
// OSX checks the virtual address of a mutex in pthread_cond_wait and fails with EINVAL
@ -137,7 +135,6 @@ template <typename Data> struct SharedMonitor
// fail if a waiter is killed.
static constexpr int buffer_size = 256;
static constexpr int internal_size = 4 * 4096;
struct InternalData
{
@ -210,9 +207,8 @@ template <typename Data> struct SharedMonitor
static_assert(buffer_size >= 2, "buffer size is too small");
#endif
static_assert(sizeof(InternalData) + sizeof(Data) <= internal_size, "not enough space");
static_assert(sizeof(InternalData) % alignof(Data) == 0, "incorrect data alignment");
static constexpr int rounded_internal_size = ((sizeof(InternalData) + alignof(Data) - 1) / alignof(Data)) * alignof(Data);
static_assert(rounded_internal_size < sizeof(InternalData) + sizeof(Data), "Data and internal data need to fit into shared memory");
InternalData &internal() const
{

8
src/engine/datafacade/shared_memory_allocator.cpp
View File

@ -13,12 +13,12 @@ namespace engine
namespace datafacade
{
SharedMemoryAllocator::SharedMemoryAllocator(storage::SharedDataType data_region)
SharedMemoryAllocator::SharedMemoryAllocator(storage::SharedRegionRegister::ShmKey data_shm_key)
{
util::Log(logDEBUG) << "Loading new data for region " << regionToString(data_region);
util::Log(logDEBUG) << "Loading new data for region " << data_shm_key;
BOOST_ASSERT(storage::SharedMemory::RegionExists(data_region));
m_large_memory = storage::makeSharedMemory(data_region);
BOOST_ASSERT(storage::SharedMemory::RegionExists(data_shm_key));
m_large_memory = storage::makeSharedMemory(data_shm_key);
storage::io::BufferReader reader(reinterpret_cast<char *>(m_large_memory->Ptr()),
m_large_memory->Size());

6
src/osrm/osrm.cpp
View File

@ -28,12 +28,6 @@ OSRM::OSRM(engine::EngineConfig &config)
throw util::exception("Required files are missing, cannot continue. Have all the "
"pre-processing steps been run?");
}
else if (config.use_shared_memory)
{
storage::SharedMonitor<storage::SharedDataTimestamp> barrier;
using mutex_type = typename decltype(barrier)::mutex_type;
boost::interprocess::scoped_lock<mutex_type> current_region_lock(barrier.get_mutex());
}
// Now, check that the algorithm requested can be used with the data
// that's available.

70
src/storage/storage.cpp
View File

@ -61,7 +61,7 @@ inline void readBlocks(const boost::filesystem::path &path, DataLayout &layout)
}
}
using Monitor = SharedMonitor<SharedDataTimestamp>;
using Monitor = SharedMonitor<SharedRegionRegister>;
Storage::Storage(StorageConfig config_) : config(std::move(config_)) {}
@ -100,25 +100,24 @@ int Storage::Run(int max_wait)
// Get the next region ID and time stamp without locking shared barriers.
// Because of datastore_lock the only write operation can occur sequentially later.
Monitor monitor(SharedDataTimestamp{REGION_NONE, 0});
auto in_use_region = monitor.data().region;
auto next_timestamp = monitor.data().timestamp + 1;
auto next_region =
in_use_region == REGION_2 || in_use_region == REGION_NONE ? REGION_1 : REGION_2;
Monitor monitor(SharedRegionRegister{});
auto &shared_register = monitor.data();
// This is safe because we have an exclusive lock for all osrm-datastore processes.
auto shm_key = shared_register.ReserveKey();
// ensure that the shared memory region we want to write to is really removed
// this is only needef for failure recovery because we actually wait for all clients
// to detach at the end of the function
if (storage::SharedMemory::RegionExists(next_region))
if (storage::SharedMemory::RegionExists(shm_key))
{
util::Log(logWARNING) << "Old shared memory region " << regionToString(next_region)
<< " still exists.";
util::Log(logWARNING) << "Old shared memory region " << shm_key << " still exists.";
util::UnbufferedLog() << "Retrying removal... ";
storage::SharedMemory::Remove(next_region);
storage::SharedMemory::Remove(shm_key);
util::UnbufferedLog() << "ok.";
}
util::Log() << "Loading data into " << regionToString(next_region);
util::Log() << "Loading data into " << shm_key;
// Populate a memory layout into stack memory
DataLayout layout;
@ -132,13 +131,16 @@ int Storage::Run(int max_wait)
auto regions_size = encoded_layout.size() + layout.GetSizeOfLayout();
util::Log() << "Data layout has a size of " << encoded_layout.size() << " bytes";
util::Log() << "Allocating shared memory of " << regions_size << " bytes";
auto data_memory = makeSharedMemory(next_region, regions_size);
auto data_memory = makeSharedMemory(shm_key, regions_size);
// Copy memory layout to shared memory and populate data
char *shared_memory_ptr = static_cast<char *>(data_memory->Ptr());
std::copy_n(encoded_layout.data(), encoded_layout.size(), shared_memory_ptr);
PopulateData(layout, shared_memory_ptr + encoded_layout.size());
std::uint32_t next_timestamp = 0;
std::uint8_t in_use_key = SharedRegionRegister::MAX_SHM_KEYS;
{ // Lock for write access shared region mutex
boost::interprocess::scoped_lock<Monitor::mutex_type> lock(monitor.get_mutex(),
boost::interprocess::defer_lock);
@ -148,13 +150,10 @@ int Storage::Run(int max_wait)
if (!lock.timed_lock(boost::posix_time::microsec_clock::universal_time() +
boost::posix_time::seconds(max_wait)))
{
util::Log(logWARNING)
<< "Could not aquire current region lock after " << max_wait
<< " seconds. Removing locked block and creating a new one. All currently "
"attached processes will not receive notifications and must be restarted";
Monitor::remove();
in_use_region = REGION_NONE;
monitor = Monitor(SharedDataTimestamp{REGION_NONE, 0});
util::Log(logERROR) << "Could not aquire current region lock after " << max_wait
<< " seconds. Data update failed.";
SharedMemory::Remove(shm_key);
return EXIT_FAILURE;
}
}
else
@ -162,32 +161,45 @@ int Storage::Run(int max_wait)
lock.lock();
}
// Update the current region ID and timestamp
monitor.data().region = next_region;
monitor.data().timestamp = next_timestamp;
auto region_id = shared_register.Find("data");
if (region_id == SharedRegionRegister::INVALID_REGION_ID)
{
region_id = shared_register.Register("data", shm_key);
}
else
{
auto& shared_region = shared_register.GetRegion(region_id);
next_timestamp = shared_region.timestamp + 1;
in_use_key = shared_region.shm_key;
shared_region.shm_key = shm_key;
shared_region.timestamp = next_timestamp;
}
}
util::Log() << "All data loaded. Notify all client about new data in "
<< regionToString(next_region) << " with timestamp " << next_timestamp;
util::Log() << "All data loaded. Notify all client about new data in " << shm_key
<< " with timestamp " << next_timestamp;
monitor.notify_all();
// SHMCTL(2): Mark the segment to be destroyed. The segment will actually be destroyed
// only after the last process detaches it.
if (in_use_region != REGION_NONE && storage::SharedMemory::RegionExists(in_use_region))
if (in_use_key != SharedRegionRegister::MAX_SHM_KEYS &&
storage::SharedMemory::RegionExists(in_use_key))
{
util::UnbufferedLog() << "Marking old shared memory region "
<< regionToString(in_use_region) << " for removal... ";
util::UnbufferedLog() << "Marking old shared memory region " << in_use_key
<< " for removal... ";
// aquire a handle for the old shared memory region before we mark it for deletion
// we will need this to wait for all users to detach
auto in_use_shared_memory = makeSharedMemory(in_use_region);
auto in_use_shared_memory = makeSharedMemory(in_use_key);
storage::SharedMemory::Remove(in_use_region);
storage::SharedMemory::Remove(in_use_key);
util::UnbufferedLog() << "ok.";
util::UnbufferedLog() << "Waiting for clients to detach... ";
in_use_shared_memory->WaitForDetach();
util::UnbufferedLog() << " ok.";
shared_register.ReleaseKey(in_use_key);
}
util::Log() << "All clients switched.";

15
src/tools/store.cpp
View File

@ -15,14 +15,13 @@
using namespace osrm;
void removeLocks() { storage::SharedMonitor<storage::SharedDataTimestamp>::remove(); }
void removeLocks() { storage::SharedMonitor<storage::SharedRegionRegister>::remove(); }
void deleteRegion(const storage::SharedDataType region)
void deleteRegion(const storage::SharedRegionRegister::ShmKey key)
{
if (storage::SharedMemory::RegionExists(region) && !storage::SharedMemory::Remove(region))
if (storage::SharedMemory::RegionExists(key) && !storage::SharedMemory::Remove(key))
{
util::Log(logWARNING) << "could not delete shared memory region "
<< storage::regionToString(region);
util::Log(logWARNING) << "could not delete shared memory region " << key;
}
}
@ -43,8 +42,10 @@ void springClean()
}
else
{
deleteRegion(storage::REGION_1);
deleteRegion(storage::REGION_2);
for (auto key : util::irange<std::uint8_t>(0, storage::SharedRegionRegister::MAX_SHM_KEYS))
{
deleteRegion(key);
}
removeLocks();
}
}