Refactor:maptool:Compress the tiles of a map in parallel

Phase 13 writes the zip file and needs 58 to 66 percent of the time of a
conversion. Measurements on five country maps show that the phase does
almost nothing but deflate: it moves 5.6 to 7.3 MB/s, and an isolated
benchmark of zlib level 9 on real tile data gives 6.0 MB/s.

The tiles of a slice do not depend on each other, and process_slice()
already holds the whole slice in memory. This change compresses
thread_count tiles at the same time and then writes them in the order of
the list.

write_zipmember() is now two functions. zip_compress_member() computes
the CRC and the compressed data. It only reads zip_info, so threads can
call it at the same time. write_zipmember_compressed() writes the result
and must keep the order of the members, because the position of a member
in the file depends on the members before it.

The output does not change. The order of the members and the parameters
of deflate stay the same. A run with one thread gives a file that is
identical to the file of the version before this change, byte for byte.
A run with 8 or 20 threads gives a file in which every member has a
correct CRC.

A batch holds no more than thread_count tiles, so the memory for the
compressed data stays small against the memory for the slice.

Measurement of the compression of a map of Luxembourg, 117.8 MB of tile
data on a machine with 20 cores:

    threads    time      factor
          1    18.4 s      1.00
          4     8.1 s      2.24
          8     4.9 s      3.55
         20     3.0 s      6.03

Author: pgrandin

navit/maptool/maptool.h

@@ -412,7 +412,32 @@ void index_init(struct zip_info *info, int version);
 void index_submap_add(struct tile_info *info, struct tile_head *th);
 
 /* zip.c */
+
+/**
+ * One zip member after the compression and before the write.
+ *
+ * zip_compress_member() fills this struct and write_zipmember_compressed()
+ * writes it. The two steps are separate because the compression of different
+ * members can run at the same time, but the write cannot.
+ */
+struct zip_member {
+    /** The data to write. This is buffer, or the input when deflate did not help. */
+    char *data;
+    /** Size of data in bytes. */
+    int data_size;
+    /** Size of the member before the compression. */
+    int uncomp_size;
+    /** CRC-32 of the data before the compression. */
+    int crc;
+    /** 8 for deflate, 0 for no compression. */
+    int method;
+    /** The buffer for the compressed data. write_zipmember_compressed() frees it. */
+    char *buffer;
+};
+
 void write_zipmember(struct zip_info *zip_info, char *name, int filelen, char *data, int data_size);
+void zip_compress_member(struct zip_info *zip_info, char *data, int data_size, struct zip_member *m);
+void write_zipmember_compressed(struct zip_info *zip_info, char *name, int filelen, struct zip_member *m);
 int zip_write_index(struct zip_info *info);
 int zip_write_directory(struct zip_info *info);
 struct zip_info *zip_new(void);

navit/maptool/misc.c

@@ -300,11 +300,92 @@ int phase4(FILE **in, int in_count, int with_range, char *suffix, FILE *tilesdir
     return phase34(&info, zip_info, in, NULL, in_count, with_range);
 }
 
