nominatim/index.c

   1 /*
   2 */
   3
   4 #include <stdio.h>
   5 #include <unistd.h>
   6 #include <stdlib.h>
   7 #include <string.h>
   8 #include <assert.h>
   9 #include <pthread.h>
  10 #include <time.h>
  11 #include <stdint.h>
  12
  13 #include <libpq-fe.h>
  14
  15 #include "nominatim.h"
  16 #include "index.h"
  17 #include "export.h"
  18 #include "postgresql.h"
  19
  20 extern int verbose;
  21
  22 void nominatim_index(int rank_min, int rank_max, int num_threads, const char *conninfo, const char *structuredoutputfile)
  23 {
  24         struct index_thread_data * thread_data;
  25         pthread_mutex_t count_mutex = PTHREAD_MUTEX_INITIALIZER;
  26         int tuples, count, sleepcount;
  27
  28         time_t rankStartTime;
  29         int rankTotalTuples;
  30         int rankCountTuples;
  31         float rankPerSecond;
  32
  33         PGconn *conn;
  34         PGresult * res;
  35         PGresult * resSectors;
  36         PGresult * resPlaces;
  37
  38         int rank;
  39         int i;
  40         int iSector;
  41
  42     const char *paramValues[2];
  43     int         paramLengths[2];
  44     int         paramFormats[2];
  45     uint32_t    paramRank;
  46     uint32_t    paramSector;
  47     uint32_t    sector;
  48
  49     xmlTextWriterPtr writer;
  50         pthread_mutex_t writer_mutex = PTHREAD_MUTEX_INITIALIZER;
  51
  52     Oid pg_prepare_params[2];
  53
  54         conn = PQconnectdb(conninfo);
  55     if (PQstatus(conn) != CONNECTION_OK) {
  56         fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(conn));
  57         exit(EXIT_FAILURE);
  58     }
  59
  60     pg_prepare_params[0] = PG_OID_INT4;
  61     res = PQprepare(conn, "index_sectors",
  62         "select geometry_sector,count(*) from placex where rank_search = $1 and indexed = false and name is not null group by geometry_sector order by geometry_sector",
  63         1, pg_prepare_params);
  64     if (PQresultStatus(res) != PGRES_COMMAND_OK) exit(EXIT_FAILURE);
  65     PQclear(res);
  66
  67     pg_prepare_params[0] = PG_OID_INT4;
  68     pg_prepare_params[1] = PG_OID_INT4;
  69     res = PQprepare(conn, "index_sector_places",
  70         "select place_id from placex where rank_search = $1 and geometry_index(geometry,indexed,name) = $2",
  71         2, pg_prepare_params);
  72     if (PQresultStatus(res) != PGRES_COMMAND_OK) exit(EXIT_FAILURE);
  73     PQclear(res);
  74
  75     // Build the data for each thread
  76     thread_data = (struct index_thread_data *)malloc(sizeof(struct index_thread_data)*num_threads);
  77         for (i = 0; i < num_threads; i++)
  78         {
  79                 thread_data[i].conn = PQconnectdb(conninfo);
  80             if (PQstatus(thread_data[i].conn) != CONNECTION_OK) {
  81                 fprintf(stderr, "Connection to database failed: %s\n", PQerrorMessage(thread_data[i].conn));
  82                 exit(EXIT_FAILURE);
  83             }
  84
  85             pg_prepare_params[0] = PG_OID_INT8;
  86             res = PQprepare(thread_data[i].conn, "index_placex",
  87                 "update placex set indexed = true where place_id = $1",
  88                 1, pg_prepare_params);
  89             if (PQresultStatus(res) != PGRES_COMMAND_OK) exit(EXIT_FAILURE);
  90             PQclear(res);
  91
  92             nominatim_exportCreatePreparedQueries(thread_data[i].conn);
  93         }
  94
  95         // Create the output file
  96         writer = NULL;
  97         if (structuredoutputfile)
  98         {
  99                 writer = nominatim_exportXMLStart(structuredoutputfile);
 100         }
 101
 102     fprintf(stderr, "Starting indexing rank (%i > %i ) using %i treads\n", rank_min, rank_max, num_threads);
 103
 104     for (rank = rank_min; rank <= rank_max; rank++)
 105     {
 106         printf("Starting rank %d\n", rank);
 107         rankStartTime = time(0);
 108         rankCountTuples = 0;
 109         rankPerSecond = 0;
 110
 111         paramRank = PGint32(rank);
 112         paramValues[0] = (char *)&paramRank;
 113         paramLengths[0] = sizeof(paramRank);
 114         paramFormats[0] = 1;
 115         resSectors = PQexecPrepared(conn, "index_sectors", 1, paramValues, paramLengths, paramFormats, 1);
 116         if (PQresultStatus(resSectors) != PGRES_TUPLES_OK)
 117         {
 118             fprintf(stderr, "index_sectors: SELECT failed: %s", PQerrorMessage(conn));
 119             PQclear(resSectors);
 120             exit(EXIT_FAILURE);
 121         }
 122                 if (PQftype(resSectors, 0) != PG_OID_INT4)
 123                 {
 124             fprintf(stderr, "Sector value has unexpected type\n");
 125             PQclear(resSectors);
 126             exit(EXIT_FAILURE);
 127                 }
 128                 if (PQftype(resSectors, 1) != PG_OID_INT8)
 129                 {
 130             fprintf(stderr, "Sector value has unexpected type\n");
 131             PQclear(resSectors);
 132             exit(EXIT_FAILURE);
 133                 }
 134
 135                 rankTotalTuples = 0;
 136         for (iSector = 0; iSector < PQntuples(resSectors); iSector++)
 137         {
 138                 rankTotalTuples += PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1)));
 139         }
 140
 141         for (iSector = 0; iSector < PQntuples(resSectors); iSector++)
 142         {
 143                         sector = PGint32(*((uint32_t *)PQgetvalue(resSectors, iSector, 0)));
 144                 //printf("\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));
 145
 146                         // Get all the place_id's for this sector
 147                 paramRank = PGint32(rank);
 148                 paramValues[0] = (char *)&paramRank;
 149                 paramLengths[0] = sizeof(paramRank);
 150                 paramFormats[0] = 1;
 151                 paramSector = PGint32(sector);
 152                 paramValues[1] = (char *)&paramSector;
 153                 paramLengths[1] = sizeof(paramSector);
 154                 paramFormats[1] = 1;
 155                 resPlaces = PQexecPrepared(conn, "index_sector_places", 2, paramValues, paramLengths, paramFormats, 1);
 156                 if (PQresultStatus(resPlaces) != PGRES_TUPLES_OK)
 157                 {
 158                     fprintf(stderr, "index_sector_places: SELECT failed: %s", PQerrorMessage(conn));
 159                     PQclear(resPlaces);
 160                     exit(EXIT_FAILURE);
 161                 }
 162                         if (PQftype(resPlaces, 0) != PG_OID_INT8)
 163                         {
 164                     fprintf(stderr, "Place_id value has unexpected type\n");
 165                     PQclear(resPlaces);
 166                     exit(EXIT_FAILURE);
 167                         }
 168
 169                         count = 0;
 170                         rankPerSecond = 0;
 171                         tuples = PQntuples(resPlaces);
 172
 173                         if (tuples > 0)
 174                 {
 175                                 // Spawn threads
 176                                 for (i = 0; i < num_threads; i++)
 177                                 {
 178                                         thread_data[i].res = resPlaces;
 179                                         thread_data[i].tuples = tuples;
 180                                         thread_data[i].count = &count;
 181                                         thread_data[i].count_mutex = &count_mutex;
 182                                         thread_data[i].writer = writer;
 183                                         thread_data[i].writer_mutex = &writer_mutex;
 184                                         pthread_create(&thread_data[i].thread, NULL, &nominatim_indexThread, (void *)&thread_data[i]);
 185                                 }
 186
 187                                 // Monitor threads to give user feedback
 188                                 sleepcount = 0;
 189                                 while(count < tuples)
 190                                 {
 191                                         usleep(1000);
 192
 193                                         // Aim for one update per second
 194                                         if (sleepcount++ > 500)
 195                                         {
 196                                                 rankPerSecond = ((float)rankCountTuples + (float)count) / MAX(difftime(time(0), rankStartTime),1);
 197                                                 printf("  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);
 198                                                 sleepcount = 0;
 199                                         }
 200                                 }
 201
 202                                 // Wait for everything to finish
 203                                 for (i = 0; i < num_threads; i++)
 204                                 {
 205                                         pthread_join(thread_data[i].thread, NULL);
 206                                 }
 207
 208                                 rankCountTuples += tuples;
 209                 }
 210
 211                         // Finished sector
 212                         rankPerSecond = (float)rankCountTuples / MAX(difftime(time(0), rankStartTime),1);
 213                         printf("  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);
 214
 215             PQclear(resPlaces);
 216
 217         }
 218         // Finished rank
 219                 printf("\r  Done %i in %i @ %f per second - FINISHED                      \n\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond);
 220
 221         PQclear(resSectors);
 222     }
 223
 224     if (writer)
 225     {
 226         nominatim_exportXMLEnd(writer);
 227     }
 228 }
 229
 230 void *nominatim_indexThread(void * thread_data_in)
 231 {
 232         struct index_thread_data * thread_data = (struct index_thread_data * )thread_data_in;
 233
 234         PGresult * res;
 235
 236     const char *paramValues[1];
 237     int         paramLengths[1];
 238     int         paramFormats[1];
 239     uint64_t    paramPlaceID;
 240     uint64_t place_id;
 241
 242         while(1)
 243         {
 244                 pthread_mutex_lock( thread_data->count_mutex );
 245                 if (*(thread_data->count) >= thread_data->tuples)
 246                 {
 247                         pthread_mutex_unlock( thread_data->count_mutex );
 248                         break;
 249                 }
 250
 251                 place_id = PGint64(*((uint64_t *)PQgetvalue(thread_data->res, *thread_data->count, 0)));
 252                 (*thread_data->count)++;
 253
 254                 pthread_mutex_unlock( thread_data->count_mutex );
 255
 256                 //printf("  Processing place_id %ld\n", place_id);
 257                 paramPlaceID = PGint64(place_id);
 258         paramValues[0] = (char *)&paramPlaceID;
 259         paramLengths[0] = sizeof(paramPlaceID);
 260         paramFormats[0] = 1;
 261         res = PQexecPrepared(thread_data->conn, "index_placex", 1, paramValues, paramLengths, paramFormats, 1);
 262         if (PQresultStatus(res) != PGRES_COMMAND_OK)
 263         {
 264             fprintf(stderr, "index_placex: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
 265             PQclear(res);
 266             exit(EXIT_FAILURE);
 267         }
 268         PQclear(res);
 269
 270         if (thread_data->writer)
 271         {
 272                 nominatim_exportPlace(place_id, thread_data->conn, thread_data->writer, thread_data->writer_mutex);
 273         }
 274         }
 275
 276         return NULL;
 277 }