@matheusbsbs / matheus-bagliano / commits / f08cbf60fe
Fork 0

index-pack: make quantum of work smaller

Currently, when index-pack resolves deltas, it does not split up delta trees into threads: each delta base root (an object that is not a REF_DELTA or OFS_DELTA) can go into its own thread, but all deltas on that root (direct or indirect) are processed in the same thread. This is a problem when a repository contains a large text file (thus, delta-able) that is modified many times - delta resolution time during fetching is dominated by processing the deltas corresponding to that text file. This patch contains a solution to that. When cloning using git -c core.deltabasecachelimit=1g clone \ https://fuchsia.googlesource.com/third_party/vulkan-cts on my laptop, clone time improved from 3m2s to 2m5s (using 3 threads, which is the default). The solution is to have a global work stack. This stack contains delta bases (objects, whether appearing directly in the packfile or generated by delta resolution, that themselves have delta children) that need to be processed; whenever a thread needs work, it peeks at the top of the stack and processes its next unprocessed child. If a thread finds the stack empty, it will look for more delta base roots to push on the stack instead. The main weakness of having a global work stack is that more time is spent in the mutex, but profiling has shown that most time is spent in the resolution of the deltas themselves, so this shouldn't be an issue in practice. In any case, experimentation (as described in the clone command above) shows that this patch is a net improvement. Signed-off-by: Jonathan Tan <jonathantanmy@google.com> Signed-off-by: Junio C Hamano <gitster@pobox.com>

