method_worker.c File Reference

#include "postgres.h"
#include "libpq/pqsignal.h"
#include "miscadmin.h"
#include "port/pg_bitutils.h"
#include "postmaster/auxprocess.h"
#include "postmaster/interrupt.h"
#include "storage/aio.h"
#include "storage/aio_internal.h"
#include "storage/aio_subsys.h"
#include "storage/io_worker.h"
#include "storage/ipc.h"
#include "storage/latch.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/injection_point.h"
#include "utils/memdebug.h"
#include "utils/ps_status.h"
#include "utils/wait_event.h"

Include dependency graph for method_worker.c:

Go to the source code of this file.

Data Structures
struct	PgAioWorkerSubmissionQueue
struct	PgAioWorkerSlot
struct	PgAioWorkerControl

Macros
#define	IO_WORKER_WAKEUP_FANOUT   2

Typedefs
typedef struct PgAioWorkerSubmissionQueue	PgAioWorkerSubmissionQueue
typedef struct PgAioWorkerSlot	PgAioWorkerSlot
typedef struct PgAioWorkerControl	PgAioWorkerControl

Functions
static size_t	pgaio_worker_shmem_size (void)
static void	pgaio_worker_shmem_init (bool first_time)
static bool	pgaio_worker_needs_synchronous_execution (PgAioHandle *ioh)
static int	pgaio_worker_submit (uint16 num_staged_ios, PgAioHandle **staged_ios)
static size_t	pgaio_worker_queue_shmem_size (int *queue_size)
static size_t	pgaio_worker_control_shmem_size (void)
static int	pgaio_worker_choose_idle (void)
static bool	pgaio_worker_submission_queue_insert (PgAioHandle *ioh)
static int	pgaio_worker_submission_queue_consume (void)
static uint32	pgaio_worker_submission_queue_depth (void)
static void	pgaio_worker_submit_internal (int num_staged_ios, PgAioHandle **staged_ios)
static void	pgaio_worker_die (int code, Datum arg)
static void	pgaio_worker_register (void)
static void	pgaio_worker_error_callback (void *arg)
void	IoWorkerMain (const void *startup_data, size_t startup_data_len)
bool	pgaio_workers_enabled (void)

Variables
const IoMethodOps	pgaio_worker_ops
int	io_workers = 3
static int	io_worker_queue_size = 64
static int	MyIoWorkerId
static PgAioWorkerSubmissionQueue *	io_worker_submission_queue
static PgAioWorkerControl *	io_worker_control

Macro Definition Documentation

IO_WORKER_WAKEUP_FANOUT

#define IO_WORKER_WAKEUP_FANOUT   2

Definition at line 52 of file method_worker.c.

Typedef Documentation

PgAioWorkerControl

typedef struct PgAioWorkerControl PgAioWorkerControl

PgAioWorkerSlot

typedef struct PgAioWorkerSlot PgAioWorkerSlot

PgAioWorkerSubmissionQueue

typedef struct PgAioWorkerSubmissionQueue PgAioWorkerSubmissionQueue

Function Documentation

IoWorkerMain()

void IoWorkerMain	(	const void *	startup_data,
		size_t	startup_data_len
	)

Definition at line 396 of file method_worker.c.

