aio.c File Reference

#include "postgres.h"
#include "lib/ilist.h"
#include "miscadmin.h"
#include "port/atomics.h"
#include "storage/aio.h"
#include "storage/aio_internal.h"
#include "storage/aio_subsys.h"
#include "utils/guc.h"
#include "utils/guc_hooks.h"
#include "utils/injection_point.h"
#include "utils/resowner.h"
#include "utils/wait_event_types.h"

Include dependency graph for aio.c:

Go to the source code of this file.

Macros

#define PGAIO_HS_TOSTR_CASE(sym) case PGAIO_HS_##sym: return #sym

Functions

static void pgaio_io_update_state (PgAioHandle *ioh, PgAioHandleState new_state)

static void pgaio_io_reclaim (PgAioHandle *ioh)

static void pgaio_io_resowner_register (PgAioHandle *ioh)

static void pgaio_io_wait_for_free (void)

static PgAioHandle * pgaio_io_from_wref (PgAioWaitRef *iow, uint64 *ref_generation)

static const char * pgaio_io_state_get_name (PgAioHandleState s)

static void pgaio_io_wait (PgAioHandle *ioh, uint64 ref_generation)

PgAioHandle * pgaio_io_acquire (struct ResourceOwnerData *resowner, PgAioReturn *ret)

PgAioHandle * pgaio_io_acquire_nb (struct ResourceOwnerData *resowner, PgAioReturn *ret)

void pgaio_io_release (PgAioHandle *ioh)

void pgaio_io_release_resowner (dlist_node *ioh_node, bool on_error)

void pgaio_io_set_flag (PgAioHandle *ioh, PgAioHandleFlags flag)

int pgaio_io_get_id (PgAioHandle *ioh)

ProcNumber pgaio_io_get_owner (PgAioHandle *ioh)

void pgaio_io_get_wref (PgAioHandle *ioh, PgAioWaitRef *iow)

void pgaio_io_stage (PgAioHandle *ioh, PgAioOp op)

bool pgaio_io_needs_synchronous_execution (PgAioHandle *ioh)

void pgaio_io_prepare_submit (PgAioHandle *ioh)

void pgaio_io_process_completion (PgAioHandle *ioh, int result)

bool pgaio_io_was_recycled (PgAioHandle *ioh, uint64 ref_generation, PgAioHandleState *state)

const char * pgaio_io_get_state_name (PgAioHandle *ioh)

const char * pgaio_result_status_string (PgAioResultStatus rs)

void pgaio_wref_clear (PgAioWaitRef *iow)

bool pgaio_wref_valid (PgAioWaitRef *iow)

int pgaio_wref_get_id (PgAioWaitRef *iow)

void pgaio_wref_wait (PgAioWaitRef *iow)

bool pgaio_wref_check_done (PgAioWaitRef *iow)

void pgaio_enter_batchmode (void)

void pgaio_exit_batchmode (void)

bool pgaio_have_staged (void)

void pgaio_submit_staged (void)

void pgaio_error_cleanup (void)

void AtEOXact_Aio (bool is_commit)

void pgaio_closing_fd (int fd)

void pgaio_shutdown (int code, Datum arg)

void assign_io_method (int newval, void *extra)

bool check_io_max_concurrency (int *newval, void **extra, GucSource source)

Variables

const struct config_enum_entry io_method_options []

int io_method = DEFAULT_IO_METHOD

int io_max_concurrency = -1

PgAioCtl * pgaio_ctl

PgAioBackend * pgaio_my_backend

static const IoMethodOps *const pgaio_method_ops_table []

const IoMethodOps * pgaio_method_ops

Macro Definition Documentation

◆ PGAIO_HS_TOSTR_CASE

#define PGAIO_HS_TOSTR_CASE ( sym ) case PGAIO_HS_##sym: return #sym

Function Documentation

◆ assign_io_method()

void assign_io_method ( int newval,

void * extra

)

Definition at line 1319 of file aio.c.

1320{

1321 Assert(pgaio_method_ops_table[newval] != NULL);

1322 Assert(newval < lengthof(io_method_options));

1323

1324 pgaio_method_ops = pgaio_method_ops_table[newval];

1325}

pgaio_method_ops

const IoMethodOps * pgaio_method_ops

Definition: aio.c:93

pgaio_method_ops_table

static const IoMethodOps *const pgaio_method_ops_table[]

Definition: aio.c:84

io_method_options

const struct config_enum_entry io_method_options[]

Definition: aio.c:64

lengthof

#define lengthof(array)

Definition: c.h:787

newval

#define newval

Assert

Assert(PointerIsAligned(start, uint64))

References Assert(), io_method_options, lengthof, newval, pgaio_method_ops, and pgaio_method_ops_table.

◆ AtEOXact_Aio()

void AtEOXact_Aio ( bool is_commit )

Definition at line 1190 of file aio.c.

1191{

1192 /*

1193 * We should never be in batch mode at transactional boundaries. In case

1194 * an error was thrown while in batch mode, pgaio_error_cleanup() should

1195 * have exited batchmode.

1196 *

1197 * In case we are in batchmode somehow, make sure to submit all staged

1198 * IOs, other backends may need them to complete to continue.

1199 */

1200 if (pgaio_my_backend->in_batchmode)

1201 {

1202 pgaio_error_cleanup();

1203 elog(WARNING, "open AIO batch at end of (sub-)transaction");

1204 }

1205

1206 /*

1207 * As we aren't in batchmode, there shouldn't be any unsubmitted IOs.

1208 */

1209 Assert(pgaio_my_backend->num_staged_ios == 0);

1210}

pgaio_my_backend

PgAioBackend * pgaio_my_backend

Definition: aio.c:81

pgaio_error_cleanup

void pgaio_error_cleanup(void)

Definition: aio.c:1162

WARNING

#define WARNING

Definition: elog.h:36

elog

#define elog(elevel,...)

Definition: elog.h:226

PgAioBackend::in_batchmode

bool in_batchmode

Definition: aio_internal.h:209

PgAioBackend::num_staged_ios

uint16 num_staged_ios

Definition: aio_internal.h:214

References Assert(), elog, PgAioBackend::in_batchmode, PgAioBackend::num_staged_ios, pgaio_error_cleanup(), pgaio_my_backend, and WARNING.

Referenced by AbortSubTransaction(), AbortTransaction(), CommitTransaction(), pgaio_shutdown(), and PrepareTransaction().

◆ check_io_max_concurrency()

bool check_io_max_concurrency ( int * newval,

void ** extra,

GucSource source

)

Definition at line 1328 of file aio.c.

1329{

1330 if (*newval == -1)

1331 {

1332 /*

1333 * Auto-tuning will be applied later during startup, as auto-tuning

1334 * depends on the value of various GUCs.

1335 */

1336 return true;

1337 }

1338 else if (*newval == 0)

1339 {

1340 GUC_check_errdetail("Only -1 or values bigger than 0 are valid.");

1341 return false;

1342 }

1343

1344 return true;

1345}

GUC_check_errdetail

#define GUC_check_errdetail

Definition: guc.h:505

References GUC_check_errdetail, and newval.

◆ pgaio_closing_fd()

void pgaio_closing_fd ( int fd )

Definition at line 1217 of file aio.c.

1218{

1219 /*

1220 * Might be called before AIO is initialized or in a subprocess that

1221 * doesn't use AIO.

1222 */

1223 if (!pgaio_my_backend)

1224 return;

1225

1226 /*

1227 * For now just submit all staged IOs - we could be more selective, but

1228 * it's probably not worth it.

1229 */

1230 if (pgaio_my_backend->num_staged_ios > 0)

1231 {

1232 pgaio_debug(DEBUG2,

1233 "submitting %d IOs before FD %d gets closed",

1234 pgaio_my_backend->num_staged_ios, fd);

1235 pgaio_submit_staged();

1236 }

1237

1238 /*

1239 * If requested by the IO method, wait for all IOs that use the

1240 * to-be-closed FD.

1241 */

1242 if (pgaio_method_ops->wait_on_fd_before_close)

1243 {

1244 /*

1245 * As waiting for one IO to complete may complete multiple IOs, we

1246 * can't just use a mutable list iterator. The maximum number of

1247 * in-flight IOs is fairly small, so just restart the loop after

1248 * waiting for an IO.

1249 */

1250 while (!dclist_is_empty(&pgaio_my_backend->in_flight_ios))

1251 {

1252 dlist_iter iter;

1253 PgAioHandle *ioh = NULL;

1254 uint64 generation;

1255

1256 dclist_foreach(iter, &pgaio_my_backend->in_flight_ios)

1257 {

1258 ioh = dclist_container(PgAioHandle, node, iter.cur);

1259

1260 generation = ioh->generation;

1261

1262 if (pgaio_io_uses_fd(ioh, fd))

1263 break;

1264 else

1265 ioh = NULL;

1266 }

1267

1268 if (!ioh)

1269 break;

1270

1271 pgaio_debug_io(DEBUG2, ioh,

1272 "waiting for IO before FD %d gets closed, %u in-flight IOs",

1273 fd, dclist_count(&pgaio_my_backend->in_flight_ios));

1274

1275 /* see comment in pgaio_io_wait_for_free() about raciness */

1276 pgaio_io_wait(ioh, generation);

1277 }

1278 }

1279}

