diff --git a/include/attribute_store.h b/include/attribute_store.h index d5a2d386..c6020a41 100644 --- a/include/attribute_store.h +++ b/include/attribute_store.h @@ -167,7 +167,7 @@ struct AttributePair { else if(hasBoolValue()) boost::hash_combine(rv, boolValue()); else { - throw new std::out_of_range("cannot hash pair, unknown value"); + throw std::out_of_range("cannot hash pair, unknown value"); } return rv; diff --git a/include/helpers.h b/include/helpers.h index a945db90..4dcb134e 100644 --- a/include/helpers.h +++ b/include/helpers.h @@ -2,7 +2,9 @@ #ifndef _HELPERS_H #define _HELPERS_H +#include #include +#include #include #include @@ -41,6 +43,16 @@ struct OffsetAndLength { uint64_t length; }; +struct FileCloser { + void operator()(FILE* fp) const { + if (fp != nullptr) + fclose(fp); + } +}; + +using FilePtr = std::unique_ptr; + +FilePtr openFile(const std::string &filename, const char *mode); uint64_t getFileSize(std::string filename); std::vector getNewlineChunks(const std::string &filename, uint64_t chunks); diff --git a/include/shp_processor.h b/include/shp_processor.h index aa4fc2a7..386ab93b 100644 --- a/include/shp_processor.h +++ b/include/shp_processor.h @@ -38,7 +38,7 @@ class ShpProcessor { void fillPointArrayFromShapefile(std::vector *points, SHPObject *shape, uint part); // Read requested attributes from a shapefile, and encode into an OutputObject - AttributeIndex readShapefileAttributes(DBFHandle &dbf, int recordNum, + AttributeIndex readShapefileAttributes(DBFHandle dbf, int recordNum, std::unordered_map &columnMap, std::unordered_map &columnTypeMap, LayerDef &layer, uint &minzoom); @@ -49,4 +49,3 @@ class ShpProcessor { }; #endif //_SHP_PROCESSOR_H - diff --git a/src/geojson_processor.cpp b/src/geojson_processor.cpp index 9cb0ddfe..131458f7 100644 --- a/src/geojson_processor.cpp +++ b/src/geojson_processor.cpp @@ -3,6 +3,7 @@ #include "helpers.h" #include #include +#include #include "rapidjson/document.h" #include "rapidjson/writer.h" @@ -24,25 +25,34 @@ void GeoJSONProcessor::read(class LayerDef &layer, uint layerNum) { void GeoJSONProcessor::readFeatureCollection(class LayerDef &layer, uint layerNum) { // Read a JSON file containing a single GeoJSON FeatureCollection object. rapidjson::Document doc; - FILE* fp = fopen(layer.source.c_str(), "r"); + FilePtr fp = openFile(layer.source, "r"); char readBuffer[65536]; - rapidjson::FileReadStream is(fp, readBuffer, sizeof(readBuffer)); + rapidjson::FileReadStream is(fp.get(), readBuffer, sizeof(readBuffer)); doc.ParseStream(is); if (doc.HasParseError()) { throw std::runtime_error("Invalid JSON file."); } - fclose(fp); if (strcmp(doc["type"].GetString(), "FeatureCollection") != 0) { throw std::runtime_error("Top-level GeoJSON object must be a FeatureCollection."); } // Process each feature + std::exception_ptr error; + std::mutex errorMutex; boost::asio::thread_pool pool(threadNum); for (auto &feature : doc["features"].GetArray()) { boost::asio::post(pool, [&]() { - processFeature(std::move(feature.GetObject()), layer, layerNum); + try { + processFeature(std::move(feature.GetObject()), layer, layerNum); + } catch (...) { + std::lock_guard lock(errorMutex); + if (!error) + error = std::current_exception(); + } }); } pool.join(); + if (error) + std::rethrow_exception(error); } void GeoJSONProcessor::readFeatureLines(class LayerDef &layer, uint layerNum) { @@ -50,30 +60,39 @@ void GeoJSONProcessor::readFeatureLines(class LayerDef &layer, uint layerNum) { std::vector chunks = getNewlineChunks(layer.source, threadNum * 4); // Process each feature + std::exception_ptr error; + std::mutex errorMutex; boost::asio::thread_pool pool(threadNum); for (auto &chunk : chunks) { boost::asio::post(pool, [&]() { - FILE* fp = fopen(layer.source.c_str(), "r"); - if (fseek(fp, chunk.offset, SEEK_SET) != 0) throw std::runtime_error("unable to seek to " + std::to_string(chunk.offset) + " in " + layer.source); - char readBuffer[65536]; - rapidjson::FileReadStream is(fp, readBuffer, sizeof(readBuffer)); + try { + FilePtr fp = openFile(layer.source, "r"); + if (fseek(fp.get(), chunk.offset, SEEK_SET) != 0) throw std::runtime_error("unable to seek to " + std::to_string(chunk.offset) + " in " + layer.source); + char readBuffer[65536]; + rapidjson::FileReadStream is(fp.get(), readBuffer, sizeof(readBuffer)); - // Skip leading whitespace. - while(is.Tell() < chunk.length && isspace(is.Peek())) is.Take(); + // Skip leading whitespace. + while(is.Tell() < chunk.length && isspace(is.Peek())) is.Take(); - while(is.Tell() < chunk.length) { - auto doc = rapidjson::Document(); - doc.ParseStream(is); - if (doc.HasParseError()) { throw std::runtime_error("Invalid JSON file."); } - processFeature(std::move(doc.GetObject()), layer, layerNum); + while(is.Tell() < chunk.length) { + auto doc = rapidjson::Document(); + doc.ParseStream(is); + if (doc.HasParseError()) { throw std::runtime_error("Invalid JSON file."); } + processFeature(std::move(doc.GetObject()), layer, layerNum); - // Skip trailing whitespace. - while(is.Tell() < chunk.length && isspace(is.Peek())) is.Take(); + // Skip trailing whitespace. + while(is.Tell() < chunk.length && isspace(is.Peek())) is.Take(); + } + } catch (...) { + std::lock_guard lock(errorMutex); + if (!error) + error = std::current_exception(); } - fclose(fp); }); } pool.join(); + if (error) + std::rethrow_exception(error); } template diff --git a/src/helpers.cpp b/src/helpers.cpp index 751bd7ce..e05c3fde 100644 --- a/src/helpers.cpp +++ b/src/helpers.cpp @@ -203,6 +203,13 @@ uint64_t getFileSize(std::string filename) { throw std::runtime_error("unable to stat " + filename); } +FilePtr openFile(const std::string &filename, const char *mode) { + FilePtr fp(fopen(filename.c_str(), mode)); + if (fp == nullptr) + throw std::runtime_error("unable to open " + filename); + return fp; +} + // Given a file, attempt to divide it into N chunks, with each chunk separated // by a newline. // @@ -212,7 +219,7 @@ std::vector getNewlineChunks(const std::string &filename, uint6 const uint64_t size = getFileSize(filename); const uint64_t chunkSize = std::max(size / chunks, 1ul); - FILE* fp = fopen(filename.c_str(), "r"); + FilePtr fp = openFile(filename, "r"); // Our approach is naive: skip chunkSize bytes, scan for a newline, repeat. // @@ -226,12 +233,12 @@ std::vector getNewlineChunks(const std::string &filename, uint6 // The last chunk will not be a full `chunkSize`. length = std::min(chunkSize, size - offset); - if (fseek(fp, offset + length, SEEK_SET) != 0) throw std::runtime_error("unable to seek to " + std::to_string(offset) + " in " + filename); + if (fseek(fp.get(), offset + length, SEEK_SET) != 0) throw std::runtime_error("unable to seek to " + std::to_string(offset) + " in " + filename); bool foundNewline = false; while(!foundNewline) { - size_t read = fread(buffer, 1, sizeof(buffer), fp); + size_t read = fread(buffer, 1, sizeof(buffer), fp.get()); if (read == 0) break; for (int i = 0; i < read; i++) { if (buffer[i] == '\n') { @@ -248,6 +255,5 @@ std::vector getNewlineChunks(const std::string &filename, uint6 offset += length; } - fclose(fp); return rv; } diff --git a/src/pooled_string.cpp b/src/pooled_string.cpp index 393e9f9b..b0659247 100644 --- a/src/pooled_string.cpp +++ b/src/pooled_string.cpp @@ -2,10 +2,18 @@ #include #include #include +#include #include +#include namespace PooledStringNS { - std::deque tables; + struct TableDeleter { + void operator()(char* table) const { + std::free(table); + } + }; + + std::deque> tables; std::mutex mutex; const uint8_t ShortString = 0b00; @@ -33,10 +41,10 @@ PooledString::PooledString(const std::string& str) { if (spaceLeft < 0 || spaceLeft < str.size()) { std::lock_guard lock(mutex); spaceLeft = 65536; - char* buffer = (char*)malloc(spaceLeft); - if (buffer == 0) + std::unique_ptr buffer(static_cast(std::malloc(spaceLeft))); + if (!buffer) throw std::runtime_error("PooledString could not malloc"); - tables.push_back(buffer); + tables.emplace_back(std::move(buffer)); tableIndex = tables.size() - 1; } @@ -52,7 +60,7 @@ PooledString::PooledString(const std::string& str) { storage[6] = length >> 8; storage[7] = length; - memcpy(tables[tableIndex] + offset, str.data(), str.size()); + memcpy(tables[tableIndex].get() + offset, str.data(), str.size()); spaceLeft -= str.size(); } @@ -106,7 +114,7 @@ const char* PooledStringNS::PooledString::data() const { uint32_t tableIndex = (storage[1] << 16) + (storage[2] << 8) + storage[3]; uint16_t offset = (storage[4] << 8) + storage[5]; - const char* data = tables[tableIndex] + offset; + const char* data = tables[tableIndex].get() + offset; return data; } @@ -136,7 +144,7 @@ std::string PooledStringNS::PooledString::toString() const { uint32_t tableIndex = (storage[1] << 16) + (storage[2] << 8) + storage[3]; uint16_t offset = (storage[4] << 8) + storage[5]; - char* data = tables[tableIndex] + offset; + char* data = tables[tableIndex].get() + offset; rv.append(data, size()); return rv; } @@ -169,4 +177,3 @@ bool PooledStringNS::PooledString::operator<(const PooledString& other) const { return memcmp(data(), other.data(), mySize) < 0; } - diff --git a/src/shp_processor.cpp b/src/shp_processor.cpp index 2cf164d7..469932c8 100644 --- a/src/shp_processor.cpp +++ b/src/shp_processor.cpp @@ -2,12 +2,32 @@ #include #include +#include +#include +#include extern bool verbose; using namespace std; namespace geom = boost::geometry; +namespace { + struct ShpHandleCloser { + void operator()(SHPHandle shp) const { + SHPClose(shp); + } + }; + + struct DbfHandleCloser { + void operator()(DBFHandle dbf) const { + DBFClose(dbf); + } + }; + + using ShpHandlePtr = std::unique_ptr::type, ShpHandleCloser>; + using DbfHandlePtr = std::unique_ptr::type, DbfHandleCloser>; +} + /* Read shapefiles into Boost geometries */ @@ -36,7 +56,7 @@ void ShpProcessor::fillPointArrayFromShapefile(vector *points, SHPObject // Read requested attributes from a shapefile, and encode into an OutputObject // columnTypeMap: 0 string, 1 int, 2 double, 3 boolean AttributeIndex ShpProcessor::readShapefileAttributes( - DBFHandle &dbf, + DBFHandle dbf, int recordNum, unordered_map &columnMap, unordered_map &columnTypeMap, LayerDef &layer, uint &minzoom) { @@ -113,39 +133,42 @@ void ShpProcessor::read(class LayerDef &layer, uint layerNum) const string &indexName = layer.indexName; // open shapefile - SHPHandle shp = SHPOpen(filename.c_str(), "rb"); - DBFHandle dbf = DBFOpen(filename.c_str(), "rb"); + ShpHandlePtr shp(SHPOpen(filename.c_str(), "rb")); + DbfHandlePtr dbf(DBFOpen(filename.c_str(), "rb")); if(shp == nullptr || dbf == nullptr) return; int numEntities=0, shpType=0; double adfMinBound[4], adfMaxBound[4]; - SHPGetInfo(shp, &numEntities, &shpType, adfMinBound, adfMaxBound); + SHPGetInfo(shp.get(), &numEntities, &shpType, adfMinBound, adfMaxBound); // prepare columns unordered_map columnMap; unordered_map columnTypeMap; if (layer.allSourceColumns) { - for (size_t i=0; i-1) { columnMap[dbfLoc]=columns[i]; - columnTypeMap[dbfLoc]=DBFGetFieldInfo(dbf,dbfLoc,NULL,NULL,NULL); + columnTypeMap[dbfLoc]=DBFGetFieldInfo(dbf.get(),dbfLoc,NULL,NULL,NULL); } } } int indexField=-1; - if (indexName!="") { indexField = DBFGetFieldIndex(dbf,indexName.c_str()); } + if (indexName!="") { indexField = DBFGetFieldIndex(dbf.get(),indexName.c_str()); } + std::exception_ptr error; + std::mutex errorMutex; boost::asio::thread_pool pool(threadNum); for (int i=0; i shape(rawShape, SHPDestroyObject); // Check shape is in clippingBox Box shapeBox(Point(shape->dfXMin, lat2latp(shape->dfYMin)), Point(shape->dfXMax, lat2latp(shape->dfYMax))); @@ -153,27 +176,31 @@ void ShpProcessor::read(class LayerDef &layer, uint layerNum) shapeBox.max_corner().get<0>() < clippingBox.min_corner().get<0>() || shapeBox.min_corner().get<1>() > clippingBox.max_corner().get<1>() || shapeBox.max_corner().get<1>() < clippingBox.min_corner().get<1>()) { - SHPDestroyObject(shape); continue; } boost::asio::post(pool, [&, i, shape]() { - // process attributes - string name; - bool hasName = false; - if (indexField>-1) { - std::lock_guard lock(attributeMutex); - name=DBFReadStringAttribute(dbf, i, indexField); hasName = true; + try { + // process attributes + string name; + bool hasName = false; + if (indexField>-1) { + std::lock_guard lock(attributeMutex); + name=DBFReadStringAttribute(dbf.get(), i, indexField); hasName = true; + } + AttributeIndex attrIdx = readShapefileAttributes(dbf.get(), i, columnMap, columnTypeMap, layer, layer.minzoom); + // process geometry + processShapeGeometry(shape.get(), attrIdx, layer, layerNum, hasName, name); + } catch (...) { + std::lock_guard lock(errorMutex); + if (!error) + error = std::current_exception(); } - AttributeIndex attrIdx = readShapefileAttributes(dbf, i, columnMap, columnTypeMap, layer, layer.minzoom); - // process geometry - processShapeGeometry(shape, attrIdx, layer, layerNum, hasName, name); - SHPDestroyObject(shape); }); } pool.join(); - SHPClose(shp); - DBFClose(dbf); + if (error) + std::rethrow_exception(error); } void ShpProcessor::processShapeGeometry(SHPObject* shape, AttributeIndex attrIdx, diff --git a/src/tilemaker.cpp b/src/tilemaker.cpp index 6cbd920a..5fadd51c 100644 --- a/src/tilemaker.cpp +++ b/src/tilemaker.cpp @@ -169,12 +169,11 @@ int main(const int argc, const char* argv[]) { rapidjson::Document jsonConfig; class Config config; try { - FILE* fp = fopen(options.jsonFile.c_str(), "r"); + FilePtr fp = openFile(options.jsonFile, "r"); char readBuffer[65536]; - rapidjson::FileReadStream is(fp, readBuffer, sizeof(readBuffer)); + rapidjson::FileReadStream is(fp.get(), readBuffer, sizeof(readBuffer)); jsonConfig.ParseStream(is); if (jsonConfig.HasParseError()) { cerr << "Invalid JSON file." << endl; return -1; } - fclose(fp); config.readConfig(jsonConfig, hasClippingBox, clippingBox); } catch (...) { @@ -272,12 +271,22 @@ int main(const int argc, const char* argv[]) { if (!hasClippingBox) { cerr << "Can't read shapefiles unless a bounding box is provided." << endl; exit(EXIT_FAILURE); - } else if (ends_with(layer.source, "json") || ends_with(layer.source, "jsonl") || ends_with(layer.source, "JSON") || ends_with(layer.source, "JSONL") || ends_with(layer.source, "jsonseq") || ends_with(layer.source, "JSONSEQ")) { - cout << "Reading GeoJSON " << layer.name << endl; - geoJSONProcessor.read(layers.layers[layerNum], layerNum); } else { - cout << "Reading shapefile " << layer.name << endl; - shpProcessor.read(layers.layers[layerNum], layerNum); + try { + if (ends_with(layer.source, "json") || ends_with(layer.source, "jsonl") || ends_with(layer.source, "JSON") || ends_with(layer.source, "JSONL") || ends_with(layer.source, "jsonseq") || ends_with(layer.source, "JSONSEQ")) { + cout << "Reading GeoJSON " << layer.name << endl; + geoJSONProcessor.read(layers.layers[layerNum], layerNum); + } else { + cout << "Reading shapefile " << layer.name << endl; + shpProcessor.read(layers.layers[layerNum], layerNum); + } + } catch (const std::exception& e) { + cerr << "Error reading external source " << layer.source << ": " << e.what() << endl; + return -1; + } catch (...) { + cerr << "Unknown error reading external source " << layer.source << endl; + return -1; + } } } } @@ -564,4 +573,3 @@ int main(const int argc, const char* argv[]) { cout << endl << "Filled the tileset with good things at " << sharedData.outputFile << endl; } - diff --git a/test/attribute_store.test.cpp b/test/attribute_store.test.cpp index 9b2fad54..e0eee1ee 100644 --- a/test/attribute_store.test.cpp +++ b/test/attribute_store.test.cpp @@ -122,7 +122,7 @@ MU_TEST(test_attribute_store_capacity) { try { keys.key2index("key512"); - } catch (std::out_of_range) { + } catch (const std::out_of_range&) { caughtException = true; } diff --git a/test/helpers.test.cpp b/test/helpers.test.cpp index c5e2be49..7a07ac62 100644 --- a/test/helpers.test.cpp +++ b/test/helpers.test.cpp @@ -1,7 +1,22 @@ #include +#include #include "external/minunit.h" #include "helpers.h" +MU_TEST(test_open_file) { + FilePtr fp = openFile("test/test.jsonl", "r"); + mu_check(fp != nullptr); + mu_check(fgetc(fp.get()) == '{'); + + bool caughtException = false; + try { + openFile("test/this-file-does-not-exist", "r"); + } catch (const std::runtime_error&) { + caughtException = true; + } + mu_check(caughtException); +} + MU_TEST(test_get_chunks) { { auto rv = getNewlineChunks("test/test.jsonl", 1); @@ -83,6 +98,7 @@ MU_TEST(test_compression_zlib) { MU_TEST_SUITE(test_suite_helpers) { + MU_RUN_TEST(test_open_file); MU_RUN_TEST(test_get_chunks); MU_RUN_TEST(test_compression_gzip); MU_RUN_TEST(test_compression_zlib); diff --git a/test/pooled_string.test.cpp b/test/pooled_string.test.cpp index 93a05843..9c8baa94 100644 --- a/test/pooled_string.test.cpp +++ b/test/pooled_string.test.cpp @@ -1,4 +1,8 @@ #include +#include +#include +#include +#include #include "external/minunit.h" #include "pooled_string.h" @@ -46,8 +50,89 @@ MU_TEST(test_pooled_string) { mu_check(stdShortString != stdLongString); } +MU_TEST(test_pooled_string_boundaries) { + const std::string shortBoundary(15, 'a'); + const std::string pooledBoundary(16, 'b'); + const std::string largest(65535, 'c'); + + PooledString shortString(shortBoundary); + PooledString pooledString(pooledBoundary); + PooledString largestString(largest); + + mu_check(shortString.size() == shortBoundary.size()); + mu_check(shortString.toString() == shortBoundary); + mu_check(pooledString.size() == pooledBoundary.size()); + mu_check(pooledString.toString() == pooledBoundary); + mu_check(largestString.size() == largest.size()); + mu_check(largestString.toString() == largest); + + bool caughtException = false; + try { + PooledString tooLarge(std::string(65536, 'd')); + } catch (const std::runtime_error&) { + caughtException = true; + } + mu_check(caughtException); + + // The largest string leaves a one-byte tail. Allocating another pooled + // string must not invalidate the preceding table. + PooledString nextTable(std::string(16, 'e')); + mu_check(largestString.toString() == largest); + mu_check(nextTable.toString() == std::string(16, 'e')); +} + +MU_TEST(test_pooled_string_multiple_tables) { + std::vector originals; + std::vector pooled; + for (int i = 0; i < 2048; i++) { + originals.emplace_back("pooled string " + std::to_string(i) + + std::string(48, char('a' + i % 26))); + pooled.emplace_back(originals.back()); + } + + for (size_t i = 0; i < pooled.size(); i++) + mu_check(pooled[i].toString() == originals[i]); +} + +MU_TEST(test_pooled_string_threaded) { + const int threadCount = 8; + const int stringsPerThread = 256; + std::atomic start(false); + std::atomic failed(false); + std::vector threads; + + for (int thread = 0; thread < threadCount; thread++) { + threads.emplace_back([&, thread]() { + std::vector originals; + std::vector pooled; + for (int i = 0; i < stringsPerThread; i++) + originals.emplace_back("thread " + std::to_string(thread) + + " string " + std::to_string(i) + + std::string(48, char('a' + thread))); + + while (!start.load(std::memory_order_acquire)) {} + for (const std::string& value : originals) + pooled.emplace_back(value); + + for (size_t i = 0; i < pooled.size(); i++) { + if (pooled[i].toString() != originals[i]) + failed.store(true, std::memory_order_release); + } + }); + } + + start.store(true, std::memory_order_release); + for (std::thread& thread : threads) + thread.join(); + + mu_check(failed.load(std::memory_order_acquire) == false); +} + MU_TEST_SUITE(test_suite_pooled_string) { MU_RUN_TEST(test_pooled_string); + MU_RUN_TEST(test_pooled_string_boundaries); + MU_RUN_TEST(test_pooled_string_multiple_tables); + MU_RUN_TEST(test_pooled_string_threaded); } int main() {