Jonathan Tan committed Sep 8, 2020 at 12:48 UTC f08cbf60fe11507b5ff722ceae95dfb86ce654ee
1 file changed +200 -148
builtin/index-pack.c
+200 -148
@@ -38,15 +38,56 @@ struct base_data {
38 struct object_entry *obj;
39 int ref_first, ref_last;
40 int ofs_first, ofs_last;
41 + /*
42 + * Threads should increment retain_data if they are about to call
43 + * patch_delta() using this struct's data as a base, and decrement this
44 + * when they are done. While retain_data is nonzero, this struct's data
45 + * will not be freed even if the delta base cache limit is exceeded.
46 + */
47 + int retain_data;
48 + /*
49 + * The number of direct children that have not been fully processed
50 + * (entered work_head, entered done_head, left done_head). When this
51 + * number reaches zero, this struct base_data can be freed.
52 + */
53 + int children_remaining;
54
55 /* Not initialized by make_base(). */
56 + struct list_head list;
57 void *data;
58 unsigned long size;
59 };
60
61 +/*
62 + * Stack of struct base_data that have unprocessed children.
63 + * threaded_second_pass() uses this as a source of work (the other being the
64 + * objects array).
65 + *
66 + * Guarded by work_mutex.
67 + */
68 +static LIST_HEAD(work_head);
69 +
70 +/*
71 + * Stack of struct base_data that have children, all of whom have been
72 + * processed or are being processed, and at least one child is being processed.
73 + * These struct base_data must be kept around until the last child is
74 + * processed.
75 + *
76 + * Guarded by work_mutex.
77 + */
78 +static LIST_HEAD(done_head);
79 +
80 +/*
81 + * All threads share one delta base cache.
82 + *
83 + * base_cache_used is guarded by work_mutex, and base_cache_limit is read-only
84 + * in a thread.
85 + */
86 +static size_t base_cache_used;
87 +static size_t base_cache_limit;
88 +
89 struct thread_local {
90 pthread_t thread;
49 - size_t base_cache_used;
91 int pack_fd;
92 };
93
@@ -369,36 +410,38 @@ static void free_base_data(struct base_data *c)
410 {
411 if (c->data) {
412 FREE_AND_NULL(c->data);
372 - get_thread_data()->base_cache_used -= c->size;
413 + base_cache_used -= c->size;
414 }
415 }
416
376 -static void prune_base_data(struct base_data *youngest_child)
417 +static void prune_base_data(struct base_data *retain)
418 {
378 - struct base_data *b;
379 - struct thread_local *data = get_thread_data();
380 - struct base_data **ancestry = NULL;
381 - size_t nr = 0, alloc = 0;
382 - ssize_t i;
419 + struct list_head *pos;
420
384 - if (data->base_cache_used <= delta_base_cache_limit)
421 + if (base_cache_used <= base_cache_limit)
422 return;
423
387 - /*
388 - * Free all ancestors of youngest_child until we have enough space,
389 - * starting with the oldest. (We cannot free youngest_child itself.)
390 - */
391 - for (b = youngest_child->base; b != NULL; b = b->base) {
392 - ALLOC_GROW(ancestry, nr + 1, alloc);
393 - ancestry[nr++] = b;
424 + list_for_each_prev(pos, &done_head) {
425 + struct base_data *b = list_entry(pos, struct base_data, list);
426 + if (b->retain_data || b == retain)
427 + continue;
428 + if (b->data) {
429 + free_base_data(b);
430 + if (base_cache_used <= base_cache_limit)
431 + return;
432 + }
433 }
395 - for (i = nr - 1;
396 - i >= 0 && data->base_cache_used > delta_base_cache_limit;
397 - i--) {
398 - if (ancestry[i]->data)
399 - free_base_data(ancestry[i]);
434 +
435 + list_for_each_prev(pos, &work_head) {
436 + struct base_data *b = list_entry(pos, struct base_data, list);
437 + if (b->retain_data || b == retain)
438 + continue;
439 + if (b->data) {
440 + free_base_data(b);
441 + if (base_cache_used <= base_cache_limit)
442 + return;
443 + }
444 }
401 - free(ancestry);
445 }
446
447 static int is_delta_type(enum object_type type)
@@ -851,15 +894,7 @@ static void sha1_object(const void *data, struct object_entry *obj_entry,
894 }
895
896 /*
854 - * This function is part of find_unresolved_deltas(). There are two
855 - * walkers going in the opposite ways.
856 - *
857 - * The first one in find_unresolved_deltas() traverses down from
858 - * parent node to children, deflating nodes along the way. However,
859 - * memory for deflated nodes is limited by delta_base_cache_limit, so
860 - * at some point parent node's deflated content may be freed.
861 - *
862 - * The second walker is this function, which goes from current node up
897 + * Walk from current node up
898 * to top parent if necessary to deflate the node. In normal
899 * situation, its parent node would be already deflated, so it just
900 * needs to apply delta.
@@ -887,7 +922,7 @@ static void *get_base_data(struct base_data *c)
922 if (!delta_nr) {
923 c->data = get_data_from_pack(obj);
924 c->size = obj->size;
890 - get_thread_data()->base_cache_used += c->size;
925 + base_cache_used += c->size;
926 prune_base_data(c);
927 }
928 for (; delta_nr > 0; delta_nr--) {
@@ -903,7 +938,7 @@ static void *get_base_data(struct base_data *c)
938 free(raw);
939 if (!c->data)
940 bad_object(obj->idx.offset, _("failed to apply delta"));
906 - get_thread_data()->base_cache_used += c->size;
941 + base_cache_used += c->size;
942 prune_base_data(c);
943 }
944 free(delta);
@@ -921,6 +956,8 @@ static struct base_data *make_base(struct object_entry *obj,
956 &base->ref_first, &base->ref_last);
957 find_ofs_delta_children(obj->idx.offset,
958 &base->ofs_first, &base->ofs_last);
959 + base->children_remaining = base->ref_last - base->ref_first +
960 + base->ofs_last - base->ofs_first + 2;
961 return base;
962 }
963
@@ -954,14 +991,8 @@ static struct base_data *resolve_delta(struct object_entry *delta_obj,
991 &delta_obj->idx.oid);
992
993 result = make_base(delta_obj, base);
957 - if (result->ref_last == -1 && result->ofs_last == -1) {
958 - free(result_data);
959 - } else {
960 - result->data = result_data;
961 - result->size = result_size;
962 - get_thread_data()->base_cache_used += result->size;
963 - prune_base_data(result);
964 - }
994 + result->data = result_data;
995 + result->size = result_size;
996
997 counter_lock();
998 nr_resolved_deltas++;
@@ -970,86 +1001,6 @@ static struct base_data *resolve_delta(struct object_entry *delta_obj,
1001 return result;
1002 }
1003
973 -/*
974 - * Standard boolean compare-and-swap: atomically check whether "*type" is
975 - * "want"; if so, swap in "set" and return true. Otherwise, leave it untouched
976 - * and return false.
977 - */
978 -static int compare_and_swap_type(signed char *type,
979 - enum object_type want,
980 - enum object_type set)
981 -{
982 - enum object_type old;
983 -
984 - type_cas_lock();
985 - old = *type;
986 - if (old == want)
987 - *type = set;
988 - type_cas_unlock();
989 -
990 - return old == want;
991 -}
992 -
993 -static struct base_data *find_unresolved_deltas_1(struct base_data *base,
994 - struct base_data *prev_base)
995 -{
996 - if (base->ref_first <= base->ref_last) {
997 - struct object_entry *child = objects + ref_deltas[base->ref_first].obj_no;
998 - struct base_data *result;
999 -
1000 - if (!compare_and_swap_type(&child->real_type, OBJ_REF_DELTA,
1001 - base->obj->real_type))
1002 - die("REF_DELTA at offset %"PRIuMAX" already resolved (duplicate base %s?)",
1003 - (uintmax_t)child->idx.offset,
1004 - oid_to_hex(&base->obj->idx.oid));
1005 -
1006 - get_base_data(base);
1007 - result = resolve_delta(child, base);
1008 - if (base->ref_first == base->ref_last && base->ofs_last == -1)
1009 - free_base_data(base);
1010 -
1011 - base->ref_first++;
1012 - return result;
1013 - }
1014 -
1015 - if (base->ofs_first <= base->ofs_last) {
1016 - struct object_entry *child = objects + ofs_deltas[base->ofs_first].obj_no;
1017 - struct base_data *result;
1018 -
1019 - assert(child->real_type == OBJ_OFS_DELTA);
1020 - child->real_type = base->obj->real_type;
1021 - get_base_data(base);
1022 - result = resolve_delta(child, base);
1023 - if (base->ofs_first == base->ofs_last)
1024 - free_base_data(base);
1025 -
1026 - base->ofs_first++;
1027 - return result;
1028 - }
1029 -
1030 - free_base_data(base);
1031 - return NULL;
1032 -}
1033 -
1034 -static void find_unresolved_deltas(struct base_data *base)
1035 -{
1036 - struct base_data *new_base, *prev_base = NULL;
1037 - for (;;) {
1038 - new_base = find_unresolved_deltas_1(base, prev_base);
1039 -
1040 - if (new_base) {
1041 - prev_base = base;
1042 - base = new_base;
1043 - } else {
1044 - free(base);
1045 - base = prev_base;
1046 - if (!base)
1047 - return;
1048 - prev_base = base->base;
1049 - }
1050 - }
1051 -}
1052 -
1004 static int compare_ofs_delta_entry(const void *a, const void *b)
1005 {
1006 const struct ofs_delta_entry *delta_a = a;
@@ -1068,33 +1019,131 @@ static int compare_ref_delta_entry(const void *a, const void *b)
1019 return oidcmp(&delta_a->oid, &delta_b->oid);
1020 }
1021
1071 -static void resolve_base(struct object_entry *obj)
1072 -{
1073 - struct base_data *base_obj = make_base(obj, NULL);
1074 -
1075 - find_unresolved_deltas(base_obj);
1076 -}
1077 -
1022 static void *threaded_second_pass(void *data)
1023 {
1080 - set_thread_data(data);
1024 + if (data)
1025 + set_thread_data(data);
1026 for (;;) {
1082 - int i;
1083 - counter_lock();
1084 - display_progress(progress, nr_resolved_deltas);
1085 - counter_unlock();
1027 + struct base_data *parent = NULL;
1028 + struct object_entry *child_obj;
1029 + struct base_data *child;
1030 +
1031 work_lock();
1087 - while (nr_dispatched < nr_objects &&
1088 - is_delta_type(objects[nr_dispatched].type))
1089 - nr_dispatched++;
1090 - if (nr_dispatched >= nr_objects) {
1091 - work_unlock();
1092 - break;
1032 + if (list_empty(&work_head)) {
1033 + /*
1034 + * Take an object from the object array.
1035 + */
1036 + while (nr_dispatched < nr_objects &&
1037 + is_delta_type(objects[nr_dispatched].type))
1038 + nr_dispatched++;
1039 + if (nr_dispatched >= nr_objects) {
1040 + work_unlock();
1041 + break;
1042 + }
1043 + child_obj = &objects[nr_dispatched++];
1044 + } else {
1045 + /*
1046 + * Peek at the top of the stack, and take a child from
1047 + * it.
1048 + */
1049 + parent = list_first_entry(&work_head, struct base_data,
1050 + list);
1051 +
1052 + if (parent->ref_first <= parent->ref_last) {
1053 + int offset = ref_deltas[parent->ref_first++].obj_no;
1054 + child_obj = objects + offset;
1055 + if (child_obj->real_type != OBJ_REF_DELTA)
1056 + die("REF_DELTA at offset %"PRIuMAX" already resolved (duplicate base %s?)",
1057 + (uintmax_t) child_obj->idx.offset,
1058 + oid_to_hex(&parent->obj->idx.oid));
1059 + child_obj->real_type = parent->obj->real_type;
1060 + } else {
1061 + child_obj = objects +
1062 + ofs_deltas[parent->ofs_first++].obj_no;
1063 + assert(child_obj->real_type == OBJ_OFS_DELTA);
1064 + child_obj->real_type = parent->obj->real_type;
1065 + }
1066 +
1067 + if (parent->ref_first > parent->ref_last &&
1068 + parent->ofs_first > parent->ofs_last) {
1069 + /*
1070 + * This parent has run out of children, so move
1071 + * it to done_head.
1072 + */
1073 + list_del(&parent->list);
1074 + list_add(&parent->list, &done_head);
1075 + }
1076 +
1077 + /*
1078 + * Ensure that the parent has data, since we will need
1079 + * it later.
1080 + *
1081 + * NEEDSWORK: If parent data needs to be reloaded, this
1082 + * prolongs the time that the current thread spends in
1083 + * the mutex. A mitigating factor is that parent data
1084 + * needs to be reloaded only if the delta base cache
1085 + * limit is exceeded, so in the typical case, this does
1086 + * not happen.
1087 + */
1088 + get_base_data(parent);
1089 + parent->retain_data++;
1090 }
1094 - i = nr_dispatched++;
1091 work_unlock();
1092
1097 - resolve_base(&objects[i]);
1093 + if (parent) {
1094 + child = resolve_delta(child_obj, parent);
1095 + if (!child->children_remaining)
1096 + FREE_AND_NULL(child->data);
1097 + } else {
1098 + child = make_base(child_obj, NULL);
1099 + if (child->children_remaining) {
1100 + /*
1101 + * Since this child has its own delta children,
1102 + * we will need this data in the future.
1103 + * Inflate now so that future iterations will
1104 + * have access to this object's data while
1105 + * outside the work mutex.
1106 + */
1107 + child->data = get_data_from_pack(child_obj);
1108 + child->size = child_obj->size;
1109 + }
1110 + }
1111 +
1112 + work_lock();
1113 + if (parent)
1114 + parent->retain_data--;
1115 + if (child->data) {
1116 + /*
1117 + * This child has its own children, so add it to
1118 + * work_head.
1119 + */
1120 + list_add(&child->list, &work_head);
1121 + base_cache_used += child->size;
1122 + prune_base_data(NULL);
1123 + } else {
1124 + /*
1125 + * This child does not have its own children. It may be
1126 + * the last descendant of its ancestors; free those
1127 + * that we can.
1128 + */
1129 + struct base_data *p = parent;
1130 +
1131 + while (p) {
1132 + struct base_data *next_p;
1133 +
1134 + p->children_remaining--;
1135 + if (p->children_remaining)
1136 + break;
1137 +
1138 + next_p = p->base;
1139 + free_base_data(p);
1140 + list_del(&p->list);
1141 + free(p);
1142 +
1143 + p = next_p;
1144 + }
1145 + }
1146 + work_unlock();
1147 }
1148 return NULL;
1149 }
@@ -1195,6 +1244,7 @@ static void resolve_deltas(void)
1244 nr_ref_deltas + nr_ofs_deltas);
1245
1246 nr_dispatched = 0;
1247 + base_cache_limit = delta_base_cache_limit * nr_threads;
1248 if (nr_threads > 1 || getenv("GIT_FORCE_THREADS")) {
1249 init_thread();
1250 for (i = 0; i < nr_threads; i++) {
@@ -1364,10 +1414,8 @@ static void fix_unresolved_deltas(struct hashfile *f)
1414 for (i = 0; i < nr_ref_deltas; i++) {
1415 struct ref_delta_entry *d = sorted_by_pos[i];
1416 enum object_type type;
1367 - struct base_data *base;
1417 void *data;
1418 unsigned long size;
1370 - struct object_entry *obj;
1419
1420 if (objects[d->obj_no].real_type != OBJ_REF_DELTA)
1421 continue;
@@ -1379,11 +1427,15 @@ static void fix_unresolved_deltas(struct hashfile *f)
1427 data, size,
1428 type_name(type)))
1429 die(_("local object %s is corrupt"), oid_to_hex(&d->oid));
1382 - obj = append_obj_to_pack(f, d->oid.hash, data, size, type);
1383 - base = make_base(obj, NULL);
1384 - base->data = data;
1385 - base->size = size;
1386 - find_unresolved_deltas(base);
1430 +
1431 + /*
1432 + * Add this as an object to the objects array and call
1433 + * threaded_second_pass() (which will pick up the added
1434 + * object).
1435 + */
1436 + append_obj_to_pack(f, d->oid.hash, data, size, type);
1437 + threaded_second_pass(NULL);
1438 +
1439 display_progress(progress, nr_resolved_deltas);
1440 }
1441 free(sorted_by_pos);