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

Use shared memory region mapping at fixed address for OSX

Browse Source
This commit is contained in:
Michael Krasnyk 2017-01-27 10:05:34 +01:00 committed by Patrick Niklaus
parent 36c944ef53
commit c22ce3ae1f
9 changed files with 184 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
include/engine/data_watchdog.hpp
View File

@ -4,7 +4,7 @@
#include "engine/datafacade/contiguous_internalmem_datafacade.hpp"
#include "engine/datafacade/shared_memory_allocator.hpp"
#include "storage/shared_barriers.hpp"
#include "storage/shared_barrier.hpp"
#include "storage/shared_datatype.hpp"
#include "storage/shared_memory.hpp"
@ -27,19 +27,16 @@ namespace engine
class DataWatchdog
{
public:
DataWatchdog() : active(true), timestamp(0)
DataWatchdog() : barrier(boost::interprocess::open_only), active(true), timestamp(0)
{
// create the initial facade before launching the watchdog thread
{
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> current_region_lock(
barrier.region_mutex);
auto shared_memory = makeSharedMemory(storage::CURRENT_REGION);
auto current = static_cast<storage::SharedDataTimestamp *>(shared_memory->Ptr());
boost::interprocess::scoped_lock<storage::SharedBarrier::mutex_type>
current_region_lock(barrier.GetMutex());
facade = std::make_shared<datafacade::ContiguousInternalMemoryDataFacade>(
std::make_unique<datafacade::SharedMemoryAllocator>(current->region));
timestamp = current->timestamp;
std::make_unique<datafacade::SharedMemoryAllocator>(barrier.GetRegion()));
timestamp = barrier.GetTimestamp();
}
watcher = std::thread(&DataWatchdog::Run, this);
@ -48,49 +45,40 @@ class DataWatchdog
~DataWatchdog()
{
active = false;
barrier.region_condition.notify_all();
barrier.NotifyAll();
watcher.join();
}
// Tries to connect to the shared memory containing the regions table
static bool TryConnect()
{
return storage::SharedMemory::RegionExists(storage::CURRENT_REGION);
}
auto GetDataFacade() const { return facade; }
private:
void Run()
{
auto shared_memory = makeSharedMemory(storage::CURRENT_REGION);
auto current = static_cast<storage::SharedDataTimestamp *>(shared_memory->Ptr());
while (active)
{
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> current_region_lock(
barrier.region_mutex);
boost::interprocess::scoped_lock<storage::SharedBarrier::mutex_type>
current_region_lock(barrier.GetMutex());
while (active && timestamp == current->timestamp)
while (active && timestamp == barrier.GetTimestamp())
{
barrier.region_condition.wait(current_region_lock);
barrier.Wait(current_region_lock);
}
if (timestamp != current->timestamp)
if (timestamp != barrier.GetTimestamp())
{
facade = std::make_shared<datafacade::ContiguousInternalMemoryDataFacade>(
std::make_unique<datafacade::SharedMemoryAllocator>(current->region));
timestamp = current->timestamp;
std::make_unique<datafacade::SharedMemoryAllocator>(barrier.GetRegion()));
timestamp = barrier.GetTimestamp();
util::Log() << "updated facade to region "
<< storage::regionToString(current->region) << " with timestamp "
<< current->timestamp;
<< storage::regionToString(barrier.GetRegion()) << " with timestamp "
<< timestamp;
}
}
util::Log() << "DataWatchdog thread stopped";
}
storage::SharedBarriers barrier;
storage::SharedBarrier barrier;
std::thread watcher;
bool active;
unsigned timestamp;

3
include/engine/datafacade/contiguous_internalmem_datafacade.hpp
View File

@ -26,9 +26,6 @@
#include "util/typedefs.hpp"
#include <boost/assert.hpp>
#include <boost/interprocess/sync/named_sharable_mutex.hpp>
#include <boost/interprocess/sync/sharable_lock.hpp>
#include <boost/thread/tss.hpp>
#include <algorithm>
#include <cstddef>

131
include/storage/shared_barrier.hpp Normal file
View File