{
    sigjmp_buf  local_sigjmp_buf;
    PgAioHandle *volatile error_ioh = NULL;
    ErrorContextCallback errcallback = {0};
    volatile int error_errno = 0;
    char        cmd[128];
 
    AuxiliaryProcessMainCommon();
 
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, die);      /* to allow manually triggering worker restart */
 
    /*
     * Ignore SIGTERM, will get explicit shutdown via SIGUSR2 later in the
     * shutdown sequence, similar to checkpointer.
     */
    pqsignal(SIGTERM, SIG_IGN);
    /* SIGQUIT handler was already set up by InitPostmasterChild */
    pqsignal(SIGALRM, SIG_IGN);
    pqsignal(SIGPIPE, SIG_IGN);
    pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    pqsignal(SIGUSR2, SignalHandlerForShutdownRequest);
 
    /* also registers a shutdown callback to unregister */
    pgaio_worker_register();
 
    sprintf(cmd, "%d", MyIoWorkerId);
    set_ps_display(cmd);
 
    errcallback.callback = pgaio_worker_error_callback;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;
 
    /* see PostgresMain() */
    if (sigsetjmp(local_sigjmp_buf, 1) != 0)
    {
        error_context_stack = NULL;
        HOLD_INTERRUPTS();
 
        EmitErrorReport();
 
        /*
         * In the - very unlikely - case that the IO failed in a way that
         * raises an error we need to mark the IO as failed.
         *
         * Need to do just enough error recovery so that we can mark the IO as
         * failed and then exit (postmaster will start a new worker).
         */
        LWLockReleaseAll();
 
        if (error_ioh != NULL)
        {
            /* should never fail without setting error_errno */
            Assert(error_errno != 0);
 
            errno = error_errno;
 
            START_CRIT_SECTION();
            pgaio_io_process_completion(error_ioh, -error_errno);
            END_CRIT_SECTION();
        }
 
        proc_exit(1);
    }
 
    /* We can now handle ereport(ERROR) */
    PG_exception_stack = &local_sigjmp_buf;
 
    sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
 
    while (!ShutdownRequestPending)
    {
        uint32      io_index;
        Latch      *latches[IO_WORKER_WAKEUP_FANOUT];
        int         nlatches = 0;
        int         nwakeups = 0;
        int         worker;
 
        /*
         * Try to get a job to do.
         *
         * The lwlock acquisition also provides the necessary memory barrier
         * to ensure that we don't see an outdated data in the handle.
         */
        LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
        if ((io_index = pgaio_worker_submission_queue_consume()) == -1)
        {
            /*
             * Nothing to do.  Mark self idle.
             *
             * XXX: Invent some kind of back pressure to reduce useless
             * wakeups?
             */
            io_worker_control->idle_worker_mask |= (UINT64_C(1) << MyIoWorkerId);
        }
        else
        {
            /* Got one.  Clear idle flag. */
            io_worker_control->idle_worker_mask &= ~(UINT64_C(1) << MyIoWorkerId);
 
            /* See if we can wake up some peers. */
            nwakeups = Min(pgaio_worker_submission_queue_depth(),
                           IO_WORKER_WAKEUP_FANOUT);
            for (int i = 0; i < nwakeups; ++i)
            {
                if ((worker = pgaio_worker_choose_idle()) < 0)
                    break;
                latches[nlatches++] = io_worker_control->workers[worker].latch;
            }
        }
        LWLockRelease(AioWorkerSubmissionQueueLock);
 
        for (int i = 0; i < nlatches; ++i)
            SetLatch(latches[i]);
 
        if (io_index != -1)
        {
            PgAioHandle *ioh = NULL;
 
            ioh = &pgaio_ctl->io_handles[io_index];
            error_ioh = ioh;
            errcallback.arg = ioh;
 
            pgaio_debug_io(DEBUG4, ioh,
                           "worker %d processing IO",
                           MyIoWorkerId);
 
            /*
             * Prevent interrupts between pgaio_io_reopen() and
             * pgaio_io_perform_synchronously() that otherwise could lead to
             * the FD getting closed in that window.
             */
            HOLD_INTERRUPTS();
 
            /*
             * It's very unlikely, but possible, that reopen fails. E.g. due
             * to memory allocations failing or file permissions changing or
             * such.  In that case we need to fail the IO.
             *
             * There's not really a good errno we can report here.
             */
            error_errno = ENOENT;
            pgaio_io_reopen(ioh);
 
            /*
             * To be able to exercise the reopen-fails path, allow injection
             * points to trigger a failure at this point.
             */
            INJECTION_POINT("aio-worker-after-reopen", ioh);
 
            error_errno = 0;
            error_ioh = NULL;
 
            /*
             * As part of IO completion the buffer will be marked as NOACCESS,
             * until the buffer is pinned again - which never happens in io
             * workers. Therefore the next time there is IO for the same
             * buffer, the memory will be considered inaccessible. To avoid
             * that, explicitly allow access to the memory before reading data
             * into it.
             */
#ifdef USE_VALGRIND
            {
                struct iovec *iov;
                uint16      iov_length = pgaio_io_get_iovec_length(ioh, &iov);
 
                for (int i = 0; i < iov_length; i++)
                    VALGRIND_MAKE_MEM_UNDEFINED(iov[i].iov_base, iov[i].iov_len);
            }
#endif
 
            /*
             * We don't expect this to ever fail with ERROR or FATAL, no need
             * to keep error_ioh set to the IO.
             * pgaio_io_perform_synchronously() contains a critical section to
             * ensure we don't accidentally fail.
             */
            pgaio_io_perform_synchronously(ioh);
 
            RESUME_INTERRUPTS();
            errcallback.arg = NULL;
        }
        else
        {
            WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH, -1,
                      WAIT_EVENT_IO_WORKER_MAIN);
            ResetLatch(MyLatch);
        }
 
        CHECK_FOR_INTERRUPTS();
 
        if (ConfigReloadPending)
        {
            ConfigReloadPending = false;
            ProcessConfigFile(PGC_SIGHUP);
        }
    }
 
    error_context_stack = errcallback.previous;
    proc_exit(0);
}

