Merge remote-tracking branch 'upstream/master'

[nominatim.git] / nominatim / index.c

diff --git a/nominatim/index.c b/nominatim/index.c
index 55f3c578f55b73b0aaadb59d81e8686dbb46e9b4..13f292737082f5a4185c45086b6ff90d265c91eb 100644 (file)
--- a/nominatim/index.c
+++ b/nominatim/index.c
@@ -40,7 +40,6 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
     int i;
     int iSector;
     int iResult;
     int i;
     int iSector;
     int iResult;

-    int bSkip;

 
     const char *paramValues[2];
     int         paramLengths[2];
 
     const char *paramValues[2];
     int         paramLengths[2];


@@ -117,7 +116,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             exit(EXIT_FAILURE);
         }
 
             exit(EXIT_FAILURE);
         }
 

-        pg_prepare_params[0] = PG_OID_INT4;
+        pg_prepare_params[0] = PG_OID_INT8;

         res = PQprepare(thread_data[i].conn, "index_placex",
                         "update placex set indexed_status = 0 where place_id = $1",
                         1, pg_prepare_params);
         res = PQprepare(thread_data[i].conn, "index_placex",
                         "update placex set indexed_status = 0 where place_id = $1",
                         1, pg_prepare_params);


@@ -128,13 +127,13 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         }
         PQclear(res);
 
         }
         PQclear(res);
 

-        res = PQexec(thread_data[i].conn, "set enable_seqscan = false");
+        /*res = PQexec(thread_data[i].conn, "set enable_seqscan = false");

         if (PQresultStatus(res) != PGRES_COMMAND_OK)
         {
             fprintf(stderr, "Failed disabling sequential scan: %s\n", PQerrorMessage(conn));
             exit(EXIT_FAILURE);
         }
         if (PQresultStatus(res) != PGRES_COMMAND_OK)
         {
             fprintf(stderr, "Failed disabling sequential scan: %s\n", PQerrorMessage(conn));
             exit(EXIT_FAILURE);
         }

-        PQclear(res);
+        PQclear(res);*/

 
         nominatim_exportCreatePreparedQueries(thread_data[i].conn);
     }
 
         nominatim_exportCreatePreparedQueries(thread_data[i].conn);
     }


@@ -146,11 +145,11 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
         writer = nominatim_exportXMLStart(structuredoutputfile);
     }
 
         writer = nominatim_exportXMLStart(structuredoutputfile);
     }
 

-    fprintf(stderr, "Starting indexing rank (%i to %i) using %i treads\n", rank_min, rank_max, num_threads);
+    fprintf(stderr, "Starting indexing rank (%i to %i) using %i threads\n", rank_min, rank_max, num_threads);

 
     for (rank = rank_min; rank <= rank_max; rank++)
     {
 
     for (rank = rank_min; rank <= rank_max; rank++)
     {

-        printf("Starting rank %d\n", rank);
+        fprintf(stderr, "Starting rank %d\n", rank);

         rankCountTuples = 0;
         rankPerSecond = 0;
 
         rankCountTuples = 0;
         rankPerSecond = 0;
 


@@ -200,7 +199,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
                     PQclear(resPlaces);
                     exit(EXIT_FAILURE);
                 }
                     PQclear(resPlaces);
                     exit(EXIT_FAILURE);
                 }

-                if (PQftype(resPlaces, 0) != PG_OID_INT4)
+                if (PQftype(resPlaces, 0) != PG_OID_INT8)

                 {
                     fprintf(stderr, "Place_id value has unexpected type\n");
                     PQclear(resPlaces);
                 {
                     fprintf(stderr, "Place_id value has unexpected type\n");
                     PQclear(resPlaces);


@@ -217,7 +216,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             if (iSector < PQntuples(resSectors))
             {
                 sector = PGint32(*((uint32_t *)PQgetvalue(resSectors, iSector, 0)));
             if (iSector < PQntuples(resSectors))
             {
                 sector = PGint32(*((uint32_t *)PQgetvalue(resSectors, iSector, 0)));

-//                printf("\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));
+//                fprintf(stderr, "\n Starting sector %d size %ld\n", sector, PGint64(*((uint64_t *)PQgetvalue(resSectors, iSector, 1))));

 
                 // Get all the place_id's for this sector
                 paramRank = PGint32(rank);
 
                 // Get all the place_id's for this sector
                 paramRank = PGint32(rank);


@@ -274,7 +273,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
                         if (sleepcount++ > 500)
                         {
                             rankPerSecond = ((float)rankCountTuples + (float)count) / MAX(difftime(time(0), rankStartTime),1);
                         if (sleepcount++ > 500)
                         {
                             rankPerSecond = ((float)rankCountTuples + (float)count) / MAX(difftime(time(0), rankStartTime),1);

-                            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);
+                            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);

                             sleepcount = 0;
                         }
                     }
                             sleepcount = 0;
                         }
                     }


@@ -290,7 +289,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
 
                 // Finished sector
                 rankPerSecond = (float)rankCountTuples / MAX(difftime(time(0), rankStartTime),1);
 
                 // Finished sector
                 rankPerSecond = (float)rankCountTuples / MAX(difftime(time(0), rankStartTime),1);

