From 71d09c78d4ffd159cfcd83c4c1b919a4c7eef7c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arpad=20M=C3=BCller?= Date: Sat, 19 Oct 2024 00:23:49 +0200 Subject: [PATCH 01/12] Accept basebackup --gzip requests (#9456) In #9453, we want to remove the non-gzipped basebackup code in the computes, and always request gzipped basebackups. However, right now the pageserver's page service only accepts basebackup requests in the following formats: * `basebackup `, lsn is determined by the pageserver as the most recent one (`timeline.get_last_record_rlsn()`) * `basebackup ` * `basebackup --gzip` We add a fourth case, `basebackup --gzip` to allow gzipping the request for the latest lsn as well. --- pageserver/src/page_service.rs | 32 ++++++++++++++++---------------- 1 file changed, 16 insertions(+), 16 deletions(-) diff --git a/pageserver/src/page_service.rs b/pageserver/src/page_service.rs index afb2f92ff8..62b14cb83e 100644 --- a/pageserver/src/page_service.rs +++ b/pageserver/src/page_service.rs @@ -1326,22 +1326,22 @@ where .for_command(ComputeCommandKind::Basebackup) .inc(); - let lsn = if let Some(lsn_str) = params.get(2) { - Some( - Lsn::from_str(lsn_str) - .with_context(|| format!("Failed to parse Lsn from {lsn_str}"))?, - ) - } else { - None - }; - - let gzip = match params.get(3) { - Some(&"--gzip") => true, - None => false, - Some(third_param) => { - return Err(QueryError::Other(anyhow::anyhow!( - "Parameter in position 3 unknown {third_param}", - ))) + let (lsn, gzip) = match (params.get(2), params.get(3)) { + (None, _) => (None, false), + (Some(&"--gzip"), _) => (None, true), + (Some(lsn_str), gzip_str_opt) => { + let lsn = Lsn::from_str(lsn_str) + .with_context(|| format!("Failed to parse Lsn from {lsn_str}"))?; + let gzip = match gzip_str_opt { + Some(&"--gzip") => true, + None => false, + Some(third_param) => { + return Err(QueryError::Other(anyhow::anyhow!( + "Parameter in position 3 unknown {third_param}", + ))) + } + }; + (Some(lsn), gzip) } }; From cc25ef73423ea0108986436501481b0154443932 Mon Sep 17 00:00:00 2001 From: Conrad Ludgate Date: Sun, 20 Oct 2024 13:42:50 +0100 Subject: [PATCH 02/12] bump pg-session-jwt version (#9455) forgot to bump this before --- proxy/src/serverless/local_conn_pool.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/proxy/src/serverless/local_conn_pool.rs b/proxy/src/serverless/local_conn_pool.rs index beb2ad4e8f..e1ad46c751 100644 --- a/proxy/src/serverless/local_conn_pool.rs +++ b/proxy/src/serverless/local_conn_pool.rs @@ -39,7 +39,7 @@ use crate::usage_metrics::{Ids, MetricCounter, USAGE_METRICS}; use crate::{DbName, RoleName}; pub(crate) const EXT_NAME: &str = "pg_session_jwt"; -pub(crate) const EXT_VERSION: &str = "0.1.1"; +pub(crate) const EXT_VERSION: &str = "0.1.2"; pub(crate) const EXT_SCHEMA: &str = "auth"; struct ConnPoolEntry { From ed958da38a0edf7853ee999f43737ac2ff69f920 Mon Sep 17 00:00:00 2001 From: Folke Behrens Date: Mon, 21 Oct 2024 10:29:23 +0200 Subject: [PATCH 03/12] proxy: Make tests fail fast when test proxy exited early (#9432) This currently happens when proxy is not compiled with feature `testing`. Also fix an adjacent function. --- test_runner/fixtures/neon_fixtures.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/test_runner/fixtures/neon_fixtures.py b/test_runner/fixtures/neon_fixtures.py index 3cd8019e32..747c2c0d63 100644 --- a/test_runner/fixtures/neon_fixtures.py +++ b/test_runner/fixtures/neon_fixtures.py @@ -3175,10 +3175,13 @@ class NeonProxy(PgProtocol): # two seconds. Raises subprocess.TimeoutExpired if the proxy does not exit in time. def wait_for_exit(self, timeout=2): if self._popen: - self._popen.wait(timeout=2) + self._popen.wait(timeout=timeout) @backoff.on_exception(backoff.expo, requests.exceptions.RequestException, max_time=10) def _wait_until_ready(self): + assert ( + self._popen and self._popen.poll() is None + ), "Proxy exited unexpectedly. Check test log." requests.get(f"http://{self.host}:{self.http_port}/v1/status") def http_query(self, query, args, **kwargs): From 5b37485c99836abb060bed8eb1172870b31504b2 Mon Sep 17 00:00:00 2001 From: Alexander Bayandin Date: Mon, 21 Oct 2024 09:51:12 +0100 Subject: [PATCH 04/12] Rename dockerfiles from `Dockerfile.` to `.Dockerfile` (#9446) ## Problem Our dockerfiles, for some historical reason, have unconventional names `Dockerfile.`, and some tools (like GitHub UI) fail to highlight the syntax in them. > Some projects may need distinct Dockerfiles for specific purposes. A common convention is to name these `.Dockerfile` From: https://docs.docker.com/build/concepts/dockerfile/#filename ## Summary of changes - Rename `Dockerfile.build-tools` -> `build-tools.Dockerfile` - Rename `compute/Dockerfile.compute-node` -> `compute/compute-node.Dockerfile` --- .github/workflows/_build-and-test-locally.yml | 8 ++++---- .github/workflows/build-build-tools-image.yml | 2 +- .github/workflows/build_and_test.yml | 6 +++--- .github/workflows/check-build-tools-image.yml | 2 +- .github/workflows/trigger-e2e-tests.yml | 2 +- Dockerfile.build-tools => build-tools.Dockerfile | 2 +- compute/README.md | 6 +++--- .../{Dockerfile.compute-node => compute-node.Dockerfile} | 0 docs/docker.md | 6 +++--- 9 files changed, 17 insertions(+), 17 deletions(-) rename Dockerfile.build-tools => build-tools.Dockerfile (99%) rename compute/{Dockerfile.compute-node => compute-node.Dockerfile} (100%) diff --git a/.github/workflows/_build-and-test-locally.yml b/.github/workflows/_build-and-test-locally.yml index 3aa671fab1..c0f59fbdd5 100644 --- a/.github/workflows/_build-and-test-locally.yml +++ b/.github/workflows/_build-and-test-locally.yml @@ -124,28 +124,28 @@ jobs: uses: actions/cache@v4 with: path: pg_install/v14 - key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v14_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }} + key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v14_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }} - name: Cache postgres v15 build id: cache_pg_15 uses: actions/cache@v4 with: path: pg_install/v15 - key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v15_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }} + key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v15_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }} - name: Cache postgres v16 build id: cache_pg_16 uses: actions/cache@v4 with: path: pg_install/v16 - key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v16_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }} + key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v16_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }} - name: Cache postgres v17 build id: cache_pg_17 uses: actions/cache@v4 with: path: pg_install/v17 - key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v17_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'Dockerfile.build-tools') }} + key: v1-${{ runner.os }}-${{ runner.arch }}-${{ inputs.build-type }}-pg-${{ steps.pg_v17_rev.outputs.pg_rev }}-bookworm-${{ hashFiles('Makefile', 'build-tools.Dockerfile') }} - name: Build postgres v14 if: steps.cache_pg_14.outputs.cache-hit != 'true' diff --git a/.github/workflows/build-build-tools-image.yml b/.github/workflows/build-build-tools-image.yml index 0f05276579..10750089b2 100644 --- a/.github/workflows/build-build-tools-image.yml +++ b/.github/workflows/build-build-tools-image.yml @@ -82,7 +82,7 @@ jobs: - uses: docker/build-push-action@v6 with: - file: Dockerfile.build-tools + file: build-tools.Dockerfile context: . provenance: false push: true diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index b669eaeb11..1186b9927b 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -683,7 +683,7 @@ jobs: provenance: false push: true pull: true - file: compute/Dockerfile.compute-node + file: compute/compute-node.Dockerfile cache-from: type=registry,ref=cache.neon.build/compute-node-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }} cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/compute-node-{0}:cache-{1}-{2},mode=max', matrix.version.pg, matrix.version.debian, matrix.arch) || '' }} tags: | @@ -703,7 +703,7 @@ jobs: provenance: false push: true pull: true - file: compute/Dockerfile.compute-node + file: compute/compute-node.Dockerfile target: neon-pg-ext-test cache-from: type=registry,ref=cache.neon.build/neon-test-extensions-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }} cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/neon-test-extensions-{0}:cache-{1}-{2},mode=max', matrix.version.pg, matrix.version.debian, matrix.arch) || '' }} @@ -728,7 +728,7 @@ jobs: provenance: false push: true pull: true - file: compute/Dockerfile.compute-node + file: compute/compute-node.Dockerfile cache-from: type=registry,ref=cache.neon.build/neon-test-extensions-${{ matrix.version.pg }}:cache-${{ matrix.version.debian }}-${{ matrix.arch }} cache-to: ${{ github.ref_name == 'main' && format('type=registry,ref=cache.neon.build/compute-tools-{0}:cache-{1}-{2},mode=max', matrix.version.pg, matrix.version.debian, matrix.arch) || '' }} tags: | diff --git a/.github/workflows/check-build-tools-image.yml b/.github/workflows/check-build-tools-image.yml index 807a9ef3bd..a7a15ad58b 100644 --- a/.github/workflows/check-build-tools-image.yml +++ b/.github/workflows/check-build-tools-image.yml @@ -31,7 +31,7 @@ jobs: id: get-build-tools-tag env: IMAGE_TAG: | - ${{ hashFiles('Dockerfile.build-tools', + ${{ hashFiles('build-tools.Dockerfile', '.github/workflows/check-build-tools-image.yml', '.github/workflows/build-build-tools-image.yml') }} run: | diff --git a/.github/workflows/trigger-e2e-tests.yml b/.github/workflows/trigger-e2e-tests.yml index 5c5423e252..1e7264c55a 100644 --- a/.github/workflows/trigger-e2e-tests.yml +++ b/.github/workflows/trigger-e2e-tests.yml @@ -112,7 +112,7 @@ jobs: # This isn't exhaustive, just the paths that are most directly compute-related. # For example, compute_ctl also depends on libs/utils, but we don't trigger # an e2e run on that. - vendor/*|pgxn/*|compute_tools/*|libs/vm_monitor/*|compute/Dockerfile.compute-node) + vendor/*|pgxn/*|compute_tools/*|libs/vm_monitor/*|compute/compute-node.Dockerfile) platforms=$(echo "${platforms}" | jq --compact-output '. += ["k8s-neonvm"] | unique') ;; *) diff --git a/Dockerfile.build-tools b/build-tools.Dockerfile similarity index 99% rename from Dockerfile.build-tools rename to build-tools.Dockerfile index f05c60661c..818cc1b6db 100644 --- a/Dockerfile.build-tools +++ b/build-tools.Dockerfile @@ -142,7 +142,7 @@ RUN wget -O /tmp/openssl-${OPENSSL_VERSION}.tar.gz https://www.openssl.org/sourc # Use the same version of libicu as the compute nodes so that # clusters created using inidb on pageserver can be used by computes. # -# TODO: at this time, Dockerfile.compute-node uses the debian bullseye libicu +# TODO: at this time, compute-node.Dockerfile uses the debian bullseye libicu # package, which is 67.1. We're duplicating that knowledge here, and also, technically, # Debian has a few patches on top of 67.1 that we're not adding here. ENV ICU_VERSION=67.1 diff --git a/compute/README.md b/compute/README.md index bb1e42ab53..61e0eee4be 100644 --- a/compute/README.md +++ b/compute/README.md @@ -1,7 +1,7 @@ This directory contains files that are needed to build the compute images, or included in the compute images. -Dockerfile.compute-node +compute-node.Dockerfile To build the compute image vm-image-spec.yaml @@ -14,8 +14,8 @@ etc/ patches/ Some extensions need to be patched to work with Neon. This directory contains such patches. They are applied to the extension - sources in Dockerfile.compute-node + sources in compute-node.Dockerfile In addition to these, postgres itself, the neon postgres extension, and compute_ctl are built and copied into the compute image by -Dockerfile.compute-node. +compute-node.Dockerfile. diff --git a/compute/Dockerfile.compute-node b/compute/compute-node.Dockerfile similarity index 100% rename from compute/Dockerfile.compute-node rename to compute/compute-node.Dockerfile diff --git a/docs/docker.md b/docs/docker.md index d16311c27b..0914a00082 100644 --- a/docs/docker.md +++ b/docs/docker.md @@ -5,7 +5,7 @@ Currently we build two main images: - [neondatabase/neon](https://hub.docker.com/repository/docker/neondatabase/neon) — image with pre-built `pageserver`, `safekeeper` and `proxy` binaries and all the required runtime dependencies. Built from [/Dockerfile](/Dockerfile). -- [neondatabase/compute-node-v16](https://hub.docker.com/repository/docker/neondatabase/compute-node-v16) — compute node image with pre-built Postgres binaries from [neondatabase/postgres](https://github.com/neondatabase/postgres). Similar images exist for v15 and v14. Built from [/compute-node/Dockerfile](/compute/Dockerfile.compute-node). +- [neondatabase/compute-node-v16](https://hub.docker.com/repository/docker/neondatabase/compute-node-v16) — compute node image with pre-built Postgres binaries from [neondatabase/postgres](https://github.com/neondatabase/postgres). Similar images exist for v15 and v14. Built from [/compute-node/Dockerfile](/compute/compute-node.Dockerfile). And additional intermediate image: @@ -56,7 +56,7 @@ CREATE TABLE postgres=# insert into t values(1, 1); INSERT 0 1 postgres=# select * from t; - key | value + key | value -----+------- 1 | 1 (1 row) @@ -84,4 +84,4 @@ Access http://localhost:9001 and sign in. - Username: `minio` - Password: `password` -You can see durable pages and WAL data in `neon` bucket. \ No newline at end of file +You can see durable pages and WAL data in `neon` bucket. From 163beaf9ad8521ec28d451d1ea884039efcb8897 Mon Sep 17 00:00:00 2001 From: Alexander Bayandin Date: Mon, 21 Oct 2024 12:14:19 +0100 Subject: [PATCH 05/12] CI: use build-tools on Debian 12 whenever we use Neon artifact (#9463) ## Problem ``` + /tmp/neon/pg_install/v16/bin/psql '***' -c 'SELECT version()' /tmp/neon/pg_install/v16/bin/psql: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.33' not found (required by /tmp/neon/pg_install/v16/bin/psql) /tmp/neon/pg_install/v16/bin/psql: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.34' not found (required by /tmp/neon/pg_install/v16/bin/psql) /tmp/neon/pg_install/v16/bin/psql: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.32' not found (required by /tmp/neon/pg_install/v16/lib/libpq.so.5) /tmp/neon/pg_install/v16/bin/psql: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.33' not found (required by /tmp/neon/pg_install/v16/lib/libpq.so.5) /tmp/neon/pg_install/v16/bin/psql: /lib/x86_64-linux-gnu/libc.so.6: version `GLIBC_2.34' not found (required by /tmp/neon/pg_install/v16/lib/libpq.so.5) ``` ## Summary of changes - Use `build-tools:pinned-bookworm` whenever we download Neon artefact --- .../workflows/_benchmarking_preparation.yml | 2 +- .github/workflows/benchmarking.yml | 18 +++++++++--------- .github/workflows/cloud-regress.yml | 2 +- 3 files changed, 11 insertions(+), 11 deletions(-) diff --git a/.github/workflows/_benchmarking_preparation.yml b/.github/workflows/_benchmarking_preparation.yml index d60f97320b..5cdc16f248 100644 --- a/.github/workflows/_benchmarking_preparation.yml +++ b/.github/workflows/_benchmarking_preparation.yml @@ -27,7 +27,7 @@ jobs: runs-on: [ self-hosted, us-east-2, x64 ] container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm credentials: username: ${{ secrets.NEON_DOCKERHUB_USERNAME }} password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }} diff --git a/.github/workflows/benchmarking.yml b/.github/workflows/benchmarking.yml index 32806b89ab..5ccfe48684 100644 --- a/.github/workflows/benchmarking.yml +++ b/.github/workflows/benchmarking.yml @@ -83,7 +83,7 @@ jobs: runs-on: ${{ matrix.RUNNER }} container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm credentials: username: ${{ secrets.NEON_DOCKERHUB_USERNAME }} password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }} @@ -178,7 +178,7 @@ jobs: runs-on: [ self-hosted, us-east-2, x64 ] container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm credentials: username: ${{ secrets.NEON_DOCKERHUB_USERNAME }} password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }} @@ -280,7 +280,7 @@ jobs: region_id_default=${{ env.DEFAULT_REGION_ID }} runner_default='["self-hosted", "us-east-2", "x64"]' runner_azure='["self-hosted", "eastus2", "x64"]' - image_default="neondatabase/build-tools:pinned" + image_default="neondatabase/build-tools:pinned-bookworm" matrix='{ "pg_version" : [ 16 @@ -299,9 +299,9 @@ jobs: "include": [{ "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-freetier", "db_size": "3gb" ,"runner": '"$runner_default"', "image": "'"$image_default"'" }, { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new", "db_size": "10gb","runner": '"$runner_default"', "image": "'"$image_default"'" }, { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-new", "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'" }, - { "pg_version": 16, "region_id": "azure-eastus2", "platform": "neonvm-azure-captest-freetier", "db_size": "3gb" ,"runner": '"$runner_azure"', "image": "neondatabase/build-tools:pinned" }, - { "pg_version": 16, "region_id": "azure-eastus2", "platform": "neonvm-azure-captest-new", "db_size": "10gb","runner": '"$runner_azure"', "image": "neondatabase/build-tools:pinned" }, - { "pg_version": 16, "region_id": "azure-eastus2", "platform": "neonvm-azure-captest-new", "db_size": "50gb","runner": '"$runner_azure"', "image": "neondatabase/build-tools:pinned" }, + { "pg_version": 16, "region_id": "azure-eastus2", "platform": "neonvm-azure-captest-freetier", "db_size": "3gb" ,"runner": '"$runner_azure"', "image": "neondatabase/build-tools:pinned-bookworm" }, + { "pg_version": 16, "region_id": "azure-eastus2", "platform": "neonvm-azure-captest-new", "db_size": "10gb","runner": '"$runner_azure"', "image": "neondatabase/build-tools:pinned-bookworm" }, + { "pg_version": 16, "region_id": "azure-eastus2", "platform": "neonvm-azure-captest-new", "db_size": "50gb","runner": '"$runner_azure"', "image": "neondatabase/build-tools:pinned-bookworm" }, { "pg_version": 16, "region_id": "'"$region_id_default"'", "platform": "neonvm-captest-sharding-reuse", "db_size": "50gb","runner": '"$runner_default"', "image": "'"$image_default"'" }] }' @@ -665,7 +665,7 @@ jobs: runs-on: [ self-hosted, us-east-2, x64 ] container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm credentials: username: ${{ secrets.NEON_DOCKERHUB_USERNAME }} password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }} @@ -772,7 +772,7 @@ jobs: runs-on: [ self-hosted, us-east-2, x64 ] container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm credentials: username: ${{ secrets.NEON_DOCKERHUB_USERNAME }} password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }} @@ -877,7 +877,7 @@ jobs: runs-on: [ self-hosted, us-east-2, x64 ] container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm credentials: username: ${{ secrets.NEON_DOCKERHUB_USERNAME }} password: ${{ secrets.NEON_DOCKERHUB_PASSWORD }} diff --git a/.github/workflows/cloud-regress.yml b/.github/workflows/cloud-regress.yml index ecafe183f8..19ebf457b8 100644 --- a/.github/workflows/cloud-regress.yml +++ b/.github/workflows/cloud-regress.yml @@ -31,7 +31,7 @@ jobs: runs-on: us-east-2 container: - image: neondatabase/build-tools:pinned + image: neondatabase/build-tools:pinned-bookworm options: --init steps: From ababa50cce5e05df4d3d9fcf617a1b2625ed3b4a Mon Sep 17 00:00:00 2001 From: Ivan Efremov Date: Mon, 21 Oct 2024 16:20:39 +0300 Subject: [PATCH 06/12] Use '-f' for make clean in Makefile compute (#9464) Use '-f' instead of '--force' because it is impossible to clean the targets on MacOS --- compute/Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/compute/Makefile b/compute/Makefile index e2896fe390..08e3c7a68b 100644 --- a/compute/Makefile +++ b/compute/Makefile @@ -34,7 +34,7 @@ sql_exporter_autoscaling.yml: $(jsonnet_files) .PHONY: clean clean: - rm --force \ + rm -f \ etc/neon_collector.yml \ etc/neon_collector_autoscaling.yml \ etc/sql_exporter.yml \ From 2dcac94194bedb47e06bc6a98467125e3cfaf07b Mon Sep 17 00:00:00 2001 From: Ivan Efremov Date: Mon, 21 Oct 2024 17:20:09 +0300 Subject: [PATCH 07/12] proxy: Use common error interface for error handling with cplane (#9454) - Remove obsolete error handles. - Use one source of truth for cplane errors. #18468 --- proxy/src/control_plane/messages.rs | 6 ++- proxy/src/control_plane/provider/mod.rs | 32 +------------ proxy/src/metrics.rs | 16 +------ proxy/src/proxy/wake_compute.rs | 62 ++----------------------- 4 files changed, 12 insertions(+), 104 deletions(-) diff --git a/proxy/src/control_plane/messages.rs b/proxy/src/control_plane/messages.rs index dae23f7c53..13a54145b1 100644 --- a/proxy/src/control_plane/messages.rs +++ b/proxy/src/control_plane/messages.rs @@ -161,6 +161,9 @@ pub(crate) enum Reason { /// LockAlreadyTaken indicates that the we attempted to take a lock that was already taken. #[serde(rename = "LOCK_ALREADY_TAKEN")] LockAlreadyTaken, + /// ActiveEndpointsLimitExceeded indicates that the limit of concurrently active endpoints was exceeded. + #[serde(rename = "ACTIVE_ENDPOINTS_LIMIT_EXCEEDED")] + ActiveEndpointsLimitExceeded, #[default] #[serde(other)] Unknown, @@ -194,7 +197,8 @@ impl Reason { | Reason::ComputeTimeQuotaExceeded | Reason::WrittenDataQuotaExceeded | Reason::DataTransferQuotaExceeded - | Reason::LogicalSizeQuotaExceeded => false, + | Reason::LogicalSizeQuotaExceeded + | Reason::ActiveEndpointsLimitExceeded => false, // transitive error. control plane is currently busy // but might be ready soon Reason::RunningOperations diff --git a/proxy/src/control_plane/provider/mod.rs b/proxy/src/control_plane/provider/mod.rs index a4a330cd5f..88399dffa8 100644 --- a/proxy/src/control_plane/provider/mod.rs +++ b/proxy/src/control_plane/provider/mod.rs @@ -87,36 +87,8 @@ pub(crate) mod errors { Reason::ConcurrencyLimitReached => ErrorKind::ControlPlane, Reason::LockAlreadyTaken => ErrorKind::ControlPlane, Reason::RunningOperations => ErrorKind::ControlPlane, - Reason::Unknown => match &**e { - ControlPlaneError { - http_status_code: - http::StatusCode::NOT_FOUND | http::StatusCode::NOT_ACCEPTABLE, - .. - } => crate::error::ErrorKind::User, - ControlPlaneError { - http_status_code: http::StatusCode::UNPROCESSABLE_ENTITY, - error, - .. - } if error - .contains("compute time quota of non-primary branches is exceeded") => - { - crate::error::ErrorKind::Quota - } - ControlPlaneError { - http_status_code: http::StatusCode::LOCKED, - error, - .. - } if error.contains("quota exceeded") - || error.contains("the limit for current plan reached") => - { - crate::error::ErrorKind::Quota - } - ControlPlaneError { - http_status_code: http::StatusCode::TOO_MANY_REQUESTS, - .. - } => crate::error::ErrorKind::ServiceRateLimit, - ControlPlaneError { .. } => crate::error::ErrorKind::ControlPlane, - }, + Reason::ActiveEndpointsLimitExceeded => ErrorKind::ControlPlane, + Reason::Unknown => ErrorKind::ControlPlane, }, ApiError::Transport(_) => crate::error::ErrorKind::ControlPlane, } diff --git a/proxy/src/metrics.rs b/proxy/src/metrics.rs index 542826e833..f91fcd4120 100644 --- a/proxy/src/metrics.rs +++ b/proxy/src/metrics.rs @@ -14,6 +14,7 @@ use metrics::{CounterPairAssoc, CounterPairVec, HyperLogLog, HyperLogLogVec}; use tokio::time::{self, Instant}; use crate::control_plane::messages::ColdStartInfo; +use crate::error::ErrorKind; #[derive(MetricGroup)] #[metric(new(thread_pool: Arc))] @@ -325,23 +326,10 @@ pub enum ConnectionFailureKind { ComputeUncached, } -#[derive(FixedCardinalityLabel, Copy, Clone)] -#[label(singleton = "kind")] -pub enum WakeupFailureKind { - BadComputeAddress, - ApiTransportError, - QuotaExceeded, - ApiConsoleLocked, - ApiConsoleBadRequest, - ApiConsoleOtherServerError, - ApiConsoleOtherError, - TimeoutError, -} - #[derive(LabelGroup)] #[label(set = ConnectionFailuresBreakdownSet)] pub struct ConnectionFailuresBreakdownGroup { - pub kind: WakeupFailureKind, + pub kind: ErrorKind, pub retry: Bool, } diff --git a/proxy/src/proxy/wake_compute.rs b/proxy/src/proxy/wake_compute.rs index 9dfa485fa4..4e61094264 100644 --- a/proxy/src/proxy/wake_compute.rs +++ b/proxy/src/proxy/wake_compute.rs @@ -1,15 +1,13 @@ -use hyper::StatusCode; use tracing::{error, info, warn}; use super::connect_compute::ComputeConnectBackend; use crate::config::RetryConfig; use crate::context::RequestMonitoring; use crate::control_plane::errors::WakeComputeError; -use crate::control_plane::messages::{ControlPlaneError, Reason}; use crate::control_plane::provider::CachedNodeInfo; +use crate::error::ReportableError; use crate::metrics::{ ConnectOutcome, ConnectionFailuresBreakdownGroup, Metrics, RetriesMetricGroup, RetryType, - WakeupFailureKind, }; use crate::proxy::retry::{retry_after, should_retry}; @@ -60,62 +58,8 @@ pub(crate) async fn wake_compute( } fn report_error(e: &WakeComputeError, retry: bool) { - use crate::control_plane::errors::ApiError; - let kind = match e { - WakeComputeError::BadComputeAddress(_) => WakeupFailureKind::BadComputeAddress, - WakeComputeError::ApiError(ApiError::Transport(_)) => WakeupFailureKind::ApiTransportError, - WakeComputeError::ApiError(ApiError::ControlPlane(e)) => match e.get_reason() { - Reason::RoleProtected => WakeupFailureKind::ApiConsoleBadRequest, - Reason::ResourceNotFound => WakeupFailureKind::ApiConsoleBadRequest, - Reason::ProjectNotFound => WakeupFailureKind::ApiConsoleBadRequest, - Reason::EndpointNotFound => WakeupFailureKind::ApiConsoleBadRequest, - Reason::BranchNotFound => WakeupFailureKind::ApiConsoleBadRequest, - Reason::RateLimitExceeded => WakeupFailureKind::ApiConsoleLocked, - Reason::NonDefaultBranchComputeTimeExceeded => WakeupFailureKind::QuotaExceeded, - Reason::ActiveTimeQuotaExceeded => WakeupFailureKind::QuotaExceeded, - Reason::ComputeTimeQuotaExceeded => WakeupFailureKind::QuotaExceeded, - Reason::WrittenDataQuotaExceeded => WakeupFailureKind::QuotaExceeded, - Reason::DataTransferQuotaExceeded => WakeupFailureKind::QuotaExceeded, - Reason::LogicalSizeQuotaExceeded => WakeupFailureKind::QuotaExceeded, - Reason::ConcurrencyLimitReached => WakeupFailureKind::ApiConsoleLocked, - Reason::LockAlreadyTaken => WakeupFailureKind::ApiConsoleLocked, - Reason::RunningOperations => WakeupFailureKind::ApiConsoleLocked, - Reason::Unknown => match **e { - ControlPlaneError { - http_status_code: StatusCode::LOCKED, - ref error, - .. - } if error.contains("written data quota exceeded") - || error.contains("the limit for current plan reached") => - { - WakeupFailureKind::QuotaExceeded - } - ControlPlaneError { - http_status_code: StatusCode::UNPROCESSABLE_ENTITY, - ref error, - .. - } if error.contains("compute time quota of non-primary branches is exceeded") => { - WakeupFailureKind::QuotaExceeded - } - ControlPlaneError { - http_status_code: StatusCode::LOCKED, - .. - } => WakeupFailureKind::ApiConsoleLocked, - ControlPlaneError { - http_status_code: StatusCode::BAD_REQUEST, - .. - } => WakeupFailureKind::ApiConsoleBadRequest, - ControlPlaneError { - http_status_code, .. - } if http_status_code.is_server_error() => { - WakeupFailureKind::ApiConsoleOtherServerError - } - ControlPlaneError { .. } => WakeupFailureKind::ApiConsoleOtherError, - }, - }, - WakeComputeError::TooManyConnections => WakeupFailureKind::ApiConsoleLocked, - WakeComputeError::TooManyConnectionAttempts(_) => WakeupFailureKind::TimeoutError, - }; + let kind = e.get_error_kind(); + Metrics::get() .proxy .connection_failures_breakdown From aca81f5fa4f3e0f882a9b0d55eef1cdee8ffc168 Mon Sep 17 00:00:00 2001 From: "Alex Chi Z." <4198311+skyzh@users.noreply.github.com> Date: Mon, 21 Oct 2024 10:59:48 -0400 Subject: [PATCH 08/12] fix(pageserver): make image split layer writer finish atomic (#8841) Part of https://github.com/neondatabase/neon/issues/8836 ## Summary of changes This pull request makes the image layer split writer atomic when finishing the layers. All the produced layers either finish at the same time, or discard at the same time. Note that this does not prevent atomicity when crash, but anyways, it will be cleaned up on pageserver restart. --------- Signed-off-by: Alex Chi Z Co-authored-by: Christian Schwarz --- .../src/tenant/storage_layer/delta_layer.rs | 4 +- .../src/tenant/storage_layer/image_layer.rs | 19 ++ .../src/tenant/storage_layer/split_writer.rs | 231 ++++++++++-------- pageserver/src/tenant/timeline/compaction.rs | 7 +- 4 files changed, 149 insertions(+), 112 deletions(-) diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index d1079876f8..6332d36dc3 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -515,8 +515,8 @@ impl DeltaLayerWriterInner { ) -> anyhow::Result<(PersistentLayerDesc, Utf8PathBuf)> { let temp_path = self.path.clone(); let result = self.finish0(key_end, ctx).await; - if result.is_err() { - tracing::info!(%temp_path, "cleaning up temporary file after error during writing"); + if let Err(ref e) = result { + tracing::info!(%temp_path, "cleaning up temporary file after error during writing: {e}"); if let Err(e) = std::fs::remove_file(&temp_path) { tracing::warn!(error=%e, %temp_path, "error cleaning up temporary layer file after error during writing"); } diff --git a/pageserver/src/tenant/storage_layer/image_layer.rs b/pageserver/src/tenant/storage_layer/image_layer.rs index 6c1a943470..b1f2557038 100644 --- a/pageserver/src/tenant/storage_layer/image_layer.rs +++ b/pageserver/src/tenant/storage_layer/image_layer.rs @@ -827,6 +827,25 @@ impl ImageLayerWriterInner { self, ctx: &RequestContext, end_key: Option, + ) -> anyhow::Result<(PersistentLayerDesc, Utf8PathBuf)> { + let temp_path = self.path.clone(); + let result = self.finish0(ctx, end_key).await; + if let Err(ref e) = result { + tracing::info!(%temp_path, "cleaning up temporary file after error during writing: {e}"); + if let Err(e) = std::fs::remove_file(&temp_path) { + tracing::warn!(error=%e, %temp_path, "error cleaning up temporary layer file after error during writing"); + } + } + result + } + + /// + /// Finish writing the image layer. + /// + async fn finish0( + self, + ctx: &RequestContext, + end_key: Option, ) -> anyhow::Result<(PersistentLayerDesc, Utf8PathBuf)> { let index_start_blk = self.blob_writer.size().div_ceil(PAGE_SZ as u64) as u32; diff --git a/pageserver/src/tenant/storage_layer/split_writer.rs b/pageserver/src/tenant/storage_layer/split_writer.rs index b499a0eef4..5bd9a47e2b 100644 --- a/pageserver/src/tenant/storage_layer/split_writer.rs +++ b/pageserver/src/tenant/storage_layer/split_writer.rs @@ -42,7 +42,7 @@ impl SplitWriterResult { pub struct SplitImageLayerWriter { inner: ImageLayerWriter, target_layer_size: u64, - generated_layers: Vec, + generated_layer_writers: Vec<(ImageLayerWriter, PersistentLayerKey)>, conf: &'static PageServerConf, timeline_id: TimelineId, tenant_shard_id: TenantShardId, @@ -71,7 +71,7 @@ impl SplitImageLayerWriter { ctx, ) .await?, - generated_layers: Vec::new(), + generated_layer_writers: Vec::new(), conf, timeline_id, tenant_shard_id, @@ -80,18 +80,12 @@ impl SplitImageLayerWriter { }) } - pub async fn put_image_with_discard_fn( + pub async fn put_image( &mut self, key: Key, img: Bytes, - tline: &Arc, ctx: &RequestContext, - discard: D, - ) -> anyhow::Result<()> - where - D: FnOnce(&PersistentLayerKey) -> F, - F: Future, - { + ) -> anyhow::Result<()> { // The current estimation is an upper bound of the space that the key/image could take // because we did not consider compression in this estimation. The resulting image layer // could be smaller than the target size. @@ -108,72 +102,83 @@ impl SplitImageLayerWriter { ctx, ) .await?; - let prev_image_writer = std::mem::replace(&mut self.inner, next_image_writer); let layer_key = PersistentLayerKey { key_range: self.start_key..key, lsn_range: PersistentLayerDesc::image_layer_lsn_range(self.lsn), is_delta: false, }; + let prev_image_writer = std::mem::replace(&mut self.inner, next_image_writer); self.start_key = key; - if discard(&layer_key).await { - drop(prev_image_writer); - self.generated_layers - .push(SplitWriterResult::Discarded(layer_key)); - } else { - let (desc, path) = prev_image_writer.finish_with_end_key(key, ctx).await?; - - let layer = Layer::finish_creating(self.conf, tline, desc, &path)?; - self.generated_layers - .push(SplitWriterResult::Produced(layer)); - } + self.generated_layer_writers + .push((prev_image_writer, layer_key)); } self.inner.put_image(key, img, ctx).await } - #[cfg(test)] - pub async fn put_image( - &mut self, - key: Key, - img: Bytes, - tline: &Arc, - ctx: &RequestContext, - ) -> anyhow::Result<()> { - self.put_image_with_discard_fn(key, img, tline, ctx, |_| async { false }) - .await - } - pub(crate) async fn finish_with_discard_fn( self, tline: &Arc, ctx: &RequestContext, end_key: Key, - discard: D, + discard_fn: D, ) -> anyhow::Result> where - D: FnOnce(&PersistentLayerKey) -> F, + D: Fn(&PersistentLayerKey) -> F, F: Future, { let Self { - mut generated_layers, + mut generated_layer_writers, inner, .. } = self; - if inner.num_keys() == 0 { - return Ok(generated_layers); + if inner.num_keys() != 0 { + let layer_key = PersistentLayerKey { + key_range: self.start_key..end_key, + lsn_range: PersistentLayerDesc::image_layer_lsn_range(self.lsn), + is_delta: false, + }; + generated_layer_writers.push((inner, layer_key)); } - let layer_key = PersistentLayerKey { - key_range: self.start_key..end_key, - lsn_range: PersistentLayerDesc::image_layer_lsn_range(self.lsn), - is_delta: false, + let clean_up_layers = |generated_layers: Vec| { + for produced_layer in generated_layers { + if let SplitWriterResult::Produced(image_layer) = produced_layer { + let layer: Layer = image_layer.into(); + layer.delete_on_drop(); + } + } }; - if discard(&layer_key).await { - generated_layers.push(SplitWriterResult::Discarded(layer_key)); - } else { - let (desc, path) = inner.finish_with_end_key(end_key, ctx).await?; - let layer = Layer::finish_creating(self.conf, tline, desc, &path)?; - generated_layers.push(SplitWriterResult::Produced(layer)); + // BEGIN: catch every error and do the recovery in the below section + let mut generated_layers = Vec::new(); + for (inner, layer_key) in generated_layer_writers { + if discard_fn(&layer_key).await { + generated_layers.push(SplitWriterResult::Discarded(layer_key)); + } else { + let layer = match inner + .finish_with_end_key(layer_key.key_range.end, ctx) + .await + { + Ok((desc, path)) => { + match Layer::finish_creating(self.conf, tline, desc, &path) { + Ok(layer) => layer, + Err(e) => { + tokio::fs::remove_file(&path).await.ok(); + clean_up_layers(generated_layers); + return Err(e); + } + } + } + Err(e) => { + // ImageLayerWriter::finish will clean up the temporary layer if anything goes wrong, + // so we don't need to remove it by ourselves. + clean_up_layers(generated_layers); + return Err(e); + } + }; + generated_layers.push(SplitWriterResult::Produced(layer)); + } } + // END: catch every error and do the recovery in the above section Ok(generated_layers) } @@ -187,11 +192,6 @@ impl SplitImageLayerWriter { self.finish_with_discard_fn(tline, ctx, end_key, |_| async { false }) .await } - - /// This function will be deprecated with #8841. - pub(crate) fn take(self) -> anyhow::Result<(Vec, ImageLayerWriter)> { - Ok((self.generated_layers, self.inner)) - } } /// A delta writer that takes key-lsn-values and produces multiple delta layers. @@ -296,8 +296,16 @@ impl SplitDeltaLayerWriter { self.generated_layers .push(SplitWriterResult::Discarded(layer_key)); } else { + // `finish` will remove the file if anything goes wrong, while we need to handle deleting temporary + // files for `finish_creating`. let (desc, path) = prev_delta_writer.finish(key, ctx).await?; - let delta_layer = Layer::finish_creating(self.conf, tline, desc, &path)?; + let delta_layer = match Layer::finish_creating(self.conf, tline, desc, &path) { + Ok(layer) => layer, + Err(e) => { + tokio::fs::remove_file(&path).await.ok(); + return Err(e); + } + }; self.generated_layers .push(SplitWriterResult::Produced(delta_layer)); } @@ -357,8 +365,16 @@ impl SplitDeltaLayerWriter { if discard(&layer_key).await { generated_layers.push(SplitWriterResult::Discarded(layer_key)); } else { + // `finish` will remove the file if anything goes wrong, while we need to handle deleting temporary + // files for `finish_creating`. let (desc, path) = inner.finish(end_key, ctx).await?; - let delta_layer = Layer::finish_creating(self.conf, tline, desc, &path)?; + let delta_layer = match Layer::finish_creating(self.conf, tline, desc, &path) { + Ok(layer) => layer, + Err(e) => { + tokio::fs::remove_file(&path).await.ok(); + return Err(e); + } + }; generated_layers.push(SplitWriterResult::Produced(delta_layer)); } Ok(generated_layers) @@ -447,7 +463,7 @@ mod tests { .unwrap(); image_writer - .put_image(get_key(0), get_img(0), &tline, &ctx) + .put_image(get_key(0), get_img(0), &ctx) .await .unwrap(); let layers = image_writer @@ -486,14 +502,18 @@ mod tests { #[tokio::test] async fn write_split() { + // Test the split writer with retaining all the layers we have produced (discard=false) write_split_helper("split_writer_write_split", false).await; } #[tokio::test] async fn write_split_discard() { - write_split_helper("split_writer_write_split_discard", false).await; + // Test the split writer with discarding all the layers we have produced (discard=true) + write_split_helper("split_writer_write_split_discard", true).await; } + /// Test the image+delta writer by writing a large number of images and deltas. If discard is + /// set to true, all layers will be discarded. async fn write_split_helper(harness_name: &'static str, discard: bool) { let harness = TenantHarness::create(harness_name).await.unwrap(); let (tenant, ctx) = harness.load().await; @@ -527,9 +547,7 @@ mod tests { for i in 0..N { let i = i as u32; image_writer - .put_image_with_discard_fn(get_key(i), get_large_img(), &tline, &ctx, |_| async { - discard - }) + .put_image(get_key(i), get_large_img(), &ctx) .await .unwrap(); delta_writer @@ -545,51 +563,54 @@ mod tests { .unwrap(); } let image_layers = image_writer - .finish(&tline, &ctx, get_key(N as u32)) + .finish_with_discard_fn(&tline, &ctx, get_key(N as u32), |_| async { discard }) .await .unwrap(); - let delta_layers = delta_writer.finish(&tline, &ctx).await.unwrap(); - if discard { - for layer in image_layers { - layer.into_discarded_layer(); - } - for layer in delta_layers { - layer.into_discarded_layer(); - } - } else { - let image_layers = image_layers - .into_iter() - .map(|x| x.into_resident_layer()) - .collect_vec(); - let delta_layers = delta_layers - .into_iter() - .map(|x| x.into_resident_layer()) - .collect_vec(); - assert_eq!(image_layers.len(), N / 512 + 1); - assert_eq!(delta_layers.len(), N / 512 + 1); - assert_eq!( - delta_layers.first().unwrap().layer_desc().key_range.start, - get_key(0) - ); - assert_eq!( - delta_layers.last().unwrap().layer_desc().key_range.end, - get_key(N as u32) - ); - for idx in 0..image_layers.len() { - assert_ne!(image_layers[idx].layer_desc().key_range.start, Key::MIN); - assert_ne!(image_layers[idx].layer_desc().key_range.end, Key::MAX); - assert_ne!(delta_layers[idx].layer_desc().key_range.start, Key::MIN); - assert_ne!(delta_layers[idx].layer_desc().key_range.end, Key::MAX); - if idx > 0 { - assert_eq!( - image_layers[idx - 1].layer_desc().key_range.end, - image_layers[idx].layer_desc().key_range.start - ); - assert_eq!( - delta_layers[idx - 1].layer_desc().key_range.end, - delta_layers[idx].layer_desc().key_range.start - ); + let delta_layers = delta_writer + .finish_with_discard_fn(&tline, &ctx, |_| async { discard }) + .await + .unwrap(); + let image_layers = image_layers + .into_iter() + .map(|x| { + if discard { + x.into_discarded_layer() + } else { + x.into_resident_layer().layer_desc().key() } + }) + .collect_vec(); + let delta_layers = delta_layers + .into_iter() + .map(|x| { + if discard { + x.into_discarded_layer() + } else { + x.into_resident_layer().layer_desc().key() + } + }) + .collect_vec(); + assert_eq!(image_layers.len(), N / 512 + 1); + assert_eq!(delta_layers.len(), N / 512 + 1); + assert_eq!(delta_layers.first().unwrap().key_range.start, get_key(0)); + assert_eq!( + delta_layers.last().unwrap().key_range.end, + get_key(N as u32) + ); + for idx in 0..image_layers.len() { + assert_ne!(image_layers[idx].key_range.start, Key::MIN); + assert_ne!(image_layers[idx].key_range.end, Key::MAX); + assert_ne!(delta_layers[idx].key_range.start, Key::MIN); + assert_ne!(delta_layers[idx].key_range.end, Key::MAX); + if idx > 0 { + assert_eq!( + image_layers[idx - 1].key_range.end, + image_layers[idx].key_range.start + ); + assert_eq!( + delta_layers[idx - 1].key_range.end, + delta_layers[idx].key_range.start + ); } } } @@ -629,11 +650,11 @@ mod tests { .unwrap(); image_writer - .put_image(get_key(0), get_img(0), &tline, &ctx) + .put_image(get_key(0), get_img(0), &ctx) .await .unwrap(); image_writer - .put_image(get_key(1), get_large_img(), &tline, &ctx) + .put_image(get_key(1), get_large_img(), &ctx) .await .unwrap(); let layers = image_writer diff --git a/pageserver/src/tenant/timeline/compaction.rs b/pageserver/src/tenant/timeline/compaction.rs index 5588363330..5cb1460b29 100644 --- a/pageserver/src/tenant/timeline/compaction.rs +++ b/pageserver/src/tenant/timeline/compaction.rs @@ -141,9 +141,7 @@ impl KeyHistoryRetention { }; stat.produce_image_key(img); if let Some(image_writer) = image_writer.as_mut() { - image_writer - .put_image_with_discard_fn(key, img.clone(), tline, ctx, discard) - .await?; + image_writer.put_image(key, img.clone(), ctx).await?; } else { delta_writer .put_value_with_discard_fn( @@ -2041,8 +2039,7 @@ impl Timeline { .finish_with_discard_fn(self, ctx, Key::MAX, discard) .await? } else { - let (layers, _) = writer.take()?; - assert!(layers.is_empty(), "image layers produced in dry run mode?"); + drop(writer); Vec::new() } } else { From 49d5e56c084f1fd694cde75d56c2d8ed9049c06e Mon Sep 17 00:00:00 2001 From: Yuchen Liang <70461588+yliang412@users.noreply.github.com> Date: Mon, 21 Oct 2024 11:01:25 -0400 Subject: [PATCH 09/12] pageserver: use direct IO for delta and image layer reads (#9326) Part of #8130 ## Problem Pageserver previously goes through the kernel page cache for all the IOs. The kernel page cache makes light-loaded pageserver have deceptive fast performance. Using direct IO would offer predictable latencies of our virtual file IO operations. In particular for reads, the data pages also have an extremely low temporal locality because the most frequently accessed pages are cached on the compute side. ## Summary of changes This PR enables pageserver to use direct IO for delta layer and image layer reads. We can ship them separately because these layers are write-once, read-many, so we will not be mixing buffered IO with direct IO. - implement `IoBufferMut`, an buffer type with aligned allocation (currently set to 512). - use `IoBufferMut` at all places we are doing reads on image + delta layers. - leverage Rust type system and use `IoBufAlignedMut` marker trait to guarantee that the input buffers for the IO operations are aligned. - page cache allocation is also made aligned. _* in-memory layer reads and the write path will be shipped separately._ ## Testing Integration test suite run with O_DIRECT enabled: https://github.com/neondatabase/neon/pull/9350 ## Performance We evaluated performance based on the `get-page-at-latest-lsn` benchmark. The results demonstrate a decrease in the number of IOps, no sigificant change in the latency mean, and an slight improvement on the p99.9 and p99.99 latencies. [Benchmark](https://www.notion.so/neondatabase/Benchmark-O_DIRECT-for-image-and-delta-layers-2024-10-01-112f189e00478092a195ea5a0137e706?pvs=4) ## Rollout We will add `virtual_file_io_mode=direct` region by region to enable direct IO on image + delta layers. Signed-off-by: Yuchen Liang --- pageserver/benches/bench_ingest.rs | 6 +- pageserver/ctl/src/layer_map_analyzer.rs | 7 +- pageserver/ctl/src/layers.rs | 13 +- pageserver/ctl/src/main.rs | 8 +- pageserver/src/bin/pageserver.rs | 6 +- pageserver/src/page_cache.rs | 16 +- pageserver/src/tenant/block_io.rs | 6 +- pageserver/src/tenant/ephemeral_file.rs | 28 +- .../src/tenant/storage_layer/delta_layer.rs | 15 +- .../src/tenant/storage_layer/image_layer.rs | 19 +- .../tenant/storage_layer/inmemory_layer.rs | 8 +- .../inmemory_layer/vectored_dio_read.rs | 23 +- pageserver/src/tenant/vectored_blob_io.rs | 9 +- pageserver/src/virtual_file.rs | 40 +- .../owned_buffers_io/aligned_buffer.rs | 9 + .../aligned_buffer/alignment.rs | 26 ++ .../owned_buffers_io/aligned_buffer/buffer.rs | 124 +++++++ .../aligned_buffer/buffer_mut.rs | 347 ++++++++++++++++++ .../owned_buffers_io/aligned_buffer/raw.rs | 216 +++++++++++ .../owned_buffers_io/aligned_buffer/slice.rs | 40 ++ .../owned_buffers_io/io_buf_aligned.rs | 9 + .../owned_buffers_io/io_buf_ext.rs | 3 + 22 files changed, 899 insertions(+), 79 deletions(-) create mode 100644 pageserver/src/virtual_file/owned_buffers_io/aligned_buffer.rs create mode 100644 pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/alignment.rs create mode 100644 pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer.rs create mode 100644 pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer_mut.rs create mode 100644 pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/raw.rs create mode 100644 pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/slice.rs create mode 100644 pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs diff --git a/pageserver/benches/bench_ingest.rs b/pageserver/benches/bench_ingest.rs index 821c8008a9..d98b23acce 100644 --- a/pageserver/benches/bench_ingest.rs +++ b/pageserver/benches/bench_ingest.rs @@ -164,7 +164,11 @@ fn criterion_benchmark(c: &mut Criterion) { let conf: &'static PageServerConf = Box::leak(Box::new( pageserver::config::PageServerConf::dummy_conf(temp_dir.path().to_path_buf()), )); - virtual_file::init(16384, virtual_file::io_engine_for_bench()); + virtual_file::init( + 16384, + virtual_file::io_engine_for_bench(), + conf.virtual_file_io_mode, + ); page_cache::init(conf.page_cache_size); { diff --git a/pageserver/ctl/src/layer_map_analyzer.rs b/pageserver/ctl/src/layer_map_analyzer.rs index 151b94cf62..7dd2a5d05c 100644 --- a/pageserver/ctl/src/layer_map_analyzer.rs +++ b/pageserver/ctl/src/layer_map_analyzer.rs @@ -7,6 +7,7 @@ use camino::{Utf8Path, Utf8PathBuf}; use pageserver::context::{DownloadBehavior, RequestContext}; use pageserver::task_mgr::TaskKind; use pageserver::tenant::{TENANTS_SEGMENT_NAME, TIMELINES_SEGMENT_NAME}; +use pageserver::virtual_file::api::IoMode; use std::cmp::Ordering; use std::collections::BinaryHeap; use std::ops::Range; @@ -152,7 +153,11 @@ pub(crate) async fn main(cmd: &AnalyzeLayerMapCmd) -> Result<()> { let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error); // Initialize virtual_file (file desriptor cache) and page cache which are needed to access layer persistent B-Tree. - pageserver::virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs); + pageserver::virtual_file::init( + 10, + virtual_file::api::IoEngineKind::StdFs, + IoMode::preferred(), + ); pageserver::page_cache::init(100); let mut total_delta_layers = 0usize; diff --git a/pageserver/ctl/src/layers.rs b/pageserver/ctl/src/layers.rs index fd948bf2ef..c0b2b6ae89 100644 --- a/pageserver/ctl/src/layers.rs +++ b/pageserver/ctl/src/layers.rs @@ -11,6 +11,7 @@ use pageserver::tenant::storage_layer::delta_layer::{BlobRef, Summary}; use pageserver::tenant::storage_layer::{delta_layer, image_layer}; use pageserver::tenant::storage_layer::{DeltaLayer, ImageLayer}; use pageserver::tenant::{TENANTS_SEGMENT_NAME, TIMELINES_SEGMENT_NAME}; +use pageserver::virtual_file::api::IoMode; use pageserver::{page_cache, virtual_file}; use pageserver::{ repository::{Key, KEY_SIZE}, @@ -59,7 +60,11 @@ pub(crate) enum LayerCmd { async fn read_delta_file(path: impl AsRef, ctx: &RequestContext) -> Result<()> { let path = Utf8Path::from_path(path.as_ref()).expect("non-Unicode path"); - virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs); + virtual_file::init( + 10, + virtual_file::api::IoEngineKind::StdFs, + IoMode::preferred(), + ); page_cache::init(100); let file = VirtualFile::open(path, ctx).await?; let file_id = page_cache::next_file_id(); @@ -190,7 +195,11 @@ pub(crate) async fn main(cmd: &LayerCmd) -> Result<()> { new_tenant_id, new_timeline_id, } => { - pageserver::virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs); + pageserver::virtual_file::init( + 10, + virtual_file::api::IoEngineKind::StdFs, + IoMode::preferred(), + ); pageserver::page_cache::init(100); let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error); diff --git a/pageserver/ctl/src/main.rs b/pageserver/ctl/src/main.rs index c96664d346..f506caec5b 100644 --- a/pageserver/ctl/src/main.rs +++ b/pageserver/ctl/src/main.rs @@ -24,7 +24,7 @@ use pageserver::{ page_cache, task_mgr::TaskKind, tenant::{dump_layerfile_from_path, metadata::TimelineMetadata}, - virtual_file, + virtual_file::{self, api::IoMode}, }; use pageserver_api::shard::TenantShardId; use postgres_ffi::ControlFileData; @@ -205,7 +205,11 @@ fn read_pg_control_file(control_file_path: &Utf8Path) -> anyhow::Result<()> { async fn print_layerfile(path: &Utf8Path) -> anyhow::Result<()> { // Basic initialization of things that don't change after startup - virtual_file::init(10, virtual_file::api::IoEngineKind::StdFs); + virtual_file::init( + 10, + virtual_file::api::IoEngineKind::StdFs, + IoMode::preferred(), + ); page_cache::init(100); let ctx = RequestContext::new(TaskKind::DebugTool, DownloadBehavior::Error); dump_layerfile_from_path(path, true, &ctx).await diff --git a/pageserver/src/bin/pageserver.rs b/pageserver/src/bin/pageserver.rs index f71a3d2653..c6659345f9 100644 --- a/pageserver/src/bin/pageserver.rs +++ b/pageserver/src/bin/pageserver.rs @@ -167,7 +167,11 @@ fn main() -> anyhow::Result<()> { let scenario = failpoint_support::init(); // Basic initialization of things that don't change after startup - virtual_file::init(conf.max_file_descriptors, conf.virtual_file_io_engine); + virtual_file::init( + conf.max_file_descriptors, + conf.virtual_file_io_engine, + conf.virtual_file_io_mode, + ); page_cache::init(conf.page_cache_size); start_pageserver(launch_ts, conf).context("Failed to start pageserver")?; diff --git a/pageserver/src/page_cache.rs b/pageserver/src/page_cache.rs index f386c825b8..45bf02362a 100644 --- a/pageserver/src/page_cache.rs +++ b/pageserver/src/page_cache.rs @@ -82,6 +82,7 @@ use once_cell::sync::OnceCell; use crate::{ context::RequestContext, metrics::{page_cache_eviction_metrics, PageCacheSizeMetrics}, + virtual_file::{IoBufferMut, IoPageSlice}, }; static PAGE_CACHE: OnceCell = OnceCell::new(); @@ -144,7 +145,7 @@ struct SlotInner { key: Option, // for `coalesce_readers_permit` permit: std::sync::Mutex>, - buf: &'static mut [u8; PAGE_SZ], + buf: IoPageSlice<'static>, } impl Slot { @@ -234,13 +235,13 @@ impl std::ops::Deref for PageReadGuard<'_> { type Target = [u8; PAGE_SZ]; fn deref(&self) -> &Self::Target { - self.slot_guard.buf + self.slot_guard.buf.deref() } } impl AsRef<[u8; PAGE_SZ]> for PageReadGuard<'_> { fn as_ref(&self) -> &[u8; PAGE_SZ] { - self.slot_guard.buf + self.slot_guard.buf.as_ref() } } @@ -266,7 +267,7 @@ enum PageWriteGuardState<'i> { impl std::ops::DerefMut for PageWriteGuard<'_> { fn deref_mut(&mut self) -> &mut Self::Target { match &mut self.state { - PageWriteGuardState::Invalid { inner, _permit } => inner.buf, + PageWriteGuardState::Invalid { inner, _permit } => inner.buf.deref_mut(), PageWriteGuardState::Downgraded => unreachable!(), } } @@ -277,7 +278,7 @@ impl std::ops::Deref for PageWriteGuard<'_> { fn deref(&self) -> &Self::Target { match &self.state { - PageWriteGuardState::Invalid { inner, _permit } => inner.buf, + PageWriteGuardState::Invalid { inner, _permit } => inner.buf.deref(), PageWriteGuardState::Downgraded => unreachable!(), } } @@ -643,7 +644,7 @@ impl PageCache { // We could use Vec::leak here, but that potentially also leaks // uninitialized reserved capacity. With into_boxed_slice and Box::leak // this is avoided. - let page_buffer = Box::leak(vec![0u8; num_pages * PAGE_SZ].into_boxed_slice()); + let page_buffer = IoBufferMut::with_capacity_zeroed(num_pages * PAGE_SZ).leak(); let size_metrics = &crate::metrics::PAGE_CACHE_SIZE; size_metrics.max_bytes.set_page_sz(num_pages); @@ -652,7 +653,8 @@ impl PageCache { let slots = page_buffer .chunks_exact_mut(PAGE_SZ) .map(|chunk| { - let buf: &mut [u8; PAGE_SZ] = chunk.try_into().unwrap(); + // SAFETY: Each chunk has `PAGE_SZ` (8192) bytes, greater than 512, still aligned. + let buf = unsafe { IoPageSlice::new_unchecked(chunk.try_into().unwrap()) }; Slot { inner: tokio::sync::RwLock::new(SlotInner { diff --git a/pageserver/src/tenant/block_io.rs b/pageserver/src/tenant/block_io.rs index 1c82e5454d..2bd7f2d619 100644 --- a/pageserver/src/tenant/block_io.rs +++ b/pageserver/src/tenant/block_io.rs @@ -5,6 +5,8 @@ use super::storage_layer::delta_layer::{Adapter, DeltaLayerInner}; use crate::context::RequestContext; use crate::page_cache::{self, FileId, PageReadGuard, PageWriteGuard, ReadBufResult, PAGE_SZ}; +#[cfg(test)] +use crate::virtual_file::IoBufferMut; use crate::virtual_file::VirtualFile; use bytes::Bytes; use std::ops::Deref; @@ -40,7 +42,7 @@ pub enum BlockLease<'a> { #[cfg(test)] Arc(std::sync::Arc<[u8; PAGE_SZ]>), #[cfg(test)] - Vec(Vec), + IoBufferMut(IoBufferMut), } impl From> for BlockLease<'static> { @@ -67,7 +69,7 @@ impl Deref for BlockLease<'_> { #[cfg(test)] BlockLease::Arc(v) => v.deref(), #[cfg(test)] - BlockLease::Vec(v) => { + BlockLease::IoBufferMut(v) => { TryFrom::try_from(&v[..]).expect("caller must ensure that v has PAGE_SZ") } } diff --git a/pageserver/src/tenant/ephemeral_file.rs b/pageserver/src/tenant/ephemeral_file.rs index a62a47f9a7..de0abab4c0 100644 --- a/pageserver/src/tenant/ephemeral_file.rs +++ b/pageserver/src/tenant/ephemeral_file.rs @@ -6,10 +6,11 @@ use crate::config::PageServerConf; use crate::context::RequestContext; use crate::page_cache; use crate::tenant::storage_layer::inmemory_layer::vectored_dio_read::File; +use crate::virtual_file::owned_buffers_io::io_buf_aligned::IoBufAlignedMut; use crate::virtual_file::owned_buffers_io::slice::SliceMutExt; use crate::virtual_file::owned_buffers_io::util::size_tracking_writer; use crate::virtual_file::owned_buffers_io::write::Buffer; -use crate::virtual_file::{self, owned_buffers_io, VirtualFile}; +use crate::virtual_file::{self, owned_buffers_io, IoBufferMut, VirtualFile}; use bytes::BytesMut; use camino::Utf8PathBuf; use num_traits::Num; @@ -107,15 +108,18 @@ impl EphemeralFile { self.page_cache_file_id } - pub(crate) async fn load_to_vec(&self, ctx: &RequestContext) -> Result, io::Error> { + pub(crate) async fn load_to_io_buf( + &self, + ctx: &RequestContext, + ) -> Result { let size = self.len().into_usize(); - let vec = Vec::with_capacity(size); - let (slice, nread) = self.read_exact_at_eof_ok(0, vec.slice_full(), ctx).await?; + let buf = IoBufferMut::with_capacity(size); + let (slice, nread) = self.read_exact_at_eof_ok(0, buf.slice_full(), ctx).await?; assert_eq!(nread, size); - let vec = slice.into_inner(); - assert_eq!(vec.len(), nread); - assert_eq!(vec.capacity(), size, "we shouldn't be reallocating"); - Ok(vec) + let buf = slice.into_inner(); + assert_eq!(buf.len(), nread); + assert_eq!(buf.capacity(), size, "we shouldn't be reallocating"); + Ok(buf) } /// Returns the offset at which the first byte of the input was written, for use @@ -158,7 +162,7 @@ impl EphemeralFile { } impl super::storage_layer::inmemory_layer::vectored_dio_read::File for EphemeralFile { - async fn read_exact_at_eof_ok<'a, 'b, B: tokio_epoll_uring::IoBufMut + Send>( + async fn read_exact_at_eof_ok<'a, 'b, B: IoBufAlignedMut + Send>( &'b self, start: u64, dst: tokio_epoll_uring::Slice, @@ -345,7 +349,7 @@ mod tests { assert!(file.len() as usize == write_nbytes); for i in 0..write_nbytes { assert_eq!(value_offsets[i], i.into_u64()); - let buf = Vec::with_capacity(1); + let buf = IoBufferMut::with_capacity(1); let (buf_slice, nread) = file .read_exact_at_eof_ok(i.into_u64(), buf.slice_full(), &ctx) .await @@ -385,7 +389,7 @@ mod tests { // assert the state is as this test expects it to be assert_eq!( - &file.load_to_vec(&ctx).await.unwrap(), + &file.load_to_io_buf(&ctx).await.unwrap(), &content[0..cap + cap / 2] ); let md = file @@ -440,7 +444,7 @@ mod tests { let (buf, nread) = file .read_exact_at_eof_ok( start.into_u64(), - Vec::with_capacity(len).slice_full(), + IoBufferMut::with_capacity(len).slice_full(), ctx, ) .await diff --git a/pageserver/src/tenant/storage_layer/delta_layer.rs b/pageserver/src/tenant/storage_layer/delta_layer.rs index 6332d36dc3..ceae1d4b1a 100644 --- a/pageserver/src/tenant/storage_layer/delta_layer.rs +++ b/pageserver/src/tenant/storage_layer/delta_layer.rs @@ -44,11 +44,11 @@ use crate::tenant::vectored_blob_io::{ }; use crate::tenant::PageReconstructError; use crate::virtual_file::owned_buffers_io::io_buf_ext::{FullSlice, IoBufExt}; +use crate::virtual_file::IoBufferMut; use crate::virtual_file::{self, MaybeFatalIo, VirtualFile}; use crate::{walrecord, TEMP_FILE_SUFFIX}; use crate::{DELTA_FILE_MAGIC, STORAGE_FORMAT_VERSION}; use anyhow::{anyhow, bail, ensure, Context, Result}; -use bytes::BytesMut; use camino::{Utf8Path, Utf8PathBuf}; use futures::StreamExt; use itertools::Itertools; @@ -1002,7 +1002,7 @@ impl DeltaLayerInner { .0 .into(); let buf_size = Self::get_min_read_buffer_size(&reads, max_vectored_read_bytes); - let mut buf = Some(BytesMut::with_capacity(buf_size)); + let mut buf = Some(IoBufferMut::with_capacity(buf_size)); // Note that reads are processed in reverse order (from highest key+lsn). // This is the order that `ReconstructState` requires such that it can @@ -1029,7 +1029,7 @@ impl DeltaLayerInner { // We have "lost" the buffer since the lower level IO api // doesn't return the buffer on error. Allocate a new one. - buf = Some(BytesMut::with_capacity(buf_size)); + buf = Some(IoBufferMut::with_capacity(buf_size)); continue; } @@ -1203,7 +1203,7 @@ impl DeltaLayerInner { .map(|x| x.0.get()) .unwrap_or(8192); - let mut buffer = Some(BytesMut::with_capacity(max_read_size)); + let mut buffer = Some(IoBufferMut::with_capacity(max_read_size)); // FIXME: buffering of DeltaLayerWriter let mut per_blob_copy = Vec::new(); @@ -1561,12 +1561,11 @@ impl<'a> DeltaLayerIterator<'a> { let vectored_blob_reader = VectoredBlobReader::new(&self.delta_layer.file); let mut next_batch = std::collections::VecDeque::new(); let buf_size = plan.size(); - let buf = BytesMut::with_capacity(buf_size); + let buf = IoBufferMut::with_capacity(buf_size); let blobs_buf = vectored_blob_reader .read_blobs(&plan, buf, self.ctx) .await?; - let frozen_buf = blobs_buf.buf.freeze(); - let view = BufView::new_bytes(frozen_buf); + let view = BufView::new_slice(&blobs_buf.buf); for meta in blobs_buf.blobs.iter() { let blob_read = meta.read(&view).await?; let value = Value::des(&blob_read)?; @@ -1941,7 +1940,7 @@ pub(crate) mod test { &vectored_reads, constants::MAX_VECTORED_READ_BYTES, ); - let mut buf = Some(BytesMut::with_capacity(buf_size)); + let mut buf = Some(IoBufferMut::with_capacity(buf_size)); for read in vectored_reads { let blobs_buf = vectored_blob_reader diff --git a/pageserver/src/tenant/storage_layer/image_layer.rs b/pageserver/src/tenant/storage_layer/image_layer.rs index b1f2557038..fa058833d4 100644 --- a/pageserver/src/tenant/storage_layer/image_layer.rs +++ b/pageserver/src/tenant/storage_layer/image_layer.rs @@ -41,10 +41,11 @@ use crate::tenant::vectored_blob_io::{ }; use crate::tenant::PageReconstructError; use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt; +use crate::virtual_file::IoBufferMut; use crate::virtual_file::{self, MaybeFatalIo, VirtualFile}; use crate::{IMAGE_FILE_MAGIC, STORAGE_FORMAT_VERSION, TEMP_FILE_SUFFIX}; use anyhow::{anyhow, bail, ensure, Context, Result}; -use bytes::{Bytes, BytesMut}; +use bytes::Bytes; use camino::{Utf8Path, Utf8PathBuf}; use hex; use itertools::Itertools; @@ -547,10 +548,10 @@ impl ImageLayerInner { for read in plan.into_iter() { let buf_size = read.size(); - let buf = BytesMut::with_capacity(buf_size); + let buf = IoBufferMut::with_capacity(buf_size); let blobs_buf = vectored_blob_reader.read_blobs(&read, buf, ctx).await?; - let frozen_buf = blobs_buf.buf.freeze(); - let view = BufView::new_bytes(frozen_buf); + + let view = BufView::new_slice(&blobs_buf.buf); for meta in blobs_buf.blobs.iter() { let img_buf = meta.read(&view).await?; @@ -609,13 +610,12 @@ impl ImageLayerInner { } } - let buf = BytesMut::with_capacity(buf_size); + let buf = IoBufferMut::with_capacity(buf_size); let res = vectored_blob_reader.read_blobs(&read, buf, ctx).await; match res { Ok(blobs_buf) => { - let frozen_buf = blobs_buf.buf.freeze(); - let view = BufView::new_bytes(frozen_buf); + let view = BufView::new_slice(&blobs_buf.buf); for meta in blobs_buf.blobs.iter() { let img_buf = meta.read(&view).await; @@ -1069,12 +1069,11 @@ impl<'a> ImageLayerIterator<'a> { let vectored_blob_reader = VectoredBlobReader::new(&self.image_layer.file); let mut next_batch = std::collections::VecDeque::new(); let buf_size = plan.size(); - let buf = BytesMut::with_capacity(buf_size); + let buf = IoBufferMut::with_capacity(buf_size); let blobs_buf = vectored_blob_reader .read_blobs(&plan, buf, self.ctx) .await?; - let frozen_buf = blobs_buf.buf.freeze(); - let view = BufView::new_bytes(frozen_buf); + let view = BufView::new_slice(&blobs_buf.buf); for meta in blobs_buf.blobs.iter() { let img_buf = meta.read(&view).await?; next_batch.push_back(( diff --git a/pageserver/src/tenant/storage_layer/inmemory_layer.rs b/pageserver/src/tenant/storage_layer/inmemory_layer.rs index e487bee1f2..7573ddb5cc 100644 --- a/pageserver/src/tenant/storage_layer/inmemory_layer.rs +++ b/pageserver/src/tenant/storage_layer/inmemory_layer.rs @@ -14,7 +14,6 @@ use crate::tenant::PageReconstructError; use crate::virtual_file::owned_buffers_io::io_buf_ext::IoBufExt; use crate::{l0_flush, page_cache}; use anyhow::{anyhow, Context, Result}; -use bytes::Bytes; use camino::Utf8PathBuf; use pageserver_api::key::CompactKey; use pageserver_api::keyspace::KeySpace; @@ -809,9 +808,8 @@ impl InMemoryLayer { match l0_flush_global_state { l0_flush::Inner::Direct { .. } => { - let file_contents: Vec = inner.file.load_to_vec(ctx).await?; - - let file_contents = Bytes::from(file_contents); + let file_contents = inner.file.load_to_io_buf(ctx).await?; + let file_contents = file_contents.freeze(); for (key, vec_map) in inner.index.iter() { // Write all page versions @@ -825,7 +823,7 @@ impl InMemoryLayer { len, will_init, } = entry; - let buf = Bytes::slice(&file_contents, pos as usize..(pos + len) as usize); + let buf = file_contents.slice(pos as usize..(pos + len) as usize); let (_buf, res) = delta_layer_writer .put_value_bytes( Key::from_compact(*key), diff --git a/pageserver/src/tenant/storage_layer/inmemory_layer/vectored_dio_read.rs b/pageserver/src/tenant/storage_layer/inmemory_layer/vectored_dio_read.rs index 0683e15659..a4bb3a6bfc 100644 --- a/pageserver/src/tenant/storage_layer/inmemory_layer/vectored_dio_read.rs +++ b/pageserver/src/tenant/storage_layer/inmemory_layer/vectored_dio_read.rs @@ -9,6 +9,7 @@ use tokio_epoll_uring::{BoundedBuf, IoBufMut, Slice}; use crate::{ assert_u64_eq_usize::{U64IsUsize, UsizeIsU64}, context::RequestContext, + virtual_file::{owned_buffers_io::io_buf_aligned::IoBufAlignedMut, IoBufferMut}, }; /// The file interface we require. At runtime, this is a [`crate::tenant::ephemeral_file::EphemeralFile`]. @@ -24,7 +25,7 @@ pub trait File: Send { /// [`std::io::ErrorKind::UnexpectedEof`] error if the file is shorter than `start+dst.len()`. /// /// No guarantees are made about the remaining bytes in `dst` in case of a short read. - async fn read_exact_at_eof_ok<'a, 'b, B: IoBufMut + Send>( + async fn read_exact_at_eof_ok<'a, 'b, B: IoBufAlignedMut + Send>( &'b self, start: u64, dst: Slice, @@ -227,7 +228,7 @@ where // Execute physical reads and fill the logical read buffers // TODO: pipelined reads; prefetch; - let get_io_buffer = |nchunks| Vec::with_capacity(nchunks * DIO_CHUNK_SIZE); + let get_io_buffer = |nchunks| IoBufferMut::with_capacity(nchunks * DIO_CHUNK_SIZE); for PhysicalRead { start_chunk_no, nchunks, @@ -459,7 +460,7 @@ mod tests { let ctx = RequestContext::new(TaskKind::UnitTest, DownloadBehavior::Error); let file = InMemoryFile::new_random(10); let test_read = |pos, len| { - let buf = vec![0; len]; + let buf = IoBufferMut::with_capacity_zeroed(len); let fut = file.read_exact_at_eof_ok(pos, buf.slice_full(), &ctx); use futures::FutureExt; let (slice, nread) = fut @@ -470,9 +471,9 @@ mod tests { buf.truncate(nread); buf }; - assert_eq!(test_read(0, 1), &file.content[0..1]); - assert_eq!(test_read(1, 2), &file.content[1..3]); - assert_eq!(test_read(9, 2), &file.content[9..]); + assert_eq!(&test_read(0, 1), &file.content[0..1]); + assert_eq!(&test_read(1, 2), &file.content[1..3]); + assert_eq!(&test_read(9, 2), &file.content[9..]); assert!(test_read(10, 2).is_empty()); assert!(test_read(11, 2).is_empty()); } @@ -609,7 +610,7 @@ mod tests { } impl<'x> File for RecorderFile<'x> { - async fn read_exact_at_eof_ok<'a, 'b, B: IoBufMut + Send>( + async fn read_exact_at_eof_ok<'a, 'b, B: IoBufAlignedMut + Send>( &'b self, start: u64, dst: Slice, @@ -782,7 +783,7 @@ mod tests { 2048, 1024 => Err("foo".to_owned()), }; - let buf = Vec::with_capacity(512); + let buf = IoBufferMut::with_capacity(512); let (buf, nread) = mock_file .read_exact_at_eof_ok(0, buf.slice_full(), &ctx) .await @@ -790,7 +791,7 @@ mod tests { assert_eq!(nread, 512); assert_eq!(&buf.into_inner()[..nread], &[0; 512]); - let buf = Vec::with_capacity(512); + let buf = IoBufferMut::with_capacity(512); let (buf, nread) = mock_file .read_exact_at_eof_ok(512, buf.slice_full(), &ctx) .await @@ -798,7 +799,7 @@ mod tests { assert_eq!(nread, 512); assert_eq!(&buf.into_inner()[..nread], &[1; 512]); - let buf = Vec::with_capacity(512); + let buf = IoBufferMut::with_capacity(512); let (buf, nread) = mock_file .read_exact_at_eof_ok(1024, buf.slice_full(), &ctx) .await @@ -806,7 +807,7 @@ mod tests { assert_eq!(nread, 10); assert_eq!(&buf.into_inner()[..nread], &[2; 10]); - let buf = Vec::with_capacity(1024); + let buf = IoBufferMut::with_capacity(1024); let err = mock_file .read_exact_at_eof_ok(2048, buf.slice_full(), &ctx) .await diff --git a/pageserver/src/tenant/vectored_blob_io.rs b/pageserver/src/tenant/vectored_blob_io.rs index 0c03791034..dfe2352310 100644 --- a/pageserver/src/tenant/vectored_blob_io.rs +++ b/pageserver/src/tenant/vectored_blob_io.rs @@ -18,7 +18,7 @@ use std::collections::BTreeMap; use std::ops::Deref; -use bytes::{Bytes, BytesMut}; +use bytes::Bytes; use pageserver_api::key::Key; use tokio::io::AsyncWriteExt; use tokio_epoll_uring::BoundedBuf; @@ -27,6 +27,7 @@ use utils::vec_map::VecMap; use crate::context::RequestContext; use crate::tenant::blob_io::{BYTE_UNCOMPRESSED, BYTE_ZSTD, LEN_COMPRESSION_BIT_MASK}; +use crate::virtual_file::IoBufferMut; use crate::virtual_file::{self, VirtualFile}; /// Metadata bundled with the start and end offset of a blob. @@ -158,7 +159,7 @@ impl std::fmt::Display for VectoredBlob { /// Return type of [`VectoredBlobReader::read_blobs`] pub struct VectoredBlobsBuf { /// Buffer for all blobs in this read - pub buf: BytesMut, + pub buf: IoBufferMut, /// Offsets into the buffer and metadata for all blobs in this read pub blobs: Vec, } @@ -441,7 +442,7 @@ impl<'a> VectoredBlobReader<'a> { pub async fn read_blobs( &self, read: &VectoredRead, - buf: BytesMut, + buf: IoBufferMut, ctx: &RequestContext, ) -> Result { assert!(read.size() > 0); @@ -916,7 +917,7 @@ mod tests { // Multiply by two (compressed data might need more space), and add a few bytes for the header let reserved_bytes = blobs.iter().map(|bl| bl.len()).max().unwrap() * 2 + 16; - let mut buf = BytesMut::with_capacity(reserved_bytes); + let mut buf = IoBufferMut::with_capacity(reserved_bytes); let vectored_blob_reader = VectoredBlobReader::new(&file); let meta = BlobMeta { diff --git a/pageserver/src/virtual_file.rs b/pageserver/src/virtual_file.rs index 5a364b7aaf..daa8b99ab0 100644 --- a/pageserver/src/virtual_file.rs +++ b/pageserver/src/virtual_file.rs @@ -18,6 +18,9 @@ use crate::page_cache::{PageWriteGuard, PAGE_SZ}; use crate::tenant::TENANTS_SEGMENT_NAME; use camino::{Utf8Path, Utf8PathBuf}; use once_cell::sync::OnceCell; +use owned_buffers_io::aligned_buffer::buffer::AlignedBuffer; +use owned_buffers_io::aligned_buffer::{AlignedBufferMut, AlignedSlice, ConstAlign}; +use owned_buffers_io::io_buf_aligned::IoBufAlignedMut; use owned_buffers_io::io_buf_ext::FullSlice; use pageserver_api::config::defaults::DEFAULT_IO_BUFFER_ALIGNMENT; use pageserver_api::shard::TenantShardId; @@ -55,6 +58,8 @@ pub(crate) mod owned_buffers_io { //! but for the time being we're proving out the primitives in the neon.git repo //! for faster iteration. + pub(crate) mod aligned_buffer; + pub(crate) mod io_buf_aligned; pub(crate) mod io_buf_ext; pub(crate) mod slice; pub(crate) mod write; @@ -196,7 +201,7 @@ impl VirtualFile { ctx: &RequestContext, ) -> Result, Error> where - Buf: IoBufMut + Send, + Buf: IoBufAlignedMut + Send, { self.inner.read_exact_at(slice, offset, ctx).await } @@ -771,7 +776,7 @@ impl VirtualFileInner { ctx: &RequestContext, ) -> Result, Error> where - Buf: IoBufMut + Send, + Buf: IoBufAlignedMut + Send, { let assert_we_return_original_bounds = if cfg!(debug_assertions) { Some((slice.stable_ptr() as usize, slice.bytes_total())) @@ -1222,12 +1227,14 @@ impl VirtualFileInner { ctx: &RequestContext, ) -> Result, std::io::Error> { use crate::page_cache::PAGE_SZ; - let slice = Vec::with_capacity(PAGE_SZ).slice_full(); + let slice = IoBufferMut::with_capacity(PAGE_SZ).slice_full(); assert_eq!(slice.bytes_total(), PAGE_SZ); let slice = self .read_exact_at(slice, blknum as u64 * (PAGE_SZ as u64), ctx) .await?; - Ok(crate::tenant::block_io::BlockLease::Vec(slice.into_inner())) + Ok(crate::tenant::block_io::BlockLease::IoBufferMut( + slice.into_inner(), + )) } async fn read_to_end(&mut self, buf: &mut Vec, ctx: &RequestContext) -> Result<(), Error> { @@ -1325,10 +1332,11 @@ impl OpenFiles { /// server startup. /// #[cfg(not(test))] -pub fn init(num_slots: usize, engine: IoEngineKind) { +pub fn init(num_slots: usize, engine: IoEngineKind, mode: IoMode) { if OPEN_FILES.set(OpenFiles::new(num_slots)).is_err() { panic!("virtual_file::init called twice"); } + set_io_mode(mode); io_engine::init(engine); crate::metrics::virtual_file_descriptor_cache::SIZE_MAX.set(num_slots as u64); } @@ -1357,6 +1365,11 @@ pub(crate) const fn get_io_buffer_alignment() -> usize { DEFAULT_IO_BUFFER_ALIGNMENT } +pub(crate) type IoBufferMut = AlignedBufferMut>; +pub(crate) type IoBuffer = AlignedBuffer>; +pub(crate) type IoPageSlice<'a> = + AlignedSlice<'a, PAGE_SZ, ConstAlign<{ get_io_buffer_alignment() }>>; + static IO_MODE: AtomicU8 = AtomicU8::new(IoMode::preferred() as u8); pub(crate) fn set_io_mode(mode: IoMode) { @@ -1395,10 +1408,10 @@ mod tests { impl MaybeVirtualFile { async fn read_exact_at( &self, - mut slice: tokio_epoll_uring::Slice>, + mut slice: tokio_epoll_uring::Slice, offset: u64, ctx: &RequestContext, - ) -> Result>, Error> { + ) -> Result, Error> { match self { MaybeVirtualFile::VirtualFile(file) => file.read_exact_at(slice, offset, ctx).await, MaybeVirtualFile::File(file) => { @@ -1466,12 +1479,13 @@ mod tests { len: usize, ctx: &RequestContext, ) -> Result { - let slice = Vec::with_capacity(len).slice_full(); + let slice = IoBufferMut::with_capacity(len).slice_full(); assert_eq!(slice.bytes_total(), len); let slice = self.read_exact_at(slice, pos, ctx).await?; - let vec = slice.into_inner(); - assert_eq!(vec.len(), len); - Ok(String::from_utf8(vec).unwrap()) + let buf = slice.into_inner(); + assert_eq!(buf.len(), len); + + Ok(String::from_utf8(buf.to_vec()).unwrap()) } } @@ -1695,7 +1709,7 @@ mod tests { let files = files.clone(); let ctx = ctx.detached_child(TaskKind::UnitTest, DownloadBehavior::Error); let hdl = rt.spawn(async move { - let mut buf = vec![0u8; SIZE]; + let mut buf = IoBufferMut::with_capacity_zeroed(SIZE); let mut rng = rand::rngs::OsRng; for _ in 1..1000 { let f = &files[rng.gen_range(0..files.len())]; @@ -1704,7 +1718,7 @@ mod tests { .await .unwrap() .into_inner(); - assert!(buf == SAMPLE); + assert!(buf[..] == SAMPLE); } }); hdls.push(hdl); diff --git a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer.rs b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer.rs new file mode 100644 index 0000000000..8ffc29b93d --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer.rs @@ -0,0 +1,9 @@ +pub mod alignment; +pub mod buffer; +pub mod buffer_mut; +pub mod raw; +pub mod slice; + +pub use alignment::*; +pub use buffer_mut::AlignedBufferMut; +pub use slice::AlignedSlice; diff --git a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/alignment.rs b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/alignment.rs new file mode 100644 index 0000000000..933b78a13b --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/alignment.rs @@ -0,0 +1,26 @@ +pub trait Alignment: std::marker::Unpin + 'static { + /// Returns the required alignments. + fn align(&self) -> usize; +} + +/// Alignment at compile time. +#[derive(Debug)] +pub struct ConstAlign; + +impl Alignment for ConstAlign { + fn align(&self) -> usize { + A + } +} + +/// Alignment at run time. +#[derive(Debug)] +pub struct RuntimeAlign { + align: usize, +} + +impl Alignment for RuntimeAlign { + fn align(&self) -> usize { + self.align + } +} diff --git a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer.rs b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer.rs new file mode 100644 index 0000000000..2fba6d699b --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer.rs @@ -0,0 +1,124 @@ +use std::{ + ops::{Deref, Range, RangeBounds}, + sync::Arc, +}; + +use super::{alignment::Alignment, raw::RawAlignedBuffer}; + +/// An shared, immutable aligned buffer type. +pub struct AlignedBuffer { + /// Shared raw buffer. + raw: Arc>, + /// Range that specifies the current slice. + range: Range, +} + +impl AlignedBuffer { + /// Creates an immutable `IoBuffer` from the raw buffer + pub(super) fn from_raw(raw: RawAlignedBuffer, range: Range) -> Self { + AlignedBuffer { + raw: Arc::new(raw), + range, + } + } + + /// Returns the number of bytes in the buffer, also referred to as its 'length'. + #[inline] + pub fn len(&self) -> usize { + self.range.len() + } + + /// Returns the alignment of the buffer. + #[inline] + pub fn align(&self) -> usize { + self.raw.align() + } + + #[inline] + fn as_ptr(&self) -> *const u8 { + // SAFETY: `self.range.start` is guaranteed to be within [0, self.len()). + unsafe { self.raw.as_ptr().add(self.range.start) } + } + + /// Extracts a slice containing the entire buffer. + /// + /// Equivalent to `&s[..]`. + #[inline] + fn as_slice(&self) -> &[u8] { + &self.raw.as_slice()[self.range.start..self.range.end] + } + + /// Returns a slice of self for the index range `[begin..end)`. + pub fn slice(&self, range: impl RangeBounds) -> Self { + use core::ops::Bound; + let len = self.len(); + + let begin = match range.start_bound() { + Bound::Included(&n) => n, + Bound::Excluded(&n) => n.checked_add(1).expect("out of range"), + Bound::Unbounded => 0, + }; + + let end = match range.end_bound() { + Bound::Included(&n) => n.checked_add(1).expect("out of range"), + Bound::Excluded(&n) => n, + Bound::Unbounded => len, + }; + + assert!( + begin <= end, + "range start must not be greater than end: {:?} <= {:?}", + begin, + end, + ); + assert!( + end <= len, + "range end out of bounds: {:?} <= {:?}", + end, + len, + ); + + let begin = self.range.start + begin; + let end = self.range.start + end; + + AlignedBuffer { + raw: Arc::clone(&self.raw), + range: begin..end, + } + } +} + +impl Deref for AlignedBuffer { + type Target = [u8]; + + fn deref(&self) -> &Self::Target { + self.as_slice() + } +} + +impl AsRef<[u8]> for AlignedBuffer { + fn as_ref(&self) -> &[u8] { + self.as_slice() + } +} + +impl PartialEq<[u8]> for AlignedBuffer { + fn eq(&self, other: &[u8]) -> bool { + self.as_slice().eq(other) + } +} + +/// SAFETY: the underlying buffer references a stable memory region. +unsafe impl tokio_epoll_uring::IoBuf for AlignedBuffer { + fn stable_ptr(&self) -> *const u8 { + self.as_ptr() + } + + fn bytes_init(&self) -> usize { + self.len() + } + + fn bytes_total(&self) -> usize { + self.len() + } +} diff --git a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer_mut.rs b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer_mut.rs new file mode 100644 index 0000000000..b3675d1aea --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/buffer_mut.rs @@ -0,0 +1,347 @@ +use std::ops::{Deref, DerefMut}; + +use super::{ + alignment::{Alignment, ConstAlign}, + buffer::AlignedBuffer, + raw::RawAlignedBuffer, +}; + +/// A mutable aligned buffer type. +#[derive(Debug)] +pub struct AlignedBufferMut { + raw: RawAlignedBuffer, +} + +impl AlignedBufferMut> { + /// Constructs a new, empty `IoBufferMut` with at least the specified capacity and alignment. + /// + /// The buffer will be able to hold at most `capacity` elements and will never resize. + /// + /// + /// # Panics + /// + /// Panics if the new capacity exceeds `isize::MAX` _bytes_, or if the following alignment requirement is not met: + /// * `align` must not be zero, + /// + /// * `align` must be a power of two, + /// + /// * `capacity`, when rounded up to the nearest multiple of `align`, + /// must not overflow isize (i.e., the rounded value must be + /// less than or equal to `isize::MAX`). + pub fn with_capacity(capacity: usize) -> Self { + AlignedBufferMut { + raw: RawAlignedBuffer::with_capacity(capacity), + } + } + + /// Constructs a new `IoBufferMut` with at least the specified capacity and alignment, filled with zeros. + pub fn with_capacity_zeroed(capacity: usize) -> Self { + use bytes::BufMut; + let mut buf = Self::with_capacity(capacity); + buf.put_bytes(0, capacity); + // SAFETY: `put_bytes` filled the entire buffer. + unsafe { buf.set_len(capacity) }; + buf + } +} + +impl AlignedBufferMut { + /// Returns the total number of bytes the buffer can hold. + #[inline] + pub fn capacity(&self) -> usize { + self.raw.capacity() + } + + /// Returns the alignment of the buffer. + #[inline] + pub fn align(&self) -> usize { + self.raw.align() + } + + /// Returns the number of bytes in the buffer, also referred to as its 'length'. + #[inline] + pub fn len(&self) -> usize { + self.raw.len() + } + + /// Force the length of the buffer to `new_len`. + #[inline] + unsafe fn set_len(&mut self, new_len: usize) { + self.raw.set_len(new_len) + } + + #[inline] + fn as_ptr(&self) -> *const u8 { + self.raw.as_ptr() + } + + #[inline] + fn as_mut_ptr(&mut self) -> *mut u8 { + self.raw.as_mut_ptr() + } + + /// Extracts a slice containing the entire buffer. + /// + /// Equivalent to `&s[..]`. + #[inline] + fn as_slice(&self) -> &[u8] { + self.raw.as_slice() + } + + /// Extracts a mutable slice of the entire buffer. + /// + /// Equivalent to `&mut s[..]`. + fn as_mut_slice(&mut self) -> &mut [u8] { + self.raw.as_mut_slice() + } + + /// Drops the all the contents of the buffer, setting its length to `0`. + #[inline] + pub fn clear(&mut self) { + self.raw.clear() + } + + /// Reserves capacity for at least `additional` more bytes to be inserted + /// in the given `IoBufferMut`. The collection may reserve more space to + /// speculatively avoid frequent reallocations. After calling `reserve`, + /// capacity will be greater than or equal to `self.len() + additional`. + /// Does nothing if capacity is already sufficient. + /// + /// # Panics + /// + /// Panics if the new capacity exceeds `isize::MAX` _bytes_. + pub fn reserve(&mut self, additional: usize) { + self.raw.reserve(additional); + } + + /// Shortens the buffer, keeping the first len bytes. + pub fn truncate(&mut self, len: usize) { + self.raw.truncate(len); + } + + /// Consumes and leaks the `IoBufferMut`, returning a mutable reference to the contents, &'a mut [u8]. + pub fn leak<'a>(self) -> &'a mut [u8] { + self.raw.leak() + } + + pub fn freeze(self) -> AlignedBuffer { + let len = self.len(); + AlignedBuffer::from_raw(self.raw, 0..len) + } +} + +impl Deref for AlignedBufferMut { + type Target = [u8]; + + fn deref(&self) -> &Self::Target { + self.as_slice() + } +} + +impl DerefMut for AlignedBufferMut { + fn deref_mut(&mut self) -> &mut Self::Target { + self.as_mut_slice() + } +} + +impl AsRef<[u8]> for AlignedBufferMut { + fn as_ref(&self) -> &[u8] { + self.as_slice() + } +} + +impl AsMut<[u8]> for AlignedBufferMut { + fn as_mut(&mut self) -> &mut [u8] { + self.as_mut_slice() + } +} + +impl PartialEq<[u8]> for AlignedBufferMut { + fn eq(&self, other: &[u8]) -> bool { + self.as_slice().eq(other) + } +} + +/// SAFETY: When advancing the internal cursor, the caller needs to make sure the bytes advcanced past have been initialized. +unsafe impl bytes::BufMut for AlignedBufferMut { + #[inline] + fn remaining_mut(&self) -> usize { + // Although a `Vec` can have at most isize::MAX bytes, we never want to grow `IoBufferMut`. + // Thus, it can have at most `self.capacity` bytes. + self.capacity() - self.len() + } + + // SAFETY: Caller needs to make sure the bytes being advanced past have been initialized. + #[inline] + unsafe fn advance_mut(&mut self, cnt: usize) { + let len = self.len(); + let remaining = self.remaining_mut(); + + if remaining < cnt { + panic_advance(cnt, remaining); + } + + // Addition will not overflow since the sum is at most the capacity. + self.set_len(len + cnt); + } + + #[inline] + fn chunk_mut(&mut self) -> &mut bytes::buf::UninitSlice { + let cap = self.capacity(); + let len = self.len(); + + // SAFETY: Since `self.ptr` is valid for `cap` bytes, `self.ptr.add(len)` must be + // valid for `cap - len` bytes. The subtraction will not underflow since + // `len <= cap`. + unsafe { + bytes::buf::UninitSlice::from_raw_parts_mut(self.as_mut_ptr().add(len), cap - len) + } + } +} + +/// Panic with a nice error message. +#[cold] +fn panic_advance(idx: usize, len: usize) -> ! { + panic!( + "advance out of bounds: the len is {} but advancing by {}", + len, idx + ); +} + +/// Safety: [`AlignedBufferMut`] has exclusive ownership of the io buffer, +/// and the underlying pointer remains stable while io-uring is owning the buffer. +/// The tokio-epoll-uring crate itself will not resize the buffer and will respect +/// [`tokio_epoll_uring::IoBuf::bytes_total`]. +unsafe impl tokio_epoll_uring::IoBuf for AlignedBufferMut { + fn stable_ptr(&self) -> *const u8 { + self.as_ptr() + } + + fn bytes_init(&self) -> usize { + self.len() + } + + fn bytes_total(&self) -> usize { + self.capacity() + } +} + +// SAFETY: See above. +unsafe impl tokio_epoll_uring::IoBufMut for AlignedBufferMut { + fn stable_mut_ptr(&mut self) -> *mut u8 { + self.as_mut_ptr() + } + + unsafe fn set_init(&mut self, init_len: usize) { + if self.len() < init_len { + self.set_len(init_len); + } + } +} + +#[cfg(test)] +mod tests { + + use super::*; + + const ALIGN: usize = 4 * 1024; + type TestIoBufferMut = AlignedBufferMut>; + + #[test] + fn test_with_capacity() { + let v = TestIoBufferMut::with_capacity(ALIGN * 4); + assert_eq!(v.len(), 0); + assert_eq!(v.capacity(), ALIGN * 4); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + + let v = TestIoBufferMut::with_capacity(ALIGN / 2); + assert_eq!(v.len(), 0); + assert_eq!(v.capacity(), ALIGN / 2); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + } + + #[test] + fn test_with_capacity_zeroed() { + let v = TestIoBufferMut::with_capacity_zeroed(ALIGN); + assert_eq!(v.len(), ALIGN); + assert_eq!(v.capacity(), ALIGN); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + assert_eq!(&v[..], &[0; ALIGN]) + } + + #[test] + fn test_reserve() { + use bytes::BufMut; + let mut v = TestIoBufferMut::with_capacity(ALIGN); + let capacity = v.capacity(); + v.reserve(capacity); + assert_eq!(v.capacity(), capacity); + let data = [b'a'; ALIGN]; + v.put(&data[..]); + v.reserve(capacity); + assert!(v.capacity() >= capacity * 2); + assert_eq!(&v[..], &data[..]); + let capacity = v.capacity(); + v.clear(); + v.reserve(capacity); + assert_eq!(capacity, v.capacity()); + } + + #[test] + fn test_bytes_put() { + use bytes::BufMut; + let mut v = TestIoBufferMut::with_capacity(ALIGN * 4); + let x = [b'a'; ALIGN]; + + for _ in 0..2 { + for _ in 0..4 { + v.put(&x[..]); + } + assert_eq!(v.len(), ALIGN * 4); + assert_eq!(v.capacity(), ALIGN * 4); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + v.clear() + } + assert_eq!(v.len(), 0); + assert_eq!(v.capacity(), ALIGN * 4); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + } + + #[test] + #[should_panic] + fn test_bytes_put_panic() { + use bytes::BufMut; + const ALIGN: usize = 4 * 1024; + let mut v = TestIoBufferMut::with_capacity(ALIGN * 4); + let x = [b'a'; ALIGN]; + for _ in 0..5 { + v.put_slice(&x[..]); + } + } + + #[test] + fn test_io_buf_put_slice() { + use tokio_epoll_uring::BoundedBufMut; + const ALIGN: usize = 4 * 1024; + let mut v = TestIoBufferMut::with_capacity(ALIGN); + let x = [b'a'; ALIGN]; + + for _ in 0..2 { + v.put_slice(&x[..]); + assert_eq!(v.len(), ALIGN); + assert_eq!(v.capacity(), ALIGN); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + v.clear() + } + assert_eq!(v.len(), 0); + assert_eq!(v.capacity(), ALIGN); + assert_eq!(v.align(), ALIGN); + assert_eq!(v.as_ptr().align_offset(ALIGN), 0); + } +} diff --git a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/raw.rs b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/raw.rs new file mode 100644 index 0000000000..6c26dec0db --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/raw.rs @@ -0,0 +1,216 @@ +use core::slice; +use std::{ + alloc::{self, Layout}, + cmp, + mem::ManuallyDrop, +}; + +use super::alignment::{Alignment, ConstAlign}; + +#[derive(Debug)] +struct AlignedBufferPtr(*mut u8); + +// SAFETY: We gurantees no one besides `IoBufferPtr` itself has the raw pointer. +unsafe impl Send for AlignedBufferPtr {} + +// SAFETY: We gurantees no one besides `IoBufferPtr` itself has the raw pointer. +unsafe impl Sync for AlignedBufferPtr {} + +/// An aligned buffer type. +#[derive(Debug)] +pub struct RawAlignedBuffer { + ptr: AlignedBufferPtr, + capacity: usize, + len: usize, + align: A, +} + +impl RawAlignedBuffer> { + /// Constructs a new, empty `IoBufferMut` with at least the specified capacity and alignment. + /// + /// The buffer will be able to hold at most `capacity` elements and will never resize. + /// + /// + /// # Panics + /// + /// Panics if the new capacity exceeds `isize::MAX` _bytes_, or if the following alignment requirement is not met: + /// * `align` must not be zero, + /// + /// * `align` must be a power of two, + /// + /// * `capacity`, when rounded up to the nearest multiple of `align`, + /// must not overflow isize (i.e., the rounded value must be + /// less than or equal to `isize::MAX`). + pub fn with_capacity(capacity: usize) -> Self { + let align = ConstAlign::; + let layout = Layout::from_size_align(capacity, align.align()).expect("Invalid layout"); + + // SAFETY: Making an allocation with a sized and aligned layout. The memory is manually freed with the same layout. + let ptr = unsafe { + let ptr = alloc::alloc(layout); + if ptr.is_null() { + alloc::handle_alloc_error(layout); + } + AlignedBufferPtr(ptr) + }; + + RawAlignedBuffer { + ptr, + capacity, + len: 0, + align, + } + } +} + +impl RawAlignedBuffer { + /// Returns the total number of bytes the buffer can hold. + #[inline] + pub fn capacity(&self) -> usize { + self.capacity + } + + /// Returns the alignment of the buffer. + #[inline] + pub fn align(&self) -> usize { + self.align.align() + } + + /// Returns the number of bytes in the buffer, also referred to as its 'length'. + #[inline] + pub fn len(&self) -> usize { + self.len + } + + /// Force the length of the buffer to `new_len`. + #[inline] + pub unsafe fn set_len(&mut self, new_len: usize) { + debug_assert!(new_len <= self.capacity()); + self.len = new_len; + } + + #[inline] + pub fn as_ptr(&self) -> *const u8 { + self.ptr.0 + } + + #[inline] + pub fn as_mut_ptr(&mut self) -> *mut u8 { + self.ptr.0 + } + + /// Extracts a slice containing the entire buffer. + /// + /// Equivalent to `&s[..]`. + #[inline] + pub fn as_slice(&self) -> &[u8] { + // SAFETY: The pointer is valid and `len` bytes are initialized. + unsafe { slice::from_raw_parts(self.as_ptr(), self.len) } + } + + /// Extracts a mutable slice of the entire buffer. + /// + /// Equivalent to `&mut s[..]`. + pub fn as_mut_slice(&mut self) -> &mut [u8] { + // SAFETY: The pointer is valid and `len` bytes are initialized. + unsafe { slice::from_raw_parts_mut(self.as_mut_ptr(), self.len) } + } + + /// Drops the all the contents of the buffer, setting its length to `0`. + #[inline] + pub fn clear(&mut self) { + self.len = 0; + } + + /// Reserves capacity for at least `additional` more bytes to be inserted + /// in the given `IoBufferMut`. The collection may reserve more space to + /// speculatively avoid frequent reallocations. After calling `reserve`, + /// capacity will be greater than or equal to `self.len() + additional`. + /// Does nothing if capacity is already sufficient. + /// + /// # Panics + /// + /// Panics if the new capacity exceeds `isize::MAX` _bytes_. + pub fn reserve(&mut self, additional: usize) { + if additional > self.capacity() - self.len() { + self.reserve_inner(additional); + } + } + + fn reserve_inner(&mut self, additional: usize) { + let Some(required_cap) = self.len().checked_add(additional) else { + capacity_overflow() + }; + + let old_capacity = self.capacity(); + let align = self.align(); + // This guarantees exponential growth. The doubling cannot overflow + // because `cap <= isize::MAX` and the type of `cap` is `usize`. + let cap = cmp::max(old_capacity * 2, required_cap); + + if !is_valid_alloc(cap) { + capacity_overflow() + } + let new_layout = Layout::from_size_align(cap, self.align()).expect("Invalid layout"); + + let old_ptr = self.as_mut_ptr(); + + // SAFETY: old allocation was allocated with std::alloc::alloc with the same layout, + // and we panics on null pointer. + let (ptr, cap) = unsafe { + let old_layout = Layout::from_size_align_unchecked(old_capacity, align); + let ptr = alloc::realloc(old_ptr, old_layout, new_layout.size()); + if ptr.is_null() { + alloc::handle_alloc_error(new_layout); + } + (AlignedBufferPtr(ptr), cap) + }; + + self.ptr = ptr; + self.capacity = cap; + } + + /// Shortens the buffer, keeping the first len bytes. + pub fn truncate(&mut self, len: usize) { + if len > self.len { + return; + } + self.len = len; + } + + /// Consumes and leaks the `IoBufferMut`, returning a mutable reference to the contents, &'a mut [u8]. + pub fn leak<'a>(self) -> &'a mut [u8] { + let mut buf = ManuallyDrop::new(self); + // SAFETY: leaking the buffer as intended. + unsafe { slice::from_raw_parts_mut(buf.as_mut_ptr(), buf.len) } + } +} + +fn capacity_overflow() -> ! { + panic!("capacity overflow") +} + +// We need to guarantee the following: +// * We don't ever allocate `> isize::MAX` byte-size objects. +// * We don't overflow `usize::MAX` and actually allocate too little. +// +// On 64-bit we just need to check for overflow since trying to allocate +// `> isize::MAX` bytes will surely fail. On 32-bit and 16-bit we need to add +// an extra guard for this in case we're running on a platform which can use +// all 4GB in user-space, e.g., PAE or x32. +#[inline] +fn is_valid_alloc(alloc_size: usize) -> bool { + !(usize::BITS < 64 && alloc_size > isize::MAX as usize) +} + +impl Drop for RawAlignedBuffer { + fn drop(&mut self) { + // SAFETY: memory was allocated with std::alloc::alloc with the same layout. + unsafe { + alloc::dealloc( + self.as_mut_ptr(), + Layout::from_size_align_unchecked(self.capacity, self.align.align()), + ) + } + } +} diff --git a/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/slice.rs b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/slice.rs new file mode 100644 index 0000000000..6cecf34c1c --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/aligned_buffer/slice.rs @@ -0,0 +1,40 @@ +use std::ops::{Deref, DerefMut}; + +use super::alignment::{Alignment, ConstAlign}; + +/// Newtype for an aligned slice. +pub struct AlignedSlice<'a, const N: usize, A: Alignment> { + /// underlying byte slice + buf: &'a mut [u8; N], + /// alignment marker + _align: A, +} + +impl<'a, const N: usize, const A: usize> AlignedSlice<'a, N, ConstAlign> { + /// Create a new aligned slice from a mutable byte slice. The input must already satisify the alignment. + pub unsafe fn new_unchecked(buf: &'a mut [u8; N]) -> Self { + let _align = ConstAlign::; + assert_eq!(buf.as_ptr().align_offset(_align.align()), 0); + AlignedSlice { buf, _align } + } +} + +impl<'a, const N: usize, A: Alignment> Deref for AlignedSlice<'a, N, A> { + type Target = [u8; N]; + + fn deref(&self) -> &Self::Target { + self.buf + } +} + +impl<'a, const N: usize, A: Alignment> DerefMut for AlignedSlice<'a, N, A> { + fn deref_mut(&mut self) -> &mut Self::Target { + self.buf + } +} + +impl<'a, const N: usize, A: Alignment> AsRef<[u8; N]> for AlignedSlice<'a, N, A> { + fn as_ref(&self) -> &[u8; N] { + self.buf + } +} diff --git a/pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs b/pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs new file mode 100644 index 0000000000..dba695196e --- /dev/null +++ b/pageserver/src/virtual_file/owned_buffers_io/io_buf_aligned.rs @@ -0,0 +1,9 @@ +use tokio_epoll_uring::IoBufMut; + +use crate::virtual_file::{IoBufferMut, PageWriteGuardBuf}; + +pub trait IoBufAlignedMut: IoBufMut {} + +impl IoBufAlignedMut for IoBufferMut {} + +impl IoBufAlignedMut for PageWriteGuardBuf {} diff --git a/pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs b/pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs index 7c773b6b21..c3940cf6ce 100644 --- a/pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs +++ b/pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs @@ -1,5 +1,6 @@ //! See [`FullSlice`]. +use crate::virtual_file::{IoBuffer, IoBufferMut}; use bytes::{Bytes, BytesMut}; use std::ops::{Deref, Range}; use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice}; @@ -76,3 +77,5 @@ macro_rules! impl_io_buf_ext { impl_io_buf_ext!(Bytes); impl_io_buf_ext!(BytesMut); impl_io_buf_ext!(Vec); +impl_io_buf_ext!(IoBufferMut); +impl_io_buf_ext!(IoBuffer); From 34b6bd416a8df8cf0d51f707beaca30dbdbe2adc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Arpad=20M=C3=BCller?= Date: Mon, 21 Oct 2024 17:33:05 +0200 Subject: [PATCH 10/12] offloaded timeline list API (#9461) Add a way to list the offloaded timelines. Before, one had to look at logs to figure out if a timeline has been offloaded or not, or use the non-presence of a certain timeline in the list of normal timelines. Now, one can list them directly. Part of #8088 --- libs/pageserver_api/src/models.rs | 17 +++++++ pageserver/src/http/routes.rs | 80 ++++++++++++++++++++++++++++++- pageserver/src/tenant.rs | 25 +++++++++- 3 files changed, 120 insertions(+), 2 deletions(-) diff --git a/libs/pageserver_api/src/models.rs b/libs/pageserver_api/src/models.rs index 5b0b6bebe3..e08bf40801 100644 --- a/libs/pageserver_api/src/models.rs +++ b/libs/pageserver_api/src/models.rs @@ -684,6 +684,23 @@ pub struct TimelineArchivalConfigRequest { pub state: TimelineArchivalState, } +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct TimelinesInfoAndOffloaded { + pub timelines: Vec, + pub offloaded: Vec, +} + +/// Analog of [`TimelineInfo`] for offloaded timelines. +#[derive(Debug, Serialize, Deserialize, Clone)] +pub struct OffloadedTimelineInfo { + pub tenant_id: TenantShardId, + pub timeline_id: TimelineId, + /// Whether the timeline has a parent it has been branched off from or not + pub ancestor_timeline_id: Option, + /// Whether to retain the branch lsn at the ancestor or not + pub ancestor_retain_lsn: Option, +} + /// This represents the output of the "timeline_detail" and "timeline_list" API calls. #[derive(Debug, Serialize, Deserialize, Clone)] pub struct TimelineInfo { diff --git a/pageserver/src/http/routes.rs b/pageserver/src/http/routes.rs index 8f928fd81b..a254f1683d 100644 --- a/pageserver/src/http/routes.rs +++ b/pageserver/src/http/routes.rs @@ -26,6 +26,7 @@ use pageserver_api::models::LocationConfigListResponse; use pageserver_api::models::LocationConfigMode; use pageserver_api::models::LsnLease; use pageserver_api::models::LsnLeaseRequest; +use pageserver_api::models::OffloadedTimelineInfo; use pageserver_api::models::ShardParameters; use pageserver_api::models::TenantDetails; use pageserver_api::models::TenantLocationConfigRequest; @@ -37,6 +38,7 @@ use pageserver_api::models::TenantShardSplitRequest; use pageserver_api::models::TenantShardSplitResponse; use pageserver_api::models::TenantSorting; use pageserver_api::models::TimelineArchivalConfigRequest; +use pageserver_api::models::TimelinesInfoAndOffloaded; use pageserver_api::models::TopTenantShardItem; use pageserver_api::models::TopTenantShardsRequest; use pageserver_api::models::TopTenantShardsResponse; @@ -81,6 +83,7 @@ use crate::tenant::timeline::CompactFlags; use crate::tenant::timeline::CompactionError; use crate::tenant::timeline::Timeline; use crate::tenant::GetTimelineError; +use crate::tenant::OffloadedTimeline; use crate::tenant::{LogicalSizeCalculationCause, PageReconstructError}; use crate::{disk_usage_eviction_task, tenant}; use pageserver_api::models::{ @@ -477,6 +480,22 @@ async fn build_timeline_info_common( Ok(info) } +fn build_timeline_offloaded_info(offloaded: &Arc) -> OffloadedTimelineInfo { + let &OffloadedTimeline { + tenant_shard_id, + timeline_id, + ancestor_retain_lsn, + ancestor_timeline_id, + .. + } = offloaded.as_ref(); + OffloadedTimelineInfo { + tenant_id: tenant_shard_id, + timeline_id, + ancestor_retain_lsn, + ancestor_timeline_id, + } +} + // healthcheck handler async fn status_handler( request: Request, @@ -643,7 +662,7 @@ async fn timeline_list_handler( ) .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id)) .await - .context("Failed to convert tenant timeline {timeline_id} into the local one: {e:?}") + .context("Failed to build timeline info") .map_err(ApiError::InternalServerError)?; response_data.push(timeline_info); @@ -658,6 +677,62 @@ async fn timeline_list_handler( json_response(StatusCode::OK, response_data) } +async fn timeline_and_offloaded_list_handler( + request: Request, + _cancel: CancellationToken, +) -> Result, ApiError> { + let tenant_shard_id: TenantShardId = parse_request_param(&request, "tenant_shard_id")?; + let include_non_incremental_logical_size: Option = + parse_query_param(&request, "include-non-incremental-logical-size")?; + let force_await_initial_logical_size: Option = + parse_query_param(&request, "force-await-initial-logical-size")?; + check_permission(&request, Some(tenant_shard_id.tenant_id))?; + + let state = get_state(&request); + let ctx = RequestContext::new(TaskKind::MgmtRequest, DownloadBehavior::Download); + + let response_data = async { + let tenant = state + .tenant_manager + .get_attached_tenant_shard(tenant_shard_id)?; + + tenant.wait_to_become_active(ACTIVE_TENANT_TIMEOUT).await?; + + let (timelines, offloadeds) = tenant.list_timelines_and_offloaded(); + + let mut timeline_infos = Vec::with_capacity(timelines.len()); + for timeline in timelines { + let timeline_info = build_timeline_info( + &timeline, + include_non_incremental_logical_size.unwrap_or(false), + force_await_initial_logical_size.unwrap_or(false), + &ctx, + ) + .instrument(info_span!("build_timeline_info", timeline_id = %timeline.timeline_id)) + .await + .context("Failed to build timeline info") + .map_err(ApiError::InternalServerError)?; + + timeline_infos.push(timeline_info); + } + let offloaded_infos = offloadeds + .into_iter() + .map(|offloaded| build_timeline_offloaded_info(&offloaded)) + .collect::>(); + let res = TimelinesInfoAndOffloaded { + timelines: timeline_infos, + offloaded: offloaded_infos, + }; + Ok::(res) + } + .instrument(info_span!("timeline_and_offloaded_list", + tenant_id = %tenant_shard_id.tenant_id, + shard_id = %tenant_shard_id.shard_slug())) + .await?; + + json_response(StatusCode::OK, response_data) +} + async fn timeline_preserve_initdb_handler( request: Request, _cancel: CancellationToken, @@ -2993,6 +3068,9 @@ pub fn make_router( .get("/v1/tenant/:tenant_shard_id/timeline", |r| { api_handler(r, timeline_list_handler) }) + .get("/v1/tenant/:tenant_shard_id/timeline_and_offloaded", |r| { + api_handler(r, timeline_and_offloaded_list_handler) + }) .post("/v1/tenant/:tenant_shard_id/timeline", |r| { api_handler(r, timeline_create_handler) }) diff --git a/pageserver/src/tenant.rs b/pageserver/src/tenant.rs index 1066d165cd..41d21ef041 100644 --- a/pageserver/src/tenant.rs +++ b/pageserver/src/tenant.rs @@ -1755,7 +1755,7 @@ impl Tenant { } /// Lists timelines the tenant contains. - /// Up to tenant's implementation to omit certain timelines that ar not considered ready for use. + /// It's up to callers to omit certain timelines that are not considered ready for use. pub fn list_timelines(&self) -> Vec> { self.timelines .lock() @@ -1765,6 +1765,29 @@ impl Tenant { .collect() } + /// Lists timelines the tenant manages, including offloaded ones. + /// + /// It's up to callers to omit certain timelines that are not considered ready for use. + pub fn list_timelines_and_offloaded( + &self, + ) -> (Vec>, Vec>) { + let timelines = self + .timelines + .lock() + .unwrap() + .values() + .map(Arc::clone) + .collect(); + let offloaded = self + .timelines_offloaded + .lock() + .unwrap() + .values() + .map(Arc::clone) + .collect(); + (timelines, offloaded) + } + pub fn list_timeline_ids(&self) -> Vec { self.timelines.lock().unwrap().keys().cloned().collect() } From 94369af7825ca26fb5aea805f2c85bdb877ceb74 Mon Sep 17 00:00:00 2001 From: David Gomes Date: Mon, 21 Oct 2024 18:39:30 -0500 Subject: [PATCH 11/12] chore(compute): bumps pg_session_jwt to latest version (#9474) --- compute/compute-node.Dockerfile | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/compute/compute-node.Dockerfile b/compute/compute-node.Dockerfile index 74970696b5..6451e309f0 100644 --- a/compute/compute-node.Dockerfile +++ b/compute/compute-node.Dockerfile @@ -975,8 +975,8 @@ ARG PG_VERSION RUN case "${PG_VERSION}" in "v17") \ echo "pg_session_jwt does not yet have a release that supports pg17" && exit 0;; \ esac && \ - wget https://github.com/neondatabase/pg_session_jwt/archive/e642528f429dd3f5403845a50191b78d434b84a6.tar.gz -O pg_session_jwt.tar.gz && \ - echo "1a69210703cc91224785e59a0a67562dd9eed9a0914ac84b11447582ca0d5b93 pg_session_jwt.tar.gz" | sha256sum --check && \ + wget https://github.com/neondatabase/pg_session_jwt/archive/e1310b08ba51377a19e0559e4d1194883b9b2ba2.tar.gz -O pg_session_jwt.tar.gz && \ + echo "837932a077888d5545fd54b0abcc79e5f8e37017c2769a930afc2f5c94df6f4e pg_session_jwt.tar.gz" | sha256sum --check && \ mkdir pg_session_jwt-src && cd pg_session_jwt-src && tar xzf ../pg_session_jwt.tar.gz --strip-components=1 -C . && \ sed -i 's/pgrx = "=0.11.3"/pgrx = { version = "=0.11.3", features = [ "unsafe-postgres" ] }/g' Cargo.toml && \ cargo pgrx install --release From 1e8e04bb2c9b2cdb17f680c8d0df697289114e17 Mon Sep 17 00:00:00 2001 From: Arseny Sher Date: Tue, 22 Oct 2024 09:11:36 +0300 Subject: [PATCH 12/12] safekeeper: refactor timeline initialization (#9362) Always do timeline init through atomic rename of temp directory. Add GlobalTimelines::load_temp_timeline which does this, and use it from both pull_timeline and basic timeline creation. Fixes a collection of issues: - previously timeline creation didn't really flushed cfile to disk due to 'nothing to do if state didn't change' check; - even if it did, without tmp dir it is possible to lose the cfile but leave timeline dir in place, making it look corrupted; - tenant directory creation fsync was missing in timeline creation; - pull_timeline is now protected from concurrent both itself and timeline creation; - now global timelines map entry got special CreationInProgress entry type which prevents from anyone getting access to timeline while it is being created (previously one could get access to it, but it was locked during creation, which is valid but confusing if creation failed). fixes #8927 --- safekeeper/src/control_file.rs | 21 +- safekeeper/src/copy_timeline.rs | 11 +- safekeeper/src/pull_timeline.rs | 106 +----- safekeeper/src/receive_wal.rs | 3 +- safekeeper/src/state.rs | 39 +- safekeeper/src/timeline.rs | 91 +---- safekeeper/src/timelines_global_map.rs | 339 +++++++++++------- safekeeper/src/wal_storage.rs | 10 +- .../tests/walproposer_sim/safekeeper.rs | 20 +- 9 files changed, 290 insertions(+), 350 deletions(-) diff --git a/safekeeper/src/control_file.rs b/safekeeper/src/control_file.rs index 8b252b4ab4..cd82e43780 100644 --- a/safekeeper/src/control_file.rs +++ b/safekeeper/src/control_file.rs @@ -66,22 +66,25 @@ impl FileStorage { }) } - /// Create file storage for a new timeline, but don't persist it yet. - pub fn create_new( - timeline_dir: Utf8PathBuf, + /// Create and reliably persist new control file at given location. + /// + /// Note: we normally call this in temp directory for atomic init, so + /// interested in FileStorage as a result only in tests. + pub async fn create_new( + dir: Utf8PathBuf, conf: &SafeKeeperConf, state: TimelinePersistentState, ) -> Result { // we don't support creating new timelines in offloaded state assert!(matches!(state.eviction_state, EvictionState::Present)); - let store = FileStorage { - timeline_dir, + let mut store = FileStorage { + timeline_dir: dir, no_sync: conf.no_sync, - state, + state: state.clone(), last_persist_at: Instant::now(), }; - + store.persist(&state).await?; Ok(store) } @@ -190,8 +193,6 @@ impl TimelinePersistentState { impl Storage for FileStorage { /// Persists state durably to the underlying storage. - /// - /// For a description, see . async fn persist(&mut self, s: &TimelinePersistentState) -> Result<()> { let _timer = PERSIST_CONTROL_FILE_SECONDS.start_timer(); @@ -269,7 +270,7 @@ mod test { .await .expect("failed to create timeline dir"); let state = TimelinePersistentState::empty(); - let storage = FileStorage::create_new(timeline_dir, conf, state.clone())?; + let storage = FileStorage::create_new(timeline_dir, conf, state.clone()).await?; Ok((storage, state)) } diff --git a/safekeeper/src/copy_timeline.rs b/safekeeper/src/copy_timeline.rs index 220988c3ce..52b13dc5e3 100644 --- a/safekeeper/src/copy_timeline.rs +++ b/safekeeper/src/copy_timeline.rs @@ -12,10 +12,10 @@ use tracing::{info, warn}; use utils::{id::TenantTimelineId, lsn::Lsn}; use crate::{ - control_file::{FileStorage, Storage}, - pull_timeline::{create_temp_timeline_dir, load_temp_timeline, validate_temp_timeline}, + control_file::FileStorage, state::TimelinePersistentState, timeline::{Timeline, TimelineError, WalResidentTimeline}, + timelines_global_map::{create_temp_timeline_dir, validate_temp_timeline}, wal_backup::copy_s3_segments, wal_storage::{wal_file_paths, WalReader}, GlobalTimelines, @@ -149,17 +149,16 @@ pub async fn handle_request(request: Request) -> Result<()> { vec![], request.until_lsn, start_lsn, - ); + )?; new_state.timeline_start_lsn = start_lsn; new_state.peer_horizon_lsn = request.until_lsn; new_state.backup_lsn = new_backup_lsn; - let mut file_storage = FileStorage::create_new(tli_dir_path.clone(), conf, new_state.clone())?; - file_storage.persist(&new_state).await?; + FileStorage::create_new(tli_dir_path.clone(), conf, new_state.clone()).await?; // now we have a ready timeline in a temp directory validate_temp_timeline(conf, request.destination_ttid, &tli_dir_path).await?; - load_temp_timeline(conf, request.destination_ttid, &tli_dir_path).await?; + GlobalTimelines::load_temp_timeline(request.destination_ttid, &tli_dir_path, true).await?; Ok(()) } diff --git a/safekeeper/src/pull_timeline.rs b/safekeeper/src/pull_timeline.rs index c772ae6de7..c7f5165f90 100644 --- a/safekeeper/src/pull_timeline.rs +++ b/safekeeper/src/pull_timeline.rs @@ -1,7 +1,6 @@ use anyhow::{anyhow, bail, Context, Result}; use bytes::Bytes; use camino::Utf8PathBuf; -use camino_tempfile::Utf8TempDir; use chrono::{DateTime, Utc}; use futures::{SinkExt, StreamExt, TryStreamExt}; use postgres_ffi::{XLogFileName, XLogSegNo, PG_TLI}; @@ -9,7 +8,6 @@ use serde::{Deserialize, Serialize}; use std::{ cmp::min, io::{self, ErrorKind}, - sync::Arc, }; use tokio::{fs::OpenOptions, io::AsyncWrite, sync::mpsc, task}; use tokio_tar::{Archive, Builder, Header}; @@ -20,7 +18,7 @@ use tokio_util::{ use tracing::{error, info, instrument}; use crate::{ - control_file::{self, CONTROL_FILE_NAME}, + control_file::CONTROL_FILE_NAME, debug_dump, http::{ client::{self, Client}, @@ -28,13 +26,14 @@ use crate::{ }, safekeeper::Term, state::TimelinePersistentState, - timeline::{get_tenant_dir, get_timeline_dir, Timeline, TimelineError, WalResidentTimeline}, + timeline::WalResidentTimeline, + timelines_global_map::{create_temp_timeline_dir, validate_temp_timeline}, wal_backup, - wal_storage::{self, open_wal_file, Storage}, - GlobalTimelines, SafeKeeperConf, + wal_storage::open_wal_file, + GlobalTimelines, }; use utils::{ - crashsafe::{durable_rename, fsync_async_opt}, + crashsafe::fsync_async_opt, id::{NodeId, TenantId, TenantTimelineId, TimelineId}, logging::SecretString, lsn::Lsn, @@ -428,100 +427,9 @@ async fn pull_timeline( assert!(status.commit_lsn <= status.flush_lsn); // Finally, load the timeline. - let _tli = load_temp_timeline(conf, ttid, &tli_dir_path).await?; + let _tli = GlobalTimelines::load_temp_timeline(ttid, &tli_dir_path, false).await?; Ok(Response { safekeeper_host: host, }) } - -/// Create temp directory for a new timeline. It needs to be located on the same -/// filesystem as the rest of the timelines. It will be automatically deleted when -/// Utf8TempDir goes out of scope. -pub async fn create_temp_timeline_dir( - conf: &SafeKeeperConf, - ttid: TenantTimelineId, -) -> Result<(Utf8TempDir, Utf8PathBuf)> { - // conf.workdir is usually /storage/safekeeper/data - // will try to transform it into /storage/safekeeper/tmp - let temp_base = conf - .workdir - .parent() - .ok_or(anyhow::anyhow!("workdir has no parent"))? - .join("tmp"); - - tokio::fs::create_dir_all(&temp_base).await?; - - let tli_dir = camino_tempfile::Builder::new() - .suffix("_temptli") - .prefix(&format!("{}_{}_", ttid.tenant_id, ttid.timeline_id)) - .tempdir_in(temp_base)?; - - let tli_dir_path = tli_dir.path().to_path_buf(); - - Ok((tli_dir, tli_dir_path)) -} - -/// Do basic validation of a temp timeline, before moving it to the global map. -pub async fn validate_temp_timeline( - conf: &SafeKeeperConf, - ttid: TenantTimelineId, - path: &Utf8PathBuf, -) -> Result<(Lsn, Lsn)> { - let control_path = path.join("safekeeper.control"); - - let control_store = control_file::FileStorage::load_control_file(control_path)?; - if control_store.server.wal_seg_size == 0 { - bail!("wal_seg_size is not set"); - } - - let wal_store = wal_storage::PhysicalStorage::new(&ttid, path.clone(), conf, &control_store)?; - - let commit_lsn = control_store.commit_lsn; - let flush_lsn = wal_store.flush_lsn(); - - Ok((commit_lsn, flush_lsn)) -} - -/// Move timeline from a temp directory to the main storage, and load it to the global map. -/// -/// This operation is done under a lock to prevent bugs if several concurrent requests are -/// trying to load the same timeline. Note that it doesn't guard against creating the -/// timeline with the same ttid, but no one should be doing this anyway. -pub async fn load_temp_timeline( - conf: &SafeKeeperConf, - ttid: TenantTimelineId, - tmp_path: &Utf8PathBuf, -) -> Result> { - // Take a lock to prevent concurrent loadings - let load_lock = GlobalTimelines::loading_lock().await; - let guard = load_lock.lock().await; - - if !matches!(GlobalTimelines::get(ttid), Err(TimelineError::NotFound(_))) { - bail!("timeline already exists, cannot overwrite it") - } - - // Move timeline dir to the correct location - let timeline_path = get_timeline_dir(conf, &ttid); - - info!( - "moving timeline {} from {} to {}", - ttid, tmp_path, timeline_path - ); - tokio::fs::create_dir_all(get_tenant_dir(conf, &ttid.tenant_id)).await?; - // fsync tenant dir creation - fsync_async_opt(&conf.workdir, !conf.no_sync).await?; - durable_rename(tmp_path, &timeline_path, !conf.no_sync).await?; - - let tli = GlobalTimelines::load_timeline(&guard, ttid) - .await - .context("Failed to load timeline after copy")?; - - info!( - "loaded timeline {}, flush_lsn={}", - ttid, - tli.get_flush_lsn().await - ); - - Ok(tli) -} diff --git a/safekeeper/src/receive_wal.rs b/safekeeper/src/receive_wal.rs index 2a9ca85299..3dbf72298f 100644 --- a/safekeeper/src/receive_wal.rs +++ b/safekeeper/src/receive_wal.rs @@ -339,7 +339,8 @@ impl<'a, IO: AsyncRead + AsyncWrite + Unpin> NetworkReader<'a, IO> { }; let tli = GlobalTimelines::create(self.ttid, server_info, Lsn::INVALID, Lsn::INVALID) - .await?; + .await + .context("create timeline")?; tli.wal_residence_guard().await? } _ => { diff --git a/safekeeper/src/state.rs b/safekeeper/src/state.rs index 8ae749ded5..8dd873ee77 100644 --- a/safekeeper/src/state.rs +++ b/safekeeper/src/state.rs @@ -3,7 +3,7 @@ use std::{cmp::max, ops::Deref}; -use anyhow::Result; +use anyhow::{bail, Result}; use safekeeper_api::models::TimelineTermBumpResponse; use serde::{Deserialize, Serialize}; use utils::{ @@ -13,7 +13,11 @@ use utils::{ use crate::{ control_file, - safekeeper::{AcceptorState, PersistedPeerInfo, PgUuid, ServerInfo, Term, TermHistory}, + safekeeper::{ + AcceptorState, PersistedPeerInfo, PgUuid, ServerInfo, Term, TermHistory, + UNKNOWN_SERVER_VERSION, + }, + timeline::TimelineError, wal_backup_partial::{self}, }; @@ -91,8 +95,24 @@ impl TimelinePersistentState { peers: Vec, commit_lsn: Lsn, local_start_lsn: Lsn, - ) -> TimelinePersistentState { - TimelinePersistentState { + ) -> anyhow::Result { + if server_info.wal_seg_size == 0 { + bail!(TimelineError::UninitializedWalSegSize(*ttid)); + } + + if server_info.pg_version == UNKNOWN_SERVER_VERSION { + bail!(TimelineError::UninitialinzedPgVersion(*ttid)); + } + + if commit_lsn < local_start_lsn { + bail!( + "commit_lsn {} is smaller than local_start_lsn {}", + commit_lsn, + local_start_lsn + ); + } + + Ok(TimelinePersistentState { tenant_id: ttid.tenant_id, timeline_id: ttid.timeline_id, acceptor_state: AcceptorState { @@ -115,24 +135,23 @@ impl TimelinePersistentState { ), partial_backup: wal_backup_partial::State::default(), eviction_state: EvictionState::Present, - } + }) } #[cfg(test)] pub fn empty() -> Self { - use crate::safekeeper::UNKNOWN_SERVER_VERSION; - TimelinePersistentState::new( &TenantTimelineId::empty(), ServerInfo { - pg_version: UNKNOWN_SERVER_VERSION, /* Postgres server version */ - system_id: 0, /* Postgres system identifier */ - wal_seg_size: 0, + pg_version: 17, /* Postgres server version */ + system_id: 0, /* Postgres system identifier */ + wal_seg_size: 16 * 1024 * 1024, }, vec![], Lsn::INVALID, Lsn::INVALID, ) + .unwrap() } } diff --git a/safekeeper/src/timeline.rs b/safekeeper/src/timeline.rs index 41b9490088..dd4d161226 100644 --- a/safekeeper/src/timeline.rs +++ b/safekeeper/src/timeline.rs @@ -27,11 +27,11 @@ use utils::{ use storage_broker::proto::SafekeeperTimelineInfo; use storage_broker::proto::TenantTimelineId as ProtoTenantTimelineId; +use crate::control_file; use crate::rate_limit::RateLimiter; use crate::receive_wal::WalReceivers; use crate::safekeeper::{ - AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, ServerInfo, Term, TermLsn, - INVALID_TERM, + AcceptorProposerMessage, ProposerAcceptorMessage, SafeKeeper, Term, TermLsn, }; use crate::send_wal::WalSenders; use crate::state::{EvictionState, TimelineMemState, TimelinePersistentState, TimelineState}; @@ -40,7 +40,6 @@ use crate::timeline_manager::{AtomicStatus, ManagerCtl}; use crate::timelines_set::TimelinesSet; use crate::wal_backup::{self, remote_timeline_path}; use crate::wal_backup_partial::PartialRemoteSegment; -use crate::{control_file, safekeeper::UNKNOWN_SERVER_VERSION}; use crate::metrics::{FullTimelineInfo, WalStorageMetrics, MISC_OPERATION_SECONDS}; use crate::wal_storage::{Storage as wal_storage_iface, WalReader}; @@ -326,44 +325,6 @@ pub struct SharedState { } impl SharedState { - /// Initialize fresh timeline state without persisting anything to disk. - fn create_new( - conf: &SafeKeeperConf, - ttid: &TenantTimelineId, - state: TimelinePersistentState, - ) -> Result { - if state.server.wal_seg_size == 0 { - bail!(TimelineError::UninitializedWalSegSize(*ttid)); - } - - if state.server.pg_version == UNKNOWN_SERVER_VERSION { - bail!(TimelineError::UninitialinzedPgVersion(*ttid)); - } - - if state.commit_lsn < state.local_start_lsn { - bail!( - "commit_lsn {} is higher than local_start_lsn {}", - state.commit_lsn, - state.local_start_lsn - ); - } - - // We don't want to write anything to disk, because we may have existing timeline there. - // These functions should not change anything on disk. - let timeline_dir = get_timeline_dir(conf, ttid); - let control_store = - control_file::FileStorage::create_new(timeline_dir.clone(), conf, state)?; - let wal_store = - wal_storage::PhysicalStorage::new(ttid, timeline_dir, conf, &control_store)?; - let sk = SafeKeeper::new(TimelineState::new(control_store), wal_store, conf.my_id)?; - - Ok(Self { - sk: StateSK::Loaded(sk), - peers_info: PeersInfo(vec![]), - wal_removal_on_hold: false, - }) - } - /// Restore SharedState from control file. If file doesn't exist, bails out. fn restore(conf: &SafeKeeperConf, ttid: &TenantTimelineId) -> Result { let timeline_dir = get_timeline_dir(conf, ttid); @@ -450,6 +411,8 @@ pub enum TimelineError { Cancelled(TenantTimelineId), #[error("Timeline {0} was not found in global map")] NotFound(TenantTimelineId), + #[error("Timeline {0} creation is in progress")] + CreationInProgress(TenantTimelineId), #[error("Timeline {0} exists on disk, but wasn't loaded on startup")] Invalid(TenantTimelineId), #[error("Timeline {0} is already exists")] @@ -514,7 +477,7 @@ pub struct Timeline { impl Timeline { /// Load existing timeline from disk. - pub fn load_timeline(conf: &SafeKeeperConf, ttid: TenantTimelineId) -> Result { + pub fn load_timeline(conf: &SafeKeeperConf, ttid: TenantTimelineId) -> Result> { let _enter = info_span!("load_timeline", timeline = %ttid.timeline_id).entered(); let shared_state = SharedState::restore(conf, &ttid)?; @@ -528,7 +491,7 @@ impl Timeline { let walreceivers = WalReceivers::new(); let remote_path = remote_timeline_path(&ttid)?; - Ok(Timeline { + Ok(Arc::new(Timeline { ttid, remote_path, commit_lsn_watch_tx, @@ -547,47 +510,7 @@ impl Timeline { wal_backup_active: AtomicBool::new(false), last_removed_segno: AtomicU64::new(0), mgr_status: AtomicStatus::new(), - }) - } - - /// Create a new timeline, which is not yet persisted to disk. - pub fn create_empty( - conf: &SafeKeeperConf, - ttid: TenantTimelineId, - server_info: ServerInfo, - commit_lsn: Lsn, - local_start_lsn: Lsn, - ) -> Result { - let (commit_lsn_watch_tx, commit_lsn_watch_rx) = watch::channel(Lsn::INVALID); - let (term_flush_lsn_watch_tx, term_flush_lsn_watch_rx) = - watch::channel(TermLsn::from((INVALID_TERM, Lsn::INVALID))); - let (shared_state_version_tx, shared_state_version_rx) = watch::channel(0); - - let state = - TimelinePersistentState::new(&ttid, server_info, vec![], commit_lsn, local_start_lsn); - - let walreceivers = WalReceivers::new(); - let remote_path = remote_timeline_path(&ttid)?; - Ok(Timeline { - ttid, - remote_path, - commit_lsn_watch_tx, - commit_lsn_watch_rx, - term_flush_lsn_watch_tx, - term_flush_lsn_watch_rx, - shared_state_version_tx, - shared_state_version_rx, - mutex: RwLock::new(SharedState::create_new(conf, &ttid, state)?), - walsenders: WalSenders::new(walreceivers.clone()), - walreceivers, - cancel: CancellationToken::default(), - timeline_dir: get_timeline_dir(conf, &ttid), - manager_ctl: ManagerCtl::new(), - broker_active: AtomicBool::new(false), - wal_backup_active: AtomicBool::new(false), - last_removed_segno: AtomicU64::new(0), - mgr_status: AtomicStatus::new(), - }) + })) } /// Initialize fresh timeline on disk and start background tasks. If init diff --git a/safekeeper/src/timelines_global_map.rs b/safekeeper/src/timelines_global_map.rs index 866cde3339..538bb6e5d2 100644 --- a/safekeeper/src/timelines_global_map.rs +++ b/safekeeper/src/timelines_global_map.rs @@ -5,11 +5,14 @@ use crate::defaults::DEFAULT_EVICTION_CONCURRENCY; use crate::rate_limit::RateLimiter; use crate::safekeeper::ServerInfo; +use crate::state::TimelinePersistentState; use crate::timeline::{get_tenant_dir, get_timeline_dir, Timeline, TimelineError}; use crate::timelines_set::TimelinesSet; -use crate::SafeKeeperConf; +use crate::wal_storage::Storage; +use crate::{control_file, wal_storage, SafeKeeperConf}; use anyhow::{bail, Context, Result}; use camino::Utf8PathBuf; +use camino_tempfile::Utf8TempDir; use once_cell::sync::Lazy; use serde::Serialize; use std::collections::HashMap; @@ -17,12 +20,22 @@ use std::str::FromStr; use std::sync::atomic::Ordering; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; +use tokio::fs; use tracing::*; +use utils::crashsafe::{durable_rename, fsync_async_opt}; use utils::id::{TenantId, TenantTimelineId, TimelineId}; use utils::lsn::Lsn; +// Timeline entry in the global map: either a ready timeline, or mark that it is +// being created. +#[derive(Clone)] +enum GlobalMapTimeline { + CreationInProgress, + Timeline(Arc), +} + struct GlobalTimelinesState { - timelines: HashMap>, + timelines: HashMap, // A tombstone indicates this timeline used to exist has been deleted. These are used to prevent // on-demand timeline creation from recreating deleted timelines. This is only soft-enforced, as @@ -31,13 +44,9 @@ struct GlobalTimelinesState { conf: Option, broker_active_set: Arc, - load_lock: Arc>, global_rate_limiter: RateLimiter, } -// Used to prevent concurrent timeline loading. -pub struct TimelineLoadLock; - impl GlobalTimelinesState { /// Get configuration, which must be set once during init. fn get_conf(&self) -> &SafeKeeperConf { @@ -55,22 +64,16 @@ impl GlobalTimelinesState { ) } - /// Insert timeline into the map. Returns error if timeline with the same id already exists. - fn try_insert(&mut self, timeline: Arc) -> Result<()> { - let ttid = timeline.ttid; - if self.timelines.contains_key(&ttid) { - bail!(TimelineError::AlreadyExists(ttid)); - } - self.timelines.insert(ttid, timeline); - Ok(()) - } - - /// Get timeline from the map. Returns error if timeline doesn't exist. + /// Get timeline from the map. Returns error if timeline doesn't exist or + /// creation is in progress. fn get(&self, ttid: &TenantTimelineId) -> Result, TimelineError> { - self.timelines - .get(ttid) - .cloned() - .ok_or(TimelineError::NotFound(*ttid)) + match self.timelines.get(ttid).cloned() { + Some(GlobalMapTimeline::Timeline(tli)) => Ok(tli), + Some(GlobalMapTimeline::CreationInProgress) => { + Err(TimelineError::CreationInProgress(*ttid)) + } + None => Err(TimelineError::NotFound(*ttid)), + } } fn delete(&mut self, ttid: TenantTimelineId) { @@ -85,7 +88,6 @@ static TIMELINES_STATE: Lazy> = Lazy::new(|| { tombstones: HashMap::new(), conf: None, broker_active_set: Arc::new(TimelinesSet::default()), - load_lock: Arc::new(tokio::sync::Mutex::new(TimelineLoadLock)), global_rate_limiter: RateLimiter::new(1, 1), }) }); @@ -141,11 +143,10 @@ impl GlobalTimelines { /// Loads all timelines for the given tenant to memory. Returns fs::read_dir /// errors if any. /// - /// It is async for update_status_notify sake. Since TIMELINES_STATE lock is - /// sync and there is no important reason to make it async (it is always - /// held for a short while) we just lock and unlock it for each timeline -- - /// this function is called during init when nothing else is running, so - /// this is fine. + /// It is async, but TIMELINES_STATE lock is sync and there is no important + /// reason to make it async (it is always held for a short while), so we + /// just lock and unlock it for each timeline -- this function is called + /// during init when nothing else is running, so this is fine. async fn load_tenant_timelines(tenant_id: TenantId) -> Result<()> { let (conf, broker_active_set, partial_backup_rate_limiter) = { let state = TIMELINES_STATE.lock().unwrap(); @@ -163,14 +164,13 @@ impl GlobalTimelines { { let ttid = TenantTimelineId::new(tenant_id, timeline_id); match Timeline::load_timeline(&conf, ttid) { - Ok(timeline) => { - let tli = Arc::new(timeline); + Ok(tli) => { let mut shared_state = tli.write_shared_state().await; TIMELINES_STATE .lock() .unwrap() .timelines - .insert(ttid, tli.clone()); + .insert(ttid, GlobalMapTimeline::Timeline(tli.clone())); tli.bootstrap( &mut shared_state, &conf, @@ -199,51 +199,6 @@ impl GlobalTimelines { Ok(()) } - /// Take a lock for timeline loading. - pub async fn loading_lock() -> Arc> { - TIMELINES_STATE.lock().unwrap().load_lock.clone() - } - - /// Load timeline from disk to the memory. - pub async fn load_timeline<'a>( - _guard: &tokio::sync::MutexGuard<'a, TimelineLoadLock>, - ttid: TenantTimelineId, - ) -> Result> { - let (conf, broker_active_set, partial_backup_rate_limiter) = - TIMELINES_STATE.lock().unwrap().get_dependencies(); - - match Timeline::load_timeline(&conf, ttid) { - Ok(timeline) => { - let tli = Arc::new(timeline); - let mut shared_state = tli.write_shared_state().await; - - // TODO: prevent concurrent timeline creation/loading - { - let mut state = TIMELINES_STATE.lock().unwrap(); - - // We may be have been asked to load a timeline that was previously deleted (e.g. from `pull_timeline.rs`). We trust - // that the human doing this manual intervention knows what they are doing, and remove its tombstone. - if state.tombstones.remove(&ttid).is_some() { - warn!("Un-deleted timeline {ttid}"); - } - - state.timelines.insert(ttid, tli.clone()); - } - - tli.bootstrap( - &mut shared_state, - &conf, - broker_active_set, - partial_backup_rate_limiter, - ); - drop(shared_state); - Ok(tli) - } - // If we can't load a timeline, it's bad. Caller will figure it out. - Err(e) => bail!("failed to load timeline {}, reason: {:?}", ttid, e), - } - } - /// Get the number of timelines in the map. pub fn timelines_count() -> usize { TIMELINES_STATE.lock().unwrap().timelines.len() @@ -266,7 +221,7 @@ impl GlobalTimelines { commit_lsn: Lsn, local_start_lsn: Lsn, ) -> Result> { - let (conf, broker_active_set, partial_backup_rate_limiter) = { + let (conf, _, _) = { let state = TIMELINES_STATE.lock().unwrap(); if let Ok(timeline) = state.get(&ttid) { // Timeline already exists, return it. @@ -282,55 +237,146 @@ impl GlobalTimelines { info!("creating new timeline {}", ttid); - let timeline = Arc::new(Timeline::create_empty( - &conf, - ttid, - server_info, - commit_lsn, - local_start_lsn, - )?); + // Do on disk initialization in tmp dir. + let (_tmp_dir, tmp_dir_path) = create_temp_timeline_dir(&conf, ttid).await?; - // Take a lock and finish the initialization holding this mutex. No other threads - // can interfere with creation after we will insert timeline into the map. - { - let mut shared_state = timeline.write_shared_state().await; + // TODO: currently we create only cfile. It would be reasonable to + // immediately initialize first WAL segment as well. + let state = + TimelinePersistentState::new(&ttid, server_info, vec![], commit_lsn, local_start_lsn)?; + control_file::FileStorage::create_new(tmp_dir_path.clone(), &conf, state).await?; + let timeline = GlobalTimelines::load_temp_timeline(ttid, &tmp_dir_path, true).await?; + Ok(timeline) + } - // We can get a race condition here in case of concurrent create calls, but only - // in theory. create() will return valid timeline on the next try. - TIMELINES_STATE - .lock() - .unwrap() - .try_insert(timeline.clone())?; + /// Move timeline from a temp directory to the main storage, and load it to + /// the global map. Creating timeline in this way ensures atomicity: rename + /// is atomic, so either move of the whole datadir succeeds or it doesn't, + /// but corrupted data dir shouldn't be possible. + /// + /// We'd like to avoid holding map lock while doing IO, so it's a 3 step + /// process: + /// 1) check the global map that timeline doesn't exist and mark that we're + /// creating it; + /// 2) move the directory and load the timeline + /// 3) take lock again and insert the timeline into the global map. + pub async fn load_temp_timeline( + ttid: TenantTimelineId, + tmp_path: &Utf8PathBuf, + check_tombstone: bool, + ) -> Result> { + // Check for existence and mark that we're creating it. + let (conf, broker_active_set, partial_backup_rate_limiter) = { + let mut state = TIMELINES_STATE.lock().unwrap(); + match state.timelines.get(&ttid) { + Some(GlobalMapTimeline::CreationInProgress) => { + bail!(TimelineError::CreationInProgress(ttid)); + } + Some(GlobalMapTimeline::Timeline(_)) => { + bail!(TimelineError::AlreadyExists(ttid)); + } + _ => {} + } + if check_tombstone { + if state.tombstones.contains_key(&ttid) { + anyhow::bail!("timeline {ttid} is deleted, refusing to recreate"); + } + } else { + // We may be have been asked to load a timeline that was previously deleted (e.g. from `pull_timeline.rs`). We trust + // that the human doing this manual intervention knows what they are doing, and remove its tombstone. + if state.tombstones.remove(&ttid).is_some() { + warn!("un-deleted timeline {ttid}"); + } + } + state + .timelines + .insert(ttid, GlobalMapTimeline::CreationInProgress); + state.get_dependencies() + }; - // Write the new timeline to the disk and start background workers. - // Bootstrap is transactional, so if it fails, the timeline will be deleted, - // and the state on disk should remain unchanged. - if let Err(e) = timeline - .init_new( - &mut shared_state, + // Do the actual move and reflect the result in the map. + match GlobalTimelines::install_temp_timeline(ttid, tmp_path, &conf).await { + Ok(timeline) => { + let mut timeline_shared_state = timeline.write_shared_state().await; + let mut state = TIMELINES_STATE.lock().unwrap(); + assert!(matches!( + state.timelines.get(&ttid), + Some(GlobalMapTimeline::CreationInProgress) + )); + + state + .timelines + .insert(ttid, GlobalMapTimeline::Timeline(timeline.clone())); + drop(state); + timeline.bootstrap( + &mut timeline_shared_state, &conf, broker_active_set, partial_backup_rate_limiter, - ) - .await - { - // Note: the most likely reason for init failure is that the timeline - // directory already exists on disk. This happens when timeline is corrupted - // and wasn't loaded from disk on startup because of that. We want to preserve - // the timeline directory in this case, for further inspection. - - // TODO: this is an unusual error, perhaps we should send it to sentry - // TODO: compute will try to create timeline every second, we should add backoff - error!("failed to init new timeline {}: {}", ttid, e); - - // Timeline failed to init, it cannot be used. Remove it from the map. - TIMELINES_STATE.lock().unwrap().timelines.remove(&ttid); - return Err(e); + ); + drop(timeline_shared_state); + Ok(timeline) + } + Err(e) => { + // Init failed, remove the marker from the map + let mut state = TIMELINES_STATE.lock().unwrap(); + assert!(matches!( + state.timelines.get(&ttid), + Some(GlobalMapTimeline::CreationInProgress) + )); + state.timelines.remove(&ttid); + Err(e) } - // We are done with bootstrap, release the lock, return the timeline. - // {} block forces release before .await } - Ok(timeline) + } + + /// Main part of load_temp_timeline: do the move and load. + async fn install_temp_timeline( + ttid: TenantTimelineId, + tmp_path: &Utf8PathBuf, + conf: &SafeKeeperConf, + ) -> Result> { + let tenant_path = get_tenant_dir(conf, &ttid.tenant_id); + let timeline_path = get_timeline_dir(conf, &ttid); + + // We must have already checked that timeline doesn't exist in the map, + // but there might be existing datadir: if timeline is corrupted it is + // not loaded. We don't want to overwrite such a dir, so check for its + // existence. + match fs::metadata(&timeline_path).await { + Ok(_) => { + // Timeline directory exists on disk, we should leave state unchanged + // and return error. + bail!(TimelineError::Invalid(ttid)); + } + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => { + return Err(e.into()); + } + } + + info!( + "moving timeline {} from {} to {}", + ttid, tmp_path, timeline_path + ); + + // Now it is safe to move the timeline directory to the correct + // location. First, create tenant directory. Ignore error if it already + // exists. + if let Err(e) = tokio::fs::create_dir(&tenant_path).await { + if e.kind() != std::io::ErrorKind::AlreadyExists { + return Err(e.into()); + } + } + // fsync it + fsync_async_opt(&tenant_path, !conf.no_sync).await?; + // and its creation + fsync_async_opt(&conf.workdir, !conf.no_sync).await?; + + // Do the move. + durable_rename(tmp_path, &timeline_path, !conf.no_sync).await?; + + Timeline::load_timeline(conf, ttid) } /// Get a timeline from the global map. If it's not present, it doesn't exist on disk, @@ -358,8 +404,16 @@ impl GlobalTimelines { global_lock .timelines .values() - .filter(|t| !t.is_cancelled()) - .cloned() + .filter_map(|t| match t { + GlobalMapTimeline::Timeline(t) => { + if t.is_cancelled() { + None + } else { + Some(t.clone()) + } + } + _ => None, + }) .collect() } @@ -370,8 +424,11 @@ impl GlobalTimelines { global_lock .timelines .values() + .filter_map(|t| match t { + GlobalMapTimeline::Timeline(t) => Some(t.clone()), + _ => None, + }) .filter(|t| t.ttid.tenant_id == tenant_id) - .cloned() .collect() } @@ -504,3 +561,45 @@ fn delete_dir(path: Utf8PathBuf) -> Result { Err(e) => Err(e.into()), } } + +/// Create temp directory for a new timeline. It needs to be located on the same +/// filesystem as the rest of the timelines. It will be automatically deleted when +/// Utf8TempDir goes out of scope. +pub async fn create_temp_timeline_dir( + conf: &SafeKeeperConf, + ttid: TenantTimelineId, +) -> Result<(Utf8TempDir, Utf8PathBuf)> { + let temp_base = conf.workdir.join("tmp"); + + tokio::fs::create_dir_all(&temp_base).await?; + + let tli_dir = camino_tempfile::Builder::new() + .suffix("_temptli") + .prefix(&format!("{}_{}_", ttid.tenant_id, ttid.timeline_id)) + .tempdir_in(temp_base)?; + + let tli_dir_path = tli_dir.path().to_path_buf(); + + Ok((tli_dir, tli_dir_path)) +} + +/// Do basic validation of a temp timeline, before moving it to the global map. +pub async fn validate_temp_timeline( + conf: &SafeKeeperConf, + ttid: TenantTimelineId, + path: &Utf8PathBuf, +) -> Result<(Lsn, Lsn)> { + let control_path = path.join("safekeeper.control"); + + let control_store = control_file::FileStorage::load_control_file(control_path)?; + if control_store.server.wal_seg_size == 0 { + bail!("wal_seg_size is not set"); + } + + let wal_store = wal_storage::PhysicalStorage::new(&ttid, path.clone(), conf, &control_store)?; + + let commit_lsn = control_store.commit_lsn; + let flush_lsn = wal_store.flush_lsn(); + + Ok((commit_lsn, flush_lsn)) +} diff --git a/safekeeper/src/wal_storage.rs b/safekeeper/src/wal_storage.rs index 6e7da94973..61d7825ae6 100644 --- a/safekeeper/src/wal_storage.rs +++ b/safekeeper/src/wal_storage.rs @@ -186,8 +186,14 @@ impl PhysicalStorage { "initialized storage for timeline {}, flush_lsn={}, commit_lsn={}, peer_horizon_lsn={}", ttid.timeline_id, flush_lsn, state.commit_lsn, state.peer_horizon_lsn, ); - if flush_lsn < state.commit_lsn || flush_lsn < state.peer_horizon_lsn { - warn!("timeline {} potential data loss: flush_lsn by find_end_of_wal is less than either commit_lsn or peer_horizon_lsn from control file", ttid.timeline_id); + if flush_lsn < state.commit_lsn { + bail!("timeline {} potential data loss: flush_lsn {} by find_end_of_wal is less than commit_lsn {} from control file", ttid.timeline_id, flush_lsn, state.commit_lsn); + } + if flush_lsn < state.peer_horizon_lsn { + warn!( + "timeline {}: flush_lsn {} is less than cfile peer_horizon_lsn {}", + ttid.timeline_id, flush_lsn, state.peer_horizon_lsn + ); } Ok(PhysicalStorage { diff --git a/safekeeper/tests/walproposer_sim/safekeeper.rs b/safekeeper/tests/walproposer_sim/safekeeper.rs index 047b4be8fa..12aa025771 100644 --- a/safekeeper/tests/walproposer_sim/safekeeper.rs +++ b/safekeeper/tests/walproposer_sim/safekeeper.rs @@ -59,7 +59,7 @@ impl GlobalMap { if state.commit_lsn < state.local_start_lsn { bail!( - "commit_lsn {} is higher than local_start_lsn {}", + "commit_lsn {} is smaller than local_start_lsn {}", state.commit_lsn, state.local_start_lsn ); @@ -96,23 +96,7 @@ impl GlobalMap { let local_start_lsn = Lsn::INVALID; let state = - TimelinePersistentState::new(&ttid, server_info, vec![], commit_lsn, local_start_lsn); - - if state.server.wal_seg_size == 0 { - bail!(TimelineError::UninitializedWalSegSize(ttid)); - } - - if state.server.pg_version == UNKNOWN_SERVER_VERSION { - bail!(TimelineError::UninitialinzedPgVersion(ttid)); - } - - if state.commit_lsn < state.local_start_lsn { - bail!( - "commit_lsn {} is higher than local_start_lsn {}", - state.commit_lsn, - state.local_start_lsn - ); - } + TimelinePersistentState::new(&ttid, server_info, vec![], commit_lsn, local_start_lsn)?; let disk_timeline = self.disk.put_state(&ttid, state); let control_store = DiskStateStorage::new(disk_timeline.clone());