From 51185432a85dea16d55a49537d4bcef813cf0b40 Mon Sep 17 00:00:00 2001 From: Andres Freund Date: Wed, 22 Jan 2025 13:44:54 -0500 Subject: [PATCH v2.4 22/29] aio: Add test_aio module --- src/include/storage/buf_internals.h | 4 + src/backend/storage/buffer/bufmgr.c | 3 +- src/test/modules/Makefile | 1 + src/test/modules/meson.build | 1 + src/test/modules/test_aio/.gitignore | 2 + src/test/modules/test_aio/Makefile | 27 + src/test/modules/test_aio/meson.build | 37 ++ src/test/modules/test_aio/t/001_aio.pl | 401 +++++++++++++++ src/test/modules/test_aio/test_aio--1.0.sql | 84 ++++ src/test/modules/test_aio/test_aio.c | 518 ++++++++++++++++++++ src/test/modules/test_aio/test_aio.control | 3 + 11 files changed, 1079 insertions(+), 2 deletions(-) create mode 100644 src/test/modules/test_aio/.gitignore create mode 100644 src/test/modules/test_aio/Makefile create mode 100644 src/test/modules/test_aio/meson.build create mode 100644 src/test/modules/test_aio/t/001_aio.pl create mode 100644 src/test/modules/test_aio/test_aio--1.0.sql create mode 100644 src/test/modules/test_aio/test_aio.c create mode 100644 src/test/modules/test_aio/test_aio.control diff --git a/src/include/storage/buf_internals.h b/src/include/storage/buf_internals.h index 2a0c70c9998..396642415bc 100644 --- a/src/include/storage/buf_internals.h +++ b/src/include/storage/buf_internals.h @@ -421,6 +421,10 @@ extern void IssuePendingWritebacks(WritebackContext *wb_context, IOContext io_co extern void ScheduleBufferTagForWriteback(WritebackContext *wb_context, IOContext io_context, BufferTag *tag); +/* solely to make it easier to write tests */ +extern bool StartBufferIO(BufferDesc *buf, bool forInput, bool nowait); + + /* freelist.c */ extern IOContext IOContextForStrategy(BufferAccessStrategy strategy); extern BufferDesc *StrategyGetBuffer(BufferAccessStrategy strategy, diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index ee9a9f70167..b641bc3982b 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -516,7 +516,6 @@ static uint32 WaitBufHdrUnlocked(BufferDesc *buf); static int SyncOneBuffer(int buf_id, bool skip_recently_used, WritebackContext *wb_context); static void WaitIO(BufferDesc *buf); -static bool StartBufferIO(BufferDesc *buf, bool forInput, bool nowait); static void TerminateBufferIO(BufferDesc *buf, bool clear_dirty, uint32 set_flag_bits, bool forget_owner, bool syncio); @@ -5831,7 +5830,7 @@ WaitIO(BufferDesc *buf) * find out if they can perform the I/O as part of a larger operation, without * waiting for the answer or distinguishing the reasons why not. */ -static bool +bool StartBufferIO(BufferDesc *buf, bool forInput, bool nowait) { uint32 buf_state; diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile index 89e78b7d114..0ae2d4b6669 100644 --- a/src/test/modules/Makefile +++ b/src/test/modules/Makefile @@ -13,6 +13,7 @@ SUBDIRS = \ libpq_pipeline \ plsample \ spgist_name_ops \ + test_aio \ test_bloomfilter \ test_copy_callbacks \ test_custom_rmgrs \ diff --git a/src/test/modules/meson.build b/src/test/modules/meson.build index a57077b682e..94a9ebfdf60 100644 --- a/src/test/modules/meson.build +++ b/src/test/modules/meson.build @@ -1,5 +1,6 @@ # Copyright (c) 2022-2025, PostgreSQL Global Development Group +subdir('test_aio') subdir('brin') subdir('commit_ts') subdir('delay_execution') diff --git a/src/test/modules/test_aio/.gitignore b/src/test/modules/test_aio/.gitignore new file mode 100644 index 00000000000..716e17f5a2a --- /dev/null +++ b/src/test/modules/test_aio/.gitignore @@ -0,0 +1,2 @@ +# Generated subdirectories +/tmp_check/ diff --git a/src/test/modules/test_aio/Makefile b/src/test/modules/test_aio/Makefile new file mode 100644 index 00000000000..87d5315ba00 --- /dev/null +++ b/src/test/modules/test_aio/Makefile @@ -0,0 +1,27 @@ +# src/test/modules/delay_execution/Makefile + +PGFILEDESC = "test_aio - test code for AIO" + +MODULE_big = test_aio +OBJS = \ + $(WIN32RES) \ + test_aio.o + +EXTENSION = test_aio +DATA = test_aio--1.0.sql + +TAP_TESTS = 1 + +export enable_injection_points +export with_liburing + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = src/test/modules/test_aio +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif diff --git a/src/test/modules/test_aio/meson.build b/src/test/modules/test_aio/meson.build new file mode 100644 index 00000000000..ac846b2b6f3 --- /dev/null +++ b/src/test/modules/test_aio/meson.build @@ -0,0 +1,37 @@ +# Copyright (c) 2022-2024, PostgreSQL Global Development Group + +test_aio_sources = files( + 'test_aio.c', +) + +if host_system == 'windows' + test_aio_sources += rc_lib_gen.process(win32ver_rc, extra_args: [ + '--NAME', 'test_aio', + '--FILEDESC', 'test_aio - test code for AIO',]) +endif + +test_aio = shared_module('test_aio', + test_aio_sources, + kwargs: pg_test_mod_args, +) +test_install_libs += test_aio + +test_install_data += files( + 'test_aio.control', + 'test_aio--1.0.sql', +) + +tests += { + 'name': 'test_aio', + 'sd': meson.current_source_dir(), + 'bd': meson.current_build_dir(), + 'tap': { + 'env': { + 'enable_injection_points': get_option('injection_points') ? 'yes' : 'no', + 'with_liburing': liburing.found() ? 'yes' : 'no', + }, + 'tests': [ + 't/001_aio.pl', + ], + }, +} diff --git a/src/test/modules/test_aio/t/001_aio.pl b/src/test/modules/test_aio/t/001_aio.pl new file mode 100644 index 00000000000..225dce08e82 --- /dev/null +++ b/src/test/modules/test_aio/t/001_aio.pl @@ -0,0 +1,401 @@ +# Copyright (c) 2025, PostgreSQL Global Development Group + +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + + +### +# Test io_method=worker +### +my $node_worker = create_node('worker'); +$node_worker->start(); + +run_generic_test('worker', $node_worker); +SKIP: +{ + skip 'Injection points not supported by this build', 1 + unless $ENV{enable_injection_points} eq 'yes'; + test_inject_worker('worker', $node_worker); +} + +$node_worker->stop(); + + +### +# Test io_method=io_uring +### + +if ($ENV{with_liburing} eq 'yes') +{ + my $node_uring = create_node('io_uring'); + $node_uring->start(); + run_generic_test('io_uring', $node_uring); + $node_uring->stop(); +} + + +### +# Test io_method=sync +### + +my $node_sync = create_node('sync'); + +# just to have one test not use the default auto-tuning + +$node_sync->append_conf('postgresql.conf', qq( +io_max_concurrency=4 +)); + +$node_sync->start(); +run_generic_test('sync', $node_sync); +$node_sync->stop(); + +done_testing(); + + +### +# Test Helpers +### + + +sub create_node +{ + my $io_method = shift; + + my $node = PostgreSQL::Test::Cluster->new($io_method); + + # Want to test initdb for each IO method, otherwise we could just reuse + # the cluster. + $node->init(extra => ['-c', "io_method=$io_method"]); + + $node->append_conf('postgresql.conf', qq( +shared_preload_libraries=test_aio +log_min_messages = 'DEBUG3' +log_statement=all +restart_after_crash=false +)); + + ok(1, "$io_method: initdb"); + + return $node; +} + + +sub psql_like +{ + my $io_method = shift; + my $psql = shift; + my $name = shift; + my $sql = shift; + my $expected_stdout = shift; + my $expected_stderr = shift; + my ($cmdret, $output); + + ($output, $cmdret) = $psql->query($sql); + + like($output, $expected_stdout, "$io_method: $name: expected stdout"); + like($psql->{stderr}, $expected_stderr, "$io_method: $name: expected stderr"); + $psql->{stderr} = ''; +} + + +sub test_handle +{ + my $io_method = shift; + my $node = shift; + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + + # leak warning: implicit xact + psql_like($io_method, $psql, + "handle_get() leak in implicit xact", + qq(SELECT handle_get()), + qr/^$/, + qr/leaked AIO handle/, "$io_method: leaky handle_get() warns"); + + # leak warning: explicit xact + psql_like($io_method, $psql, + "handle_get() leak in explicit xact", + qq(BEGIN; SELECT handle_get(); COMMIT), + qr/^$/, + qr/leaked AIO handle/); + + + # leak warning: explicit xact, rollback + psql_like($io_method, $psql, + "handle_get() leak in explicit xact, rollback", + qq(BEGIN; SELECT handle_get(); ROLLBACK;), + qr/^$/, + qr/leaked AIO handle/); + + # leak warning: subtrans + psql_like($io_method, $psql, + "handle_get() leak in subxact", + qq(BEGIN; SAVEPOINT foo; SELECT handle_get(); COMMIT;), + qr/^$/, + qr/leaked AIO handle/); + + # leak warning + error: released in different command (thus resowner) + psql_like($io_method, $psql, + "handle_release() in different command", + qq(BEGIN; SELECT handle_get(); SELECT handle_release_last(); COMMIT;), + qr/^$/, + qr/leaked AIO handle.*release in unexpected state/ms); + + # no leak, release in same command + psql_like($io_method, $psql, + "handle_release() in same command", + qq(BEGIN; SELECT handle_get() UNION ALL SELECT handle_release_last(); COMMIT;), + qr/^$/, + qr/^$/); + + # normal handle use + psql_like($io_method, $psql, + "handle_get_release()", + qq(SELECT handle_get_release()), + qr/^$/, + qr/^$/); + + # should error out, API violation + psql_like($io_method, $psql, + "handle_get_twice()", + qq(SELECT handle_get_release()), + qr/^$/, + qr/^$/); + + # recover after error in implicit xact + psql_like($io_method, $psql, + "handle error recovery in implicit xact", + qq(SELECT handle_get_and_error(); SELECT 'ok', handle_get_release()), + qr/^|ok$/, + qr/ERROR.*as you command/); + + # recover after error in implicit xact + psql_like($io_method, $psql, + "handle error recovery in explicit xact", + qq(BEGIN; SELECT handle_get_and_error(); SELECT handle_get_release(), 'ok'; COMMIT;), + qr/^|ok$/, + qr/ERROR.*as you command/); + + # recover after error in subtrans + psql_like($io_method, $psql, + "handle error recovery in explicit subxact", + qq(BEGIN; SAVEPOINT foo; SELECT handle_get_and_error(); ROLLBACK TO SAVEPOINT foo; SELECT handle_get_release(); ROLLBACK;), + qr/^|ok$/, + qr/ERROR.*as you command/); + + $psql->quit(); +} + + +sub test_batch +{ + my $io_method = shift; + my $node = shift; + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + + # leak warning & recovery: implicit xact + psql_like($io_method, $psql, + "batch_start() leak & cleanup in implicit xact", + qq(SELECT batch_start()), + qr/^$/, + qr/open AIO batch at end/, "$io_method: leaky batch_start() warns"); + + # leak warning & recovery: explicit xact + psql_like($io_method, $psql, + "batch_start() leak & cleanup in explicit xact", + qq(BEGIN; SELECT batch_start(); COMMIT;), + qr/^$/, + qr/open AIO batch at end/, "$io_method: leaky batch_start() warns"); + + + # leak warning & recovery: explicit xact, rollback + # + # FIXME: This doesn't fail right now, due to not getting a chance to do + # something at transaction command commit. That's not a correctness issue, + # it just means it's a bit harder to find buggy code. + #psql_like($io_method, $psql, + # "batch_start() leak & cleanup after abort", + # qq(BEGIN; SELECT batch_start(); ROLLBACK;), + # qr/^$/, + # qr/open AIO batch at end/, "$io_method: leaky batch_start() warns"); + + # no warning, batch closed in same command + psql_like($io_method, $psql, + "batch_start(), batch_end() works", + qq(SELECT batch_start() UNION ALL SELECT batch_end()), + qr/^$/, + qr/^$/, "$io_method: batch_start(), batch_end()"); + + $psql->quit(); +} + +sub test_io_error +{ + my $io_method = shift; + my $node = shift; + my ($ret, $output); + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + + # verify the error is reported in custom C code + ($output, $ret) = $psql->query(qq(SELECT read_corrupt_rel_block('tbl_a', 1);)); + is($ret, 1, "$io_method: read_corrupt_rel_block() fails"); + like($psql->{stderr}, qr/invalid page in block 1 of relation base\/.*/, + "$io_method: read_corrupt_rel_block() reports error"); + $psql->{stderr} = ''; + + # verify the error is reported for bufmgr reads + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_a WHERE ctid = '(1, 1)')); + is($ret, 1, "$io_method: tid scan reading corrupt block fails"); + like($psql->{stderr}, qr/invalid page in block 1 of relation base\/.*/, + "$io_method: tid scan reading corrupt block reports error"); + $psql->{stderr} = ''; + + # verify the error is reported for bufmgr reads + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_a WHERE ctid = '(1, 1)')); + is($ret, 1, "$io_method: sequential scan reading corrupt block fails"); + like($psql->{stderr}, qr/invalid page in block 1 of relation base\/.*/, + "$io_method: sequential scan reading corrupt block reports error"); + $psql->{stderr} = ''; + + $psql->quit(); +} + + +sub test_inject +{ + my $io_method = shift; + my $node = shift; + my ($ret, $output); + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + + # injected what we'd expect + $psql->query_safe(qq(SELECT inj_io_short_read_attach(8192);)); + $psql->query_safe(qq(SELECT invalidate_rel_block('tbl_b', 2);)); + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_b WHERE ctid = '(2, 1)';)); + is($ret, 0, "$io_method: injection point not triggering failure succeeds"); + + # injected a read shorter than a single block, expecting error + $psql->query_safe(qq(SELECT inj_io_short_read_attach(17);)); + $psql->query_safe(qq(SELECT invalidate_rel_block('tbl_b', 2);)); + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_b WHERE ctid = '(2, 1)';)); + is($ret, 1, "$io_method: single block short read fails"); + like($psql->{stderr}, qr/ERROR:.*could not read blocks 2\.\.2 in file "base\/.*": read only 0 of 8192 bytes/, + "$io_method: single block short read reports error"); + $psql->{stderr} = ''; + + # shorten multi-block read to a single block, should retry + $psql->query_safe(qq( +SELECT invalidate_rel_block('tbl_b', 0); +SELECT invalidate_rel_block('tbl_b', 1); +SELECT invalidate_rel_block('tbl_b', 2); +SELECT inj_io_short_read_attach(8192); + )); + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_b;)); + is($ret, 0, "$io_method: multi block short read is retried"); + + # verify that page verification errors are detected even as part of a + # shortened multi-block read (tbl_a, block 1 is corrupted) + $psql->query_safe(qq( +SELECT invalidate_rel_block('tbl_a', 0); +SELECT invalidate_rel_block('tbl_a', 1); +SELECT invalidate_rel_block('tbl_a', 2); +SELECT inj_io_short_read_attach(8192); + )); + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_a WHERE ctid < '(2, 1)')); + is($ret, 1, "$io_method: shortened multi-block read detects invalid page"); + like($psql->{stderr}, qr/ERROR:.*invalid page in block 1 of relation base\/.*/, + "$io_method: shortened multi-block reads reports invalid page"); + $psql->{stderr} = ''; + + # trigger a hard error, should error out + $psql->query_safe(qq( +SELECT inj_io_short_read_attach(-errno_from_string('EIO')); +SELECT invalidate_rel_block('tbl_b', 2); + )); + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_b; SELECT 1;)); + is($ret, 1, "$io_method: hard IO error is detected"); + like($psql->{stderr}, qr/ERROR:.*could not read blocks 2..3 in file \"base\/.*\": Input\/output error/, + "$io_method: hard IO error is reported"); + $psql->{stderr} = ''; + + $psql->query_safe(qq( +SELECT inj_io_short_read_detach(); + )); + + $psql->quit(); +} + + +sub test_inject_worker +{ + my $io_method = shift; + my $node = shift; + my ($ret, $output); + + my $psql = $node->background_psql('postgres', on_error_stop => 0); + + # trigger a failure to reopen, should error out, but should recover + $psql->query_safe(qq( +SELECT inj_io_reopen_attach(); +SELECT invalidate_rel_block('tbl_b', 1); + )); + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_b;)); + is($ret, 1, "$io_method: failure to open is detected"); + like($psql->{stderr}, qr/ERROR:.*could not read blocks 1..2 in file "base\/.*": No such file or directory/, + "$io_method: failure to open is reported"); + $psql->{stderr} = ''; + + $psql->query_safe(qq( +SELECT inj_io_reopen_detach(); + )); + + # check that we indeed recover + ($output, $ret) = $psql->query(qq(SELECT count(*) FROM tbl_b;)); + is($ret, 0, "$io_method: recovers from failure to open "); + + + $psql->quit(); +} + + +sub run_generic_test +{ + my $io_method = shift; + my $node = shift; + + is($node->safe_psql('postgres', 'SHOW io_method'), + $io_method, + "$io_method: io_method set correctly"); + + $node->safe_psql('postgres', qq( +CREATE EXTENSION test_aio; +CREATE TABLE tbl_a(data int not null) WITH (AUTOVACUUM_ENABLED = false); +CREATE TABLE tbl_b(data int not null) WITH (AUTOVACUUM_ENABLED = false); + +INSERT INTO tbl_a SELECT generate_series(1, 10000); +INSERT INTO tbl_b SELECT generate_series(1, 10000); +SELECT grow_rel('tbl_a', 500); +SELECT grow_rel('tbl_b', 550); + +SELECT corrupt_rel_block('tbl_a', 1); +)); + + test_handle($io_method, $node); + test_io_error($io_method, $node); + test_batch($io_method, $node); + + SKIP: + { + skip 'Injection points not supported by this build', 1 + unless $ENV{enable_injection_points} eq 'yes'; + test_inject($io_method, $node); + } +} diff --git a/src/test/modules/test_aio/test_aio--1.0.sql b/src/test/modules/test_aio/test_aio--1.0.sql new file mode 100644 index 00000000000..e7c7c6a6db6 --- /dev/null +++ b/src/test/modules/test_aio/test_aio--1.0.sql @@ -0,0 +1,84 @@ +/* src/test/modules/test_aio/test_aio--1.0.sql */ + +-- complain if script is sourced in psql, rather than via CREATE EXTENSION +\echo Use "CREATE EXTENSION test_aio" to load this file. \quit + + +CREATE FUNCTION errno_from_string(sym text) +RETURNS pg_catalog.int4 STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + + +CREATE FUNCTION grow_rel(rel regclass, nblocks int) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + + +CREATE FUNCTION corrupt_rel_block(rel regclass, blockno int) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION read_corrupt_rel_block(rel regclass, blockno int) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION invalidate_rel_block(rel regclass, blockno int) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + + +/* + * Handle related functions + */ +CREATE FUNCTION handle_get_and_error() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION handle_get_twice() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION handle_get() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION handle_get_release() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION handle_release_last() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + + +/* + * Batchmode related functions + */ +CREATE FUNCTION batch_start() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION batch_end() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + + + +/* + * Injection point related functions + */ +CREATE FUNCTION inj_io_short_read_attach(result int) +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION inj_io_short_read_detach() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION inj_io_reopen_attach() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; + +CREATE FUNCTION inj_io_reopen_detach() +RETURNS pg_catalog.void STRICT +AS 'MODULE_PATHNAME' LANGUAGE C; diff --git a/src/test/modules/test_aio/test_aio.c b/src/test/modules/test_aio/test_aio.c new file mode 100644 index 00000000000..15851565853 --- /dev/null +++ b/src/test/modules/test_aio/test_aio.c @@ -0,0 +1,518 @@ +/*------------------------------------------------------------------------- + * + * delay_execution.c + * Test module to allow delay between parsing and execution of a query. + * + * The delay is implemented by taking and immediately releasing a specified + * advisory lock. If another process has previously taken that lock, the + * current process will be blocked until the lock is released; otherwise, + * there's no effect. This allows an isolationtester script to reliably + * test behaviors where some specified action happens in another backend + * between parsing and execution of any desired query. + * + * Copyright (c) 2020-2025, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/test/modules/delay_execution/delay_execution.c + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "access/relation.h" +#include "fmgr.h" +#include "storage/aio.h" +#include "storage/aio_internal.h" +#include "storage/buf_internals.h" +#include "storage/bufmgr.h" +#include "storage/ipc.h" +#include "storage/lwlock.h" +#include "utils/builtins.h" +#include "utils/injection_point.h" +#include "utils/rel.h" + + +PG_MODULE_MAGIC; + + +typedef struct InjIoErrorState +{ + bool enabled_short_read; + bool enabled_reopen; + + bool short_read_result_set; + int short_read_result; +} InjIoErrorState; + +static InjIoErrorState * inj_io_error_state; + +/* Shared memory init callbacks */ +static shmem_request_hook_type prev_shmem_request_hook = NULL; +static shmem_startup_hook_type prev_shmem_startup_hook = NULL; + + +static PgAioHandle *last_handle; + + + +static void +test_aio_shmem_request(void) +{ + if (prev_shmem_request_hook) + prev_shmem_request_hook(); + + RequestAddinShmemSpace(sizeof(InjIoErrorState)); +} + +static void +test_aio_shmem_startup(void) +{ + bool found; + + if (prev_shmem_startup_hook) + prev_shmem_startup_hook(); + + /* Create or attach to the shared memory state */ + LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE); + + inj_io_error_state = ShmemInitStruct("injection_points", + sizeof(InjIoErrorState), + &found); + + if (!found) + { + /* + * First time through, so initialize. This is shared with the dynamic + * initialization using a DSM. + */ + inj_io_error_state->enabled_short_read = false; + inj_io_error_state->enabled_reopen = false; + +#ifdef USE_INJECTION_POINTS + InjectionPointAttach("AIO_PROCESS_COMPLETION_BEFORE_SHARED", + "test_aio", + "inj_io_short_read", + NULL, + 0); + InjectionPointLoad("AIO_PROCESS_COMPLETION_BEFORE_SHARED"); + + InjectionPointAttach("AIO_WORKER_AFTER_REOPEN", + "test_aio", + "inj_io_reopen", + NULL, + 0); + InjectionPointLoad("AIO_WORKER_AFTER_REOPEN"); + +#endif + } + else + { +#ifdef USE_INJECTION_POINTS + InjectionPointLoad("AIO_PROCESS_COMPLETION_BEFORE_SHARED"); + InjectionPointLoad("AIO_WORKER_AFTER_REOPEN"); + elog(LOG, "injection point loaded"); +#endif + } + + LWLockRelease(AddinShmemInitLock); +} + +void +_PG_init(void) +{ + if (!process_shared_preload_libraries_in_progress) + return; + + /* Shared memory initialization */ + prev_shmem_request_hook = shmem_request_hook; + shmem_request_hook = test_aio_shmem_request; + prev_shmem_startup_hook = shmem_startup_hook; + shmem_startup_hook = test_aio_shmem_startup; +} + + +PG_FUNCTION_INFO_V1(errno_from_string); +Datum +errno_from_string(PG_FUNCTION_ARGS) +{ + const char *sym = text_to_cstring(PG_GETARG_TEXT_PP(0)); + + if (strcmp(sym, "EIO") == 0) + PG_RETURN_INT32(EIO); + else if (strcmp(sym, "EAGAIN") == 0) + PG_RETURN_INT32(EAGAIN); + else if (strcmp(sym, "EINTR") == 0) + PG_RETURN_INT32(EINTR); + else if (strcmp(sym, "ENOSPC") == 0) + PG_RETURN_INT32(ENOSPC); + else if (strcmp(sym, "EROFS") == 0) + PG_RETURN_INT32(EROFS); + + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg_internal("%s is not a supported errno value", sym)); + PG_RETURN_INT32(0); +} + + +PG_FUNCTION_INFO_V1(grow_rel); +Datum +grow_rel(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + uint32 nblocks = PG_GETARG_UINT32(1); + Relation rel; +#define MAX_BUFFERS_TO_EXTEND_BY 64 + Buffer victim_buffers[MAX_BUFFERS_TO_EXTEND_BY]; + + rel = relation_open(relid, AccessExclusiveLock); + + while (nblocks > 0) + { + uint32 extend_by_pages; + + extend_by_pages = Min(nblocks, MAX_BUFFERS_TO_EXTEND_BY); + + ExtendBufferedRelBy(BMR_REL(rel), + MAIN_FORKNUM, + NULL, + 0, + extend_by_pages, + victim_buffers, + &extend_by_pages); + + nblocks -= extend_by_pages; + + for (uint32 i = 0; i < extend_by_pages; i++) + { + ReleaseBuffer(victim_buffers[i]); + } + } + + relation_close(rel, NoLock); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(corrupt_rel_block); +Datum +corrupt_rel_block(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + uint32 block = PG_GETARG_UINT32(1); + Relation rel; + Buffer buf; + Page page; + PageHeader ph; + + rel = relation_open(relid, AccessExclusiveLock); + + buf = ReadBuffer(rel, block); + page = BufferGetPage(buf); + + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); + + MarkBufferDirty(buf); + + PageInit(page, BufferGetPageSize(buf), 0); + + ph = (PageHeader) page; + ph->pd_special = BLCKSZ + 1; + + FlushOneBuffer(buf); + + LockBuffer(buf, BUFFER_LOCK_UNLOCK); + + ReleaseBuffer(buf); + + EvictUnpinnedBuffer(buf); + + relation_close(rel, NoLock); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(read_corrupt_rel_block); +Datum +read_corrupt_rel_block(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + uint32 block = PG_GETARG_UINT32(1); + Relation rel; + Buffer buf; + BufferDesc *buf_hdr; + Page page; + PgAioReturn ior; + PgAioHandle *ioh; + PgAioWaitRef iow; + SMgrRelation smgr; + uint32 buf_state; + + rel = relation_open(relid, AccessExclusiveLock); + + /* read buffer without erroring out */ + buf = ReadBufferExtended(rel, MAIN_FORKNUM, block, RBM_ZERO_AND_LOCK, NULL); + LockBuffer(buf, BUFFER_LOCK_UNLOCK); + + page = BufferGetBlock(buf); + + ioh = pgaio_io_acquire(CurrentResourceOwner, &ior); + pgaio_io_get_wref(ioh, &iow); + + buf_hdr = GetBufferDescriptor(buf - 1); + smgr = RelationGetSmgr(rel); + + /* FIXME: even if just a test, we should verify nobody else uses this */ + buf_state = LockBufHdr(buf_hdr); + buf_state &= ~(BM_VALID | BM_DIRTY); + UnlockBufHdr(buf_hdr, buf_state); + + StartBufferIO(buf_hdr, true, false); + + pgaio_io_set_handle_data_32(ioh, (uint32 *) &buf, 1); + pgaio_io_register_callbacks(ioh, PGAIO_HCB_SHARED_BUFFER_READV); + + smgrstartreadv(ioh, smgr, MAIN_FORKNUM, block, + (void *) &page, 1); + + ReleaseBuffer(buf); + + pgaio_wref_wait(&iow); + + if (ior.result.status != ARS_OK) + pgaio_result_report(ior.result, &ior.target_data, + ior.result.status == ARS_PARTIAL ? WARNING : ERROR); + + relation_close(rel, NoLock); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(invalidate_rel_block); +Datum +invalidate_rel_block(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + uint32 block = PG_GETARG_UINT32(1); + Relation rel; + PrefetchBufferResult pr; + Buffer buf; + + rel = relation_open(relid, AccessExclusiveLock); + + /* this is a gross hack, but there's no good API exposed */ + pr = PrefetchBuffer(rel, MAIN_FORKNUM, block); + buf = pr.recent_buffer; + elog(LOG, "recent: %d", buf); + if (BufferIsValid(buf)) + { + /* if the buffer contents aren't valid, this'll return false */ + if (ReadRecentBuffer(rel->rd_locator, MAIN_FORKNUM, block, buf)) + { + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); + FlushOneBuffer(buf); + LockBuffer(buf, BUFFER_LOCK_UNLOCK); + ReleaseBuffer(buf); + + if (!EvictUnpinnedBuffer(buf)) + elog(ERROR, "couldn't unpin"); + } + } + + relation_close(rel, AccessExclusiveLock); + + PG_RETURN_VOID(); +} + +#if 0 +PG_FUNCTION_INFO_V1(test_unsubmitted_vs_close); +Datum +test_unsubmitted_vs_close(PG_FUNCTION_ARGS) +{ + Oid relid = PG_GETARG_OID(0); + uint32 block = PG_GETARG_UINT32(1); + Relation rel; + Buffer buf; + Page page; + PageHeader ph; + + rel = relation_open(relid, AccessExclusiveLock); + + buf = ReadBufferExtended(rel, MAIN_FORKNUM, block, RBM_ZERO_AND_LOCK, NULL); + + buf = ReadBuffer(rel, block); + page = BufferGetPage(buf); + + EvictUnpinnedBuffer(buf); + + LockBuffer(buf, BUFFER_LOCK_UNLOCK); + + + MarkBufferDirty(buf); + ph->pd_special = BLCKSZ + 1; + + /* last_handle = pgaio_io_acquire(); */ + + PG_RETURN_VOID(); +} +#endif + +PG_FUNCTION_INFO_V1(handle_get); +Datum +handle_get(PG_FUNCTION_ARGS) +{ + last_handle = pgaio_io_acquire(CurrentResourceOwner, NULL); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(handle_release_last); +Datum +handle_release_last(PG_FUNCTION_ARGS) +{ + if (!last_handle) + elog(ERROR, "no handle"); + + pgaio_io_release(last_handle); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(handle_get_and_error); +Datum +handle_get_and_error(PG_FUNCTION_ARGS) +{ + pgaio_io_acquire(CurrentResourceOwner, NULL); + + elog(ERROR, "as you command"); + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(handle_get_twice); +Datum +handle_get_twice(PG_FUNCTION_ARGS) +{ + pgaio_io_acquire(CurrentResourceOwner, NULL); + pgaio_io_acquire(CurrentResourceOwner, NULL); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(handle_get_release); +Datum +handle_get_release(PG_FUNCTION_ARGS) +{ + PgAioHandle *handle; + + handle = pgaio_io_acquire(CurrentResourceOwner, NULL); + pgaio_io_release(handle); + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(batch_start); +Datum +batch_start(PG_FUNCTION_ARGS) +{ + pgaio_enter_batchmode(); + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(batch_end); +Datum +batch_end(PG_FUNCTION_ARGS) +{ + pgaio_exit_batchmode(); + PG_RETURN_VOID(); +} + +#ifdef USE_INJECTION_POINTS +extern PGDLLEXPORT void inj_io_short_read(const char *name, const void *private_data); +extern PGDLLEXPORT void inj_io_reopen(const char *name, const void *private_data); + +void +inj_io_short_read(const char *name, const void *private_data) +{ + PgAioHandle *ioh; + + elog(LOG, "short read called: %d", inj_io_error_state->enabled_short_read); + + if (inj_io_error_state->enabled_short_read) + { + ioh = pgaio_inj_io_get(); + + if (inj_io_error_state->short_read_result_set) + { + elog(LOG, "short read, changing result from %d to %d", + ioh->result, inj_io_error_state->short_read_result); + + ioh->result = inj_io_error_state->short_read_result; + } + } +} + +void +inj_io_reopen(const char *name, const void *private_data) +{ + elog(LOG, "reopen called: %d", inj_io_error_state->enabled_reopen); + + if (inj_io_error_state->enabled_reopen) + { + elog(ERROR, "injection point triggering failure to reopen "); + } +} +#endif + +PG_FUNCTION_INFO_V1(inj_io_short_read_attach); +Datum +inj_io_short_read_attach(PG_FUNCTION_ARGS) +{ +#ifdef USE_INJECTION_POINTS + inj_io_error_state->enabled_short_read = true; + inj_io_error_state->short_read_result_set = !PG_ARGISNULL(0); + if (inj_io_error_state->short_read_result_set) + inj_io_error_state->short_read_result = PG_GETARG_INT32(0); +#else + elog(ERROR, "injection points not supported"); +#endif + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(inj_io_short_read_detach); +Datum +inj_io_short_read_detach(PG_FUNCTION_ARGS) +{ +#ifdef USE_INJECTION_POINTS + inj_io_error_state->enabled_short_read = false; +#else + elog(ERROR, "injection points not supported"); +#endif + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(inj_io_reopen_attach); +Datum +inj_io_reopen_attach(PG_FUNCTION_ARGS) +{ +#ifdef USE_INJECTION_POINTS + inj_io_error_state->enabled_reopen = true; +#else + elog(ERROR, "injection points not supported"); +#endif + + PG_RETURN_VOID(); +} + +PG_FUNCTION_INFO_V1(inj_io_reopen_detach); +Datum +inj_io_reopen_detach(PG_FUNCTION_ARGS) +{ +#ifdef USE_INJECTION_POINTS + inj_io_error_state->enabled_reopen = false; +#else + elog(ERROR, "injection points not supported"); +#endif + PG_RETURN_VOID(); +} diff --git a/src/test/modules/test_aio/test_aio.control b/src/test/modules/test_aio/test_aio.control new file mode 100644 index 00000000000..cd91c3ed16b --- /dev/null +++ b/src/test/modules/test_aio/test_aio.control @@ -0,0 +1,3 @@ +comment = 'Test code for AIO' +default_version = '1.0' +module_pathname = '$libdir/test_aio' -- 2.48.1.76.g4e746b1a31.dirty