References ErrorContextCallback::arg, Assert, AuxiliaryProcessMainCommon(), ErrorContextCallback::callback, CHECK_FOR_INTERRUPTS, ConfigReloadPending, DEBUG4, die, EmitErrorReport(), END_CRIT_SECTION, error_context_stack, fb(), HOLD_INTERRUPTS, i, PgAioWorkerControl::idle_worker_mask, INJECTION_POINT, PgAioCtl::io_handles, io_worker_control, IO_WORKER_WAKEUP_FANOUT, PgAioWorkerSlot::latch, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LWLockReleaseAll(), Min, MyIoWorkerId, MyLatch, PG_exception_stack, pgaio_ctl, pgaio_debug_io, pgaio_io_get_iovec_length(), pgaio_io_perform_synchronously(), pgaio_io_process_completion(), pgaio_io_reopen(), pgaio_worker_choose_idle(), pgaio_worker_error_callback(), pgaio_worker_register(), pgaio_worker_submission_queue_consume(), pgaio_worker_submission_queue_depth(), PGC_SIGHUP, pqsignal, ErrorContextCallback::previous, proc_exit(), ProcessConfigFile(), procsignal_sigusr1_handler(), ResetLatch(), RESUME_INTERRUPTS, set_ps_display(), SetLatch(), ShutdownRequestPending, SIGALRM, SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SIGPIPE, SIGUSR1, SIGUSR2, sprintf, START_CRIT_SECTION, UnBlockSig, VALGRIND_MAKE_MEM_UNDEFINED, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and PgAioWorkerControl::workers.

pgaio_worker_choose_idle()

static int pgaio_worker_choose_idle ( void  )

static

Definition at line 164 of file method_worker.c.

{
    int         worker;
 
    if (io_worker_control->idle_worker_mask == 0)
        return -1;
 
    /* Find the lowest bit position, and clear it. */
    worker = pg_rightmost_one_pos64(io_worker_control->idle_worker_mask);
    io_worker_control->idle_worker_mask &= ~(UINT64_C(1) << worker);
    Assert(io_worker_control->workers[worker].in_use);
 
    return worker;
}

References Assert, fb(), PgAioWorkerControl::idle_worker_mask, PgAioWorkerSlot::in_use, io_worker_control, pg_rightmost_one_pos64(), and PgAioWorkerControl::workers.

Referenced by IoWorkerMain(), and pgaio_worker_submit_internal().

pgaio_worker_control_shmem_size()

static size_t pgaio_worker_control_shmem_size ( void  )

static

Definition at line 113 of file method_worker.c.

{
    return offsetof(PgAioWorkerControl, workers) +
        sizeof(PgAioWorkerSlot) * MAX_IO_WORKERS;
}

References fb(), and MAX_IO_WORKERS.

Referenced by pgaio_worker_shmem_init(), and pgaio_worker_shmem_size().

pgaio_worker_die()

static void pgaio_worker_die ( int  code, Datum  arg  )

static

Definition at line 324 of file method_worker.c.

{
    LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
    Assert(io_worker_control->workers[MyIoWorkerId].in_use);
    Assert(io_worker_control->workers[MyIoWorkerId].latch == MyLatch);
 
    io_worker_control->idle_worker_mask &= ~(UINT64_C(1) << MyIoWorkerId);
    io_worker_control->workers[MyIoWorkerId].in_use = false;
    io_worker_control->workers[MyIoWorkerId].latch = NULL;
    LWLockRelease(AioWorkerSubmissionQueueLock);
}

References Assert, fb(), PgAioWorkerControl::idle_worker_mask, PgAioWorkerSlot::in_use, io_worker_control, PgAioWorkerSlot::latch, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MyIoWorkerId, MyLatch, and PgAioWorkerControl::workers.

Referenced by pgaio_worker_register().

pgaio_worker_error_callback()

