Re: [RFC PATCH 2/3] run-commands: add an async queue processor

From: Junio C Hamano <gitster@pobox.com>
Date: Aug 21, 2015, 20:47 UTC
Message-ID: <xmqqy4h44bdp.fsf@gitster.dls.corp.google.com>
In-Reply-To: <CAGZ79kZrcPAAt+miHDGQp=052S-q=JaKvvLgKHaPG+G6cDjBtg@mail.gmail.com>

Stefan Beller <sbeller@google.com> writes:

Show 30 quoted lines

>>> +struct task_queue *create_task_queue(unsigned max_threads)
>>> +{
>>> +     struct task_queue *aq = xmalloc(sizeof(*aq));
>>> +
>>> +#ifndef NO_PTHREADS
>>> +     int i;
>>> +     if (!max_threads)
>>> +             aq->max_threads = online_cpus();
>>> +     else
>>> +             aq->max_threads = max_threads;
>>> +
>>> +     sem_init(&aq->mutex, 0, 1);
>>> +     sem_init(&aq->workingcount, 0, 0);
>>> +     sem_init(&aq->freecount, 0, aq->max_threads);
>>> +     aq->threads = xmalloc(aq->max_threads * sizeof(pthread_t));
>>> +
>>> +     for (i = 0; i < aq->max_threads; i++)
>>> +             pthread_create(&aq->threads[i], 0, &dispatcher, aq);
>>> +
>>> +     aq->first = NULL;
>>> +     aq->last = NULL;
>>
>>
>> Shouldn't these be initialized before letting threads call into
>> dispatcher?  The workingcount semaphore that is initialized to 0 may
>> prevent them from peeking into these pointers and barfing, but still...
>
> They are initialized to NULL as the empty queue doesn't need a
> container element.
> Do we do queues in another way usually?

I do not think we are on the same wavelength. What I meant was to do this:

	aq = xmalloc(...);
        set up _everything_ in aq and make it a consistent state;
        /* aq->first and aq->last are part of _everything_ in aq */
        for (many times)
        	pthread_create(...);

	/* No aq->first = aq->last = NULL assignment here */

instead of

	aq = xmalloc(...);
        set up part of aq;
        for (many times)
        	pthread_create(...);
	belatedly initialize aq->first and aq->last and finally
        aq becomes a consistent state.

which is what we see above. The latter works _only_ because the threads created are blocked waiting on aq->workingcount which is initialized to block before threads are created to run dispatch, and one of the early things dispatch does is to try acquiring that semaphore to block before accessing aq->first and aq->last.