Merge branch 'master' of https://github.com/DennisOSRM/Project-OSRM

Contractor/Contractor.h

@@ -35,7 +35,9 @@ or see http://www.gnu.org/licenses/agpl.txt.
 #include <set>
 #include <stack>
 #include <limits>
+#ifdef _OPEMP
 #include <omp.h>
+#endif
 
 class Contractor {
 

DataStructures/BaseParser.h

@@ -28,7 +28,6 @@ public:
     virtual bool Init() = 0;
     virtual bool RegisterCallbacks(bool (*nodeCallbackPointer)(NodeT), bool (*restrictionCallbackPointer)(RestrictionT), bool (*wayCallbackPointer)(WayT), bool (*addressCallbackPointer)(NodeT, HashTable<std::string, std::string>)) = 0;
     virtual bool Parse() = 0;
-private:
 };
 
 #endif /* BASEPARSER_H_ */

DataStructures/ConcurrentQueue.h

@@ -0,0 +1,103 @@
+/*
+    open source routing machine
+    Copyright (C) Dennis Luxen, others 2010
+
+This program is free software; you can redistribute it and/or modify
+it under the terms of the GNU AFFERO General Public License as published by
+the Free Software Foundation; either version 3 of the License, or
+any later version.
+
+This program is distributed in the hope that it will be useful,
+but WITHOUT ANY WARRANTY; without even the implied warranty of
+MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+GNU General Public License for more details.
+
+You should have received a copy of the GNU Affero General Public License
+along with this program; if not, write to the Free Software
+Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+or see http://www.gnu.org/licenses/agpl.txt.
+ */
+
+#ifndef CONCURRENTQUEUE_H_INCLUDED
+#define CONCURRENTQUEUE_H_INCLUDED
+
+#include "typedefs.h"
+
+/* 
+   Concurrent Queue written by Anthony Williams: 
+   http://www.justsoftwaresolutions.co.uk/threading/implementing-a-thread-safe-queue-using-condition-variables.html  
+*/
+template<typename Data>
+class concurrent_queue
+{
+private:
+    std::queue<Data> internal_queue;
+    mutable boost::mutex queue_mutex;
+    mutable boost::mutex queue_full_mutex;
+    boost::condition_variable queue_cv;
+    boost::condition_variable queue_full_cv;
+    const size_t max_queue_size;
+
+    bool size_exceeded() const {
+        boost::mutex::scoped_lock lock(queue_mutex);
+        return internal_queue.size() >= max_queue_size;
+    }
+
+public:
+    concurrent_queue(const size_t max_size) 
+        : max_queue_size(max_size) {
+    }
+
+    void push(Data const& data)
+    {
+        if (size_exceeded()) {
+            boost::mutex::scoped_lock qf_lock(queue_full_mutex);
+            queue_full_cv.wait(qf_lock);
+        }
+
+        boost::mutex::scoped_lock lock(queue_mutex);
+        internal_queue.push(data);
+        lock.unlock();
+        queue_cv.notify_one();
+    }
+
+    bool empty() const
+    {
+        boost::mutex::scoped_lock lock(queue_mutex);
+        return internal_queue.empty();
+    }
+
+    bool try_pop(Data& popped_value)
+    {
+        boost::mutex::scoped_lock lock(queue_mutex);
+        if(internal_queue.empty())
+        {
+            return false;
+        }
+        
+        popped_value=internal_queue.front();
+        internal_queue.pop();
+        queue_full_cv.notify_one();
+        return true;
+    }
+
+    void wait_and_pop(Data& popped_value)
+    {
+        boost::mutex::scoped_lock lock(queue_mutex);
+        while(internal_queue.empty())
+        {
+            queue_cv.wait(lock);
+        }
+        
+        popped_value=internal_queue.front();
+        internal_queue.pop();
+        queue_full_cv.notify_one();
+    }
+
+    int size() const {
+        boost::mutex::scoped_lock lock(queue_mutex);
+        return static_cast<int>(internal_queue.size());
+    }
+};
+
+#endif //#ifndef CONCURRENTQUEUE_H_INCLUDED