@ -0,0 +1,131 @@
#ifndef SHARED_BARRIER_HPP
#define SHARED_BARRIER_HPP
#include "storage/shared_datatype.hpp"
#include <boost/format.hpp>
#include <boost/interprocess/mapped_region.hpp>
#include <boost/interprocess/shared_memory_object.hpp>
#include <boost/interprocess/sync/interprocess_condition.hpp>
#include <boost/interprocess/sync/interprocess_mutex.hpp>
#include <iostream>
namespace osrm
{
namespace storage
{
namespace
{
namespace bi = boost::interprocess;
}
struct SharedBarrier
{
struct internal_data
{
typedef bi::interprocess_mutex mutex_type;
typedef bi::interprocess_condition condvar_type;
internal_data() : region(REGION_NONE), timestamp(0){};
mutex_type mutex;
condvar_type condvar;
SharedDataType region;
unsigned timestamp;
};
typedef internal_data::mutex_type mutex_type;
typedef internal_data::condvar_type condvar_type;
SharedBarrier(bi::open_or_create_t)
{
shmem = bi::shared_memory_object(bi::open_or_create, block_name, bi::read_write);
const void *address = reinterpret_cast<void *>(block_address);
bi::offset_t size = 0;
if (!shmem.get_size(size) || size != block_size)
{
shmem.truncate(block_size);
region = bi::mapped_region(shmem, bi::read_write, 0, block_size, address, block_flags);
new (region.get_address()) internal_data;
}
else
{
region = bi::mapped_region(shmem, bi::read_write, 0, block_size, address, block_flags);
}
}
SharedBarrier(bi::open_only_t)
{
try
{
shmem = bi::shared_memory_object(bi::open_only, block_name, bi::read_write);
bi::offset_t size = 0;
if (!shmem.get_size(size) || size != block_size)
{
auto message =
boost::format("Wrong shared memory block size %1%, expected %2% bytes") % size %
static_cast<const std::size_t>(block_size);
throw util::exception(message.str() + SOURCE_REF);
}
}
catch (const bi::interprocess_exception &exception)
{
throw util::exception(
std::string(
"No shared memory blocks found, have you forgotten to run osrm-datastore?") +
SOURCE_REF);
}
const void *address = reinterpret_cast<void *>(block_address);
region = bi::mapped_region(shmem, bi::read_write, 0, block_size, address, block_flags);
}
auto &GetMutex() const { return internal()->mutex; }
template <typename L> void Wait(L &lock) const { internal()->condvar.wait(lock); }
void NotifyAll() const { internal()->condvar.notify_all(); }
void SetRegion(SharedDataType region) const
{
internal()->region = region;
internal()->timestamp += 1;
}
auto GetRegion() const { return internal()->region; }
auto GetTimestamp() const { return internal()->timestamp; }
static void Remove() { bi::shared_memory_object::remove(block_name); }
private:
bi::shared_memory_object shmem;
bi::mapped_region region;
static constexpr const char *const block_name = "osrm-region";
static constexpr std::size_t block_size = sizeof(internal_data);
#if defined(__APPLE__)
// OSX pthread_cond_wait implementation internally checks address of the mutex
// so a fixed mapping address is needed to share control memory block between
// processes and engine instances in one process
// https://github.com/Project-OSRM/osrm-backend/issues/3619
static constexpr std::intptr_t block_address = 0x00000caffee00000;
static constexpr bi::map_options_t block_flags = MAP_FIXED;
#else
static constexpr std::intptr_t block_address = 0x0;
static constexpr bi::map_options_t block_flags = bi::default_map_options;
#endif
inline internal_data *internal() const
{
return reinterpret_cast<internal_data *>(region.get_address());
}
};
}
}
#endif // SHARED_BARRIER_HPP

40
include/storage/shared_barriers.hpp
View File