pgaio_submit_staged

void pgaio_submit_staged(void)

Definition: aio.c:1120

pgaio_io_wait

static void pgaio_io_wait(PgAioHandle *ioh, uint64 ref_generation)

Definition: aio.c:576

pgaio_debug

#define pgaio_debug(elevel, msg,...)

Definition: aio_internal.h:382

pgaio_debug_io

#define pgaio_debug_io(elevel, ioh, msg,...)

Definition: aio_internal.h:395

pgaio_io_uses_fd

bool pgaio_io_uses_fd(PgAioHandle *ioh, int fd)

Definition: aio_io.c:197

uint64

uint64_t uint64

Definition: c.h:539

DEBUG2

#define DEBUG2

Definition: elog.h:29

dclist_container

#define dclist_container(type, membername, ptr)

Definition: ilist.h:947

dclist_count

static uint32 dclist_count(const dclist_head *head)

Definition: ilist.h:932

dclist_is_empty

static bool dclist_is_empty(const dclist_head *head)

Definition: ilist.h:682

dclist_foreach

#define dclist_foreach(iter, lhead)

Definition: ilist.h:970

fd

static int fd(const char *x, int i)

Definition: preproc-init.c:105

IoMethodOps::wait_on_fd_before_close

bool wait_on_fd_before_close

Definition: aio_internal.h:268

PgAioBackend::in_flight_ios

dclist_head in_flight_ios

Definition: aio_internal.h:225

PgAioHandle

Definition: aio_internal.h:103

PgAioHandle::generation

uint64 generation

Definition: aio_internal.h:152

dlist_iter

Definition: ilist.h:178

dlist_iter::cur

dlist_node * cur

Definition: ilist.h:179

References dlist_iter::cur, dclist_container, dclist_count(), dclist_foreach, dclist_is_empty(), DEBUG2, fd(), PgAioHandle::generation, PgAioBackend::in_flight_ios, PgAioBackend::num_staged_ios, pgaio_debug, pgaio_debug_io, pgaio_io_uses_fd(), pgaio_io_wait(), pgaio_method_ops, pgaio_my_backend, pgaio_submit_staged(), and IoMethodOps::wait_on_fd_before_close.

Referenced by CloseTransientFile(), FileClose(), FreeDesc(), and LruDelete().

◆ pgaio_enter_batchmode()

void pgaio_enter_batchmode ( void )

Definition at line 1078 of file aio.c.

1079{

1080 if (pgaio_my_backend->in_batchmode)

1081 elog(ERROR, "starting batch while batch already in progress");

1082 pgaio_my_backend->in_batchmode = true;

1083}

ERROR

#define ERROR

Definition: elog.h:39

References elog, ERROR, PgAioBackend::in_batchmode, and pgaio_my_backend.

Referenced by batch_start(), read_rel_block_ll(), and read_stream_look_ahead().

◆ pgaio_error_cleanup()

void pgaio_error_cleanup ( void )

Definition at line 1162 of file aio.c.

1163{

1164 /*

1165 * It is possible that code errored out after pgaio_enter_batchmode() but

1166 * before pgaio_exit_batchmode() was called. In that case we need to

1167 * submit the IO now.

1168 */

1169 if (pgaio_my_backend->in_batchmode)

1170 {

1171 pgaio_my_backend->in_batchmode = false;

1172

1173 pgaio_submit_staged();

1174 }

1175

1176 /*

1177 * As we aren't in batchmode, there shouldn't be any unsubmitted IOs.

1178 */

1179 Assert(pgaio_my_backend->num_staged_ios == 0);

1180}

References Assert(), PgAioBackend::in_batchmode, PgAioBackend::num_staged_ios, pgaio_my_backend, and pgaio_submit_staged().

Referenced by AbortSubTransaction(), AbortTransaction(), AtEOXact_Aio(), AutoVacLauncherMain(), BackgroundWriterMain(), CheckpointerMain(), pgarch_archiveXlog(), WalSndErrorCleanup(), WalSummarizerMain(), and WalWriterMain().

◆ pgaio_exit_batchmode()

void pgaio_exit_batchmode ( void )

Definition at line 1089 of file aio.c.

1090{

1091 Assert(pgaio_my_backend->in_batchmode);

1092

1093 pgaio_submit_staged();

1094 pgaio_my_backend->in_batchmode = false;

1095}

References Assert(), PgAioBackend::in_batchmode, pgaio_my_backend, and pgaio_submit_staged().

Referenced by batch_end(), read_rel_block_ll(), and read_stream_look_ahead().

◆ pgaio_have_staged()

bool pgaio_have_staged ( void )

Definition at line 1104 of file aio.c.

1105{

1106 Assert(pgaio_my_backend->in_batchmode ||

1107 pgaio_my_backend->num_staged_ios == 0);

1108 return pgaio_my_backend->num_staged_ios > 0;

1109}

References Assert(), PgAioBackend::in_batchmode, PgAioBackend::num_staged_ios, and pgaio_my_backend.

Referenced by ReadBuffersCanStartIO().

◆ pgaio_io_acquire()

PgAioHandle * pgaio_io_acquire ( struct ResourceOwnerData * resowner,

PgAioReturn * ret

)

Definition at line 159 of file aio.c.

160{

161 PgAioHandle *h;

162

163 while (true)

164 {

165 h = pgaio_io_acquire_nb(resowner, ret);

166

167 if (h != NULL)

168 return h;

169

170 /*

171 * Evidently all handles by this backend are in use. Just wait for

172 * some to complete.

173 */

174 pgaio_io_wait_for_free();

175 }

176}

pgaio_io_wait_for_free

static void pgaio_io_wait_for_free(void)

Definition: aio.c:758

pgaio_io_acquire_nb

PgAioHandle * pgaio_io_acquire_nb(struct ResourceOwnerData *resowner, PgAioReturn *ret)

Definition: aio.c:185

References pgaio_io_acquire_nb(), and pgaio_io_wait_for_free().

Referenced by AsyncReadBuffers(), handle_get(), handle_get_and_error(), handle_get_release(), handle_get_twice(), and read_rel_block_ll().

◆ pgaio_io_acquire_nb()

PgAioHandle * pgaio_io_acquire_nb ( struct ResourceOwnerData * resowner,

PgAioReturn * ret

)

Definition at line 185 of file aio.c.

186{

187 PgAioHandle *ioh = NULL;

188

189 if (pgaio_my_backend->num_staged_ios >= PGAIO_SUBMIT_BATCH_SIZE)

190 {

191 Assert(pgaio_my_backend->num_staged_ios == PGAIO_SUBMIT_BATCH_SIZE);

192 pgaio_submit_staged();

193 }

194

195 if (pgaio_my_backend->handed_out_io)

196 elog(ERROR, "API violation: Only one IO can be handed out");

197

198 /*

199 * Probably not needed today, as interrupts should not process this IO,

200 * but...

201 */

202 HOLD_INTERRUPTS();

203

204 if (!dclist_is_empty(&pgaio_my_backend->idle_ios))

205 {

206 dlist_node *ion = dclist_pop_head_node(&pgaio_my_backend->idle_ios);

207

208 ioh = dclist_container(PgAioHandle, node, ion);

209

210 Assert(ioh->state == PGAIO_HS_IDLE);

211 Assert(ioh->owner_procno == MyProcNumber);

212

213 pgaio_io_update_state(ioh, PGAIO_HS_HANDED_OUT);

214 pgaio_my_backend->handed_out_io = ioh;

215

216 if (resowner)

217 pgaio_io_resowner_register(ioh);

218

219 if (ret)

220 {

221 ioh->report_return = ret;

222 ret->result.status = PGAIO_RS_UNKNOWN;

223 }

224 }

225

226 RESUME_INTERRUPTS();

227

228 return ioh;

229}

pgaio_io_update_state

static void pgaio_io_update_state(PgAioHandle *ioh, PgAioHandleState new_state)

Definition: aio.c:383

pgaio_io_resowner_register

static void pgaio_io_resowner_register(PgAioHandle *ioh)

Definition: aio.c:406

PGAIO_HS_IDLE

@ PGAIO_HS_IDLE

Definition: aio_internal.h:46

PGAIO_HS_HANDED_OUT

@ PGAIO_HS_HANDED_OUT

Definition: aio_internal.h:53

PGAIO_SUBMIT_BATCH_SIZE

#define PGAIO_SUBMIT_BATCH_SIZE

Definition: aio_internal.h:28

PGAIO_RS_UNKNOWN

@ PGAIO_RS_UNKNOWN

Definition: aio_types.h:80

MyProcNumber

ProcNumber MyProcNumber

Definition: globals.c:90

dclist_pop_head_node

static dlist_node * dclist_pop_head_node(dclist_head *head)

Definition: ilist.h:789

RESUME_INTERRUPTS

#define RESUME_INTERRUPTS()

Definition: miscadmin.h:135