static void pgaio_worker_error_callback ( void *  arg )

static

Definition at line 375 of file method_worker.c.

{
    ProcNumber  owner;
    PGPROC     *owner_proc;
    int32       owner_pid;
    PgAioHandle *ioh = arg;
 
    if (!ioh)
        return;
 
    Assert(ioh->owner_procno != MyProcNumber);
    Assert(MyBackendType == B_IO_WORKER);
 
    owner = ioh->owner_procno;
    owner_proc = GetPGProcByNumber(owner);
    owner_pid = owner_proc->pid;
 
    errcontext("I/O worker executing I/O on behalf of process %d", owner_pid);
}

References arg, Assert, B_IO_WORKER, errcontext, fb(), GetPGProcByNumber, MyBackendType, and MyProcNumber.

Referenced by IoWorkerMain().

pgaio_worker_needs_synchronous_execution()

static bool pgaio_worker_needs_synchronous_execution ( PgAioHandle *  ioh )

static

Definition at line 234 of file method_worker.c.

{
    return
        !IsUnderPostmaster
        || ioh->flags & PGAIO_HF_REFERENCES_LOCAL
        || !pgaio_io_can_reopen(ioh);
}

References fb(), IsUnderPostmaster, PGAIO_HF_REFERENCES_LOCAL, and pgaio_io_can_reopen().

Referenced by pgaio_worker_submit_internal().

pgaio_worker_queue_shmem_size()

static size_t pgaio_worker_queue_shmem_size ( int *  queue_size )

static

Definition at line 103 of file method_worker.c.

{
    /* Round size up to next power of two so we can make a mask. */
    *queue_size = pg_nextpower2_32(io_worker_queue_size);
 
    return offsetof(PgAioWorkerSubmissionQueue, sqes) +
        sizeof(int) * *queue_size;
}

References fb(), io_worker_queue_size, and pg_nextpower2_32().

Referenced by pgaio_worker_shmem_init(), and pgaio_worker_shmem_size().

pgaio_worker_register()

static void pgaio_worker_register ( void  )

static

Definition at line 341 of file method_worker.c.

{
    MyIoWorkerId = -1;
 
    /*
     * XXX: This could do with more fine-grained locking. But it's also not
     * very common for the number of workers to change at the moment...
     */
    LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
 
    for (int i = 0; i < MAX_IO_WORKERS; ++i)
    {
        if (!io_worker_control->workers[i].in_use)
        {
            Assert(io_worker_control->workers[i].latch == NULL);
            io_worker_control->workers[i].in_use = true;
            MyIoWorkerId = i;
            break;
        }
        else
            Assert(io_worker_control->workers[i].latch != NULL);
    }
 
    if (MyIoWorkerId == -1)
        elog(ERROR, "couldn't find a free worker slot");
 
    io_worker_control->idle_worker_mask |= (UINT64_C(1) << MyIoWorkerId);
    io_worker_control->workers[MyIoWorkerId].latch = MyLatch;
    LWLockRelease(AioWorkerSubmissionQueueLock);
 
    on_shmem_exit(pgaio_worker_die, 0);
}

References Assert, elog, ERROR, fb(), i, PgAioWorkerControl::idle_worker_mask, PgAioWorkerSlot::in_use, io_worker_control, PgAioWorkerSlot::latch, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), MAX_IO_WORKERS, MyIoWorkerId, MyLatch, on_shmem_exit(), pgaio_worker_die(), and PgAioWorkerControl::workers.

Referenced by IoWorkerMain().

pgaio_worker_shmem_init()

static void pgaio_worker_shmem_init ( bool  first_time )

static

Definition at line 132 of file method_worker.c.

{
    bool        found;
    int         queue_size;
 
    io_worker_submission_queue =
        ShmemInitStruct("AioWorkerSubmissionQueue",
                        pgaio_worker_queue_shmem_size(&queue_size),
                        &found);
    if (!found)
    {
        io_worker_submission_queue->size = queue_size;
        io_worker_submission_queue->head = 0;
        io_worker_submission_queue->tail = 0;
    }
 
    io_worker_control =
        ShmemInitStruct("AioWorkerControl",
                        pgaio_worker_control_shmem_size(),
                        &found);
    if (!found)
    {
        io_worker_control->idle_worker_mask = 0;
        for (int i = 0; i < MAX_IO_WORKERS; ++i)
        {
            io_worker_control->workers[i].latch = NULL;
            io_worker_control->workers[i].in_use = false;
        }
    }
}

