Re: [PATCH 03/10] run-command: add an asynchronous parallel child processor

[Stefan Beller <sbeller@google.com>]

On Thu, Sep 17, 2015 at 2:44 PM, Junio C Hamano <gitster@pobox.com> wrote:

> Hmm, you are relying on the fact that a valid pid can never be 0, so
> you can just use pp->children[i].child.pid to see if a "slot" is
> occupied without even using pp->slots[] (or pp->children[i].in_use).

We could either use the pid as accessed via children[i].process.pid
or we could rely on the children[i].process.err != -1 as start_process
will set it to an actual fd, and when it's done we reset it to -1.

I am unsure if this make it less readable though (all your other
suggestions improve readability a lot!)

>
>> +                     if (i == pp->max_number_processes)
>> +                             /*
>> +                              * waitpid returned another process id
>> +                              * which we are not waiting for.
>> +                              */
>> +                             return;
>> +             }
>> +             strbuf_read_noblock(&pp->err[i], pp->children[i].err, 0);
>
> This is to read leftover output?
>
> As discussed elsewhere, read_nonblock() will have to have "read
> some, not necessarily to the end" semantics to serve the caller in
> run_processes_parallel_buffer_stderr(), so you'd need a loop around
> it here to read until you see EOF.
>
> Or you may be able to just call strbuf_read() and the function may
> do the right thing to read things through to the EOF.  It depends on
> how you redo the patch [2/10].

strbuf_read sounds like the logical choice here (and the redo of 2/10
supports that).


>> +                      * NEEDSWORK:
>> +                      * For now we pick it randomly by doing a round
>> +                      * robin. Later we may want to pick the one with
>> +                      * the most output or the longest or shortest
>> +                      * running process time.
>> +                      */
>> +                     for (i = 0; i < n; i++)
>> +                             if (pp->slots[(pp->foreground_child + i) % n])
>> +                                     break;
>> +                     pp->foreground_child = (pp->foreground_child + i) % n;
>
> ... and then picks a new owner of the output channel.
>
> Up to this point it looks sensible.
>
>> +                     fputs(pp->err[pp->foreground_child].buf, stderr);
>> +                     strbuf_reset(&pp->err[pp->foreground_child]);
>
> I do not think these two lines need to be here, especially if you
> follow the above advice of separating buffering and draining.

They are outputting the buffer, if the next selected child is still running.
I mean it would output eventually anyway, but it would only output after
the next start of processes and poll() is done. Yeah maybe that's what we
want (starting new processes earlier is better than outputting earlier as we're
talking microseconds here).

>
>> +int run_processes_parallel(int n, void *data,
>> +                        get_next_task fn,
>> +                        handle_child_starting_failure fn_err,
>> +                        handle_child_return_value fn_exit)
>> +{
>> +     struct parallel_processes pp;
>> +     run_processes_parallel_init(&pp, n, data, fn, fn_err, fn_exit);
>> +
>> +     while (!pp.all_tasks_started || pp.nr_processes > 0) {
>
> The former is true as long as more_task() says there may be more.
> The latter is true as long as we have something already running.
>
> In either case, we should keep collecting and spawning as needed.
>
>> +             run_processes_parallel_start_new(&pp);
>
> But calling start_new() unconditionally feels sloppy.  It should at
> least be something like
>
>         if (pp.nr_processes < pp.max_processes &&
>             !pp.all_task_started)
>                 start_new_process()
>
> no?

That's the exact condition we have in start_new_process
so I don't want to repeat it here? We could move the while
loop outside of it though. That looks better, done.