#include "postgres.h"
#include "miscadmin.h"
#include "storage/aio.h"
#include "storage/aio_internal.h"
#include "storage/bufmgr.h"
#include "storage/md.h"

Include dependency graph for aio_callback.c:

Go to the source code of this file.

Data Structures
struct	PgAioHandleCallbacksEntry

Macros
#define	CALLBACK_ENTRY(id, callback) [id] = {.cb = &callback, .name = #callback}

Typedefs
typedef struct PgAioHandleCallbacksEntry	PgAioHandleCallbacksEntry

Functions
void	pgaio_io_register_callbacks (PgAioHandle *ioh, PgAioHandleCallbackID cb_id, uint8 cb_data)

void	pgaio_io_set_handle_data_64 (PgAioHandle ioh, uint64 data, uint8 len)

void	pgaio_io_set_handle_data_32 (PgAioHandle ioh, uint32 data, uint8 len)

uint64 *	pgaio_io_get_handle_data (PgAioHandle ioh, uint8 len)

void	pgaio_result_report (PgAioResult result, const PgAioTargetData *target_data, int elevel)

void	pgaio_io_call_stage (PgAioHandle *ioh)

void	pgaio_io_call_complete_shared (PgAioHandle *ioh)

PgAioResult	pgaio_io_call_complete_local (PgAioHandle *ioh)

Variables
static const PgAioHandleCallbacks	aio_invalid_cb = {0}

static const PgAioHandleCallbacksEntry	aio_handle_cbs []

Macro Definition Documentation

◆ CALLBACK_ENTRY