References fb(), PgAioWorkerSubmissionQueue::head, i, PgAioWorkerControl::idle_worker_mask, PgAioWorkerSlot::in_use, io_worker_control, io_worker_submission_queue, PgAioWorkerSlot::latch, MAX_IO_WORKERS, pgaio_worker_control_shmem_size(), pgaio_worker_queue_shmem_size(), ShmemInitStruct(), PgAioWorkerSubmissionQueue::size, PgAioWorkerSubmissionQueue::tail, and PgAioWorkerControl::workers.

pgaio_worker_shmem_size()

static size_t pgaio_worker_shmem_size ( void  )

static

Definition at line 120 of file method_worker.c.

{
    size_t      sz;
    int         queue_size;
 
    sz = pgaio_worker_queue_shmem_size(&queue_size);
    sz = add_size(sz, pgaio_worker_control_shmem_size());
 
    return sz;
}

References add_size(), fb(), pgaio_worker_control_shmem_size(), and pgaio_worker_queue_shmem_size().

pgaio_worker_submission_queue_consume()

static int pgaio_worker_submission_queue_consume ( void  )

static

Definition at line 201 of file method_worker.c.

{
    PgAioWorkerSubmissionQueue *queue;
    int         result;
 
    queue = io_worker_submission_queue;
    if (queue->tail == queue->head)
        return -1;              /* empty */
 
    result = queue->sqes[queue->tail];
    queue->tail = (queue->tail + 1) & (queue->size - 1);
 
    return result;
}

References PgAioWorkerSubmissionQueue::head, io_worker_submission_queue, PgAioWorkerSubmissionQueue::size, PgAioWorkerSubmissionQueue::sqes, and PgAioWorkerSubmissionQueue::tail.

Referenced by IoWorkerMain().

pgaio_worker_submission_queue_depth()

static uint32 pgaio_worker_submission_queue_depth ( void  )

static

Definition at line 217 of file method_worker.c.

{
    uint32      head;
    uint32      tail;
 
    head = io_worker_submission_queue->head;
    tail = io_worker_submission_queue->tail;
 
    if (tail > head)
        head += io_worker_submission_queue->size;
 
    Assert(head >= tail);
 
    return head - tail;
}

References Assert, PgAioWorkerSubmissionQueue::head, io_worker_submission_queue, PgAioWorkerSubmissionQueue::size, and PgAioWorkerSubmissionQueue::tail.

Referenced by IoWorkerMain().

pgaio_worker_submission_queue_insert()

static bool pgaio_worker_submission_queue_insert ( PgAioHandle *  ioh )

static

Definition at line 180 of file method_worker.c.

{
    PgAioWorkerSubmissionQueue *queue;
    uint32      new_head;
 
    queue = io_worker_submission_queue;
    new_head = (queue->head + 1) & (queue->size - 1);
    if (new_head == queue->tail)
    {
        pgaio_debug(DEBUG3, "io queue is full, at %u elements",
                    io_worker_submission_queue->size);
        return false;           /* full */
    }
 
    queue->sqes[queue->head] = pgaio_io_get_id(ioh);
    queue->head = new_head;
 
    return true;
}

References DEBUG3, fb(), PgAioWorkerSubmissionQueue::head, io_worker_submission_queue, pgaio_debug, pgaio_io_get_id(), PgAioWorkerSubmissionQueue::size, PgAioWorkerSubmissionQueue::sqes, and PgAioWorkerSubmissionQueue::tail.

Referenced by pgaio_worker_submit_internal().

pgaio_worker_submit()

static int pgaio_worker_submit ( uint16  num_staged_ios, PgAioHandle **  staged_ios  )

static

Definition at line 305 of file method_worker.c.

{
    for (int i = 0; i < num_staged_ios; i++)
    {
        PgAioHandle *ioh = staged_ios[i];
 
        pgaio_io_prepare_submit(ioh);
    }
 
    pgaio_worker_submit_internal(num_staged_ios, staged_ios);
 
    return num_staged_ios;
}

References fb(), i, pgaio_io_prepare_submit(), and pgaio_worker_submit_internal().

pgaio_worker_submit_internal()

static void pgaio_worker_submit_internal ( int  num_staged_ios, PgAioHandle **  staged_ios  )

static

Definition at line 243 of file method_worker.c.