DataStructures/ExtractorCallBacks.h

@@ -167,13 +167,13 @@ public:
                 w.nameID = strit->second;
             }
 
+            GUARANTEE(w.id != UINT_MAX, "found way with unknown type");
+            GUARANTEE(-1 != w.speed, "found way with unknown speed");
+
             if ( w.direction == _Way::opposite ){
                 std::reverse( w.path.begin(), w.path.end() );
             }
 
-            GUARANTEE(w.id != UINT_MAX, "found way with unknown type");
-            GUARANTEE(-1 != w.speed, "found way with unknown speed");
-
             for(vector< NodeID >::size_type n = 0; n < w.path.size()-1; ++n) {
                 externalMemory->allEdges.push_back(_Edge(w.path[n], w.path[n+1], w.type, w.direction, w.speed, w.nameID));
                 externalMemory->usedNodeIDs.push_back(w.path[n]);

DataStructures/ImportEdge.h

@@ -23,8 +23,7 @@ or see http://www.gnu.org/licenses/agpl.txt.
 
 #include <cassert>
 
-class Edge
+class Edge {
-{
 public:
 
     bool operator< (const Edge& e) const {

DataStructures/InputReaderFactory.h

@@ -32,8 +32,7 @@ struct BZ2Context {
     char unused[BZ_MAX_UNUSED];
 };
 
-int readFromBz2Stream( void* pointer, char* buffer, int len )
+int readFromBz2Stream( void* pointer, char* buffer, int len ) {
-{
     void *unusedTmpVoid=NULL;
     char *unusedTmp=NULL;
     BZ2Context* context = (BZ2Context*) pointer;

DataStructures/NNGrid.h

@@ -28,7 +28,9 @@ or see http://www.gnu.org/licenses/agpl.txt.
 #include <limits>
 #include <vector>
 #include <stxxl.h>
+#ifdef _OPENMP
 #include <omp.h>
+#endif
 
 #ifdef _WIN32
 #include <math.h>

DataStructures/PBFParser.h

@@ -30,7 +30,7 @@ or see http://www.gnu.org/licenses/agpl.txt.
 #include "../typedefs.h"
 #include "HashTable.h"
 #include "ExtractorStructs.h"
-
+#include "ConcurrentQueue.h"
 
 class PBFParser : public BaseParser<_Node, _RawRestrictionContainer, _Way> {
 
@@ -61,9 +61,9 @@ class PBFParser : public BaseParser<_Node, _RawRestrictionContainer, _Way> {
     };
 
 public:
-    PBFParser(const char * fileName) {
+    PBFParser(const char * fileName) 
+        : threadDataQueue( new concurrent_queue<_ThreadData*>(25) ) { /* Max 25 items in queue */
         GOOGLE_PROTOBUF_VERIFY_VERSION;
-        omp_set_num_threads(1);
         input.open(fileName, std::ios::in | std::ios::binary);
 
         if (!input) {
@@ -86,10 +86,12 @@ public:
         if(input.is_open())
             input.close();
 
-        unsigned maxThreads = omp_get_max_threads();
+        // Clean up any leftover ThreadData objects in the queue
-        for ( unsigned threadNum = 0; threadNum < maxThreads; ++threadNum ) {
+        _ThreadData* td;
-            delete threadDataVector[threadNum];
+        while (threadDataQueue->try_pop(td)) {
+            delete td;
         }
+        delete threadDataQueue;
 
         google::protobuf::ShutdownProtobufLibrary();
 
@@ -100,12 +102,6 @@ public:
     }
 
     bool Init() {
-        /** Init Vector with ThreadData Objects */
-        unsigned maxThreads = omp_get_max_threads();
-        for ( unsigned threadNum = 0; threadNum < maxThreads; ++threadNum ) {
-            threadDataVector.push_back( new _ThreadData( ) );
-        }
-
         _ThreadData initData;
         /** read Header */
         if(!readPBFBlobHeader(input, &initData)) {
@@ -137,36 +133,58 @@ public:
         return true;
     }
 
-    bool Parse() {
+    void ReadData() {
-#pragma omp parallel
+        bool keepRunning = true;
-        {
+        do {
-            _ThreadData * threadData = threadDataVector[omp_get_thread_num()];
+            _ThreadData *threadData = new _ThreadData();
-            //parse through all Blocks
+            keepRunning = readNextBlock(input, threadData);
-            bool keepRunning = true;
-            //        while(readNextBlock(input)) {
-            do{
-#pragma omp critical
-                {
-                    keepRunning = readNextBlock(input, threadData);
-                }
-                if(keepRunning) {
-                    loadBlock(threadData);
-                    for(int i = 0; i < threadData->PBFprimitiveBlock.primitivegroup_size(); i++) {
-                        threadData->currentGroupID = i;
-                        loadGroup(threadData);
 
-                        if(threadData->entityTypeIndicator == TypeNode)
+            if (keepRunning)
-                            parseNode(threadData);
+                threadDataQueue->push(threadData);
-                        if(threadData->entityTypeIndicator == TypeWay)
+            else
-                            parseWay(threadData);
+                threadDataQueue->push(NULL); // No more data to read, parse stops when NULL encountered
-                        if(threadData->entityTypeIndicator == TypeRelation)
+        } while(keepRunning);
-                            parseRelation(threadData);
+    }
-                        if(threadData->entityTypeIndicator == TypeDenseNode)
+
-                            parseDenseNode(threadData);
+    void ParseData() {
-                    }
+        while (1) {
-                }
+            _ThreadData *threadData;
-            }while(keepRunning);
+            threadDataQueue->wait_and_pop(threadData);
+            if (threadData == NULL) {
+                cout << "Parse Data Thread Finished" << endl;
+                threadDataQueue->push(NULL); // Signal end of data for other threads
+                break;
+            }
+
+            loadBlock(threadData);
+            for(int i = 0; i < threadData->PBFprimitiveBlock.primitivegroup_size(); i++) {
+                threadData->currentGroupID = i;
+                loadGroup(threadData);
+
+                if(threadData->entityTypeIndicator == TypeNode)
+                    parseNode(threadData);
+                if(threadData->entityTypeIndicator == TypeWay)
+                    parseWay(threadData);
+                if(threadData->entityTypeIndicator == TypeRelation)
+                    parseRelation(threadData);
+                if(threadData->entityTypeIndicator == TypeDenseNode)
+                    parseDenseNode(threadData);
+            }
+
+            delete threadData;
+            threadData = NULL;
         }
+    }
+
+    bool Parse() {
+        // Start the read and parse threads
+        boost::thread readThread(boost::bind(&PBFParser::ReadData, this));
+        boost::thread parseThread(boost::bind(&PBFParser::ParseData, this));
+        
+        // Wait for the threads to finish
+        readThread.join();
+        parseThread.join();
+
         return true;
     }
 
@@ -200,27 +218,18 @@ private:
                 keyVals.Add(key, value);
                 denseTagIndex += 2;
             }
-#pragma omp critical
+            if(!(*addressCallback)(n, keyVals))
-            {
+                std::cerr << "[PBFParser] adress not parsed" << std::endl;
-                if(!(*addressCallback)(n, keyVals))
+            
-                    std::cerr << "[PBFParser] adress not parsed" << std::endl;
+            if(!(*nodeCallback)(n))
-            }
+                std::cerr << "[PBFParser] dense node not parsed" << std::endl;
-
-#pragma omp critical
-            {
-                if(!(*nodeCallback)(n))
-                    std::cerr << "[PBFParser] dense node not parsed" << std::endl;
-            }
         }
     }
 
     void parseNode(_ThreadData * threadData) {
         _Node n;
-#pragma omp critical
+        if(!(*nodeCallback)(n))
-        {
+            std::cerr << "[PBFParser] simple node not parsed" << std::endl;
-            if(!(*nodeCallback)(n))
-                std::cerr << "[PBFParser] simple node not parsed" << std::endl;
-        }
     }
 
     void parseRelation(_ThreadData * threadData) {
@@ -291,11 +300,8 @@ private:
 //                    cout << "node " << currentRestriction.viaNode;
 //                    cout << " to " << currentRestriction.to << endl;
 //                }
-#pragma omp critical
+                if(!(*restrictionCallback)(currentRestrictionContainer))
-                {
+                    std::cerr << "[PBFParser] relation not parsed" << std::endl;
-                    if(!(*restrictionCallback)(currentRestrictionContainer))
-                        std::cerr << "[PBFParser] relation not parsed" << std::endl;
-                }
             }
         }
     }
@@ -317,18 +323,15 @@ private:
                     const std::string val = threadData->PBFprimitiveBlock.stringtable().s(inputWay.vals(i));
                     w.keyVals.Add(key, val);
                 }
-#pragma omp critical
+
-                {
+                if(!(*wayCallback)(w)) {
-                    if(!(*wayCallback)(w)) {
+                    std::cerr << "[PBFParser] way not parsed" << std::endl;
-                        std::cerr << "[PBFParser] way not parsed" << std::endl;
-                    }
                 }
             }
         }
     }
 
     void loadGroup(_ThreadData * threadData) {
-#pragma omp atomic
         groupCount++;
 
         const OSMPBF::PrimitiveGroup& group = threadData->PBFprimitiveBlock.primitivegroup( threadData->currentGroupID );
@@ -350,7 +353,6 @@ private:
     }
 
     void loadBlock(_ThreadData * threadData) {
-#pragma omp critical
         blockCount++;
         threadData->currentGroupID = 0;
         threadData->currentEntityID = 0;
@@ -522,9 +524,8 @@ private:
     /* the input stream to parse */
     std::fstream input;
 
-    /* ThreadData Array */
+    /* ThreadData Queue */
-    std::vector < _ThreadData* > threadDataVector;
+    concurrent_queue < _ThreadData* >* threadDataQueue;
-
 };
 
 #endif /* PBFPARSER_H_ */

README.TXT

@@ -5,7 +5,7 @@ Compiling the source code is easy. If you are running a decent linux
 installing dependencies and running make should suffice. Make sure the following
 dependencies are installed:
 
- - Boost		1.37+
+ - Boost		1.41+
  - sparsehash   1.4+
  - g++			4.2+ (with warnings and unsupported), 4.4+
  - libxml2		2.7+

createHierarchy.cpp

@@ -38,6 +38,11 @@ or see http://www.gnu.org/licenses/agpl.txt.
 
 #ifdef _OPENMP
 #include <omp.h>
+#else
+int omp_get_num_procs() { return 1; }
+int omp_get_max_threads() { return 1; }
+int omp_get_thread_num() { return 0; }
+int omp_set_num_threads(int i) {}
 #endif
 
 #include "typedefs.h"

extractor.cpp

@@ -229,11 +229,11 @@ int main (int argc, char *argv[]) {
 
         while(wayStartAndEndEdgeIT != externalMemory.wayStartEndVector.end() && restrictionsIT != externalMemory.restrictionsVector.end()) {
             if(wayStartAndEndEdgeIT->wayID < restrictionsIT->fromWay){
-                wayStartAndEndEdgeIT++;
+            	++wayStartAndEndEdgeIT;
                 continue;
             }
             if(wayStartAndEndEdgeIT->wayID > restrictionsIT->fromWay) {
-                restrictionsIT++;
+            	++restrictionsIT;
                 continue;
             }
             assert(wayStartAndEndEdgeIT->wayID == restrictionsIT->fromWay);
@@ -248,7 +248,7 @@ int main (int argc, char *argv[]) {
             } else if(wayStartAndEndEdgeIT->lastTarget == viaNode) {
                 restrictionsIT->restriction.fromNode = wayStartAndEndEdgeIT->lastStart;
             }
-            restrictionsIT++;
+            ++restrictionsIT;
         }
 
         cout << "ok, after " << get_timestamp() - time << "s" << endl;
@@ -266,11 +266,11 @@ int main (int argc, char *argv[]) {
         while(wayStartAndEndEdgeIT != externalMemory.wayStartEndVector.end() &&
                 restrictionsIT != externalMemory.restrictionsVector.end()) {
             if(wayStartAndEndEdgeIT->wayID < restrictionsIT->toWay){
-                wayStartAndEndEdgeIT++;
+            	++wayStartAndEndEdgeIT;
                 continue;
             }
             if(wayStartAndEndEdgeIT->wayID > restrictionsIT->toWay) {
-                restrictionsIT++;
+            	++restrictionsIT;
                 continue;
             }
             NodeID viaNode = restrictionsIT->restriction.viaNode;
@@ -285,9 +285,9 @@ int main (int argc, char *argv[]) {
             }
 
             if(UINT_MAX != restrictionsIT->restriction.fromNode && UINT_MAX != restrictionsIT->restriction.toNode) {
-                usableRestrictionsCounter++;
+            	++usableRestrictionsCounter;
             }
-            restrictionsIT++;
+            ++restrictionsIT;
         }
 
         cout << "ok, after " << get_timestamp() - time << "s" << endl;
@@ -295,7 +295,7 @@ int main (int argc, char *argv[]) {
         ofstream restrictionsOutstream;
         restrictionsOutstream.open(restrictionsFileName.c_str(), ios::binary);
         restrictionsOutstream.write((char*)&usableRestrictionsCounter, sizeof(unsigned));
-        for(restrictionsIT = externalMemory.restrictionsVector.begin(); restrictionsIT != externalMemory.restrictionsVector.end(); restrictionsIT++) {
+        for(restrictionsIT = externalMemory.restrictionsVector.begin(); restrictionsIT != externalMemory.restrictionsVector.end(); ++restrictionsIT) {
             if(UINT_MAX != restrictionsIT->restriction.fromNode && UINT_MAX != restrictionsIT->restriction.toNode) {
                 restrictionsOutstream.write((char *)&(restrictionsIT->restriction), sizeof(_Restriction));
             }
@@ -311,20 +311,20 @@ int main (int argc, char *argv[]) {
         STXXLNodeIDVector::iterator usedNodeIDsIT = externalMemory.usedNodeIDs.begin();
         while(usedNodeIDsIT != externalMemory.usedNodeIDs.end() && nodesIT != externalMemory.allNodes.end()) {
             if(*usedNodeIDsIT < nodesIT->id){
-                usedNodeIDsIT++;
+                ++usedNodeIDsIT;
                 continue;
             }
             if(*usedNodeIDsIT > nodesIT->id) {
-                nodesIT++;
+                ++nodesIT;
                 continue;
             }
             if(*usedNodeIDsIT == nodesIT->id) {
                 fout.write((char*)&(nodesIT->id), sizeof(unsigned));
                 fout.write((char*)&(nodesIT->lon), sizeof(int));
                 fout.write((char*)&(nodesIT->lat), sizeof(int));
-                usedNodeCounter++;
+                ++usedNodeCounter;
-                usedNodeIDsIT++;
+                ++usedNodeIDsIT;
-                nodesIT++;
+                ++nodesIT;
             }
         }
 
@@ -353,7 +353,7 @@ int main (int argc, char *argv[]) {
         STXXLEdgeVector::iterator edgeIT = externalMemory.allEdges.begin();
         while(edgeIT != externalMemory.allEdges.end() && nodesIT != externalMemory.allNodes.end()) {
             if(edgeIT->start < nodesIT->id){
-                edgeIT++;
+            	++edgeIT;
                 continue;
             }
             if(edgeIT->start > nodesIT->id) {
@@ -363,7 +363,7 @@ int main (int argc, char *argv[]) {
             if(edgeIT->start == nodesIT->id) {
                 edgeIT->startCoord.lat = nodesIT->lat;
                 edgeIT->startCoord.lon = nodesIT->lon;
-                edgeIT++;
+                ++edgeIT;
             }
         }
         cout << "ok, after " << get_timestamp() - time << "s" << endl;
@@ -381,11 +381,11 @@ int main (int argc, char *argv[]) {
         edgeIT = externalMemory.allEdges.begin();
         while(edgeIT != externalMemory.allEdges.end() && nodesIT != externalMemory.allNodes.end()) {
             if(edgeIT->target < nodesIT->id){
-                edgeIT++;
+            	++edgeIT;
                 continue;
             }
             if(edgeIT->target > nodesIT->id) {
-                nodesIT++;
+            	++nodesIT;
                 continue;
             }
             if(edgeIT->target == nodesIT->id) {
@@ -430,8 +430,8 @@ int main (int argc, char *argv[]) {
                     fout.write((char*)&edgeType, sizeof(short));
                     fout.write((char*)&edgeIT->nameID, sizeof(unsigned));
                 }
-                usedEdgeCounter++;
+                ++usedEdgeCounter;
-                edgeIT++;
+                ++edgeIT;
             }
         }
         cout << "ok, after " << get_timestamp() - time << "s" << endl;
@@ -494,7 +494,7 @@ bool adressFunction(_Node n, HashTable<string, string> keyVals){
 
 bool restrictionFunction(_RawRestrictionContainer r) {
     extractCallBacks->restrictionFunction(r);
-    globalRestrictionCounter++;
+    ++globalRestrictionCounter;
     return true;
 }
 bool wayFunction(_Way w) {

routed.cpp

@@ -23,6 +23,10 @@ or see http://www.gnu.org/licenses/agpl.txt.
 
 #ifdef _OPENMP
 #include <omp.h>
+#else 
+int omp_get_num_procs() { return 1; }
+int omp_get_max_threads() { return 1; }
+int omp_get_thread_num() { return 0; }
 #endif
 
 #include <boost/thread.hpp>

vcprojects/createHierarchy.vcproj

@@ -99,7 +99,7 @@
 		<Configuration
 			Name="Release|Win32"
 			OutputDirectory="$(SolutionDir)..\bin"
-			IntermediateDirectory="(SolutionDir)..\bin"
+			IntermediateDirectory="$(SolutionDir)..\bin"
 			ConfigurationType="1"
 			InheritedPropertySheets=".\osrm.vsprops"
 			CharacterSet="1"

vcprojects/extractor.vcproj

@@ -177,7 +177,7 @@
 			/>
 			<Tool
 				Name="VCPostBuildEventTool"
-				CommandLine="xcopy &quot;$(SolutionDir)..\.stxxl&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;xcopy &quot;$(SolutionDir)..\extractor.ini&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;xcopy &quot;$(Bzip2Path)\bin\libbz2.dll&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;"
+				CommandLine="xcopy &quot;$(SolutionDir)..\.stxxl&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;xcopy &quot;$(SolutionDir)..\extractor.ini&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;xcopy &quot;$(SolutionDir)..\speedprofile.ini&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;xcopy &quot;$(Bzip2Path)\bin\libbz2.dll&quot; &quot;$(OutDir)\&quot; /f /y&#x0D;&#x0A;"
 			/>
 		</Configuration>
 	</Configurations>
@@ -250,6 +250,10 @@
 					RelativePath="..\DataStructures\BinaryHeap.h"
 					>
 				</File>
+				<File
+					RelativePath="..\DataStructures\ConcurrentQueue.h"
+					>
+				</File>
 				<File
 					RelativePath="..\DataStructures\DynamicGraph.h"
 					>