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

implements #801

Browse Source
This commit is contained in:
Dennis Luxen 2013-11-27 23:24:00 +01:00
parent cb86d3d829
commit 4911601b24
3 changed files with 133 additions and 117 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
Contractor/Contractor.h
View File

@ -214,14 +214,14 @@ public:
//Create temporary file //Create temporary file
// GetTemporaryFileName(temporaryEdgeStorageFilename); // GetTemporaryFileName(temporaryEdgeStorageFilename);
temporaryStorageSlotID = TemporaryStorage::GetInstance().allocateSlot(); temporaryStorageSlotID = TemporaryStorage::GetInstance().AllocateSlot();
std::cout << "contractor finished initalization" << std::endl; std::cout << "contractor finished initalization" << std::endl;
} }
~Contractor() { ~Contractor() {
//Delete temporary file //Delete temporary file
// remove(temporaryEdgeStorageFilename.c_str()); // remove(temporaryEdgeStorageFilename.c_str());
TemporaryStorage::GetInstance().deallocateSlot(temporaryStorageSlotID); TemporaryStorage::GetInstance().DeallocateSlot(temporaryStorageSlotID);
} }
void Run() { void Run() {
@ -287,9 +287,9 @@ public:
TemporaryStorage & tempStorage = TemporaryStorage::GetInstance(); TemporaryStorage & tempStorage = TemporaryStorage::GetInstance();
//Write dummy number of edges to temporary file //Write dummy number of edges to temporary file
// std::ofstream temporaryEdgeStorage(temporaryEdgeStorageFilename.c_str(), std::ios::binary); // std::ofstream temporaryEdgeStorage(temporaryEdgeStorageFilename.c_str(), std::ios::binary);
uint64_t initialFilePosition = tempStorage.tell(temporaryStorageSlotID); uint64_t initialFilePosition = tempStorage.Tell(temporaryStorageSlotID);
unsigned numberOfTemporaryEdges = 0; unsigned numberOfTemporaryEdges = 0;
tempStorage.writeToSlot(temporaryStorageSlotID, (char*)&numberOfTemporaryEdges, sizeof(unsigned)); tempStorage.WriteToSlot(temporaryStorageSlotID, (char*)&numberOfTemporaryEdges, sizeof(unsigned));
//walk over all nodes //walk over all nodes
for(unsigned i = 0; i < _graph->GetNumberOfNodes(); ++i) { for(unsigned i = 0; i < _graph->GetNumberOfNodes(); ++i) {
@ -299,11 +299,11 @@ public:
const NodeID target = _graph->GetTarget(currentEdge); const NodeID target = _graph->GetTarget(currentEdge);
if(UINT_MAX == newNodeIDFromOldNodeIDMap[i] ){ if(UINT_MAX == newNodeIDFromOldNodeIDMap[i] ){
//Save edges of this node w/o renumbering. //Save edges of this node w/o renumbering.
tempStorage.writeToSlot(temporaryStorageSlotID, (char*)&start, sizeof(NodeID)); tempStorage.WriteToSlot(temporaryStorageSlotID, (char*)&start, sizeof(NodeID));
tempStorage.writeToSlot(temporaryStorageSlotID, (char*)&target, sizeof(NodeID)); tempStorage.WriteToSlot(temporaryStorageSlotID, (char*)&target, sizeof(NodeID));
tempStorage.writeToSlot(temporaryStorageSlotID, (char*)&data, sizeof(_DynamicGraph::EdgeData)); tempStorage.WriteToSlot(temporaryStorageSlotID, (char*)&data, sizeof(_DynamicGraph::EdgeData));
++numberOfTemporaryEdges; ++numberOfTemporaryEdges;
}else { } else {
//node is not yet contracted. //node is not yet contracted.
//add (renumbered) outgoing edges to new DynamicGraph. //add (renumbered) outgoing edges to new DynamicGraph.
_ContractorEdge newEdge; _ContractorEdge newEdge;
@ -324,8 +324,8 @@ public:
} }
} }
//Note the number of temporarily stored edges //Note the number of temporarily stored edges
tempStorage.seek(temporaryStorageSlotID, initialFilePosition); tempStorage.Seek(temporaryStorageSlotID, initialFilePosition);
tempStorage.writeToSlot(temporaryStorageSlotID, (char*)&numberOfTemporaryEdges, sizeof(unsigned)); tempStorage.WriteToSlot(temporaryStorageSlotID, (char*)&numberOfTemporaryEdges, sizeof(unsigned));
//Delete map from old NodeIDs to new ones. //Delete map from old NodeIDs to new ones.
std::vector<NodeID>().swap(newNodeIDFromOldNodeIDMap); std::vector<NodeID>().swap(newNodeIDFromOldNodeIDMap);
@ -498,16 +498,16 @@ public:
TemporaryStorage & tempStorage = TemporaryStorage::GetInstance(); TemporaryStorage & tempStorage = TemporaryStorage::GetInstance();
//Also get the edges from temporary storage //Also get the edges from temporary storage
unsigned numberOfTemporaryEdges = 0; unsigned numberOfTemporaryEdges = 0;
tempStorage.readFromSlot(temporaryStorageSlotID, (char*)&numberOfTemporaryEdges, sizeof(unsigned)); tempStorage.ReadFromSlot(temporaryStorageSlotID, (char*)&numberOfTemporaryEdges, sizeof(unsigned));
//loads edges of graph before renumbering, no need for further numbering action. //loads edges of graph before renumbering, no need for further numbering action.
NodeID start; NodeID start;
NodeID target; NodeID target;
//edges.reserve(edges.size()+numberOfTemporaryEdges); //edges.reserve(edges.size()+numberOfTemporaryEdges);
_DynamicGraph::EdgeData data; _DynamicGraph::EdgeData data;
for(unsigned i = 0; i < numberOfTemporaryEdges; ++i) { for(unsigned i = 0; i < numberOfTemporaryEdges; ++i) {
tempStorage.readFromSlot(temporaryStorageSlotID, (char*)&start, sizeof(NodeID)); tempStorage.ReadFromSlot(temporaryStorageSlotID, (char*)&start, sizeof(NodeID));
tempStorage.readFromSlot(temporaryStorageSlotID, (char*)&target, sizeof(NodeID)); tempStorage.ReadFromSlot(temporaryStorageSlotID, (char*)&target, sizeof(NodeID));
tempStorage.readFromSlot(temporaryStorageSlotID, (char*)&data, sizeof(_DynamicGraph::EdgeData)); tempStorage.ReadFromSlot(temporaryStorageSlotID, (char*)&data, sizeof(_DynamicGraph::EdgeData));
Edge newEdge; Edge newEdge;
newEdge.source = start; newEdge.source = start;
newEdge.target = target; newEdge.target = target;
@ -518,7 +518,7 @@ public:
newEdge.data.backward = data.backward; newEdge.data.backward = data.backward;
edges.push_back( newEdge ); edges.push_back( newEdge );
} }
tempStorage.deallocateSlot(temporaryStorageSlotID); tempStorage.DeallocateSlot(temporaryStorageSlotID);
} }
private: private:

120
Contractor/TemporaryStorage.cpp
View File

@ -28,113 +28,137 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#include "TemporaryStorage.h" #include "TemporaryStorage.h"
TemporaryStorage::TemporaryStorage() { TemporaryStorage::TemporaryStorage() {
tempDirectory = boost::filesystem::temp_directory_path(); temp_directory = boost::filesystem::temp_directory_path();
} }
TemporaryStorage & TemporaryStorage::GetInstance(){ TemporaryStorage & TemporaryStorage::GetInstance(){
static TemporaryStorage runningInstance; static TemporaryStorage static_instance;
return runningInstance; return static_instance;
} }
TemporaryStorage::~TemporaryStorage() { TemporaryStorage::~TemporaryStorage() {
removeAll(); RemoveAll();
mutex.unlock();
} }
void TemporaryStorage::removeAll() { void TemporaryStorage::RemoveAll() {
boost::mutex::scoped_lock lock(mutex); boost::mutex::scoped_lock lock(mutex);
for(unsigned slot_id = 0; slot_id < vectorOfStreamDatas.size(); ++slot_id) { for(unsigned slot_id = 0; slot_id < stream_data_list.size(); ++slot_id) {
deallocateSlot(slot_id); DeallocateSlot(slot_id);
} }
vectorOfStreamDatas.clear(); stream_data_list.clear();
} }
int TemporaryStorage::allocateSlot() { int TemporaryStorage::AllocateSlot() {
boost::mutex::scoped_lock lock(mutex); boost::mutex::scoped_lock lock(mutex);
try { try {
vectorOfStreamDatas.push_back(StreamData()); stream_data_list.push_back(StreamData());
//SimpleLogger().Write() << "created new temporary file: " << vectorOfStreamDatas.back().pathToTemporaryFile;
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
return vectorOfStreamDatas.size() - 1; CheckIfTemporaryDeviceFull();
return stream_data_list.size() - 1;
} }
void TemporaryStorage::deallocateSlot(int slotID) { void TemporaryStorage::DeallocateSlot(const int slot_id) {
try { try {
StreamData & data = vectorOfStreamDatas[slotID]; StreamData & data = stream_data_list[slot_id];
boost::mutex::scoped_lock lock(*data.readWriteMutex); boost::mutex::scoped_lock lock(*data.readWriteMutex);
if(!boost::filesystem::exists(data.pathToTemporaryFile)) { if(!boost::filesystem::exists(data.temp_path)) {
return; return;
} }
if(data.streamToTemporaryFile->is_open()) { if(data.temp_file->is_open()) {
data.streamToTemporaryFile->close(); data.temp_file->close();
} }
boost::filesystem::remove(data.pathToTemporaryFile); boost::filesystem::remove(data.temp_path);
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
} }
void TemporaryStorage::writeToSlot(int slotID, char * pointer, std::streamsize size) { void TemporaryStorage::WriteToSlot(
const int slot_id,
char * pointer,
std::streamsize size
) {
try { try {
StreamData & data = vectorOfStreamDatas[slotID]; StreamData & data = stream_data_list[slot_id];
boost::mutex::scoped_lock lock(*data.readWriteMutex); boost::mutex::scoped_lock lock(*data.readWriteMutex);
BOOST_ASSERT_MSG( BOOST_ASSERT_MSG(
data.writeMode, data.write_mode,
"Writing after first read is not allowed" "Writing after first read is not allowed"
); );
data.streamToTemporaryFile->write(pointer, size); data.temp_file->write(pointer, size);
CheckIfTemporaryDeviceFull();
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
} }
void TemporaryStorage::readFromSlot(int slotID, char * pointer, std::streamsize size) { void TemporaryStorage::ReadFromSlot(
const int slot_id,
char * pointer,
std::streamsize size
) {
try { try {
StreamData & data = vectorOfStreamDatas[slotID]; StreamData & data = stream_data_list[slot_id];
boost::mutex::scoped_lock lock(*data.readWriteMutex); boost::mutex::scoped_lock lock(*data.readWriteMutex);
if(data.writeMode) { if(data.write_mode) {
data.writeMode = false; data.write_mode = false;
data.streamToTemporaryFile->seekg(0, data.streamToTemporaryFile->beg); data.temp_file->seekg(0, data.temp_file->beg);
} }
data.streamToTemporaryFile->read(pointer, size); data.temp_file->read(pointer, size);
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
} }
unsigned TemporaryStorage::getFreeBytesOnTemporaryDevice() { uint64_t TemporaryStorage::GetFreeBytesOnTemporaryDevice() {
boost::filesystem::space_info tempSpaceInfo; uint64_t value = -1;
try { try {
tempSpaceInfo = boost::filesystem::space(tempDirectory); boost::filesystem::path p = boost::filesystem::temp_directory_path();
boost::filesystem::space_info s = boost::filesystem::space( p );
value = s.free;
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
return tempSpaceInfo.available; return value;
} }
boost::filesystem::fstream::pos_type TemporaryStorage::tell(int slotID) { void TemporaryStorage::CheckIfTemporaryDeviceFull() {
boost::filesystem::path p = boost::filesystem::temp_directory_path();
boost::filesystem::space_info s = boost::filesystem::space( p );
if(1024*1024 > s.free) {
throw OSRMException("temporary device is full");
}
}
boost::filesystem::fstream::pos_type TemporaryStorage::Tell(int slot_id) {
boost::filesystem::fstream::pos_type position; boost::filesystem::fstream::pos_type position;
try { try {
StreamData & data = vectorOfStreamDatas[slotID]; StreamData & data = stream_data_list[slot_id];
boost::mutex::scoped_lock lock(*data.readWriteMutex); boost::mutex::scoped_lock lock(*data.readWriteMutex);
position = data.streamToTemporaryFile->tellp(); position = data.temp_file->tellp();
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
return position; return position;
} }
void TemporaryStorage::abort(boost::filesystem::filesystem_error& ) { void TemporaryStorage::Abort(boost::filesystem::filesystem_error& e) {
removeAll(); RemoveAll();
throw OSRMException(e.what());
} }
void TemporaryStorage::seek(int slotID, boost::filesystem::fstream::pos_type position) { void TemporaryStorage::Seek(
const int slot_id,
const boost::filesystem::fstream::pos_type position
) {
try { try {
StreamData & data = vectorOfStreamDatas[slotID]; StreamData & data = stream_data_list[slot_id];
boost::mutex::scoped_lock lock(*data.readWriteMutex); boost::mutex::scoped_lock lock(*data.readWriteMutex);
data.streamToTemporaryFile->seekg(position); data.temp_file->seekg(position);
} catch(boost::filesystem::filesystem_error & e) { } catch(boost::filesystem::filesystem_error & e) {
abort(e); Abort(e);
} }
} }

100
Contractor/TemporaryStorage.h
View File

@ -28,46 +28,23 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#ifndef TEMPORARYSTORAGE_H_ #ifndef TEMPORARYSTORAGE_H_
#define TEMPORARYSTORAGE_H_ #define TEMPORARYSTORAGE_H_
#include <vector> #include "../Util/BoostFileSystemFix.h"
#include <fstream>
#include <boost/assert.hpp>
#include <boost/foreach.hpp>
#include <boost/filesystem.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include "../Util/OSRMException.h" #include "../Util/OSRMException.h"
#include "../Util/SimpleLogger.h" #include "../Util/SimpleLogger.h"
#include "../typedefs.h" #include "../typedefs.h"
//This is one big workaround for latest boost renaming woes. #include <boost/assert.hpp>
#include <boost/foreach.hpp>
#include <boost/integer.hpp>
#include <boost/filesystem.hpp>
#include <boost/filesystem/fstream.hpp>
#include <boost/make_shared.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread/mutex.hpp>
#if BOOST_FILESYSTEM_VERSION < 3 #include <vector>
#warning Boost Installation with Filesystem3 missing, activating workaround #include <fstream>
#include <cstdio>
namespace boost {
namespace filesystem {
inline path temp_directory_path() {
char * buffer;
buffer = tmpnam (NULL);
return path(buffer);
}
inline path unique_path(const path&) {
return temp_directory_path();
}
}
}
#endif
#ifndef BOOST_FILESYSTEM_VERSION
#define BOOST_FILESYSTEM_VERSION 3
#endif
/** /**
* This class implements a singleton file storage for temporary data. * This class implements a singleton file storage for temporary data.
* temporary slots can be accessed by other objects through an int * temporary slots can be accessed by other objects through an int
@ -77,48 +54,63 @@ inline path unique_path(const path&) {
* -> Data is written in first phase and reread in second. * -> Data is written in first phase and reread in second.
*/ */
static boost::filesystem::path tempDirectory; static boost::filesystem::path temp_directory;
static std::string TemporaryFilePattern("OSRM-%%%%-%%%%-%%%%"); static std::string TemporaryFilePattern("OSRM-%%%%-%%%%-%%%%");
class TemporaryStorage { class TemporaryStorage {
public: public:
static TemporaryStorage & GetInstance(); static TemporaryStorage & GetInstance();
virtual ~TemporaryStorage(); virtual ~TemporaryStorage();
int allocateSlot(); int AllocateSlot();
void deallocateSlot(int slotID); void DeallocateSlot(const int slot_id);
void writeToSlot(int slotID, char * pointer, std::streamsize size); void WriteToSlot(const int slot_id, char * pointer, std::streamsize size);
void readFromSlot(int slotID, char * pointer, std::streamsize size); void ReadFromSlot(const int slot_id, char * pointer, std::streamsize size);
//returns the number of free bytes //returns the number of free bytes
unsigned getFreeBytesOnTemporaryDevice(); uint64_t GetFreeBytesOnTemporaryDevice();
boost::filesystem::fstream::pos_type tell(int slotID); boost::filesystem::fstream::pos_type Tell(int slot_id);
void seek(int slotID, boost::filesystem::fstream::pos_type); void Seek(const int slot_id, boost::filesystem::fstream::pos_type);
void removeAll(); void RemoveAll();
private: private:
TemporaryStorage(); TemporaryStorage();
TemporaryStorage(TemporaryStorage const &){}; TemporaryStorage(TemporaryStorage const &){};
TemporaryStorage& operator=(TemporaryStorage const &) {
TemporaryStorage & operator=(TemporaryStorage const &) {
return *this; return *this;
} }
void abort(boost::filesystem::filesystem_error& e);
void Abort(boost::filesystem::filesystem_error& e);
void CheckIfTemporaryDeviceFull();
struct StreamData { struct StreamData {
bool writeMode; bool write_mode;
boost::filesystem::path pathToTemporaryFile; boost::filesystem::path temp_path;
boost::shared_ptr<boost::filesystem::fstream> streamToTemporaryFile; boost::shared_ptr<boost::filesystem::fstream> temp_file;
boost::shared_ptr<boost::mutex> readWriteMutex; boost::shared_ptr<boost::mutex> readWriteMutex;
StreamData() : StreamData() :
writeMode(true), write_mode(true),
pathToTemporaryFile (boost::filesystem::unique_path(tempDirectory.append(TemporaryFilePattern.begin(), TemporaryFilePattern.end()))), temp_path(
streamToTemporaryFile(new boost::filesystem::fstream(pathToTemporaryFile, std::ios::in | std::ios::out | std::ios::trunc | std::ios::binary)), boost::filesystem::unique_path(
readWriteMutex(new boost::mutex) temp_directory.append(
TemporaryFilePattern.begin(),
TemporaryFilePattern.end()
)
)
),
temp_file(
new boost::filesystem::fstream(
temp_path,
std::ios::in|std::ios::out|std::ios::trunc|std::ios::binary
)
),
readWriteMutex(boost::make_shared<boost::mutex>())
{ {
if(streamToTemporaryFile->fail()) { if(temp_file->fail()) {
throw OSRMException("temporary file could not be created"); throw OSRMException("temporary file could not be created");
} }
} }
}; };
//vector of file streams that is used to store temporary data //vector of file streams that is used to store temporary data
std::vector<StreamData> vectorOfStreamDatas; std::vector<StreamData> stream_data_list;
boost::mutex mutex; boost::mutex mutex;
}; };