{
    PgAioHandle **synchronous_ios = NULL;
    int         nsync = 0;
    Latch      *wakeup = NULL;
    int         worker;
 
    Assert(num_staged_ios <= PGAIO_SUBMIT_BATCH_SIZE);
 
    if (LWLockConditionalAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE))
    {
        for (int i = 0; i < num_staged_ios; ++i)
        {
            Assert(!pgaio_worker_needs_synchronous_execution(staged_ios[i]));
            if (!pgaio_worker_submission_queue_insert(staged_ios[i]))
            {
                /*
                 * Do the rest synchronously. If the queue is full, give up
                 * and do the rest synchronously. We're holding an exclusive
                 * lock on the queue so nothing can consume entries.
                 */
                synchronous_ios = &staged_ios[i];
                nsync = (num_staged_ios - i);
 
                break;
            }
 
            if (wakeup == NULL)
            {
                /* Choose an idle worker to wake up if we haven't already. */
                worker = pgaio_worker_choose_idle();
                if (worker >= 0)
                    wakeup = io_worker_control->workers[worker].latch;
 
                pgaio_debug_io(DEBUG4, staged_ios[i],
                               "choosing worker %d",
                               worker);
            }
        }
        LWLockRelease(AioWorkerSubmissionQueueLock);
    }
    else
    {
        /* do everything synchronously, no wakeup needed */
        synchronous_ios = staged_ios;
        nsync = num_staged_ios;
    }
 
    if (wakeup)
        SetLatch(wakeup);
 
    /* Run whatever is left synchronously. */
    if (nsync > 0)
    {
        for (int i = 0; i < nsync; ++i)
        {
            pgaio_io_perform_synchronously(synchronous_ios[i]);
        }
    }
}

References Assert, DEBUG4, fb(), i, io_worker_control, PgAioWorkerSlot::latch, LW_EXCLUSIVE, LWLockConditionalAcquire(), LWLockRelease(), pgaio_debug_io, pgaio_io_perform_synchronously(), PGAIO_SUBMIT_BATCH_SIZE, pgaio_worker_choose_idle(), pgaio_worker_needs_synchronous_execution(), pgaio_worker_submission_queue_insert(), SetLatch(), wakeup, and PgAioWorkerControl::workers.

Referenced by pgaio_worker_submit().

pgaio_workers_enabled()

bool pgaio_workers_enabled

(

void

)

Definition at line 600 of file method_worker.c.

{
    return io_method == IOMETHOD_WORKER;
}

References io_method, and IOMETHOD_WORKER.

Referenced by maybe_adjust_io_workers().

Variable Documentation

io_worker_control

PgAioWorkerControl* io_worker_control

static

Definition at line 99 of file method_worker.c.

Referenced by IoWorkerMain(), pgaio_worker_choose_idle(), pgaio_worker_die(), pgaio_worker_register(), pgaio_worker_shmem_init(), and pgaio_worker_submit_internal().

io_worker_queue_size

int io_worker_queue_size = 64

static

Definition at line 96 of file method_worker.c.

Referenced by pgaio_worker_queue_shmem_size().

io_worker_submission_queue

PgAioWorkerSubmissionQueue* io_worker_submission_queue

static

Definition at line 98 of file method_worker.c.

Referenced by pgaio_worker_shmem_init(), pgaio_worker_submission_queue_consume(), pgaio_worker_submission_queue_depth(), and pgaio_worker_submission_queue_insert().

io_workers

int io_workers = 3

Definition at line 93 of file method_worker.c.

Referenced by maybe_adjust_io_workers().

MyIoWorkerId

int MyIoWorkerId

static

Definition at line 97 of file method_worker.c.

Referenced by IoWorkerMain(), pgaio_worker_die(), and pgaio_worker_register().

pgaio_worker_ops

const IoMethodOps pgaio_worker_ops

Initial value:

= {
    .shmem_size = pgaio_worker_shmem_size,
    .shmem_init = pgaio_worker_shmem_init,
 
    .needs_synchronous_execution = pgaio_worker_needs_synchronous_execution,
    .submit = pgaio_worker_submit,
}

Definition at line 83 of file method_worker.c.

                                     {
    .shmem_size = pgaio_worker_shmem_size,
    .shmem_init = pgaio_worker_shmem_init,
 
    .needs_synchronous_execution = pgaio_worker_needs_synchronous_execution,
    .submit = pgaio_worker_submit,
};