X-Git-Url: https://git.tokkee.org/?a=blobdiff_plain;f=builtin-pack-objects.c;h=7dff6536dea3b725a95fcd17b001c7297b251151;hb=8e0f70033b2bd1679a6e5971978fdc3ee09bdb72;hp=4f446588ac9ce9d8aa50d0c288817a16855422e6;hpb=08e1812db1585c450bfe7f41e7106222346c88da;p=git.git diff --git a/builtin-pack-objects.c b/builtin-pack-objects.c index 4f446588a..7dff6536d 100644 --- a/builtin-pack-objects.c +++ b/builtin-pack-objects.c @@ -68,7 +68,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 +445,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. @@ -1245,28 +1245,37 @@ static void get_object_details(void) free(sorted_by_offset); } +/* + * We search for deltas in a list sorted by type, by filename hash, and then + * by size, so that we see progressively smaller and smaller files. + * That's because we prefer deltas to be from the bigger file + * to the smaller -- deletes are potentially cheaper, but perhaps + * more importantly, the bigger file is likely the more recent + * one. The deepest deltas are therefore the oldest objects which are + * less susceptible to be accessed often. + */ static int type_size_sort(const void *_a, const void *_b) { const struct object_entry *a = *(struct object_entry **)_a; const struct object_entry *b = *(struct object_entry **)_b; - if (a->type < b->type) - return -1; if (a->type > b->type) - return 1; - if (a->hash < b->hash) return -1; - if (a->hash > b->hash) + if (a->type < b->type) return 1; - if (a->preferred_base < b->preferred_base) + if (a->hash > b->hash) return -1; - if (a->preferred_base > b->preferred_base) + if (a->hash < b->hash) return 1; - if (a->size < b->size) + if (a->preferred_base > b->preferred_base) return -1; + if (a->preferred_base < b->preferred_base) + return 1; if (a->size > b->size) + return -1; + if (a->size < b->size) return 1; - return a > b ? -1 : (a < b); /* newest last */ + return a < b ? -1 : (a > b); /* newest first */ } struct unpacked { @@ -1317,14 +1326,6 @@ static pthread_mutex_t progress_mutex = PTHREAD_MUTEX_INITIALIZER; #endif -/* - * We search for deltas _backwards_ in a list sorted by type and - * by size, so that we see progressively smaller and smaller files. - * That's because we prefer deltas to be from the bigger file - * to the smaller - deletes are potentially cheaper, but perhaps - * more importantly, the bigger file is likely the more recent - * one. - */ static int try_delta(struct unpacked *trg, struct unpacked *src, unsigned max_depth, unsigned long *mem_usage) { @@ -1422,24 +1423,19 @@ static int try_delta(struct unpacked *trg, struct unpacked *src, } } - trg_entry->delta = src_entry; - trg_entry->delta_size = delta_size; - trg->depth = src->depth + 1; - /* * Handle memory allocation outside of the cache * 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; trg_entry->delta_data = NULL; } if (delta_cacheable(src_size, trg_size, delta_size)) { - delta_cache_size += trg_entry->delta_size; + delta_cache_size += delta_size; cache_unlock(); trg_entry->delta_data = xrealloc(delta_buf, delta_size); } else { @@ -1447,6 +1443,10 @@ static int try_delta(struct unpacked *trg, struct unpacked *src, free(delta_buf); } + trg_entry->delta = src_entry; + trg_entry->delta_size = delta_size; + trg->depth = src->depth + 1; + return 1; } @@ -1478,10 +1478,10 @@ static unsigned long free_unpacked(struct unpacked *n) return freed_mem; } -static void find_deltas(struct object_entry **list, unsigned list_size, +static void find_deltas(struct object_entry **list, unsigned *list_size, int window, int depth, unsigned *processed) { - uint32_t i = list_size, idx = 0, count = 0; + uint32_t i, idx = 0, count = 0; unsigned int array_size = window * sizeof(struct unpacked); struct unpacked *array; unsigned long mem_usage = 0; @@ -1489,11 +1489,23 @@ static void find_deltas(struct object_entry **list, unsigned list_size, array = xmalloc(array_size); memset(array, 0, array_size); - do { - struct object_entry *entry = list[--i]; + for (;;) { + struct object_entry *entry = *list++; struct unpacked *n = array + idx; int j, max_depth, best_base = -1; + progress_lock(); + if (!*list_size) { + progress_unlock(); + break; + } + (*list_size)--; + if (!entry->preferred_base) { + (*processed)++; + display_progress(progress_state, *processed); + } + progress_unlock(); + mem_usage -= free_unpacked(n); n->entry = entry; @@ -1511,11 +1523,6 @@ static void find_deltas(struct object_entry **list, unsigned list_size, if (entry->preferred_base) goto next; - progress_lock(); - (*processed)++; - display_progress(progress_state, *processed); - progress_unlock(); - /* * If the current object is at pack edge, take the depth the * objects that depend on the current object into account @@ -1575,7 +1582,7 @@ static void find_deltas(struct object_entry **list, unsigned list_size, count++; if (idx >= window) idx = 0; - } while (i > 0); + } for (i = 0; i < window; ++i) { free_delta_index(array[i].index); @@ -1586,94 +1593,180 @@ 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; unsigned list_size; + 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); + while (me->remaining) { + find_deltas(me->list, &me->remaining, + me->window, me->depth, me->processed); - if (!me->list_size) - return NULL; + progress_lock(); + me->working = 0; + pthread_cond_signal(&progress_cond); + progress_unlock(); - find_deltas(me->list, me->list_size, - me->window, me->depth, me->processed); + /* + * 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]; - int i, ret; - unsigned chunk_size; + struct thread_params p[delta_search_threads]; + int i, ret, active_threads = 0; if (delta_search_threads <= 1) { - find_deltas(list, list_size, window, depth, processed); + find_deltas(list, &list_size, window, depth, processed); return; } - pthread_mutex_lock(&data_provider); - pthread_mutex_lock(&data_ready); - + /* 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].working = 1; + p[i].data_ready = 0; + + /* try to split chunks on "path" boundaries */ + while (sub_size && sub_size < list_size && + list[sub_size]->hash && + list[sub_size]->hash == list[sub_size-1]->hash) + sub_size++; + + 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++; } - /* this should be auto-tuned somehow */ - chunk_size = window * 1000; + /* + * 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 + * thread with the largest number of unprocessed objects and give + * it to that newly idle thread. This ensure good load balancing + * until the remaining object list segments are simply too short + * to be worth splitting anymore. + */ + while (active_threads) { + struct thread_params *target = NULL; + struct thread_params *victim = NULL; + unsigned sub_size = 0; - do { - unsigned sublist_size = chunk_size; - if (sublist_size > list_size) - sublist_size = list_size; + 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); + } - /* try to split chunks on "path" boundaries */ - while (sublist_size < list_size && list[sublist_size]->hash && - list[sublist_size]->hash == list[sublist_size-1]->hash) - sublist_size++; - - pthread_mutex_lock(&data_provider); - target = data_requester; - target->list = list; - target->list_size = sublist_size; - pthread_mutex_unlock(&data_ready); - - list += sublist_size; - list_size -= sublist_size; - if (!sublist_size) { + for (i = 0; i < delta_search_threads; i++) + if (p[i].remaining > 2*window && + (!victim || victim->remaining < p[i].remaining)) + victim = &p[i]; + if (victim) { + sub_size = victim->remaining / 2; + list = victim->list + victim->list_size - sub_size; + while (sub_size && list[0]->hash && + list[0]->hash == list[-1]->hash) { + list++; + sub_size--; + } + if (!sub_size) { + /* + * It is possible for some "paths" to have + * so many objects that no hash boundary + * might be found. Let's just steal the + * exact half in that case. + */ + sub_size = victim->remaining / 2; + list -= sub_size; + } + target->list = list; + victim->list_size -= sub_size; + victim->remaining -= sub_size; + } + target->list_size = sub_size; + target->remaining = sub_size; + 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); - i--; + pthread_cond_destroy(&target->cond); + pthread_mutex_destroy(&target->mutex); + active_threads--; } - } while (i); + } } #else -#define ll_find_deltas find_deltas +#define ll_find_deltas(l, s, w, d, p) find_deltas(l, &s, w, d, p) #endif static void prepare_pack(int window, int depth) @@ -1773,6 +1866,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); } @@ -1920,7 +2017,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; @@ -1935,7 +2032,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); @@ -2002,6 +2100,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); @@ -2126,6 +2225,9 @@ 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.");