Re: [PATCH 1/5] unpack-trees: add basic support for parallel checkout

> +struct parallel_checkout_item {
> +       /* pointer to a istate->cache[] entry. Not owned by us. */
> +       struct cache_entry *ce;
> +       struct conv_attrs ca;
> +       struct stat st;
> +       enum pc_item_status status;
> +};
> +
> +struct parallel_checkout {
> +       enum pc_status status;
> +       struct parallel_checkout_item *items;
> +       size_t nr, alloc;
> +};

> +static void finish_parallel_checkout(void)
> +{
> +       if (parallel_checkout.status == PC_UNINITIALIZED)
> +               BUG("cannot finish parallel checkout: not initialized yet");
> +
> +       free(parallel_checkout.items);

> +}
> +
> +static int is_eligible_for_parallel_checkout(const struct cache_entry *ce,
> +                                            const struct conv_attrs *ca)
> +{
> +       enum conv_attrs_classification c;
> +
> +       /*
> +        * Symlinks cannot be checked out in parallel as, in case of path
> +        * collision, they could racily replace leading directories of other
> +        * entries being checked out. Submodules are checked out in child
> +        * processes, which have their own parallel checkout queues.
> +        */
> +       if (!S_ISREG(ce->ce_mode))
> +               return 0;
> +
> +       c = classify_conv_attrs(ca);
> +       switch (c) {
> +       case CA_CLASS_INCORE:
> +               return 1;
> +
> +       case CA_CLASS_INCORE_FILTER:
> +               /*
> +                * It would be safe to allow concurrent instances of
> +                * single-file smudge filters, like rot13, but we should not
> +                * assume that all filters are parallel-process safe. So we
> +                * don't allow this.
> +                */
> +               return 0;
> +
> +       case CA_CLASS_INCORE_PROCESS:
> +               /*
> +                * The parallel queue and the delayed queue are not compatible,
> +                * so they must be kept completely separated. And we can't tell
> +                * if a long-running process will delay its response without
> +                * actually asking it to perform the filtering. Therefore, this
> +                * type of filter is not allowed in parallel checkout.
> +                *
> +                * Furthermore, there should only be one instance of the
> +                * long-running process filter as we don't know how it is
> +                * managing its own concurrency. So, spreading the entries that
> +                * requisite such a filter among the parallel workers would
> +                * require a lot more inter-process communication. We would
> +                * probably have to designate a single process to interact with
> +                * the filter and send all the necessary data to it, for each
> +                * entry.
> +                */
> +               return 0;

> +       case CA_CLASS_STREAMABLE:
> +               return 1;
> +
> +       default:
> +               BUG("unsupported conv_attrs classification '%d'", c);
> +       }
> +}

> +static int handle_results(struct checkout *state)
> +{
> +       int ret = 0;
> +       size_t i;
> +       int have_pending = 0;
> +
> +       /*
> +        * We first update the successfully written entries with the collected
> +        * stat() data, so that they can be found by mark_colliding_entries(),
> +        * in the next loop, when necessary.
> +        */
> +       for (i = 0; i < parallel_checkout.nr; i++) {
> +               struct parallel_checkout_item *pc_item = &parallel_checkout.items[i];
> +               if (pc_item->status == PC_ITEM_WRITTEN)
> +                       update_ce_after_write(state, pc_item->ce, &pc_item->st);
> +       }
> +
> +       for (i = 0; i < parallel_checkout.nr; i++) {
> +               struct parallel_checkout_item *pc_item = &parallel_checkout.items[i];
> +
> +               switch(pc_item->status) {
> +               case PC_ITEM_WRITTEN:
> +                       /* Already handled */
> +                       break;
> +               case PC_ITEM_COLLIDED:
> +                       /*
> +                        * The entry could not be checked out due to a path
> +                        * collision with another entry. Since there can only
> +                        * be one entry of each colliding group on the disk, we
> +                        * could skip trying to check out this one and move on.
> +                        * However, this would leave the unwritten entries with
> +                        * null stat() fields on the index, which could
> +                        * potentially slow down subsequent operations that
> +                        * require refreshing it: git would not be able to
> +                        * trust st_size and would have to go to the filesystem
> +                        * to see if the contents match (see ie_modified()).
> +                        *
> +                        * Instead, let's pay the overhead only once, now, and
> +                        * call checkout_entry_ca() again for this file, to
> +                        * have it's stat() data stored in the index. This also

> +                        * has the benefit of adding this entry and its
> +                        * colliding pair to the collision report message.
> +                        * Additionally, this overwriting behavior is consistent
> +                        * with what the sequential checkout does, so it doesn't
> +                        * add any extra overhead.
> +                        */
> +                       ret |= checkout_entry_ca(pc_item->ce, &pc_item->ca,
> +                                                state, NULL, NULL);
> +                       break;
> +               case PC_ITEM_PENDING:
> +                       have_pending = 1;
> +                       /* fall through */
> +               case PC_ITEM_FAILED:
> +                       ret = -1;
> +                       break;
> +               default:
> +                       BUG("unknown checkout item status in parallel checkout");
> +               }
> +       }
> +
> +       if (have_pending)
> +               error(_("parallel checkout finished with pending entries"));

> +static int write_pc_item_to_fd(struct parallel_checkout_item *pc_item, int fd,
> +                              const char *path)
> +{
> +       int ret;
> +       struct stream_filter *filter;
> +       struct strbuf buf = STRBUF_INIT;
> +       char *new_blob;
> +       unsigned long size;

> +       ssize_t wrote;
> +
> +       /* Sanity check */
> +       assert(is_eligible_for_parallel_checkout(pc_item->ce, &pc_item->ca));
> +
> +       filter = get_stream_filter_ca(&pc_item->ca, &pc_item->ce->oid);
> +       if (filter) {
> +               if (stream_blob_to_fd(fd, &pc_item->ce->oid, filter, 1)) {
> +                       /* On error, reset fd to try writing without streaming */
> +                       if (reset_fd(fd, path))
> +                               return -1;
> +               } else {
> +                       return 0;
> +               }
> +       }
> +
> +       new_blob = read_blob_entry(pc_item->ce, &size);
> +       if (!new_blob)
> +               return error("unable to read sha1 file of %s (%s)", path,

> +                            oid_to_hex(&pc_item->ce->oid));
> +
> +       /*
> +        * checkout metadata is used to give context for external process
> +        * filters. Files requiring such filters are not eligible for parallel
> +        * checkout, so pass NULL.
> +        */
> +       ret = convert_to_working_tree_ca(&pc_item->ca, pc_item->ce->name,
> +                                        new_blob, size, &buf, NULL);
> +
> +       if (ret) {
> +               free(new_blob);
> +               new_blob = strbuf_detach(&buf, &newsize);
> +               size = newsize;

> +       }
> +
> +       wrote = write_in_full(fd, new_blob, size);
> +       free(new_blob);
> +       if (wrote < 0)
> +               return error("unable to write file %s", path);
> +
> +       return 0;
> +}
> +