Приветствую.

Продолжаем изучать работу PostgreSQL под капотом. Сегодня рассмотрим работу главного цикла бэкэнда. Продолжаем с места где остановились — прямо перед главным циклом. Файл входной точки располагается в src/backend/tcop/postgres.c

Обработчик исключений

Для обработки исключений используется setjmp

Перед началом цикла настраивается обработчик исключений и сохраняется точка стека вызовов для возврата при исключениях.

Обработка исключений работает следующим образом:

• Сбрасываются флаги (например, DoingCommandRead)

• Сбрасываются таймауты

• Откатывается текущая транзакция

• Библиотека libpq откатывается

• Сбрасываются слоты репликации

• Контекст памяти сбрасывается

• Пользователю отправляется сообщение об ошибке

if (sigsetjmp(local_sigjmp_buf, 1) != 0)
{
    // Обработка исключения
}

// Сохранение стека для возврата
PG_exception_stack = &local_sigjmp_buf;

// ...

for (;;)
{
  // Главный цикл
}

jmp_buf *global_buf;

void doSomeStaff();

// Всегда будет возвращать 1
int main() {
  jmp_buf buf;
  if (setjmp(&buf) != 0) {
    printf("Вернулись обратно!")
    return 1;
  }

  global_buf = &buf;
  doSomeStaff();
  return 0;
}

void doSomeStaff() {
  printf("Внутри doSomeStaff");
  longjmp(global_buf, 1);
}

#define PG_TRY()  \
  do { \
      sigjmp_buf *_save_exception_stack = PG_exception_stack; \
      ErrorContextCallback *_save_context_stack = error_context_stack; \
      sigjmp_buf _local_sigjmp_buf; \
      bool _do_rethrow = false; \
      if (sigsetjmp(_local_sigjmp_buf, 0) == 0) \
      { \
          PG_exception_stack = &_local_sigjmp_buf

#define PG_CATCH()	\
  } \
  else \
  { \
      PG_exception_stack = _save_exception_stack; \
      error_context_stack = _save_context_stack

#define PG_FINALLY() \
  } \
  else \
      _do_rethrow = true; \
  { \
      PG_exception_stack = _save_exception_stack; \
      error_context_stack = _save_context_stack

#define PG_END_TRY()  \
    } \
      if (_do_rethrow) \
              PG_RE_THROW(); \
      PG_exception_stack = _save_exception_stack; \
      error_context_stack = _save_context_stack; \
  } while (0)

PG_TRY();
{
  // Ошибка
  if (!IsValidValue(1)) 
  {
    ereport(ERROR,
            (errcode(ERRCODE_SOME_ERRCODE),
            errmsg("value is not valid")));
  }
}
PG_CATCH();
{
    GlobalValue = old_value;
}
PG_END_TRY();

// ===================

PG_TRY();
{
  fwrite("Hello, world!", sizeof(char), 13, handle)
}
PG_FINALLY();
{
    fclose(handle);
}
PG_END_TRY();

После настройки обработчика исключений, мы входим в цикл.

Инициализация/сброс переменных

В начале каждой итерации настраиваются локальные переменные и окружение.

Перед началом принятия и обработки пакета инициализируются переменные:

• MessageContext — контекст памяти при обработке входящего сообщения.

• input_message — хранение строки входящего запроса.

typedef struct StringInfoData
{
    // Буфер, хранящий строку
	char	   *data;
  
	// Длина строки
    int			len;
	
    // Максимально возможная длина строки, 
    // т.к. память была аллоцирована через palloc
    int			maxlen; 
	
    // Переменная для нужд других функций. 
    // Например, отслеживание текущий позиции считывания
    int			cursor;
} StringInfoData;

typedef StringInfoData *StringInfo;

// 1584 строка
if (es->format == EXPLAIN_FORMAT_TEXT)
{
    appendStringInfo(es->str, "  (cost=%.2f..%.2f rows=%.0f width=%d)",
                     plan->startup_cost, plan->total_cost,
                     plan->plan_rows, plan->plan_width);
}