#define CALLBACK_ENTRY	(	id,
		callback
	)	[id] = {.cb = &callback, .name = #callback}

Typedef Documentation

◆ PgAioHandleCallbacksEntry

typedef struct PgAioHandleCallbacksEntry PgAioHandleCallbacksEntry

Function Documentation

◆ pgaio_io_call_complete_local()

PgAioResult pgaio_io_call_complete_local ( PgAioHandle * ioh )

Definition at line 285 of file aio_callback.c.

{
    PgAioResult result;
 
    START_CRIT_SECTION();
 
    Assert(ioh->target > PGAIO_TID_INVALID && ioh->target < PGAIO_TID_COUNT);
    Assert(ioh->op > PGAIO_OP_INVALID && ioh->op < PGAIO_OP_COUNT);
 
    /* start with distilled result from shared callback */
    result = ioh->distilled_result;
    Assert(result.status != PGAIO_RS_UNKNOWN);
 
    for (int i = ioh->num_callbacks; i > 0; i--)
    {
        PgAioHandleCallbackID cb_id = ioh->callbacks[i - 1];
        uint8       cb_data = ioh->callbacks_data[i - 1];
        const PgAioHandleCallbacksEntry *ce = &aio_handle_cbs[cb_id];
 
        if (!ce->cb->complete_local)
            continue;
 
        pgaio_debug_io(DEBUG4, ioh,
                       "calling cb #%d, id %d/%s->complete_local(%u) with distilled result: status %s, id %u, error_data %d, result %d",
                       i, cb_id, ce->name, cb_data,
                       pgaio_result_status_string(result.status),
                       result.id, result.error_data, result.result);
        result = ce->cb->complete_local(ioh, result, cb_data);
 
        /* the callback should never transition to unknown */
        Assert(result.status != PGAIO_RS_UNKNOWN);
    }
 
    /*
     * Note that we don't save the result in ioh->distilled_result, the local
     * callback's result should not ever matter to other waiters. However, the
     * local backend does care, so we return the result as modified by local
     * callbacks, which then can be passed to ioh->report_return->result.
     */
    pgaio_debug_io(DEBUG3, ioh,
                   "after local completion: result: (status %s, id %u, error_data %d, result %d), raw_result: %d",
                   pgaio_result_status_string(result.status),
                   result.id, result.error_data, result.result,
                   ioh->result);
 
    END_CRIT_SECTION();
 
    return result;
}

References aio_handle_cbs, Assert, DEBUG3, DEBUG4, END_CRIT_SECTION, PgAioResult::error_data, fb(), i, PgAioResult::id, pgaio_debug_io, PGAIO_OP_COUNT, PGAIO_OP_INVALID, pgaio_result_status_string(), PGAIO_RS_UNKNOWN, PGAIO_TID_COUNT, PGAIO_TID_INVALID, PgAioResult::result, START_CRIT_SECTION, and PgAioResult::status.

Referenced by pgaio_io_reclaim().

◆ pgaio_io_call_complete_shared()

void pgaio_io_call_complete_shared ( PgAioHandle * ioh )

Definition at line 225 of file aio_callback.c.

{
    PgAioResult result;
 
    START_CRIT_SECTION();
 
    Assert(ioh->target > PGAIO_TID_INVALID && ioh->target < PGAIO_TID_COUNT);
    Assert(ioh->op > PGAIO_OP_INVALID && ioh->op < PGAIO_OP_COUNT);
 
    result.status = PGAIO_RS_OK;    /* low level IO is always considered OK */
    result.result = ioh->result;
    result.id = PGAIO_HCB_INVALID;
    result.error_data = 0;
 
    /*
     * Call callbacks with the last registered (innermost) callback first.
     * Each callback can modify the result forwarded to the next callback.
     */
    for (int i = ioh->num_callbacks; i > 0; i--)
    {
        PgAioHandleCallbackID cb_id = ioh->callbacks[i - 1];
        uint8       cb_data = ioh->callbacks_data[i - 1];
        const PgAioHandleCallbacksEntry *ce = &aio_handle_cbs[cb_id];
 
        if (!ce->cb->complete_shared)
            continue;
 
        pgaio_debug_io(DEBUG4, ioh,
                       "calling cb #%d, id %d/%s->complete_shared(%u) with distilled result: (status %s, id %u, error_data %d, result %d)",
                       i, cb_id, ce->name,
                       cb_data,
                       pgaio_result_status_string(result.status),
                       result.id, result.error_data, result.result);
        result = ce->cb->complete_shared(ioh, result, cb_data);
 
        /* the callback should never transition to unknown */
        Assert(result.status != PGAIO_RS_UNKNOWN);
    }
 
    ioh->distilled_result = result;
 
    pgaio_debug_io(DEBUG3, ioh,
                   "after shared completion: distilled result: (status %s, id %u, error_data: %d, result %d), raw_result: %d",
                   pgaio_result_status_string(result.status),
                   result.id, result.error_data, result.result,
                   ioh->result);
 
    END_CRIT_SECTION();
}

References aio_handle_cbs, Assert, DEBUG3, DEBUG4, END_CRIT_SECTION, PgAioResult::error_data, fb(), i, PgAioResult::id, pgaio_debug_io, PGAIO_HCB_INVALID, PGAIO_OP_COUNT, PGAIO_OP_INVALID, pgaio_result_status_string(), PGAIO_RS_OK, PGAIO_RS_UNKNOWN, PGAIO_TID_COUNT, PGAIO_TID_INVALID, PgAioResult::result, START_CRIT_SECTION, and PgAioResult::status.

Referenced by pgaio_io_process_completion().

◆ pgaio_io_call_stage()

void pgaio_io_call_stage ( PgAioHandle * ioh )

Definition at line 199 of file aio_callback.c.

{
    Assert(ioh->target > PGAIO_TID_INVALID && ioh->target < PGAIO_TID_COUNT);
    Assert(ioh->op > PGAIO_OP_INVALID && ioh->op < PGAIO_OP_COUNT);
 
    for (int i = ioh->num_callbacks; i > 0; i--)
    {
        PgAioHandleCallbackID cb_id = ioh->callbacks[i - 1];
        uint8       cb_data = ioh->callbacks_data[i - 1];
        const PgAioHandleCallbacksEntry *ce = &aio_handle_cbs[cb_id];
 
        if (!ce->cb->stage)
            continue;
 
        pgaio_debug_io(DEBUG3, ioh,
                       "calling cb #%d %d/%s->stage(%u)",
                       i, cb_id, ce->name, cb_data);
        ce->cb->stage(ioh, cb_data);
    }
}

References aio_handle_cbs, Assert, DEBUG3, fb(), i, pgaio_debug_io, PGAIO_OP_COUNT, PGAIO_OP_INVALID, PGAIO_TID_COUNT, and PGAIO_TID_INVALID.

Referenced by pgaio_io_stage().

◆ pgaio_io_get_handle_data()

uint64 * pgaio_io_get_handle_data	(	PgAioHandle *	ioh,
		uint8 *	len
	)

Definition at line 156 of file aio_callback.c.

{
    Assert(ioh->handle_data_len > 0);
 
    *len = ioh->handle_data_len;
 
    return &pgaio_ctl->handle_data[ioh->iovec_off];
}

References Assert, fb(), PgAioCtl::handle_data, len, and pgaio_ctl.

Referenced by buffer_readv_complete(), and buffer_stage_common().

◆ pgaio_io_register_callbacks()

void pgaio_io_register_callbacks	(	PgAioHandle *	ioh,
		PgAioHandleCallbackID	cb_id,
		uint8	cb_data
	)

Definition at line 86 of file aio_callback.c.

{
    const PgAioHandleCallbacksEntry *ce = &aio_handle_cbs[cb_id];
 
    Assert(cb_id <= PGAIO_HCB_MAX);
    if (cb_id >= lengthof(aio_handle_cbs))
        elog(ERROR, "callback %d is out of range", cb_id);
    if (aio_handle_cbs[cb_id].cb->complete_shared == NULL &&
        aio_handle_cbs[cb_id].cb->complete_local == NULL)
        elog(ERROR, "callback %d does not have a completion callback", cb_id);
    if (ioh->num_callbacks >= PGAIO_HANDLE_MAX_CALLBACKS)
        elog(PANIC, "too many callbacks, the max is %d",
             PGAIO_HANDLE_MAX_CALLBACKS);
    ioh->callbacks[ioh->num_callbacks] = cb_id;
    ioh->callbacks_data[ioh->num_callbacks] = cb_data;
 
    pgaio_debug_io(DEBUG3, ioh,
                   "adding cb #%d, id %d/%s",
                   ioh->num_callbacks + 1,
                   cb_id, ce->name);
 
    ioh->num_callbacks++;
}

References aio_handle_cbs, Assert, PgAioHandleCallbacksEntry::cb, PgAioHandleCallbacks::complete_local, DEBUG3, elog, ERROR, fb(), lengthof, PANIC, pgaio_debug_io, PGAIO_HANDLE_MAX_CALLBACKS, and PGAIO_HCB_MAX.

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

◆ pgaio_io_set_handle_data_32()

void pgaio_io_set_handle_data_32	(	PgAioHandle *	ioh,
		uint32 *	data,
		uint8	len
	)

Definition at line 140 of file aio_callback.c.

{
    Assert(ioh->state == PGAIO_HS_HANDED_OUT);
    Assert(ioh->handle_data_len == 0);
    Assert(len <= PG_IOV_MAX);
    Assert(len <= io_max_combine_limit);
 
    for (int i = 0; i < len; i++)
        pgaio_ctl->handle_data[ioh->iovec_off + i] = data[i];
    ioh->handle_data_len = len;
}

References Assert, data, fb(), PgAioCtl::handle_data, i, io_max_combine_limit, len, PG_IOV_MAX, pgaio_ctl, and PGAIO_HS_HANDED_OUT.

Referenced by AsyncReadBuffers(), and read_rel_block_ll().

◆ pgaio_io_set_handle_data_64()

void pgaio_io_set_handle_data_64	(	PgAioHandle *	ioh,
		uint64 *	data,
		uint8	len
	)

Definition at line 122 of file aio_callback.c.

{
    Assert(ioh->state == PGAIO_HS_HANDED_OUT);
    Assert(ioh->handle_data_len == 0);
    Assert(len <= PG_IOV_MAX);
    Assert(len <= io_max_combine_limit);
 
    for (int i = 0; i < len; i++)
        pgaio_ctl->handle_data[ioh->iovec_off + i] = data[i];
    ioh->handle_data_len = len;
}

References Assert, data, fb(), PgAioCtl::handle_data, i, io_max_combine_limit, len, PG_IOV_MAX, pgaio_ctl, and PGAIO_HS_HANDED_OUT.

◆ pgaio_result_report()

void pgaio_result_report	(	PgAioResult	result,
		const PgAioTargetData *	target_data,
		int	elevel
	)

Definition at line 173 of file aio_callback.c.

{
    PgAioHandleCallbackID cb_id = result.id;
    const PgAioHandleCallbacksEntry *ce = &aio_handle_cbs[cb_id];
 
    Assert(result.status != PGAIO_RS_UNKNOWN);
    Assert(result.status != PGAIO_RS_OK);
 
    if (ce->cb->report == NULL)
        elog(ERROR, "callback %d/%s does not have report callback",
             result.id, ce->name);
 
    ce->cb->report(result, target_data, elevel);
}

References aio_handle_cbs, Assert, elog, ERROR, fb(), PgAioResult::id, PGAIO_RS_OK, PGAIO_RS_UNKNOWN, and PgAioResult::status.

Referenced by buffer_readv_complete(), buffer_readv_complete_one(), md_readv_complete(), ProcessReadBuffersResult(), and read_rel_block_ll().

Variable Documentation

◆ aio_handle_cbs

const PgAioHandleCallbacksEntry aio_handle_cbs[]

static

Initial value:

= {
#define CALLBACK_ENTRY(id, callback)                                             
    CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
 
    CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
 
    CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
 
    CALLBACK_ENTRY(PGAIO_HCB_LOCAL_BUFFER_READV, aio_local_buffer_readv_cb),
 
}

Definition at line 39 of file aio_callback.c.

                                                          {
#define CALLBACK_ENTRY(id, callback)  [id] = {.cb = &callback, .name = #callback}
    CALLBACK_ENTRY(PGAIO_HCB_INVALID, aio_invalid_cb),
 
    CALLBACK_ENTRY(PGAIO_HCB_MD_READV, aio_md_readv_cb),
 
    CALLBACK_ENTRY(PGAIO_HCB_SHARED_BUFFER_READV, aio_shared_buffer_readv_cb),
 
    CALLBACK_ENTRY(PGAIO_HCB_LOCAL_BUFFER_READV, aio_local_buffer_readv_cb),
#undef CALLBACK_ENTRY
};

Referenced by pgaio_io_call_complete_local(), pgaio_io_call_complete_shared(), pgaio_io_call_stage(), pgaio_io_register_callbacks(), and pgaio_result_report().

◆ aio_invalid_cb

const PgAioHandleCallbacks aio_invalid_cb = {0}