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

Продолжаем серию постов о работе PostgreSQL на уровне кода. Сегодня, познакомимся с главным циклом сервера. Цикл расположен в src/backend/postmaster/postmaster.c

Предыдущий пост

Инициализация

В начале цикла инициализируем переменные времени для проверок локфайлов текущим временем:

• last_lockfile_recheck_time - время последней проверки postmaster.pid;

• last_touch_time - время последней проверки сокет-файлов.

Далее инициализируем прослушиваемые порты. 

Для прослушивания используется мультиплексирование, используя select(). Инициализируем множество прослушиваемых портов и одновременно находим значение числа наибольшего дескриптора.

// nSockets - число наибольшего дескриптора + 1
nSockets = initMasks(&readmask);
for (;;)
{
    // Копируем перед использованием
    memcpy((char *) &rmask, (char *) &readmask, sizeof(fd_set));
    // Рабочий код
}

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

* Цикл создан через for(;;), а не while (1).

Бесконечный цикл

Каждую итерацию цикла можно разбить на несколько логических секций.

Начальная часть определяется текущим состоянием:

• PM_WAIT_DEAD_END - Спим

• Остальное - Обрабатываем входящие подключения

Спим

Если Postmaster в состоянии PM_WAIT_DEAD_END , то ждем пока dead_end бэкэнды завершатся. Для этого просто засыпаем на 100 мс.

Почему на 100 мс, объяснений нет:

pg_usleep(100000L); /* 100 msec seems reasonable */

Обработка входящих подключений

Первым делом копируем переменную дескриптора портов в локальную, чтобы не затирать порты между обработкой клиентов в select().

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

Как расчитывается таймаут:

• Нормальное состояние: 60 секунд;

• Принят SIGKILL: время ожидания = 5 - (ТЕКУЩЕЕ ВРЕМЯ - ВРЕМЯ ПОЛУЧЕНИЯ SIGKILL);

• Запрошен старт воркера: ждем 0 секунд;

  Тогда select() сделает возврат немедленно, а значит процесс старта воркера запустится как можно скорее;

• Воркер упал: таймаут - наибольшее время необходимое для восстановление воркера, но не более 1 минуты.

Теперь начинаем принимать подключения:

selres = select(nSockets, &rmask, NULL, NULL, &timeout);

Как только функция вернулась проверяем результат.

Если select() вернул ошибку (-1), то падаем.

Если select() вернул не 0, то на какой-то порт пришел запрос. Проходимся по всем портам, и для каждого, на который постучались:

1. Создаем соединение

Соединение создается через сокеты:

• Принимается сокетное соединение на порту - accept()

• Сохраняется адрес - getsockname()

• Если TCP - настраиваем

  • TCP_NODELAY

  • TCP_KEEPALIVE

  • Оптимизация буфера для Windows

int optlen;
int newopt;

optlen = sizeof(oldopt);
if (getsockopt(port->sock, SOL_SOCKET, SO_SNDBUF, (char *) &oldopt,
               &optlen) < 0)
{
    ereport(LOG,
            (errmsg("%s(%s) failed: %m", "getsockopt", "SO_SNDBUF")));
    return STATUS_ERROR;
}
newopt = PQ_SEND_BUFFER_SIZE * 4;
if (oldopt < newopt)
{
    if (setsockopt(port->sock, SOL_SOCKET, SO_SNDBUF, (char *) &newopt,
                   sizeof(newopt)) < 0)
    {
        ereport(LOG,
                (errmsg("%s(%s) failed: %m", "setsockopt", "SO_SNDBUF")));
        return STATUS_ERROR;
    }
}

2. Старт бэкэнда

Создаем структуру Бэкэнда:

typedef struct bkend
{
	pid_t		pid;			/* process id of backend */
	int32		cancel_key;		/* cancel key for cancels for this backend */
	int			child_slot;		/* PMChildSlot for this backend, if any */
	int			bkend_type;		/* child process flavor, see above */
	bool		dead_end;		/* is it going to send an error and quit? */
	bool		bgworker_notify;	/* gets bgworker start/stop notifications */
	dlist_node	elem;			/* list link in BackendList */
} Backend;