Сигнализация клиенту о готовности принимать запрос

Пакет ReadyForQuery бэкэнд отправляет, когда хочет уведомить фронтэнд о готовности принимать запросы. Чтобы понимать, нужно ли отправлять этот пакет используется флаг send_ready_for_query.

После инициализации (на первом шаге) он будет выставлен, но затем может быть снят

В начале каждой итерации проверяется этот флаг. Если он установлен:

1. Отправляется статистика сборщику статистики.

2. Обновляется заголовок программы соответствующим образом.

3. Инициализируются таймеры таймаутов.

4. Отправляются измененные GUC настройки.

5. Отправляется ReadyForQuery пакет.

6. Флаг send_ready_for_query снимается.

Для первых 3 шагов поведение/параметры зависят от состояния транзакции:

• Простой в транзакции

• Прерванная транзакция

• Остальные

typedef struct TransactionStateData
{
	FullTransactionId fullTransactionId;	/* my FullTransactionId */
	SubTransactionId subTransactionId;	/* my subxact ID */
	char	   *name;			/* savepoint name, if any */
	int			savepointLevel; /* savepoint level */
	TransState	state;			/* low-level state */
	TBlockState blockState;		/* high-level state */
	int			nestingLevel;	/* transaction nesting depth */
	int			gucNestLevel;	/* GUC context nesting depth */
	MemoryContext curTransactionContext;	/* my xact-lifetime context */
	ResourceOwner curTransactionOwner;	/* my query resources */
	TransactionId *childXids;	/* subcommitted child XIDs, in XID order */
	int			nChildXids;		/* # of subcommitted child XIDs */
	int			maxChildXids;	/* allocated size of childXids[] */
	Oid			prevUser;		/* previous CurrentUserId setting */
	int			prevSecContext; /* previous SecurityRestrictionContext */
	bool		prevXactReadOnly;	/* entry-time xact r/o state */
	bool		startedInRecovery;	/* did we start in recovery? */
	bool		didLogXid;		/* has xid been included in WAL record? */
	int			parallelModeLevel;	/* Enter/ExitParallelMode counter */
	bool		chain;			/* start a new block after this one */
	bool		assigned;		/* assigned to top-level XID */
	struct TransactionStateData *parent;	/* back link to parent */
} TransactionStateData;

/*
 * CurrentTransactionState always points to the current transaction state
 * block.  It will point to TopTransactionStateData when not in a
 * transaction at all, or when in a top-level transaction.
 */
static TransactionStateData TopTransactionStateData = {
	.state = TRANS_DEFAULT,
	.blockState = TBLOCK_DEFAULT,
	.assigned = false,
};

static TransactionState CurrentTransactionState = &TopTransactionStateData;

/*
 *	transaction states - transaction state from server perspective
 */
typedef enum TransState
{
	TRANS_DEFAULT,				/* idle */
	TRANS_START,				/* transaction starting */
	TRANS_INPROGRESS,			/* inside a valid transaction */
	TRANS_COMMIT,				/* commit in progress */
	TRANS_ABORT,				/* abort in progress */
	TRANS_PREPARE				/* prepare in progress */
} TransState;

/*
 *	transaction block states - transaction state of client queries
 *
 * Note: the subtransaction states are used only for non-topmost
 * transactions; the others appear only in the topmost transaction.
 */
