Function process_pipe_input

src/backend/postmaster/syslogger.c:1470–1653

* Process data received through the syslogger pipe. * * This routine interprets the log pipe protocol which sends log messages as * (hopefully atomic) chunks - such chunks are detected and reassembled here. * * The protocol has a header that starts with two nul bytes, then has a 16 bit * length, the pid of the sending process, and a flag to indicate if it is * the last chunk in a message. I

 * logbuffer, and *bytes_in_logbuffer is updated.
 */
static void
process_pipe_input(char *logbuffer, int *bytes_in_logbuffer)
{
	char	   *cursor = logbuffer;
	int			count = *bytes_in_logbuffer;
	int			dest = LOG_DESTINATION_STDERR;

	/* While we have enough for a header, process data... */
	while (count >= sizeof(PipeProtoHeader))
	{
		PipeProtoHeader p;
		int chunklen;

		/* Do we have a valid header? */
		memcpy(&p, cursor, PIPE_HEADER_SIZE);
		if (chunk_is_postgres_chunk(&p))
		{
			List	   *buffer_list;
			ListCell   *cell;
			save_buffer *existing_slot = NULL,
					   *free_slot = NULL;
			StringInfo	str;

			chunklen = PIPE_HEADER_SIZE + p.len;

			/* Fall out of loop if we don't have the whole chunk yet */
			if (count < chunklen)
				break;

			dest = (p.log_format == 'c' || p.log_format == 'f') ?
				LOG_DESTINATION_CSVLOG : LOG_DESTINATION_STDERR;

			/* Locate any existing buffer for this source pid */
			buffer_list = buffer_lists[p.pid % NBUFFER_LISTS];
			foreach(cell, buffer_list)
			{
				save_buffer *buf = (save_buffer *) lfirst(cell);

				/*
				 * Different threads in the same process may write
				 * log messages concurrently. The chunk messages
				 * should be treat differently.
				 */
				if (buf->pid == p.pid && buf->tid == p.thid)
				{
					existing_slot = buf;
					break;
				}
				if (buf->pid == 0 && free_slot == NULL)
					free_slot = buf;
			}

			if (p.is_last == 'f' || p.is_last == 'F')
			{
				/*
				 * Save a complete non-final chunk in a per-pid buffer
				 */
				if (existing_slot != NULL)