nominatim/index.c

   1 /*
   2  * triggers indexing (reparenting etc.) through setting resetting indexed_status: update placex/osmline set indexed_status = 0 where indexed_status > 0
   3  * triggers placex_update and osmline_update
   4 */
   5
   6 #include <stdio.h>
   7 #include <unistd.h>
   8 #include <stdlib.h>
   9 #include <string.h>
  10 #include <assert.h>
  11 #include <pthread.h>
  12 #include <time.h>
  13 #include <stdint.h>
  14
  15 #include <libpq-fe.h>
  16
  17 #include "nominatim.h"
  18 #include "index.h"
  19 #include "export.h"
  20 #include "postgresql.h"
  21
  22 extern int verbose;
  23
  24 void run_indexing(int rank, int interpolation, PGconn *conn, int num_threads,
  25 struct index_thread_data * thread_data, const char *structuredoutputfile)
  26 {
  27     int tuples, count, sleepcount;
  28     pthread_mutex_t count_mutex = PTHREAD_MUTEX_INITIALIZER;
  29
  30     time_t rankStartTime;
  31     int rankTotalTuples;
  32     int rankCountTuples;
  33     float rankPerSecond;
  34
  35     PGresult * resSectors;
  36     PGresult * resPlaces;
  37     PGresult * resNULL;
  38
  39     int i;
  40     int iSector;
  41     int iResult;
  42
  43     const char *paramValues[2];
  44     int         paramLengths[2];
  45     int         paramFormats[2];
  46     uint32_t    paramRank;
  47     uint32_t    paramSector;
  48     uint32_t    sector;
  49
  50     xmlTextWriterPtr writer;
  51     pthread_mutex_t writer_mutex = PTHREAD_MUTEX_INITIALIZER;
  52
  53     // Create the output file
  54     writer = NULL;
  55     if (structuredoutputfile)
  56     {
  57         writer = nominatim_exportXMLStart(structuredoutputfile);
  58     }
  59
  60     if (interpolation)
  61     {
  62         fprintf(stderr, "Starting interpolation lines (location_property_osmline)\n");
  63     }
  64     else
  65     {
  66         fprintf(stderr, "Starting rank %d\n", rank);
  67     }
  68
  69     rankCountTuples = 0;
  70     rankPerSecond = 0;
  71
  72     paramRank = PGint32(rank);
  73     paramValues[0] = (char *)&paramRank;
  74     paramLengths[0] = sizeof(paramRank);
  75     paramFormats[0] = 1;
  76
  77     if (interpolation)
  78     {
  79         resSectors = PQexecPrepared(conn, "index_sectors_osmline", 0, NULL, 0, NULL, 1);
  80     }
  81     else
  82     {
  83         resSectors = PQexecPrepared(conn, "index_sectors", 1, paramValues, paramLengths, paramFormats, 1);
  84     }
  85     if (PQresultStatus(resSectors) != PGRES_TUPLES_OK)
  86     {
  87         fprintf(stderr, "index_sectors: SELECT failed: %s", PQerrorMessage(conn));
  88         PQclear(resSectors);
  89         exit(EXIT_FAILURE);
  90     }
  91     if (PQftype(resSectors, 0) != PG_OID_INT4)
  92     {
  93         fprintf(stderr, "Sector value has unexpected type\n");
  94         PQclear(resSectors);
  95         exit(EXIT_FAILURE);
  96     }
  97     if (PQftype(resSectors, 1) != PG_OID_INT8)
  98     {
  99         fprintf(stderr, "Sector value has unexpected type\n");
 100         PQclear(resSectors);
 101         exit(EXIT_FAILURE);
 102     }
 103
 104     rankTotalTuples = 0;
 105     for (iSector = 0; iSector < PQntuples(resSectors); iSector++)
 106     {
 107         rankTotalTuples += PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1)));
 108     }
 109
 110     rankStartTime = time(0);
 111     for (iSector = 0; iSector <= PQntuples(resSectors); iSector++)
 112     {
 113         if (iSector > 0)
 114         {
 115             resPlaces = PQgetResult(conn);
 116             if (PQresultStatus(resPlaces) != PGRES_TUPLES_OK)
 117             {
 118                 fprintf(stderr, "index_sector_places: SELECT failed: %s", PQerrorMessage(conn));
 119                 PQclear(resPlaces);
 120                 exit(EXIT_FAILURE);
 121             }
 122             if (PQftype(resPlaces, 0) != PG_OID_INT8)
 123             {
 124                 fprintf(stderr, "Place_id value has unexpected type\n");
 125                 PQclear(resPlaces);
 126                 exit(EXIT_FAILURE);
 127             }
 128             resNULL = PQgetResult(conn);
 129             if (resNULL != NULL)
 130             {
 131                 fprintf(stderr, "Unexpected non-null response\n");
 132                 exit(EXIT_FAILURE);
 133             }
 134         }
 135
 136         if (iSector < PQntuples(resSectors))
 137         {
 138             sector = PGint32(*((uint32_t *)PQgetvalue(resSectors, iSector, 0)));
 139 //                fprintf(stderr, "\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));
 140
 141             // Get all the place_id's for this sector
 142             paramRank = PGint32(rank);
 143             paramValues[0] = (char *)&paramRank;
 144             paramLengths[0] = sizeof(paramRank);
 145             paramFormats[0] = 1;
 146             paramSector = PGint32(sector);
 147             paramValues[1] = (char *)&paramSector;
 148             paramLengths[1] = sizeof(paramSector);
 149             paramFormats[1] = 1;
 150             if (rankTotalTuples-rankCountTuples < num_threads*1000)
 151             {
 152                 // no sectors
 153                 if (interpolation)
 154                 {
 155                     iResult = PQsendQueryPrepared(conn, "index_nosector_places_osmline", 0, NULL, 0, NULL, 1);
 156                 }
 157                 else
 158                 {
 159                     iResult = PQsendQueryPrepared(conn, "index_nosector_places", 2, paramValues, paramLengths, paramFormats, 1);
 160                 }
 161             }
 162             else
 163             {
 164                 if (interpolation)
 165                 {
 166                                         iResult = PQsendQueryPrepared(conn, "index_sector_places_osmline", 2, paramValues, paramLengths, paramFormats, 1);
 167
 168                 }
 169                 else
 170                 {
 171                     iResult = PQsendQueryPrepared(conn, "index_sector_places", 2, paramValues, paramLengths, paramFormats, 1);
 172                 }
 173             }
 174             if (!iResult)
 175             {
 176                 fprintf(stderr, "index_sector_places: SELECT failed: %s", PQerrorMessage(conn));
 177                 PQclear(resPlaces);
 178                 exit(EXIT_FAILURE);
 179             }
 180         }
 181         if (iSector > 0)
 182         {
 183             count = 0;
 184             rankPerSecond = 0;
 185             tuples = PQntuples(resPlaces);
 186
 187             if (tuples > 0)
 188             {
 189                 // Spawn threads
 190                 for (i = 0; i < num_threads; i++)
 191                 {
 192                     thread_data[i].res = resPlaces;
 193                     thread_data[i].tuples = tuples;
 194                     thread_data[i].count = &count;
 195                     thread_data[i].count_mutex = &count_mutex;
 196                     thread_data[i].writer = writer;
 197                     thread_data[i].writer_mutex = &writer_mutex;
 198                     if (interpolation)
 199                     {
 200                         thread_data[i].table = 0;  // use interpolations table
 201                     }
 202                     else
 203                     {
 204                         thread_data[i].table = 1;  // use placex table
 205                     }
 206                     pthread_create(&thread_data[i].thread, NULL, &nominatim_indexThread, (void *)&thread_data[i]);
 207                 }
 208
 209                 // Monitor threads to give user feedback
 210                 sleepcount = 0;
 211                 while (count < tuples)
 212                 {
 213                     usleep(1000);
 214
 215                     // Aim for one update per second
 216                     if (sleepcount++ > 500)
 217                     {
 218                         rankPerSecond = ((float)rankCountTuples + (float)count) / MAX(difftime(time(0), rankStartTime),1);
 219                         if(interpolation)
 220                         {
 221                             fprintf(stderr, "  Done %i in %i @ %f per second - Interpolation lines ETA (seconds): %f\n", (rankCountTuples + count), (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - (rankCountTuples + count)))/rankPerSecond);
 222                         }
 223                         else
 224                         {
 225                             fprintf(stderr, "  Done %i in %i @ %f per second - Rank %i ETA (seconds): %f\n", (rankCountTuples + count), (int)(difftime(time(0), rankStartTime)), rankPerSecond, rank, ((float)(rankTotalTuples - (rankCountTuples + count)))/rankPerSecond);
 226                         }
 227
 228                         sleepcount = 0;
 229                     }
 230                 }
 231
 232                 // Wait for everything to finish
 233                 for (i = 0; i < num_threads; i++)
 234                 {
 235                     pthread_join(thread_data[i].thread, NULL);
 236                 }
 237
 238                 rankCountTuples += tuples;
 239             }
 240
 241             // Finished sector
 242             rankPerSecond = (float)rankCountTuples / MAX(difftime(time(0), rankStartTime),1);
 243             fprintf(stderr, "  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);
 244
 245             PQclear(resPlaces);
 246         }
 247         if (rankTotalTuples-rankCountTuples < num_threads*20 && iSector < PQntuples(resSectors))
 248         {
 249             iSector = PQntuples(resSectors) - 1;
 250         }
 251     }
 252     // Finished rank
 253     fprintf(stderr, "\r  Done %i in %i @ %f per second - FINISHED\n\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond);
 254
 255     PQclear(resSectors);
 256 }
 257
 258 void nominatim_index(int rank_min, int rank_max, int num_threads, const char *conninfo, const char *structuredoutputfile)
 259 {
 260     struct index_thread_data * thread_data;
 261
 262     PGconn *conn;
 263     PGresult * res;
 264
 265     int rank;
 266
 267     int i;
 268
 269     xmlTextWriterPtr writer;
 270     pthread_mutex_t writer_mutex = PTHREAD_MUTEX_INITIALIZER;
 271
 272     Oid pg_prepare_params[2];
 273
 274     conn = PQconnectdb(conninfo);
 275     if (PQstatus(conn) != CONNECTION_OK)
 276     {
 277         fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn));
 278         exit(EXIT_FAILURE);
 279     }
 280
 281     pg_prepare_params[0] = PG_OID_INT4;
 282     res = PQprepare(conn, "index_sectors",
 283                     "select geometry_sector,count(*) from placex where rank_search = $1 and indexed_status > 0 group by geometry_sector order by geometry_sector",
 284                     1, pg_prepare_params);
 285     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 286     {
 287         fprintf(stderr, "Failed preparing index_sectors: %s\n", PQerrorMessage(conn));
 288         exit(EXIT_FAILURE);
 289     }
 290     PQclear(res);
 291
 292     res = PQprepare(conn, "index_sectors_osmline",
 293                     "select geometry_sector,count(*) from location_property_osmline where indexed_status > 0 group by geometry_sector order by geometry_sector",
 294                     0, NULL);
 295     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 296     {
 297         fprintf(stderr, "Failed preparing index_sectors: %s\n", PQerrorMessage(conn));
 298         exit(EXIT_FAILURE);
 299     }
 300     PQclear(res);
 301
 302     pg_prepare_params[0] = PG_OID_INT4;
 303     res = PQprepare(conn, "index_nosectors",
 304                     "select 0::integer,count(*) from placex where rank_search = $1 and indexed_status > 0",
 305                     1, pg_prepare_params);
 306     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 307     {
 308         fprintf(stderr, "Failed preparing index_sectors: %s\n", PQerrorMessage(conn));
 309         exit(EXIT_FAILURE);
 310     }
 311     PQclear(res);
 312
 313     pg_prepare_params[0] = PG_OID_INT4;
 314     pg_prepare_params[1] = PG_OID_INT4;
 315     res = PQprepare(conn, "index_sector_places",
 316                     "select place_id from placex where rank_search = $1 and geometry_sector = $2 and indexed_status > 0",
 317                     2, pg_prepare_params);
 318     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 319     {
 320         fprintf(stderr, "Failed preparing index_sector_places: %s\n", PQerrorMessage(conn));
 321         exit(EXIT_FAILURE);
 322     }
 323     PQclear(res);
 324
 325     pg_prepare_params[0] = PG_OID_INT4;
 326     res = PQprepare(conn, "index_nosector_places",
 327                     "select place_id from placex where rank_search = $1 and indexed_status > 0 order by geometry_sector",
 328                     1, pg_prepare_params);
 329     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 330     {
 331         fprintf(stderr, "Failed preparing index_nosector_places: %s\n", PQerrorMessage(conn));
 332         exit(EXIT_FAILURE);
 333     }
 334     PQclear(res);
 335
 336     pg_prepare_params[0] = PG_OID_INT4;
 337     res = PQprepare(conn, "index_sector_places_osmline",
 338                     "select place_id from location_property_osmline where geometry_sector = $2 and indexed_status > 0",
 339                     1, pg_prepare_params);
 340     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 341     {
 342         fprintf(stderr, "Failed preparing index_sector_places: %s\n", PQerrorMessage(conn));
 343         exit(EXIT_FAILURE);
 344     }
 345     PQclear(res);
 346
 347     res = PQprepare(conn, "index_nosector_places_osmline",
 348                     "select place_id from location_property_osmline where indexed_status > 0 order by geometry_sector",
 349                     0, NULL);
 350     if (PQresultStatus(res) != PGRES_COMMAND_OK)
 351     {
 352         fprintf(stderr, "Failed preparing index_nosector_places: %s\n", PQerrorMessage(conn));
 353         exit(EXIT_FAILURE);
 354     }
 355     PQclear(res);
 356
 357     // Build the data for each thread
 358     thread_data = (struct index_thread_data *)malloc(sizeof(struct index_thread_data)*num_threads);
 359     for (i = 0; i < num_threads; i++)
 360     {
 361         thread_data[i].conn = PQconnectdb(conninfo);
 362         if (PQstatus(thread_data[i].conn) != CONNECTION_OK)
 363         {
 364             fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(thread_data[i].conn));
 365             exit(EXIT_FAILURE);
 366         }
 367
 368         pg_prepare_params[0] = PG_OID_INT8;
 369         res = PQprepare(thread_data[i].conn, "index_placex",
 370                         "update placex set indexed_status = 0 where place_id = $1",
 371                         1, pg_prepare_params);
 372         if (PQresultStatus(res) != PGRES_COMMAND_OK)
 373         {
 374             fprintf(stderr, "Failed preparing index_placex: %s\n", PQerrorMessage(conn));
 375             exit(EXIT_FAILURE);
 376         }
 377         PQclear(res);
 378
 379         pg_prepare_params[0] = PG_OID_INT8;
 380         res = PQprepare(thread_data[i].conn, "index_osmline",
 381                         "update location_property_osmline set indexed_status = 0 where place_id = $1",
 382                         1, pg_prepare_params);
 383         if (PQresultStatus(res) != PGRES_COMMAND_OK)
 384         {
 385             fprintf(stderr, "Failed preparing index_osmline: %s\n", PQerrorMessage(conn));
 386             exit(EXIT_FAILURE);
 387         }
 388         PQclear(res);
 389
 390         /*res = PQexec(thread_data[i].conn, "set enable_seqscan = false");
 391         if (PQresultStatus(res) != PGRES_COMMAND_OK)
 392         {
 393             fprintf(stderr, "Failed disabling sequential scan: %s\n", PQerrorMessage(conn));
 394             exit(EXIT_FAILURE);
 395         }
 396         PQclear(res);*/
 397
 398         nominatim_exportCreatePreparedQueries(thread_data[i].conn);
 399     }
 400
 401
 402     fprintf(stderr, "Starting indexing rank (%i to %i) using %i threads\n", rank_min, rank_max, num_threads);
 403
 404     for (rank = rank_min; rank <= rank_max; rank++)
 405     {
 406         // OSMLINE: do reindexing (=> reparenting) for interpolation lines at rank 30, but before all other objects of rank 30
 407         // reason: houses (rank 30) depend on the updated interpolation line, when reparenting (see placex_update in functions.sql)
 408         if (rank == 30)
 409         {
 410             run_indexing(rank, 1, conn, num_threads, thread_data, structuredoutputfile);
 411         }
 412         run_indexing(rank, 0, conn, num_threads, thread_data, structuredoutputfile);
 413     }
 414         // Close all connections
 415         for (i = 0; i < num_threads; i++)
 416         {
 417                 PQfinish(thread_data[i].conn);
 418         }
 419         PQfinish(conn);
 420 }
 421
 422 void *nominatim_indexThread(void * thread_data_in)
 423 {
 424     struct index_thread_data * thread_data = (struct index_thread_data * )thread_data_in;
 425     struct export_data  querySet;
 426
 427     PGresult   *res;
 428
 429     const char  *paramValues[1];
 430     int         paramLengths[1];
 431     int         paramFormats[1];
 432     uint64_t    paramPlaceID;
 433     uint64_t    place_id;
 434     time_t      updateStartTime;
 435     uint        table;
 436
 437     table = (uint)(thread_data->table);
 438
 439     while (1)
 440     {
 441         pthread_mutex_lock( thread_data->count_mutex );
 442         if (*(thread_data->count) >= thread_data->tuples)
 443         {
 444             pthread_mutex_unlock( thread_data->count_mutex );
 445             break;
 446         }
 447
 448         place_id = PGint64(*((uint64_t *)PQgetvalue(thread_data->res, *thread_data->count, 0)));
 449         (*thread_data->count)++;
 450
 451         pthread_mutex_unlock( thread_data->count_mutex );
 452
 453         if (verbose) fprintf(stderr, "  Processing place_id %ld\n", place_id);
 454
 455         updateStartTime = time(0);
 456         int done = 0;
 457
 458         if (thread_data->writer)
 459         {
 460              nominatim_exportPlaceQueries(place_id, thread_data->conn, &querySet);
 461         }
 462
 463         while(!done)
 464         {
 465             paramPlaceID = PGint64(place_id);
 466             paramValues[0] = (char *)&paramPlaceID;
 467             paramLengths[0] = sizeof(paramPlaceID);
 468             paramFormats[0] = 1;
 469             if (table == 1) // table=1 for placex
 470             {
 471                 res = PQexecPrepared(thread_data->conn, "index_placex", 1, paramValues, paramLengths, paramFormats, 1);
 472             }
 473             else // table=0 for osmline
 474             {
 475                 res = PQexecPrepared(thread_data->conn, "index_osmline", 1, paramValues, paramLengths, paramFormats, 1);
 476             }
 477             if (PQresultStatus(res) == PGRES_COMMAND_OK)
 478                 done = 1;
 479             else
 480             {
 481                 if (!strncmp(PQerrorMessage(thread_data->conn), "ERROR:  deadlock detected", 25))
 482                 {
 483                     if (table == 1)
 484                     {
 485                         fprintf(stderr, "index_placex: UPDATE failed - deadlock, retrying (%ld)\n", place_id);
 486                     }
 487                     else
 488                     {
 489                         fprintf(stderr, "index_osmline: UPDATE failed - deadlock, retrying (%ld)\n", place_id);
 490                     }
 491                     PQclear(res);
 492                     sleep(rand() % 10);
 493                 }
 494                 else
 495                 {
 496                     if (table == 1)
 497                     {
 498                         fprintf(stderr, "index_placex: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
 499                     }
 500                     else
 501                     {
 502                         fprintf(stderr, "index_osmline: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
 503                     }
 504                     PQclear(res);
 505                     exit(EXIT_FAILURE);
 506                 }
 507             }
 508         }
 509         PQclear(res);
 510         if (difftime(time(0), updateStartTime) > 1) fprintf(stderr, "  Slow place_id %ld\n", place_id);
 511
 512         if (thread_data->writer)
 513         {
 514             nominatim_exportPlace(place_id, thread_data->conn, thread_data->writer, thread_data->writer_mutex, &querySet);
 515             nominatim_exportFreeQueries(&querySet);
 516         }
 517     }
 518
 519     return NULL;
 520 }