+/** One member of a batch that process_slice() compresses in parallel. */
+struct slice_compress_job {
+    struct zip_info *zip_info;
+    struct tile_head *th;
+    struct zip_member member;
+    GThread *thread;
+};
+
+static gpointer slice_compress_worker(gpointer data) {
+    struct slice_compress_job *job = data;
+    zip_compress_member(job->zip_info, job->th->zip_data, job->th->total_size, &job->member);
+    return NULL;
+}
+
+/**
+ * @brief Compresses and writes the tiles of one slice.
+ *
+ * The compression is the largest part of this phase and one member does not
+ * depend on another member, so this function compresses thread_count members at
+ * the same time. It then writes the members of the batch in the order of the
+ * list, because the position of a member in the zip file depends on the members
+ * before it.
+ *
+ * The result is the same file that a sequential run gives. The order of the
+ * members and the parameters of deflate do not change.
+ *
+ * A batch holds no more than thread_count members. The memory that the
+ * compressed data needs is therefore small against slice_data, which already
+ * holds the whole slice.
+ *
+ * @param zip_info the zip file that receives the tiles
+ * @param tiles the tiles of the slice, in the order of the write
+ * @param count the number of tiles
+ *
+ * @returns the number of tiles that went into the zip file
+ */
+static int write_tiles(struct zip_info *zip_info, struct tile_head **tiles, int count) {
+    /* thread_count comes from the command line, so do not trust it for the
+     * stride of the loop below. A stride of 0 would never end. */
+    int batch_max = thread_count > 0 ? thread_count : 1;
+    struct slice_compress_job *jobs = g_malloc0(sizeof(struct slice_compress_job) * batch_max);
+    int maxnamelen = zip_get_maxnamelen(zip_info);
+    gint64 start = g_get_monotonic_time();
+    long long bytes = 0;
+    int zipfiles = 0;
+    int first, i;
+
+    for (first = 0; first < count; first += batch_max) {
+        int batch = MIN(batch_max, count - first);
+        for (i = 0; i < batch; i++) {
+            jobs[i].zip_info = zip_info;
+            jobs[i].th = tiles[first + i];
+            jobs[i].thread = NULL;
+            /* A tile without a name goes to the index and needs no compression. */
+            if (jobs[i].th->name[0])
+                jobs[i].thread = g_thread_new("zip_compress_worker", slice_compress_worker, &(jobs[i]));
+        }
+        for (i = 0; i < batch; i++) {
+            if (jobs[i].thread)
+                g_thread_join(jobs[i].thread);
+        }
+        for (i = 0; i < batch; i++) {
+            struct tile_head *th = jobs[i].th;
+            if (jobs[i].thread) {
+                bytes += th->total_size;
+                write_zipmember_compressed(zip_info, th->name, maxnamelen, &(jobs[i].member));
+                zipfiles++;
+            } else {
+                dbg_assert(fwrite(th->zip_data, th->total_size, 1, zip_get_index(zip_info)) == 1);
+            }
+        }
+    }
+    g_free(jobs);
+
+    fprintf(stderr, "Compressed %d tiles, " LONGLONG_FMT " bytes in %.1f s with %d threads\n", zipfiles, bytes,
+            (g_get_monotonic_time() - start) / 1000000.0, batch_max);
+
+    return zipfiles;
+}
+
 static int process_slice(FILE **in, FILE **reference, int in_count, int with_range, long long size, char *suffix,
                          struct zip_info *zip_info) {
     struct tile_head *th;
+    struct tile_head **tiles;
     char *slice_data, *zip_data;
-    int zipfiles = 0;
+    int zipfiles = 0, tile_count = 0;
     struct tile_info info;
     int i;
 
@@ -332,20 +413,23 @@ static int process_slice(FILE **in, FILE **reference, int in_count, int with_ran
     info.tilesdir_out = NULL;
     phase34(&info, zip_info, in, reference, in_count, with_range);
 
+    for (th = tile_head_root; th; th = th->next) {
+        if (th->process)
+            tile_count++;
+    }
+    tiles = g_malloc(sizeof(struct tile_head *) * (tile_count ? tile_count : 1));
+    tile_count = 0;
     for (th = tile_head_root; th; th = th->next) {
         if (!th->process)
             continue;
-        if (th->name[0]) {
-            if (th->total_size != th->total_size_used) {
-                fprintf(stderr, "Size error '%s': %d vs %d\n", th->name, th->total_size, th->total_size_used);
-                exit(1);
-            }
-            write_zipmember(zip_info, th->name, zip_get_maxnamelen(zip_info), th->zip_data, th->total_size);
-            zipfiles++;
-        } else {
-            dbg_assert(fwrite(th->zip_data, th->total_size, 1, zip_get_index(zip_info)) == 1);
+        if (th->name[0] && th->total_size != th->total_size_used) {
+            fprintf(stderr, "Size error '%s': %d vs %d\n", th->name, th->total_size, th->total_size_used);
+            exit(1);
         }
+        tiles[tile_count++] = th;
     }
+    zipfiles = write_tiles(zip_info, tiles, tile_count);
+    g_free(tiles);
     g_free(slice_data);
 
     return zipfiles;

navit/maptool/zip.c

@@ -81,7 +81,60 @@ static int compress2_int(Byte *dest, uLongf *destLen, const Bytef *source, uLong
 }
 #endif
 
-void write_zipmember(struct zip_info *zip_info, char *name, int filelen, char *data, int data_size) {
+/**
+ * @brief Compresses the data of one zip member.
+ *
+ * This function reads zip_info but does not change it. Several threads can
+ * therefore compress different members at the same time. The caller must then
+ * give the results to write_zipmember_compressed() in the order of the members,
+ * because the position of a member in the file depends on the members before it.
+ *
+ * @param zip_info the zip file that receives the member
+ * @param data the data of the member
+ * @param data_size the size of data in bytes
+ * @param m receives the result. Give it to write_zipmember_compressed().
+ */
+void zip_compress_member(struct zip_info *zip_info, char *data, int data_size, struct zip_member *m) {
+    m->data = data;
+    m->data_size = data_size;
+    m->uncomp_size = data_size;
+    m->buffer = NULL;
+    m->crc = crc32(crc32(0, NULL, 0), (unsigned char *)data, data_size);
+    m->method = zip_info->compression_level ? 8 : 0;
+#ifdef HAVE_ZLIB
+    if (zip_info->compression_level) {
+        uLongf destlen = data_size + data_size / 500 + 12;
+        int error;
+        m->buffer = g_malloc(destlen);
+        error = compress2_int((Byte *)m->buffer, &destlen, (Bytef *)data, data_size, zip_info->compression_level);
+        if (error == Z_OK) {
+            if (destlen < data_size) {
+                m->data = m->buffer;
+                m->data_size = destlen;
+            } else
+                m->method = 0;
+        } else {
+            /* Note that this keeps method 8 for data that is not compressed.
+             * The behavior is the same as before the members were compressed in
+             * parallel. deflate needs more room than the buffer only for data
+             * that grows, and tile data always becomes smaller. */
+            fprintf(stderr, "compress2 returned %d\n", error);
+        }
+    }
+#endif
+}
+
+/**
+ * @brief Writes one compressed zip member and frees its buffer.
+ *
+ * Call this function for the members of a slice in the order of the members.
+ *
+ * @param zip_info the zip file that receives the member
+ * @param name the name of the member
+ * @param filelen the length that every name in this file uses
+ * @param m the result of zip_compress_member(). This function frees its buffer.
+ */
+void write_zipmember_compressed(struct zip_info *zip_info, char *name, int filelen, struct zip_member *m) {
     struct zip_lfh lfh = {
         0x04034b50, 0x0a, 0x0, 0x0, zip_info->time, zip_info->date, 0x0, 0x0, 0x0, filelen, 0x0,
     };
@@ -95,35 +148,16 @@ void write_zipmember(struct zip_info *zip_info, char *name, int filelen, char *d
         zip_info->offset,
     };
     char *filename;
-    int crc = 0, len, comp_size = data_size;
-    uLongf destlen = data_size + data_size / 500 + 12;
-    char *compbuffer;
-
-    compbuffer = g_malloc(destlen);
-    crc = crc32(0, NULL, 0);
-    crc = crc32(crc, (unsigned char *)data, data_size);
-    lfh.zipmthd = zip_info->compression_level ? 8 : 0;
-#ifdef HAVE_ZLIB
-    if (zip_info->compression_level) {
-        int error = compress2_int((Byte *)compbuffer, &destlen, (Bytef *)data, data_size, zip_info->compression_level);
-        if (error == Z_OK) {
-            if (destlen < data_size) {
-                data = compbuffer;
-                comp_size = destlen;
-            } else
-                lfh.zipmthd = 0;
-        } else {
-            fprintf(stderr, "compress2 returned %d\n", error);
-        }
-    }
-#endif
-    lfh.zipcrc = crc;
-    lfh.zipsize = comp_size;
-    lfh.zipuncmp = data_size;
-    cd.zipccrc = crc;
-    cd.zipcsiz = lfh.zipsize;
-    cd.zipcunc = data_size;
-    cd.zipcmthd = lfh.zipmthd;
+    int len;
+
+    lfh.zipmthd = m->method;
+    lfh.zipcrc = m->crc;
+    lfh.zipsize = m->data_size;
+    lfh.zipuncmp = m->uncomp_size;
+    cd.zipccrc = m->crc;
+    cd.zipcsiz = m->data_size;
+    cd.zipcunc = m->uncomp_size;
+    cd.zipcmthd = m->method;
     if (zip_info->zip64) {
         cd.zipofst = 0xffffffff;
         cd.zipcxtl += sizeof(cd_ext);
@@ -138,8 +172,8 @@ void write_zipmember(struct zip_info *zip_info, char *name, int filelen, char *d
     zip_write(zip_info, &lfh, sizeof(lfh));
     zip_write(zip_info, filename, filelen);
     zip_info->offset += sizeof(lfh) + filelen;
-    zip_write(zip_info, data, comp_size);
-    zip_info->offset += comp_size;
+    zip_write(zip_info, m->data, m->data_size);
+    zip_info->offset += m->data_size;
     dbg_assert(fwrite(&cd, sizeof(cd), 1, zip_info->dir) == 1);
     dbg_assert(fwrite(filename, filelen, 1, zip_info->dir) == 1);
     zip_info->dir_size += sizeof(cd) + filelen;
@@ -148,7 +182,14 @@ void write_zipmember(struct zip_info *zip_info, char *name, int filelen, char *d
         zip_info->dir_size += sizeof(cd_ext);
     }
 
-    g_free(compbuffer);
+    g_free(m->buffer);
+    m->buffer = NULL;
+}
+
+void write_zipmember(struct zip_info *zip_info, char *name, int filelen, char *data, int data_size) {
+    struct zip_member m;
+    zip_compress_member(zip_info, data, data_size, &m);
+    write_zipmember_compressed(zip_info, name, filelen, &m);
 }
 
 int zip_write_index(struct zip_info *info) {