typedef enum TBlockState
{
	/* not-in-transaction-block states */
	TBLOCK_DEFAULT,				/* idle */
	TBLOCK_STARTED,				/* running single-query transaction */

	/* transaction block states */
	TBLOCK_BEGIN,				/* starting transaction block */
	TBLOCK_INPROGRESS,			/* live transaction */
	TBLOCK_IMPLICIT_INPROGRESS, /* live transaction after implicit BEGIN */
	TBLOCK_PARALLEL_INPROGRESS, /* live transaction inside parallel worker */
	TBLOCK_END,					/* COMMIT received */
	TBLOCK_ABORT,				/* failed xact, awaiting ROLLBACK */
	TBLOCK_ABORT_END,			/* failed xact, ROLLBACK received */
	TBLOCK_ABORT_PENDING,		/* live xact, ROLLBACK received */
	TBLOCK_PREPARE,				/* live xact, PREPARE received */

	/* subtransaction states */
	TBLOCK_SUBBEGIN,			/* starting a subtransaction */
	TBLOCK_SUBINPROGRESS,		/* live subtransaction */
	TBLOCK_SUBRELEASE,			/* RELEASE received */
	TBLOCK_SUBCOMMIT,			/* COMMIT received while TBLOCK_SUBINPROGRESS */
	TBLOCK_SUBABORT,			/* failed subxact, awaiting ROLLBACK */
	TBLOCK_SUBABORT_END,		/* failed subxact, ROLLBACK received */
	TBLOCK_SUBABORT_PENDING,	/* live subxact, ROLLBACK received */
	TBLOCK_SUBRESTART,			/* live subxact, ROLLBACK TO received */
	TBLOCK_SUBABORT_RESTART		/* failed subxact, ROLLBACK TO received */
} TBlockState;

bool
IsTransactionOrTransactionBlock(void)
{
	TransactionState s = CurrentTransactionState;

	if (s->blockState == TBLOCK_DEFAULT)
		return false;

	return true;
}

bool
IsAbortedTransactionBlockState(void)
{
	TransactionState s = CurrentTransactionState;

	if (s->blockState == TBLOCK_ABORT ||
		s->blockState == TBLOCK_SUBABORT)
		return true;

	return false;
}

if (IsAbortedTransactionBlockState())
{
    set_ps_display("idle in transaction (aborted)");
    pgstat_report_activity(STATE_IDLEINTRANSACTION_ABORTED, NULL);

    /* Start the idle-in-transaction timer */
    if (IdleInTransactionSessionTimeout > 0)
    {
        idle_in_transaction_timeout_enabled = true;
        enable_timeout_after(IDLE_IN_TRANSACTION_SESSION_TIMEOUT,
                             IdleInTransactionSessionTimeout);
    }
}
else if (IsTransactionOrTransactionBlock())
{
    set_ps_display("idle in transaction");
    pgstat_report_activity(STATE_IDLEINTRANSACTION, NULL);

    /* Start the idle-in-transaction timer */
    if (IdleInTransactionSessionTimeout > 0)
    {
        idle_in_transaction_timeout_enabled = true;
        enable_timeout_after(IDLE_IN_TRANSACTION_SESSION_TIMEOUT,
                             IdleInTransactionSessionTimeout);
    }
}
else
{
    /*
     * Process incoming notifies (including self-notifies), if
     * any, and send relevant messages to the client.  Doing it
     * here helps ensure stable behavior in tests: if any notifies
     * were received during the just-finished transaction, they'll
     * be seen by the client before ReadyForQuery is.
     */
    if (notifyInterruptPending)
        ProcessNotifyInterrupt(false);

    pgstat_report_stat(false);

    set_ps_display("idle");
    pgstat_report_activity(STATE_IDLE, NULL);

    /* Start the idle-session timer */
    if (IdleSessionTimeout > 0)
    {
        idle_session_timeout_enabled = true;
        enable_timeout_after(IDLE_SESSION_TIMEOUT,
                             IdleSessionTimeout);
    }
}

Чтение запроса

Так как во время чтения и парсинга пришедшего пакета могут прийти асинхронные сигналы. Чтобы учитывать текущее состояние принятия пакета при обработке сигналов, существует флаг DoingCommandRead. Перед началом получения пакета этот флаг выставляется, а после затирается.

Дальше начинается само чтение запроса.

Для получения запроса используется функция ReadCommand. При работе через сокеты внутри происходит делегирование выполнения функции SocketBackend.