HOLD_INTERRUPTS

#define HOLD_INTERRUPTS()

Definition: miscadmin.h:133

PgAioBackend::idle_ios

dclist_head idle_ios

Definition: aio_internal.h:197

PgAioBackend::handed_out_io

PgAioHandle * handed_out_io

Definition: aio_internal.h:206

PgAioHandle::owner_procno

int32 owner_procno

Definition: aio_internal.h:131

PgAioHandle::report_return

PgAioReturn * report_return

Definition: aio_internal.h:177

PgAioHandle::state

uint8 state

Definition: aio_internal.h:105

PgAioResult::status

uint32 status

Definition: aio_types.h:108

PgAioReturn::result

PgAioResult result

Definition: aio_types.h:132

dlist_node

Definition: ilist.h:138

References Assert(), dclist_container, dclist_is_empty(), dclist_pop_head_node(), elog, ERROR, PgAioBackend::handed_out_io, HOLD_INTERRUPTS, PgAioBackend::idle_ios, MyProcNumber, PgAioBackend::num_staged_ios, PgAioHandle::owner_procno, PGAIO_HS_HANDED_OUT, PGAIO_HS_IDLE, pgaio_io_resowner_register(), pgaio_io_update_state(), pgaio_my_backend, PGAIO_RS_UNKNOWN, PGAIO_SUBMIT_BATCH_SIZE, pgaio_submit_staged(), PgAioHandle::report_return, PgAioReturn::result, RESUME_INTERRUPTS, PgAioHandle::state, and PgAioResult::status.

Referenced by AsyncReadBuffers(), and pgaio_io_acquire().

◆ pgaio_io_from_wref()

static PgAioHandle * pgaio_io_from_wref ( PgAioWaitRef * iow,

uint64 * ref_generation

)

static

Definition at line 888 of file aio.c.

889{

890 PgAioHandle *ioh;

891

892 Assert(iow->aio_index < pgaio_ctl->io_handle_count);

893

894 ioh = &pgaio_ctl->io_handles[iow->aio_index];

895

896 *ref_generation = ((uint64) iow->generation_upper) << 32 |

897 iow->generation_lower;

898

899 Assert(*ref_generation != 0);

900

901 return ioh;

902}

pgaio_ctl

PgAioCtl * pgaio_ctl

Definition: aio.c:78

PgAioCtl::io_handles

PgAioHandle * io_handles

Definition: aio_internal.h:252

PgAioCtl::io_handle_count

uint32 io_handle_count

Definition: aio_internal.h:251

PgAioWaitRef::generation_upper

uint32 generation_upper

Definition: aio_types.h:45

PgAioWaitRef::aio_index

uint32 aio_index

Definition: aio_types.h:35

PgAioWaitRef::generation_lower

uint32 generation_lower

Definition: aio_types.h:46

References PgAioWaitRef::aio_index, Assert(), PgAioWaitRef::generation_lower, PgAioWaitRef::generation_upper, PgAioCtl::io_handle_count, PgAioCtl::io_handles, and pgaio_ctl.

Referenced by pgaio_wref_check_done(), and pgaio_wref_wait().

◆ pgaio_io_get_id()

int pgaio_io_get_id ( PgAioHandle * ioh )

Definition at line 339 of file aio.c.

340{

341 Assert(ioh >= pgaio_ctl->io_handles &&

342 ioh < (pgaio_ctl->io_handles + pgaio_ctl->io_handle_count));

343 return ioh - pgaio_ctl->io_handles;

344}

References Assert(), PgAioCtl::io_handle_count, PgAioCtl::io_handles, and pgaio_ctl.

Referenced by pg_get_aios(), pgaio_io_wait(), pgaio_io_wait_for_free(), and pgaio_worker_submission_queue_insert().

◆ pgaio_io_get_owner()

ProcNumber pgaio_io_get_owner ( PgAioHandle * ioh )

Definition at line 352 of file aio.c.

353{

354 return ioh->owner_procno;

355}

References PgAioHandle::owner_procno.

Referenced by buffer_readv_complete(), and smgr_aio_reopen().

◆ pgaio_io_get_state_name()

const char * pgaio_io_get_state_name ( PgAioHandle * ioh )

Definition at line 925 of file aio.c.

926{

927 return pgaio_io_state_get_name(ioh->state);

928}

pgaio_io_state_get_name

static const char * pgaio_io_state_get_name(PgAioHandleState s)

Definition: aio.c:905

References pgaio_io_state_get_name(), and PgAioHandle::state.

Referenced by pg_get_aios(), and pgaio_io_wait().

◆ pgaio_io_get_wref()

void pgaio_io_get_wref ( PgAioHandle * ioh,

PgAioWaitRef * iow

)

Definition at line 363 of file aio.c.

364{

365 Assert(ioh->state == PGAIO_HS_HANDED_OUT ||

366 ioh->state == PGAIO_HS_DEFINED ||

367 ioh->state == PGAIO_HS_STAGED);

368 Assert(ioh->generation != 0);

369

370 iow->aio_index = ioh - pgaio_ctl->io_handles;

371 iow->generation_upper = (uint32) (ioh->generation >> 32);

372 iow->generation_lower = (uint32) ioh->generation;

373}

PGAIO_HS_STAGED

@ PGAIO_HS_STAGED

Definition: aio_internal.h:66

PGAIO_HS_DEFINED

@ PGAIO_HS_DEFINED

Definition: aio_internal.h:59

uint32

uint32_t uint32

Definition: c.h:538

References PgAioWaitRef::aio_index, Assert(), PgAioHandle::generation, PgAioWaitRef::generation_lower, PgAioWaitRef::generation_upper, PgAioCtl::io_handles, pgaio_ctl, PGAIO_HS_DEFINED, PGAIO_HS_HANDED_OUT, PGAIO_HS_STAGED, and PgAioHandle::state.

Referenced by AsyncReadBuffers(), buffer_stage_common(), and read_rel_block_ll().

◆ pgaio_io_needs_synchronous_execution()

bool pgaio_io_needs_synchronous_execution ( PgAioHandle * ioh )

Definition at line 480 of file aio.c.

481{

482 /*

483 * If the caller said to execute the IO synchronously, do so.

484 *

485 * XXX: We could optimize the logic when to execute synchronously by first

486 * checking if there are other IOs in flight and only synchronously

487 * executing if not. Unclear whether that'll be sufficiently common to be

488 * worth worrying about.

489 */

490 if (ioh->flags & PGAIO_HF_SYNCHRONOUS)

491 return true;

492

493 /* Check if the IO method requires synchronous execution of IO */

494 if (pgaio_method_ops->needs_synchronous_execution)

495 return pgaio_method_ops->needs_synchronous_execution(ioh);

496

497 return false;

498}

PGAIO_HF_SYNCHRONOUS

@ PGAIO_HF_SYNCHRONOUS

Definition: aio.h:70

IoMethodOps::needs_synchronous_execution

bool(* needs_synchronous_execution)(PgAioHandle *ioh)

Definition: aio_internal.h:294

PgAioHandle::flags

uint8 flags

Definition: aio_internal.h:114

References PgAioHandle::flags, IoMethodOps::needs_synchronous_execution, PGAIO_HF_SYNCHRONOUS, and pgaio_method_ops.

Referenced by pgaio_io_stage().

◆ pgaio_io_prepare_submit()

void pgaio_io_prepare_submit ( PgAioHandle * ioh )

Definition at line 507 of file aio.c.

508{

509 pgaio_io_update_state(ioh, PGAIO_HS_SUBMITTED);

510

511 dclist_push_tail(&pgaio_my_backend->in_flight_ios, &ioh->node);

512}

PGAIO_HS_SUBMITTED

@ PGAIO_HS_SUBMITTED

Definition: aio_internal.h:69

dclist_push_tail

static void dclist_push_tail(dclist_head *head, dlist_node *node)

Definition: ilist.h:709

PgAioHandle::node

dlist_node node

Definition: aio_internal.h:146

References dclist_push_tail(), PgAioBackend::in_flight_ios, PgAioHandle::node, PGAIO_HS_SUBMITTED, pgaio_io_update_state(), and pgaio_my_backend.

Referenced by pgaio_io_stage(), and pgaio_worker_submit().

◆ pgaio_io_process_completion()

void pgaio_io_process_completion ( PgAioHandle * ioh,

int result

)

Definition at line 525 of file aio.c.

526{

527 Assert(ioh->state == PGAIO_HS_SUBMITTED);

528

529 Assert(CritSectionCount > 0);

530

531 ioh->result = result;

532

533 pgaio_io_update_state(ioh, PGAIO_HS_COMPLETED_IO);

534

535 INJECTION_POINT("aio-process-completion-before-shared", ioh);

536

537 pgaio_io_call_complete_shared(ioh);

538

539 pgaio_io_update_state(ioh, PGAIO_HS_COMPLETED_SHARED);

540

541 /* condition variable broadcast ensures state is visible before wakeup */

542 ConditionVariableBroadcast(&ioh->cv);

543

544 /* contains call to pgaio_io_call_complete_local() */

545 if (ioh->owner_procno == MyProcNumber)

546 pgaio_io_reclaim(ioh);

547}

