Merge branch 'jm/free'
[git/git.git] / builtin-pack-objects.c
index 7dd0d7f..1bba6e6 100644 (file)
@@ -16,6 +16,7 @@
 #include "progress.h"
 
 #ifdef THREADED_DELTA_SEARCH
+#include "thread-utils.h"
 #include <pthread.h>
 #endif
 
@@ -68,7 +69,7 @@ static int allow_ofs_delta;
 static const char *base_name;
 static int progress = 1;
 static int window = 10;
-static uint32_t pack_size_limit;
+static uint32_t pack_size_limit, pack_size_limit_cfg;
 static int depth = 50;
 static int delta_search_threads = 1;
 static int pack_to_stdout;
@@ -445,7 +446,7 @@ static unsigned long write_object(struct sha1file *f,
                        /* nothing */;
                deflateEnd(&stream);
                datalen = stream.total_out;
-               deflateEnd(&stream);
+
                /*
                 * The object header is a byte of 'type' followed by zero or
                 * more bytes of length.
@@ -1428,8 +1429,7 @@ static int try_delta(struct unpacked *trg, struct unpacked *src,
         * accounting lock.  Compiler will optimize the strangeness
         * away when THREADED_DELTA_SEARCH is not defined.
         */
-       if (trg_entry->delta_data)
-               free(trg_entry->delta_data);
+       free(trg_entry->delta_data);
        cache_lock();
        if (trg_entry->delta_data) {
                delta_cache_size -= trg_entry->delta_size;
@@ -1594,6 +1594,15 @@ static void find_deltas(struct object_entry **list, unsigned *list_size,
 
 #ifdef THREADED_DELTA_SEARCH
 
+/*
+ * The main thread waits on the condition that (at least) one of the workers
+ * has stopped working (which is indicated in the .working member of
+ * struct thread_params).
+ * When a work thread has completed its work, it sets .working to 0 and
+ * signals the main thread and waits on the condition that .data_ready
+ * becomes 1.
+ */
+
 struct thread_params {
        pthread_t thread;
        struct object_entry **list;
@@ -1601,37 +1610,50 @@ struct thread_params {
        unsigned remaining;
        int window;
        int depth;
+       int working;
+       int data_ready;
+       pthread_mutex_t mutex;
+       pthread_cond_t cond;
        unsigned *processed;
 };
 
-static pthread_mutex_t data_request  = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t data_ready    = PTHREAD_MUTEX_INITIALIZER;
-static pthread_mutex_t data_provider = PTHREAD_MUTEX_INITIALIZER;
-static struct thread_params *data_requester;
+static pthread_cond_t progress_cond = PTHREAD_COND_INITIALIZER;
 
 static void *threaded_find_deltas(void *arg)
 {
        struct thread_params *me = arg;
 
-       for (;;) {
-               pthread_mutex_lock(&data_request);
-               data_requester = me;
-               pthread_mutex_unlock(&data_provider);
-               pthread_mutex_lock(&data_ready);
-               pthread_mutex_unlock(&data_request);
-
-               if (!me->remaining)
-                       return NULL;
-
+       while (me->remaining) {
                find_deltas(me->list, &me->remaining,
                            me->window, me->depth, me->processed);
+
+               progress_lock();
+               me->working = 0;
+               pthread_cond_signal(&progress_cond);
+               progress_unlock();
+
+               /*
+                * We must not set ->data_ready before we wait on the
+                * condition because the main thread may have set it to 1
+                * before we get here. In order to be sure that new
+                * work is available if we see 1 in ->data_ready, it
+                * was initialized to 0 before this thread was spawned
+                * and we reset it to 0 right away.
+                */
+               pthread_mutex_lock(&me->mutex);
+               while (!me->data_ready)
+                       pthread_cond_wait(&me->cond, &me->mutex);
+               me->data_ready = 0;
+               pthread_mutex_unlock(&me->mutex);
        }
+       /* leave ->working 1 so that this doesn't get more work assigned */
+       return NULL;
 }
 
 static void ll_find_deltas(struct object_entry **list, unsigned list_size,
                           int window, int depth, unsigned *processed)
 {
-       struct thread_params *target, p[delta_search_threads];
+       struct thread_params p[delta_search_threads];
        int i, ret, active_threads = 0;
 
        if (delta_search_threads <= 1) {
@@ -1639,49 +1661,43 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
                return;
        }
 
-       pthread_mutex_lock(&data_provider);
-       pthread_mutex_lock(&data_ready);
-
-       /* Start work threads. */
+       /* Partition the work amongst work threads. */
        for (i = 0; i < delta_search_threads; i++) {
+               unsigned sub_size = list_size / (delta_search_threads - i);
+
                p[i].window = window;
                p[i].depth = depth;
                p[i].processed = processed;
-               p[i].remaining = 0;
-               ret = pthread_create(&p[i].thread, NULL,
-                                    threaded_find_deltas, &p[i]);
-               if (ret)
-                       die("unable to create thread: %s", strerror(ret));
-               active_threads++;
-       }
-
-       /* Then partition the work amongst them. */
-       for (i = 0; i < delta_search_threads; i++) {
-               unsigned sub_size = list_size / (delta_search_threads - i);
-
-               pthread_mutex_lock(&data_provider);
-               target = data_requester;
-               if (!sub_size) {
-                       pthread_mutex_unlock(&data_ready);
-                       pthread_join(target->thread, NULL);
-                       active_threads--;
-                       continue;
-               }
+               p[i].working = 1;
+               p[i].data_ready = 0;
 
                /* try to split chunks on "path" boundaries */
-               while (sub_size < list_size && list[sub_size]->hash &&
+               while (sub_size && sub_size < list_size &&
+                      list[sub_size]->hash &&
                       list[sub_size]->hash == list[sub_size-1]->hash)
                        sub_size++;
 
-               target->list = list;
-               target->list_size = sub_size;
-               target->remaining = sub_size;
-               pthread_mutex_unlock(&data_ready);
+               p[i].list = list;
+               p[i].list_size = sub_size;
+               p[i].remaining = sub_size;
 
                list += sub_size;
                list_size -= sub_size;
        }
 
+       /* Start work threads. */
+       for (i = 0; i < delta_search_threads; i++) {
+               if (!p[i].list_size)
+                       continue;
+               pthread_mutex_init(&p[i].mutex, NULL);
+               pthread_cond_init(&p[i].cond, NULL);
+               ret = pthread_create(&p[i].thread, NULL,
+                                    threaded_find_deltas, &p[i]);
+               if (ret)
+                       die("unable to create thread: %s", strerror(ret));
+               active_threads++;
+       }
+
        /*
         * Now let's wait for work completion.  Each time a thread is done
         * with its work, we steal half of the remaining work from the
@@ -1690,13 +1706,21 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
         * until the remaining object list segments are simply too short
         * to be worth splitting anymore.
         */
-       do {
+       while (active_threads) {
+               struct thread_params *target = NULL;
                struct thread_params *victim = NULL;
                unsigned sub_size = 0;
-               pthread_mutex_lock(&data_provider);
-               target = data_requester;
 
                progress_lock();
+               for (;;) {
+                       for (i = 0; !target && i < delta_search_threads; i++)
+                               if (!p[i].working)
+                                       target = &p[i];
+                       if (target)
+                               break;
+                       pthread_cond_wait(&progress_cond, &progress_mutex);
+               }
+
                for (i = 0; i < delta_search_threads; i++)
                        if (p[i].remaining > 2*window &&
                            (!victim || victim->remaining < p[i].remaining))
@@ -1723,17 +1747,23 @@ static void ll_find_deltas(struct object_entry **list, unsigned list_size,
                        victim->list_size -= sub_size;
                        victim->remaining -= sub_size;
                }
-               progress_unlock();
-
                target->list_size = sub_size;
                target->remaining = sub_size;
-               pthread_mutex_unlock(&data_ready);
+               target->working = 1;
+               progress_unlock();
+
+               pthread_mutex_lock(&target->mutex);
+               target->data_ready = 1;
+               pthread_cond_signal(&target->cond);
+               pthread_mutex_unlock(&target->mutex);
 
                if (!sub_size) {
                        pthread_join(target->thread, NULL);
+                       pthread_cond_destroy(&target->cond);
+                       pthread_mutex_destroy(&target->mutex);
                        active_threads--;
                }
-       } while (active_threads);
+       }
 }
 
 #else
@@ -1822,11 +1852,11 @@ static int git_pack_config(const char *k, const char *v)
        }
        if (!strcmp(k, "pack.threads")) {
                delta_search_threads = git_config_int(k, v);
-               if (delta_search_threads < 1)
+               if (delta_search_threads < 0)
                        die("invalid number of threads specified (%d)",
                            delta_search_threads);
 #ifndef THREADED_DELTA_SEARCH
-               if (delta_search_threads > 1)
+               if (delta_search_threads != 1)
                        warning("no threads support, ignoring %s", k);
 #endif
                return 0;
@@ -1837,6 +1867,10 @@ static int git_pack_config(const char *k, const char *v)
                        die("bad pack.indexversion=%d", pack_idx_default_version);
                return 0;
        }
+       if (!strcmp(k, "pack.packsizelimit")) {
+               pack_size_limit_cfg = git_config_ulong(k, v);
+               return 0;
+       }
        return git_default_config(k, v);
 }
 
@@ -1984,7 +2018,7 @@ static void get_object_list(int ac, const char **av)
 
        while (fgets(line, sizeof(line), stdin) != NULL) {
                int len = strlen(line);
-               if (line[len - 1] == '\n')
+               if (len && line[len - 1] == '\n')
                        line[--len] = 0;
                if (!len)
                        break;
@@ -1999,7 +2033,8 @@ static void get_object_list(int ac, const char **av)
                        die("bad revision '%s'", line);
        }
 
-       prepare_revision_walk(&revs);
+       if (prepare_revision_walk(&revs))
+               die("revision walk setup failed");
        mark_edges_uninteresting(revs.commits, &revs, show_edge);
        traverse_commit_list(&revs, show_commit, show_object);
 
@@ -2066,6 +2101,7 @@ int cmd_pack_objects(int argc, const char **argv, const char *prefix)
                }
                if (!prefixcmp(arg, "--max-pack-size=")) {
                        char *end;
+                       pack_size_limit_cfg = 0;
                        pack_size_limit = strtoul(arg+16, &end, 0) * 1024 * 1024;
                        if (!arg[16] || *end)
                                usage(pack_usage);
@@ -2086,10 +2122,10 @@ int cmd_pack_objects(int argc, const char **argv, const char *prefix)
                if (!prefixcmp(arg, "--threads=")) {
                        char *end;
                        delta_search_threads = strtoul(arg+10, &end, 0);
-                       if (!arg[10] || *end || delta_search_threads < 1)
+                       if (!arg[10] || *end || delta_search_threads < 0)
                                usage(pack_usage);
 #ifndef THREADED_DELTA_SEARCH
-                       if (delta_search_threads > 1)
+                       if (delta_search_threads != 1)
                                warning("no threads support, "
                                        "ignoring %s", arg);
 #endif
@@ -2190,12 +2226,20 @@ int cmd_pack_objects(int argc, const char **argv, const char *prefix)
        if (pack_to_stdout != !base_name)
                usage(pack_usage);
 
+       if (!pack_to_stdout && !pack_size_limit)
+               pack_size_limit = pack_size_limit_cfg;
+
        if (pack_to_stdout && pack_size_limit)
                die("--max-pack-size cannot be used to build a pack for transfer.");
 
        if (!pack_to_stdout && thin)
                die("--thin cannot be used to build an indexable pack.");
 
+#ifdef THREADED_DELTA_SEARCH
+       if (!delta_search_threads)      /* --threads=0 means autodetect */
+               delta_search_threads = online_cpus();
+#endif
+
        prepare_packed_git();
 
        if (progress)