Вначале читается первый байт. На его основании определяются максимальный размер буфера записи и тип режима работы: Simple Query или Extended Query режимы.

Когда чтение началось отменить запрос нельзя, поэтому процесс чтения текущего запроса оборачивается макросами HOLD_CANCEL_INTERRUPTS и RESUME_CANCEL_INTERRUPTS. Грубо говоря, они размечают область, при входе в которую не нужно выполнять команды на отмену запросов.

/* ----------------
 *	SocketBackend()		Is called for frontend-backend connections
 *
 *	Returns the message type code, and loads message body data into inBuf.
 *
 *	EOF is returned if the connection is lost.
 * ----------------
 */
static int
SocketBackend(StringInfo inBuf)
{
	int			qtype;
	int			maxmsglen;

	/*
	 * Get message type code from the frontend.
	 */
	HOLD_CANCEL_INTERRUPTS();
	pq_startmsgread();
	qtype = pq_getbyte();

	if (qtype == EOF)			/* frontend disconnected */
	{
		if (IsTransactionState())
			ereport(COMMERROR,
					(errcode(ERRCODE_CONNECTION_FAILURE),
					 errmsg("unexpected EOF on client connection with an open transaction")));
		else
		{
			/*
			 * Can't send DEBUG log messages to client at this point. Since
			 * we're disconnecting right away, we don't need to restore
			 * whereToSendOutput.
			 */
			whereToSendOutput = DestNone;
			ereport(DEBUG1,
					(errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
					 errmsg_internal("unexpected EOF on client connection")));
		}
		return qtype;
	}

	/*
	 * Validate message type code before trying to read body; if we have lost
	 * sync, better to say "command unknown" than to run out of memory because
	 * we used garbage as a length word.  We can also select a type-dependent
	 * limit on what a sane length word could be.  (The limit could be chosen
	 * more granularly, but it's not clear it's worth fussing over.)
	 *
	 * This also gives us a place to set the doing_extended_query_message flag
	 * as soon as possible.
	 */
	switch (qtype)
	{
		case 'Q':				/* simple query */
			maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
			doing_extended_query_message = false;
			break;

		case 'F':				/* fastpath function call */
			maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
			doing_extended_query_message = false;
			break;

		case 'X':				/* terminate */
			maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
			doing_extended_query_message = false;
			ignore_till_sync = false;
			break;

		case 'B':				/* bind */
		case 'P':				/* parse */
			maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
			doing_extended_query_message = true;
			break;

		case 'C':				/* close */
		case 'D':				/* describe */
		case 'E':				/* execute */
		case 'H':				/* flush */
			maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
			doing_extended_query_message = true;
			break;

		case 'S':				/* sync */
			maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
			/* stop any active skip-till-Sync */
			ignore_till_sync = false;
			/* mark not-extended, so that a new error doesn't begin skip */
			doing_extended_query_message = false;
			break;

		case 'd':				/* copy data */
			maxmsglen = PQ_LARGE_MESSAGE_LIMIT;
			doing_extended_query_message = false;
			break;

		case 'c':				/* copy done */
		case 'f':				/* copy fail */
			maxmsglen = PQ_SMALL_MESSAGE_LIMIT;
			doing_extended_query_message = false;
			break;

		default:

			/*
			 * Otherwise we got garbage from the frontend.  We treat this as
			 * fatal because we have probably lost message boundary sync, and
			 * there's no good way to recover.
			 */
			ereport(FATAL,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("invalid frontend message type %d", qtype)));
			maxmsglen = 0;		/* keep compiler quiet */
			break;
	}

	/*
	 * In protocol version 3, all frontend messages have a length word next
	 * after the type code; we can read the message contents independently of
	 * the type.
	 */
	if (pq_getmessage(inBuf, maxmsglen))
		return EOF;				/* suitable message already logged */
	RESUME_CANCEL_INTERRUPTS();

	return qtype;
}