pgaio_io_reclaim

static void pgaio_io_reclaim(PgAioHandle *ioh)

Definition: aio.c:672

pgaio_io_call_complete_shared

void pgaio_io_call_complete_shared(PgAioHandle *ioh)

Definition: aio_callback.c:225

PGAIO_HS_COMPLETED_SHARED

@ PGAIO_HS_COMPLETED_SHARED

Definition: aio_internal.h:82

PGAIO_HS_COMPLETED_IO

@ PGAIO_HS_COMPLETED_IO

Definition: aio_internal.h:72

ConditionVariableBroadcast

void ConditionVariableBroadcast(ConditionVariable *cv)

Definition: condition_variable.c:282

CritSectionCount

volatile uint32 CritSectionCount

Definition: globals.c:45

INJECTION_POINT

#define INJECTION_POINT(name, arg)

Definition: injection_point.h:37

PgAioHandle::result

int32 result

Definition: aio_internal.h:134

PgAioHandle::cv

ConditionVariable cv

Definition: aio_internal.h:159

References Assert(), ConditionVariableBroadcast(), CritSectionCount, PgAioHandle::cv, INJECTION_POINT, MyProcNumber, PgAioHandle::owner_procno, PGAIO_HS_COMPLETED_IO, PGAIO_HS_COMPLETED_SHARED, PGAIO_HS_SUBMITTED, pgaio_io_call_complete_shared(), pgaio_io_reclaim(), pgaio_io_update_state(), PgAioHandle::result, and PgAioHandle::state.

Referenced by IoWorkerMain(), and pgaio_io_perform_synchronously().

◆ pgaio_io_reclaim()

static void pgaio_io_reclaim ( PgAioHandle * ioh )

static

Definition at line 672 of file aio.c.

673{

674 /* This is only ok if it's our IO */

675 Assert(ioh->owner_procno == MyProcNumber);

676 Assert(ioh->state != PGAIO_HS_IDLE);

677

678 /* see comment in function header */

679 HOLD_INTERRUPTS();

680

681 /*

682 * It's a bit ugly, but right now the easiest place to put the execution

683 * of local completion callbacks is this function, as we need to execute

684 * local callbacks just before reclaiming at multiple callsites.

685 */

686 if (ioh->state == PGAIO_HS_COMPLETED_SHARED)

687 {

688 PgAioResult local_result;

689

690 local_result = pgaio_io_call_complete_local(ioh);

691 pgaio_io_update_state(ioh, PGAIO_HS_COMPLETED_LOCAL);

692

693 if (ioh->report_return)

694 {

695 ioh->report_return->result = local_result;

696 ioh->report_return->target_data = ioh->target_data;

697 }

698 }

699

700 pgaio_debug_io(DEBUG4, ioh,

701 "reclaiming: distilled_result: (status %s, id %u, error_data %d), raw_result: %d",

702 pgaio_result_status_string(ioh->distilled_result.status),

703 ioh->distilled_result.id,

704 ioh->distilled_result.error_data,

705 ioh->result);

706

707 /* if the IO has been defined, it's on the in-flight list, remove */

708 if (ioh->state != PGAIO_HS_HANDED_OUT)

709 dclist_delete_from(&pgaio_my_backend->in_flight_ios, &ioh->node);

710

711 if (ioh->resowner)

712 {

713 ResourceOwnerForgetAioHandle(ioh->resowner, &ioh->resowner_node);

714 ioh->resowner = NULL;

715 }

716

717 Assert(!ioh->resowner);

718

719 /*

720 * Update generation & state first, before resetting the IO's fields,

721 * otherwise a concurrent "viewer" could think the fields are valid, even

722 * though they are being reset. Increment the generation first, so that

723 * we can assert elsewhere that we never wait for an IDLE IO. While it's

724 * a bit weird for the state to go backwards for a generation, it's OK

725 * here, as there cannot be references to the "reborn" IO yet. Can't

726 * update both at once, so something has to give.

727 */

728 ioh->generation++;

729 pgaio_io_update_state(ioh, PGAIO_HS_IDLE);

730

731 /* ensure the state update is visible before we reset fields */

732 pg_write_barrier();

733

734 ioh->op = PGAIO_OP_INVALID;

735 ioh->target = PGAIO_TID_INVALID;

736 ioh->flags = 0;

737 ioh->num_callbacks = 0;

738 ioh->handle_data_len = 0;

739 ioh->report_return = NULL;

740 ioh->result = 0;

741 ioh->distilled_result.status = PGAIO_RS_UNKNOWN;

742

743 /*

744 * We push the IO to the head of the idle IO list, that seems more cache

745 * efficient in cases where only a few IOs are used.

746 */

747 dclist_push_head(&pgaio_my_backend->idle_ios, &ioh->node);

748

749 RESUME_INTERRUPTS();

750}

pgaio_result_status_string

const char * pgaio_result_status_string(PgAioResultStatus rs)

Definition: aio.c:931

PGAIO_TID_INVALID

@ PGAIO_TID_INVALID

Definition: aio.h:119

PGAIO_OP_INVALID

@ PGAIO_OP_INVALID

Definition: aio.h:90

pgaio_io_call_complete_local

PgAioResult pgaio_io_call_complete_local(PgAioHandle *ioh)

Definition: aio_callback.c:285

PGAIO_HS_COMPLETED_LOCAL

@ PGAIO_HS_COMPLETED_LOCAL

Definition: aio_internal.h:89

pg_write_barrier

#define pg_write_barrier()

Definition: atomics.h:155

DEBUG4

#define DEBUG4

Definition: elog.h:27

dclist_delete_from

static void dclist_delete_from(dclist_head *head, dlist_node *node)

Definition: ilist.h:763

dclist_push_head

static void dclist_push_head(dclist_head *head, dlist_node *node)

Definition: ilist.h:693

ResourceOwnerForgetAioHandle

void ResourceOwnerForgetAioHandle(ResourceOwner owner, struct dlist_node *ioh_node)

Definition: resowner.c:1107

PgAioHandle::target

uint8 target

Definition: aio_internal.h:108

PgAioHandle::target_data

PgAioTargetData target_data

Definition: aio_internal.h:187

PgAioHandle::resowner

struct ResourceOwnerData * resowner

Definition: aio_internal.h:148

PgAioHandle::distilled_result

PgAioResult distilled_result

Definition: aio_internal.h:162

PgAioHandle::handle_data_len

uint8 handle_data_len

Definition: aio_internal.h:128

PgAioHandle::op

uint8 op

Definition: aio_internal.h:111

PgAioHandle::num_callbacks

uint8 num_callbacks

Definition: aio_internal.h:116

PgAioHandle::resowner_node

dlist_node resowner_node

Definition: aio_internal.h:149

PgAioResult

Definition: aio_types.h:100

PgAioResult::error_data

uint32 error_data

Definition: aio_types.h:111

PgAioResult::id

uint32 id

Definition: aio_types.h:105

PgAioReturn::target_data

PgAioTargetData target_data

Definition: aio_types.h:133

References Assert(), dclist_delete_from(), dclist_push_head(), DEBUG4, PgAioHandle::distilled_result, PgAioResult::error_data, PgAioHandle::flags, PgAioHandle::generation, PgAioHandle::handle_data_len, HOLD_INTERRUPTS, PgAioResult::id, PgAioBackend::idle_ios, PgAioBackend::in_flight_ios, MyProcNumber, PgAioHandle::node, PgAioHandle::num_callbacks, PgAioHandle::op, PgAioHandle::owner_procno, pg_write_barrier, pgaio_debug_io, PGAIO_HS_COMPLETED_LOCAL, PGAIO_HS_COMPLETED_SHARED, PGAIO_HS_HANDED_OUT, PGAIO_HS_IDLE, pgaio_io_call_complete_local(), pgaio_io_update_state(), pgaio_my_backend, PGAIO_OP_INVALID, pgaio_result_status_string(), PGAIO_RS_UNKNOWN, PGAIO_TID_INVALID, PgAioHandle::report_return, ResourceOwnerForgetAioHandle(), PgAioHandle::resowner, PgAioHandle::resowner_node, PgAioHandle::result, PgAioReturn::result, RESUME_INTERRUPTS, PgAioHandle::state, PgAioResult::status, PgAioHandle::target, PgAioHandle::target_data, and PgAioReturn::target_data.

Referenced by pgaio_io_process_completion(), pgaio_io_release(), pgaio_io_release_resowner(), pgaio_io_wait(), pgaio_io_wait_for_free(), and pgaio_wref_check_done().

◆ pgaio_io_release()

void pgaio_io_release ( PgAioHandle * ioh )

Definition at line 237 of file aio.c.

238{

239 if (ioh == pgaio_my_backend->handed_out_io)

240 {

241 Assert(ioh->state == PGAIO_HS_HANDED_OUT);

242 Assert(ioh->resowner);

243

244 pgaio_my_backend->handed_out_io = NULL;

245

246 /*

247 * Note that no interrupts are processed between the handed_out_io

248 * check and the call to reclaim - that's important as otherwise an

249 * interrupt could have already reclaimed the handle.

250 */

251 pgaio_io_reclaim(ioh);

252 }

253 else

254 {

255 elog(ERROR, "release in unexpected state");

256 }

257}

