This patch helps address the CPU cost of loading the index by creating
multiple threads to divide the work of loading and converting the cache
entries across all available CPU cores.
It accomplishes this by having the primary thread loop across the index file
tracking the offset and (for V4 indexes) expanding the name. It creates a
thread to process each block of entries as it comes to them. Once the
threads are complete and the cache entries are loaded, the rest of the
extensions can be loaded and processed normally on the primary thread.
I used p0002-read-cache.sh to generate some performance data:
100,000 entries
Test HEAD~3 HEAD~2
---------------------------------------------------------------------------
read_cache/discard_cache 1000 times 14.02(0.01+0.12) 9.81(0.01+0.07) -30.0%
1,000,000 entries
Test HEAD~3 HEAD~2
------------------------------------------------------------------------------
read_cache/discard_cache 1000 times 202.06(0.06+0.09) 155.72(0.03+0.06) -22.9%
Signed-off-by: Ben Peart <Ben.Peart@xxxxxxxxxxxxx>
---
Documentation/config.txt | 6 +
config.c | 14 +++
config.h | 1 +
read-cache.c | 240 +++++++++++++++++++++++++++++++++++----
4 files changed, 237 insertions(+), 24 deletions(-)
diff --git a/Documentation/config.txt b/Documentation/config.txt
index 1c42364988..79f8296d9c 100644
--- a/Documentation/config.txt
+++ b/Documentation/config.txt
@@ -2391,6 +2391,12 @@ imap::
The configuration variables in the 'imap' section are described
in linkgit:git-imap-send[1].
+index.threads::
+ Specifies the number of threads to spawn when loading the index.
+ This is meant to reduce index load time on multiprocessor machines.
+ Specifying 0 or 'true' will cause Git to auto-detect the number of
+ CPU's and set the number of threads accordingly. Defaults to 'true'.
+
index.version::
Specify the version with which new index files should be
initialized. This does not affect existing repositories.
diff --git a/config.c b/config.c
index 9a0b10d4bc..3bda124550 100644
--- a/config.c
+++ b/config.c
@@ -2289,6 +2289,20 @@ int git_config_get_fsmonitor(void)
return 0;
}
+int git_config_get_index_threads(void)
+{
+ int is_bool, val;
+
+ if (!git_config_get_bool_or_int("index.threads", &is_bool, &val)) {
+ if (is_bool)
+ return val ? 0 : 1;
+ else
+ return val;
+ }
+
+ return 0; /* auto-detect */
+}
+
NORETURN
void git_die_config_linenr(const char *key, const char *filename, int linenr)
{
diff --git a/config.h b/config.h
index ab46e0165d..a06027e69b 100644
--- a/config.h
+++ b/config.h
@@ -250,6 +250,7 @@ extern int git_config_get_untracked_cache(void);
extern int git_config_get_split_index(void);
extern int git_config_get_max_percent_split_change(void);
extern int git_config_get_fsmonitor(void);
+extern int git_config_get_index_threads(void);
/* This dies if the configured or default date is in the future */
extern int git_config_get_expiry(const char *key, const char **output);
diff --git a/read-cache.c b/read-cache.c
index 7b1354d759..c30346388a 100644
--- a/read-cache.c
+++ b/read-cache.c
@@ -1889,16 +1889,229 @@ static size_t estimate_cache_size(size_t ondisk_size, unsigned int entries)
return ondisk_size + entries * per_entry;
}
+/*
+ * A helper function that will load the specified range of cache entries
+ * from the memory mapped file and add them to the given index.
+ */
+static unsigned long load_cache_entry_block(struct index_state *istate,
+ struct mem_pool *ce_mem_pool, int offset, int nr, void *mmap,
+ unsigned long start_offset, struct strbuf *previous_name)
+{
+ int i;
+ unsigned long src_offset = start_offset;
+
+ for (i = offset; i < offset + nr; i++) {
+ struct ondisk_cache_entry *disk_ce;
+ struct cache_entry *ce;
+ unsigned long consumed;
+
+ disk_ce = (struct ondisk_cache_entry *)((char *)mmap + src_offset);
+ ce = create_from_disk(ce_mem_pool, disk_ce, &consumed, previous_name);
+ set_index_entry(istate, i, ce);
+
+ src_offset += consumed;
+ }
+ return src_offset - start_offset;
+}
+
+static unsigned long load_all_cache_entries(struct index_state *istate,
+ void *mmap, size_t mmap_size, unsigned long src_offset)
+{
+ struct strbuf previous_name_buf = STRBUF_INIT, *previous_name;
+ unsigned long consumed;
+
+ if (istate->version == 4) {
+ previous_name = &previous_name_buf;
+ mem_pool_init(&istate->ce_mem_pool,
+ estimate_cache_size_from_compressed(istate->cache_nr));
+ } else {
+ previous_name = NULL;
+ mem_pool_init(&istate->ce_mem_pool,
+ estimate_cache_size(mmap_size, istate->cache_nr));
+ }
+
+ consumed = load_cache_entry_block(istate, istate->ce_mem_pool,
+ 0, istate->cache_nr, mmap, src_offset, previous_name);
+ strbuf_release(&previous_name_buf);
+ return consumed;
+}
+
+#ifdef NO_PTHREADS
+
+#define load_cache_entries load_all_cache_entries
+
+#else
+
+#include "thread-utils.h"
+
+/*
+* Mostly randomly chosen maximum thread counts: we
+* cap the parallelism to online_cpus() threads, and we want
+* to have at least 7500 cache entries per thread for it to
+* be worth starting a thread.
+*/
+#define THREAD_COST (7500)
+
+struct load_cache_entries_thread_data
+{
+ pthread_t pthread;
+ struct index_state *istate;
+ struct mem_pool *ce_mem_pool;
+ int offset, nr;
+ void *mmap;
+ unsigned long start_offset;
+ struct strbuf previous_name_buf;
+ struct strbuf *previous_name;
+ unsigned long consumed; /* return # of bytes in index file processed */
+};
+
+/*
+* A thread proc to run the load_cache_entries() computation
+* across multiple background threads.
+*/
+static void *load_cache_entries_thread(void *_data)
+{
+ struct load_cache_entries_thread_data *p = _data;
+
+ p->consumed += load_cache_entry_block(p->istate, p->ce_mem_pool,
+ p->offset, p->nr, p->mmap, p->start_offset, p->previous_name);
+ return NULL;
+}
+
+static unsigned long load_cache_entries(struct index_state *istate,
+ void *mmap, size_t mmap_size, unsigned long src_offset)
+{
+ struct strbuf previous_name_buf = STRBUF_INIT, *previous_name;
+ struct load_cache_entries_thread_data *data;
+ int nr_threads, cpus, ce_per_thread;
+ unsigned long consumed;
+ int i, thread;
+
+ nr_threads = git_config_get_index_threads();
+ if (!nr_threads) {
+ cpus = online_cpus();
+ nr_threads = istate->cache_nr / THREAD_COST;
+ if (nr_threads > cpus)
+ nr_threads = cpus;
+ }
+
+ /* enable testing with fewer than default minimum of entries */
+ if ((istate->cache_nr > 1) && (nr_threads < 2) && git_env_bool("GIT_INDEX_THREADS_TEST", 0))
+ nr_threads = 2;
+
+ if (nr_threads < 2)
+ return load_all_cache_entries(istate, mmap, mmap_size, src_offset);
+
+ /* a little sanity checking */
+ if (istate->name_hash_initialized)
+ die("the name hash isn't thread safe");
+
+ mem_pool_init(&istate->ce_mem_pool, 0);
+ if (istate->version == 4)
+ previous_name = &previous_name_buf;
+ else
+ previous_name = NULL;
+
+ ce_per_thread = DIV_ROUND_UP(istate->cache_nr, nr_threads);
+ data = xcalloc(nr_threads, sizeof(struct load_cache_entries_thread_data));
+
+ /*
+ * Loop through index entries starting a thread for every ce_per_thread
+ * entries. Exit the loop when we've created the final thread (no need
+ * to parse the remaining entries.
+ */
+ consumed = thread = 0;
+ for (i = 0; ; i++) {
+ struct ondisk_cache_entry *ondisk;
+ const char *name;
+ unsigned int flags;
+
+ /*
+ * we've reached the beginning of a block of cache entries,
+ * kick off a thread to process them
+ */
+ if (0 == i % ce_per_thread) {
+ struct load_cache_entries_thread_data *p = &data[thread];
+
+ p->istate = istate;
+ p->offset = i;
+ p->nr = ce_per_thread < istate->cache_nr - i ? ce_per_thread : istate->cache_nr - i;
+
+ /* create a mem_pool for each thread */
+ if (istate->version == 4)
+ mem_pool_init(&p->ce_mem_pool,
+ estimate_cache_size_from_compressed(p->nr));
+ else
+ mem_pool_init(&p->ce_mem_pool,
+ estimate_cache_size(mmap_size, p->nr));
+
+ p->mmap = mmap;
+ p->start_offset = src_offset;
+ if (previous_name) {
+ strbuf_addbuf(&p->previous_name_buf, previous_name);
+ p->previous_name = &p->previous_name_buf;
+ }
+
+ if (pthread_create(&p->pthread, NULL, load_cache_entries_thread, p))
+ die("unable to create load_cache_entries_thread");
+
+ /* exit the loop when we've created the last thread */
+ if (++thread == nr_threads)
+ break;
+ }
+
+ ondisk = (struct ondisk_cache_entry *)((char *)mmap + src_offset);
+
+ /* On-disk flags are just 16 bits */
+ flags = get_be16(&ondisk->flags);
+
+ if (flags & CE_EXTENDED) {
+ struct ondisk_cache_entry_extended *ondisk2;
+ ondisk2 = (struct ondisk_cache_entry_extended *)ondisk;
+ name = ondisk2->name;
+ } else
+ name = ondisk->name;
+
+ if (!previous_name) {
+ size_t len;
+
+ /* v3 and earlier */
+ len = flags & CE_NAMEMASK;
+ if (len == CE_NAMEMASK)
+ len = strlen(name);
+ src_offset += (flags & CE_EXTENDED) ?
+ ondisk_cache_entry_extended_size(len) :
+ ondisk_cache_entry_size(len);
+ } else
+ src_offset += (name - ((char *)ondisk)) + expand_name_field(previous_name, name);
+ }
+
+ for (i = 0; i < nr_threads; i++) {
+ struct load_cache_entries_thread_data *p = data + i;
+ if (pthread_join(p->pthread, NULL))
+ die("unable to join load_cache_entries_thread");
+ mem_pool_combine(istate->ce_mem_pool, p->ce_mem_pool);
+ strbuf_release(&p->previous_name_buf);
+ consumed += p->consumed;
+ }
+
+ free(data);
+ strbuf_release(&previous_name_buf);
+
+ return consumed;
+}
+
+#endif
+
/* remember to discard_cache() before reading a different cache! */
int do_read_index(struct index_state *istate, const char *path, int must_exist)
{
- int fd, i;
+ int fd;
struct stat st;
unsigned long src_offset;
struct cache_header *hdr;
void *mmap;
size_t mmap_size;
- struct strbuf previous_name_buf = STRBUF_INIT, *previous_name;
if (istate->initialized)
return istate->cache_nr;
@@ -1935,29 +2148,8 @@ int do_read_index(struct index_state *istate, const char *path, int must_exist)
istate->cache = xcalloc(istate->cache_alloc, sizeof(*istate->cache));
istate->initialized = 1;
- if (istate->version == 4) {
- previous_name = &previous_name_buf;
- mem_pool_init(&istate->ce_mem_pool,
- estimate_cache_size_from_compressed(istate->cache_nr));
- } else {
- previous_name = NULL;
- mem_pool_init(&istate->ce_mem_pool,
- estimate_cache_size(mmap_size, istate->cache_nr));
- }
-
src_offset = sizeof(*hdr);
- for (i = 0; i < istate->cache_nr; i++) {
- struct ondisk_cache_entry *disk_ce;
- struct cache_entry *ce;
- unsigned long consumed;
-
- disk_ce = (struct ondisk_cache_entry *)((char *)mmap + src_offset);
- ce = create_from_disk(istate->ce_mem_pool, disk_ce, &consumed, previous_name);
- set_index_entry(istate, i, ce);
-
- src_offset += consumed;
- }
- strbuf_release(&previous_name_buf);
+ src_offset += load_cache_entries(istate, mmap, mmap_size, src_offset);
istate->timestamp.sec = st.st_mtime;
istate->timestamp.nsec = ST_MTIME_NSEC(st);
--
2.18.0.windows.1