static int
InteractiveBackend(StringInfo inBuf)
{
	int			c;				/* character read from getc() */

	/*
	 * display a prompt and obtain input from the user
	 */
	printf("backend> ");
	fflush(stdout);

	resetStringInfo(inBuf);

	/*
	 * Read characters until EOF or the appropriate delimiter is seen.
	 */
	while ((c = interactive_getc()) != EOF)
	{
		if (c == '\n')
		{
			if (UseSemiNewlineNewline)
			{
				/*
				 * In -j mode, semicolon followed by two newlines ends the
				 * command; otherwise treat newline as regular character.
				 */
				if (inBuf->len > 1 &&
					inBuf->data[inBuf->len - 1] == '\n' &&
					inBuf->data[inBuf->len - 2] == ';')
				{
					/* might as well drop the second newline */
					break;
				}
			}
			else
			{
				/*
				 * In plain mode, newline ends the command unless preceded by
				 * backslash.
				 */
				if (inBuf->len > 0 &&
					inBuf->data[inBuf->len - 1] == '\\')
				{
					/* discard backslash from inBuf */
					inBuf->data[--inBuf->len] = '\0';
					/* discard newline too */
					continue;
				}
				else
				{
					/* keep the newline character, but end the command */
					appendStringInfoChar(inBuf, '\n');
					break;
				}
			}
		}

		/* Not newline, or newline treated as regular character */
		appendStringInfoChar(inBuf, (char) c);
	}

	/* No input before EOF signal means time to quit. */
	if (c == EOF && inBuf->len == 0)
		return EOF;

	/*
	 * otherwise we have a user query so process it.
	 */

	/* Add '\0' to make it look the same as message case. */
	appendStringInfoChar(inBuf, (char) '\0');

	/*
	 * if the query echo flag was given, print the query..
	 */
	if (EchoQuery)
		printf("statement: %s\n", inBuf->data);
	fflush(stdout);

	return 'Q';
}

После завершения принятия сообщения сбрасываются таймауты ожиданий:

• IDLE_IN_TRANSACTION_SESSION_TIMEOUT — таймаут ожидания в транзакции;

• IDLE_SESSION_TIMEOUT — таймаут ожидания в сессии.

Проверка внешних прерываний (сигналы)

Во время чтения команды от пользователя была заблокирована обработка сигналов. Теперь мы можем проверить наличие пришедших сигналов. За обработку сигналов отвечает функция ProcessInterrupts().

/* Test whether an interrupt is pending */
#ifndef WIN32
#define INTERRUPTS_PENDING_CONDITION() \
	(unlikely(InterruptPending))
#else
#define INTERRUPTS_PENDING_CONDITION() \
	(unlikely(UNBLOCKED_SIGNAL_QUEUE()) ? pgwin32_dispatch_queued_signals() : 0, \
	 unlikely(InterruptPending))
#endif

/*
 * Hints to the compiler about the likelihood of a branch. Both likely() and
 * unlikely() return the boolean value of the contained expression.
 *
 * These should only be used sparingly, in very hot code paths. It's very easy
 * to mis-estimate likelihoods.
 */
#if __GNUC__ >= 3
#define likely(x)	__builtin_expect((x) != 0, 1)
#define unlikely(x) __builtin_expect((x) != 0, 0)
#else
#define likely(x)	((x) != 0)
#define unlikely(x) ((x) != 0)
#endif

В начале ProcessInterrupts имеется проверка на нахождение в особенных секциях.

1. CritSectionCount — количество исполняемых критических секций. Например, во время построение Gist индекса, мы входим в такую критическую секцию (src/backend/access/gist/gistbuild.c)

IndexBuildResult *
gistbuild(Relation heap, Relation index, IndexInfo *indexInfo)
{
  // ...
  START_CRIT_SECTION();
  
  GISTInitBuffer(buffer, F_LEAF);
  
  MarkBufferDirty(buffer);
  PageSetLSN(page, GistBuildLSN);
  
  UnlockReleaseBuffer(buffer);
  
  END_CRIT_SECTION();
  // ...
}