References Assert(), elog, ERROR, PgAioBackend::handed_out_io, PGAIO_HS_HANDED_OUT, pgaio_io_reclaim(), pgaio_my_backend, PgAioHandle::resowner, and PgAioHandle::state.

Referenced by AsyncReadBuffers(), handle_get_release(), and handle_release_last().

◆ pgaio_io_release_resowner()

void pgaio_io_release_resowner ( dlist_node * ioh_node,

bool on_error

)

Definition at line 263 of file aio.c.

264{

265 PgAioHandle *ioh = dlist_container(PgAioHandle, resowner_node, ioh_node);

266

267 Assert(ioh->resowner);

268

269 /*

270 * Otherwise an interrupt, in the middle of releasing the IO, could end up

271 * trying to wait for the IO, leading to state confusion.

272 */

273 HOLD_INTERRUPTS();

274

275 ResourceOwnerForgetAioHandle(ioh->resowner, &ioh->resowner_node);

276 ioh->resowner = NULL;

277

278 switch ((PgAioHandleState) ioh->state)

279 {

280 case PGAIO_HS_IDLE:

281 elog(ERROR, "unexpected");

282 break;

283 case PGAIO_HS_HANDED_OUT:

284 Assert(ioh == pgaio_my_backend->handed_out_io || pgaio_my_backend->handed_out_io == NULL);

285

286 if (ioh == pgaio_my_backend->handed_out_io)

287 {

288 pgaio_my_backend->handed_out_io = NULL;

289 if (!on_error)

290 elog(WARNING, "leaked AIO handle");

291 }

292

293 pgaio_io_reclaim(ioh);

294 break;

295 case PGAIO_HS_DEFINED:

296 case PGAIO_HS_STAGED:

297 if (!on_error)

298 elog(WARNING, "AIO handle was not submitted");

299 pgaio_submit_staged();

300 break;

301 case PGAIO_HS_SUBMITTED:

302 case PGAIO_HS_COMPLETED_IO:

303 case PGAIO_HS_COMPLETED_SHARED:

304 case PGAIO_HS_COMPLETED_LOCAL:

305 /* this is expected to happen */

306 break;

307 }

308

309 /*

310 * Need to unregister the reporting of the IO's result, the memory it's

311 * referencing likely has gone away.

312 */

313 if (ioh->report_return)

314 ioh->report_return = NULL;

315

316 RESUME_INTERRUPTS();

317}

PgAioHandleState

Definition: aio_internal.h:44

dlist_container

#define dlist_container(type, membername, ptr)

Definition: ilist.h:593

References Assert(), dlist_container, elog, ERROR, PgAioBackend::handed_out_io, HOLD_INTERRUPTS, PGAIO_HS_COMPLETED_IO, PGAIO_HS_COMPLETED_LOCAL, PGAIO_HS_COMPLETED_SHARED, PGAIO_HS_DEFINED, PGAIO_HS_HANDED_OUT, PGAIO_HS_IDLE, PGAIO_HS_STAGED, PGAIO_HS_SUBMITTED, pgaio_io_reclaim(), pgaio_my_backend, pgaio_submit_staged(), PgAioHandle::report_return, ResourceOwnerForgetAioHandle(), PgAioHandle::resowner, PgAioHandle::resowner_node, RESUME_INTERRUPTS, PgAioHandle::state, and WARNING.

Referenced by ResourceOwnerReleaseInternal().

◆ pgaio_io_resowner_register()

static void pgaio_io_resowner_register ( PgAioHandle * ioh )

static

Definition at line 406 of file aio.c.

407{

408 Assert(!ioh->resowner);

409 Assert(CurrentResourceOwner);

410

411 ResourceOwnerRememberAioHandle(CurrentResourceOwner, &ioh->resowner_node);

412 ioh->resowner = CurrentResourceOwner;

413}

CurrentResourceOwner

ResourceOwner CurrentResourceOwner

Definition: resowner.c:173

ResourceOwnerRememberAioHandle

void ResourceOwnerRememberAioHandle(ResourceOwner owner, struct dlist_node *ioh_node)

Definition: resowner.c:1101

References Assert(), CurrentResourceOwner, ResourceOwnerRememberAioHandle(), PgAioHandle::resowner, and PgAioHandle::resowner_node.

Referenced by pgaio_io_acquire_nb().

◆ pgaio_io_set_flag()

void pgaio_io_set_flag ( PgAioHandle * ioh,

PgAioHandleFlags flag

)

Definition at line 327 of file aio.c.

328{

329 Assert(ioh->state == PGAIO_HS_HANDED_OUT);

330

331 ioh->flags |= flag;

332}

flag

char * flag(int b)

Definition: test-ctype.c:33

References Assert(), flag(), PgAioHandle::flags, PGAIO_HS_HANDED_OUT, and PgAioHandle::state.

Referenced by AsyncReadBuffers(), mdstartreadv(), and read_rel_block_ll().

◆ pgaio_io_stage()

void pgaio_io_stage ( PgAioHandle * ioh,

PgAioOp op

)

Definition at line 421 of file aio.c.

422{

423 bool needs_synchronous;

424

425 Assert(ioh->state == PGAIO_HS_HANDED_OUT);

426 Assert(pgaio_my_backend->handed_out_io == ioh);

427 Assert(pgaio_io_has_target(ioh));

428

429 /*

430 * Otherwise an interrupt, in the middle of staging and possibly executing

431 * the IO, could end up trying to wait for the IO, leading to state

432 * confusion.

433 */

434 HOLD_INTERRUPTS();

435

436 ioh->op = op;

437 ioh->result = 0;

438

439 pgaio_io_update_state(ioh, PGAIO_HS_DEFINED);

440

441 /* allow a new IO to be staged */

442 pgaio_my_backend->handed_out_io = NULL;

443

444 pgaio_io_call_stage(ioh);

445

446 pgaio_io_update_state(ioh, PGAIO_HS_STAGED);

447

448 /*

449 * Synchronous execution has to be executed, well, synchronously, so check

450 * that first.

451 */

452 needs_synchronous = pgaio_io_needs_synchronous_execution(ioh);

453

454 pgaio_debug_io(DEBUG3, ioh,

455 "staged (synchronous: %d, in_batch: %d)",

456 needs_synchronous, pgaio_my_backend->in_batchmode);

457

458 if (!needs_synchronous)

459 {

460 pgaio_my_backend->staged_ios[pgaio_my_backend->num_staged_ios++] = ioh;

461 Assert(pgaio_my_backend->num_staged_ios <= PGAIO_SUBMIT_BATCH_SIZE);

462

463 /*

464 * Unless code explicitly opted into batching IOs, submit the IO

465 * immediately.

466 */

467 if (!pgaio_my_backend->in_batchmode)

468 pgaio_submit_staged();

469 }

470 else

471 {

472 pgaio_io_prepare_submit(ioh);

473 pgaio_io_perform_synchronously(ioh);

474 }

475

476 RESUME_INTERRUPTS();

477}

pgaio_io_needs_synchronous_execution

bool pgaio_io_needs_synchronous_execution(PgAioHandle *ioh)

Definition: aio.c:480

pgaio_io_prepare_submit

void pgaio_io_prepare_submit(PgAioHandle *ioh)

Definition: aio.c:507

pgaio_io_call_stage

void pgaio_io_call_stage(PgAioHandle *ioh)

Definition: aio_callback.c:199

pgaio_io_perform_synchronously

void pgaio_io_perform_synchronously(PgAioHandle *ioh)

Definition: aio_io.c:116

pgaio_io_has_target

bool pgaio_io_has_target(PgAioHandle *ioh)

Definition: aio_target.c:40

DEBUG3

#define DEBUG3

Definition: elog.h:28

PgAioBackend::staged_ios

PgAioHandle * staged_ios[PGAIO_SUBMIT_BATCH_SIZE]

Definition: aio_internal.h:215

References Assert(), DEBUG3, PgAioBackend::handed_out_io, HOLD_INTERRUPTS, PgAioBackend::in_batchmode, PgAioBackend::num_staged_ios, PgAioHandle::op, pgaio_debug_io, PGAIO_HS_DEFINED, PGAIO_HS_HANDED_OUT, PGAIO_HS_STAGED, pgaio_io_call_stage(), pgaio_io_has_target(), pgaio_io_needs_synchronous_execution(), pgaio_io_perform_synchronously(), pgaio_io_prepare_submit(), pgaio_io_update_state(), pgaio_my_backend, PGAIO_SUBMIT_BATCH_SIZE, pgaio_submit_staged(), PgAioHandle::result, RESUME_INTERRUPTS, PgAioBackend::staged_ios, and PgAioHandle::state.

Referenced by pgaio_io_start_readv(), and pgaio_io_start_writev().

◆ pgaio_io_state_get_name()

