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/_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/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/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/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: 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/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 \ 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. 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/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/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/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/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) } }; 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() } 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 d1079876f8..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; @@ -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"); } @@ -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 6c1a943470..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; @@ -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; @@ -1050,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/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 { 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); 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 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 { 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()); 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):