Set the pg-setting inferring from the PageserverConnectionInfo

This adds a new request type between backend and communicator, to make
a getpage request at a given LSN, bypassing the LFC. Only used by the
get_raw_page_at_lsn() debugging/testing function.

.gitignore

@@ -15,6 +15,7 @@ neon.iml
 /.neon
 /integration_tests/.neon
 compaction-suite-results.*
+pgxn/neon/communicator/communicator_bindings.h
 docker-compose/docker-compose-parallel.yml
 
 # Coverage

Cargo.lock

@@ -259,6 +259,17 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8ab6b55fe97976e46f91ddbed8d147d966475dc29b2032757ba47e02376fbc3"
 
+[[package]]
+name = "atomic_enum"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "99e1aca718ea7b89985790c94aad72d77533063fe00bc497bb79a7c2dae6a661"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.100",
+]
+
 [[package]]
 name = "autocfg"
 version = "1.1.0"
@@ -1296,13 +1307,30 @@ dependencies = [
 name = "communicator"
 version = "0.1.0"
 dependencies = [
+ "atomic_enum",
  "axum",
+ "bytes",
  "cbindgen",
+ "clashmap",
  "http 1.3.1",
+ "libc",
  "measured",
+ "metrics",
+ "neon-shmem",
+ "nix 0.30.1",
+ "pageserver_api",
+ "pageserver_client_grpc",
+ "pageserver_page_api",
+ "prometheus",
+ "prost 0.13.5",
+ "strum_macros",
+ "thiserror 1.0.69",
  "tokio",
+ "tokio-pipe",
+ "tonic",
  "tracing",
  "tracing-subscriber",
+ "uring-common",
  "utils",
  "workspace_hack",
 ]
@@ -1643,9 +1671,9 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-utils"
-version = "0.8.19"
+version = "0.8.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "248e3bacc7dc6baa3b21e405ee045c3047101a49145e7e9eca583ab4c2ca5345"
+checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28"
 
 [[package]]
 name = "crossterm"
@@ -2361,6 +2389,12 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1"
 
+[[package]]
+name = "foldhash"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
+
 [[package]]
 name = "form_urlencoded"
 version = "1.2.1"
@@ -2742,6 +2776,16 @@ version = "0.15.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
 
+[[package]]
+name = "hashbrown"
+version = "0.15.4"
+source = "git+https://github.com/quantumish/hashbrown.git?rev=6610e6d#6610e6d2b1f288ef7b0709a3efefbc846395dc5e"
+dependencies = [
+ "allocator-api2",
+ "equivalent",
+ "foldhash",
+]
+
 [[package]]
 name = "hashlink"
 version = "0.9.1"
@@ -3822,7 +3866,7 @@ dependencies = [
  "prometheus",
  "rand 0.9.1",
  "rand_distr",
- "twox-hash",
+ "twox-hash 1.6.3",
 ]
 
 [[package]]
@@ -3928,15 +3972,21 @@ checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a"
 name = "neon-shmem"
 version = "0.1.0"
 dependencies = [
+ "ahash",
+ "criterion",
+ "hashbrown 0.15.4",
  "libc",
  "lock_api",
  "nix 0.30.1",
  "rand 0.9.1",
  "rand_distr",
  "rustc-hash 2.1.1",
+ "seahash",
  "tempfile",
  "thiserror 1.0.69",
+ "twox-hash 2.1.1",
  "workspace_hack",
+ "xxhash-rust",
 ]
 
 [[package]]
@@ -4391,13 +4441,16 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "async-trait",
+ "axum",
  "bytes",
  "camino",
  "clap",
  "futures",
  "hdrhistogram",
+ "http 1.3.1",
  "humantime",
  "humantime-serde",
+ "metrics",
  "pageserver_api",
  "pageserver_client",
  "pageserver_client_grpc",
@@ -4487,6 +4540,7 @@ dependencies = [
  "pageserver_client",
  "pageserver_compaction",
  "pageserver_page_api",
+ "peekable",
  "pem",
  "pin-project-lite",
  "postgres-protocol",
@@ -4500,6 +4554,7 @@ dependencies = [
  "pprof",
  "pq_proto",
  "procfs",
+ "prost 0.13.5",
  "rand 0.9.1",
  "range-set-blaze",
  "regex",
@@ -4536,7 +4591,7 @@ dependencies = [
  "tower 0.5.2",
  "tracing",
  "tracing-utils",
- "twox-hash",
+ "twox-hash 1.6.3",
  "url",
  "utils",
  "uuid",
@@ -4748,7 +4803,7 @@ dependencies = [
  "paste",
  "seq-macro",
  "thrift",
- "twox-hash",
+ "twox-hash 1.6.3",
  "zstd",
  "zstd-sys",
 ]
@@ -4794,6 +4849,15 @@ dependencies = [
  "sha2",
 ]
 
+[[package]]
+name = "peekable"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "225f9651e475709164f871dc2f5724956be59cb9edb055372ffeeab01ec2d20b"
+dependencies = [
+ "smallvec",
+]
+
 [[package]]
 name = "pem"
 version = "3.0.3"
@@ -6493,6 +6557,12 @@ version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "621e3680f3e07db4c9c2c3fb07c6223ab2fab2e54bd3c04c3ae037990f428c32"
 
+[[package]]
+name = "seahash"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b"
+
 [[package]]
 name = "sec1"
 version = "0.3.0"
@@ -7646,6 +7716,16 @@ dependencies = [
  "syn 2.0.100",
 ]
 
+[[package]]
+name = "tokio-pipe"
+version = "0.2.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f213a84bffbd61b8fa0ba8a044b4bbe35d471d0b518867181e82bd5c15542784"
+dependencies = [
+ "libc",
+ "tokio",
+]
+
 [[package]]
 name = "tokio-postgres"
 version = "0.7.10"
@@ -8183,6 +8263,15 @@ dependencies = [
  "static_assertions",
 ]
 
+[[package]]
+name = "twox-hash"
+version = "2.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b907da542cbced5261bd3256de1b3a1bf340a3d37f93425a07362a1d687de56"
+dependencies = [
+ "rand 0.9.1",
+]
+
 [[package]]
 name = "typed-json"
 version = "0.1.1"
@@ -9013,8 +9102,8 @@ dependencies = [
  "clap",
  "clap_builder",
  "const-oid",
+ "criterion",
  "crossbeam-epoch",
- "crossbeam-utils",
  "crypto-bigint 0.5.5",
  "der 0.7.8",
  "deranged",
@@ -9057,7 +9146,6 @@ dependencies = [
  "num-iter",
  "num-rational",
  "num-traits",
- "once_cell",
  "p256 0.13.2",
  "parquet",
  "portable-atomic",
@@ -9166,6 +9254,12 @@ version = "0.13.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4d25c75bf9ea12c4040a97f829154768bbbce366287e2dc044af160cd79a13fd"
 
+[[package]]
+name = "xxhash-rust"
+version = "0.8.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fdd20c5420375476fbd4394763288da7eb0cc0b8c11deed431a91562af7335d3"
+
 [[package]]
 name = "yansi"
 version = "1.0.1"

Cargo.toml

@@ -93,6 +93,7 @@ clap = { version = "4.0", features = ["derive", "env"] }
 clashmap = { version = "1.0", features = ["raw-api"] }
 comfy-table = "7.1"
 const_format = "0.2"
+crossbeam-utils = "0.8.21"
 crc32c = "0.6"
 diatomic-waker = { version = "0.2.3" }
 either = "1.8"
@@ -152,6 +153,7 @@ parquet = { version = "53", default-features = false, features = ["zstd"] }
 parquet_derive = "53"
 pbkdf2 = { version = "0.12.1", features = ["simple", "std"] }
 pem = "3.0.3"
+peekable = "0.3.0"
 pin-project-lite = "0.2"
 pprof = { version = "0.14", features = ["criterion", "flamegraph", "frame-pointer", "prost-codec"] }
 procfs = "0.16"
@@ -190,6 +192,7 @@ smallvec = "1.11"
 smol_str = { version = "0.2.0", features = ["serde"] }
 socket2 = "0.5"
 spki = "0.7.3"
+spin = "0.9.8"
 strum = "0.26"
 strum_macros = "0.26"
 "subtle"  = "2.5.0"
@@ -201,7 +204,6 @@ thiserror = "1.0"
 tikv-jemallocator = { version = "0.6", features = ["profiling", "stats", "unprefixed_malloc_on_supported_platforms"] }
 tikv-jemalloc-ctl = { version = "0.6", features = ["stats"] }
 tokio = { version = "1.43.1", features = ["macros"] }
-tokio-epoll-uring = { git = "https://github.com/neondatabase/tokio-epoll-uring.git" , branch = "main" }
 tokio-io-timeout = "1.2.0"
 tokio-postgres-rustls = "0.12.0"
 tokio-rustls = { version = "0.26.0", default-features = false, features = ["tls12", "ring"]}
@@ -242,6 +244,9 @@ zeroize = "1.8"
 env_logger = "0.11"
 log = "0.4"
 
+tokio-epoll-uring = { git = "https://github.com/neondatabase/tokio-epoll-uring.git" , branch = "main" }
+uring-common = { git = "https://github.com/neondatabase/tokio-epoll-uring.git" , branch = "main" }
+
 ## Libraries from neondatabase/ git forks, ideally with changes to be upstreamed
 postgres = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "neon" }
 postgres-protocol = { git = "https://github.com/neondatabase/rust-postgres.git", branch = "neon" }

compute/etc/sql_exporter/checkpoints_req.17.sql

@@ -1 +1 @@
-SELECT num_requested AS checkpoints_req FROM pg_stat_checkpointer;
+SELECT num_requested AS checkpoints_req FROM pg_catalog.pg_stat_checkpointer;

compute/etc/sql_exporter/checkpoints_req.sql

@@ -1 +1 @@
-SELECT checkpoints_req FROM pg_stat_bgwriter;
+SELECT checkpoints_req FROM pg_catalog.pg_stat_bgwriter;

compute/etc/sql_exporter/checkpoints_timed.sql

@@ -1 +1 @@
-SELECT checkpoints_timed FROM pg_stat_bgwriter;
+SELECT checkpoints_timed FROM pg_catalog.pg_stat_bgwriter;

compute/etc/sql_exporter/compute_backpressure_throttling_seconds_total.sql

@@ -1 +1 @@
-SELECT (neon.backpressure_throttling_time()::float8 / 1000000) AS throttled;
+SELECT (neon.backpressure_throttling_time()::pg_catalog.float8 / 1000000) AS throttled;

compute/etc/sql_exporter/compute_current_lsn.sql

@@ -1,4 +1,4 @@
 SELECT CASE
-  WHEN pg_catalog.pg_is_in_recovery() THEN (pg_last_wal_replay_lsn() - '0/0')::FLOAT8
-  ELSE (pg_current_wal_lsn() - '0/0')::FLOAT8
+  WHEN pg_catalog.pg_is_in_recovery() THEN (pg_catalog.pg_last_wal_replay_lsn() - '0/0')::pg_catalog.FLOAT8
+  ELSE (pg_catalog.pg_current_wal_lsn() - '0/0')::pg_catalog.FLOAT8
 END AS lsn;

compute/etc/sql_exporter/compute_logical_snapshot_files.sql

@@ -1,7 +1,7 @@
 SELECT
-  (SELECT setting FROM pg_settings WHERE name = 'neon.timeline_id') AS timeline_id,
+  (SELECT setting FROM pg_catalog.pg_settings WHERE name = 'neon.timeline_id') AS timeline_id,
   -- Postgres creates temporary snapshot files of the form %X-%X.snap.%d.tmp.
   -- These temporary snapshot files are renamed to the actual snapshot files
   -- after they are completely built. We only WAL-log the completely built
   -- snapshot files
-  (SELECT COUNT(*) FROM pg_ls_dir('pg_logical/snapshots') AS name WHERE name LIKE '%.snap') AS num_logical_snapshot_files;
+  (SELECT COUNT(*) FROM pg_catalog.pg_ls_dir('pg_logical/snapshots') AS name WHERE name LIKE '%.snap') AS num_logical_snapshot_files;

compute/etc/sql_exporter/compute_logical_snapshots_bytes.15.sql

@@ -1,7 +1,7 @@
 SELECT
-  (SELECT current_setting('neon.timeline_id')) AS timeline_id,
+  (SELECT pg_catalog.current_setting('neon.timeline_id')) AS timeline_id,
   -- Postgres creates temporary snapshot files of the form %X-%X.snap.%d.tmp.
   -- These temporary snapshot files are renamed to the actual snapshot files
   -- after they are completely built. We only WAL-log the completely built
   -- snapshot files
-  (SELECT COALESCE(sum(size), 0) FROM pg_ls_logicalsnapdir() WHERE name LIKE '%.snap') AS logical_snapshots_bytes;
+  (SELECT COALESCE(pg_catalog.sum(size), 0) FROM pg_catalog.pg_ls_logicalsnapdir() WHERE name LIKE '%.snap') AS logical_snapshots_bytes;

compute/etc/sql_exporter/compute_logical_snapshots_bytes.sql

@@ -1,9 +1,9 @@
 SELECT
-  (SELECT setting FROM pg_settings WHERE name = 'neon.timeline_id') AS timeline_id,
+  (SELECT setting FROM pg_catalog.pg_settings WHERE name = 'neon.timeline_id') AS timeline_id,
   -- Postgres creates temporary snapshot files of the form %X-%X.snap.%d.tmp.
   -- These temporary snapshot files are renamed to the actual snapshot files
   -- after they are completely built. We only WAL-log the completely built
   -- snapshot files
-  (SELECT COALESCE(sum((pg_stat_file('pg_logical/snapshots/' || name, missing_ok => true)).size), 0)
-    FROM (SELECT * FROM pg_ls_dir('pg_logical/snapshots') WHERE pg_ls_dir LIKE '%.snap') AS name
+  (SELECT COALESCE(pg_catalog.sum((pg_catalog.pg_stat_file('pg_logical/snapshots/' || name, missing_ok => true)).size), 0)
+   FROM (SELECT * FROM pg_catalog.pg_ls_dir('pg_logical/snapshots') WHERE pg_ls_dir LIKE '%.snap') AS name
   ) AS logical_snapshots_bytes;

compute/etc/sql_exporter/compute_max_connections.sql

@@ -1 +1 @@
-SELECT current_setting('max_connections') as max_connections;
+SELECT pg_catalog.current_setting('max_connections') AS max_connections;

compute/etc/sql_exporter/compute_pg_oldest_frozen_xid_age.sql

@@ -1,4 +1,4 @@
 SELECT datname database_name,
-  age(datfrozenxid) frozen_xid_age
-FROM pg_database
+   pg_catalog.age(datfrozenxid) frozen_xid_age
+FROM pg_catalog.pg_database
 ORDER BY frozen_xid_age DESC LIMIT 10;

compute/etc/sql_exporter/compute_pg_oldest_mxid_age.sql

@@ -1,4 +1,4 @@
 SELECT datname database_name,
-  mxid_age(datminmxid) min_mxid_age
-FROM pg_database
+  pg_catalog.mxid_age(datminmxid) min_mxid_age
+FROM pg_catalog.pg_database
 ORDER BY min_mxid_age DESC LIMIT 10;

compute/etc/sql_exporter/compute_receive_lsn.sql

@@ -1,4 +1,4 @@
 SELECT CASE
-  WHEN pg_catalog.pg_is_in_recovery() THEN (pg_last_wal_receive_lsn() - '0/0')::FLOAT8
+  WHEN pg_catalog.pg_is_in_recovery() THEN (pg_catalog.pg_last_wal_receive_lsn() - '0/0')::pg_catalog.FLOAT8
   ELSE 0
 END AS lsn;

compute/etc/sql_exporter/compute_subscriptions_count.sql

@@ -1 +1 @@
-SELECT subenabled::text AS enabled, count(*) AS subscriptions_count FROM pg_subscription GROUP BY subenabled;
+SELECT subenabled::pg_catalog.text AS enabled, pg_catalog.count(*) AS subscriptions_count FROM pg_catalog.pg_subscription GROUP BY subenabled;

compute/etc/sql_exporter/connection_counts.sql

@@ -1 +1 @@
-SELECT datname, state, count(*) AS count FROM pg_stat_activity WHERE state <> '' GROUP BY datname, state;
+SELECT datname, state, pg_catalog.count(*) AS count FROM pg_catalog.pg_stat_activity WHERE state <> '' GROUP BY datname, state;

compute/etc/sql_exporter/db_total_size.sql

@@ -1,5 +1,5 @@
-SELECT sum(pg_database_size(datname)) AS total
-FROM pg_database
+SELECT pg_catalog.sum(pg_catalog.pg_database_size(datname)) AS total
+FROM pg_catalog.pg_database
 -- Ignore invalid databases, as we will likely have problems with
 -- getting their size from the Pageserver.
 WHERE datconnlimit != -2;

compute/etc/sql_exporter/lfc_approximate_working_set_size_windows.autoscaling.sql

@@ -3,6 +3,6 @@
 -- minutes.
 
 SELECT
-  x::text as duration_seconds,
+  x::pg_catalog.text AS duration_seconds,
   neon.approximate_working_set_size_seconds(x) AS size
 FROM (SELECT generate_series * 60 AS x FROM generate_series(1, 60)) AS t (x);

compute/etc/sql_exporter/lfc_approximate_working_set_size_windows.sql

@@ -3,6 +3,6 @@
 
 SELECT
   x AS duration,
-  neon.approximate_working_set_size_seconds(extract('epoch' FROM x::interval)::int) AS size FROM (
+  neon.approximate_working_set_size_seconds(extract('epoch' FROM x::pg_catalog.interval)::pg_catalog.int4) AS size FROM (
     VALUES ('5m'), ('15m'), ('1h')
   ) AS t (x);

compute/etc/sql_exporter/lfc_cache_size_limit.sql

@@ -1 +1 @@
-SELECT pg_size_bytes(current_setting('neon.file_cache_size_limit')) AS lfc_cache_size_limit;
+SELECT pg_catalog.pg_size_bytes(pg_catalog.current_setting('neon.file_cache_size_limit')) AS lfc_cache_size_limit;

compute/etc/sql_exporter/logical_slot_restart_lsn.sql

@@ -1,3 +1,3 @@
-SELECT slot_name, (restart_lsn - '0/0')::FLOAT8 as restart_lsn
-FROM pg_replication_slots
+SELECT slot_name, (restart_lsn - '0/0')::pg_catalog.FLOAT8 AS restart_lsn
+FROM pg_catalog.pg_replication_slots
 WHERE slot_type = 'logical';

compute/etc/sql_exporter/max_cluster_size.sql

@@ -1 +1 @@
-SELECT setting::int AS max_cluster_size FROM pg_settings WHERE name = 'neon.max_cluster_size';
+SELECT setting::pg_catalog.int4 AS max_cluster_size FROM pg_catalog.pg_settings WHERE name = 'neon.max_cluster_size';

compute/etc/sql_exporter/pg_stats_userdb.sql

@@ -1,13 +1,13 @@
 -- We export stats for 10 non-system databases. Without this limit it is too
 -- easy to abuse the system by creating lots of databases.
 
-SELECT pg_database_size(datname) AS db_size,
+SELECT pg_catalog.pg_database_size(datname) AS db_size,
   deadlocks,
   tup_inserted AS inserted,
   tup_updated AS updated,
   tup_deleted AS deleted,
   datname
-FROM pg_stat_database
+FROM pg_catalog.pg_stat_database
 WHERE datname IN (
   SELECT datname FROM pg_database
   -- Ignore invalid databases, as we will likely have problems with

compute/etc/sql_exporter/replication_delay_bytes.sql

@@ -3,4 +3,4 @@
 -- replay LSN may have advanced past the receive LSN we are using for the
 -- calculation.
 
-SELECT GREATEST(0, pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn())) AS replication_delay_bytes;
+SELECT GREATEST(0, pg_catalog.pg_wal_lsn_diff(pg_catalog.pg_last_wal_receive_lsn(), pg_catalog.pg_last_wal_replay_lsn())) AS replication_delay_bytes;

compute/etc/sql_exporter/replication_delay_seconds.sql

@@ -1,5 +1,5 @@
 SELECT
   CASE
-    WHEN pg_last_wal_receive_lsn() = pg_last_wal_replay_lsn() THEN 0
-    ELSE GREATEST(0, EXTRACT (EPOCH FROM now() - pg_last_xact_replay_timestamp()))
+    WHEN pg_catalog.pg_last_wal_receive_lsn() = pg_catalog.pg_last_wal_replay_lsn() THEN 0
+    ELSE GREATEST(0, EXTRACT (EPOCH FROM pg_catalog.now() - pg_catalog.pg_last_xact_replay_timestamp()))
   END AS replication_delay_seconds;

compute/etc/sql_exporter/retained_wal.sql

@@ -1,10 +1,10 @@
 SELECT
   slot_name,
-  pg_wal_lsn_diff(
+  pg_catalog.pg_wal_lsn_diff(
     CASE
-      WHEN pg_is_in_recovery() THEN pg_last_wal_replay_lsn()
-      ELSE pg_current_wal_lsn()
+      WHEN pg_catalog.pg_is_in_recovery() THEN pg_catalog.pg_last_wal_replay_lsn()
+      ELSE pg_catalog.pg_current_wal_lsn()
     END,
-    restart_lsn)::FLOAT8 AS retained_wal
-FROM pg_replication_slots
+    restart_lsn)::pg_catalog.FLOAT8 AS retained_wal
+FROM pg_catalog.pg_replication_slots
 WHERE active = false;

compute/etc/sql_exporter/wal_is_lost.sql

@@ -4,4 +4,4 @@ SELECT
     WHEN wal_status = 'lost' THEN 1
     ELSE 0
   END AS wal_is_lost
-FROM pg_replication_slots;
+FROM pg_catalog.pg_replication_slots;

compute_tools/src/bin/compute_ctl.rs

@@ -279,7 +279,7 @@ fn main() -> Result<()> {
         config,
     )?;
 
-    let exit_code = compute_node.run()?;
+    let exit_code = compute_node.run().context("running compute node")?;
 
     scenario.teardown();
 

compute_tools/src/checker.rs

@@ -24,9 +24,9 @@ pub async fn check_writability(compute: &ComputeNode) -> Result<()> {
     });
 
     let query = "
-    INSERT INTO health_check VALUES (1, now())
+    INSERT INTO public.health_check VALUES (1, pg_catalog.now())
         ON CONFLICT (id) DO UPDATE
-         SET updated_at = now();";
+         SET updated_at = pg_catalog.now();";
 
     match client.simple_query(query).await {
         Result::Ok(result) => {

compute_tools/src/compute.rs

@@ -583,7 +583,7 @@ impl ComputeNode {
         // that can affect `compute_ctl` and prevent it from properly configuring the database schema.
         // Unset them via connection string options before connecting to the database.
         // N.B. keep it in sync with `ZENITH_OPTIONS` in `get_maintenance_client()`.
-        const EXTRA_OPTIONS: &str = "-c role=cloud_admin -c default_transaction_read_only=off -c search_path=public -c statement_timeout=0 -c pgaudit.log=none";
+        const EXTRA_OPTIONS: &str = "-c role=cloud_admin -c default_transaction_read_only=off -c search_path='' -c statement_timeout=0 -c pgaudit.log=none";
         let options = match conn_conf.get_options() {
             // Allow the control plane to override any options set by the
             // compute
@@ -1884,7 +1884,7 @@ impl ComputeNode {
 
                     // It doesn't matter what were the options before, here we just want
                     // to connect and create a new superuser role.
-                    const ZENITH_OPTIONS: &str = "-c role=zenith_admin -c default_transaction_read_only=off -c search_path=public -c statement_timeout=0";
+                    const ZENITH_OPTIONS: &str = "-c role=zenith_admin -c default_transaction_read_only=off -c search_path='' -c statement_timeout=0";
                     zenith_admin_conf.options(ZENITH_OPTIONS);
 
                     let mut client =
@@ -2339,13 +2339,13 @@ impl ComputeNode {
         let result = client
             .simple_query(
                 "SELECT
-    row_to_json(pg_stat_statements)
+    pg_catalog.row_to_json(pss)
 FROM
-    pg_stat_statements
+    public.pg_stat_statements pss
 WHERE
-    userid != 'cloud_admin'::regrole::oid
+    pss.userid != 'cloud_admin'::pg_catalog.regrole::pg_catalog.oid
 ORDER BY
-    (mean_exec_time + mean_plan_time) DESC
+    (pss.mean_exec_time + pss.mean_plan_time) DESC
 LIMIT 100",
             )
             .await;
@@ -2473,11 +2473,11 @@ LIMIT 100",
 
         // check the role grants first - to gracefully handle read-replicas.
         let select = "SELECT privilege_type
-            FROM pg_namespace
-                JOIN LATERAL (SELECT * FROM aclexplode(nspacl) AS x) acl ON true
-                JOIN pg_user users ON acl.grantee = users.usesysid
-            WHERE users.usename = $1
-                AND nspname = $2";
+            FROM pg_catalog.pg_namespace
+                JOIN LATERAL (SELECT * FROM aclexplode(nspacl) AS x) AS acl ON true
+                JOIN pg_catalog.pg_user users ON acl.grantee = users.usesysid
+            WHERE users.usename OPERATOR(pg_catalog.=) $1::pg_catalog.name
+                AND nspname OPERATOR(pg_catalog.=) $2::pg_catalog.name";
         let rows = db_client
             .query(select, &[role_name, schema_name])
             .await
@@ -2546,8 +2546,9 @@ LIMIT 100",
                 .await
                 .with_context(|| format!("Failed to execute query: {query}"))?;
         } else {
-            let query =
-                format!("CREATE EXTENSION IF NOT EXISTS {ext_name} WITH VERSION {quoted_version}");
+            let query = format!(
+                "CREATE EXTENSION IF NOT EXISTS {ext_name} WITH SCHEMA public VERSION {quoted_version}"
+            );
             db_client
                 .simple_query(&query)
                 .await

compute_tools/src/compute_promote.rs

@@ -78,7 +78,7 @@ impl ComputeNode {
         const RETRIES: i32 = 20;
         for i in 0..=RETRIES {
             let row = client
-                .query_one("SELECT pg_last_wal_replay_lsn()", &[])
+                .query_one("SELECT pg_catalog.pg_last_wal_replay_lsn()", &[])
                 .await
                 .context("getting last replay lsn")?;
             let lsn: u64 = row.get::<usize, postgres_types::PgLsn>(0).into();
@@ -103,7 +103,7 @@ impl ComputeNode {
             .await
             .context("setting safekeepers")?;
         client
-            .query("SELECT pg_reload_conf()", &[])
+            .query("SELECT pg_catalog.pg_reload_conf()", &[])
             .await
             .context("reloading postgres config")?;
 
@@ -113,7 +113,7 @@ impl ComputeNode {
         });
 
         let row = client
-            .query_one("SELECT * FROM pg_promote()", &[])
+            .query_one("SELECT * FROM pg_catalog.pg_promote()", &[])
             .await
             .context("pg_promote")?;
         if !row.get::<usize, bool>(0) {

compute_tools/src/config.rs

@@ -65,15 +65,21 @@ pub fn write_postgres_conf(
         writeln!(file, "{conf}")?;
     }
 
-    // Stripe size GUC should be defined prior to connection string
-    if let Some(stripe_size) = spec.shard_stripe_size {
-        writeln!(file, "neon.stripe_size={stripe_size}")?;
-    }
     // Add options for connecting to storage
     writeln!(file, "# Neon storage settings")?;
     writeln!(file)?;
     if let Some(conninfo) = &spec.pageserver_connection_info {
+        // Stripe size GUC should be defined prior to connection string
+        if let Some(stripe_size) = conninfo.stripe_size {
+            writeln!(
+                file,
+                "# from compute spec's pageserver_conninfo.stripe_size field"
+            )?;
+            writeln!(file, "neon.stripe_size={stripe_size}")?;
+        }
+
         let mut libpq_urls: Option<Vec<String>> = Some(Vec::new());
+        let mut grpc_urls: Option<Vec<String>> = Some(Vec::new());
         let num_shards = if conninfo.shard_count.0 == 0 {
             1 // unsharded, treat it as a single shard
         } else {
@@ -106,6 +112,14 @@ pub fn write_postgres_conf(
             } else {
                 libpq_urls = None
             }
+            // Similarly for gRPC URLs
+            if let Some(url) = &first_pageserver.grpc_url {
+                if let Some(ref mut urls) = grpc_urls {
+                    urls.push(url.clone());
+                }
+            } else {
+                grpc_urls = None
+            }
         }
         if let Some(libpq_urls) = libpq_urls {
             writeln!(
@@ -117,27 +131,34 @@ pub fn write_postgres_conf(
                 "neon.pageserver_connstring={}",
                 escape_conf_value(&libpq_urls.join(","))
             )?;
+            writeln!(file, "neon.use_communicator_worker=false")?;
         } else {
             writeln!(file, "# no neon.pageserver_connstring")?;
         }
-
-        if let Some(stripe_size) = conninfo.stripe_size {
+        if let Some(grpc_urls) = grpc_urls {
             writeln!(
                 file,
-                "# from compute spec's pageserver_conninfo.stripe_size field"
+                "# derived from compute spec's pageserver_conninfo field"
             )?;
-            writeln!(file, "neon.stripe_size={stripe_size}")?;
+            writeln!(
+                file,
+                "neon.pageserver_grpc_urls={}",
+                escape_conf_value(&grpc_urls.join(","))
+            )?;
+            writeln!(file, "neon.use_communicator_worker=true")?;
+        } else {
+            writeln!(file, "# no neon.pageserver_grpc_urls")?;
         }
     } else {
-        if let Some(s) = &spec.pageserver_connstring {
-            writeln!(file, "# from compute spec's pageserver_connstring field")?;
-            writeln!(file, "neon.pageserver_connstring={}", escape_conf_value(s))?;
-        }
-
+        // Stripe size GUC should be defined prior to connection string
         if let Some(stripe_size) = spec.shard_stripe_size {
             writeln!(file, "# from compute spec's shard_stripe_size field")?;
             writeln!(file, "neon.stripe_size={stripe_size}")?;
         }
+        if let Some(s) = &spec.pageserver_connstring {
+            writeln!(file, "# from compute spec's pageserver_connstring field")?;
+            writeln!(file, "neon.pageserver_connstring={}", escape_conf_value(s))?;
+        }
     }
 
     if !spec.safekeeper_connstrings.is_empty() {

compute_tools/src/installed_extensions.rs

@@ -19,7 +19,7 @@ async fn list_dbs(client: &mut Client) -> Result<Vec<String>, PostgresError> {
         .query(
             "SELECT datname FROM pg_catalog.pg_database
                 WHERE datallowconn
-                AND datconnlimit <> - 2
+                AND datconnlimit OPERATOR(pg_catalog.<>) (OPERATOR(pg_catalog.-) 2::pg_catalog.int4)
                 LIMIT 500",
             &[],
         )
@@ -67,7 +67,7 @@ pub async fn get_installed_extensions(
 
         let extensions: Vec<(String, String, i32)> = client
             .query(
-                "SELECT extname, extversion, extowner::integer FROM pg_catalog.pg_extension",
+                "SELECT extname, extversion, extowner::pg_catalog.int4 FROM pg_catalog.pg_extension",
                 &[],
             )
             .await?

compute_tools/src/migration.rs

@@ -76,7 +76,7 @@ impl<'m> MigrationRunner<'m> {
         self.client
             .simple_query("CREATE SCHEMA IF NOT EXISTS neon_migration")
             .await?;
-        self.client.simple_query("CREATE TABLE IF NOT EXISTS neon_migration.migration_id (key INT NOT NULL PRIMARY KEY, id bigint NOT NULL DEFAULT 0)").await?;
+        self.client.simple_query("CREATE TABLE IF NOT EXISTS neon_migration.migration_id (key pg_catalog.int4 NOT NULL PRIMARY KEY, id pg_catalog.int8 NOT NULL DEFAULT 0)").await?;
         self.client
             .simple_query(
                 "INSERT INTO neon_migration.migration_id VALUES (0, 0) ON CONFLICT DO NOTHING",

compute_tools/src/migrations/0002-alter_roles.sql

@@ -15,17 +15,17 @@ DO $$
 DECLARE
     role_name text;
 BEGIN
-    FOR role_name IN SELECT rolname FROM pg_roles WHERE pg_has_role(rolname, '{privileged_role_name}', 'member')
+    FOR role_name IN SELECT rolname FROM pg_catalog.pg_roles WHERE pg_catalog.pg_has_role(rolname, '{privileged_role_name}', 'member')
     LOOP
-        RAISE NOTICE 'EXECUTING ALTER ROLE % INHERIT', quote_ident(role_name);
-        EXECUTE 'ALTER ROLE ' || quote_ident(role_name) || ' INHERIT';
+        RAISE NOTICE 'EXECUTING ALTER ROLE % INHERIT', pg_catalog.quote_ident(role_name);
+        EXECUTE pg_catalog.format('ALTER ROLE %I INHERIT;', role_name);
     END LOOP;
 
-    FOR role_name IN SELECT rolname FROM pg_roles
+    FOR role_name IN SELECT rolname FROM pg_catalog.pg_roles
         WHERE
-            NOT pg_has_role(rolname, '{privileged_role_name}', 'member') AND NOT starts_with(rolname, 'pg_')
+            NOT pg_catalog.pg_has_role(rolname, '{privileged_role_name}', 'member') AND NOT pg_catalog.starts_with(rolname, 'pg_')
     LOOP
-        RAISE NOTICE 'EXECUTING ALTER ROLE % NOBYPASSRLS', quote_ident(role_name);
-        EXECUTE 'ALTER ROLE ' || quote_ident(role_name) || ' NOBYPASSRLS';
+        RAISE NOTICE 'EXECUTING ALTER ROLE % NOBYPASSRLS', pg_catalog.quote_ident(role_name);
+        EXECUTE pg_catalog.format('ALTER ROLE %I NOBYPASSRLS;', role_name);
     END LOOP;
 END $$;

compute_tools/src/migrations/0003-grant_pg_create_subscription_to_privileged_role.sql

@@ -1,6 +1,6 @@
 DO $$
 BEGIN
-    IF (SELECT setting::numeric >= 160000 FROM pg_settings WHERE name = 'server_version_num') THEN
+    IF (SELECT setting::pg_catalog.numeric >= 160000 FROM pg_catalog.pg_settings WHERE name = 'server_version_num') THEN
         EXECUTE 'GRANT pg_create_subscription TO {privileged_role_name}';
     END IF;
 END $$;

compute_tools/src/migrations/0009-revoke_replication_for_previously_allowed_roles.sql

@@ -5,9 +5,9 @@ DO $$
 DECLARE
     role_name TEXT;
 BEGIN
-    FOR role_name IN SELECT rolname FROM pg_roles WHERE rolreplication IS TRUE
+    FOR role_name IN SELECT rolname FROM pg_catalog.pg_roles WHERE rolreplication IS TRUE
     LOOP
-        RAISE NOTICE 'EXECUTING ALTER ROLE % NOREPLICATION', quote_ident(role_name);
-        EXECUTE 'ALTER ROLE ' || quote_ident(role_name) || ' NOREPLICATION';
+        RAISE NOTICE 'EXECUTING ALTER ROLE % NOREPLICATION', pg_catalog.quote_ident(role_name);
+        EXECUTE pg_catalog.format('ALTER ROLE %I NOREPLICATION;', role_name);
     END LOOP;
 END $$;

compute_tools/src/migrations/0010-grant_snapshot_synchronization_funcs_to_privileged_role.sql

@@ -1,6 +1,6 @@
 DO $$
 BEGIN
-    IF (SELECT setting::numeric >= 160000 FROM pg_settings WHERE name = 'server_version_num') THEN
+    IF (SELECT setting::pg_catalog.numeric >= 160000 FROM pg_catalog.pg_settings WHERE name OPERATOR(pg_catalog.=) 'server_version_num'::pg_catalog.text) THEN
        EXECUTE 'GRANT EXECUTE ON FUNCTION pg_export_snapshot TO {privileged_role_name}';
        EXECUTE 'GRANT EXECUTE ON FUNCTION pg_log_standby_snapshot TO {privileged_role_name}';
     END IF;

compute_tools/src/migrations/tests/0001-add_bypass_rls_to_privileged_role.sql

@@ -2,7 +2,7 @@ DO $$
 DECLARE
     bypassrls boolean;
 BEGIN
-    SELECT rolbypassrls INTO bypassrls FROM pg_roles WHERE rolname = 'neon_superuser';
+    SELECT rolbypassrls INTO bypassrls FROM pg_catalog.pg_roles WHERE rolname = 'neon_superuser';
     IF NOT bypassrls THEN
         RAISE EXCEPTION 'neon_superuser cannot bypass RLS';
     END IF;

compute_tools/src/migrations/tests/0002-alter_roles.sql

@@ -4,8 +4,8 @@ DECLARE
 BEGIN
     FOR role IN
         SELECT rolname AS name, rolinherit AS inherit
-        FROM pg_roles
-        WHERE pg_has_role(rolname, 'neon_superuser', 'member')
+        FROM pg_catalog.pg_roles
+        WHERE pg_catalog.pg_has_role(rolname, 'neon_superuser', 'member')
     LOOP
         IF NOT role.inherit THEN
             RAISE EXCEPTION '% cannot inherit', quote_ident(role.name);
@@ -14,12 +14,12 @@ BEGIN
 
     FOR role IN
         SELECT rolname AS name, rolbypassrls AS bypassrls
-        FROM pg_roles
-        WHERE NOT pg_has_role(rolname, 'neon_superuser', 'member')
-            AND NOT starts_with(rolname, 'pg_')
+        FROM pg_catalog.pg_roles
+        WHERE NOT pg_catalog.pg_has_role(rolname, 'neon_superuser', 'member')
+            AND NOT pg_catalog.starts_with(rolname, 'pg_')
     LOOP
         IF role.bypassrls THEN
-            RAISE EXCEPTION  '% can bypass RLS', quote_ident(role.name);
+            RAISE EXCEPTION  '% can bypass RLS', pg_catalog.quote_ident(role.name);
         END IF;
     END LOOP;
 END $$;

compute_tools/src/migrations/tests/0003-grant_pg_create_subscription_to_privileged_role.sql

@@ -1,10 +1,10 @@
 DO $$
 BEGIN
-    IF (SELECT current_setting('server_version_num')::numeric < 160000) THEN
+    IF (SELECT pg_catalog.current_setting('server_version_num')::pg_catalog.numeric < 160000) THEN
         RETURN;
     END IF;
 
-    IF NOT (SELECT pg_has_role('neon_superuser', 'pg_create_subscription', 'member')) THEN
+    IF NOT (SELECT pg_catalog.pg_has_role('neon_superuser', 'pg_create_subscription', 'member')) THEN
         RAISE EXCEPTION 'neon_superuser cannot execute pg_create_subscription';
     END IF;
 END $$;

compute_tools/src/migrations/tests/0004-grant_pg_monitor_to_privileged_role.sql

@@ -2,12 +2,12 @@ DO $$
 DECLARE
     monitor record;
 BEGIN
-    SELECT pg_has_role('neon_superuser', 'pg_monitor', 'member') AS member,
+    SELECT pg_catalog.pg_has_role('neon_superuser', 'pg_monitor', 'member') AS member,
             admin_option AS admin
         INTO monitor
-        FROM pg_auth_members
-        WHERE roleid = 'pg_monitor'::regrole
-            AND member = 'neon_superuser'::regrole;
+        FROM pg_catalog.pg_auth_members
+        WHERE roleid = 'pg_monitor'::pg_catalog.regrole
+            AND member = 'neon_superuser'::pg_catalog.regrole;
 
     IF monitor IS NULL THEN
         RAISE EXCEPTION 'no entry in pg_auth_members for neon_superuser and pg_monitor';

compute_tools/src/migrations/tests/0010-grant_snapshot_synchronization_funcs_to_privileged_role.sql

@@ -2,11 +2,11 @@ DO $$
 DECLARE
     can_execute boolean;
 BEGIN
-    SELECT bool_and(has_function_privilege('neon_superuser', oid, 'execute'))
+    SELECT pg_catalog.bool_and(pg_catalog.has_function_privilege('neon_superuser', oid, 'execute'))
        INTO can_execute
-       FROM pg_proc
+       FROM pg_catalog.pg_proc
        WHERE proname IN ('pg_export_snapshot', 'pg_log_standby_snapshot')
-           AND pronamespace = 'pg_catalog'::regnamespace;
+           AND pronamespace = 'pg_catalog'::pg_catalog.regnamespace;
     IF NOT can_execute THEN
         RAISE EXCEPTION 'neon_superuser cannot execute both pg_export_snapshot and pg_log_standby_snapshot';
     END IF;

compute_tools/src/migrations/tests/0011-grant_pg_show_replication_origin_status_to_privileged_role.sql

@@ -2,9 +2,9 @@ DO $$
 DECLARE
     can_execute boolean;
 BEGIN
-    SELECT has_function_privilege('neon_superuser', oid, 'execute')
+    SELECT pg_catalog.has_function_privilege('neon_superuser', oid, 'execute')
        INTO can_execute
-       FROM pg_proc
+       FROM pg_catalog.pg_proc
        WHERE proname = 'pg_show_replication_origin_status'
            AND pronamespace = 'pg_catalog'::regnamespace;
     IF NOT can_execute THEN

compute_tools/src/migrations/tests/0012-grant_pg_signal_backend_to_privileged_role.sql

@@ -2,10 +2,10 @@ DO $$
 DECLARE
     signal_backend record;
 BEGIN
-    SELECT pg_has_role('neon_superuser', 'pg_signal_backend', 'member') AS member,
+    SELECT pg_catalog.pg_has_role('neon_superuser', 'pg_signal_backend', 'member') AS member,
             admin_option AS admin
         INTO signal_backend
-        FROM pg_auth_members
+        FROM pg_catalog.pg_auth_members
         WHERE roleid = 'pg_signal_backend'::regrole
             AND member = 'neon_superuser'::regrole;
 

compute_tools/src/monitor.rs

@@ -407,9 +407,9 @@ fn get_database_stats(cli: &mut Client) -> anyhow::Result<(f64, i64)> {
     // like `postgres_exporter` use it to query Postgres statistics.
     // Use explicit 8 bytes type casts to match Rust types.
     let stats = cli.query_one(
-        "SELECT coalesce(sum(active_time), 0.0)::float8 AS total_active_time,
-            coalesce(sum(sessions), 0)::bigint AS total_sessions
-        FROM pg_stat_database
+        "SELECT pg_catalog.coalesce(pg_catalog.sum(active_time), 0.0)::pg_catalog.float8 AS total_active_time,
+            pg_catalog.coalesce(pg_catalog.sum(sessions), 0)::pg_catalog.bigint AS total_sessions
+        FROM pg_catalog.pg_stat_database
         WHERE datname NOT IN (
                 'postgres',
                 'template0',
@@ -445,11 +445,11 @@ fn get_backends_state_change(cli: &mut Client) -> anyhow::Result<Option<DateTime
     let mut last_active: Option<DateTime<Utc>> = None;
     // Get all running client backends except ourself, use RFC3339 DateTime format.
     let backends = cli.query(
-        "SELECT state, to_char(state_change, 'YYYY-MM-DD\"T\"HH24:MI:SS.US\"Z\"') AS state_change
+        "SELECT state, pg_catalog.to_char(state_change, 'YYYY-MM-DD\"T\"HH24:MI:SS.US\"Z\"'::pg_catalog.text) AS state_change
                 FROM pg_stat_activity
-                    WHERE backend_type = 'client backend'
-                    AND pid != pg_backend_pid()
-                    AND usename != 'cloud_admin';", // XXX: find a better way to filter other monitors?
+                    WHERE backend_type OPERATOR(pg_catalog.=) 'client backend'::pg_catalog.text
+                    AND pid OPERATOR(pg_catalog.!=) pg_catalog.pg_backend_pid()
+                    AND usename OPERATOR(pg_catalog.!=) 'cloud_admin'::pg_catalog.name;", // XXX: find a better way to filter other monitors?
         &[],
     );
 

compute_tools/src/pg_helpers.rs

@@ -299,9 +299,9 @@ pub async fn get_existing_dbs_async(
         .query_raw::<str, &String, &[String; 0]>(
             "SELECT
                 datname AS name,
-                (SELECT rolname FROM pg_roles WHERE oid = datdba) AS owner,
+                (SELECT rolname FROM pg_catalog.pg_roles WHERE oid OPERATOR(pg_catalog.=) datdba) AS owner,
                 NOT datallowconn AS restrict_conn,
-                datconnlimit = - 2 AS invalid
+                datconnlimit OPERATOR(pg_catalog.=) (OPERATOR(pg_catalog.-) 2) AS invalid
             FROM
                 pg_catalog.pg_database;",
             &[],

compute_tools/src/spec_apply.rs

@@ -82,7 +82,7 @@ impl ComputeNode {
                 info!("Checking if drop subscription operation was already performed for timeline_id: {}", timeline_id);
 
                 drop_subscriptions_done = match
-                    client.query("select 1 from neon.drop_subscriptions_done where timeline_id = $1", &[&timeline_id.to_string()]).await {
+                    client.query("select 1 from neon.drop_subscriptions_done where timeline_id OPERATOR(pg_catalog.=) $1", &[&timeline_id.to_string()]).await {
                     Ok(result) => !result.is_empty(),
                     Err(e) =>
                     {
@@ -1142,7 +1142,9 @@ async fn get_operations<'a>(
             if let Some(libs) = spec.cluster.settings.find("shared_preload_libraries") {
                 if libs.contains("pg_stat_statements") {
                     return Ok(Box::new(once(Operation {
-                        query: String::from("CREATE EXTENSION IF NOT EXISTS pg_stat_statements"),
+                        query: String::from(
+                            "CREATE EXTENSION IF NOT EXISTS pg_stat_statements WITH SCHEMA public",
+                        ),
                         comment: Some(String::from("create system extensions")),
                     })));
                 }
@@ -1150,11 +1152,13 @@ async fn get_operations<'a>(
             Ok(Box::new(empty()))
         }
         ApplySpecPhase::CreatePgauditExtension => Ok(Box::new(once(Operation {
-            query: String::from("CREATE EXTENSION IF NOT EXISTS pgaudit"),
+            query: String::from("CREATE EXTENSION IF NOT EXISTS pgaudit WITH SCHEMA public"),
             comment: Some(String::from("create pgaudit extensions")),
         }))),
         ApplySpecPhase::CreatePgauditlogtofileExtension => Ok(Box::new(once(Operation {
-            query: String::from("CREATE EXTENSION IF NOT EXISTS pgauditlogtofile"),
+            query: String::from(
+                "CREATE EXTENSION IF NOT EXISTS pgauditlogtofile WITH SCHEMA public",
+            ),
             comment: Some(String::from("create pgauditlogtofile extensions")),
         }))),
         // Disable pgaudit logging for postgres database.
@@ -1178,7 +1182,7 @@ async fn get_operations<'a>(
                 },
                 Operation {
                     query: String::from(
-                        "UPDATE pg_extension SET extrelocatable = true WHERE extname = 'neon'",
+                        "UPDATE pg_catalog.pg_extension SET extrelocatable = true WHERE extname OPERATOR(pg_catalog.=) 'neon'::pg_catalog.name AND extrelocatable OPERATOR(pg_catalog.=) false",
                     ),
                     comment: Some(String::from("compat/fix: make neon relocatable")),
                 },

compute_tools/src/sql/add_availabilitycheck_tables.sql

@@ -3,16 +3,17 @@ BEGIN
     IF NOT EXISTS(
         SELECT 1
         FROM pg_catalog.pg_tables
-        WHERE tablename = 'health_check'
+        WHERE tablename::pg_catalog.name OPERATOR(pg_catalog.=) 'health_check'::pg_catalog.name
+        AND schemaname::pg_catalog.name OPERATOR(pg_catalog.=) 'public'::pg_catalog.name
     )
     THEN
-    CREATE TABLE health_check (
-        id serial primary key,
-        updated_at timestamptz default now()
+    CREATE TABLE public.health_check (
+        id pg_catalog.int4 primary key generated by default as identity,
+        updated_at pg_catalog.timestamptz default pg_catalog.now()
     );
-    INSERT INTO health_check VALUES (1, now())
+    INSERT INTO public.health_check VALUES (1, pg_catalog.now())
         ON CONFLICT (id) DO UPDATE
-         SET updated_at = now();
+         SET updated_at = pg_catalog.now();
     END IF;
 END
 $$

compute_tools/src/sql/anon_ext_fn_reassign.sql

@@ -2,10 +2,11 @@ DO $$
 DECLARE
     query varchar;
 BEGIN
-    FOR query IN SELECT 'ALTER FUNCTION '||nsp.nspname||'.'||p.proname||'('||pg_get_function_identity_arguments(p.oid)||') OWNER TO {db_owner};'
-    FROM pg_proc p
-        JOIN pg_namespace nsp ON p.pronamespace = nsp.oid
-    WHERE nsp.nspname = 'anon' LOOP
+    FOR query IN
+    SELECT pg_catalog.format('ALTER FUNCTION %I(%s) OWNER TO {db_owner};', p.oid::regproc, pg_catalog.pg_get_function_identity_arguments(p.oid))
+    FROM pg_catalog.pg_proc p
+        WHERE p.pronamespace OPERATOR(pg_catalog.=) 'anon'::regnamespace::oid
+    LOOP
         EXECUTE query;
     END LOOP;
 END

compute_tools/src/sql/create_privileged_role.sql

@@ -1,6 +1,6 @@
 DO $$
     BEGIN
-        IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = '{privileged_role_name}')
+        IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname OPERATOR(pg_catalog.=) '{privileged_role_name}'::pg_catalog.name)
         THEN
             CREATE ROLE {privileged_role_name} {privileges} IN ROLE pg_read_all_data, pg_write_all_data;
         END IF;

compute_tools/src/sql/default_grants.sql

@@ -4,14 +4,14 @@ $$
         IF EXISTS(
             SELECT nspname
             FROM pg_catalog.pg_namespace
-            WHERE nspname = 'public'
+            WHERE nspname OPERATOR(pg_catalog.=) 'public'
         ) AND
-           current_setting('server_version_num')::int / 10000 >= 15
+           pg_catalog.current_setting('server_version_num')::int OPERATOR(pg_catalog./) 10000 OPERATOR(pg_catalog.>=) 15
         THEN
             IF EXISTS(
                 SELECT rolname
                 FROM pg_catalog.pg_roles
-                WHERE rolname = 'web_access'
+                WHERE rolname OPERATOR(pg_catalog.=) 'web_access'
             )
             THEN
                 GRANT CREATE ON SCHEMA public TO web_access;
@@ -20,7 +20,7 @@ $$
         IF EXISTS(
             SELECT nspname
             FROM pg_catalog.pg_namespace
-            WHERE nspname = 'public'
+            WHERE nspname OPERATOR(pg_catalog.=) 'public'
         )
         THEN
             ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT ALL ON TABLES TO neon_superuser WITH GRANT OPTION;

compute_tools/src/sql/drop_subscriptions.sql

@@ -2,11 +2,17 @@ DO ${outer_tag}$
 DECLARE
     subname TEXT;
 BEGIN
-    LOCK TABLE pg_subscription IN ACCESS EXCLUSIVE MODE;
-    FOR subname IN SELECT pg_subscription.subname FROM pg_subscription WHERE subdbid = (SELECT oid FROM pg_database WHERE datname = {datname_str}) LOOP
-        EXECUTE format('ALTER SUBSCRIPTION %I DISABLE;', subname);
-        EXECUTE format('ALTER SUBSCRIPTION %I SET (slot_name = NONE);', subname);
-        EXECUTE format('DROP SUBSCRIPTION %I;', subname);
+    LOCK TABLE pg_catalog.pg_subscription IN ACCESS EXCLUSIVE MODE;
+    FOR subname IN
+        SELECT pg_subscription.subname
+        FROM pg_catalog.pg_subscription
+        WHERE subdbid OPERATOR(pg_catalog.=) (
+            SELECT oid FROM pg_database WHERE datname OPERATOR(pg_catalog.=) {datname_str}::pg_catalog.name
+        )
+    LOOP
+        EXECUTE pg_catalog.format('ALTER SUBSCRIPTION %I DISABLE;', subname);
+        EXECUTE pg_catalog.format('ALTER SUBSCRIPTION %I SET (slot_name = NONE);', subname);
+        EXECUTE pg_catalog.format('DROP SUBSCRIPTION %I;', subname);
     END LOOP;
 END;
 ${outer_tag}$;

compute_tools/src/sql/finalize_drop_subscriptions.sql

@@ -3,19 +3,19 @@ BEGIN
     IF NOT EXISTS(
         SELECT 1
         FROM pg_catalog.pg_tables
-        WHERE tablename = 'drop_subscriptions_done'
-        AND schemaname = 'neon'
+        WHERE tablename OPERATOR(pg_catalog.=) 'drop_subscriptions_done'::pg_catalog.name
+        AND schemaname OPERATOR(pg_catalog.=) 'neon'::pg_catalog.name
     )
     THEN
         CREATE TABLE neon.drop_subscriptions_done
-        (id serial primary key, timeline_id text);
+        (id pg_catalog.int4 primary key generated by default as identity, timeline_id pg_catalog.text);
     END IF;
 
     -- preserve the timeline_id of the last drop_subscriptions run
     -- to ensure that the cleanup of a timeline is executed only once.
     -- use upsert to avoid the table bloat in case of cascade branching (branch of a branch)
-    INSERT INTO neon.drop_subscriptions_done VALUES (1, current_setting('neon.timeline_id'))
+    INSERT INTO neon.drop_subscriptions_done VALUES (1, pg_catalog.current_setting('neon.timeline_id'))
     ON CONFLICT (id) DO UPDATE
-    SET timeline_id = current_setting('neon.timeline_id');
+    SET timeline_id = pg_catalog.current_setting('neon.timeline_id')::pg_catalog.text;
 END
 $$

compute_tools/src/sql/pre_drop_role_revoke_privileges.sql

@@ -15,15 +15,15 @@ BEGIN
         WHERE schema_name IN ('public')
     LOOP
         FOR grantor IN EXECUTE
-            format(
-                'SELECT DISTINCT rtg.grantor FROM information_schema.role_table_grants AS rtg WHERE grantee = %s',
+            pg_catalog.format(
+                'SELECT DISTINCT rtg.grantor FROM information_schema.role_table_grants AS rtg WHERE grantee OPERATOR(pg_catalog.=) %s',
                 -- N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`
                 quote_literal({role_name})
             )
         LOOP
-            EXECUTE format('SET LOCAL ROLE %I', grantor);
+            EXECUTE pg_catalog.format('SET LOCAL ROLE %I', grantor);
 
-            revoke_query := format(
+            revoke_query := pg_catalog.format(
                 'REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA %I FROM %I GRANTED BY %I',
                 schema,
                 -- N.B. this has to be properly dollar-escaped with `pg_quote_dollar()`

compute_tools/src/sql/set_public_schema_owner.sql

@@ -5,17 +5,17 @@ DO ${outer_tag}$
         IF EXISTS(
             SELECT nspname
             FROM pg_catalog.pg_namespace
-            WHERE nspname = 'public'
+            WHERE nspname OPERATOR(pg_catalog.=) 'public'::pg_catalog.name
         )
         THEN
             SELECT nspowner::regrole::text
             FROM pg_catalog.pg_namespace
-            WHERE nspname = 'public'
+            WHERE nspname OPERATOR(pg_catalog.=) 'public'::pg_catalog.text
             INTO schema_owner;
 
-            IF schema_owner = 'cloud_admin' OR schema_owner = 'zenith_admin'
+            IF schema_owner OPERATOR(pg_catalog.=) 'cloud_admin'::pg_catalog.text OR schema_owner OPERATOR(pg_catalog.=) 'zenith_admin'::pg_catalog.text
             THEN
-                EXECUTE format('ALTER SCHEMA public OWNER TO %I', {db_owner});
+                EXECUTE pg_catalog.format('ALTER SCHEMA public OWNER TO %I', {db_owner});
             END IF;
         END IF;
     END

compute_tools/src/sql/unset_template_for_drop_dbs.sql

@@ -3,10 +3,10 @@ DO ${outer_tag}$
         IF EXISTS(
             SELECT 1
             FROM pg_catalog.pg_database
-            WHERE datname = {datname}
+            WHERE datname OPERATOR(pg_catalog.=) {datname}::pg_catalog.name
         )
         THEN
-            EXECUTE format('ALTER DATABASE %I is_template false', {datname});
+            EXECUTE pg_catalog.format('ALTER DATABASE %I is_template false', {datname});
         END IF;
     END
 ${outer_tag}$;

docker-compose/compute_wrapper/var/db/postgres/configs/config.json

@@ -120,6 +120,11 @@
                     "value": "host=pageserver port=6400",
                     "vartype": "string"
                 },
+                {
+                    "name": "neon.pageserver_grpc_urls",
+                    "value": "grpc://pageserver:6401/",
+                    "vartype": "string"
+                },
                 {
                     "name": "max_replication_write_lag",
                     "value": "500MB",

docker-compose/pageserver_config/pageserver.toml

@@ -1,6 +1,7 @@
 broker_endpoint='http://storage_broker:50051'
 pg_distrib_dir='/usr/local/'
 listen_pg_addr='0.0.0.0:6400'
+listen_grpc_addr='0.0.0.0:6401'
 listen_http_addr='0.0.0.0:9898'
 remote_storage={ endpoint='http://minio:9000', bucket_name='neon', bucket_region='eu-north-1', prefix_in_bucket='/pageserver' }
 control_plane_api='http://0.0.0.0:6666' # No storage controller in docker compose, specify a junk address

libs/neon-shmem/Cargo.toml

@@ -6,15 +6,26 @@ license.workspace = true
 
 [dependencies]
 thiserror.workspace = true
-nix.workspace=true
+nix.workspace = true
 workspace_hack = { version = "0.1", path = "../../workspace_hack" }
 libc.workspace = true
 lock_api.workspace = true
 rustc-hash.workspace = true
 
+[dev-dependencies]
+criterion = { workspace = true, features = ["html_reports"] }
+rand = "0.9"
+rand_distr = "0.5.1"
+xxhash-rust = { version = "0.8.15", features = ["xxh3"] }
+ahash.workspace = true
+twox-hash = { version = "2.1.1" }
+seahash = "4.1.0"
+hashbrown = { git = "https://github.com/quantumish/hashbrown.git", rev = "6610e6d" }
+
+
 [target.'cfg(target_os = "macos")'.dependencies]
 tempfile = "3.14.0"
 
-[dev-dependencies]
-rand.workspace = true
-rand_distr = "0.5.1"
+[[bench]]
+name = "hmap_resize"
+harness = false

libs/neon-shmem/benches/hmap_resize.rs

@@ -0,0 +1,330 @@
+use criterion::{BatchSize, BenchmarkId, Criterion, criterion_group, criterion_main};
+use neon_shmem::hash::HashMapAccess;
+use neon_shmem::hash::HashMapInit;
+use neon_shmem::hash::entry::Entry;
+use rand::distr::{Distribution, StandardUniform};
+use rand::prelude::*;
+use std::default::Default;
+use std::hash::BuildHasher;
+
+// Taken from bindings to C code
+
+#[derive(Clone, Debug, Hash, Eq, PartialEq)]
+#[repr(C)]
+pub struct FileCacheKey {
+    pub _spc_id: u32,
+    pub _db_id: u32,
+    pub _rel_number: u32,
+    pub _fork_num: u32,
+    pub _block_num: u32,
+}
+
+impl Distribution<FileCacheKey> for StandardUniform {
+    // questionable, but doesn't need to be good randomness
+    fn sample<R: Rng + ?Sized>(&self, rng: &mut R) -> FileCacheKey {
+        FileCacheKey {
+            _spc_id: rng.random(),
+            _db_id: rng.random(),
+            _rel_number: rng.random(),
+            _fork_num: rng.random(),
+            _block_num: rng.random(),
+        }
+    }
+}
+
+#[derive(Clone, Debug)]
+#[repr(C)]
+pub struct FileCacheEntry {
+    pub _offset: u32,
+    pub _access_count: u32,
+    pub _prev: *mut FileCacheEntry,
+    pub _next: *mut FileCacheEntry,
+    pub _state: [u32; 8],
+}
+
+impl FileCacheEntry {
+    fn dummy() -> Self {
+        Self {
+            _offset: 0,
+            _access_count: 0,
+            _prev: std::ptr::null_mut(),
+            _next: std::ptr::null_mut(),
+            _state: [0; 8],
+        }
+    }
+}
+
+// Utilities for applying operations.
+
+#[derive(Clone, Debug)]
+struct TestOp<K, V>(K, Option<V>);
+
+fn apply_op<K: Clone + std::hash::Hash + Eq, V, S: std::hash::BuildHasher>(
+    op: TestOp<K, V>,
+    map: &mut HashMapAccess<K, V, S>,
+) {
+    let entry = map.entry(op.0);
+
+    match op.1 {
+        Some(new) => match entry {
+            Entry::Occupied(mut e) => Some(e.insert(new)),
+            Entry::Vacant(e) => {
+                _ = e.insert(new).unwrap();
+                None
+            }
+        },
+        None => match entry {
+            Entry::Occupied(e) => Some(e.remove()),
+            Entry::Vacant(_) => None,
+        },
+    };
+}
+
+// Hash utilities
+
+struct SeaRandomState {
+    k1: u64,
+    k2: u64,
+    k3: u64,
+    k4: u64,
+}
+
+impl std::hash::BuildHasher for SeaRandomState {
+    type Hasher = seahash::SeaHasher;
+
+    fn build_hasher(&self) -> Self::Hasher {
+        seahash::SeaHasher::with_seeds(self.k1, self.k2, self.k3, self.k4)
+    }
+}
+
+impl SeaRandomState {
+    fn new() -> Self {
+        let mut rng = rand::rng();
+        Self {
+            k1: rng.random(),
+            k2: rng.random(),
+            k3: rng.random(),
+            k4: rng.random(),
+        }
+    }
+}
+
+fn small_benchs(c: &mut Criterion) {
+    let mut group = c.benchmark_group("Small maps");
+    group.sample_size(10);
+
+    group.bench_function("small_rehash", |b| {
+        let ideal_filled = 4_000_000;
+        let size = 5_000_000;
+        let mut writer = HashMapInit::new_resizeable(size, size * 2).attach_writer();
+        let mut rng = rand::rng();
+        while writer.get_num_buckets_in_use() < ideal_filled as usize {
+            let key: FileCacheKey = rng.random();
+            let val = FileCacheEntry::dummy();
+            apply_op(TestOp(key, Some(val)), &mut writer);
+        }
+        b.iter(|| writer.shuffle());
+    });
+
+    group.bench_function("small_rehash_xxhash", |b| {
+        let ideal_filled = 4_000_000;
+        let size = 5_000_000;
+        let mut writer = HashMapInit::new_resizeable(size, size * 2)
+            .with_hasher(twox_hash::xxhash64::RandomState::default())
+            .attach_writer();
+        let mut rng = rand::rng();
+        while writer.get_num_buckets_in_use() < ideal_filled as usize {
+            let key: FileCacheKey = rng.random();
+            let val = FileCacheEntry::dummy();
+            apply_op(TestOp(key, Some(val)), &mut writer);
+        }
+        b.iter(|| writer.shuffle());
+    });
+
+    group.bench_function("small_rehash_ahash", |b| {
+        let ideal_filled = 4_000_000;
+        let size = 5_000_000;
+        let mut writer = HashMapInit::new_resizeable(size, size * 2)
+            .with_hasher(ahash::RandomState::default())
+            .attach_writer();
+        let mut rng = rand::rng();
+        while writer.get_num_buckets_in_use() < ideal_filled as usize {
+            let key: FileCacheKey = rng.random();
+            let val = FileCacheEntry::dummy();
+            apply_op(TestOp(key, Some(val)), &mut writer);
+        }
+        b.iter(|| writer.shuffle());
+    });
+
+    group.bench_function("small_rehash_seahash", |b| {
+        let ideal_filled = 4_000_000;
+        let size = 5_000_000;
+        let mut writer = HashMapInit::new_resizeable(size, size * 2)
+            .with_hasher(SeaRandomState::new())
+            .attach_writer();
+        let mut rng = rand::rng();
+        while writer.get_num_buckets_in_use() < ideal_filled as usize {
+            let key: FileCacheKey = rng.random();
+            let val = FileCacheEntry::dummy();
+            apply_op(TestOp(key, Some(val)), &mut writer);
+        }
+        b.iter(|| writer.shuffle());
+    });
+
+    group.finish();
+}
+
+fn real_benchs(c: &mut Criterion) {
+    let mut group = c.benchmark_group("Realistic workloads");
+    group.sample_size(10);
+    group.bench_function("real_bulk_insert", |b| {
+        let size = 125_000_000;
+        let ideal_filled = 100_000_000;
+        let mut rng = rand::rng();
+        b.iter_batched(
+            || HashMapInit::new_resizeable(size, size * 2).attach_writer(),
+            |writer| {
+                for _ in 0..ideal_filled {
+                    let key: FileCacheKey = rng.random();
+                    let val = FileCacheEntry::dummy();
+                    let entry = writer.entry(key);
+                    match entry {
+                        Entry::Occupied(mut e) => {
+                            std::hint::black_box(e.insert(val));
+                        }
+                        Entry::Vacant(e) => {
+                            let _ = std::hint::black_box(e.insert(val).unwrap());
+                        }
+                    }
+                }
+            },
+            BatchSize::SmallInput,
+        )
+    });
+
+    group.bench_function("real_rehash", |b| {
+        let size = 125_000_000;
+        let ideal_filled = 100_000_000;
+        let mut writer = HashMapInit::new_resizeable(size, size).attach_writer();
+        let mut rng = rand::rng();
+        while writer.get_num_buckets_in_use() < ideal_filled {
+            let key: FileCacheKey = rng.random();
+            let val = FileCacheEntry::dummy();
+            apply_op(TestOp(key, Some(val)), &mut writer);
+        }
+        b.iter(|| writer.shuffle());
+    });
+
+    group.bench_function("real_rehash_hashbrown", |b| {
+        let size = 125_000_000;
+        let ideal_filled = 100_000_000;
+        let mut writer = hashbrown::raw::RawTable::new();
+        let mut rng = rand::rng();
+        let hasher = rustc_hash::FxBuildHasher;
+        unsafe {
+            writer
+                .resize(
+                    size,
+                    |(k, _)| hasher.hash_one(k),
+                    hashbrown::raw::Fallibility::Infallible,
+                )
+                .unwrap();
+        }
+        while writer.len() < ideal_filled as usize {
+            let key: FileCacheKey = rng.random();
+            let val = FileCacheEntry::dummy();
+            writer.insert(hasher.hash_one(&key), (key, val), |(k, _)| {
+                hasher.hash_one(k)
+            });
+        }
+        b.iter(|| unsafe {
+            writer.table.rehash_in_place(
+                &|table, index| {
+                    hasher.hash_one(
+                        &table
+                            .bucket::<(FileCacheKey, FileCacheEntry)>(index)
+                            .as_ref()
+                            .0,
+                    )
+                },
+                std::mem::size_of::<(FileCacheKey, FileCacheEntry)>(),
+                if std::mem::needs_drop::<(FileCacheKey, FileCacheEntry)>() {
+                    Some(|ptr| std::ptr::drop_in_place(ptr as *mut (FileCacheKey, FileCacheEntry)))
+                } else {
+                    None
+                },
+            )
+        });
+    });
+
+    for elems in [2, 4, 8, 16, 32, 64, 96, 112] {
+        group.bench_with_input(
+            BenchmarkId::new("real_rehash_varied", elems),
+            &elems,
+            |b, &size| {
+                let ideal_filled = size * 1_000_000;
+                let size = 125_000_000;
+                let mut writer = HashMapInit::new_resizeable(size, size).attach_writer();
+                let mut rng = rand::rng();
+                while writer.get_num_buckets_in_use() < ideal_filled as usize {
+                    let key: FileCacheKey = rng.random();
+                    let val = FileCacheEntry::dummy();
+                    apply_op(TestOp(key, Some(val)), &mut writer);
+                }
+                b.iter(|| writer.shuffle());
+            },
+        );
+        group.bench_with_input(
+            BenchmarkId::new("real_rehash_varied_hashbrown", elems),
+            &elems,
+            |b, &size| {
+                let ideal_filled = size * 1_000_000;
+                let size = 125_000_000;
+                let mut writer = hashbrown::raw::RawTable::new();
+                let mut rng = rand::rng();
+                let hasher = rustc_hash::FxBuildHasher;
+                unsafe {
+                    writer
+                        .resize(
+                            size,
+                            |(k, _)| hasher.hash_one(k),
+                            hashbrown::raw::Fallibility::Infallible,
+                        )
+                        .unwrap();
+                }
+                while writer.len() < ideal_filled as usize {
+                    let key: FileCacheKey = rng.random();
+                    let val = FileCacheEntry::dummy();
+                    writer.insert(hasher.hash_one(&key), (key, val), |(k, _)| {
+                        hasher.hash_one(k)
+                    });
+                }
+                b.iter(|| unsafe {
+                    writer.table.rehash_in_place(
+                        &|table, index| {
+                            hasher.hash_one(
+                                &table
+                                    .bucket::<(FileCacheKey, FileCacheEntry)>(index)
+                                    .as_ref()
+                                    .0,
+                            )
+                        },
+                        std::mem::size_of::<(FileCacheKey, FileCacheEntry)>(),
+                        if std::mem::needs_drop::<(FileCacheKey, FileCacheEntry)>() {
+                            Some(|ptr| {
+                                std::ptr::drop_in_place(ptr as *mut (FileCacheKey, FileCacheEntry))
+                            })
+                        } else {
+                            None
+                        },
+                    )
+                });
+            },
+        );
+    }
+
+    group.finish();
+}
+
+criterion_group!(benches, small_benchs, real_benchs);
+criterion_main!(benches);

libs/neon-shmem/src/hash.rs

@@ -16,6 +16,7 @@
 //!
 //! Concurrency is managed very simply: the entire map is guarded by one shared-memory RwLock.
 
+use std::fmt::Debug;
 use std::hash::{BuildHasher, Hash};
 use std::mem::MaybeUninit;
 
@@ -56,6 +57,22 @@ pub struct HashMapInit<'a, K, V, S = rustc_hash::FxBuildHasher> {
     num_buckets: u32,
 }
 
+impl<'a, K, V, S> Debug for HashMapInit<'a, K, V, S>
+where
+    K: Debug,
+    V: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("HashMapInit")
+            .field("shmem_handle", &self.shmem_handle)
+            .field("shared_ptr", &self.shared_ptr)
+            .field("shared_size", &self.shared_size)
+            // .field("hasher", &self.hasher)
+            .field("num_buckets", &self.num_buckets)
+            .finish()
+    }
+}
+
 /// This is a per-process handle to a hash table that (possibly) lives in shared memory.
 /// If a child process is launched with fork(), the child process should
 /// get its own HashMapAccess by calling HashMapInit::attach_writer/reader().
@@ -71,6 +88,20 @@ pub struct HashMapAccess<'a, K, V, S = rustc_hash::FxBuildHasher> {
 unsafe impl<K: Sync, V: Sync, S> Sync for HashMapAccess<'_, K, V, S> {}
 unsafe impl<K: Send, V: Send, S> Send for HashMapAccess<'_, K, V, S> {}
 
+impl<'a, K, V, S> Debug for HashMapAccess<'a, K, V, S>
+where
+    K: Debug,
+    V: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("HashMapAccess")
+            .field("shmem_handle", &self.shmem_handle)
+            .field("shared_ptr", &self.shared_ptr)
+            // .field("hasher", &self.hasher)
+            .finish()
+    }
+}
+
 impl<'a, K: Clone + Hash + Eq, V, S> HashMapInit<'a, K, V, S> {
     /// Change the 'hasher' used by the hash table.
     ///
@@ -298,7 +329,7 @@ where
 
     /// Get a reference to the entry containing a key.
     ///
-    /// NB: THis takes a write lock as there's no way to distinguish whether the intention
+    /// NB: This takes a write lock as there's no way to distinguish whether the intention
     /// is to use the entry for reading or for writing in advance.
     pub fn entry(&self, key: K) -> Entry<'a, '_, K, V> {
         let hash = self.get_hash_value(&key);

libs/neon-shmem/src/hash/core.rs

@@ -1,5 +1,6 @@
 //! Simple hash table with chaining.
 
+use std::fmt::Debug;
 use std::hash::Hash;
 use std::mem::MaybeUninit;
 
@@ -17,6 +18,19 @@ pub(crate) struct Bucket<K, V> {
     pub(crate) inner: Option<(K, V)>,
 }
 
+impl<K, V> Debug for Bucket<K, V>
+where
+    K: Debug,
+    V: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("Bucket")
+            .field("next", &self.next)
+            .field("inner", &self.inner)
+            .finish()
+    }
+}
+
 /// Core hash table implementation.
 pub(crate) struct CoreHashMap<'a, K, V> {
     /// Dictionary used to map hashes to bucket indices.
@@ -31,6 +45,22 @@ pub(crate) struct CoreHashMap<'a, K, V> {
     pub(crate) buckets_in_use: u32,
 }
 
+impl<'a, K, V> Debug for CoreHashMap<'a, K, V>
+where
+    K: Debug,
+    V: Debug,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("CoreHashMap")
+            .field("dictionary", &self.dictionary)
+            .field("buckets", &self.buckets)
+            .field("free_head", &self.free_head)
+            .field("alloc_limit", &self.alloc_limit)
+            .field("buckets_in_use", &self.buckets_in_use)
+            .finish()
+    }
+}
+
 /// Error for when there are no empty buckets left but one is needed.
 #[derive(Debug, PartialEq)]
 pub struct FullError;

libs/neon-shmem/src/hash/entry.rs

@@ -61,6 +61,10 @@ impl<K, V> OccupiedEntry<'_, '_, K, V> {
     ///
     /// This may result in multiple bucket accesses if the entry was obtained by index as the
     /// previous chain entry needs to be discovered in this case.
+    ///
+    /// # Panics
+    /// Panics if the `prev_pos` field is equal to [`PrevPos::Unknown`]. In practice, this means
+    /// the entry was obtained via calling something like [`super::HashMapAccess::entry_at_bucket`].
     pub fn remove(mut self) -> V {
         // If this bucket was queried by index, go ahead and follow its chain from the start.
         let prev = if let PrevPos::Unknown(hash) = self.prev_pos {

libs/neon-shmem/src/shmem.rs

@@ -21,6 +21,7 @@ use nix::unistd::ftruncate as nix_ftruncate;
 /// the underlying file is resized. Do not access the area beyond the current size. Currently, that
 /// will cause the file to be expanded, but we might use `mprotect()` etc. to enforce that in the
 /// future.
+#[derive(Debug)]
 pub struct ShmemHandle {
     /// memfd file descriptor
     fd: OwnedFd,
@@ -35,6 +36,7 @@ pub struct ShmemHandle {
 }
 
 /// This is stored at the beginning in the shared memory area.
+#[derive(Debug)]
 struct SharedStruct {
     max_size: usize,
 

libs/utils/src/lsn.rs

@@ -310,6 +310,11 @@ impl AtomicLsn {
         }
     }
 
+    /// Consumes the atomic and returns the contained value.
+    pub const fn into_inner(self) -> Lsn {
+        Lsn(self.inner.into_inner())
+    }
+
     /// Atomically retrieve the `Lsn` value from memory.
     pub fn load(&self) -> Lsn {
         Lsn(self.inner.load(Ordering::Acquire))

libs/walproposer/src/api_bindings.rs

@@ -341,6 +341,34 @@ extern "C-unwind" fn log_internal(
     }
 }
 
+/* BEGIN_HADRON */
+extern "C" fn reset_safekeeper_statuses_for_metrics(wp: *mut WalProposer, num_safekeepers: u32) {
+    unsafe {
+        let callback_data = (*(*wp).config).callback_data;
+        let api = callback_data as *mut Box<dyn ApiImpl>;
+        if api.is_null() {
+            return;
+        }
+        (*api).reset_safekeeper_statuses_for_metrics(&mut (*wp), num_safekeepers);
+    }
+}
+
+extern "C" fn update_safekeeper_status_for_metrics(
+    wp: *mut WalProposer,
+    sk_index: u32,
+    status: u8,
+) {
+    unsafe {
+        let callback_data = (*(*wp).config).callback_data;
+        let api = callback_data as *mut Box<dyn ApiImpl>;
+        if api.is_null() {
+            return;
+        }
+        (*api).update_safekeeper_status_for_metrics(&mut (*wp), sk_index, status);
+    }
+}
+/* END_HADRON */
+
 #[derive(Debug, PartialEq)]
 pub enum Level {
     Debug5,
@@ -414,6 +442,10 @@ pub(crate) fn create_api() -> walproposer_api {
         finish_sync_safekeepers: Some(finish_sync_safekeepers),
         process_safekeeper_feedback: Some(process_safekeeper_feedback),
         log_internal: Some(log_internal),
+        /* BEGIN_HADRON */
+        reset_safekeeper_statuses_for_metrics: Some(reset_safekeeper_statuses_for_metrics),
+        update_safekeeper_status_for_metrics: Some(update_safekeeper_status_for_metrics),
+        /* END_HADRON */
     }
 }
 
@@ -451,6 +483,8 @@ pub fn empty_shmem() -> crate::bindings::WalproposerShmemState {
         replica_promote: false,
         min_ps_feedback: empty_feedback,
         wal_rate_limiter: empty_wal_rate_limiter,
+        num_safekeepers: 0,
+        safekeeper_status: [0; 32],
     }
 }
 

libs/walproposer/src/walproposer.rs

@@ -159,6 +159,21 @@ pub trait ApiImpl {
     fn after_election(&self, _wp: &mut WalProposer) {
         todo!()
     }
+
+    /* BEGIN_HADRON */
+    fn reset_safekeeper_statuses_for_metrics(&self, _wp: &mut WalProposer, _num_safekeepers: u32) {
+        // Do nothing for testing purposes.
+    }
+
+    fn update_safekeeper_status_for_metrics(
+        &self,
+        _wp: &mut WalProposer,
+        _sk_index: u32,
+        _status: u8,
+    ) {
+        // Do nothing for testing purposes.
+    }
+    /* END_HADRON */
 }
 
 #[derive(Debug)]

pageserver/Cargo.toml

@@ -54,6 +54,7 @@ pageserver_api.workspace = true
 pageserver_client.workspace = true # for ResponseErrorMessageExt TOOD refactor that
 pageserver_compaction.workspace = true
 pageserver_page_api.workspace = true
+peekable.workspace = true
 pem.workspace = true
 pin-project-lite.workspace = true
 postgres_backend.workspace = true
@@ -66,6 +67,7 @@ postgres-types.workspace = true
 posthog_client_lite.workspace = true
 pprof.workspace = true
 pq_proto.workspace = true
+prost.workspace = true
 rand.workspace = true
 range-set-blaze = { version = "0.1.16", features = ["alloc"] }
 regex.workspace = true

pageserver/client_grpc/src/lib.rs

@@ -3,3 +3,4 @@ mod pool;
 mod retry;
 
 pub use client::{PageserverClient, ShardSpec};
+pub use pageserver_api::shard::ShardStripeSize; // used in ShardSpec

pageserver/page_api/src/model.rs

@@ -33,6 +33,8 @@ pub enum ProtocolError {
     Invalid(&'static str, String),
     #[error("required field '{0}' is missing")]
     Missing(&'static str),
+    #[error("invalid combination of not_modified_lsn '{0}' and request_lsn '{1}'")]
+    InvalidLsns(Lsn, Lsn),
 }
 
 impl ProtocolError {
@@ -85,9 +87,9 @@ impl TryFrom<proto::ReadLsn> for ReadLsn {
             return Err(ProtocolError::invalid("request_lsn", pb.request_lsn));
         }
         if pb.not_modified_since_lsn > pb.request_lsn {
-            return Err(ProtocolError::invalid(
-                "not_modified_since_lsn",
-                pb.not_modified_since_lsn,
+            return Err(ProtocolError::InvalidLsns(
+                Lsn(pb.not_modified_since_lsn),
+                Lsn(pb.request_lsn),
             ));
         }
         Ok(Self {

pageserver/pagebench/Cargo.toml

@@ -25,6 +25,9 @@ tracing.workspace = true
 tokio.workspace = true
 tokio-stream.workspace = true
 tokio-util.workspace = true
+axum.workspace = true
+http.workspace = true
+metrics.workspace = true
 tonic.workspace = true
 url.workspace = true
 

pageserver/pagebench/src/cmd/getpage_latest_lsn.rs

@@ -34,6 +34,10 @@ use crate::util::{request_stats, tokio_thread_local_stats};
 /// GetPage@LatestLSN, uniformly distributed across the compute-accessible keyspace.
 #[derive(clap::Parser)]
 pub(crate) struct Args {
+    #[clap(long, default_value = "false")]
+    grpc: bool,
+    #[clap(long, default_value = "false")]
+    grpc_stream: bool,
     #[clap(long, default_value = "http://localhost:9898")]
     mgmt_api_endpoint: String,
     /// Pageserver connection string. Supports postgresql:// and grpc:// protocols.
@@ -78,6 +82,9 @@ pub(crate) struct Args {
     #[clap(long)]
     set_io_mode: Option<pageserver_api::models::virtual_file::IoMode>,
 
+    #[clap(long)]
+    only_relnode: Option<u32>,
+
     /// Queue depth generated in each client.
     #[clap(long, default_value = "1")]
     queue_depth: NonZeroUsize,
@@ -92,10 +99,31 @@ pub(crate) struct Args {
     #[clap(long, default_value = "1")]
     batch_size: NonZeroUsize,
 
-    #[clap(long)]
-    only_relnode: Option<u32>,
-
     targets: Option<Vec<TenantTimelineId>>,
+
+    #[clap(long, default_value = "100")]
+    pool_max_consumers: NonZeroUsize,
+
+    #[clap(long, default_value = "5")]
+    pool_error_threshold: NonZeroUsize,
+
+    #[clap(long, default_value = "5000")]
+    pool_connect_timeout: NonZeroUsize,
+
+    #[clap(long, default_value = "1000")]
+    pool_connect_backoff: NonZeroUsize,
+
+    #[clap(long, default_value = "60000")]
+    pool_max_idle_duration: NonZeroUsize,
+
+    #[clap(long, default_value = "0")]
+    max_delay_ms: usize,
+
+    #[clap(long, default_value = "0")]
+    percent_drops: usize,
+
+    #[clap(long, default_value = "0")]
+    percent_hangs: usize,
 }
 
 /// State shared by all clients
@@ -152,7 +180,6 @@ pub(crate) fn main(args: Args) -> anyhow::Result<()> {
         main_impl(args, thread_local_stats)
     })
 }
-
 async fn main_impl(
     args: Args,
     all_thread_local_stats: AllThreadLocalStats<request_stats::Stats>,
@@ -317,6 +344,7 @@ async fn main_impl(
     let rps_period = args
         .per_client_rate
         .map(|rps_limit| Duration::from_secs_f64(1.0 / (rps_limit as f64)));
+
     let make_worker: &dyn Fn(WorkerId) -> Pin<Box<dyn Send + Future<Output = ()>>> = &|worker_id| {
         let ss = shared_state.clone();
         let cancel = cancel.clone();

pageserver/src/page_service.rs

@@ -453,6 +453,7 @@ impl TimelineHandles {
             handles: Default::default(),
         }
     }
+
     async fn get(
         &mut self,
         tenant_id: TenantId,

pgxn/neon/Makefile

@@ -5,10 +5,12 @@ MODULE_big = neon
 OBJS = \
 	$(WIN32RES) \
 	communicator.o \
+	communicator_new.o \
 	communicator_process.o \
 	extension_server.o \
 	file_cache.o \
 	hll.o \
+	lfc_prewarm.o \
 	libpagestore.o \
 	logical_replication_monitor.o \
 	neon.o \
@@ -67,6 +69,7 @@ WALPROP_OBJS = \
 
 # libcommunicator.a is built by cargo from the Rust sources under communicator/
 # subdirectory. `cargo build` also generates communicator_bindings.h.
+communicator_new.o: communicator/communicator_bindings.h
 communicator_process.o: communicator/communicator_bindings.h
 file_cache.o: communicator/communicator_bindings.h
 

pgxn/neon/communicator/Cargo.toml

@@ -17,12 +17,31 @@ rest_broker = []
 
 [dependencies]
 axum.workspace = true
+bytes.workspace = true
+clashmap.workspace = true
 http.workspace = true
+libc.workspace = true
+nix.workspace = true
+atomic_enum = "0.3.0"
+measured.workspace = true
+prometheus.workspace = true
+prost.workspace = true
+strum_macros.workspace = true
+thiserror.workspace = true
+tonic = { workspace = true, default-features = false, features=["codegen", "prost", "transport"] }
 tokio = { workspace = true, features = ["macros", "net", "io-util", "rt", "rt-multi-thread"] }
+tokio-pipe = { version = "0.2.12" }
 tracing.workspace = true
 tracing-subscriber.workspace = true
 
-measured.workspace = true
+metrics.workspace = true
+uring-common = { workspace = true, features = ["bytes"] }
+
+pageserver_client_grpc.workspace = true
+pageserver_api.workspace = true
+pageserver_page_api.workspace = true
+
+neon-shmem.workspace = true
 utils.workspace = true
 workspace_hack = { version = "0.1", path = "../../../workspace_hack" }
 

pgxn/neon/communicator/README.md

@@ -3,9 +3,18 @@
 This package provides the so-called "compute-pageserver communicator",
 or just "communicator" in short. The communicator is a separate
 background worker process that runs in the PostgreSQL server. It's
-part of the neon extension. Currently, it only provides an HTTP
-endpoint for metrics, but in the future it will evolve to handle all
-communications with the pageservers.
+part of the neon extension.
+
+The commuicator handles the communication with the pageservers, and
+also provides an HTTP endpoint for metrics over a local Unix Domain
+socket (aka. the "communicator control socket"). On the PostgreSQL
+side, the glue code in pgxn/neon/ uses the communicator to implement
+the PostgreSQL Storage Manager (SMGR) interface.
+
+## Design criteria
+
+- Low latency
+- Saturate a 10 Gbit / s network interface without becoming a bottleneck
 
 ## Source code view
 
@@ -14,10 +23,122 @@ pgxn/neon/communicator_process.c
     the glue that interacts with PostgreSQL code and the Rust
     code in the communicator process.
 
+pgxn/neon/communicator_new.c
+	Contains the backend code that interacts with the communicator
+	process.
 
-pgxn/neon/communicator/src/worker_process/
-    Worker process main loop and glue code
+pgxn/neon/communicator/src/backend_interface.rs
+	The entry point for calls from each backend.
+
+pgxn/neon/communicator/src/init.rs
+	Initialization at server startup
 
 At compilation time, pgxn/neon/communicator/ produces a static
 library, libcommunicator.a. It is linked to the neon.so extension
 library.
+
+The real networking code, which is independent of PostgreSQL, is in
+the pageserver/client_grpc crate.
+
+## Process view
+
+The communicator runs in a dedicated background worker process, the
+"communicator process". The communicator uses a multi-threaded Tokio
+runtime to execute the IO requests. So the communicator process has
+multiple threads running. That's unusual for Postgres processes and
+care must be taken to make that work.
+
+### Backend <-> worker communication
+
+Each backend has a number of I/O request slots in shared memory. The
+slots are statically allocated for each backend, and must not be
+accessed by other backends. The worker process reads requests from the
+shared memory slots, and writes responses back to the slots.
+
+Here's an example snapshot of the system, when two requests from two
+different backends are in progress:
+
+```
+Backends           Request slots          Communicator process
+---------          -------------          --------------------
+
+Backend 1          1: Idle
+                   2: Idle
+                   3: Processing          tokio task handling request 3
+
+Backend 2          4: Completed
+                   5: Processing          tokio task handling request 5
+                   6: Idle
+
+...                ...
+```
+
+To submit an IO request, the backend first picks one of its Idle
+slots, writes the IO request in the slot, and updates it to
+'Submitted' state. That transfers the ownership of the slot to the
+worker process, until the worker process marks the request as
+Completed. The worker process spawns a separate Tokio task for each
+request.
+
+To inform the worker process that a request slot has a pending IO
+request, there's a pipe shared by the worker process and all backend
+processes. The backend writes the index of the request slot to the
+pipe after changing the slot's state to Submitted. This wakes up the
+worker process.
+
+(Note that the pipe is just used for wakeups, but the worker process
+is free to pick up Submitted IO requests even without receiving the
+wakeup. As of this writing, it doesn't do that, but it might be useful
+in the future to reduce latency even further, for example.)
+
+When the worker process has completed processing the request, it
+writes the result back in the request slot. A GetPage request can also
+contain a pointer to buffer in the shared buffer cache. In that case,
+the worker process writes the resulting page contents directly to the
+buffer, and just a result code in the request slot. It then updates
+the 'state' field to Completed, which passes the owner ship back to
+the originating backend. Finally, it signals the process Latch of the
+originating backend, waking it up.
+
+### Differences between PostgreSQL v16, v17 and v18
+
+PostgreSQL v18 introduced the new AIO mechanism. The PostgreSQL AIO
+mechanism uses a very similar mechanism as described in the previous
+section, for the communication between AIO worker processes and
+backends. With our communicator, the AIO worker processes are not
+used, but we use the same PgAioHandle request slots as in upstream.
+For Neon-specific IO requests like GetDbSize, a neon request slot is
+used. But for the actual IO requests, the request slot merely contains
+a pointer to the PgAioHandle slot. The worker process updates the
+status of that, calls the IO callbacks upon completionetc, just like
+the upstream AIO worker processes do.
+
+## Sequence diagram
+
+                      neon
+    PostgreSQL     extension       backend_interface.rs  worker_process.rs    processor    tonic
+       |               .                    .                   .                 .
+	   | smgr_read()   .                    .                   .                 .
+	   +-------------> +                    .                   .                 .
+	   .               |                    .                   .                 .
+	   .               |  rcommunicator_    .                   .                 .
+	   .               | get_page_at_lsn    .                   .                 .
+	   .               +------------------> +                   .                 .
+                                            |                   .                 .
+                                            | write request to  .                 .                 .
+                                            | slot              .                 .
+                                            |                   .                 .
+                                            |                   .                 .
+											| submit_request()  .                 .
+											+-----------------> +                 .
+											|                   |                 .
+											|					| db_size_request .               .
+																+---------------->.
+																                  . TODO
+
+
+
+### Compute <-> pageserver protocol
+
+The protocol between Compute and the pageserver is based on gRPC. See `protos/`.
+

pgxn/neon/communicator/src/backend_comms.rs

@@ -0,0 +1,224 @@
+//! This module implements a request/response "slot" for submitting
+//! requests from backends to the communicator process.
+//!
+//! NB: The "backend" side of this code runs in Postgres backend processes,
+//! which means that it is not safe to use the 'tracing' crate for logging, nor
+//! to launch threads or use tokio tasks!
+
+use std::cell::UnsafeCell;
+use std::sync::atomic::{AtomicI32, Ordering};
+
+use crate::neon_request::{NeonIORequest, NeonIOResult};
+
+use atomic_enum::atomic_enum;
+
+/// One request/response slot. Each backend has its own set of slots that it
+/// uses.
+///
+/// This is the moral equivalent of PgAioHandle for Postgres AIO requests
+/// Like PgAioHandle, try to keep this small.
+///
+/// There is an array of these in shared memory. Therefore, this must be Sized.
+///
+/// ## Lifecycle of a request
+///
+/// A slot is always owned by either the backend process or the communicator
+/// process, depending on the 'state'. Only the owning process is allowed to
+/// read or modify the slot, except for reading the 'state' itself to check who
+/// owns it.
+///
+/// A slot begins in the Idle state, where it is owned by the backend process.
+/// To submit a request, the backend process fills the slot with the request
+/// data, and changes it to the Submitted state. After changing the state, the
+/// slot is owned by the communicator process, and the backend is not allowed
+/// to access it until the communicator process marks it as Completed.
+///
+/// When the communicator process sees that the slot is in Submitted state, it
+/// starts to process the request. After processing the request, it stores the
+/// result in the slot, and changes the state to Completed. It is now owned by
+/// the backend process again, which may now read the result, and reuse the
+/// slot for a new request.
+///
+/// For correctness of the above protocol, we really only need two states:
+/// "owned by backend" and "owned by communicator process". But to help with
+/// debugging and better assertions, there are a few more states. When the
+/// backend starts to fill in the request details in the slot, it first sets the
+/// state from Idle to Filling, and when it's done with that, from Filling to
+/// Submitted. In the Filling state, the slot is still owned by the
+/// backend. Similarly, when the communicator process starts to process a
+/// request, it sets it to Processing state first, but the slot is still owned
+/// by the communicator process.
+///
+/// This struct doesn't handle waking up the communicator process when a request
+/// has been submitted or when a response is ready. The 'owner_procno' is used
+/// for waking up the backend on completion, but that happens elsewhere.
+pub struct NeonIORequestSlot {
+    /// similar to PgAioHandleState
+    state: AtomicNeonIORequestSlotState,
+
+    /// The owning process's ProcNumber. The worker process uses this to set the
+    /// process's latch on completion.
+    ///
+    /// (This could be calculated from num_neon_request_slots_per_backend and
+    /// the index of this slot in the overall 'neon_requst_slots array'. But we
+    /// prefer the communicator process to not know how the request slots are
+    /// divided between the backends.)
+    owner_procno: AtomicI32,
+
+    /// SAFETY: This is modified by submit_request(), after it has established
+    /// ownership of the slot by setting state from Idle to Filling
+    request: UnsafeCell<NeonIORequest>,
+
+    /// Valid when state is Completed
+    ///
+    /// SAFETY: This is modified by RequestProcessingGuard::complete(). There
+    /// can be only one RequestProcessingGuard outstanding for a slot at a time,
+    /// because it is returned by start_processing_request() which checks the
+    /// state, so RequestProcessingGuard has exclusive access to the slot.
+    result: UnsafeCell<NeonIOResult>,
+}
+
+// The protocol described in the "Lifecycle of a request" section above ensures
+// the safe access to the fields
+unsafe impl Send for NeonIORequestSlot {}
+unsafe impl Sync for NeonIORequestSlot {}
+
+impl Default for NeonIORequestSlot {
+    fn default() -> NeonIORequestSlot {
+        NeonIORequestSlot {
+            owner_procno: AtomicI32::new(-1),
+            request: UnsafeCell::new(NeonIORequest::Empty),
+            result: UnsafeCell::new(NeonIOResult::Empty),
+            state: AtomicNeonIORequestSlotState::new(NeonIORequestSlotState::Idle),
+        }
+    }
+}
+
+#[atomic_enum]
+#[derive(Eq, PartialEq)]
+pub enum NeonIORequestSlotState {
+    Idle,
+
+    /// Backend is filling in the request
+    Filling,
+
+    /// Backend has submitted the request to the communicator, but the
+    /// communicator process has not yet started processing it.
+    Submitted,
+
+    /// Communicator is processing the request
+    Processing,
+
+    /// Communicator has completed the request, and the 'result' field is now
+    /// valid, but the backend has not read the result yet.
+    Completed,
+}
+
+impl NeonIORequestSlot {
+    /// Write a request to the slot, and mark it as Submitted.
+    ///
+    /// Note: This does not wake up the worker process to actually process
+    /// the request. It's the caller's responsibility to do that.
+    pub fn submit_request(&self, request: &NeonIORequest, proc_number: i32) {
+        // Verify that the slot is in Idle state previously, and put it in
+        // Filling state.
+        //
+        // XXX: This step isn't strictly necessary. Assuming the caller didn't
+        // screw up and try to use a slot that's already in use, we could fill
+        // the slot and switch it directly from Idle to Submitted state.
+        if let Err(s) = self.state.compare_exchange(
+            NeonIORequestSlotState::Idle,
+            NeonIORequestSlotState::Filling,
+            Ordering::Relaxed,
+            Ordering::Relaxed,
+        ) {
+            panic!("unexpected state in request slot: {s:?}");
+        }
+
+        // Fill in the request details
+        self.owner_procno.store(proc_number, Ordering::Relaxed);
+        unsafe { *self.request.get() = *request }
+
+        // This synchronizes-with store/swap in [`start_processing_request`].
+        // Note that this ensures that the previous non-atomic writes visible
+        // to other threads too.
+        self.state
+            .store(NeonIORequestSlotState::Submitted, Ordering::Release);
+    }
+
+    pub fn get_state(&self) -> NeonIORequestSlotState {
+        self.state.load(Ordering::Relaxed)
+    }
+
+    pub fn try_get_result(&self) -> Option<NeonIOResult> {
+        // This synchronizes-with the store/swap in [`RequestProcessingGuard::completed`]
+        let state = self.state.load(Ordering::Acquire);
+        if state == NeonIORequestSlotState::Completed {
+            let result = unsafe { *self.result.get() };
+            self.state
+                .store(NeonIORequestSlotState::Idle, Ordering::Relaxed);
+            Some(result)
+        } else {
+            None
+        }
+    }
+
+    /// Read the IO request from the slot indicated in the wakeup
+    pub fn start_processing_request<'a>(&'a self) -> Option<RequestProcessingGuard<'a>> {
+        // XXX: using atomic load rather than compare_exchange would be
+        // sufficient here, as long as the communicator process has _some_ means
+        // of tracking which requests it's already processing. That could be a
+        // flag somewhere in communicator's private memory, for example.
+        //
+        // This synchronizes-with the store in [`submit_request`].
+        if let Err(s) = self.state.compare_exchange(
+            NeonIORequestSlotState::Submitted,
+            NeonIORequestSlotState::Processing,
+            Ordering::Acquire,
+            Ordering::Relaxed,
+        ) {
+            // FIXME surprising state. This is unexpected at the moment, but if we
+            // started to process requests more aggressively, without waiting for the
+            // read from the pipe, then this could happen
+            panic!("unexpected state in request slot: {s:?}");
+        }
+
+        Some(RequestProcessingGuard(self))
+    }
+}
+
+/// [`NeonIORequestSlot::start_processing_request`] returns this guard object to
+/// indicate that the the caller now "owns" the slot, until it calls
+/// [`RequestProcessingGuard::completed`].
+///
+/// TODO: implement Drop on this, to mark the request as Aborted or Errored
+/// if [`RequestProcessingGuard::completed`] is not called.
+pub struct RequestProcessingGuard<'a>(&'a NeonIORequestSlot);
+
+unsafe impl<'a> Send for RequestProcessingGuard<'a> {}
+unsafe impl<'a> Sync for RequestProcessingGuard<'a> {}
+
+impl<'a> RequestProcessingGuard<'a> {
+    pub fn get_request(&self) -> &NeonIORequest {
+        unsafe { &*self.0.request.get() }
+    }
+
+    pub fn get_owner_procno(&self) -> i32 {
+        self.0.owner_procno.load(Ordering::Relaxed)
+    }
+
+    pub fn completed(self, result: NeonIOResult) {
+        // Store the result to the slot.
+        unsafe {
+            *self.0.result.get() = result;
+        };
+
+        // Mark the request as completed. After that, we no longer have
+        // ownership of the slot, and must not modify it.
+        let old_state = self
+            .0
+            .state
+            .swap(NeonIORequestSlotState::Completed, Ordering::Release);
+        assert!(old_state == NeonIORequestSlotState::Processing);
+    }
+}

pgxn/neon/communicator/src/backend_interface.rs

@@ -0,0 +1,296 @@
+//! This code runs in each backend process. That means that launching Rust threads, panicking
+//! etc. is forbidden!
+
+use std::os::fd::OwnedFd;
+
+use crate::backend_comms::NeonIORequestSlot;
+use crate::init::CommunicatorInitStruct;
+use crate::integrated_cache::{BackendCacheReadOp, IntegratedCacheReadAccess};
+use crate::neon_request::{CCachedGetPageVResult, CLsn, COid};
+use crate::neon_request::{NeonIORequest, NeonIOResult};
+
+use utils::lsn::Lsn;
+
+pub struct CommunicatorBackendStruct<'t> {
+    my_proc_number: i32,
+
+    neon_request_slots: &'t [NeonIORequestSlot],
+
+    submission_pipe_write_fd: OwnedFd,
+
+    pending_cache_read_op: Option<BackendCacheReadOp<'t>>,
+
+    integrated_cache: &'t IntegratedCacheReadAccess<'t>,
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn rcommunicator_backend_init(
+    cis: Box<CommunicatorInitStruct>,
+    my_proc_number: i32,
+) -> &'static mut CommunicatorBackendStruct<'static> {
+    if my_proc_number < 0 {
+        panic!("cannot attach to communicator shared memory with procnumber {my_proc_number}");
+    }
+
+    let integrated_cache = Box::leak(Box::new(cis.integrated_cache_init_struct.backend_init()));
+
+    let bs: &'static mut CommunicatorBackendStruct =
+        Box::leak(Box::new(CommunicatorBackendStruct {
+            my_proc_number,
+            neon_request_slots: cis.neon_request_slots,
+
+            submission_pipe_write_fd: cis.submission_pipe_write_fd,
+            pending_cache_read_op: None,
+
+            integrated_cache,
+        }));
+    bs
+}
+
+/// Start a request. You can poll for its completion and get the result by
+/// calling bcomm_poll_dbsize_request_completion(). The communicator will wake
+/// us up by setting our process latch, so to wait for the completion, wait on
+/// the latch and call bcomm_poll_dbsize_request_completion() every time the
+/// latch is set.
+///
+/// Safety: The C caller must ensure that the references are valid.
+/// The requested slot must be free, or this panics.
+#[unsafe(no_mangle)]
+pub extern "C" fn bcomm_start_io_request(
+    bs: &'_ mut CommunicatorBackendStruct,
+    slot_idx: i32,
+    request: &NeonIORequest,
+    immediate_result_ptr: &mut NeonIOResult,
+) -> i32 {
+    assert!(bs.pending_cache_read_op.is_none());
+
+    // Check if the request can be satisfied from the cache first
+    if let NeonIORequest::RelSize(req) = request {
+        if let Some(nblocks) = bs.integrated_cache.get_rel_size(&req.reltag()) {
+            *immediate_result_ptr = NeonIOResult::RelSize(nblocks);
+            return -1;
+        }
+    }
+
+    // Create neon request and submit it
+    bs.start_neon_io_request(slot_idx, request);
+
+    slot_idx
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn bcomm_start_get_page_v_request(
+    bs: &mut CommunicatorBackendStruct,
+    slot_idx: i32,
+    request: &NeonIORequest,
+    immediate_result_ptr: &mut CCachedGetPageVResult,
+) -> i32 {
+    let NeonIORequest::GetPageV(get_pagev_request) = request else {
+        panic!("invalid request passed to bcomm_start_get_page_v_request()");
+    };
+    assert!(matches!(request, NeonIORequest::GetPageV(_)));
+    assert!(bs.pending_cache_read_op.is_none());
+
+    // Check if the request can be satisfied from the cache first
+    let mut all_cached = true;
+    let mut read_op = bs.integrated_cache.start_read_op();
+    for i in 0..get_pagev_request.nblocks {
+        if let Some(cache_block) = read_op.get_page(
+            &get_pagev_request.reltag(),
+            get_pagev_request.block_number + i as u32,
+        ) {
+            immediate_result_ptr.cache_block_numbers[i as usize] = cache_block;
+        } else {
+            // not found in cache
+            all_cached = false;
+            break;
+        }
+    }
+    if all_cached {
+        bs.pending_cache_read_op = Some(read_op);
+        return -1;
+    }
+
+    // Create neon request and submit it
+    bs.start_neon_io_request(slot_idx, request);
+
+    slot_idx
+}
+
+/// Check if a request has completed. Returns:
+///
+/// -1 if the request is still being processed
+/// 0 on success
+#[unsafe(no_mangle)]
+pub extern "C" fn bcomm_poll_request_completion(
+    bs: &mut CommunicatorBackendStruct,
+    request_slot_idx: u32,
+    result_p: &mut NeonIOResult,
+) -> i32 {
+    match bs.neon_request_slots[request_slot_idx as usize].try_get_result() {
+        None => -1, // still processing
+        Some(result) => {
+            *result_p = result;
+            0
+        }
+    }
+}
+
+/// Check if a request has completed. Returns:
+///
+/// 'false' if the slot is Idle. The backend process has ownership.
+/// 'true' if the slot is busy, and should be polled for result.
+#[unsafe(no_mangle)]
+pub extern "C" fn bcomm_get_request_slot_status(
+    bs: &mut CommunicatorBackendStruct,
+    request_slot_idx: u32,
+) -> bool {
+    use crate::backend_comms::NeonIORequestSlotState;
+    match bs.neon_request_slots[request_slot_idx as usize].get_state() {
+        NeonIORequestSlotState::Idle => false,
+        NeonIORequestSlotState::Filling => {
+            // 'false' would be the right result here. However, this
+            // is a very transient state. The C code should never
+            // leave a slot in this state, so if it sees that,
+            // something's gone wrong and it's not clear what to do
+            // with it.
+            panic!("unexpected Filling state in request slot {request_slot_idx}");
+        }
+        NeonIORequestSlotState::Submitted => true,
+        NeonIORequestSlotState::Processing => true,
+        NeonIORequestSlotState::Completed => true,
+    }
+}
+
+// LFC functions
+
+/// Finish a local file cache read
+///
+//
+#[unsafe(no_mangle)]
+pub extern "C" fn bcomm_finish_cache_read(bs: &mut CommunicatorBackendStruct) -> bool {
+    if let Some(op) = bs.pending_cache_read_op.take() {
+        op.finish()
+    } else {
+        panic!("bcomm_finish_cache_read() called with no cached read pending");
+    }
+}
+
+/// Check if LFC contains the given buffer, and update its last-written LSN if not.
+///
+/// This is used in WAL replay in read replica, to skip updating pages that are
+/// not in cache.
+#[unsafe(no_mangle)]
+pub extern "C" fn bcomm_update_lw_lsn_for_block_if_not_cached(
+    bs: &mut CommunicatorBackendStruct,
+    spc_oid: COid,
+    db_oid: COid,
+    rel_number: u32,
+    fork_number: u8,
+    block_number: u32,
+    lsn: CLsn,
+) -> bool {
+    bs.integrated_cache.update_lw_lsn_for_block_if_not_cached(
+        &pageserver_page_api::RelTag {
+            spcnode: spc_oid,
+            dbnode: db_oid,
+            relnode: rel_number,
+            forknum: fork_number,
+        },
+        block_number,
+        Lsn(lsn),
+    )
+}
+
+#[repr(C)]
+#[derive(Clone, Debug)]
+pub struct FileCacheIterator {
+    next_bucket: u64,
+
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+}
+
+/// Iterate over LFC contents
+#[allow(clippy::missing_safety_doc)]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn bcomm_cache_iterate_begin(
+    _bs: &mut CommunicatorBackendStruct,
+    iter: *mut FileCacheIterator,
+) {
+    unsafe { (*iter).next_bucket = 0 };
+}
+
+#[allow(clippy::missing_safety_doc)]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn bcomm_cache_iterate_next(
+    bs: &mut CommunicatorBackendStruct,
+    iter: *mut FileCacheIterator,
+) -> bool {
+    use crate::integrated_cache::GetBucketResult;
+    loop {
+        let next_bucket = unsafe { (*iter).next_bucket } as usize;
+        match bs.integrated_cache.get_bucket(next_bucket) {
+            GetBucketResult::Occupied(rel, blk) => {
+                unsafe {
+                    (*iter).spc_oid = rel.spcnode;
+                    (*iter).db_oid = rel.dbnode;
+                    (*iter).rel_number = rel.relnode;
+                    (*iter).fork_number = rel.forknum;
+                    (*iter).block_number = blk;
+
+                    (*iter).next_bucket += 1;
+                }
+                break true;
+            }
+            GetBucketResult::Vacant => {
+                unsafe {
+                    (*iter).next_bucket += 1;
+                }
+                continue;
+            }
+            GetBucketResult::OutOfBounds => {
+                break false;
+            }
+        }
+    }
+}
+
+#[allow(clippy::missing_safety_doc)]
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn bcomm_cache_get_num_pages_used(bs: &mut CommunicatorBackendStruct) -> u64 {
+    bs.integrated_cache.get_num_buckets_in_use() as u64
+}
+
+impl<'t> CommunicatorBackendStruct<'t> {
+    /// The slot must be free, or this panics.
+    pub(crate) fn start_neon_io_request(&mut self, request_slot_idx: i32, request: &NeonIORequest) {
+        let my_proc_number = self.my_proc_number;
+
+        self.neon_request_slots[request_slot_idx as usize].submit_request(request, my_proc_number);
+
+        // Tell the communicator about it
+        self.notify_about_request(request_slot_idx);
+    }
+
+    /// Send a wakeup to the communicator process
+    fn notify_about_request(self: &CommunicatorBackendStruct<'t>, request_slot_idx: i32) {
+        // wake up communicator by writing the idx to the submission pipe
+        //
+
+        // This can block, if the pipe is full. That should be very rare,
+        // because the communicator tries hard to drain the pipe to prevent
+        // that. Also, there's a natural upper bound on how many wakeups can be
+        // queued up: there is only a limited number of request slots for each
+        // backend.
+        //
+        // If it does block very briefly, that's not too serious.
+        let idxbuf = request_slot_idx.to_ne_bytes();
+
+        let _res = nix::unistd::write(&self.submission_pipe_write_fd, &idxbuf);
+        // FIXME: check result, return any errors
+    }
+}

pgxn/neon/communicator/src/file_cache.rs

@@ -0,0 +1,162 @@
+//! Implement the "low-level" parts of the file cache.
+//!
+//! This module just deals with reading and writing the file, and keeping track
+//! which blocks in the cache file are in use and which are free. The "high
+//! level" parts of tracking which block in the cache file corresponds to which
+//! relation block is handled in 'integrated_cache' instead.
+//!
+//! This module is only used to access the file from the communicator
+//! process. The backend processes *also* read the file (and sometimes also
+//! write it? ), but the backends use direct C library calls for that.
+use std::fs::File;
+use std::os::unix::fs::FileExt;
+use std::path::Path;
+use std::sync::Arc;
+use std::sync::Mutex;
+
+use crate::BLCKSZ;
+
+use tokio::task::spawn_blocking;
+
+pub type CacheBlock = u64;
+
+pub const INVALID_CACHE_BLOCK: CacheBlock = u64::MAX;
+
+#[derive(Debug)]
+pub struct FileCache {
+    file: Arc<File>,
+
+    free_list: Mutex<FreeList>,
+
+    // metrics
+    max_blocks_gauge: metrics::IntGauge,
+    num_free_blocks_gauge: metrics::IntGauge,
+}
+
+// TODO: We keep track of all free blocks in this vec. That doesn't really scale.
+// Idea: when free_blocks fills up with more than 1024 entries, write them all to
+// one block on disk.
+#[derive(Debug)]
+struct FreeList {
+    next_free_block: CacheBlock,
+    max_blocks: u64,
+
+    free_blocks: Vec<CacheBlock>,
+}
+
+impl FileCache {
+    pub fn new(file_cache_path: &Path, mut initial_size: u64) -> Result<FileCache, std::io::Error> {
+        if initial_size < 100 {
+            tracing::warn!(
+                "min size for file cache is 100 blocks, {} requested",
+                initial_size
+            );
+            initial_size = 100;
+        }
+
+        let file = std::fs::OpenOptions::new()
+            .read(true)
+            .write(true)
+            .truncate(true)
+            .create(true)
+            .open(file_cache_path)?;
+
+        let max_blocks_gauge = metrics::IntGauge::new(
+            "file_cache_max_blocks",
+            "Local File Cache size in 8KiB blocks",
+        )
+        .unwrap();
+        let num_free_blocks_gauge = metrics::IntGauge::new(
+            "file_cache_num_free_blocks",
+            "Number of free 8KiB blocks in Local File Cache",
+        )
+        .unwrap();
+
+        tracing::info!("initialized file cache with {} blocks", initial_size);
+
+        Ok(FileCache {
+            file: Arc::new(file),
+            free_list: Mutex::new(FreeList {
+                next_free_block: 0,
+                max_blocks: initial_size,
+                free_blocks: Vec::new(),
+            }),
+            max_blocks_gauge,
+            num_free_blocks_gauge,
+        })
+    }
+
+    // File cache management
+
+    pub async fn read_block(
+        &self,
+        cache_block: CacheBlock,
+        mut dst: impl uring_common::buf::IoBufMut + Send + Sync,
+    ) -> Result<(), std::io::Error> {
+        assert!(dst.bytes_total() == BLCKSZ);
+        let file = self.file.clone();
+
+        let dst_ref = unsafe { std::slice::from_raw_parts_mut(dst.stable_mut_ptr(), BLCKSZ) };
+
+        spawn_blocking(move || file.read_exact_at(dst_ref, cache_block * BLCKSZ as u64)).await??;
+        Ok(())
+    }
+
+    pub async fn write_block(
+        &self,
+        cache_block: CacheBlock,
+        src: impl uring_common::buf::IoBuf + Send + Sync,
+    ) -> Result<(), std::io::Error> {
+        assert!(src.bytes_init() == BLCKSZ);
+        let file = self.file.clone();
+
+        let src_ref = unsafe { std::slice::from_raw_parts(src.stable_ptr(), BLCKSZ) };
+
+        spawn_blocking(move || file.write_all_at(src_ref, cache_block * BLCKSZ as u64)).await??;
+
+        Ok(())
+    }
+
+    pub fn alloc_block(&self) -> Option<CacheBlock> {
+        let mut free_list = self.free_list.lock().unwrap();
+        if let Some(x) = free_list.free_blocks.pop() {
+            return Some(x);
+        }
+        if free_list.next_free_block < free_list.max_blocks {
+            let result = free_list.next_free_block;
+            free_list.next_free_block += 1;
+            return Some(result);
+        }
+        None
+    }
+
+    pub fn dealloc_block(&self, cache_block: CacheBlock) {
+        let mut free_list = self.free_list.lock().unwrap();
+        free_list.free_blocks.push(cache_block);
+    }
+}
+
+impl metrics::core::Collector for FileCache {
+    fn desc(&self) -> Vec<&metrics::core::Desc> {
+        let mut descs = Vec::new();
+        descs.append(&mut self.max_blocks_gauge.desc());
+        descs.append(&mut self.num_free_blocks_gauge.desc());
+        descs
+    }
+    fn collect(&self) -> Vec<metrics::proto::MetricFamily> {
+        // Update the gauges with fresh values first
+        {
+            let free_list = self.free_list.lock().unwrap();
+            self.max_blocks_gauge.set(free_list.max_blocks as i64);
+
+            let total_free_blocks: i64 = free_list.free_blocks.len() as i64
+                + (free_list.max_blocks as i64 - free_list.next_free_block as i64);
+            self.num_free_blocks_gauge.set(total_free_blocks);
+        }
+
+        let mut values = Vec::new();
+        values.append(&mut self.max_blocks_gauge.collect());
+        values.append(&mut self.num_free_blocks_gauge.collect());
+        values
+    }
+}

pgxn/neon/communicator/src/global_allocator.rs

@@ -0,0 +1,109 @@
+//! Global allocator, for tracking memory usage of the Rust parts
+//!
+//! Postgres is designed to handle allocation failure (ie. malloc() returning NULL) gracefully.  It
+//! rolls backs the transaction and gives the user an "ERROR: out of memory" error. Rust code
+//! however panics if an allocation fails. We don't want that to ever happen, because an unhandled
+//! panic leads to Postgres crash and restart. Our strategy is to pre-allocate a large enough chunk
+//! of memory for use by the Rust code, so that the allocations never fail.
+//!
+//! To pick the size for the pre-allocated chunk, we have a metric to track the high watermark
+//! memory usage of all the Rust allocations in total.
+//!
+//! TODO:
+//!
+//! - Currently we just export the metrics. Actual allocations are still just passed through to
+//!   the system allocator.
+//! - Take padding etc. overhead into account
+
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+
+use metrics::IntGauge;
+
+struct MyAllocator {
+    allocations: AtomicU64,
+    deallocations: AtomicU64,
+
+    allocated: AtomicUsize,
+    high: AtomicUsize,
+}
+
+unsafe impl GlobalAlloc for MyAllocator {
+    unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+        self.allocations.fetch_add(1, Ordering::Relaxed);
+        let mut allocated = self.allocated.fetch_add(layout.size(), Ordering::Relaxed);
+        allocated += layout.size();
+        self.high.fetch_max(allocated, Ordering::Relaxed);
+        unsafe { System.alloc(layout) }
+    }
+
+    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+        self.deallocations.fetch_add(1, Ordering::Relaxed);
+        self.allocated.fetch_sub(layout.size(), Ordering::Relaxed);
+        unsafe { System.dealloc(ptr, layout) }
+    }
+}
+
+#[global_allocator]
+static GLOBAL: MyAllocator = MyAllocator {
+    allocations: AtomicU64::new(0),
+    deallocations: AtomicU64::new(0),
+    allocated: AtomicUsize::new(0),
+    high: AtomicUsize::new(0),
+};
+
+pub struct MyAllocatorCollector {
+    allocations: IntGauge,
+    deallocations: IntGauge,
+    allocated: IntGauge,
+    high: IntGauge,
+}
+
+impl MyAllocatorCollector {
+    pub fn new() -> MyAllocatorCollector {
+        MyAllocatorCollector {
+            allocations: IntGauge::new("allocations_total", "Number of allocations in Rust code")
+                .unwrap(),
+            deallocations: IntGauge::new(
+                "deallocations_total",
+                "Number of deallocations in Rust code",
+            )
+            .unwrap(),
+            allocated: IntGauge::new("allocated_total", "Bytes currently allocated").unwrap(),
+            high: IntGauge::new("allocated_high", "High watermark of allocated bytes").unwrap(),
+        }
+    }
+}
+
+impl metrics::core::Collector for MyAllocatorCollector {
+    fn desc(&self) -> Vec<&metrics::core::Desc> {
+        let mut descs = Vec::new();
+
+        descs.append(&mut self.allocations.desc());
+        descs.append(&mut self.deallocations.desc());
+        descs.append(&mut self.allocated.desc());
+        descs.append(&mut self.high.desc());
+
+        descs
+    }
+
+    fn collect(&self) -> Vec<metrics::proto::MetricFamily> {
+        let mut values = Vec::new();
+
+        // update the gauges
+        self.allocations
+            .set(GLOBAL.allocations.load(Ordering::Relaxed) as i64);
+        self.deallocations
+            .set(GLOBAL.allocations.load(Ordering::Relaxed) as i64);
+        self.allocated
+            .set(GLOBAL.allocated.load(Ordering::Relaxed) as i64);
+        self.high.set(GLOBAL.high.load(Ordering::Relaxed) as i64);
+
+        values.append(&mut self.allocations.collect());
+        values.append(&mut self.deallocations.collect());
+        values.append(&mut self.allocated.collect());
+        values.append(&mut self.high.collect());
+
+        values
+    }
+}

pgxn/neon/communicator/src/init.rs

@@ -0,0 +1,166 @@
+//! Initialization functions. These are executed in the postmaster process,
+//! at different stages of server startup.
+//!
+//!
+//! Communicator initialization steps:
+//!
+//! 1. At postmaster startup, before shared memory is allocated,
+//!    rcommunicator_shmem_size() is called to get the amount of
+//!    shared memory that this module needs.
+//!
+//! 2. Later, after the shared memory has been allocated,
+//!    rcommunicator_shmem_init() is called to initialize the shmem
+//!    area.
+//!
+//! Per process initialization:
+//!
+//! When a backend process starts up, it calls rcommunicator_backend_init().
+//! In the communicator worker process, other functions are called, see
+//! `worker_process` module.
+
+use std::ffi::c_int;
+use std::mem;
+use std::mem::MaybeUninit;
+use std::os::fd::OwnedFd;
+
+use crate::backend_comms::NeonIORequestSlot;
+use crate::integrated_cache::IntegratedCacheInitStruct;
+
+/// This struct is created in the postmaster process, and inherited to
+/// the communicator process and all backend processes through fork()
+#[repr(C)]
+pub struct CommunicatorInitStruct {
+    pub submission_pipe_read_fd: OwnedFd,
+    pub submission_pipe_write_fd: OwnedFd,
+
+    // Shared memory data structures
+    pub num_neon_request_slots: u32,
+
+    pub neon_request_slots: &'static [NeonIORequestSlot],
+
+    pub integrated_cache_init_struct: IntegratedCacheInitStruct<'static>,
+}
+
+impl std::fmt::Debug for CommunicatorInitStruct {
+    fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
+        fmt.debug_struct("CommunicatorInitStruct")
+            .field("submission_pipe_read_fd", &self.submission_pipe_read_fd)
+            .field("submission_pipe_write_fd", &self.submission_pipe_write_fd)
+            .field("num_neon_request_slots", &self.num_neon_request_slots)
+            .field("neon_request_slots length", &self.neon_request_slots.len())
+            .finish()
+    }
+}
+
+#[unsafe(no_mangle)]
+pub extern "C" fn rcommunicator_shmem_size(num_neon_request_slots: u32) -> u64 {
+    let mut size = 0;
+
+    size += mem::size_of::<NeonIORequestSlot>() * num_neon_request_slots as usize;
+
+    // For integrated_cache's Allocator. TODO: make this adjustable
+    size += IntegratedCacheInitStruct::shmem_size();
+
+    size as u64
+}
+
+/// Initialize the shared memory segment. Returns a backend-private
+/// struct, which will be inherited by backend processes through fork
+#[unsafe(no_mangle)]
+pub extern "C" fn rcommunicator_shmem_init(
+    submission_pipe_read_fd: c_int,
+    submission_pipe_write_fd: c_int,
+    num_neon_request_slots: u32,
+    shmem_area_ptr: *mut MaybeUninit<u8>,
+    shmem_area_len: u64,
+    initial_file_cache_size: u64,
+    max_file_cache_size: u64,
+) -> &'static mut CommunicatorInitStruct {
+    let shmem_area: &'static mut [MaybeUninit<u8>] =
+        unsafe { std::slice::from_raw_parts_mut(shmem_area_ptr, shmem_area_len as usize) };
+
+    let (neon_request_slots, remaining_area) =
+        alloc_array_from_slice::<NeonIORequestSlot>(shmem_area, num_neon_request_slots as usize);
+
+    for slot in neon_request_slots.iter_mut() {
+        slot.write(NeonIORequestSlot::default());
+    }
+
+    // 'neon_request_slots' is initialized now. (MaybeUninit::slice_assume_init_mut() is nightly-only
+    // as of this writing.)
+    let neon_request_slots = unsafe {
+        std::mem::transmute::<&mut [MaybeUninit<NeonIORequestSlot>], &mut [NeonIORequestSlot]>(
+            neon_request_slots,
+        )
+    };
+
+    // Give the rest of the area to the integrated cache
+    let integrated_cache_init_struct = IntegratedCacheInitStruct::shmem_init(
+        remaining_area,
+        initial_file_cache_size,
+        max_file_cache_size,
+    );
+
+    let (submission_pipe_read_fd, submission_pipe_write_fd) = unsafe {
+        use std::os::fd::FromRawFd;
+        (
+            OwnedFd::from_raw_fd(submission_pipe_read_fd),
+            OwnedFd::from_raw_fd(submission_pipe_write_fd),
+        )
+    };
+
+    let cis: &'static mut CommunicatorInitStruct = Box::leak(Box::new(CommunicatorInitStruct {
+        submission_pipe_read_fd,
+        submission_pipe_write_fd,
+
+        num_neon_request_slots,
+        neon_request_slots,
+
+        integrated_cache_init_struct,
+    }));
+
+    cis
+}
+
+pub fn alloc_from_slice<T>(
+    area: &mut [MaybeUninit<u8>],
+) -> (&mut MaybeUninit<T>, &mut [MaybeUninit<u8>]) {
+    let layout = std::alloc::Layout::new::<T>();
+
+    let area_start = area.as_mut_ptr();
+
+    // pad to satisfy alignment requirements
+    let padding = area_start.align_offset(layout.align());
+    if padding + layout.size() > area.len() {
+        panic!("out of memory");
+    }
+    let area = &mut area[padding..];
+    let (result_area, remain) = area.split_at_mut(layout.size());
+
+    let result_ptr: *mut MaybeUninit<T> = result_area.as_mut_ptr().cast();
+    let result = unsafe { result_ptr.as_mut().unwrap() };
+
+    (result, remain)
+}
+
+pub fn alloc_array_from_slice<T>(
+    area: &mut [MaybeUninit<u8>],
+    len: usize,
+) -> (&mut [MaybeUninit<T>], &mut [MaybeUninit<u8>]) {
+    let layout = std::alloc::Layout::new::<T>();
+
+    let area_start = area.as_mut_ptr();
+
+    // pad to satisfy alignment requirements
+    let padding = area_start.align_offset(layout.align());
+    if padding + layout.size() * len > area.len() {
+        panic!("out of memory");
+    }
+    let area = &mut area[padding..];
+    let (result_area, remain) = area.split_at_mut(layout.size() * len);
+
+    let result_ptr: *mut MaybeUninit<T> = result_area.as_mut_ptr().cast();
+    let result = unsafe { std::slice::from_raw_parts_mut(result_ptr.as_mut().unwrap(), len) };
+
+    (result, remain)
+}

pgxn/neon/communicator/src/lib.rs

@@ -1,5 +1,29 @@
+//! Three main parts:
+//! - async tokio communicator core, which receives requests and processes them.
+//! - Main loop and requests queues, which routes requests from backends to the core
+//! - the per-backend glue code, which submits requests
+
+mod backend_comms;
+
+// mark this 'pub', because these functions are called from C code. Otherwise, the compiler
+// complains about a bunch of structs and enum variants being unused, because it thinkgs
+// the functions that use them are never called. There are some C-callable functions in
+// other modules too, but marking this as pub is currently enough to silence the warnings
+//
+// TODO: perhaps collect *all* the extern "C" functions to one module?
+pub mod backend_interface;
+
+mod file_cache;
+mod init;
+mod integrated_cache;
+mod neon_request;
 mod worker_process;
 
+mod global_allocator;
+
 /// Name of the Unix Domain Socket that serves the metrics, and other APIs in the
 /// future. This is within the Postgres data directory.
 const NEON_COMMUNICATOR_SOCKET_NAME: &str = "neon-communicator.socket";
+
+// FIXME: get this from postgres headers somehow
+pub const BLCKSZ: usize = 8192;

pgxn/neon/communicator/src/neon_request.rs

@@ -0,0 +1,466 @@
+// Definitions of some core PostgreSQL datatypes.
+
+/// XLogRecPtr is defined in "access/xlogdefs.h" as:
+///
+/// ```
+/// typedef uint64 XLogRecPtr;
+/// ```
+/// cbindgen:no-export
+pub type XLogRecPtr = u64;
+
+pub type CLsn = XLogRecPtr;
+pub type COid = u32;
+
+// This conveniently matches PG_IOV_MAX
+pub const MAX_GETPAGEV_PAGES: usize = 32;
+
+pub const INVALID_BLOCK_NUMBER: u32 = u32::MAX;
+
+use std::ffi::CStr;
+
+use pageserver_page_api::{self as page_api, SlruKind};
+
+/// Request from a Postgres backend to the communicator process
+#[allow(clippy::large_enum_variant)]
+#[repr(C)]
+#[derive(Copy, Clone, Debug, strum_macros::EnumDiscriminants)]
+#[strum_discriminants(derive(measured::FixedCardinalityLabel))]
+pub enum NeonIORequest {
+    Empty,
+
+    // Read requests. These are C-friendly variants of the corresponding structs in
+    // pageserver_page_api.
+    RelSize(CRelSizeRequest),
+    GetPageV(CGetPageVRequest),
+    ReadSlruSegment(CReadSlruSegmentRequest),
+    PrefetchV(CPrefetchVRequest),
+    DbSize(CDbSizeRequest),
+
+    /// This is like GetPageV, but bypasses the LFC and allows specifiying the
+    /// request LSNs directly. For debugging purposes only.
+    GetPageVUncached(CGetPageVUncachedRequest),
+
+    // Write requests. These are needed to keep the relation size cache and LFC up-to-date.
+    // They are not sent to the pageserver.
+    WritePage(CWritePageRequest),
+    RelExtend(CRelExtendRequest),
+    RelZeroExtend(CRelZeroExtendRequest),
+    RelCreate(CRelCreateRequest),
+    RelTruncate(CRelTruncateRequest),
+    RelUnlink(CRelUnlinkRequest),
+
+    // Other requests
+    UpdateCachedRelSize(CUpdateCachedRelSizeRequest),
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub enum NeonIOResult {
+    Empty,
+    /// InvalidBlockNumber == 0xffffffff means "rel does not exist"
+    RelSize(u32),
+
+    /// the result pages are written to the shared memory addresses given in the request
+    GetPageV,
+    /// The result is written to the file, path to which is provided
+    /// in the request. The [`u64`] value here is the number of blocks.
+    ReadSlruSegment(u64),
+
+    /// A prefetch request returns as soon as the request has been received by the communicator.
+    /// It is processed in the background.
+    PrefetchVLaunched,
+
+    DbSize(u64),
+
+    // FIXME design compact error codes. Can't easily pass a string or other dynamic data.
+    // currently, this is 'errno'
+    Error(i32),
+
+    Aborted,
+
+    /// used for all write requests
+    WriteOK,
+}
+
+impl NeonIORequest {
+    /// All requests include a unique request ID, which can be used to trace the execution
+    /// of a request all the way to the pageservers. The request ID needs to be unique
+    /// within the lifetime of the Postgres instance (but not across servers or across
+    /// restarts of the same server).
+    pub fn request_id(&self) -> u64 {
+        use NeonIORequest::*;
+        match self {
+            Empty => 0,
+            RelSize(req) => req.request_id,
+            GetPageV(req) => req.request_id,
+            GetPageVUncached(req) => req.request_id,
+            ReadSlruSegment(req) => req.request_id,
+            PrefetchV(req) => req.request_id,
+            DbSize(req) => req.request_id,
+            WritePage(req) => req.request_id,
+            RelExtend(req) => req.request_id,
+            RelZeroExtend(req) => req.request_id,
+            RelCreate(req) => req.request_id,
+            RelTruncate(req) => req.request_id,
+            RelUnlink(req) => req.request_id,
+            UpdateCachedRelSize(req) => req.request_id,
+        }
+    }
+}
+
+/// Special quick result to a CGetPageVRequest request, indicating that the
+/// the requested pages are present in the local file cache. The backend can
+/// read the blocks directly from the given LFC blocks.
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CCachedGetPageVResult {
+    pub cache_block_numbers: [u64; MAX_GETPAGEV_PAGES],
+}
+
+/// ShmemBuf represents a buffer in shared memory.
+///
+/// SAFETY: The pointer must point to an area in shared memory. The functions allow you to liberally
+/// get a mutable pointer to the contents; it is the caller's responsibility to ensure that you
+/// don't access a buffer that's you're not allowed to. Inappropriate access to the buffer doesn't
+/// violate Rust's safety semantics, but it will mess up and crash Postgres.
+///
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct ShmemBuf {
+    // Pointer to where the result is written or where to read from. Must point into a buffer in shared memory!
+    pub ptr: *mut u8,
+}
+
+unsafe impl Send for ShmemBuf {}
+unsafe impl Sync for ShmemBuf {}
+
+unsafe impl uring_common::buf::IoBuf for ShmemBuf {
+    fn stable_ptr(&self) -> *const u8 {
+        self.ptr
+    }
+
+    fn bytes_init(&self) -> usize {
+        crate::BLCKSZ
+    }
+
+    fn bytes_total(&self) -> usize {
+        crate::BLCKSZ
+    }
+}
+
+unsafe impl uring_common::buf::IoBufMut for ShmemBuf {
+    fn stable_mut_ptr(&mut self) -> *mut u8 {
+        self.ptr
+    }
+
+    unsafe fn set_init(&mut self, pos: usize) {
+        if pos > crate::BLCKSZ {
+            panic!(
+                "set_init called past end of buffer, pos {}, buffer size {}",
+                pos,
+                crate::BLCKSZ
+            );
+        }
+    }
+}
+
+impl ShmemBuf {
+    pub fn as_mut_ptr(&self) -> *mut u8 {
+        self.ptr
+    }
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CRelSizeRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub allow_missing: bool,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CGetPageVRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+    pub nblocks: u8,
+
+    // These fields define where the result is written. Must point into a buffer in shared memory!
+    pub dest: [ShmemBuf; MAX_GETPAGEV_PAGES],
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CGetPageVUncachedRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+    pub nblocks: u8,
+
+    pub request_lsn: CLsn,
+    pub not_modified_since: CLsn,
+
+    // These fields define where the result is written. Must point into a buffer in shared memory!
+    pub dest: [ShmemBuf; MAX_GETPAGEV_PAGES],
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CReadSlruSegmentRequest {
+    pub request_id: u64,
+    pub slru_kind: SlruKind,
+    pub segment_number: u32,
+    pub request_lsn: CLsn,
+    /// Must be a null-terminated C string containing the file path
+    /// where the communicator will write the SLRU segment.
+    pub destination_file_path: ShmemBuf,
+}
+
+impl CReadSlruSegmentRequest {
+    /// Returns the file path where the communicator will write the
+    /// SLRU segment.
+    pub(crate) fn destination_file_path(&self) -> String {
+        unsafe { CStr::from_ptr(self.destination_file_path.as_mut_ptr() as *const _) }
+            .to_string_lossy()
+            .into_owned()
+    }
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CPrefetchVRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+    pub nblocks: u8,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CDbSizeRequest {
+    pub request_id: u64,
+    pub db_oid: COid,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CWritePageRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+    pub lsn: CLsn,
+
+    // `src` defines the new page contents. Must point into a buffer in shared memory!
+    pub src: ShmemBuf,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CRelExtendRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+    pub lsn: CLsn,
+
+    // `src` defines the new page contents. Must point into a buffer in shared memory!
+    pub src: ShmemBuf,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CRelZeroExtendRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub block_number: u32,
+    pub nblocks: u32,
+    pub lsn: CLsn,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CRelCreateRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub lsn: CLsn,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CRelTruncateRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub nblocks: u32,
+    pub lsn: CLsn,
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CRelUnlinkRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub lsn: CLsn,
+}
+
+impl CRelSizeRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CGetPageVRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CGetPageVUncachedRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CPrefetchVRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CWritePageRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CRelExtendRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CRelZeroExtendRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CRelCreateRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CRelTruncateRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+impl CRelUnlinkRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}
+
+#[repr(C)]
+#[derive(Copy, Clone, Debug)]
+pub struct CUpdateCachedRelSizeRequest {
+    pub request_id: u64,
+    pub spc_oid: COid,
+    pub db_oid: COid,
+    pub rel_number: u32,
+    pub fork_number: u8,
+    pub nblocks: u32,
+    pub lsn: CLsn,
+}
+
+impl CUpdateCachedRelSizeRequest {
+    pub fn reltag(&self) -> page_api::RelTag {
+        page_api::RelTag {
+            spcnode: self.spc_oid,
+            dbnode: self.db_oid,
+            relnode: self.rel_number,
+            forknum: self.fork_number,
+        }
+    }
+}

pgxn/neon/communicator/src/worker_process/callbacks.rs

@@ -4,10 +4,13 @@
 //!
 //! These are called from the communicator threads! Careful what you do, most Postgres
 //! functions are not safe to call in that context.
+use utils::lsn::Lsn;
 
 #[cfg(not(test))]
 unsafe extern "C" {
+    pub fn notify_proc_unsafe(procno: std::ffi::c_int);
     pub fn callback_set_my_latch_unsafe();
+    pub fn callback_get_request_lsn_unsafe() -> crate::neon_request::CLsn;
     pub fn callback_get_lfc_metrics_unsafe() -> LfcMetrics;
 }
 
@@ -16,20 +19,36 @@ unsafe extern "C" {
 // package, but the code coverage build still builds these and tries to link with the
 // external C code.)
 #[cfg(test)]
+unsafe fn notify_proc_unsafe(_procno: std::ffi::c_int) {
+    panic!("not usable in unit tests");
+}
+#[cfg(test)]
 unsafe fn callback_set_my_latch_unsafe() {
     panic!("not usable in unit tests");
 }
 #[cfg(test)]
+unsafe fn callback_get_request_lsn_unsafe() -> crate::neon_request::CLsn {
+    panic!("not usable in unit tests");
+}
+#[cfg(test)]
 unsafe fn callback_get_lfc_metrics_unsafe() -> LfcMetrics {
     panic!("not usable in unit tests");
 }
 
 // safe wrappers
 
+pub(super) fn notify_proc(procno: std::ffi::c_int) {
+    unsafe { notify_proc_unsafe(procno) };
+}
+
 pub(super) fn callback_set_my_latch() {
     unsafe { callback_set_my_latch_unsafe() };
 }
 
+pub(super) fn get_request_lsn() -> Lsn {
+    Lsn(unsafe { callback_get_request_lsn_unsafe() })
+}
+
 pub(super) fn callback_get_lfc_metrics() -> LfcMetrics {
     unsafe { callback_get_lfc_metrics_unsafe() }
 }

pgxn/neon/communicator/src/worker_process/control_socket.rs

@@ -19,71 +19,105 @@ use http::StatusCode;
 use http::header::CONTENT_TYPE;
 
 use measured::MetricGroup;
+use measured::metric::MetricEncoding;
+use measured::metric::gauge::GaugeState;
+use measured::metric::group::Encoding;
 use measured::text::BufferedTextEncoder;
 
 use std::io::ErrorKind;
+use std::sync::Arc;
 
 use tokio::net::UnixListener;
 
 use crate::NEON_COMMUNICATOR_SOCKET_NAME;
+use crate::worker_process::lfc_metrics::LfcMetricsCollector;
 use crate::worker_process::main_loop::CommunicatorWorkerProcessStruct;
 
-impl CommunicatorWorkerProcessStruct {
-    /// Launch the listener
-    pub(crate) async fn launch_control_socket_listener(
-        &'static self,
-    ) -> Result<(), std::io::Error> {
-        use axum::routing::get;
-        let app = Router::new()
-            .route("/metrics", get(get_metrics))
-            .route("/autoscaling_metrics", get(get_autoscaling_metrics))
-            .route("/debug/panic", get(handle_debug_panic))
-            .with_state(self);
+enum ControlSocketState<'a> {
+    Full(&'a CommunicatorWorkerProcessStruct<'a>),
+    Legacy(LegacyControlSocketState),
+}
 
-        // If the server is restarted, there might be an old socket still
-        // lying around. Remove it first.
-        match std::fs::remove_file(NEON_COMMUNICATOR_SOCKET_NAME) {
-            Ok(()) => {
-                tracing::warn!("removed stale control socket");
-            }
-            Err(e) if e.kind() == ErrorKind::NotFound => {}
-            Err(e) => {
-                tracing::error!("could not remove stale control socket: {e:#}");
-                // Try to proceed anyway. It will likely fail below though.
-            }
-        };
-
-        // Create the unix domain socket and start listening on it
-        let listener = UnixListener::bind(NEON_COMMUNICATOR_SOCKET_NAME)?;
-
-        tokio::spawn(async {
-            tracing::info!("control socket listener spawned");
-            axum::serve(listener, app)
-                .await
-                .expect("axum::serve never returns")
-        });
+struct LegacyControlSocketState {
+    pub(crate) lfc_metrics: LfcMetricsCollector,
+}
 
+impl<T> MetricGroup<T> for LegacyControlSocketState
+where
+    T: Encoding,
+    GaugeState: MetricEncoding<T>,
+{
+    fn collect_group_into(&self, enc: &mut T) -> Result<(), T::Err> {
+        self.lfc_metrics.collect_group_into(enc)?;
         Ok(())
     }
 }
 
+/// Launch the listener
+pub(crate) async fn launch_listener(
+    worker: Option<&'static CommunicatorWorkerProcessStruct<'static>>,
+) -> Result<(), std::io::Error> {
+    use axum::routing::get;
+
+    let state = match worker {
+        Some(worker) => ControlSocketState::Full(worker),
+        None => ControlSocketState::Legacy(LegacyControlSocketState {
+            lfc_metrics: LfcMetricsCollector,
+        }),
+    };
+
+    let app = Router::new()
+        .route("/metrics", get(get_metrics))
+        .route("/autoscaling_metrics", get(get_autoscaling_metrics))
+        .route("/debug/panic", get(handle_debug_panic))
+        .route("/debug/dump_cache_map", get(dump_cache_map))
+        .with_state(Arc::new(state));
+
+    // If the server is restarted, there might be an old socket still
+    // lying around. Remove it first.
+    match std::fs::remove_file(NEON_COMMUNICATOR_SOCKET_NAME) {
+        Ok(()) => {
+            tracing::warn!("removed stale control socket");
+        }
+        Err(e) if e.kind() == ErrorKind::NotFound => {}
+        Err(e) => {
+            tracing::error!("could not remove stale control socket: {e:#}");
+            // Try to proceed anyway. It will likely fail below though.
+        }
+    };
+
+    // Create the unix domain socket and start listening on it
+    let listener = UnixListener::bind(NEON_COMMUNICATOR_SOCKET_NAME)?;
+
+    tokio::spawn(async {
+        tracing::info!("control socket listener spawned");
+        axum::serve(listener, app)
+            .await
+            .expect("axum::serve never returns")
+    });
+
+    Ok(())
+}
+
 /// Expose all Prometheus metrics.
-async fn get_metrics(State(state): State<&CommunicatorWorkerProcessStruct>) -> Response {
-    tracing::trace!("/metrics requested");
-    metrics_to_response(&state).await
+async fn get_metrics(State(state): State<Arc<ControlSocketState<'_>>>) -> Response {
+    match state.as_ref() {
+        ControlSocketState::Full(worker) => metrics_to_response(&worker).await,
+        ControlSocketState::Legacy(legacy) => metrics_to_response(&legacy).await,
+    }
 }
 
 /// Expose Prometheus metrics, for use by the autoscaling agent.
 ///
 /// This is a subset of all the metrics.
-async fn get_autoscaling_metrics(
-    State(state): State<&CommunicatorWorkerProcessStruct>,
-) -> Response {
-    tracing::trace!("/metrics requested");
-    metrics_to_response(&state.lfc_metrics).await
+async fn get_autoscaling_metrics(State(state): State<Arc<ControlSocketState<'_>>>) -> Response {
+    match state.as_ref() {
+        ControlSocketState::Full(worker) => metrics_to_response(&worker.lfc_metrics).await,
+        ControlSocketState::Legacy(legacy) => metrics_to_response(&legacy.lfc_metrics).await,
+    }
 }
 
-async fn handle_debug_panic(State(_state): State<&CommunicatorWorkerProcessStruct>) -> Response {
+async fn handle_debug_panic(State(_state): State<Arc<ControlSocketState<'_>>>) -> Response {
     panic!("test HTTP handler task panic");
 }
 
@@ -100,3 +134,23 @@ async fn metrics_to_response(metrics: &(dyn MetricGroup<BufferedTextEncoder> + S
         .body(Body::from(enc.finish()))
         .unwrap()
 }
+
+async fn dump_cache_map(State(state): State<Arc<ControlSocketState<'_>>>) -> Response {
+    match state.as_ref() {
+        ControlSocketState::Full(worker) => {
+            let mut buf: Vec<u8> = Vec::new();
+            worker.cache.dump_map(&mut buf);
+
+            Response::builder()
+                .status(StatusCode::OK)
+                .header(CONTENT_TYPE, "application/text")
+                .body(Body::from(buf))
+                .unwrap()
+        }
+        ControlSocketState::Legacy(_) => Response::builder()
+            .status(StatusCode::NOT_FOUND)
+            .header(CONTENT_TYPE, "application/text")
+            .body(Body::from(Vec::new()))
+            .unwrap(),
+    }
+}

pgxn/neon/communicator/src/worker_process/in_progress_ios.rs

@@ -0,0 +1,95 @@
+//! Lock table to ensure that only one IO request is in flight for a given
+//! block (or relation or database metadata) at a time
+
+use std::cmp::Eq;
+use std::hash::Hash;
+use std::sync::Arc;
+
+use tokio::sync::{Mutex, OwnedMutexGuard};
+
+use clashmap::ClashMap;
+use clashmap::Entry;
+
+use pageserver_page_api::RelTag;
+
+#[derive(Clone, Eq, Hash, PartialEq)]
+pub enum RequestInProgressKey {
+    Db(u32),
+    Rel(RelTag),
+    Block(RelTag, u32),
+}
+
+type RequestId = u64;
+
+pub type RequestInProgressTable = MutexHashMap<RequestInProgressKey, RequestId>;
+
+// more primitive locking thingie:
+
+pub struct MutexHashMap<K, V>
+where
+    K: Clone + Eq + Hash,
+{
+    lock_table: ClashMap<K, (V, Arc<Mutex<()>>)>,
+}
+
+pub struct MutexHashMapGuard<'a, K, V>
+where
+    K: Clone + Eq + Hash,
+{
+    pub key: K,
+    map: &'a MutexHashMap<K, V>,
+    mutex: Arc<Mutex<()>>,
+    _guard: OwnedMutexGuard<()>,
+}
+
+impl<'a, K, V> Drop for MutexHashMapGuard<'a, K, V>
+where
+    K: Clone + Eq + Hash,
+{
+    fn drop(&mut self) {
+        let (_old_key, old_val) = self.map.lock_table.remove(&self.key).unwrap();
+        assert!(Arc::ptr_eq(&old_val.1, &self.mutex));
+
+        // the guard will be dropped as we return
+    }
+}
+
+impl<K, V> MutexHashMap<K, V>
+where
+    K: Clone + Eq + Hash,
+    V: std::fmt::Display + Copy,
+{
+    pub fn new() -> MutexHashMap<K, V> {
+        MutexHashMap {
+            lock_table: ClashMap::new(),
+        }
+    }
+
+    pub async fn lock<'a>(&'a self, key: K, val: V) -> MutexHashMapGuard<'a, K, V> {
+        let my_mutex = Arc::new(Mutex::new(()));
+        let my_guard = Arc::clone(&my_mutex).lock_owned().await;
+
+        loop {
+            let (request_id, lock) = match self.lock_table.entry(key.clone()) {
+                Entry::Occupied(e) => {
+                    let e = e.get();
+                    (e.0, Arc::clone(&e.1))
+                }
+                Entry::Vacant(e) => {
+                    e.insert((val, Arc::clone(&my_mutex)));
+                    break;
+                }
+            };
+            tracing::info!("waiting for conflicting IO {request_id} to complete");
+            let _ = lock.lock().await;
+            tracing::info!("conflicting IO {request_id} completed");
+        }
+
+        MutexHashMapGuard {
+            key,
+            map: self,
+            mutex: my_mutex,
+            _guard: my_guard,
+        }
+    }
+}

pgxn/neon/communicator/src/worker_process/main_loop.rs

@@ -1,34 +1,126 @@
+use std::collections::HashMap;
+use std::os::fd::AsRawFd;
+use std::os::fd::OwnedFd;
+use std::path::PathBuf;
 use std::str::FromStr as _;
 
+use crate::backend_comms::NeonIORequestSlot;
+use crate::file_cache::FileCache;
+use crate::global_allocator::MyAllocatorCollector;
+use crate::init::CommunicatorInitStruct;
+use crate::integrated_cache::{CacheResult, IntegratedCacheWriteAccess};
+use crate::neon_request::{CGetPageVRequest, CGetPageVUncachedRequest, CPrefetchVRequest};
+use crate::neon_request::{INVALID_BLOCK_NUMBER, NeonIORequest, NeonIOResult};
+use crate::worker_process::control_socket;
+use crate::worker_process::in_progress_ios::{RequestInProgressKey, RequestInProgressTable};
 use crate::worker_process::lfc_metrics::LfcMetricsCollector;
+use pageserver_client_grpc::{PageserverClient, ShardSpec, ShardStripeSize};
+use pageserver_page_api as page_api;
+
+use tokio::io::AsyncReadExt;
+use tokio_pipe::PipeRead;
+use uring_common::buf::IoBuf;
 
 use measured::MetricGroup;
 use measured::metric::MetricEncoding;
 use measured::metric::gauge::GaugeState;
 use measured::metric::group::Encoding;
+use measured::{Gauge, GaugeVec};
 use utils::id::{TenantId, TimelineId};
 
-pub struct CommunicatorWorkerProcessStruct {
+use super::callbacks::{get_request_lsn, notify_proc};
+
+use tracing::{debug, error, info, info_span, trace};
+
+use utils::lsn::Lsn;
+
+pub struct CommunicatorWorkerProcessStruct<'a> {
+    /// Tokio runtime that the main loop and any other related tasks runs in.
     runtime: tokio::runtime::Runtime,
 
+    /// Client to communicate with the pageserver
+    client: PageserverClient,
+
+    /// Request slots that backends use to send IO requests to the communicator.
+    neon_request_slots: &'a [NeonIORequestSlot],
+
+    /// Notification pipe. Backends use this to notify the communicator that a request is waiting to
+    /// be processed in one of the request slots.
+    submission_pipe_read_fd: OwnedFd,
+
+    /// Locking table for all in-progress IO requests.
+    in_progress_table: RequestInProgressTable,
+
+    /// Local File Cache, relation size tracking, last-written LSN tracking
+    pub(crate) cache: IntegratedCacheWriteAccess<'a>,
+
     /*** Metrics ***/
     pub(crate) lfc_metrics: LfcMetricsCollector,
+
+    request_counters: GaugeVec<RequestTypeLabelGroupSet>,
+
+    getpage_cache_misses_counter: Gauge,
+    getpage_cache_hits_counter: Gauge,
+
+    // For the requests that affect multiple blocks, have separate counters for the # of blocks affected
+    request_nblocks_counters: GaugeVec<RequestTypeLabelGroupSet>,
+
+    #[allow(dead_code)]
+    allocator_metrics: MyAllocatorCollector,
+}
+
+// Define a label group, consisting of 1 or more label values
+#[derive(measured::LabelGroup)]
+#[label(set = RequestTypeLabelGroupSet)]
+struct RequestTypeLabelGroup {
+    request_type: crate::neon_request::NeonIORequestDiscriminants,
+}
+
+impl RequestTypeLabelGroup {
+    fn from_req(req: &NeonIORequest) -> Self {
+        RequestTypeLabelGroup {
+            request_type: req.into(),
+        }
+    }
 }
 
 /// Launch the communicator process's Rust subsystems
+#[allow(clippy::too_many_arguments)]
+pub(super) fn init_legacy() -> Result<(), String> {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .thread_name("communicator thread")
+        .build()
+        .unwrap();
+
+    // Start the listener on the control socket
+    runtime
+        .block_on(control_socket::launch_listener(None))
+        .map_err(|e| e.to_string())?;
+
+    Box::leak(Box::new(runtime));
+
+    Ok(())
+}
+
+/// Launch the communicator process's Rust subsystems
+#[allow(clippy::too_many_arguments)]
 pub(super) fn init(
-    tenant_id: Option<&str>,
-    timeline_id: Option<&str>,
-) -> Result<&'static CommunicatorWorkerProcessStruct, String> {
+    cis: CommunicatorInitStruct,
+    tenant_id: &str,
+    timeline_id: &str,
+    auth_token: Option<&str>,
+    shard_map: HashMap<utils::shard::ShardIndex, String>,
+    stripe_size: Option<ShardStripeSize>,
+    initial_file_cache_size: u64,
+    file_cache_path: Option<PathBuf>,
+) -> Result<&'static CommunicatorWorkerProcessStruct<'static>, String> {
     // The caller validated these already
-    let _tenant_id = tenant_id
-        .map(TenantId::from_str)
-        .transpose()
-        .map_err(|e| format!("invalid tenant ID: {e}"))?;
-    let _timeline_id = timeline_id
-        .map(TimelineId::from_str)
-        .transpose()
-        .map_err(|e| format!("invalid timeline ID: {e}"))?;
+    let tenant_id = TenantId::from_str(tenant_id).map_err(|e| format!("invalid tenant ID: {e}"))?;
+    let timeline_id =
+        TimelineId::from_str(timeline_id).map_err(|e| format!("invalid timeline ID: {e}"))?;
+    let shard_spec =
+        ShardSpec::new(shard_map, stripe_size).map_err(|e| format!("invalid shard spec: {e}:"))?;
 
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
@@ -36,31 +128,710 @@ pub(super) fn init(
         .build()
         .unwrap();
 
+    let last_lsn = get_request_lsn();
+
+    let file_cache = if let Some(path) = file_cache_path {
+        Some(FileCache::new(&path, initial_file_cache_size).expect("could not create cache file"))
+    } else {
+        // FIXME: temporarily for testing, use LFC even if disabled
+        Some(
+            FileCache::new(&PathBuf::from("new_filecache"), 1000)
+                .expect("could not create cache file"),
+        )
+    };
+
+    // Initialize subsystems
+    let cache = cis
+        .integrated_cache_init_struct
+        .worker_process_init(last_lsn, file_cache);
+
+    debug!("Initialised integrated cache: {cache:?}");
+
+    let client = {
+        let _guard = runtime.enter();
+        PageserverClient::new(
+            tenant_id,
+            timeline_id,
+            shard_spec,
+            auth_token.map(|s| s.to_string()),
+            None,
+        )
+        .expect("could not create client")
+    };
+
     let worker_struct = CommunicatorWorkerProcessStruct {
         // Note: it's important to not drop the runtime, or all the tasks are dropped
         // too. Including it in the returned struct is one way to keep it around.
         runtime,
+        neon_request_slots: cis.neon_request_slots,
+        client,
+        cache,
+        submission_pipe_read_fd: cis.submission_pipe_read_fd,
+        in_progress_table: RequestInProgressTable::new(),
 
         // metrics
         lfc_metrics: LfcMetricsCollector,
+
+        request_counters: GaugeVec::new(),
+
+        getpage_cache_misses_counter: Gauge::new(),
+        getpage_cache_hits_counter: Gauge::new(),
+
+        request_nblocks_counters: GaugeVec::new(),
+
+        allocator_metrics: MyAllocatorCollector::new(),
     };
+
     let worker_struct = Box::leak(Box::new(worker_struct));
 
+    let main_loop_handle = worker_struct.runtime.spawn(worker_struct.run());
+    worker_struct.runtime.spawn(async {
+        let err = main_loop_handle.await.unwrap_err();
+        error!("error: {err:?}");
+    });
+
     // Start the listener on the control socket
     worker_struct
         .runtime
-        .block_on(worker_struct.launch_control_socket_listener())
+        .block_on(control_socket::launch_listener(Some(worker_struct)))
         .map_err(|e| e.to_string())?;
 
     Ok(worker_struct)
 }
 
-impl<T> MetricGroup<T> for CommunicatorWorkerProcessStruct
+impl<'t> CommunicatorWorkerProcessStruct<'t> {
+    /// Update the configuration
+    pub(super) fn update_shard_map(
+        &self,
+        new_shard_map: HashMap<utils::shard::ShardIndex, String>,
+        stripe_size: Option<ShardStripeSize>,
+    ) {
+        let shard_spec = ShardSpec::new(new_shard_map, stripe_size).expect("invalid shard spec");
+
+        {
+            let _in_runtime = self.runtime.enter();
+            if let Err(err) = self.client.update_shards(shard_spec) {
+                tracing::error!("could not update shard map: {err:?}");
+            }
+        }
+    }
+
+    /// Main loop of the worker process. Receive requests from the backends and process them.
+    pub(super) async fn run(&'static self) {
+        let mut idxbuf: [u8; 4] = [0; 4];
+
+        let mut submission_pipe_read =
+            PipeRead::try_from(self.submission_pipe_read_fd.as_raw_fd()).expect("invalid pipe fd");
+
+        loop {
+            // Wait for a backend to ring the doorbell
+            match submission_pipe_read.read(&mut idxbuf).await {
+                Ok(4) => {}
+                Ok(nbytes) => panic!("short read ({nbytes} bytes) on communicator pipe"),
+                Err(e) => panic!("error reading from communicator pipe: {e}"),
+            }
+            let slot_idx = u32::from_ne_bytes(idxbuf) as usize;
+
+            // Read the IO request from the slot indicated in the wakeup
+            let Some(slot) = self.neon_request_slots[slot_idx].start_processing_request() else {
+                // This currently should not happen. But if we had multiple threads picking up
+                // requests, and without waiting for the notifications, it could.
+                panic!("no request in slot");
+            };
+
+            // Ok, we have ownership of this request now. We must process it now, there's no going
+            // back.
+            //
+            // Spawn a separate task for every request. That's a little excessive for requests that
+            // can be quickly satisfied from the cache, but we expect that to be rare, because the
+            // requesting backend would have already checked the cache.
+            tokio::spawn(async move {
+                use tracing::Instrument;
+
+                let request_id = slot.get_request().request_id();
+                let owner_procno = slot.get_owner_procno();
+
+                let span = info_span!(
+                    "processing",
+                    request_id = request_id,
+                    slot_idx = slot_idx,
+                    procno = owner_procno,
+                );
+                async {
+                    // FIXME: as a temporary hack, abort the request if we don't get a response
+                    // promptly.
+                    //
+                    // Lots of regression tests are getting stuck and failing at the moment,
+                    // this makes them fail a little faster, which it faster to iterate.
+                    // This needs to be removed once more regression tests are passing.
+                    // See also similar hack in the backend code, in wait_request_completion()
+                    let result = tokio::time::timeout(
+                        tokio::time::Duration::from_secs(60),
+                        self.handle_request(slot.get_request()),
+                    )
+                    .await
+                    .unwrap_or_else(|_elapsed| {
+                        info!("request {request_id} timed out");
+                        NeonIOResult::Error(libc::ETIMEDOUT)
+                    });
+                    trace!("request {request_id} at slot {slot_idx} completed");
+
+                    // Ok, we have completed the IO. Mark the request as completed. After that,
+                    // we no longer have ownership of the slot, and must not modify it.
+                    slot.completed(result);
+
+                    // Notify the backend about the completion. (Note that the backend might see
+                    // the completed status even before this; this is just a wakeup)
+                    notify_proc(owner_procno);
+                }
+                .instrument(span)
+                .await
+            });
+        }
+    }
+
+    /// Compute the 'request_lsn' to use for a pageserver request
+    fn request_lsns(&self, not_modified_since_lsn: Lsn) -> page_api::ReadLsn {
+        let mut request_lsn = get_request_lsn();
+
+        // Is it possible that the last-written LSN is ahead of last flush LSN? Generally not, we
+        // shouldn't evict a page from the buffer cache before all its modifications have been
+        // safely flushed. That's the "WAL before data" rule. However, there are a few exceptions:
+        //
+        // - when creation an index: _bt_blwritepage logs the full page without flushing WAL before
+        // smgrextend (files are fsynced before build ends).
+        //
+        // XXX: If we make a request LSN greater than the current WAL flush LSN, the pageserver would
+        // block waiting for the WAL arrive, until we flush it and it propagates through the
+        // safekeepers to the pageserver. If there's nothing that forces the WAL to be flushed,
+        // the pageserver would get stuck waiting forever. To avoid that, all the write-
+        // functions in communicator_new.c call XLogSetAsyncXactLSN(). That nudges the WAL writer to
+        // perform the flush relatively soon.
+        //
+        // It would perhaps be nicer to do the WAL flush here, but it's tricky to call back into
+        // Postgres code to do that from here. That's why we rely on communicator_new.c to do the
+        // calls "pre-emptively".
+        //
+        // FIXME: Because of the above, it can still happen that the flush LSN is ahead of
+        // not_modified_since, if the WAL writer hasn't done the flush yet. It would be nice to know
+        // if there are other cases like that that we have mised, but unfortunately we cannot turn
+        // this into an assertion because of that legit case.
+        //
+        // See also the old logic in neon_get_request_lsns() C function
+        if not_modified_since_lsn > request_lsn {
+            tracing::info!(
+                "not_modified_since_lsn {} is ahead of last flushed LSN {}",
+                not_modified_since_lsn,
+                request_lsn
+            );
+            request_lsn = not_modified_since_lsn;
+        }
+
+        page_api::ReadLsn {
+            request_lsn,
+            not_modified_since_lsn: Some(not_modified_since_lsn),
+        }
+    }
+
+    /// Handle one IO request
+    async fn handle_request(&'static self, request: &'_ NeonIORequest) -> NeonIOResult {
+        self.request_counters
+            .inc(RequestTypeLabelGroup::from_req(request));
+        match request {
+            NeonIORequest::Empty => {
+                error!("unexpected Empty IO request");
+                NeonIOResult::Error(0)
+            }
+            NeonIORequest::RelSize(req) => {
+                let rel = req.reltag();
+
+                let _in_progress_guard = self
+                    .in_progress_table
+                    .lock(RequestInProgressKey::Rel(rel), req.request_id)
+                    .await;
+
+                // Check the cache first
+                let not_modified_since = match self.cache.get_rel_size(&rel) {
+                    CacheResult::Found(nblocks) => {
+                        tracing::trace!("found relsize for {:?} in cache: {}", rel, nblocks);
+                        return NeonIOResult::RelSize(nblocks);
+                    }
+                    // XXX: we don't cache negative entries, so if there's no entry in the cache, it could mean
+                    // that the relation doesn't exist or that we don't have it cached.
+                    CacheResult::NotFound(lsn) => lsn,
+                };
+
+                let read_lsn = self.request_lsns(not_modified_since);
+                match self
+                    .client
+                    .get_rel_size(page_api::GetRelSizeRequest {
+                        read_lsn,
+                        rel,
+                        allow_missing: req.allow_missing,
+                    })
+                    .await
+                {
+                    Ok(Some(nblocks)) => {
+                        // update the cache
+                        tracing::trace!(
+                            "updated relsize for {:?} in cache: {}, lsn {}",
+                            rel,
+                            nblocks,
+                            read_lsn
+                        );
+                        self.cache
+                            .remember_rel_size(&rel, nblocks, not_modified_since);
+
+                        NeonIOResult::RelSize(nblocks)
+                    }
+                    Ok(None) => {
+                        // TODO: cache negative entry?
+                        NeonIOResult::RelSize(INVALID_BLOCK_NUMBER)
+                    }
+                    Err(err) => {
+                        info!("tonic error: {err:?}");
+                        NeonIOResult::Error(0)
+                    }
+                }
+            }
+            NeonIORequest::GetPageV(req) => match self.handle_get_pagev_request(req).await {
+                Ok(()) => NeonIOResult::GetPageV,
+                Err(errno) => NeonIOResult::Error(errno),
+            },
+            NeonIORequest::GetPageVUncached(req) => {
+                match self.handle_get_pagev_uncached_request(req).await {
+                    Ok(()) => NeonIOResult::GetPageV,
+                    Err(errno) => NeonIOResult::Error(errno),
+                }
+            }
+            NeonIORequest::ReadSlruSegment(req) => {
+                let lsn = Lsn(req.request_lsn);
+                let file_path = req.destination_file_path();
+
+                match self
+                    .client
+                    .get_slru_segment(page_api::GetSlruSegmentRequest {
+                        read_lsn: self.request_lsns(lsn),
+                        kind: req.slru_kind,
+                        segno: req.segment_number,
+                    })
+                    .await
+                {
+                    Ok(slru_bytes) => {
+                        if let Err(e) = tokio::fs::write(&file_path, &slru_bytes).await {
+                            error!("could not write slru segment to file {file_path}: {e}");
+                            return NeonIOResult::Error(e.raw_os_error().unwrap_or(libc::EIO));
+                        }
+
+                        let blocks_count = slru_bytes.len() / crate::BLCKSZ;
+
+                        NeonIOResult::ReadSlruSegment(blocks_count as _)
+                    }
+                    Err(err) => {
+                        info!("tonic error: {err:?}");
+                        NeonIOResult::Error(0)
+                    }
+                }
+            }
+            NeonIORequest::PrefetchV(req) => {
+                self.request_nblocks_counters
+                    .inc_by(RequestTypeLabelGroup::from_req(request), req.nblocks as i64);
+                let req = *req;
+                tokio::spawn(async move { self.handle_prefetchv_request(&req).await });
+                NeonIOResult::PrefetchVLaunched
+            }
+            NeonIORequest::DbSize(req) => {
+                let _in_progress_guard = self
+                    .in_progress_table
+                    .lock(RequestInProgressKey::Db(req.db_oid), req.request_id)
+                    .await;
+
+                // Check the cache first
+                let not_modified_since = match self.cache.get_db_size(req.db_oid) {
+                    CacheResult::Found(db_size) => {
+                        // get_page already copied the block content to the destination
+                        return NeonIOResult::DbSize(db_size);
+                    }
+                    CacheResult::NotFound(lsn) => lsn,
+                };
+
+                match self
+                    .client
+                    .get_db_size(page_api::GetDbSizeRequest {
+                        read_lsn: self.request_lsns(not_modified_since),
+                        db_oid: req.db_oid,
+                    })
+                    .await
+                {
+                    Ok(db_size) => NeonIOResult::DbSize(db_size),
+                    Err(err) => {
+                        info!("tonic error: {err:?}");
+                        NeonIOResult::Error(0)
+                    }
+                }
+            }
+
+            // Write requests
+            NeonIORequest::WritePage(req) => {
+                let rel = req.reltag();
+                let _in_progress_guard = self
+                    .in_progress_table
+                    .lock(
+                        RequestInProgressKey::Block(rel, req.block_number),
+                        req.request_id,
+                    )
+                    .await;
+
+                // We must at least update the last-written LSN on the page, but also store the page
+                // image in the LFC while we still have it
+                self.cache
+                    .remember_page(&rel, req.block_number, req.src, Lsn(req.lsn), true)
+                    .await;
+                NeonIOResult::WriteOK
+            }
+            NeonIORequest::RelExtend(req) => {
+                let rel = req.reltag();
+                let _in_progress_guard = self
+                    .in_progress_table
+                    .lock(
+                        RequestInProgressKey::Block(rel, req.block_number),
+                        req.request_id,
+                    )
+                    .await;
+
+                // We must at least update the last-written LSN on the page and the relation size,
+                // but also store the page image in the LFC while we still have it
+                self.cache
+                    .remember_page(&rel, req.block_number, req.src, Lsn(req.lsn), true)
+                    .await;
+                self.cache
+                    .remember_rel_size(&req.reltag(), req.block_number + 1, Lsn(req.lsn));
+                NeonIOResult::WriteOK
+            }
+            NeonIORequest::RelZeroExtend(req) => {
+                self.request_nblocks_counters
+                    .inc_by(RequestTypeLabelGroup::from_req(request), req.nblocks as i64);
+
+                // TODO: need to grab an io-in-progress lock for this? I guess not
+                // TODO: We could put the empty pages to the cache. Maybe have
+                // a marker on the block entries for all-zero pages, instead of
+                // actually storing the empty pages.
+                self.cache.remember_rel_size(
+                    &req.reltag(),
+                    req.block_number + req.nblocks,
+                    Lsn(req.lsn),
+                );
+                NeonIOResult::WriteOK
+            }
+            NeonIORequest::RelCreate(req) => {
+                // TODO: need to grab an io-in-progress lock for this? I guess not
+                self.cache.remember_rel_size(&req.reltag(), 0, Lsn(req.lsn));
+                NeonIOResult::WriteOK
+            }
+            NeonIORequest::RelTruncate(req) => {
+                // TODO: need to grab an io-in-progress lock for this? I guess not
+                self.cache
+                    .remember_rel_size(&req.reltag(), req.nblocks, Lsn(req.lsn));
+                NeonIOResult::WriteOK
+            }
+            NeonIORequest::RelUnlink(req) => {
+                // TODO: need to grab an io-in-progress lock for this? I guess not
+                self.cache.forget_rel(&req.reltag(), None, Lsn(req.lsn));
+                NeonIOResult::WriteOK
+            }
+            NeonIORequest::UpdateCachedRelSize(req) => {
+                // TODO: need to grab an io-in-progress lock for this? I guess not
+                self.cache
+                    .remember_rel_size(&req.reltag(), req.nblocks, Lsn(req.lsn));
+                NeonIOResult::WriteOK
+            }
+        }
+    }
+
+    /// Subroutine to handle a GetPageV request, since it's a little more complicated than
+    /// others.
+    async fn handle_get_pagev_request(&'t self, req: &CGetPageVRequest) -> Result<(), i32> {
+        let rel = req.reltag();
+
+        // Check the cache first
+        //
+        // Note: Because the backends perform a direct lookup in the cache before sending
+        // the request to the communicator process, we expect the pages to almost never
+        // be already in cache. It could happen if:
+        // 1. two backends try to read the same page at the same time, but that should never
+        //    happen because there's higher level locking in the Postgres buffer manager, or
+        // 2. a prefetch request finished at the same time as a backend requested the
+        //    page. That's much more likely.
+        let mut cache_misses = Vec::with_capacity(req.nblocks as usize);
+        for i in 0..req.nblocks {
+            let blkno = req.block_number + i as u32;
+
+            // note: this is deadlock-safe even though we hold multiple locks at the same time,
+            // because they're always acquired in the same order.
+            let in_progress_guard = self
+                .in_progress_table
+                .lock(RequestInProgressKey::Block(rel, blkno), req.request_id)
+                .await;
+
+            let dest = req.dest[i as usize];
+            let not_modified_since = match self.cache.get_page(&rel, blkno, dest).await {
+                Ok(CacheResult::Found(_)) => {
+                    // get_page already copied the block content to the destination
+                    trace!("found blk {} in rel {:?} in LFC", blkno, rel);
+                    continue;
+                }
+                Ok(CacheResult::NotFound(lsn)) => lsn,
+                Err(_io_error) => return Err(-1), // FIXME errno?
+            };
+            cache_misses.push((blkno, not_modified_since, dest, in_progress_guard));
+        }
+        self.getpage_cache_misses_counter
+            .inc_by(cache_misses.len() as i64);
+        self.getpage_cache_hits_counter
+            .inc_by(req.nblocks as i64 - cache_misses.len() as i64);
+
+        if cache_misses.is_empty() {
+            return Ok(());
+        }
+        let not_modified_since = cache_misses
+            .iter()
+            .map(|(_blkno, lsn, _dest, _guard)| *lsn)
+            .max()
+            .unwrap();
+
+        // Construct a pageserver request for the cache misses
+        let block_numbers: Vec<u32> = cache_misses
+            .iter()
+            .map(|(blkno, _lsn, _dest, _guard)| *blkno)
+            .collect();
+        let read_lsn = self.request_lsns(not_modified_since);
+        trace!(
+            "sending getpage request for blocks {:?} in rel {:?} lsns {}",
+            block_numbers, rel, read_lsn
+        );
+        match self
+            .client
+            .get_page(page_api::GetPageRequest {
+                request_id: req.request_id.into(),
+                request_class: page_api::GetPageClass::Normal,
+                read_lsn,
+                rel,
+                block_numbers: block_numbers.clone(),
+            })
+            .await
+        {
+            Ok(resp) => {
+                // Write the received page images directly to the shared memory location
+                // that the backend requested.
+                if resp.pages.len() != block_numbers.len() {
+                    error!(
+                        "received unexpected response with {} page images from pageserver for a request for {} pages",
+                        resp.pages.len(),
+                        block_numbers.len(),
+                    );
+                    return Err(-1);
+                }
+
+                trace!(
+                    "received getpage response for blocks {:?} in rel {:?} lsns {}",
+                    block_numbers, rel, read_lsn
+                );
+
+                for (page, (blkno, _lsn, dest, _guard)) in resp.pages.into_iter().zip(cache_misses)
+                {
+                    let src: &[u8] = page.image.as_ref();
+                    let len = std::cmp::min(src.len(), dest.bytes_total());
+                    unsafe {
+                        std::ptr::copy_nonoverlapping(src.as_ptr(), dest.as_mut_ptr(), len);
+                    };
+
+                    // Also store it in the LFC while we have it
+                    self.cache
+                        .remember_page(
+                            &rel,
+                            blkno,
+                            page.image,
+                            read_lsn.not_modified_since_lsn.unwrap(),
+                            false,
+                        )
+                        .await;
+                }
+            }
+            Err(err) => {
+                info!("tonic error: {err:?}");
+                return Err(-1);
+            }
+        }
+        Ok(())
+    }
+
+    /// Subroutine to handle an GetPageVUncached request.
+    ///
+    /// Note: this bypasses the cache, in-progress IO locking, and all other side-effects.
+    /// This request type is only used in tests.
+    async fn handle_get_pagev_uncached_request(
+        &'t self,
+        req: &CGetPageVUncachedRequest,
+    ) -> Result<(), i32> {
+        let rel = req.reltag();
+
+        // Construct a pageserver request
+        let block_numbers: Vec<u32> =
+            (req.block_number..(req.block_number + (req.nblocks as u32))).collect();
+        let read_lsn = page_api::ReadLsn {
+            request_lsn: Lsn(req.request_lsn),
+            not_modified_since_lsn: Some(Lsn(req.not_modified_since)),
+        };
+        trace!(
+            "sending (uncached) getpage request for blocks {:?} in rel {:?} lsns {}",
+            block_numbers, rel, read_lsn
+        );
+        match self
+            .client
+            .get_page(page_api::GetPageRequest {
+                request_id: req.request_id.into(),
+                request_class: page_api::GetPageClass::Normal,
+                read_lsn,
+                rel,
+                block_numbers: block_numbers.clone(),
+            })
+            .await
+        {
+            Ok(resp) => {
+                // Write the received page images directly to the shared memory location
+                // that the backend requested.
+                if resp.pages.len() != block_numbers.len() {
+                    error!(
+                        "received unexpected response with {} page images from pageserver for a request for {} pages",
+                        resp.pages.len(),
+                        block_numbers.len(),
+                    );
+                    return Err(-1);
+                }
+
+                trace!(
+                    "received getpage response for blocks {:?} in rel {:?} lsns {}",
+                    block_numbers, rel, read_lsn
+                );
+
+                for (page, dest) in resp.pages.into_iter().zip(req.dest) {
+                    let src: &[u8] = page.image.as_ref();
+                    let len = std::cmp::min(src.len(), dest.bytes_total());
+                    unsafe {
+                        std::ptr::copy_nonoverlapping(src.as_ptr(), dest.as_mut_ptr(), len);
+                    };
+                }
+            }
+            Err(err) => {
+                info!("tonic error: {err:?}");
+                return Err(-1);
+            }
+        }
+        Ok(())
+    }
+
+    /// Subroutine to handle a PrefetchV request, since it's a little more complicated than
+    /// others.
+    ///
+    /// This is very similar to a GetPageV request, but the results are only stored in the cache.
+    async fn handle_prefetchv_request(&'static self, req: &CPrefetchVRequest) -> Result<(), i32> {
+        let rel = req.reltag();
+
+        // Check the cache first
+        let mut cache_misses = Vec::with_capacity(req.nblocks as usize);
+        for i in 0..req.nblocks {
+            let blkno = req.block_number + i as u32;
+
+            // note: this is deadlock-safe even though we hold multiple locks at the same time,
+            // because they're always acquired in the same order.
+            let in_progress_guard = self
+                .in_progress_table
+                .lock(RequestInProgressKey::Block(rel, blkno), req.request_id)
+                .await;
+
+            let not_modified_since = match self.cache.page_is_cached(&rel, blkno).await {
+                Ok(CacheResult::Found(_)) => {
+                    trace!("found blk {} in rel {:?} in LFC", blkno, rel);
+                    continue;
+                }
+                Ok(CacheResult::NotFound(lsn)) => lsn,
+                Err(_io_error) => return Err(-1), // FIXME errno?
+            };
+            cache_misses.push((blkno, not_modified_since, in_progress_guard));
+        }
+        if cache_misses.is_empty() {
+            return Ok(());
+        }
+        let not_modified_since = cache_misses
+            .iter()
+            .map(|(_blkno, lsn, _guard)| *lsn)
+            .max()
+            .unwrap();
+        let block_numbers: Vec<u32> = cache_misses
+            .iter()
+            .map(|(blkno, _lsn, _guard)| *blkno)
+            .collect();
+
+        // TODO: spawn separate tasks for these. Use the integrated cache to keep track of the
+        // in-flight requests
+
+        match self
+            .client
+            .get_page(page_api::GetPageRequest {
+                request_id: req.request_id.into(),
+                request_class: page_api::GetPageClass::Prefetch,
+                read_lsn: self.request_lsns(not_modified_since),
+                rel,
+                block_numbers: block_numbers.clone(),
+            })
+            .await
+        {
+            Ok(resp) => {
+                trace!(
+                    "prefetch completed, remembering blocks {:?} in rel {:?} in LFC",
+                    block_numbers, rel
+                );
+                if resp.pages.len() != block_numbers.len() {
+                    error!(
+                        "received unexpected response with {} page images from pageserver for a request for {} pages",
+                        resp.pages.len(),
+                        block_numbers.len(),
+                    );
+                    return Err(-1);
+                }
+
+                for (page, (blkno, _lsn, _guard)) in resp.pages.into_iter().zip(cache_misses) {
+                    self.cache
+                        .remember_page(&rel, blkno, page.image, not_modified_since, false)
+                        .await;
+                }
+            }
+            Err(err) => {
+                info!("tonic error: {err:?}");
+                return Err(-1);
+            }
+        }
+        Ok(())
+    }
+}
+
+impl<T> MetricGroup<T> for CommunicatorWorkerProcessStruct<'_>
 where
     T: Encoding,
     GaugeState: MetricEncoding<T>,
 {
     fn collect_group_into(&self, enc: &mut T) -> Result<(), T::Err> {
-        self.lfc_metrics.collect_group_into(enc)
+        use measured::metric::MetricFamilyEncoding;
+        use measured::metric::name::MetricName;
+
+        self.lfc_metrics.collect_group_into(enc)?;
+        self.request_counters
+            .collect_family_into(MetricName::from_str("request_counters"), enc)?;
+        self.request_nblocks_counters
+            .collect_family_into(MetricName::from_str("request_nblocks_counters"), enc)?;
+
+        // FIXME: allocator metrics
+
+        Ok(())
     }
 }

pgxn/neon/communicator/src/worker_process/mod.rs

@@ -4,9 +4,9 @@
 //! - launch the main loop,
 //! - receive IO requests from backends and process them,
 //! - write results back to backends.
-
 mod callbacks;
 mod control_socket;
+mod in_progress_ios;
 mod lfc_metrics;
 mod logging;
 mod main_loop;

pgxn/neon/communicator/src/worker_process/worker_interface.rs

@@ -1,14 +1,21 @@
 //! Functions called from the C code in the worker process
 
+use std::collections::HashMap;
 use std::ffi::{CStr, CString, c_char};
+use std::path::PathBuf;
 
+use crate::init::CommunicatorInitStruct;
 use crate::worker_process::main_loop;
 use crate::worker_process::main_loop::CommunicatorWorkerProcessStruct;
 
+use pageserver_client_grpc::ShardStripeSize;
+
 /// Launch the communicator's tokio tasks, which do most of the work.
 ///
 /// The caller has initialized the process as a regular PostgreSQL background worker
-/// process.
+/// process. The shared memory segment used to communicate with the backends has been
+/// allocated and initialized earlier, at postmaster startup, in
+/// rcommunicator_shmem_init().
 ///
 /// Inputs:
 ///   `tenant_id` and `timeline_id` can be NULL, if we're been launched in "non-Neon" mode,
@@ -23,27 +30,63 @@ use crate::worker_process::main_loop::CommunicatorWorkerProcessStruct;
 /// This is called only once in the process, so the returned struct, and error message in
 /// case of failure, are simply leaked.
 #[unsafe(no_mangle)]
-pub extern "C" fn communicator_worker_launch(
+pub extern "C" fn communicator_worker_process_launch(
+    cis: Box<CommunicatorInitStruct>,
     tenant_id: *const c_char,
     timeline_id: *const c_char,
+    auth_token: *const c_char,
+    shard_map: *mut *mut c_char,
+    nshards: u32,
+    stripe_size: u32,
+    file_cache_path: *const c_char,
+    initial_file_cache_size: u64,
     error_p: *mut *const c_char,
-) -> Option<&'static CommunicatorWorkerProcessStruct> {
+) -> Option<&'static CommunicatorWorkerProcessStruct<'static>> {
+    tracing::warn!("starting threads in rust code");
     // Convert the arguments into more convenient Rust types
-    let tenant_id = if tenant_id.is_null() {
+    let tenant_id = {
+        let cstr = unsafe { CStr::from_ptr(tenant_id) };
+        cstr.to_str().expect("assume UTF-8")
+    };
+    let timeline_id = {
+        let cstr = unsafe { CStr::from_ptr(timeline_id) };
+        cstr.to_str().expect("assume UTF-8")
+    };
+    let auth_token = if auth_token.is_null() {
         None
     } else {
-        let cstr = unsafe { CStr::from_ptr(tenant_id) };
+        let cstr = unsafe { CStr::from_ptr(auth_token) };
         Some(cstr.to_str().expect("assume UTF-8"))
     };
-    let timeline_id = if timeline_id.is_null() {
-        None
+    let file_cache_path = {
+        if file_cache_path.is_null() {
+            None
+        } else {
+            let c_str = unsafe { CStr::from_ptr(file_cache_path) };
+            Some(PathBuf::from(c_str.to_str().unwrap()))
+        }
+    };
+    let shard_map = shard_map_to_hash(nshards, shard_map);
+    // FIXME: distinguish between unsharded, and sharded with 1 shard
+    // Also, we might go from unsharded to sharded while the system
+    // is running.
+    let stripe_size = if stripe_size > 0 && nshards > 1 {
+        Some(ShardStripeSize(stripe_size))
     } else {
-        let cstr = unsafe { CStr::from_ptr(timeline_id) };
-        Some(cstr.to_str().expect("assume UTF-8"))
+        None
     };
 
     // The `init` function does all the work.
-    let result = main_loop::init(tenant_id, timeline_id);
+    let result = main_loop::init(
+        *cis,
+        tenant_id,
+        timeline_id,
+        auth_token,
+        shard_map,
+        stripe_size,
+        initial_file_cache_size,
+        file_cache_path,
+    );
 
     // On failure, return the error message to the C caller in *error_p.
     match result {
@@ -58,3 +101,66 @@ pub extern "C" fn communicator_worker_launch(
         }
     }
 }
+
+#[unsafe(no_mangle)]
+pub extern "C" fn communicator_worker_process_launch_legacy(error_p: *mut *const c_char) -> bool {
+    // The `init` function does all the work.
+    let result = main_loop::init_legacy();
+
+    // On failure, return the error message to the C caller in *error_p.
+    match result {
+        Ok(()) => true,
+        Err(errmsg) => {
+            let errmsg = CString::new(errmsg).expect("no nuls within error message");
+            let errmsg = Box::leak(errmsg.into_boxed_c_str());
+            let p: *const c_char = errmsg.as_ptr();
+
+            unsafe { *error_p = p };
+            false
+        }
+    }
+}
+
+/// Convert the "shard map" from an array of C strings, indexed by shard no to a rust HashMap
+fn shard_map_to_hash(
+    nshards: u32,
+    shard_map: *mut *mut c_char,
+) -> HashMap<utils::shard::ShardIndex, String> {
+    use utils::shard::*;
+
+    assert!(nshards <= u8::MAX as u32);
+
+    let mut result: HashMap<ShardIndex, String> = HashMap::new();
+    let mut p = shard_map;
+
+    for i in 0..nshards {
+        let c_str = unsafe { CStr::from_ptr(*p) };
+
+        p = unsafe { p.add(1) };
+
+        let s = c_str.to_str().unwrap();
+        let k = if nshards > 1 {
+            ShardIndex::new(ShardNumber(i as u8), ShardCount(nshards as u8))
+        } else {
+            ShardIndex::unsharded()
+        };
+        result.insert(k, s.into());
+    }
+    result
+}
+
+/// Inform the rust code about a configuration change
+#[unsafe(no_mangle)]
+pub extern "C" fn communicator_worker_config_reload(
+    proc_handle: &'static CommunicatorWorkerProcessStruct<'static>,
+    file_cache_size: u64,
+    shard_map: *mut *mut c_char,
+    nshards: u32,
+    stripe_size: u32,
+) {
+    proc_handle.cache.resize_file_cache(file_cache_size as u32);
+
+    let shard_map = shard_map_to_hash(nshards, shard_map);
+    let stripe_size = (nshards > 1).then_some(ShardStripeSize(stripe_size));
+    proc_handle.update_shard_map(shard_map, stripe_size);
+}

pgxn/neon/communicator_new.h

@@ -0,0 +1,69 @@
+/*-------------------------------------------------------------------------
+ *
+ * communicator_new.h
+ *	  new implementation
+ *
+ *
+ * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef COMMUNICATOR_NEW_H
+#define COMMUNICATOR_NEW_H
+
+#include "storage/buf_internals.h"
+
+#include "lfc_prewarm.h"
+#include "neon.h"
+#include "neon_pgversioncompat.h"
+#include "pagestore_client.h"
+
+/* initialization at postmaster startup */
+extern void CommunicatorNewShmemRequest(void);
+extern void CommunicatorNewShmemInit(void);
+
+/* initialization at backend startup */
+extern void communicator_new_init(void);
+
+/* Read requests */
+extern bool communicator_new_rel_exists(NRelFileInfo rinfo, ForkNumber forkNum);
+extern BlockNumber communicator_new_rel_nblocks(NRelFileInfo rinfo, ForkNumber forknum);
+extern int64 communicator_new_dbsize(Oid dbNode);
+extern void communicator_new_readv(NRelFileInfo rinfo, ForkNumber forkNum,
+								   BlockNumber base_blockno,
+								   void **buffers, BlockNumber nblocks);
+extern void communicator_new_read_at_lsn_uncached(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blockno,
+												  void *buffer, XLogRecPtr request_lsn, XLogRecPtr not_modified_since);
+extern void communicator_new_prefetch_register_bufferv(NRelFileInfo rinfo, ForkNumber forkNum,
+													   BlockNumber blockno,
+													   BlockNumber nblocks);
+extern bool communicator_new_update_lwlsn_for_block_if_not_cached(NRelFileInfo rinfo, ForkNumber forkNum,
+																  BlockNumber blockno, XLogRecPtr lsn);
+extern int	communicator_new_read_slru_segment(
+											   SlruKind kind,
+											   uint32_t segno,
+											   neon_request_lsns * request_lsns,
+											   const char *path
+);
+
+/* Write requests, to keep the caches up-to-date */
+extern void communicator_new_write_page(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blockno,
+										const void *buffer, XLogRecPtr lsn);
+extern void communicator_new_rel_extend(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blockno,
+										const void *buffer, XLogRecPtr lsn);
+extern void communicator_new_rel_zeroextend(NRelFileInfo rinfo, ForkNumber forkNum,
+											BlockNumber blockno, BlockNumber nblocks,
+											XLogRecPtr lsn);
+extern void communicator_new_rel_create(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr lsn);
+extern void communicator_new_rel_truncate(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn);
+extern void communicator_new_rel_unlink(NRelFileInfo rinfo, ForkNumber forkNum, XLogRecPtr lsn);
+extern void communicator_new_update_cached_rel_size(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks, XLogRecPtr lsn);
+
+/* other functions */
+extern int32 communicator_new_approximate_working_set_size_seconds(time_t duration, bool reset);
+extern struct LfcMetrics communicator_new_get_lfc_metrics_unsafe(void);
+extern FileCacheState *communicator_new_get_lfc_state(size_t max_entries);
+extern struct LfcStatsEntry *communicator_new_lfc_get_stats(size_t *num_entries);
+
+#endif							/* COMMUNICATOR_NEW_H */

pgxn/neon/communicator_process.c

@@ -18,6 +18,9 @@
 #include <unistd.h>
 
 #include "miscadmin.h"
+#if PG_VERSION_NUM >= 150000
+#include "access/xlogrecovery.h"
+#endif
 #include "postmaster/bgworker.h"
 #include "postmaster/interrupt.h"
 #include "postmaster/postmaster.h"
@@ -29,14 +32,18 @@
 #include "tcop/tcopprot.h"
 #include "utils/timestamp.h"
 
+#include "communicator_new.h"
 #include "communicator_process.h"
 #include "file_cache.h"
 #include "neon.h"
 #include "neon_perf_counters.h"
+#include "pagestore_client.h"
 
 /* the rust bindings, generated by cbindgen */
 #include "communicator/communicator_bindings.h"
 
+struct CommunicatorInitStruct *cis;
+
 static void pump_logging(struct LoggingReceiver *logging);
 PGDLLEXPORT void communicator_new_bgworker_main(Datum main_arg);
 
@@ -70,9 +77,13 @@ pg_init_communicator_process(void)
 void
 communicator_new_bgworker_main(Datum main_arg)
 {
+	char	  **connstrings;
+	ShardMap	shard_map;
+	uint64		file_cache_size;
 	struct LoggingReceiver *logging;
 	const char *errmsg = NULL;
 	const struct CommunicatorWorkerProcessStruct *proc_handle;
+	bool		success;
 
 	/*
 	 * Pretend that this process is a WAL sender. That affects the shutdown
@@ -108,12 +119,42 @@ communicator_new_bgworker_main(Datum main_arg)
 
 	logging = communicator_worker_configure_logging();
 
-	proc_handle = communicator_worker_launch(
-		neon_tenant[0] == '\0' ? NULL : neon_tenant,
-		neon_timeline[0] == '\0' ? NULL : neon_timeline,
-		&errmsg
-		);
-	if (proc_handle == NULL)
+	if (cis != NULL)
+	{
+		/* lfc_size_limit is in MBs */
+		file_cache_size = lfc_size_limit * (1024 * 1024 / BLCKSZ);
+		if (file_cache_size < 100)
+			file_cache_size = 100;
+
+		if (!parse_shard_map(pageserver_grpc_urls, &shard_map))
+		{
+			/* shouldn't happen, as the GUC was verified already */
+			elog(FATAL, "could not parse neon.pageserver_grpcs_urls");
+		}
+		connstrings = palloc(shard_map.num_shards * sizeof(char *));
+		for (int i = 0; i < shard_map.num_shards; i++)
+			connstrings[i] = shard_map.connstring[i];
+		proc_handle = communicator_worker_process_launch(
+			cis,
+			neon_tenant,
+			neon_timeline,
+			neon_auth_token,
+			connstrings,
+			shard_map.num_shards,
+			neon_stripe_size,
+			lfc_path,
+			file_cache_size,
+			&errmsg);
+		pfree(connstrings);
+		cis = NULL;
+		success = proc_handle != NULL;
+	}
+	else
+	{
+		proc_handle = NULL;
+		success = communicator_worker_process_launch_legacy(&errmsg);
+	}
+	if (!success)
 	{
 		/*
 		 * Something went wrong. Before exiting, forward any log messages that
@@ -173,6 +214,31 @@ communicator_new_bgworker_main(Datum main_arg)
 		{
 			ConfigReloadPending = false;
 			ProcessConfigFile(PGC_SIGHUP);
+
+			if (proc_handle)
+			{
+				/* lfc_size_limit is in MBs */
+				file_cache_size = lfc_size_limit * (1024 * 1024 / BLCKSZ);
+				if (file_cache_size < 100)
+					file_cache_size = 100;
+
+				/* Reload pageserver URLs */
+				if (!parse_shard_map(pageserver_grpc_urls, &shard_map))
+				{
+					/* shouldn't happen, as the GUC was verified already */
+					elog(FATAL, "could not parse neon.pageserver_grpcs_urls");
+				}
+				connstrings = palloc(shard_map.num_shards * sizeof(char *));
+				for (int i = 0; i < shard_map.num_shards; i++)
+					connstrings[i] = shard_map.connstring[i];
+
+				communicator_worker_config_reload(proc_handle,
+												  file_cache_size,
+												  connstrings,
+												  shard_map.num_shards,
+												  neon_stripe_size);
+				pfree(connstrings);
+			}
 		}
 
 		duration = TimestampDifferenceMilliseconds(before, GetCurrentTimestamp());
@@ -271,3 +337,49 @@ callback_set_my_latch_unsafe(void)
 {
 	SetLatch(MyLatch);
 }
+
+/*
+ * FIXME: The logic from neon_get_request_lsns() needs to go here, except for
+ * the last-written LSN cache stuff, which is managed by the rust code now.
+ */
+XLogRecPtr
+callback_get_request_lsn_unsafe(void)
+{
+	/*
+	 * NB: be very careful with what you do here! This is called from tokio
+	 * threads, so anything tha tries to take LWLocks is unsafe, for example.
+	 *
+	 * RecoveryInProgress() is OK
+	 */
+	if (RecoveryInProgress())
+	{
+		XLogRecPtr	replay_lsn = GetXLogReplayRecPtr(NULL);
+
+		return replay_lsn;
+	}
+	else
+	{
+		XLogRecPtr	flushlsn;
+
+#if PG_VERSION_NUM >= 150000
+		flushlsn = GetFlushRecPtr(NULL);
+#else
+		flushlsn = GetFlushRecPtr();
+#endif
+
+		return flushlsn;
+	}
+}
+
+/*
+ * Get metrics, for the built-in metrics exporter that's part of the
+ * communicator process.
+ */
+struct LfcMetrics
+callback_get_lfc_metrics_unsafe(void)
+{
+	if (neon_use_communicator_worker)
+		return communicator_new_get_lfc_metrics_unsafe();
+	else
+		return lfc_get_metrics_unsafe();
+}

pgxn/neon/communicator_process.h

@@ -12,6 +12,9 @@
 #ifndef COMMUNICATOR_PROCESS_H
 #define COMMUNICATOR_PROCESS_H
 
+extern struct CommunicatorInitStruct *cis;
+
+/* initialization early at postmaster startup */
 extern void pg_init_communicator_process(void);
 
 #endif			/* COMMUNICATOR_PROCESS_H */

pgxn/neon/file_cache.c

@@ -49,6 +49,7 @@
 #include "neon.h"
 #include "neon_lwlsncache.h"
 #include "neon_perf_counters.h"
+#include "neon_utils.h"
 #include "pagestore_client.h"
 #include "communicator.h"
 
@@ -136,15 +137,6 @@ typedef struct FileCacheEntry
 #define N_COND_VARS 	64
 #define CV_WAIT_TIMEOUT	10
 
-#define MAX_PREWARM_WORKERS 8
-
-typedef struct PrewarmWorkerState
-{
-	uint32		prewarmed_pages;
-	uint32		skipped_pages;
-	TimestampTz completed;
-} PrewarmWorkerState;
-
 typedef struct FileCacheControl
 {
 	uint64		generation;		/* generation is needed to handle correct hash
@@ -190,47 +182,27 @@ typedef struct FileCacheControl
 	 *   again.
 	 */
 	HyperLogLogState wss_estimation;
-
-	/* Prewarmer state */
-	PrewarmWorkerState prewarm_workers[MAX_PREWARM_WORKERS];
-	size_t n_prewarm_workers;
-	size_t n_prewarm_entries;
-	size_t total_prewarm_pages;
-	size_t prewarm_batch;
-	bool   prewarm_active;
-	bool   prewarm_canceled;
-	dsm_handle prewarm_lfc_state_handle;
 } FileCacheControl;
 
-#define FILE_CACHE_STATE_MAGIC 0xfcfcfcfc
-
-#define FILE_CACHE_STATE_BITMAP(fcs)	((uint8*)&(fcs)->chunks[(fcs)->n_chunks])
-#define FILE_CACHE_STATE_SIZE_FOR_CHUNKS(n_chunks)	(sizeof(FileCacheState) + (n_chunks)*sizeof(BufferTag) + (((n_chunks) * lfc_blocks_per_chunk)+7)/8)
-#define FILE_CACHE_STATE_SIZE(fcs)		(sizeof(FileCacheState) + (fcs->n_chunks)*sizeof(BufferTag) + (((fcs->n_chunks) << fcs->chunk_size_log)+7)/8)
-
 static HTAB *lfc_hash;
 static int	lfc_desc = -1;
 static LWLockId lfc_lock;
-static int	lfc_max_size;
-static int	lfc_size_limit;
-static int	lfc_prewarm_limit;
-static int	lfc_prewarm_batch;
+int	lfc_max_size;
+int	lfc_size_limit;
 static int	lfc_chunk_size_log = MAX_BLOCKS_PER_CHUNK_LOG;
 static int	lfc_blocks_per_chunk = MAX_BLOCKS_PER_CHUNK;
-static char *lfc_path;
+char *lfc_path;
 static uint64 lfc_generation;
 static FileCacheControl *lfc_ctl;
-static bool lfc_do_prewarm;
 
 bool lfc_store_prefetch_result;
 bool lfc_prewarm_update_ws_estimation;
 
-bool AmPrewarmWorker;
+bool lfc_do_prewarm;
+bool lfc_prewarm_cancel;
 
 #define LFC_ENABLED() (lfc_ctl->limit != 0)
 
-PGDLLEXPORT void lfc_prewarm_main(Datum main_arg);
-
 /*
  * Close LFC file if opened.
  * All backends should close their LFC files once LFC is disabled.
@@ -256,6 +228,8 @@ lfc_switch_off(void)
 {
 	int			fd;
 
+	Assert(!neon_use_communicator_worker);
+
 	if (LFC_ENABLED())
 	{
 		HASH_SEQ_STATUS status;
@@ -321,6 +295,8 @@ lfc_maybe_disabled(void)
 static bool
 lfc_ensure_opened(void)
 {
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_generation != lfc_ctl->generation)
 	{
 		lfc_close_file();
@@ -346,6 +322,9 @@ LfcShmemInit(void)
 	bool		found;
 	static HASHCTL info;
 
+	if (neon_use_communicator_worker)
+		return;
+
 	if (lfc_max_size <= 0)
 		return;
 
@@ -535,7 +514,6 @@ lfc_init(void)
 	if (!process_shared_preload_libraries_in_progress)
 		neon_log(ERROR, "Neon module should be loaded via shared_preload_libraries");
 
-
 	DefineCustomBoolVariable("neon.store_prefetch_result_in_lfc",
 							"Immediately store received prefetch result in LFC",
 							NULL,
@@ -607,32 +585,6 @@ lfc_init(void)
 							lfc_check_chunk_size,
 							lfc_change_chunk_size,
 							NULL);
-
-	DefineCustomIntVariable("neon.file_cache_prewarm_limit",
-							"Maximal number of prewarmed chunks",
-							NULL,
-							&lfc_prewarm_limit,
-							INT_MAX,	/* no limit by default */
-							0,
-							INT_MAX,
-							PGC_SIGHUP,
-							0,
-							NULL,
-							NULL,
-							NULL);
-
-	DefineCustomIntVariable("neon.file_cache_prewarm_batch",
-							"Number of pages retrivied by prewarm from page server",
-							NULL,
-							&lfc_prewarm_batch,
-							64,
-							1,
-							INT_MAX,
-							PGC_SIGHUP,
-							0,
-							NULL,
-							NULL,
-							NULL);
 }
 
 /*
@@ -657,7 +609,7 @@ lfc_get_state(size_t max_entries)
 		uint8* bitmap;
 		size_t n_pages = 0;
 		size_t n_entries = Min(max_entries, lfc_ctl->used - lfc_ctl->pinned);
-		size_t state_size = FILE_CACHE_STATE_SIZE_FOR_CHUNKS(n_entries);
+		size_t state_size = FILE_CACHE_STATE_SIZE_FOR_CHUNKS(n_entries, lfc_blocks_per_chunk);
 		fcs = (FileCacheState*)palloc0(state_size);
 		SET_VARSIZE(fcs, state_size);
 		fcs->magic = FILE_CACHE_STATE_MAGIC;
@@ -673,8 +625,19 @@ lfc_get_state(size_t max_entries)
 			{
 				if (GET_STATE(entry, j) != UNAVAILABLE)
 				{
-					BITMAP_SET(bitmap, i*lfc_blocks_per_chunk + j);
-					n_pages += 1;
+					/* Validate the buffer tag before including it */
+					BufferTag test_tag = entry->key;
+					test_tag.blockNum += j;
+
+					if (BufferTagIsValid(&test_tag))
+					{
+						BITMAP_SET(bitmap, i*lfc_blocks_per_chunk + j);
+						n_pages += 1;
+					}
+					else
+					{
+						elog(ERROR, "LFC: Skipping invalid buffer tag during cache state capture: blockNum=%u", test_tag.blockNum);
+					}
 				}
 			}
 			if (++i == n_entries)
@@ -683,7 +646,7 @@ lfc_get_state(size_t max_entries)
 		Assert(i == n_entries);
 		fcs->n_pages = n_pages;
 		Assert(pg_popcount((char*)bitmap, ((n_entries << lfc_chunk_size_log) + 7)/8) == n_pages);
-		elog(LOG, "LFC: save state of %d chunks %d pages", (int)n_entries, (int)n_pages);
+		elog(LOG, "LFC: save state of %d chunks %d pages (validated)", (int)n_entries, (int)n_pages);
 	}
 
 	LWLockRelease(lfc_lock);
@@ -691,267 +654,6 @@ lfc_get_state(size_t max_entries)
 	return fcs;
 }
 
-/*
- * Prewarm LFC cache to the specified state. It uses lfc_prefetch function to load prewarmed page without hoilding shared buffer lock
- * and avoid race conditions with other backends.
- */
-void
-lfc_prewarm(FileCacheState* fcs, uint32 n_workers)
-{
-	size_t fcs_chunk_size_log;
-	size_t n_entries;
-	size_t prewarm_batch = Min(lfc_prewarm_batch, readahead_buffer_size);
-	size_t fcs_size;
-	dsm_segment *seg;
-	BackgroundWorkerHandle* bgw_handle[MAX_PREWARM_WORKERS];
-
-
-	if (!lfc_ensure_opened())
-		return;
-
-	if (prewarm_batch == 0 || lfc_prewarm_limit == 0 || n_workers == 0)
-	{
-		elog(LOG, "LFC: prewarm is disabled");
-		return;
-	}
-
-	if (n_workers > MAX_PREWARM_WORKERS)
-	{
-		elog(ERROR, "LFC: Too much prewarm workers, maximum is %d", MAX_PREWARM_WORKERS);
-	}
-
-	if (fcs == NULL || fcs->n_chunks == 0)
-	{
-		elog(LOG, "LFC: nothing to prewarm");
-		return;
-	}
-
-	if (fcs->magic != FILE_CACHE_STATE_MAGIC)
-	{
-		elog(ERROR, "LFC: Invalid file cache state magic: %X", fcs->magic);
-	}
-
-	fcs_size = VARSIZE(fcs);
-	if (FILE_CACHE_STATE_SIZE(fcs) != fcs_size)
-	{
-		elog(ERROR, "LFC: Invalid file cache state size: %u vs. %u", (unsigned)FILE_CACHE_STATE_SIZE(fcs), VARSIZE(fcs));
-	}
-
-	fcs_chunk_size_log = fcs->chunk_size_log;
-	if (fcs_chunk_size_log > MAX_BLOCKS_PER_CHUNK_LOG)
-	{
-		elog(ERROR, "LFC: Invalid chunk size log: %u", fcs->chunk_size_log);
-	}
-
-	n_entries = Min(fcs->n_chunks, lfc_prewarm_limit);
-	Assert(n_entries != 0);
-
-	LWLockAcquire(lfc_lock, LW_EXCLUSIVE);
-
-	/* Do not prewarm more entries than LFC limit */
-	if (lfc_ctl->limit <= lfc_ctl->size)
-	{
-		elog(LOG, "LFC: skip prewarm because LFC is already filled");
-		LWLockRelease(lfc_lock);
-		return;
-	}
-
-	if (lfc_ctl->prewarm_active)
-	{
-		LWLockRelease(lfc_lock);
-		elog(ERROR, "LFC: skip prewarm because another prewarm is still active");
-	}
-	lfc_ctl->n_prewarm_entries = n_entries;
-	lfc_ctl->n_prewarm_workers = n_workers;
-	lfc_ctl->prewarm_active = true;
-	lfc_ctl->prewarm_canceled = false;
-	lfc_ctl->prewarm_batch = prewarm_batch;
-	memset(lfc_ctl->prewarm_workers, 0, n_workers*sizeof(PrewarmWorkerState));
-
-	LWLockRelease(lfc_lock);
-
-	/* Calculate total number of pages to be prewarmed */
-	lfc_ctl->total_prewarm_pages = fcs->n_pages;
-
-	seg = dsm_create(fcs_size, 0);
-	memcpy(dsm_segment_address(seg), fcs, fcs_size);
-	lfc_ctl->prewarm_lfc_state_handle = dsm_segment_handle(seg);
-
-	/* Spawn background workers */
-	for (uint32 i = 0; i < n_workers; i++)
-	{
-		BackgroundWorker worker = {0};
-
-		worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
-		worker.bgw_start_time = BgWorkerStart_ConsistentState;
-		worker.bgw_restart_time = BGW_NEVER_RESTART;
-		strcpy(worker.bgw_library_name, "neon");
-		strcpy(worker.bgw_function_name, "lfc_prewarm_main");
-		snprintf(worker.bgw_name, BGW_MAXLEN, "LFC prewarm worker %d", i+1);
-		strcpy(worker.bgw_type, "LFC prewarm worker");
-		worker.bgw_main_arg = Int32GetDatum(i);
-		/* must set notify PID to wait for shutdown */
-		worker.bgw_notify_pid = MyProcPid;
-
-		if (!RegisterDynamicBackgroundWorker(&worker, &bgw_handle[i]))
-		{
-			ereport(LOG,
-					(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
-					 errmsg("LFC: registering dynamic bgworker prewarm failed"),
-					 errhint("Consider increasing the configuration parameter \"%s\".", "max_worker_processes")));
-			n_workers = i;
-			lfc_ctl->prewarm_canceled = true;
-			break;
-		}
-	}
-
-	for (uint32 i = 0; i < n_workers; i++)
-	{
-		bool interrupted;
-		do
-		{
-			interrupted = false;
-			PG_TRY();
-			{
-				BgwHandleStatus status = WaitForBackgroundWorkerShutdown(bgw_handle[i]);
-				if (status != BGWH_STOPPED && status != BGWH_POSTMASTER_DIED)
-				{
-					elog(LOG, "LFC: Unexpected status of prewarm worker termination: %d", status);
-				}
-			}
-			PG_CATCH();
-			{
-				elog(LOG, "LFC: cancel prewarm");
-				lfc_ctl->prewarm_canceled = true;
-				interrupted = true;
-			}
-			PG_END_TRY();
-		} while (interrupted);
-
-		if (!lfc_ctl->prewarm_workers[i].completed)
-		{
-			/* Background worker doesn't set completion time: it means that it was abnormally terminated */
-			elog(LOG, "LFC: prewarm worker %d failed", i+1);
-			/* Set completion time to prevent get_prewarm_info from considering this worker as active */
-			lfc_ctl->prewarm_workers[i].completed = GetCurrentTimestamp();
-		}
-	}
-	dsm_detach(seg);
-
-	LWLockAcquire(lfc_lock, LW_EXCLUSIVE);
-	lfc_ctl->prewarm_active = false;
-	LWLockRelease(lfc_lock);
-}
-
-void
-lfc_prewarm_main(Datum main_arg)
-{
-	size_t snd_idx = 0, rcv_idx = 0;
-	size_t n_sent = 0, n_received = 0;
-	size_t fcs_chunk_size_log;
-	size_t max_prefetch_pages;
-	size_t prewarm_batch;
-	size_t n_workers;
-	dsm_segment *seg;
-	FileCacheState* fcs;
-	uint8* bitmap;
-	BufferTag tag;
-	PrewarmWorkerState* ws;
-	uint32 worker_id = DatumGetInt32(main_arg);
-
-	AmPrewarmWorker = true;
-
-	pqsignal(SIGTERM, die);
-	BackgroundWorkerUnblockSignals();
-
-	seg = dsm_attach(lfc_ctl->prewarm_lfc_state_handle);
-	if (seg == NULL)
-		ereport(ERROR,
-				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
-				 errmsg("could not map dynamic shared memory segment")));
-
-	fcs = (FileCacheState*) dsm_segment_address(seg);
-	prewarm_batch = lfc_ctl->prewarm_batch;
-	fcs_chunk_size_log = fcs->chunk_size_log;
-	n_workers = lfc_ctl->n_prewarm_workers;
-	max_prefetch_pages = lfc_ctl->n_prewarm_entries << fcs_chunk_size_log;
-	ws = &lfc_ctl->prewarm_workers[worker_id];
-	bitmap = FILE_CACHE_STATE_BITMAP(fcs);
-
-	/* enable prefetch in LFC */
-	lfc_store_prefetch_result = true;
-	lfc_do_prewarm = true; /* Flag for lfc_prefetch preventing replacement of existed entries if LFC cache is full */
-
-	elog(LOG, "LFC: worker %d start prewarming", worker_id);
-	while (!lfc_ctl->prewarm_canceled)
-	{
-		if (snd_idx < max_prefetch_pages)
-		{
-			if ((snd_idx >> fcs_chunk_size_log) % n_workers != worker_id)
-			{
-				/* If there are multiple workers, split chunks between them */
-				snd_idx += 1 << fcs_chunk_size_log;
-			}
-			else
-			{
-				if (BITMAP_ISSET(bitmap, snd_idx))
-				{
-					tag = fcs->chunks[snd_idx >> fcs_chunk_size_log];
-					tag.blockNum += snd_idx & ((1 << fcs_chunk_size_log) - 1);
-					if (!lfc_cache_contains(BufTagGetNRelFileInfo(tag), tag.forkNum, tag.blockNum))
-					{
-						(void)communicator_prefetch_register_bufferv(tag, NULL, 1, NULL);
-						n_sent += 1;
-					}
-					else
-					{
-						ws->skipped_pages += 1;
-						BITMAP_CLR(bitmap, snd_idx);
-					}
-				}
-				snd_idx += 1;
-			}
-		}
-		if (n_sent >= n_received + prewarm_batch || snd_idx == max_prefetch_pages)
-		{
-			if (n_received == n_sent && snd_idx == max_prefetch_pages)
-			{
-				break;
-			}
-			if ((rcv_idx >> fcs_chunk_size_log) % n_workers != worker_id)
-			{
-				/* Skip chunks processed by other workers */
-				rcv_idx += 1 << fcs_chunk_size_log;
-				continue;
-			}
-
-			/* Locate next block to prefetch */
-			while (!BITMAP_ISSET(bitmap, rcv_idx))
-			{
-				rcv_idx += 1;
-			}
-			tag = fcs->chunks[rcv_idx >> fcs_chunk_size_log];
-			tag.blockNum += rcv_idx & ((1 << fcs_chunk_size_log) - 1);
-			if (communicator_prefetch_receive(tag))
-			{
-				ws->prewarmed_pages += 1;
-			}
-			else
-			{
-				ws->skipped_pages += 1;
-			}
-			rcv_idx += 1;
-			n_received += 1;
-		}
-	}
-	/* No need to perform prefetch cleanup here because prewarm worker will be terminated and
-	 * connection to PS dropped just after return from this function.
-	 */
-	Assert(n_sent == n_received || lfc_ctl->prewarm_canceled);
-	elog(LOG, "LFC: worker %d complete prewarming: loaded %ld pages", worker_id, (long)n_received);
-	lfc_ctl->prewarm_workers[worker_id].completed = GetCurrentTimestamp();
-}
-
 void
 lfc_invalidate(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks)
 {
@@ -959,6 +661,8 @@ lfc_invalidate(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks)
 	FileCacheEntry *entry;
 	uint32		hash;
 
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_maybe_disabled())	/* fast exit if file cache is disabled */
 		return;
 
@@ -1004,6 +708,8 @@ lfc_cache_contains(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno)
 	bool		found = false;
 	uint32		hash;
 
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_maybe_disabled())	/* fast exit if file cache is disabled */
 		return false;
 
@@ -1039,6 +745,8 @@ lfc_cache_containsv(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 	uint32		hash;
 	int			i = 0;
 
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_maybe_disabled())	/* fast exit if file cache is disabled */
 		return 0;
 
@@ -1146,6 +854,8 @@ lfc_readv_select(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 	int			blocks_read = 0;
 	int			buf_offset = 0;
 
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_maybe_disabled())	/* fast exit if file cache is disabled */
 		return -1;
 
@@ -1456,7 +1166,7 @@ lfc_init_new_entry(FileCacheEntry* entry, uint32 hash)
 		/* Can't add this chunk - we don't have the space for it */
 		hash_search_with_hash_value(lfc_hash, &entry->key, hash,
 									HASH_REMOVE, NULL);
-		lfc_ctl->prewarm_canceled = true; /* cancel prewarm if LFC limit is reached */
+		lfc_prewarm_cancel = true; /* cancel prewarm if LFC limit is reached */
 		return false;
 	}
 
@@ -1511,6 +1221,8 @@ lfc_prefetch(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blkno,
 
 	int		chunk_offs = BLOCK_TO_CHUNK_OFF(blkno);
 
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_maybe_disabled())	/* fast exit if file cache is disabled */
 		return false;
 
@@ -1656,6 +1368,8 @@ lfc_writev(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 	uint32		entry_offset;
 	int			buf_offset = 0;
 
+	Assert(!neon_use_communicator_worker);
+
 	if (lfc_maybe_disabled())	/* fast exit if file cache is disabled */
 		return;
 
@@ -1874,7 +1588,6 @@ lfc_get_stats(size_t *num_entries)
 	return entries;
 }
 
-
 /*
  * Function returning data from the local file cache
  * relation node/tablespace/database/blocknum and access_counter
@@ -1978,15 +1691,15 @@ lfc_approximate_working_set_size_seconds(time_t duration, bool reset)
 }
 
 /*
- * Get metrics, for the built-in metrics exporter that's part of the communicator
- * process.
+ * Get metrics, for the built-in metrics exporter that's part of the
+ * communicator process.
  *
  * NB: This is called from a Rust tokio task inside the communicator process.
  * Acquiring lwlocks, elog(), allocating memory or anything else non-trivial
  * is strictly prohibited here!
  */
 struct LfcMetrics
-callback_get_lfc_metrics_unsafe(void)
+lfc_get_metrics_unsafe(void)
 {
 	struct LfcMetrics result = {
 		.lfc_cache_size_limit = (int64) lfc_size_limit * 1024 * 1024,
@@ -2007,82 +1720,3 @@ callback_get_lfc_metrics_unsafe(void)
 
 	return result;
 }
-
-
-PG_FUNCTION_INFO_V1(get_local_cache_state);
-
-Datum
-get_local_cache_state(PG_FUNCTION_ARGS)
-{
-	size_t max_entries = PG_ARGISNULL(0) ? lfc_prewarm_limit : PG_GETARG_INT32(0);
-	FileCacheState* fcs = lfc_get_state(max_entries);
-	if (fcs != NULL)
-		PG_RETURN_BYTEA_P((bytea*)fcs);
-	else
-		PG_RETURN_NULL();
-}
-
-PG_FUNCTION_INFO_V1(prewarm_local_cache);
-
-Datum
-prewarm_local_cache(PG_FUNCTION_ARGS)
-{
-	bytea* state = PG_GETARG_BYTEA_PP(0);
-	uint32 n_workers =  PG_GETARG_INT32(1);
-	FileCacheState* fcs = (FileCacheState*)state;
-
-	lfc_prewarm(fcs, n_workers);
-
-	PG_RETURN_NULL();
-}
-
-PG_FUNCTION_INFO_V1(get_prewarm_info);
-
-Datum
-get_prewarm_info(PG_FUNCTION_ARGS)
-{
-	Datum		values[4];
-	bool		nulls[4];
-	TupleDesc	tupdesc;
-	uint32 prewarmed_pages = 0;
-	uint32 skipped_pages = 0;
-	uint32 active_workers = 0;
-	uint32 total_pages;
-	size_t n_workers;
-
-	if (lfc_size_limit == 0)
-		PG_RETURN_NULL();
-
-	LWLockAcquire(lfc_lock, LW_SHARED);
-	if (!lfc_ctl || lfc_ctl->n_prewarm_workers == 0)
-	{
-		LWLockRelease(lfc_lock);
-		PG_RETURN_NULL();
-	}
-	n_workers = lfc_ctl->n_prewarm_workers;
-	total_pages = lfc_ctl->total_prewarm_pages;
-	for (size_t i = 0; i < n_workers; i++)
-	{
-		PrewarmWorkerState* ws = &lfc_ctl->prewarm_workers[i];
-		prewarmed_pages += ws->prewarmed_pages;
-		skipped_pages += ws->skipped_pages;
-		active_workers += ws->completed != 0;
-	}
-	LWLockRelease(lfc_lock);
-
-	tupdesc = CreateTemplateTupleDesc(4);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "total_pages", INT4OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "prewarmed_pages", INT4OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "skipped_pages", INT4OID, -1, 0);
-	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "active_workers", INT4OID, -1, 0);
-	tupdesc = BlessTupleDesc(tupdesc);
-
-	MemSet(nulls, 0, sizeof(nulls));
-
-	values[0] = Int32GetDatum(total_pages);
-	values[1] = Int32GetDatum(prewarmed_pages);
-	values[2] = Int32GetDatum(skipped_pages);
-	values[3] = Int32GetDatum(active_workers);
-
-	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
-}

pgxn/neon/file_cache.h

@@ -11,21 +11,19 @@
 #ifndef FILE_CACHE_h
 #define FILE_CACHE_h
 
-#include "neon_pgversioncompat.h"
+#include "lfc_prewarm.h"
+#include "neon.h"
 
-typedef struct FileCacheState
-{
-	int32		vl_len_;		/* varlena header (do not touch directly!) */
-	uint32		magic;
-	uint32		n_chunks;
-	uint32		n_pages;
-	uint16		chunk_size_log;
-	BufferTag	chunks[FLEXIBLE_ARRAY_MEMBER];
-	/* followed by bitmap */
-} FileCacheState;
+#include "neon_pgversioncompat.h"
 
 /* GUCs */
 extern bool lfc_store_prefetch_result;
+extern int	lfc_max_size;
+extern int	lfc_size_limit;
+extern char *lfc_path;
+
+extern bool lfc_do_prewarm;
+extern bool lfc_prewarm_cancel;
 
 /* functions for local file cache */
 extern void lfc_invalidate(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber nblocks);
@@ -44,17 +42,13 @@ extern int lfc_cache_containsv(NRelFileInfo rinfo, ForkNumber forkNum,
 extern void lfc_init(void);
 extern bool lfc_prefetch(NRelFileInfo rinfo, ForkNumber forknum, BlockNumber blkno,
 						 const void* buffer, XLogRecPtr lsn);
-extern FileCacheState* lfc_get_state(size_t max_entries);
-extern void lfc_prewarm(FileCacheState* fcs, uint32 n_workers);
 
-typedef struct LfcStatsEntry
-{
-	const char *metric_name;
-	bool		isnull;
-	uint64		value;
-} LfcStatsEntry;
+extern FileCacheState* lfc_get_state(size_t max_entries);
 extern LfcStatsEntry *lfc_get_stats(size_t *num_entries);
 
+struct LfcMetrics; /* defined in communicator_bindings.h */
+extern struct LfcMetrics lfc_get_metrics_unsafe(void);
+
 typedef struct
 {
 	uint32		pageoffs;
@@ -69,7 +63,6 @@ extern LocalCachePagesRec *lfc_local_cache_pages(size_t *num_entries);
 
 extern int32 lfc_approximate_working_set_size_seconds(time_t duration, bool reset);
 
-
 static inline bool
 lfc_read(NRelFileInfo rinfo, ForkNumber forkNum, BlockNumber blkno,
 		 void *buffer)

pgxn/neon/lfc_prewarm.c

@@ -0,0 +1,661 @@
+/*-------------------------------------------------------------------------
+ *
+ * lfc_prewarm.c
+ *		Functions related to LFC prewarming
+ *
+ * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "bitmap.h"
+#include "communicator.h"
+#include "communicator_new.h"
+#include "file_cache.h"
+#include "lfc_prewarm.h"
+#include "neon.h"
+#include "neon_utils.h"
+#include "pagestore_client.h"
+
+#include "funcapi.h"
+#include "miscadmin.h"
+#include "postmaster/bgworker.h"
+#include "storage/dsm.h"
+#include "tcop/tcopprot.h"
+#include "utils/timestamp.h"
+
+#define MAX_PREWARM_WORKERS 8
+
+typedef struct PrewarmWorkerState
+{
+	uint32		prewarmed_pages;
+	uint32		skipped_pages;
+	TimestampTz completed;
+} PrewarmWorkerState;
+
+typedef struct PrewarmControl
+{
+	/* -1 when not using workers, 0 when no prewarm has been performed */
+	size_t		n_prewarm_workers;
+	size_t		total_prewarm_pages;
+	bool		prewarm_active;
+	bool		prewarm_canceled;
+
+	/* These are used in the non-worker mode */
+	uint32		prewarmed_pages;
+	uint32		skipped_pages;
+	TimestampTz completed;
+
+	/* These are used with workers */
+	PrewarmWorkerState prewarm_workers[MAX_PREWARM_WORKERS];
+	dsm_handle	prewarm_lfc_state_handle;
+	size_t		prewarm_batch;
+	size_t		n_prewarm_entries;
+} PrewarmControl;
+
+static PrewarmControl *prewarm_ctl;
+
+static int	lfc_prewarm_limit;
+static int	lfc_prewarm_batch;
+
+static LWLockId prewarm_lock;
+
+bool AmPrewarmWorker;
+
+static void lfc_prewarm_with_workers(FileCacheState *fcs, uint32 n_workers);
+static void lfc_prewarm_with_async_requests(FileCacheState *fcs);
+PGDLLEXPORT void lfc_prewarm_main(Datum main_arg);
+
+void
+pg_init_prewarm(void)
+{
+	DefineCustomIntVariable("neon.file_cache_prewarm_limit",
+							"Maximal number of prewarmed chunks",
+							NULL,
+							&lfc_prewarm_limit,
+							INT_MAX,	/* no limit by default */
+							0,
+							INT_MAX,
+							PGC_SIGHUP,
+							0,
+							NULL,
+							NULL,
+							NULL);
+
+	DefineCustomIntVariable("neon.file_cache_prewarm_batch",
+							"Number of pages retrivied by prewarm from page server",
+							NULL,
+							&lfc_prewarm_batch,
+							64,
+							1,
+							INT_MAX,
+							PGC_SIGHUP,
+							0,
+							NULL,
+							NULL,
+							NULL);
+}
+
+static size_t
+PrewarmShmemSize(void)
+{
+	return sizeof(PrewarmControl);
+}
+
+void
+PrewarmShmemRequest(void)
+{
+	RequestAddinShmemSpace(PrewarmShmemSize());
+	RequestNamedLWLockTranche("prewarm_lock", 1);
+}
+
+void
+PrewarmShmemInit(void)
+{
+	bool		found;
+
+	prewarm_ctl = (PrewarmControl *) ShmemInitStruct("Prewarmer shmem state",
+								PrewarmShmemSize(),
+								&found);
+	if (!found)
+	{
+		/* it's zeroed already */
+
+		prewarm_lock = (LWLockId) GetNamedLWLockTranche("prewarm_lock");
+	}
+}
+
+static void
+validate_fcs(FileCacheState *fcs)
+{
+	size_t fcs_size;
+#if 0
+	size_t fcs_chunk_size_log;
+#endif
+
+	if (fcs->magic != FILE_CACHE_STATE_MAGIC)
+	{
+		elog(ERROR, "LFC: Invalid file cache state magic: %X", fcs->magic);
+	}
+
+	fcs_size = VARSIZE(fcs);
+	if (FILE_CACHE_STATE_SIZE(fcs) != fcs_size)
+	{
+		elog(ERROR, "LFC: Invalid file cache state size: %u vs. %u", (unsigned)FILE_CACHE_STATE_SIZE(fcs), VARSIZE(fcs));
+	}
+
+	/* FIXME */
+#if 0
+	fcs_chunk_size_log = fcs->chunk_size_log;
+	if (fcs_chunk_size_log > MAX_BLOCKS_PER_CHUNK_LOG)
+	{
+		elog(ERROR, "LFC: Invalid chunk size log: %u", fcs->chunk_size_log);
+	}
+#endif
+}
+
+/*
+ * Prewarm LFC cache to the specified state. It uses lfc_prefetch function to
+ * load prewarmed page without hoilding shared buffer lock and avoid race
+ * conditions with other backends.
+ */
+void
+lfc_prewarm_with_workers(FileCacheState *fcs, uint32 n_workers)
+{
+	size_t n_entries;
+	size_t prewarm_batch = Min(lfc_prewarm_batch, readahead_buffer_size);
+	size_t fcs_size = VARSIZE(fcs);
+	dsm_segment *seg;
+	BackgroundWorkerHandle* bgw_handle[MAX_PREWARM_WORKERS];
+
+	Assert(!neon_use_communicator_worker);
+
+	if (prewarm_batch == 0 || lfc_prewarm_limit == 0 || n_workers == 0)
+	{
+		elog(LOG, "LFC: prewarm is disabled");
+		return;
+	}
+
+	if (n_workers > MAX_PREWARM_WORKERS)
+	{
+		elog(ERROR, "LFC: too many prewarm workers, maximum is %d", MAX_PREWARM_WORKERS);
+	}
+
+	if (fcs == NULL || fcs->n_chunks == 0)
+	{
+		elog(LOG, "LFC: nothing to prewarm");
+		return;
+	}
+
+	n_entries = Min(fcs->n_chunks, lfc_prewarm_limit);
+	Assert(n_entries != 0);
+
+	LWLockAcquire(prewarm_lock, LW_EXCLUSIVE);
+
+	/* Do not prewarm more entries than LFC limit */
+	/* FIXME */
+#if 0
+	if (prewarm_ctl->limit <= prewarm_ctl->size)
+	{
+		elog(LOG, "LFC: skip prewarm because LFC is already filled");
+		LWLockRelease(prewarm_lock);
+		return;
+	}
+#endif
+	
+	if (prewarm_ctl->prewarm_active)
+	{
+		LWLockRelease(prewarm_lock);
+		elog(ERROR, "LFC: skip prewarm because another prewarm is still active");
+	}
+	prewarm_ctl->n_prewarm_entries = n_entries;
+	prewarm_ctl->n_prewarm_workers = n_workers;
+	prewarm_ctl->prewarm_active = true;
+	prewarm_ctl->prewarm_canceled = false;
+	prewarm_ctl->prewarm_batch = prewarm_batch;
+	memset(prewarm_ctl->prewarm_workers, 0, n_workers*sizeof(PrewarmWorkerState));
+
+	/* Calculate total number of pages to be prewarmed */
+	prewarm_ctl->total_prewarm_pages = fcs->n_pages;
+
+	LWLockRelease(prewarm_lock);
+
+	seg = dsm_create(fcs_size, 0);
+	memcpy(dsm_segment_address(seg), fcs, fcs_size);
+	prewarm_ctl->prewarm_lfc_state_handle = dsm_segment_handle(seg);
+
+	/* Spawn background workers */
+	for (uint32 i = 0; i < n_workers; i++)
+	{
+		BackgroundWorker worker = {0};
+
+		worker.bgw_flags = BGWORKER_SHMEM_ACCESS;
+		worker.bgw_start_time = BgWorkerStart_ConsistentState;
+		worker.bgw_restart_time = BGW_NEVER_RESTART;
+		strcpy(worker.bgw_library_name, "neon");
+		strcpy(worker.bgw_function_name, "lfc_prewarm_main");
+		snprintf(worker.bgw_name, BGW_MAXLEN, "LFC prewarm worker %d", i+1);
+		strcpy(worker.bgw_type, "LFC prewarm worker");
+		worker.bgw_main_arg = Int32GetDatum(i);
+		/* must set notify PID to wait for shutdown */
+		worker.bgw_notify_pid = MyProcPid;
+
+		if (!RegisterDynamicBackgroundWorker(&worker, &bgw_handle[i]))
+		{
+			ereport(LOG,
+					(errcode(ERRCODE_INSUFFICIENT_RESOURCES),
+					 errmsg("LFC: registering dynamic bgworker prewarm failed"),
+					 errhint("Consider increasing the configuration parameter \"%s\".", "max_worker_processes")));
+			n_workers = i;
+			prewarm_ctl->prewarm_canceled = true;
+			break;
+		}
+	}
+
+	for (uint32 i = 0; i < n_workers; i++)
+	{
+		bool interrupted;
+		do
+		{
+			interrupted = false;
+			PG_TRY();
+			{
+				BgwHandleStatus status = WaitForBackgroundWorkerShutdown(bgw_handle[i]);
+				if (status != BGWH_STOPPED && status != BGWH_POSTMASTER_DIED)
+				{
+					elog(LOG, "LFC: Unexpected status of prewarm worker termination: %d", status);
+				}
+			}
+			PG_CATCH();
+			{
+				elog(LOG, "LFC: cancel prewarm");
+				prewarm_ctl->prewarm_canceled = true;
+				interrupted = true;
+			}
+			PG_END_TRY();
+		} while (interrupted);
+
+		if (!prewarm_ctl->prewarm_workers[i].completed)
+		{
+			/* Background worker doesn't set completion time: it means that it was abnormally terminated */
+			elog(LOG, "LFC: prewarm worker %d failed", i+1);
+			/* Set completion time to prevent get_prewarm_info from considering this worker as active */
+			prewarm_ctl->prewarm_workers[i].completed = GetCurrentTimestamp();
+		}
+	}
+	dsm_detach(seg);
+
+	LWLockAcquire(prewarm_lock, LW_EXCLUSIVE);
+	prewarm_ctl->prewarm_active = false;
+	LWLockRelease(prewarm_lock);
+}
+
+
+void
+lfc_prewarm_main(Datum main_arg)
+{
+	size_t snd_idx = 0, rcv_idx = 0;
+	size_t n_sent = 0, n_received = 0;
+	size_t fcs_chunk_size_log;
+	size_t max_prefetch_pages;
+	size_t prewarm_batch;
+	size_t n_workers;
+	dsm_segment *seg;
+	FileCacheState* fcs;
+	uint8* bitmap;
+	BufferTag tag;
+	PrewarmWorkerState* ws;
+	uint32 worker_id = DatumGetInt32(main_arg);
+
+	Assert(!neon_use_communicator_worker);
+
+	AmPrewarmWorker = true;
+
+	pqsignal(SIGTERM, die);
+	BackgroundWorkerUnblockSignals();
+
+	seg = dsm_attach(prewarm_ctl->prewarm_lfc_state_handle);
+	if (seg == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 errmsg("could not map dynamic shared memory segment")));
+
+	fcs = (FileCacheState*) dsm_segment_address(seg);
+	prewarm_batch = prewarm_ctl->prewarm_batch;
+	fcs_chunk_size_log = fcs->chunk_size_log;
+	n_workers = prewarm_ctl->n_prewarm_workers;
+	max_prefetch_pages = prewarm_ctl->n_prewarm_entries << fcs_chunk_size_log;
+	ws = &prewarm_ctl->prewarm_workers[worker_id];
+	bitmap = FILE_CACHE_STATE_BITMAP(fcs);
+
+	/* enable prefetch in LFC */
+	lfc_store_prefetch_result = true;
+	lfc_do_prewarm = true; /* Flag for lfc_prefetch preventing replacement of existed entries if LFC cache is full */
+
+	elog(LOG, "LFC: worker %d start prewarming", worker_id);
+	while (!prewarm_ctl->prewarm_canceled)
+	{
+		if (snd_idx < max_prefetch_pages)
+		{
+			if ((snd_idx >> fcs_chunk_size_log) % n_workers != worker_id)
+			{
+				/* If there are multiple workers, split chunks between them */
+				snd_idx += 1 << fcs_chunk_size_log;
+			}
+			else
+			{
+				if (BITMAP_ISSET(bitmap, snd_idx))
+				{
+					tag = fcs->chunks[snd_idx >> fcs_chunk_size_log];
+					tag.blockNum += snd_idx & ((1 << fcs_chunk_size_log) - 1);
+
+					if (!BufferTagIsValid(&tag))
+						elog(ERROR, "LFC: Invalid buffer tag: %u", tag.blockNum);
+
+					if (!lfc_cache_contains(BufTagGetNRelFileInfo(tag), tag.forkNum, tag.blockNum))
+					{
+						(void) communicator_prefetch_register_bufferv(tag, NULL, 1, NULL);
+						n_sent += 1;
+					}
+					else
+					{
+						ws->skipped_pages += 1;
+						BITMAP_CLR(bitmap, snd_idx);
+					}
+				}
+				snd_idx += 1;
+			}
+		}
+		if (n_sent >= n_received + prewarm_batch || snd_idx == max_prefetch_pages)
+		{
+			if (n_received == n_sent && snd_idx == max_prefetch_pages)
+			{
+				break;
+			}
+			if ((rcv_idx >> fcs_chunk_size_log) % n_workers != worker_id)
+			{
+				/* Skip chunks processed by other workers */
+				rcv_idx += 1 << fcs_chunk_size_log;
+				continue;
+			}
+
+			/* Locate next block to prefetch */
+			while (!BITMAP_ISSET(bitmap, rcv_idx))
+			{
+				rcv_idx += 1;
+			}
+			tag = fcs->chunks[rcv_idx >> fcs_chunk_size_log];
+			tag.blockNum += rcv_idx & ((1 << fcs_chunk_size_log) - 1);
+			if (communicator_prefetch_receive(tag))
+			{
+				ws->prewarmed_pages += 1;
+			}
+			else
+			{
+				ws->skipped_pages += 1;
+			}
+			rcv_idx += 1;
+			n_received += 1;
+		}
+	}
+	/* No need to perform prefetch cleanup here because prewarm worker will be terminated and
+	 * connection to PS dropped just after return from this function.
+	 */
+	Assert(n_sent == n_received || prewarm_ctl->prewarm_canceled);
+	elog(LOG, "LFC: worker %d complete prewarming: loaded %ld pages", worker_id, (long)n_received);
+	prewarm_ctl->prewarm_workers[worker_id].completed = GetCurrentTimestamp();
+}
+
+/*
+ * Prewarm LFC cache to the specified state. Uses the new communicator
+ *
+ * FIXME: Is there a race condition because we're not holding Postgres
+ * buffer manager locks?
+ */
+static void
+lfc_prewarm_with_async_requests(FileCacheState *fcs)
+{
+	size_t n_entries;
+	uint8	   *bitmap;
+	uint64		bitno;
+	int			blocks_per_chunk;
+
+	Assert(neon_use_communicator_worker);
+
+	if (lfc_prewarm_limit == 0)
+	{
+		elog(LOG, "LFC: prewarm is disabled");
+		return;
+	}
+
+	if (fcs == NULL || fcs->n_chunks == 0)
+	{
+		elog(LOG, "LFC: nothing to prewarm");
+		return;
+	}
+
+	n_entries = Min(fcs->n_chunks, lfc_prewarm_limit);
+	Assert(n_entries != 0);
+
+	LWLockAcquire(prewarm_lock, LW_EXCLUSIVE);
+
+	/* Do not prewarm more entries than LFC limit */
+	/* FIXME */
+#if 0
+	if (prewarm_ctl->limit <= prewarm_ctl->size)
+	{
+		elog(LOG, "LFC: skip prewarm because LFC is already filled");
+		LWLockRelease(prewarm_lock);
+		return;
+	}
+#endif
+
+	if (prewarm_ctl->prewarm_active)
+	{
+		LWLockRelease(prewarm_lock);
+		elog(ERROR, "LFC: skip prewarm because another prewarm is still active");
+	}
+	prewarm_ctl->n_prewarm_entries = n_entries;
+	prewarm_ctl->n_prewarm_workers = -1;
+	prewarm_ctl->prewarm_active = true;
+	prewarm_ctl->prewarm_canceled = false;
+
+	/* Calculate total number of pages to be prewarmed */
+	prewarm_ctl->total_prewarm_pages = fcs->n_pages;
+
+	LWLockRelease(prewarm_lock);
+
+	elog(LOG, "LFC: start prewarming");
+	lfc_do_prewarm = true;
+	lfc_prewarm_cancel = false;
+
+	bitmap = FILE_CACHE_STATE_BITMAP(fcs);
+
+	blocks_per_chunk = 1 << fcs->chunk_size_log;
+
+	bitno = 0;
+	for (uint32 chunkno = 0; chunkno < fcs->n_chunks; chunkno++)
+	{
+		BufferTag *chunk_tag = &fcs->chunks[chunkno];
+		BlockNumber request_startblkno = InvalidBlockNumber;
+		BlockNumber request_endblkno;
+
+		if (!BufferTagIsValid(chunk_tag))
+			elog(ERROR, "LFC: Invalid buffer tag: %u", chunk_tag->blockNum);
+
+		if (lfc_prewarm_cancel)
+		{
+			prewarm_ctl->prewarm_canceled = true;
+			break;
+		}
+
+		/* take next chunk */
+		for (int j = 0; j < blocks_per_chunk; j++)
+		{
+			BlockNumber blkno = chunk_tag->blockNum + j;
+
+			if (BITMAP_ISSET(bitmap, bitno))
+			{
+				if (request_startblkno != InvalidBlockNumber)
+				{
+					if (request_endblkno == blkno)
+					{
+						/* append this block to the request */
+						request_endblkno++;
+					}
+					else
+					{
+						/* flush this request, and start new one */
+						communicator_new_prefetch_register_bufferv(
+							BufTagGetNRelFileInfo(*chunk_tag),
+							chunk_tag->forkNum,
+							request_startblkno,
+							request_endblkno - request_startblkno
+							);
+						request_startblkno = blkno;
+						request_endblkno = blkno + 1;
+					}
+				}
+				else
+				{
+					/* flush this request, if any, and start new one */
+					if (request_startblkno != InvalidBlockNumber)
+					{
+						communicator_new_prefetch_register_bufferv(
+							BufTagGetNRelFileInfo(*chunk_tag),
+							chunk_tag->forkNum,
+							request_startblkno,
+							request_endblkno - request_startblkno
+							);
+					}
+					request_startblkno = blkno;
+					request_endblkno = blkno + 1;
+				}
+				prewarm_ctl->prewarmed_pages += 1;
+			}
+			bitno++;
+		}
+
+		/* flush this request */
+		communicator_new_prefetch_register_bufferv(
+			BufTagGetNRelFileInfo(*chunk_tag),
+			chunk_tag->forkNum,
+			request_startblkno,
+			request_endblkno - request_startblkno
+			);
+		request_startblkno = request_endblkno = InvalidBlockNumber;
+	}
+
+	elog(LOG, "LFC: complete prewarming: loaded %lu pages", (unsigned long) prewarm_ctl->prewarmed_pages);
+	prewarm_ctl->completed = GetCurrentTimestamp();
+
+	LWLockAcquire(prewarm_lock, LW_EXCLUSIVE);
+	prewarm_ctl->prewarm_active = false;
+	LWLockRelease(prewarm_lock);
+}
+
+PG_FUNCTION_INFO_V1(get_local_cache_state);
+
+Datum
+get_local_cache_state(PG_FUNCTION_ARGS)
+{
+	size_t max_entries = PG_ARGISNULL(0) ? lfc_prewarm_limit : PG_GETARG_INT32(0);
+	FileCacheState* fcs;
+
+	if (neon_use_communicator_worker)
+		fcs = communicator_new_get_lfc_state(max_entries);
+	else
+		fcs = lfc_get_state(max_entries);
+
+	if (fcs != NULL)
+		PG_RETURN_BYTEA_P((bytea*)fcs);
+	else
+		PG_RETURN_NULL();
+}
+
+PG_FUNCTION_INFO_V1(prewarm_local_cache);
+
+Datum
+prewarm_local_cache(PG_FUNCTION_ARGS)
+{
+	bytea* state = PG_GETARG_BYTEA_PP(0);
+	uint32 n_workers =  PG_GETARG_INT32(1);
+	FileCacheState* fcs;
+
+	fcs = (FileCacheState *)state;
+	validate_fcs(fcs);
+
+	if (neon_use_communicator_worker)
+		lfc_prewarm_with_async_requests(fcs);
+	else
+		lfc_prewarm_with_workers(fcs, n_workers);
+
+	PG_RETURN_NULL();
+}
+
+PG_FUNCTION_INFO_V1(get_prewarm_info);
+
+Datum
+get_prewarm_info(PG_FUNCTION_ARGS)
+{
+	Datum		values[4];
+	bool		nulls[4];
+	TupleDesc	tupdesc;
+	uint32		prewarmed_pages = 0;
+	uint32		skipped_pages = 0;
+	uint32		active_workers = 0;
+	uint32		total_pages;
+
+	if (lfc_size_limit == 0)
+		PG_RETURN_NULL();
+
+	LWLockAcquire(prewarm_lock, LW_SHARED);
+	if (!prewarm_ctl || prewarm_ctl->n_prewarm_workers == 0)
+	{
+		LWLockRelease(prewarm_lock);
+		PG_RETURN_NULL();
+	}
+
+	if (prewarm_ctl->n_prewarm_workers == -1)
+	{
+		total_pages = prewarm_ctl->total_prewarm_pages;
+		prewarmed_pages = prewarm_ctl->prewarmed_pages;
+		skipped_pages = prewarm_ctl->prewarmed_pages;
+		active_workers = 1;
+	}
+	else
+	{
+		size_t		n_workers;
+
+		n_workers = prewarm_ctl->n_prewarm_workers;
+		total_pages = prewarm_ctl->total_prewarm_pages;
+		for (size_t i = 0; i < n_workers; i++)
+		{
+			PrewarmWorkerState *ws = &prewarm_ctl->prewarm_workers[i];
+
+			prewarmed_pages += ws->prewarmed_pages;
+			skipped_pages += ws->skipped_pages;
+			active_workers += ws->completed != 0;
+		}
+	}
+	LWLockRelease(prewarm_lock);
+
+	tupdesc = CreateTemplateTupleDesc(4);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 1, "total_pages", INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 2, "prewarmed_pages", INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 3, "skipped_pages", INT4OID, -1, 0);
+	TupleDescInitEntry(tupdesc, (AttrNumber) 4, "active_workers", INT4OID, -1, 0);
+	tupdesc = BlessTupleDesc(tupdesc);
+
+	MemSet(nulls, 0, sizeof(nulls));
+
+	values[0] = Int32GetDatum(total_pages);
+	values[1] = Int32GetDatum(prewarmed_pages);
+	values[2] = Int32GetDatum(skipped_pages);
+	values[3] = Int32GetDatum(active_workers);
+
+	PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
+}

pgxn/neon/lfc_prewarm.h

@@ -0,0 +1,39 @@
+/*-------------------------------------------------------------------------
+ *
+ * lfc_prewarm.h
+ *	  Local File Cache prewarmer
+ *
+ * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef LFC_PREWARM_H
+#define LFC_PREWARM_H
+
+#include "storage/buf_internals.h"
+
+typedef struct FileCacheState
+{
+	int32		vl_len_;		/* varlena header (do not touch directly!) */
+	uint32		magic;
+	uint32		n_chunks;
+	uint32		n_pages;
+	uint16		chunk_size_log;
+	BufferTag	chunks[FLEXIBLE_ARRAY_MEMBER];
+	/* followed by bitmap */
+} FileCacheState;
+
+#define FILE_CACHE_STATE_MAGIC 0xfcfcfcfc
+
+#define FILE_CACHE_STATE_BITMAP(fcs)	((uint8*)&(fcs)->chunks[(fcs)->n_chunks])
+#define FILE_CACHE_STATE_SIZE_FOR_CHUNKS(n_chunks, blocks_per_chunk)	(sizeof(FileCacheState) + (n_chunks)*sizeof(BufferTag) + (((n_chunks) * blocks_per_chunk)+7)/8)
+#define FILE_CACHE_STATE_SIZE(fcs)		(sizeof(FileCacheState) + (fcs->n_chunks)*sizeof(BufferTag) + (((fcs->n_chunks) << fcs->chunk_size_log)+7)/8)
+
+extern void pg_init_prewarm(void);
+extern void PrewarmShmemRequest(void);
+extern void PrewarmShmemInit(void);
+
+#endif							/* LFC_PREWARM_H */
+
+