static const char * pgaio_io_state_get_name ( PgAioHandleState s )

static

Definition at line 905 of file aio.c.

906{

907#define PGAIO_HS_TOSTR_CASE(sym) case PGAIO_HS_##sym: return #sym

908 switch ((PgAioHandleState) s)

909 {

910 PGAIO_HS_TOSTR_CASE(IDLE);

911 PGAIO_HS_TOSTR_CASE(HANDED_OUT);

912 PGAIO_HS_TOSTR_CASE(DEFINED);

913 PGAIO_HS_TOSTR_CASE(STAGED);

914 PGAIO_HS_TOSTR_CASE(SUBMITTED);

915 PGAIO_HS_TOSTR_CASE(COMPLETED_IO);

916 PGAIO_HS_TOSTR_CASE(COMPLETED_SHARED);

917 PGAIO_HS_TOSTR_CASE(COMPLETED_LOCAL);

918 }

919#undef PGAIO_HS_TOSTR_CASE

920

921 return NULL; /* silence compiler */

922}

PGAIO_HS_TOSTR_CASE

#define PGAIO_HS_TOSTR_CASE(sym)

References PGAIO_HS_TOSTR_CASE.

Referenced by pgaio_io_get_state_name(), and pgaio_io_update_state().

◆ pgaio_io_update_state()

static void pgaio_io_update_state ( PgAioHandle * ioh,

PgAioHandleState new_state

)

inlinestatic

Definition at line 383 of file aio.c.

384{

385 /*

386 * All callers need to have held interrupts in some form, otherwise

387 * interrupt processing could wait for the IO to complete, while in an

388 * intermediary state.

389 */

390 Assert(!INTERRUPTS_CAN_BE_PROCESSED());

391

392 pgaio_debug_io(DEBUG5, ioh,

393 "updating state to %s",

394 pgaio_io_state_get_name(new_state));

395

396 /*

397 * Ensure the changes signified by the new state are visible before the

398 * new state becomes visible.

399 */

400 pg_write_barrier();

401

402 ioh->state = new_state;

403}

DEBUG5

#define DEBUG5

Definition: elog.h:26

INTERRUPTS_CAN_BE_PROCESSED

#define INTERRUPTS_CAN_BE_PROCESSED()

Definition: miscadmin.h:129

References Assert(), DEBUG5, INTERRUPTS_CAN_BE_PROCESSED, pg_write_barrier, pgaio_debug_io, pgaio_io_state_get_name(), and PgAioHandle::state.

Referenced by pgaio_io_acquire_nb(), pgaio_io_prepare_submit(), pgaio_io_process_completion(), pgaio_io_reclaim(), and pgaio_io_stage().

◆ pgaio_io_wait()

static void pgaio_io_wait ( PgAioHandle * ioh,

uint64 ref_generation

)

static

Definition at line 576 of file aio.c.

577{

578 PgAioHandleState state;

579 bool am_owner;

580

581 am_owner = ioh->owner_procno == MyProcNumber;

582

583 if (pgaio_io_was_recycled(ioh, ref_generation, &state))

584 return;

585

586 if (am_owner)

587 {

588 if (state != PGAIO_HS_SUBMITTED

589 && state != PGAIO_HS_COMPLETED_IO

590 && state != PGAIO_HS_COMPLETED_SHARED

591 && state != PGAIO_HS_COMPLETED_LOCAL)

592 {

593 elog(PANIC, "waiting for own IO %d in wrong state: %s",

594 pgaio_io_get_id(ioh), pgaio_io_get_state_name(ioh));

595 }

596 }

597

598 while (true)

599 {

600 if (pgaio_io_was_recycled(ioh, ref_generation, &state))

601 return;

602

603 switch ((PgAioHandleState) state)

604 {

605 case PGAIO_HS_IDLE:

606 case PGAIO_HS_HANDED_OUT:

607 elog(ERROR, "IO in wrong state: %d", state);

608 break;

609

610 case PGAIO_HS_SUBMITTED:

611

612 /*

613 * If we need to wait via the IO method, do so now. Don't

614 * check via the IO method if the issuing backend is executing

615 * the IO synchronously.

616 */

617 if (pgaio_method_ops->wait_one && !(ioh->flags & PGAIO_HF_SYNCHRONOUS))

618 {

619 pgaio_method_ops->wait_one(ioh, ref_generation);

620 continue;

621 }

622 /* fallthrough */

623

624 /* waiting for owner to submit */

625 case PGAIO_HS_DEFINED:

626 case PGAIO_HS_STAGED:

627 /* waiting for reaper to complete */

628 /* fallthrough */

629 case PGAIO_HS_COMPLETED_IO:

630 /* shouldn't be able to hit this otherwise */

631 Assert(IsUnderPostmaster);

632 /* ensure we're going to get woken up */

633 ConditionVariablePrepareToSleep(&ioh->cv);

634

635 while (!pgaio_io_was_recycled(ioh, ref_generation, &state))

636 {

637 if (state == PGAIO_HS_COMPLETED_SHARED ||

638 state == PGAIO_HS_COMPLETED_LOCAL)

639 break;

640 ConditionVariableSleep(&ioh->cv, WAIT_EVENT_AIO_IO_COMPLETION);

641 }

642

643 ConditionVariableCancelSleep();

644 break;

645

646 case PGAIO_HS_COMPLETED_SHARED:

647 case PGAIO_HS_COMPLETED_LOCAL:

648

649 /*

650 * Note that no interrupts are processed between

651 * pgaio_io_was_recycled() and this check - that's important

652 * as otherwise an interrupt could have already reclaimed the

653 * handle.

654 */

655 if (am_owner)

656 pgaio_io_reclaim(ioh);

657 return;

658 }

659 }

660}

pgaio_io_get_id

int pgaio_io_get_id(PgAioHandle *ioh)

Definition: aio.c:339

pgaio_io_get_state_name

const char * pgaio_io_get_state_name(PgAioHandle *ioh)

Definition: aio.c:925

pgaio_io_was_recycled

bool pgaio_io_was_recycled(PgAioHandle *ioh, uint64 ref_generation, PgAioHandleState *state)

Definition: aio.c:556

ConditionVariableCancelSleep

bool ConditionVariableCancelSleep(void)

Definition: condition_variable.c:230

ConditionVariablePrepareToSleep

void ConditionVariablePrepareToSleep(ConditionVariable *cv)

Definition: condition_variable.c:56

ConditionVariableSleep

void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)

Definition: condition_variable.c:96

PANIC

#define PANIC

Definition: elog.h:42

IsUnderPostmaster

bool IsUnderPostmaster

Definition: globals.c:120

IoMethodOps::wait_one

void(* wait_one)(PgAioHandle *ioh, uint64 ref_generation)

Definition: aio_internal.h:329

state

Definition: regguts.h:323

References Assert(), ConditionVariableCancelSleep(), ConditionVariablePrepareToSleep(), ConditionVariableSleep(), PgAioHandle::cv, elog, ERROR, PgAioHandle::flags, IsUnderPostmaster, MyProcNumber, PgAioHandle::owner_procno, PANIC, PGAIO_HF_SYNCHRONOUS, PGAIO_HS_COMPLETED_IO, PGAIO_HS_COMPLETED_LOCAL, PGAIO_HS_COMPLETED_SHARED, PGAIO_HS_DEFINED, PGAIO_HS_HANDED_OUT, PGAIO_HS_IDLE, PGAIO_HS_STAGED, PGAIO_HS_SUBMITTED, pgaio_io_get_id(), pgaio_io_get_state_name(), pgaio_io_reclaim(), pgaio_io_was_recycled(), pgaio_method_ops, and IoMethodOps::wait_one.

Referenced by pgaio_closing_fd(), pgaio_io_wait_for_free(), pgaio_shutdown(), and pgaio_wref_wait().

◆ pgaio_io_wait_for_free()

static void pgaio_io_wait_for_free ( void )

static

Definition at line 758 of file aio.c.

