Function HandleParallelMessages

src/backend/access/transam/parallel.c:1062–1146

* Handle any queued protocol messages received from parallel workers. */

 * Handle any queued protocol messages received from parallel workers.
 */
void
HandleParallelMessages(void)
{
	dlist_iter	iter;
	MemoryContext oldcontext;

	static MemoryContext hpm_context = NULL;

	/*
	 * This is invoked from ProcessInterrupts(), and since some of the
	 * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
	 * for recursive calls if more signals are received while this runs.  It's
	 * unclear that recursive entry would be safe, and it doesn't seem useful
	 * even if it is safe, so let's block interrupts until done.
	 */
	HOLD_INTERRUPTS();

	/*
	 * Moreover, CurrentMemoryContext might be pointing almost anywhere.  We
	 * don't want to risk leaking data into long-lived contexts, so let's do
	 * our work here in a private context that we can reset on each use.
	 */
	if (hpm_context == NULL)	/* first time through? */
		hpm_context = AllocSetContextCreate(TopMemoryContext,
											"HandleParallelMessages",
											ALLOCSET_DEFAULT_SIZES);
	else
		MemoryContextReset(hpm_context);

	oldcontext = MemoryContextSwitchTo(hpm_context);

	/* OK to process messages.  Reset the flag saying there are more to do. */
	ParallelMessagePending = false;

	dlist_foreach(iter, &pcxt_list)
	{
		ParallelContext *pcxt;
		int			i;

		pcxt = dlist_container(ParallelContext, node, iter.cur);
		if (pcxt->worker == NULL)
			continue;

		for (i = 0; i < pcxt->nworkers_launched; ++i)
		{
			/*
			 * Read as many messages as we can from each worker, but stop when
			 * either (1) the worker's error queue goes away, which can happen
			 * if we receive a Terminate message from the worker; or (2) no
			 * more messages can be read from the worker without blocking.
			 */
			while (pcxt->worker[i].error_mqh != NULL)
			{
				shm_mq_result res;
				Size		nbytes;
				void	   *data;

				res = shm_mq_receive(pcxt->worker[i].error_mqh, &nbytes,