2. InterruptHoldoffCount — количество секций, в которых исполнение обработчиков нежелательно. Например, обработчик SIGALRM входит в такую секцию

static void
handle_sig_alarm(SIGNAL_ARGS)
{
	/*
	 * Bump the holdoff counter, to make sure nothing we call will process
	 * interrupts directly. No timeout handler should do that, but these
	 * failures are hard to debug, so better be sure.
	 */
	HOLD_INTERRUPTS();

	// ...

	RESUME_INTERRUPTS();
}

В начале ProcessInterrupts проверяется, что обе мы не находимся ни в одной из перечисленных секций.

Внутри функции проверяется, что нужно:

• Завершить процесс

• Проверить подключение к клиенту

• Провести восстановление после конфликта

• Отменить запрос

• Обработать превышения таймаутов ожидания

• Выполнить параллельные запросы

• Вывести в лог описание текущего контекста памяти

Отдельно проверяются изменения GUC конфигурации. При их изменении, они отсылаются клиенту.

Обработка входного запроса

Когда пакет получен и все подготовки завершены, мы готовы обрабатывать запрос. На основании первого байта пакета определяются следующие шаги.

Simple Query

Сообщение Simple Query просто исполняется. Но если процесс — WALSender, то выполнение происходит только в случае, если сообщение репликационное.

case 'Q':			/* simple query */
  {
      const char *query_string;
  
      /* Set statement_timestamp() */
      SetCurrentStatementStartTimestamp();
  
      query_string = pq_getmsgstring(&input_message);
      pq_getmsgend(&input_message);
  
      if (am_walsender)
      {
          if (!exec_replication_command(query_string))
              exec_simple_query(query_string);
      }
      else
          exec_simple_query(query_string);
  
      send_ready_for_query = true;
  }
  break;

extern void replication_scanner_init(const char *query_string);
extern void replication_scanner_finish(void);
extern bool replication_scanner_is_replication_command(void);

/*
 * Check to see if the first token of a command is a WalSender keyword.
 *
 * To keep repl_scanner.l minimal, we don't ask it to know every construct
 * that the core lexer knows.  Therefore, we daren't lex more than the
 * first token of a general SQL command.  That will usually look like an
 * IDENT token here, although some other cases are possible.
 */
bool
replication_scanner_is_replication_command(void)
{
	int			first_token = replication_yylex();

	switch (first_token)
	{
		case K_IDENTIFY_SYSTEM:
		case K_BASE_BACKUP:
		case K_START_REPLICATION:
		case K_CREATE_REPLICATION_SLOT:
		case K_DROP_REPLICATION_SLOT:
		case K_TIMELINE_HISTORY:
		case K_SHOW:
			/* Yes; push back the first token so we can parse later. */
			repl_pushed_back_token = first_token;
			return true;
		default:
			/* Nope; we don't bother to push back the token. */
			return false;
	}
}

Parse

В случае подготовленного запроса нужно не только почитать строку запроса, но и получить все параметры. Так как типы в Postgres являются объектами, то и типы параметров передаются в виде идентификаторов объектов.

case 'P':			/* parse */
    {
        const char *stmt_name;
        const char *query_string;
        int			numParams;
        Oid		   *paramTypes = NULL;
    
        forbidden_in_wal_sender(firstchar);
    
        /* Set statement_timestamp() */
        SetCurrentStatementStartTimestamp();
    
        stmt_name = pq_getmsgstring(&input_message);
        query_string = pq_getmsgstring(&input_message);
        numParams = pq_getmsgint(&input_message, 2);
        if (numParams > 0)
        {
            paramTypes = (Oid *) palloc(numParams * sizeof(Oid));
            for (int i = 0; i < numParams; i++)
                paramTypes[i] = pq_getmsgint(&input_message, 4);
        }
        pq_getmsgend(&input_message);
    
        exec_parse_message(query_string, stmt_name,
                           paramTypes, numParams);
    }
    break;

