Function HandleParallelMessage

src/backend/access/transam/parallel.c:1151–1253

* Handle a single protocol message received from a single parallel worker. */

 * Handle a single protocol message received from a single parallel worker.
 */
static void
HandleParallelMessage(ParallelContext *pcxt, int i, StringInfo msg)
{
	char		msgtype;

	if (pcxt->known_attached_workers != NULL &&
		!pcxt->known_attached_workers[i])
	{
		pcxt->known_attached_workers[i] = true;
		pcxt->nknown_attached_workers++;
	}

	msgtype = pq_getmsgbyte(msg);

	switch (msgtype)
	{
		case 'K':				/* BackendKeyData */
			{
				int32		pid = pq_getmsgint(msg, 4);

				(void) pq_getmsgint(msg, 4);	/* discard cancel key */
				(void) pq_getmsgend(msg);
				pcxt->worker[i].pid = pid;
				break;
			}

		case 'E':				/* ErrorResponse */
		case 'N':				/* NoticeResponse */
			{
				ErrorData	edata;
				ErrorContextCallback *save_error_context_stack;

				/* Parse ErrorResponse or NoticeResponse. */
				pq_parse_errornotice(msg, &edata);

				/* Death of a worker isn't enough justification for suicide. */
				edata.elevel = Min(edata.elevel, ERROR);

				/*
				 * If desired, add a context line to show that this is a
				 * message propagated from a parallel worker.  Otherwise, it
				 * can sometimes be confusing to understand what actually
				 * happened.  (We don't do this in FORCE_PARALLEL_REGRESS mode
				 * because it causes test-result instability depending on
				 * whether a parallel worker is actually used or not.)
				 */
				if (force_parallel_mode != FORCE_PARALLEL_REGRESS)
				{
					if (edata.context)
						edata.context = psprintf("%s\n%s", edata.context,
												 _("parallel worker"));
					else
						edata.context = pstrdup(_("parallel worker"));
				}

				/*
				 * Context beyond that should use the error context callbacks
				 * that were in effect when the ParallelContext was created,