-                printf("  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);
+                fprintf(stderr, "  Done %i in %i @ %f per second - ETA (seconds): %f\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond, ((float)(rankTotalTuples - rankCountTuples))/rankPerSecond);

 
                 PQclear(resPlaces);
             }
 
                 PQclear(resPlaces);
             }


@@ -300,7 +299,7 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
             }
         }
         // Finished rank
             }
         }
         // Finished rank

-        printf("\r  Done %i in %i @ %f per second - FINISHED                      \n\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond);
+        fprintf(stderr, "\r  Done %i in %i @ %f per second - FINISHED                      \n\n", rankCountTuples, (int)(difftime(time(0), rankStartTime)), rankPerSecond);

 
         PQclear(resSectors);
     }
 
         PQclear(resSectors);
     }


@@ -309,6 +308,13 @@ void nominatim_index(int rank_min, int rank_max, int num_threads, const char *co
     {
         nominatim_exportXMLEnd(writer);
     }
     {
         nominatim_exportXMLEnd(writer);
     }

+
+    // Close all connections
+    for (i = 0; i < num_threads; i++)
+    {
+        PQfinish(thread_data[i].conn);
+    }
+    PQfinish(conn);

 }
 
 void *nominatim_indexThread(void * thread_data_in)
 }
 
 void *nominatim_indexThread(void * thread_data_in)


@@ -321,8 +327,8 @@ void *nominatim_indexThread(void * thread_data_in)
     const char *paramValues[1];
     int         paramLengths[1];
     int         paramFormats[1];
     const char *paramValues[1];
     int         paramLengths[1];
     int         paramFormats[1];

-    uint32_t    paramPlaceID;
-    uint32_t   place_id;
+    uint64_t    paramPlaceID;
+    uint64_t   place_id;

     time_t             updateStartTime;
 
     while (1)
     time_t             updateStartTime;
 
     while (1)


@@ -334,12 +340,12 @@ void *nominatim_indexThread(void * thread_data_in)
             break;
         }
 
             break;
         }
 

-        place_id = PGint32(*((uint32_t *)PQgetvalue(thread_data->res, *thread_data->count, 0)));
+        place_id = PGint64(*((uint64_t *)PQgetvalue(thread_data->res, *thread_data->count, 0)));

         (*thread_data->count)++;
 
         pthread_mutex_unlock( thread_data->count_mutex );
 
         (*thread_data->count)++;
 
         pthread_mutex_unlock( thread_data->count_mutex );
 

-        if (verbose) printf("  Processing place_id %d\n", place_id);
+        if (verbose) fprintf(stderr, "  Processing place_id %ld\n", place_id);

 
         updateStartTime = time(0);
        int done = 0;
 
         updateStartTime = time(0);
        int done = 0;


@@ -351,8 +357,7 @@ void *nominatim_indexThread(void * thread_data_in)
 
        while(!done)
        {
 
        while(!done)
        {

-
-               paramPlaceID = PGint32(place_id);
+               paramPlaceID = PGint64(place_id);

                paramValues[0] = (char *)&paramPlaceID;
                paramLengths[0] = sizeof(paramPlaceID);
                paramFormats[0] = 1;
                paramValues[0] = (char *)&paramPlaceID;
                paramLengths[0] = sizeof(paramPlaceID);
                paramFormats[0] = 1;


@@ -361,9 +366,9 @@ void *nominatim_indexThread(void * thread_data_in)
                        done = 1;
                else
                {
                        done = 1;
                else
                {

-                       if (strncmp(PQerrorMessage(thread_data->conn), "ERROR:  deadlock detected", 25))
+                       if (!strncmp(PQerrorMessage(thread_data->conn), "ERROR:  deadlock detected", 25))

                        {
                        {

-                           fprintf(stderr, "index_placex: UPDATE failed - deadlock, retrying\n");
+                           fprintf(stderr, "index_placex: UPDATE failed - deadlock, retrying (%ld)\n", place_id);

                            PQclear(res);
                             sleep(rand() % 10);
                        }
                            PQclear(res);
                             sleep(rand() % 10);
                        }


@@ -371,13 +376,12 @@ void *nominatim_indexThread(void * thread_data_in)
                        {
                            fprintf(stderr, "index_placex: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
                            PQclear(res);
                        {
                            fprintf(stderr, "index_placex: UPDATE failed: %s", PQerrorMessage(thread_data->conn));
                            PQclear(res);

-                            sleep(rand() % 10);
-//                         exit(EXIT_FAILURE);
+                           exit(EXIT_FAILURE);

                        }
                }
         }
         PQclear(res);
                        }
                }
         }
         PQclear(res);

-        if (difftime(time(0), updateStartTime) > 1) printf("  Slow place_id %d\n", place_id);
+        if (difftime(time(0), updateStartTime) > 1) fprintf(stderr, "  Slow place_id %ld\n", place_id);

 
         if (thread_data->writer)
         {
 
         if (thread_data->writer)
         {