Также можно заметить функцию forbidden_in_wal_sender. Эта функция инициирует ошибку, если WALSender получил недопустимую команду - он работает только в режиме Simple Query.

/*
 * Throw an error if we're a WAL sender process.
 *
 * This is used to forbid anything else than simple query protocol messages
 * in a WAL sender process.  'firstchar' specifies what kind of a forbidden
 * message was received, and is used to construct the error message.
 */
static void
forbidden_in_wal_sender(char firstchar)
{
	if (am_walsender)
	{
		if (firstchar == 'F')
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("fastpath function calls not supported in a replication connection")));
		else
			ereport(ERROR,
					(errcode(ERRCODE_PROTOCOL_VIOLATION),
					 errmsg("extended query protocol not supported in a replication connection")));
	}
}

Bind

Парсинг BIND пакета (как описывается в комментарии) довольно затратный. Поэтому практически вся логика вынесена в отдельную функцию.

case 'B':			/* bind */
    forbidden_in_wal_sender(firstchar);
  
    /* Set statement_timestamp() */
    SetCurrentStatementStartTimestamp();
  
    /*
     * this message is complex enough that it seems best to put
     * the field extraction out-of-line
     */
    exec_bind_message(&input_message);
    break;

Execute

Если на вход пришел EXECUTE пакет, то из него получаются название портала для выполнения и количество строк для возврата. Они передаются самой функции выполнения.

case 'E':			/* execute */
    {
        const char *portal_name;
        int			max_rows;
  
        forbidden_in_wal_sender(firstchar);
  
        /* Set statement_timestamp() */
        SetCurrentStatementStartTimestamp();
  
        portal_name = pq_getmsgstring(&input_message);
        max_rows = pq_getmsgint(&input_message, 4);
        pq_getmsgend(&input_message);
  
        exec_execute_message(portal_name, max_rows);
    }
    break;

Fastpath function call

В отличие от предыдущих пакетов, бизнес логика FASTPATH частично вынесена наружу. Именно: старт транзакции и переключение контекста памяти.

case 'F':			/* fastpath function call */
    forbidden_in_wal_sender(firstchar);
  
    /* Set statement_timestamp() */
    SetCurrentStatementStartTimestamp();
  
    /* Report query to various monitoring facilities. */
    pgstat_report_activity(STATE_FASTPATH, NULL);
    set_ps_display("<FASTPATH>");
  
    /* start an xact for this function invocation */
    start_xact_command();
  
    /*
     * Note: we may at this point be inside an aborted
     * transaction.  We can't throw error for that until we've
     * finished reading the function-call message, so
     * HandleFunctionRequest() must check for it after doing so.
     * Be careful not to do anything that assumes we're inside a
     * valid transaction here.
     */
  
    /* switch back to message context */
    MemoryContextSwitchTo(MessageContext);
  
    HandleFunctionRequest(&input_message);
  
    /* commit the function-invocation transaction */
    finish_xact_command();
  
    send_ready_for_query = true;
    break;

P.S. В комментарии сказано не добавлять код, предполагающий его исполнение в корректном состоянии (не прерванной транзакции), хотя этот комментарий нужно вынести из switch’а, т.к. подобное относится ко всей итерации цикла.

Close

В Расширенном запросе можно создавать порталы и именованные запросы. Они хранятся в памяти, поэтому их стоит закрывать. Для этого используется пакет CLOSE и одноименная SQL команда.  

В пакете передаются тип объекта для закрытия и его название. 

