Use a shared (!) reader writer lock to protect CURRENT_REGIONS
This fixes issue #3016.
This commit is contained in:
committed by
Patrick Niklaus
parent
036475afd0
commit
9eb7fc03ce
+9
-57
@@ -1,5 +1,5 @@
|
||||
#include "engine/engine.hpp"
|
||||
#include "engine/api/route_parameters.hpp"
|
||||
#include "engine/engine.hpp"
|
||||
#include "engine/engine_config.hpp"
|
||||
#include "engine/status.hpp"
|
||||
|
||||
@@ -20,6 +20,7 @@
|
||||
#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 <boost/thread/lock_types.hpp>
|
||||
|
||||
#include <algorithm>
|
||||
@@ -27,65 +28,13 @@
|
||||
#include <utility>
|
||||
#include <vector>
|
||||
|
||||
namespace osrm
|
||||
{
|
||||
namespace engine
|
||||
{
|
||||
struct Engine::EngineLock
|
||||
{
|
||||
// will only be initialized if shared memory is used
|
||||
storage::SharedBarriers barrier;
|
||||
// decrease number of concurrent queries
|
||||
void DecreaseQueryCount();
|
||||
// increase number of concurrent queries
|
||||
void IncreaseQueryCount();
|
||||
};
|
||||
|
||||
// decrease number of concurrent queries
|
||||
void Engine::EngineLock::DecreaseQueryCount()
|
||||
{
|
||||
// lock query
|
||||
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> query_lock(
|
||||
barrier.query_mutex);
|
||||
|
||||
// decrement query count
|
||||
--(barrier.number_of_queries);
|
||||
BOOST_ASSERT_MSG(0 <= barrier.number_of_queries, "invalid number of queries");
|
||||
|
||||
// notify all processes that were waiting for this condition
|
||||
if (0 == barrier.number_of_queries)
|
||||
{
|
||||
barrier.no_running_queries_condition.notify_all();
|
||||
}
|
||||
}
|
||||
|
||||
// increase number of concurrent queries
|
||||
void Engine::EngineLock::IncreaseQueryCount()
|
||||
{
|
||||
// lock update pending
|
||||
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> pending_lock(
|
||||
barrier.pending_update_mutex);
|
||||
|
||||
// lock query
|
||||
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> query_lock(
|
||||
barrier.query_mutex);
|
||||
|
||||
// unlock update pending
|
||||
pending_lock.unlock();
|
||||
|
||||
// increment query count
|
||||
++(barrier.number_of_queries);
|
||||
}
|
||||
} // ns engine
|
||||
} // ns osrm
|
||||
|
||||
namespace
|
||||
{
|
||||
// Abstracted away the query locking into a template function
|
||||
// Works the same for every plugin.
|
||||
template <typename ParameterT, typename PluginT, typename ResultT>
|
||||
osrm::engine::Status
|
||||
RunQuery(const std::unique_ptr<osrm::engine::Engine::EngineLock> &lock,
|
||||
RunQuery(const std::unique_ptr<osrm::storage::SharedBarriers> &lock,
|
||||
const std::shared_ptr<osrm::engine::datafacade::BaseDataFacade> &facade,
|
||||
const ParameterT ¶meters,
|
||||
PluginT &plugin,
|
||||
@@ -97,7 +46,10 @@ RunQuery(const std::unique_ptr<osrm::engine::Engine::EngineLock> &lock,
|
||||
}
|
||||
|
||||
BOOST_ASSERT(lock);
|
||||
lock->IncreaseQueryCount();
|
||||
// this locks aquires shared ownership of the query mutex: other requets are allowed
|
||||
// to run, but data updates need to wait for all queries to finish until they can aquire an exclusive lock
|
||||
boost::interprocess::sharable_lock<boost::interprocess::named_sharable_mutex> query_lock(
|
||||
lock->query_mutex);
|
||||
|
||||
auto &shared_facade = static_cast<osrm::engine::datafacade::SharedDataFacade &>(*facade);
|
||||
shared_facade.CheckAndReloadFacade();
|
||||
@@ -107,7 +59,6 @@ RunQuery(const std::unique_ptr<osrm::engine::Engine::EngineLock> &lock,
|
||||
|
||||
osrm::engine::Status status = plugin.HandleRequest(facade, parameters, result);
|
||||
|
||||
lock->DecreaseQueryCount();
|
||||
return status;
|
||||
}
|
||||
|
||||
@@ -119,10 +70,11 @@ namespace engine
|
||||
{
|
||||
|
||||
Engine::Engine(const EngineConfig &config)
|
||||
: lock(config.use_shared_memory ? std::make_unique<storage::SharedBarriers>()
|
||||
: std::unique_ptr<storage::SharedBarriers>())
|
||||
{
|
||||
if (config.use_shared_memory)
|
||||
{
|
||||
lock = std::make_unique<EngineLock>();
|
||||
query_data_facade = std::make_shared<datafacade::SharedDataFacade>();
|
||||
}
|
||||
else
|
||||
|
||||
+12
-17
@@ -42,11 +42,6 @@ namespace osrm
|
||||
namespace storage
|
||||
{
|
||||
|
||||
using RTreeLeaf = engine::datafacade::BaseDataFacade::RTreeLeaf;
|
||||
using RTreeNode =
|
||||
util::StaticRTree<RTreeLeaf, util::ShM<util::Coordinate, true>::vector, true>::TreeNode;
|
||||
using QueryGraph = util::StaticGraph<contractor::QueryEdge::EdgeData>;
|
||||
|
||||
// delete a shared memory region. report warning if it could not be deleted
|
||||
void deleteRegion(const SharedDataType region)
|
||||
{
|
||||
@@ -76,6 +71,11 @@ void deleteRegion(const SharedDataType region)
|
||||
}
|
||||
}
|
||||
|
||||
using RTreeLeaf = engine::datafacade::BaseDataFacade::RTreeLeaf;
|
||||
using RTreeNode =
|
||||
util::StaticRTree<RTreeLeaf, util::ShM<util::Coordinate, true>::vector, true>::TreeNode;
|
||||
using QueryGraph = util::StaticGraph<contractor::QueryEdge::EdgeData>;
|
||||
|
||||
Storage::Storage(StorageConfig config_) : config(std::move(config_)) {}
|
||||
|
||||
int Storage::Run()
|
||||
@@ -738,20 +738,15 @@ int Storage::Run()
|
||||
SharedDataTimestamp *data_timestamp_ptr =
|
||||
static_cast<SharedDataTimestamp *>(data_type_memory->Ptr());
|
||||
|
||||
boost::interprocess::scoped_lock<boost::interprocess::named_mutex> query_lock(
|
||||
barrier.query_mutex);
|
||||
|
||||
// notify all processes that were waiting for this condition
|
||||
if (0 < barrier.number_of_queries)
|
||||
{
|
||||
barrier.no_running_queries_condition.wait(query_lock);
|
||||
}
|
||||
boost::interprocess::scoped_lock<boost::interprocess::named_sharable_mutex> query_lock(barrier.query_mutex);
|
||||
|
||||
data_timestamp_ptr->layout = layout_region;
|
||||
data_timestamp_ptr->data = data_region;
|
||||
data_timestamp_ptr->timestamp += 1;
|
||||
deleteRegion(previous_data_region);
|
||||
deleteRegion(previous_layout_region);
|
||||
data_timestamp_ptr->layout = layout_region;
|
||||
data_timestamp_ptr->data = data_region;
|
||||
data_timestamp_ptr->timestamp += 1;
|
||||
deleteRegion(previous_data_region);
|
||||
deleteRegion(previous_layout_region);
|
||||
}
|
||||
util::SimpleLogger().Write() << "all data loaded";
|
||||
|
||||
return EXIT_SUCCESS;
|
||||
|
||||
@@ -10,6 +10,5 @@ int main()
|
||||
osrm::storage::SharedBarriers barrier;
|
||||
barrier.pending_update_mutex.unlock();
|
||||
barrier.query_mutex.unlock();
|
||||
barrier.update_mutex.unlock();
|
||||
return 0;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user