@ -1,40 +0,0 @@
#ifndef SHARED_BARRIERS_HPP
#define SHARED_BARRIERS_HPP
#include <boost/interprocess/sync/named_condition.hpp>
#include <boost/interprocess/sync/named_mutex.hpp>
#include "util/retry_lock.hpp"
namespace osrm
{
namespace storage
{
struct SharedBarriers
{
SharedBarriers()
: region_mutex(boost::interprocess::open_or_create, "osrm-region"),
region_condition(boost::interprocess::open_or_create, "osrm-region-cv")
{
}
static void remove()
{
boost::interprocess::named_mutex::remove("osrm-region");
boost::interprocess::named_condition::remove("osrm-region-cv");
}
static RetryLock getLockWithRetry(int timeout_seconds)
{
return RetryLock(timeout_seconds, "osrm-region");
}
boost::interprocess::named_mutex region_mutex;
boost::interprocess::named_condition region_condition;
};
}
}
#endif // SHARED_BARRIERS_HPP

7
include/storage/shared_datatype.hpp
View File

@ -204,10 +204,9 @@ struct DataLayout
enum SharedDataType
{
CURRENT_REGION,
REGION_NONE,
REGION_1,
REGION_2,
REGION_NONE
REGION_2
};
struct SharedDataTimestamp
@ -220,8 +219,6 @@ inline std::string regionToString(const SharedDataType region)
{
switch (region)
{
case CURRENT_REGION:
return "CURRENT_REGION";
case REGION_1:
return "REGION_1";
case REGION_2:

49
include/util/retry_lock.hpp
View File

@ -1,49 +0,0 @@
#ifndef OSRM_RETRY_TIMED_LOCK_HPP
#define OSRM_RETRY_TIMED_LOCK_HPP
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/interprocess/exceptions.hpp>
#include <boost/interprocess/sync/named_mutex.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
class RetryLock
{
public:
RetryLock(int timeout_seconds, const char *name)
: timeout_seconds(timeout_seconds), name(name),
mutex(std::make_unique<boost::interprocess::named_mutex>(
boost::interprocess::open_or_create, name)),
internal_lock(*mutex, boost::interprocess::defer_lock)
{
}
bool TryLock()
{
if (timeout_seconds >= 0)
{
return internal_lock.timed_lock(boost::posix_time::microsec_clock::universal_time() +
boost::posix_time::seconds(timeout_seconds));
}
else
{
internal_lock.lock();
return true;
}
}
void ForceLock()
{
mutex.reset();
boost::interprocess::named_mutex::remove(name);
mutex = std::make_unique<boost::interprocess::named_mutex>(
boost::interprocess::open_or_create, name);
}
private:
int timeout_seconds;
const char *name;
std::unique_ptr<boost::interprocess::named_mutex> mutex;
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> internal_lock;
};
#endif

12
src/engine/engine.cpp
View File

@ -6,13 +6,9 @@
#include "engine/datafacade/contiguous_internalmem_datafacade.hpp"
#include "engine/datafacade/process_memory_allocator.hpp"
#include "storage/shared_barriers.hpp"
#include "util/log.hpp"
#include <boost/assert.hpp>
#include <boost/interprocess/sync/named_condition.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
#include <boost/interprocess/sync/sharable_lock.hpp>
#include <algorithm>
#include <fstream>
@ -63,14 +59,6 @@ Engine::Engine(const EngineConfig &config)
{
if (config.use_shared_memory)
{
if (!DataWatchdog::TryConnect())
{
throw util::exception(
std::string(
"No shared memory blocks found, have you forgotten to run osrm-datastore?") +
SOURCE_REF);
}
watchdog = std::make_unique<DataWatchdog>();
BOOST_ASSERT(watchdog);
}

63
src/storage/storage.cpp
View File

@ -8,7 +8,7 @@
#include "extractor/travel_mode.hpp"
#include "storage/io.hpp"
#include "storage/serialization.hpp"
#include "storage/shared_barriers.hpp"
#include "storage/shared_barrier.hpp"
#include "storage/shared_datatype.hpp"
#include "storage/shared_memory.hpp"
#include "engine/datafacade/datafacade_base.hpp"
@ -29,9 +29,11 @@
#include <sys/mman.h>
#endif
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/filesystem/path.hpp>
#include <boost/interprocess/sync/file_lock.hpp>
#include <boost/interprocess/sync/scoped_lock.hpp>
#include <cstdint>
@ -88,19 +90,10 @@ 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.
auto next_region = REGION_1;
unsigned next_timestamp = 1;
auto in_use_region = REGION_NONE;
unsigned in_use_timestamp = 0;
if (SharedMemory::RegionExists(CURRENT_REGION))
{
auto shared_memory = makeSharedMemory(CURRENT_REGION);
auto current_region = static_cast<SharedDataTimestamp *>(shared_memory->Ptr());
in_use_region = current_region->region;
in_use_timestamp = current_region->timestamp;
next_region = in_use_region == REGION_1 ? REGION_2 : REGION_1;
next_timestamp = in_use_timestamp + 1;
}
SharedBarrier barrier(boost::interprocess::open_or_create);
auto in_use_region = barrier.GetRegion();
auto next_region =
in_use_region == REGION_2 || in_use_region == REGION_NONE ? REGION_1 : REGION_2;
// 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
@ -114,8 +107,7 @@ int Storage::Run(int max_wait)
util::UnbufferedLog() << "ok.";
}
util::Log() << "Loading data into " << regionToString(next_region) << " timestamp "
<< next_timestamp;
util::Log() << "Loading data into " << regionToString(next_region);
// Populate a memory layout into stack memory
DataLayout layout;
@ -131,28 +123,37 @@ int Storage::Run(int max_wait)
memcpy(shared_memory_ptr, &layout, sizeof(layout));
PopulateData(layout, shared_memory_ptr + sizeof(layout));
SharedBarriers barriers;
{ // Lock for write access shared region mutex that protects CURRENT_REGION
RetryLock current_region_lock = barriers.getLockWithRetry(max_wait);
{ // Lock for write access shared region mutex
boost::interprocess::scoped_lock<SharedBarrier::mutex_type> lock(
barrier.GetMutex(), boost::interprocess::defer_lock);
if (!current_region_lock.TryLock())
if (max_wait >= 0)
{
util::Log(logWARNING) << "Could not aquire current region lock after " << max_wait
<< " seconds. Claiming the lock by force.";
current_region_lock.ForceLock();
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";
SharedBarrier::Remove();
in_use_region = REGION_NONE;
barrier = SharedBarrier(boost::interprocess::open_or_create);
}
}
else
{
lock.lock();
}
// Get write access to CURRENT_REGION
auto shared_memory = makeSharedMemory(CURRENT_REGION, sizeof(SharedDataTimestamp));
auto current_region = static_cast<SharedDataTimestamp *>(shared_memory->Ptr());
// Update the current region ID and timestamp
current_region->region = next_region;
current_region->timestamp = next_timestamp;
barrier.SetRegion(next_region);
}
util::Log() << "All data loaded. Notify all client... ";
barriers.region_condition.notify_all();
util::Log() << "All data loaded. Notify all client about new data in "
<< regionToString(barrier.GetRegion()) << " with timestamp "
<< barrier.GetTimestamp();
barrier.NotifyAll();
// SHMCTL(2): Mark the segment to be destroyed. The segment will actually be destroyed
// only after the last process detaches it.

5
src/tools/store.cpp
View File

@ -1,4 +1,4 @@
#include "storage/shared_barriers.hpp"
#include "storage/shared_barrier.hpp"
#include "storage/shared_memory.hpp"
#include "storage/storage.hpp"
#include "util/exception.hpp"
@ -14,7 +14,7 @@
using namespace osrm;
void removeLocks() { osrm::storage::SharedBarriers::remove(); }
void removeLocks() { osrm::storage::SharedBarrier::Remove(); }
void deleteRegion(const storage::SharedDataType region)
{
@ -44,7 +44,6 @@ void springClean()
{
deleteRegion(storage::REGION_1);
deleteRegion(storage::REGION_2);
deleteRegion(storage::CURRENT_REGION);
removeLocks();
}
}