case 'C':			/* close */
    {
        int			close_type;
        const char *close_target;
  
        forbidden_in_wal_sender(firstchar);
  
        close_type = pq_getmsgbyte(&input_message);
        close_target = pq_getmsgstring(&input_message);
        pq_getmsgend(&input_message);
  
        switch (close_type)
        {
            case 'S':
                if (close_target[0] != '\0')
                    DropPreparedStatement(close_target, false);
                else
                {
                    /* special-case the unnamed statement */
                    drop_unnamed_stmt();
                }
                break;
            case 'P':
                {
                    Portal		portal;
  
                    portal = GetPortalByName(close_target);
                    if (PortalIsValid(portal))
                        PortalDrop(portal, false);
                }
                break;
            default:
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg("invalid CLOSE message subtype %d",
                                close_type)));
                break;
        }
  
        if (whereToSendOutput == DestRemote)
            pq_putemptymessage('3');	/* CloseComplete */
    }
    break;

Describe

От бэкэнда можно получить описание порталов и именованных выражений:

• Портал — описание возвращаемых данных (название, тип данных и т. д.).

• Выражения — описание передаваемых параметров.

Так же как и в CLOSE, в DESCRIBE пакете передаются тип объекта для описания и его название

case 'D':			/* describe */
    {
        int			describe_type;
        const char *describe_target;
  
        forbidden_in_wal_sender(firstchar);
  
        /* Set statement_timestamp() (needed for xact) */
        SetCurrentStatementStartTimestamp();
  
        describe_type = pq_getmsgbyte(&input_message);
        describe_target = pq_getmsgstring(&input_message);
        pq_getmsgend(&input_message);
  
        switch (describe_type)
        {
            case 'S':
                exec_describe_statement_message(describe_target);
                break;
            case 'P':
                exec_describe_portal_message(describe_target);
                break;
            default:
                ereport(ERROR,
                        (errcode(ERRCODE_PROTOCOL_VIOLATION),
                         errmsg("invalid DESCRIBE message subtype %d",
                                describe_type)));
                break;
        }
    }
    break;

Flush

В процессе работы во внутреннем буфере могла накопиться информация. Для немедленного сброса буфера отправки, определена команда FLUSH

case 'H':			/* flush */
    pq_getmsgend(&input_message);
    if (whereToSendOutput == DestRemote)
        pq_flush();
    break;

Sync

SYNC пакет используется для коммита транзакции. Логика работы простая — просто вызывается функция коммита

case 'S':			/* sync */
    pq_getmsgend(&input_message);
    finish_xact_command();
    send_ready_for_query = true;
    break;

Terminate

Закрыть соединения можно 2 способами: оборвать соединение и послать пакет закрытия соединения. Оба этих случая учтены.

    /*
     * 'X' means that the frontend is closing down the socket. EOF
     * means unexpected loss of frontend connection. Either way,
     * perform normal shutdown.
     */
case EOF:

    /* for the statistics collector */
    pgStatSessionEndCause = DISCONNECT_CLIENT_EOF;

    /* FALLTHROUGH */

case 'X':

    /*
     * Reset whereToSendOutput to prevent ereport from attempting
     * to send any more messages to client.
     */
    if (whereToSendOutput == DestRemote)
        whereToSendOutput = DestNone;

    /*
     * NOTE: if you are tempted to add more code here, DON'T!
     * Whatever you had in mind to do should be set up as an
     * on_proc_exit or on_shmem_exit callback, instead. Otherwise
     * it will fail to be called during other backend-shutdown
     * scenarios.
     */
    proc_exit(0);

Дополнительно

Также в процессе разработки обнаружился случай, когда клиент продолжает посылать пакеты команд связанными с копированием. Такие пакеты просто игноруруются.

case 'd':			/* copy data */
case 'c':			/* copy done */
case 'f':			/* copy fail */

    /*
     * Ac!cept but ignore these messages, per protocol spec; we
     * probably got here because a COPY failed, and the frontend
     * is still sending data.
     */
    break;

Также обрабатывается неизвестный тип пакета.

default:
    ereport(FATAL,
            (errcode(ERRCODE_PROTOCOL_VIOLATION),
             errmsg("invalid frontend message type %d",
                    firstchar)));

Конец

На этом итерация заканчивается и начинается новая (если не было исключений)