759{

760 int reclaimed = 0;

761

762 pgaio_debug(DEBUG2, "waiting for free IO with %d pending, %u in-flight, %u idle IOs",

763 pgaio_my_backend->num_staged_ios,

764 dclist_count(&pgaio_my_backend->in_flight_ios),

765 dclist_count(&pgaio_my_backend->idle_ios));

766

767 /*

768 * First check if any of our IOs actually have completed - when using

769 * worker, that'll often be the case. We could do so as part of the loop

770 * below, but that'd potentially lead us to wait for some IO submitted

771 * before.

772 */

773 for (int i = 0; i < io_max_concurrency; i++)

774 {

775 PgAioHandle *ioh = &pgaio_ctl->io_handles[pgaio_my_backend->io_handle_off + i];

776

777 if (ioh->state == PGAIO_HS_COMPLETED_SHARED)

778 {

779 /*

780 * Note that no interrupts are processed between the state check

781 * and the call to reclaim - that's important as otherwise an

782 * interrupt could have already reclaimed the handle.

783 *

784 * Need to ensure that there's no reordering, in the more common

785 * paths, where we wait for IO, that's done by

786 * pgaio_io_was_recycled().

787 */

788 pg_read_barrier();

789 pgaio_io_reclaim(ioh);

790 reclaimed++;

791 }

792 }

793

794 if (reclaimed > 0)

795 return;

796

797 /*

798 * If we have any unsubmitted IOs, submit them now. We'll start waiting in

799 * a second, so it's better they're in flight. This also addresses the

800 * edge-case that all IOs are unsubmitted.

801 */

802 if (pgaio_my_backend->num_staged_ios > 0)

803 pgaio_submit_staged();

804

805 /* possibly some IOs finished during submission */

806 if (!dclist_is_empty(&pgaio_my_backend->idle_ios))

807 return;

808

809 if (dclist_count(&pgaio_my_backend->in_flight_ios) == 0)

810 ereport(ERROR,

811 errmsg_internal("no free IOs despite no in-flight IOs"),

812 errdetail_internal("%d pending, %u in-flight, %u idle IOs",

813 pgaio_my_backend->num_staged_ios,

814 dclist_count(&pgaio_my_backend->in_flight_ios),

815 dclist_count(&pgaio_my_backend->idle_ios)));

816

817 /*

818 * Wait for the oldest in-flight IO to complete.

819 *

820 * XXX: Reusing the general IO wait is suboptimal, we don't need to wait

821 * for that specific IO to complete, we just need *any* IO to complete.

822 */

823 {

824 PgAioHandle *ioh = dclist_head_element(PgAioHandle, node,

825 &pgaio_my_backend->in_flight_ios);

826 uint64 generation = ioh->generation;

827

828 switch ((PgAioHandleState) ioh->state)

829 {

830 /* should not be in in-flight list */

831 case PGAIO_HS_IDLE:

832 case PGAIO_HS_DEFINED:

833 case PGAIO_HS_HANDED_OUT:

834 case PGAIO_HS_STAGED:

835 case PGAIO_HS_COMPLETED_LOCAL:

836 elog(ERROR, "shouldn't get here with io:%d in state %d",

837 pgaio_io_get_id(ioh), ioh->state);

838 break;

839

840 case PGAIO_HS_COMPLETED_IO:

841 case PGAIO_HS_SUBMITTED:

842 pgaio_debug_io(DEBUG2, ioh,

843 "waiting for free io with %u in flight",

844 dclist_count(&pgaio_my_backend->in_flight_ios));

845

846 /*

847 * In a more general case this would be racy, because the

848 * generation could increase after we read ioh->state above.

849 * But we are only looking at IOs by the current backend and

850 * the IO can only be recycled by this backend. Even this is

851 * only OK because we get the handle's generation before

852 * potentially processing interrupts, e.g. as part of

853 * pgaio_debug_io().

854 */

855 pgaio_io_wait(ioh, generation);

856 break;

857

858 case PGAIO_HS_COMPLETED_SHARED:

859

860 /*

861 * It's possible that another backend just finished this IO.

862 *

863 * Note that no interrupts are processed between the state

864 * check and the call to reclaim - that's important as

865 * otherwise an interrupt could have already reclaimed the

866 * handle.

867 *

868 * Need to ensure that there's no reordering, in the more

869 * common paths, where we wait for IO, that's done by

870 * pgaio_io_was_recycled().

871 */

872 pg_read_barrier();

873 pgaio_io_reclaim(ioh);

874 break;

875 }

876

877 if (dclist_count(&pgaio_my_backend->idle_ios) == 0)

878 elog(PANIC, "no idle IO after waiting for IO to terminate");

879 return;

880 }

881}

io_max_concurrency

int io_max_concurrency

Definition: aio.c:75

pg_read_barrier

#define pg_read_barrier()

Definition: atomics.h:154

errmsg_internal

int errmsg_internal(const char *fmt,...)

Definition: elog.c:1161

errdetail_internal

int errdetail_internal(const char *fmt,...)

Definition: elog.c:1234

ereport

#define ereport(elevel,...)

Definition: elog.h:150

dclist_head_element

#define dclist_head_element(type, membername, lhead)

Definition: ilist.h:955

i

int i

Definition: isn.c:77

PgAioBackend::io_handle_off

uint32 io_handle_off

Definition: aio_internal.h:194

References dclist_count(), dclist_head_element, dclist_is_empty(), DEBUG2, elog, ereport, errdetail_internal(), errmsg_internal(), ERROR, PgAioHandle::generation, i, PgAioBackend::idle_ios, PgAioBackend::in_flight_ios, PgAioBackend::io_handle_off, PgAioCtl::io_handles, io_max_concurrency, PgAioBackend::num_staged_ios, PANIC, pg_read_barrier, pgaio_ctl, pgaio_debug, pgaio_debug_io, PGAIO_HS_COMPLETED_IO, PGAIO_HS_COMPLETED_LOCAL, PGAIO_HS_COMPLETED_SHARED, PGAIO_HS_DEFINED, PGAIO_HS_HANDED_OUT, PGAIO_HS_IDLE, PGAIO_HS_STAGED, PGAIO_HS_SUBMITTED, pgaio_io_get_id(), pgaio_io_reclaim(), pgaio_io_wait(), pgaio_my_backend, pgaio_submit_staged(), and PgAioHandle::state.

Referenced by pgaio_io_acquire().

◆ pgaio_io_was_recycled()

bool pgaio_io_was_recycled ( PgAioHandle * ioh,

uint64 ref_generation,

PgAioHandleState * state

)

Definition at line 556 of file aio.c.

557{

558 *state = ioh->state;

559

560 /*

561 * Ensure that we don't see an earlier state of the handle than ioh->state

562 * due to compiler or CPU reordering. This protects both ->generation as

563 * directly used here, and other fields in the handle accessed in the

564 * caller if the handle was not reused.

565 */

566 pg_read_barrier();

567

568 return ioh->generation != ref_generation;

569}

References PgAioHandle::generation, pg_read_barrier, and PgAioHandle::state.

Referenced by pgaio_io_wait(), and pgaio_wref_check_done().

◆ pgaio_result_status_string()

const char * pgaio_result_status_string ( PgAioResultStatus rs )

Definition at line 931 of file aio.c.

932{

933 switch ((PgAioResultStatus) rs)

934 {

935 case PGAIO_RS_UNKNOWN:

936 return "UNKNOWN";

937 case PGAIO_RS_OK:

938 return "OK";

939 case PGAIO_RS_WARNING:

940 return "WARNING";

941 case PGAIO_RS_PARTIAL:

942 return "PARTIAL";

943 case PGAIO_RS_ERROR:

944 return "ERROR";

945 }

946

947 return NULL; /* silence compiler */

948}

PgAioResultStatus

Definition: aio_types.h:79

PGAIO_RS_OK

@ PGAIO_RS_OK

Definition: aio_types.h:81

PGAIO_RS_PARTIAL

@ PGAIO_RS_PARTIAL

Definition: aio_types.h:82

PGAIO_RS_ERROR

@ PGAIO_RS_ERROR

Definition: aio_types.h:84

PGAIO_RS_WARNING

@ PGAIO_RS_WARNING

Definition: aio_types.h:83

References PGAIO_RS_ERROR, PGAIO_RS_OK, PGAIO_RS_PARTIAL, PGAIO_RS_UNKNOWN, and PGAIO_RS_WARNING.

Referenced by pg_get_aios(), pgaio_io_call_complete_local(), pgaio_io_call_complete_shared(), and pgaio_io_reclaim().

◆ pgaio_shutdown()

void pgaio_shutdown ( int code,

Datum arg

)

Definition at line 1285 of file aio.c.

1286{

1287 Assert(pgaio_my_backend);

1288 Assert(!pgaio_my_backend->handed_out_io);

1289

1290 /* first clean up resources as we would at a transaction boundary */

1291 AtEOXact_Aio(code == 0);

1292

1293 /*

1294 * Before exiting, make sure that all IOs are finished. That has two main

1295 * purposes:

1296 *

1297 * - Some kernel-level AIO mechanisms don't deal well with the issuer of

1298 * an AIO exiting before IO completed

1299 *

1300 * - It'd be confusing to see partially finished IOs in stats views etc

1301 */

1302 while (!dclist_is_empty(&pgaio_my_backend->in_flight_ios))

1303 {

1304 PgAioHandle *ioh = dclist_head_element(PgAioHandle, node, &pgaio_my_backend->in_flight_ios);

1305 uint64 generation = ioh->generation;

1306

1307 pgaio_debug_io(DEBUG2, ioh,

1308 "waiting for IO to complete during shutdown, %u in-flight IOs",

1309 dclist_count(&pgaio_my_backend->in_flight_ios));

1310

1311 /* see comment in pgaio_io_wait_for_free() about raciness */

1312 pgaio_io_wait(ioh, generation);

1313 }

1314

1315 pgaio_my_backend = NULL;

1316}

AtEOXact_Aio

void AtEOXact_Aio(bool is_commit)

Definition: aio.c:1190

References Assert(), AtEOXact_Aio(), dclist_count(), dclist_head_element, dclist_is_empty(), DEBUG2, PgAioHandle::generation, PgAioBackend::handed_out_io, PgAioBackend::in_flight_ios, pgaio_debug_io, pgaio_io_wait(), and pgaio_my_backend.

