Silk (Shardman InterLinK) — это экспериментальная транспортная функциональность. Она подключается в тот момент, когда postgres_fdw решает передать восстановленный фрагмент запроса через соединение libpq на удалённый узел, заменяя собой соединение libpq. Эта функциональность предназначена для уменьшения количества незанятых подключений postgres_fdw во время выполнения транзакции, минимизации задержки и повышения общей пропускной способности.

Реализация Silk использует несколько обслуживающих процессов. Основной процесс маршрутизации/мультиплексирования (по одному на экземпляр PostgreSQL) называется silkroad, а группа обслуживающих процессов — silkworms. Хотя postgres_fdw использует libpq, оно порождает несколько подключений libpq от каждого обслуживающего процесса к удалённому узлу (где соответственно порождаются несколько обслуживающих процессов). Но если silk заменяет libpq, каждый процесс silkroad подключается только к одному удалённому процессу silkroad. В этой схеме удалённые silkworms играют роль удалённых обслуживающих процессов, в противном случае порождённых postgres_fdw.

Silkroad связывает локальный обслуживающий процесс с рабочими процессами удалённого узла следующим образом:

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

Процесс Silkroad запускает цикл событий на базе библиотеки libev. Очередь общей памяти каждого обслуживающего процесса отображается в цикле обработки событий с помощью дескриптора eventfd, а каждое сетевое подключение — с помощью дескриптора сокета.

Во время запуска обслуживающий процесс регистрирует себя (свои дескрипторы eventfd) в локальном процессе silkroad. Silkroad отвечает, указывая, какие сегменты памяти использовать для очереди сообщений обслуживающего процесса. С этого момента silkroad будет реагировать на события из очереди, связанной с этим обслуживающим процессом. Сетевые подключения между локальными и удалёнными silkroads будут установлены сразу по первому запросу от обслуживающего процесса к удалённому узлу и будут существовать до тех пор, пока оба участника (процессы silkroad) существуют.

Для каждого подзапроса ожидается подмножество кортежей и поэтому взаимодействие внутри подзапроса представляется как двунаправленный поток данных. Silkroad использует внутреннюю таблицу маршрутизации для регистрации этих потоков. Уникальный идентификатор потока (в кластере Shardman) формируется как пара «адрес исходного узла, адрес целевого узла» и локально (внутри узла) уникальный номер. Каждый конкретный подзапрос от обслуживающего процесса к удалённым узлам будет зарегистрирован silkroad как такой поток. Таким образом, любой обслуживающий процесс может быть связан со многими потоками одновременно.

Когда локальный процесс silkroad получает от серверной части сообщение с новым идентификатором потока, он регистрирует его в локальной таблице маршрутизации, а затем перенаправляет это сообщение в соответствующий сокет. Если подключение с удалённым silkroad не существует, оно устанавливается с помощью процедуры согласования. Исходное сообщение, инициировавшее согласование, помещается в специальный внутренний буфер до тех пор, пока согласование не завершится успешно. Удалённый процесс silkroad, получив пакет с новым идентификатором, регистрирует его в своей таблице, затем назначает рабочий процесс silkworm из пула доступных рабочих процессов и помещает сообщение в очередь общей памяти рабочего процесса. Если все рабочие процессы silkworm в данный момент заняты, сообщение будет отложено, т. е. помещено в специальную «очередь неназначенных заданий» (обратите внимание, что значение параметра конфигурации shardman.silk_unassigned_job_queue_size равно 1024). Если в очереди нет свободного места, будет сгенерировано сообщение об ошибке, которое будет отправлено обратно исходному обслуживающему процессу. Задание из этой очереди будет позже назначено первому доступному рабочему процессу, когда он освободится от предыдущего задания.

Когда рабочий процесс получает новое «задание», он выполняет его через подсистему SPI, организует получившиеся кортежи в пакеты и отправляет их обратно через общую память локальному процессу silkroad. Остальное тривиально, потому что весь маршрут известен. Последний получившийся пакет с кортежами в потоке помечается как «закрывающий». Это приказ процессам silkroads стереть этот маршрут из своих таблиц.

Обратите внимание, что серверные и удалённые рабочие процессы остаются «подписанными» на свои потоки, пока они не будут явно закрыты. Таким образом, сервер имеет возможность отправить сообщение о прерывании или уведомить удалённый рабочий процесс о преждевременном закрытии транзакции. Это позволяет отбрасывать устаревшие пакеты данных, возможно оставшиеся от предыдущих прерванных транзакций.

Для просмотра текущего состояния процесса silkroad мультиплексора используются диагностические представления Silk, описанные в Раздел 6.4.8.

Кроме того, таблица маршрутизации silkroad отслеживает конечные точки (серверы и сетевые подключения), которые были задействованы в том или ином потоке. Таким образом, когда какое-либо подключение закрывается, все задействованные серверы (и/или рабочие процессы) будут уведомлены об этом событии специальным сообщением об ошибке, а все маршруты/потоки, связанные с этим подключением, будут отклонены. Точно так же, если происходит сбой сервера, его очередь общей памяти отсоединяется, и silkroad реагирует, отправляя сообщения об ошибках удалённым участникам каждого потока, связанного со сбоем сервера. Таким образом, удалённые рабочие процессы не выполняют бесполезную работу, когда инициатор запроса уже неактивен.

Получившиеся кортежи передаются silkworm в собственном двоичном режиме. Кортежи с атрибутом хранения external будут удалены из TOAST, но те, которые были сжаты, останутся сжатыми.

Небольшие кортежи будут организованы в порции (примерно по 256 тыс.). Большие кортежи будут разделены на части отправителем и собраны в единое целое принимающим сервером.

Может возникнуть ситуация, при которой очередное сообщение, полученное от сервера, не помещается в целевом сетевом буфере. Или сообщение, полученное из сети, не помещается в целевую очередь общей памяти. В таком случае поток, вызвавший эту ситуацию, будет «приостановлен». Это означает, что silkroad приостанавливает реакцию на события исходной конечной точки (подключение или сервер) до тех пор, пока целевая конечная точка не обработает все свои сообщения. Остальные серверы и подключения, не затронутые этим маршрутом, продолжают работать. Принимающие модули серверов призваны свести к минимуму подобные ситуации. Сервер периодически проверяет и очищает входящую очередь, даже когда исполнитель плана занят обработкой других узлов плана. Полученные кортежи хранятся в хранилищах кортежей сервера в соответствии с узлами плана до тех пор, пока исполнитель не запросит следующий кортеж для выполнения конкретного узла плана.

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