diff --git a/tile-join.cpp b/tile-join.cpp index 5f96686e..499ba04f 100644 --- a/tile-join.cpp +++ b/tile-join.cpp @@ -10,6 +10,7 @@ #include #include #include +#include #include "mvt.hpp" #include "projection.hpp" #include "pool.hpp" @@ -28,7 +29,7 @@ struct stats { double minlat, minlon, maxlat, maxlon; }; -void handle(std::string message, int z, unsigned x, unsigned y, std::map &layermap, sqlite3 *outdb, std::vector &header, std::map > &mapping, std::set &exclude, int ifmatched, mvt_tile &outtile) { +void handle(std::string message, int z, unsigned x, unsigned y, std::map &layermap, std::vector &header, std::map> &mapping, std::set &exclude, int ifmatched, mvt_tile &outtile) { mvt_tile tile; int features_added = 0; @@ -132,7 +133,7 @@ void handle(std::string message, int z, unsigned x, unsigned y, std::map 0 && strcmp(key, header[0].c_str()) == 0) { - std::map >::iterator ii = mapping.find(value); + std::map>::iterator ii = mapping.find(value); if (ii != mapping.end()) { std::vector fields = ii->second; @@ -299,42 +300,154 @@ struct reader *begin_reading(char *fname) { return r; } -void decode(struct reader *readers, char *map, std::map &layermap, sqlite3 *outdb, struct stats *st, std::vector &header, std::map > &mapping, std::set &exclude, int ifmatched, std::string &attribution) { - std::vector > layermaps; +struct zxy { + long long z; + long long x; + long long y; + + zxy(long long _z, long long _x, long long _y) { + z = _z; + x = _x; + y = _y; + } + + bool operator<(zxy const &other) const { + if (z < other.z) { + return true; + } + if (z > other.z) { + return false; + } + + if (x < other.x) { + return true; + } + if (x > other.x) { + return false; + } + + if (y < other.y) { + return true; + } + + return false; + } +}; + +struct arg { + std::map> inputs; + std::map outputs; + + std::map *layermap; + + std::vector *header; + std::map> *mapping; + std::set *exclude; + int ifmatched; +}; + +void *join_worker(void *v) { + arg *a = (arg *) v; + + for (auto ai = a->inputs.begin(); ai != a->inputs.end(); ++ai) { + mvt_tile tile; + + for (size_t i = 0; i < ai->second.size(); i++) { + handle(ai->second[i], ai->first.z, ai->first.x, ai->first.y, *(a->layermap), *(a->header), *(a->mapping), *(a->exclude), a->ifmatched, tile); + } + + bool anything = false; + for (size_t i = 0; i < tile.layers.size(); i++) { + if (tile.layers[i].features.size() > 0) { + anything = true; + break; + } + } + + if (anything) { + std::string compressed = tile.encode(); + + if (!pk && compressed.size() > 500000) { + fprintf(stderr, "Tile %lld/%lld/%lld size is %lld, >500000. Skipping this tile\n.", ai->first.z, ai->first.x, ai->first.y, (long long) compressed.size()); + } else { + a->outputs.insert(std::pair(ai->first, compressed)); + } + } + } + + return NULL; +} + +void handle_tasks(std::map> &tasks, std::vector> &layermaps, sqlite3 *outdb, std::vector &header, std::map> &mapping, std::set &exclude, int ifmatched) { + pthread_t pthreads[CPUS]; + std::vector args; + + for (size_t i = 0; i < CPUS; i++) { + args.push_back(arg()); + + args[i].layermap = &layermaps[i]; + args[i].header = &header; + args[i].mapping = &mapping; + args[i].exclude = &exclude; + args[i].ifmatched = ifmatched; + } + + size_t count = 0; + // XXX Be more careful to distribute tasks evenly across CPUs + for (auto ai = tasks.begin(); ai != tasks.end(); ++ai) { + args[count].inputs.insert(*ai); + count = (count + 1) % CPUS; + + if (ai == tasks.begin()) { + fprintf(stderr, "%lld/%lld/%lld \r", ai->first.z, ai->first.x, ai->first.y); + } + } + + for (size_t i = 0; i < CPUS; i++) { + if (pthread_create(&pthreads[i], NULL, join_worker, &args[i]) != 0) { + perror("pthread_create"); + exit(EXIT_FAILURE); + } + } + + for (int i = 0; i < CPUS; i++) { + void *retval; + + if (pthread_join(pthreads[i], &retval) != 0) { + perror("pthread_join"); + } + + for (auto ai = args[i].outputs.begin(); ai != args[i].outputs.end(); ++ai) { + mbtiles_write_tile(outdb, ai->first.z, ai->first.x, ai->first.y, ai->second.data(), ai->second.size()); + } + } +} + +void decode(struct reader *readers, char *map, std::map &layermap, sqlite3 *outdb, struct stats *st, std::vector &header, std::map> &mapping, std::set &exclude, int ifmatched, std::string &attribution) { + std::vector> layermaps; for (size_t i = 0; i < CPUS; i++) { layermaps.push_back(std::map()); } - mvt_tile tile; + std::map> tasks; while (readers != NULL && readers->zoom < 32) { reader *r = readers; readers = readers->next; r->next = NULL; - fprintf(stderr, "%lld/%lld/%lld \r", r->zoom, r->x, r->y); - handle(r->data, r->zoom, r->x, r->y, layermaps[0], outdb, header, mapping, exclude, ifmatched, tile); + zxy tile = zxy(r->zoom, r->x, r->y); + if (tasks.count(tile) == 0) { + tasks.insert(std::pair>(tile, std::vector())); + } + auto f = tasks.find(tile); + f->second.push_back(r->data); if (readers == NULL || readers->zoom != r->zoom || readers->x != r->x || readers->y != r->y) { - bool anything = false; - for (size_t i = 0; i < tile.layers.size(); i++) { - if (tile.layers[i].features.size() > 0) { - anything = true; - break; - } + if (tasks.size() > 100 * CPUS) { + handle_tasks(tasks, layermaps, outdb, header, mapping, exclude, ifmatched); + tasks.clear(); } - - if (anything) { - std::string compressed = tile.encode(); - - if (!pk && compressed.size() > 500000) { - fprintf(stderr, "Tile %lld/%lld/%lld size is %lld, >500000. Skipping this tile\n.", r->zoom, r->x, r->y, (long long) compressed.size()); - } else { - mbtiles_write_tile(outdb, r->zoom, r->x, r->y, compressed.data(), compressed.size()); - } - } - - tile = mvt_tile(); } if (sqlite3_step(r->stmt) == SQLITE_ROW) { @@ -363,6 +476,7 @@ void decode(struct reader *readers, char *map, std::map &header, std::map > &mapping) { +void readcsv(char *fn, std::vector &header, std::map> &mapping) { FILE *f = fopen(fn, "r"); if (f == NULL) { perror(fn); @@ -491,7 +605,7 @@ void readcsv(char *fn, std::vector &header, std::map >(line[0], line)); + mapping.insert(std::pair>(line[0], line)); } } @@ -510,7 +624,7 @@ int main(int argc, char **argv) { } std::vector header; - std::map > mapping; + std::map> mapping; std::set exclude;