И инициализируем его:

• Выделяем память

• Создаем CancelKey

• Определяем может ли бэкэнд принимать соединения (он может стартовать сразу в DEAD_END)

Запускаем его - форкаемся. Бэкэнд уходит в свой процесс и работает там до конца, не возвращается.

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

#ifndef WIN32
#define PG_SETMASK(mask)	sigprocmask(SIG_SETMASK, mask, NULL)
#else
/* Emulate POSIX sigset_t APIs on Windows */
typedef int sigset_t;

extern int	pqsigsetmask(int mask);

#define PG_SETMASK(mask)		pqsigsetmask(*(mask))
#define sigemptyset(set)		(*(set) = 0)
#define sigfillset(set)			(*(set) = ~0)
#define sigaddset(set, signum)	(*(set) |= (sigmask(signum)))
#define sigdelset(set, signum)	(*(set) &= ~(sigmask(signum)))
#endif							/* WIN32 */

PG_SETMASK(&UnBlockSig);

selres = select(nSockets, &rmask, NULL, NULL, &timeout);

PG_SETMASK(&BlockSig);

Проверка вспомогательных процессов

Postgres - многопроцессное приложение. Если какой-то процесс упадет, то система продолжит работать.

Как узнаем, что дочерний процесс упал? Делать регулярные проверки. Postmaster хранит PID'ы основных процессов:

/* PIDs of special child processes; 0 when not running */
static pid_t StartupPID = 0,
			BgWriterPID = 0,
			CheckpointerPID = 0,
			WalWriterPID = 0,
			WalReceiverPID = 0,
			AutoVacPID = 0,
			PgArchPID = 0,
			PgStatPID = 0,
			SysLoggerPID = 0;

После обработки пришедшего сигнала, смотрим упал ли какой-нибудь воркер, и поднимаем при необходимости. Для многих процессов есть свои условия для запуска. Например:

/*
 * If no background writer process is running, and we are not in a
 * state that prevents it, start one.  It doesn't matter if this
 * fails, we'll just try again later.  Likewise for the checkpointer.
 */
if (pmState == PM_RUN || pmState == PM_RECOVERY ||
    pmState == PM_HOT_STANDBY)
{
    if (CheckpointerPID == 0)
        CheckpointerPID = StartCheckpointer();
    if (BgWriterPID == 0)
        BgWriterPID = StartBackgroundWriter();
}

Проверке подвергаются также и бэкэнды: проходимся по всем и проверяем их статус.

Проверка локфайлов

При запуске, был создан локфайл postmaster.pid. В нем хранится рабочая информация.

В частности, первая строка хранит PID запущенного Postmaster. Может ли получиться так, что 2 процесса запущено одновременно? Да, например, на различных портах. Лучше, чтобы такого не происходило, иначе могут одновременно измениться несколько файлов. Ничего хорошего это не принесет.

Раз в минуту проверяем файл. Если его нет или хранящийся в нем PID не равен нашему, то значит новый процесс БД запустился. В таком случае, экстренно закрываемся - отправляем SIGQUIT самому себе.

now = time(NULL);

if (now - last_lockfile_recheck_time >= 1 * SECS_PER_MINUTE)
{
    if (!RecheckDataDirLockFile())
    {
        ereport(LOG,
                (errmsg("performing immediate shutdown because data directory lock file is invalid")));
        kill(MyProcPid, SIGQUIT);
    }
    last_lockfile_recheck_time = now;
}

Не забываем про UNIX сокеты.

Они сохраняются в директории, указанной в настройке unix_socket_directories в postgresql.conf. По умолчанию, равен /tmp. В нем хранятся временные файлы. Чтобы их не скопилось слишком много, часто запущены процессы очищающие директорию. Чтобы они случайно не удалили сокет-файл, будем обновлять время последнего изменения файла - вызывать touch().

if (now - last_touch_time >= 58 * SECS_PER_MINUTE)
{
    TouchSocketFiles();
    TouchSocketLockFiles();
    last_touch_time = now;
}

Конец итерации

На этом итерация заканчивается и заходим на новый круг.