Referenced by pgaio_init_backend().

◆ pgaio_submit_staged()

void pgaio_submit_staged ( void )

Definition at line 1120 of file aio.c.

1121{

1122 int total_submitted = 0;

1123 int did_submit;

1124

1125 if (pgaio_my_backend->num_staged_ios == 0)

1126 return;

1127

1128

1129 START_CRIT_SECTION();

1130

1131 did_submit = pgaio_method_ops->submit(pgaio_my_backend->num_staged_ios,

1132 pgaio_my_backend->staged_ios);

1133

1134 END_CRIT_SECTION();

1135

1136 total_submitted += did_submit;

1137

1138 Assert(total_submitted == did_submit);

1139

1140 pgaio_my_backend->num_staged_ios = 0;

1141

1142 pgaio_debug(DEBUG4,

1143 "aio: submitted %d IOs",

1144 total_submitted);

1145}

START_CRIT_SECTION

#define START_CRIT_SECTION()

Definition: miscadmin.h:149

END_CRIT_SECTION

#define END_CRIT_SECTION()

Definition: miscadmin.h:151

IoMethodOps::submit

int(* submit)(uint16 num_staged_ios, PgAioHandle **staged_ios)

Definition: aio_internal.h:308

References Assert(), DEBUG4, END_CRIT_SECTION, PgAioBackend::num_staged_ios, pgaio_debug, pgaio_method_ops, pgaio_my_backend, PgAioBackend::staged_ios, START_CRIT_SECTION, and IoMethodOps::submit.

Referenced by AsyncReadBuffers(), pgaio_closing_fd(), pgaio_error_cleanup(), pgaio_exit_batchmode(), pgaio_io_acquire_nb(), pgaio_io_release_resowner(), pgaio_io_stage(), pgaio_io_wait_for_free(), and ReadBuffersCanStartIO().

◆ pgaio_wref_check_done()

bool pgaio_wref_check_done ( PgAioWaitRef * iow )

Definition at line 1002 of file aio.c.

1003{

1004 uint64 ref_generation;

1005 PgAioHandleState state;

1006 bool am_owner;

1007 PgAioHandle *ioh;

1008

1009 ioh = pgaio_io_from_wref(iow, &ref_generation);

1010

1011 if (pgaio_io_was_recycled(ioh, ref_generation, &state))

1012 return true;

1013

1014 if (state == PGAIO_HS_IDLE)

1015 return true;

1016

1017 am_owner = ioh->owner_procno == MyProcNumber;

1018

1019 if (state == PGAIO_HS_COMPLETED_SHARED ||

1020 state == PGAIO_HS_COMPLETED_LOCAL)

1021 {

1022 /*

1023 * Note that no interrupts are processed between

1024 * pgaio_io_was_recycled() and this check - that's important as

1025 * otherwise an interrupt could have already reclaimed the handle.

1026 */

1027 if (am_owner)

1028 pgaio_io_reclaim(ioh);

1029 return true;

1030 }

1031

1032 /*

1033 * XXX: It likely would be worth checking in with the io method, to give

1034 * the IO method a chance to check if there are completion events queued.

1035 */

1036

1037 return false;

1038}

pgaio_io_from_wref

static PgAioHandle * pgaio_io_from_wref(PgAioWaitRef *iow, uint64 *ref_generation)

Definition: aio.c:888

References MyProcNumber, PgAioHandle::owner_procno, PGAIO_HS_COMPLETED_LOCAL, PGAIO_HS_COMPLETED_SHARED, PGAIO_HS_IDLE, pgaio_io_from_wref(), pgaio_io_reclaim(), and pgaio_io_was_recycled().

Referenced by WaitReadBuffers().

◆ pgaio_wref_clear()

void pgaio_wref_clear ( PgAioWaitRef * iow )

Definition at line 961 of file aio.c.

962{

963 iow->aio_index = PG_UINT32_MAX;

964}

PG_UINT32_MAX

#define PG_UINT32_MAX

Definition: c.h:595

References PgAioWaitRef::aio_index, and PG_UINT32_MAX.

Referenced by AsyncReadBuffers(), BufferManagerShmemInit(), InitLocalBuffers(), StartReadBuffersImpl(), TerminateBufferIO(), and TerminateLocalBufferIO().

◆ pgaio_wref_get_id()

int pgaio_wref_get_id ( PgAioWaitRef * iow )

Definition at line 977 of file aio.c.

978{

979 Assert(pgaio_wref_valid(iow));

980 return iow->aio_index;

981}

pgaio_wref_valid

bool pgaio_wref_valid(PgAioWaitRef *iow)

Definition: aio.c:968

References PgAioWaitRef::aio_index, Assert(), and pgaio_wref_valid().

◆ pgaio_wref_valid()

bool pgaio_wref_valid ( PgAioWaitRef * iow )

Definition at line 968 of file aio.c.

969{

970 return iow->aio_index != PG_UINT32_MAX;

971}

References PgAioWaitRef::aio_index, and PG_UINT32_MAX.

Referenced by InvalidateLocalBuffer(), pgaio_wref_get_id(), ProcessReadBuffersResult(), StartLocalBufferIO(), WaitIO(), and WaitReadBuffers().

◆ pgaio_wref_wait()

void pgaio_wref_wait ( PgAioWaitRef * iow )

Definition at line 988 of file aio.c.

989{

990 uint64 ref_generation;

991 PgAioHandle *ioh;

992

993 ioh = pgaio_io_from_wref(iow, &ref_generation);

994

995 pgaio_io_wait(ioh, ref_generation);

996}

References pgaio_io_from_wref(), and pgaio_io_wait().

Referenced by InvalidateLocalBuffer(), read_rel_block_ll(), StartLocalBufferIO(), WaitIO(), and WaitReadBuffers().

Variable Documentation

◆ io_max_concurrency

int io_max_concurrency = -1

Definition at line 75 of file aio.c.

Referenced by AioHandleDataShmemSize(), AioHandleIOVShmemSize(), AioHandleShmemSize(), AioShmemInit(), AioShmemSize(), and pgaio_io_wait_for_free().

◆ io_method

int io_method = DEFAULT_IO_METHOD

Definition at line 74 of file aio.c.

Referenced by pgaio_workers_enabled(), read_stream_begin_impl(), StartReadBuffersImpl(), and WaitReadBuffers().

◆ io_method_options

const struct config_enum_entry io_method_options[]

Initial value:

= {

{"sync", IOMETHOD_SYNC, false},

{"worker", IOMETHOD_WORKER, false},

{NULL, 0, false}

}

IOMETHOD_WORKER

@ IOMETHOD_WORKER

Definition: aio.h:35

IOMETHOD_SYNC

@ IOMETHOD_SYNC

Definition: aio.h:34

Definition at line 64 of file aio.c.

Referenced by assign_io_method().

◆ pgaio_ctl

PgAioCtl* pgaio_ctl

Definition at line 78 of file aio.c.

Referenced by AioShmemInit(), IoWorkerMain(), pg_get_aios(), pgaio_init_backend(), pgaio_io_from_wref(), pgaio_io_get_handle_data(), pgaio_io_get_id(), pgaio_io_get_iovec(), pgaio_io_get_iovec_length(), pgaio_io_get_wref(), pgaio_io_perform_synchronously(), pgaio_io_set_handle_data_32(), pgaio_io_set_handle_data_64(), and pgaio_io_wait_for_free().

◆ pgaio_method_ops

const IoMethodOps* pgaio_method_ops

Definition at line 93 of file aio.c.

Referenced by AioShmemInit(), AioShmemSize(), assign_io_method(), pgaio_closing_fd(), pgaio_init_backend(), pgaio_io_needs_synchronous_execution(), pgaio_io_wait(), and pgaio_submit_staged().

◆ pgaio_method_ops_table

const IoMethodOps* const pgaio_method_ops_table[]

static

Initial value:

= {

[IOMETHOD_SYNC] = &pgaio_sync_ops,

[IOMETHOD_WORKER] = &pgaio_worker_ops,

}

pgaio_sync_ops

const IoMethodOps pgaio_sync_ops

Definition: method_sync.c:28

pgaio_worker_ops

const IoMethodOps pgaio_worker_ops

Definition: method_worker.c:84

Definition at line 84 of file aio.c.

Referenced by assign_io_method().

◆ pgaio_my_backend

PgAioBackend* pgaio_my_backend

Definition at line 81 of file aio.c.

Referenced by AtEOXact_Aio(), pgaio_closing_fd(), pgaio_enter_batchmode(), pgaio_error_cleanup(), pgaio_exit_batchmode(), pgaio_have_staged(), pgaio_init_backend(), pgaio_io_acquire_nb(), pgaio_io_before_start(), pgaio_io_prepare_submit(), pgaio_io_reclaim(), pgaio_io_release(), pgaio_io_release_resowner(), pgaio_io_stage(), pgaio_io_wait_for_free(), pgaio_shutdown(), and pgaio_submit_staged().