io_worker.h File Reference

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
pg_noreturn void	IoWorkerMain (const void *startup_data, size_t startup_data_len)
bool	pgaio_worker_pm_test_grow_signal_sent (void)
void	pgaio_worker_pm_clear_grow_signal_sent (void)
bool	pgaio_worker_pm_test_grow (void)

Variables
PGDLLIMPORT int	io_min_workers
PGDLLIMPORT int	io_max_workers
PGDLLIMPORT int	io_worker_idle_timeout
PGDLLIMPORT int	io_worker_launch_interval

Function Documentation

IoWorkerMain()

pg_noreturn void IoWorkerMain ( const void *  startup_data, size_t  startup_data_len  )

extern

Definition at line 666 of file method_worker.c.

{
    sigjmp_buf  local_sigjmp_buf;
    TimestampTz idle_timeout_abs = 0;
    int         timeout_guc_used = 0;
    PgAioHandle *volatile error_ioh = NULL;
    ErrorContextCallback errcallback = {0};
    volatile int error_errno = 0;
    char        cmd[128];
    int         hist_ios = 0;
    int         hist_wakeups = 0;
 
    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;
        int         worker = -1;
        int         queue_depth = 0;
        bool        maybe_grow = false;
 
        /*
         * 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. */
            pgaio_workerset_insert(&io_worker_control->idle_workerset,
                                   MyIoWorkerId);
        }
        else
        {
            /* Got one.  Clear idle flag. */
            pgaio_workerset_remove(&io_worker_control->idle_workerset,
                                   MyIoWorkerId);
 
            /*
             * See if we should wake up a higher numbered peer.  Only do that
             * if this worker is not receiving spurious wakeups itself.  The
             * intention is create a frontier beyond which idle workers stay
             * asleep.
             *
             * This heuristic tries to discover the useful wakeup propagation
             * chain length when IOs are very fast and workers wake up to find
             * that all IOs have already been taken.
             *
             * If we chose not to wake a worker when we ideally should have,
             * then the ratio will soon change to correct that.
             */
            if (hist_wakeups <= hist_ios)
            {
                queue_depth = pgaio_worker_submission_queue_depth();
                if (queue_depth > 0)
                {
                    /* Choose a worker higher than me to wake. */
                    worker = pgaio_worker_choose_idle(MyIoWorkerId);
                    if (worker == -1)
                        maybe_grow = true;
                }
            }
        }
        LWLockRelease(AioWorkerSubmissionQueueLock);
 
        /* Propagate wakeups. */
        if (worker != -1)
        {
            pgaio_worker_wake(worker);
        }
        else if (maybe_grow)
        {
            /*
             * We know there was at least one more item in the queue, and we
             * failed to find a higher-numbered idle worker to wake.  Now we
             * decide if we should try to start one more worker.
             *
             * We do this with a simple heuristic: is the queue depth greater
             * than the current number of workers?
             *
             * Consider the following situations:
             *
             * 1. The queue depth is constantly increasing, because IOs are
             * arriving faster than they can possibly be serviced.  It doesn't
             * matter much which threshold we choose, as we will surely hit
             * it.  Crossing the current worker count is a useful signal
             * because it's clearly too deep to avoid queuing latency already,
             * but still leaves a small window of opportunity to improve the
             * situation before the queue oveflows.
             *
             * 2. The worker pool is keeping up, no latency is being
             * introduced and an extra worker would be a waste of resources.
             * Queue depth distributions tend to be heavily skewed, with long
             * tails of low probability spikes (due to submission clustering,
             * scheduling, jitter, stalls, noisy neighbors, etc).  We want a
             * number that is very unlikely to be triggered by an outlier, and
             * we bet that an exponential or similar distribution whose
             * outliers never reach this threshold must be almost entirely
             * concentrated at the low end.  If we do see a spike as big as
             * the worker count, we take it as a signal that the distribution
             * is surely too wide.
             *
             * On its own, this is an extremely crude signal.  When combined
             * with the wakeup propagation test that precedes it (but on its
             * own tends to overshoot) and io_worker_launch_delay, the result
             * is that we gradually test each pool size until we find one that
             * doesn't trigger further expansion, and then hold it for at
             * least io_worker_idle_timeout.
             *
             * XXX Perhaps ideas from queueing theory or control theory could
             * do a better job of this.
             */
 
            /* Read nworkers without lock for this heuristic purpose. */
            if (queue_depth > io_worker_control->nworkers)
                pgaio_worker_request_grow();
        }
 
        if (io_index != -1)
        {
            PgAioHandle *ioh = NULL;
 
            /* Cancel timeout and update wakeup:work ratio. */
            idle_timeout_abs = 0;
            if (++hist_ios == PGAIO_WORKER_WAKEUP_RATIO_SATURATE)
            {
                hist_wakeups /= 2;
                hist_ios /= 2;
            }
 
            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
 
#ifdef PGAIO_WORKER_SHOW_PS_INFO
            {
                char       *description = pgaio_io_get_target_description(ioh);
 
                sprintf(cmd, "%d: [%s] %s",
                        MyIoWorkerId,
                        pgaio_io_get_op_name(ioh),
                        description);
                pfree(description);
                set_ps_display(cmd);
            }
#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
        {
            int         timeout_ms;
 
            /* Cancel new worker request if pending. */
            pgaio_worker_cancel_grow();
 
            /* Compute the remaining allowed idle time. */
            if (io_worker_idle_timeout == -1)
            {
                /* Never time out. */
                timeout_ms = -1;
            }
            else
            {
                TimestampTz now = GetCurrentTimestamp();
 
                /* If the GUC changes, reset timer. */
                if (idle_timeout_abs != 0 &&
                    io_worker_idle_timeout != timeout_guc_used)
                    idle_timeout_abs = 0;
 
                /* Only the highest-numbered worker can time out. */
                if (pgaio_worker_can_timeout())
                {
                    if (idle_timeout_abs == 0)
                    {
                        /*
                         * I have just been promoted to the timeout worker, or
                         * the GUC changed.  Compute new absolute time from
                         * now.
                         */
                        idle_timeout_abs =
                            TimestampTzPlusMilliseconds(now,
                                                        io_worker_idle_timeout);
                        timeout_guc_used = io_worker_idle_timeout;
                    }
                    timeout_ms =
                        TimestampDifferenceMilliseconds(now, idle_timeout_abs);
                }
                else
                {
                    /* No timeout for me. */
                    idle_timeout_abs = 0;
                    timeout_ms = -1;
                }
            }
 
#ifdef PGAIO_WORKER_SHOW_PS_INFO
            sprintf(cmd, "%d: idle, wakeups:ios = %d:%d",
                    MyIoWorkerId, hist_wakeups, hist_ios);
            set_ps_display(cmd);
#endif
 
            if (WaitLatch(MyLatch, WL_LATCH_SET | WL_EXIT_ON_PM_DEATH | WL_TIMEOUT,
                          timeout_ms,
                          WAIT_EVENT_IO_WORKER_MAIN) == WL_TIMEOUT)
            {
                /* WL_TIMEOUT */
                if (pgaio_worker_can_timeout())
                    if (GetCurrentTimestamp() >= idle_timeout_abs)
                        break;
            }
            else
            {
                /* WL_LATCH_SET */
                if (++hist_wakeups == PGAIO_WORKER_WAKEUP_RATIO_SATURATE)
                {
                    hist_wakeups /= 2;
                    hist_ios /= 2;
                }
            }
            ResetLatch(MyLatch);
        }
 
        CHECK_FOR_INTERRUPTS();
 
        if (ConfigReloadPending)
        {
            ConfigReloadPending = false;
            ProcessConfigFile(PGC_SIGHUP);
 
            /* If io_max_workers has been decreased, exit highest first. */
            if (MyIoWorkerId >= io_max_workers)
                break;
        }
    }
 
    error_context_stack = errcallback.previous;
    proc_exit(0);
}

References ErrorContextCallback::arg, Assert, AuxiliaryProcessMainCommon(), ErrorContextCallback::callback, CHECK_FOR_INTERRUPTS, ConfigReloadPending, DEBUG4, description, die, EmitErrorReport(), END_CRIT_SECTION, error_context_stack, fb(), GetCurrentTimestamp(), HOLD_INTERRUPTS, i, PgAioWorkerControl::idle_workerset, INJECTION_POINT, PgAioCtl::io_handles, io_max_workers, io_worker_control, io_worker_idle_timeout, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LWLockReleaseAll(), MyIoWorkerId, MyLatch, now(), PgAioWorkerControl::nworkers, pfree(), PG_exception_stack, pgaio_ctl, pgaio_debug_io, pgaio_io_get_iovec_length(), pgaio_io_get_op_name(), pgaio_io_get_target_description(), pgaio_io_perform_synchronously(), pgaio_io_process_completion(), pgaio_io_reopen(), pgaio_worker_can_timeout(), pgaio_worker_cancel_grow(), pgaio_worker_choose_idle(), pgaio_worker_error_callback(), pgaio_worker_register(), pgaio_worker_request_grow(), pgaio_worker_submission_queue_consume(), pgaio_worker_submission_queue_depth(), pgaio_worker_wake(), PGAIO_WORKER_WAKEUP_RATIO_SATURATE, pgaio_workerset_insert(), pgaio_workerset_remove(), PGC_SIGHUP, pqsignal, ErrorContextCallback::previous, proc_exit(), ProcessConfigFile(), procsignal_sigusr1_handler(), ResetLatch(), RESUME_INTERRUPTS, set_ps_display(), ShutdownRequestPending, SIGALRM, SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SIGPIPE, SIGUSR1, SIGUSR2, sprintf, START_CRIT_SECTION, TimestampDifferenceMilliseconds(), TimestampTzPlusMilliseconds, UnBlockSig, VALGRIND_MAKE_MEM_UNDEFINED, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

pgaio_worker_pm_clear_grow_signal_sent()

void pgaio_worker_pm_clear_grow_signal_sent ( void  )

extern

Definition at line 349 of file method_worker.c.

{
    if (!io_worker_control)
        return;
 
    io_worker_control->grow = false;
    io_worker_control->grow_signal_sent = false;
    pg_memory_barrier();
}

References PgAioWorkerControl::grow, PgAioWorkerControl::grow_signal_sent, io_worker_control, and pg_memory_barrier.

Referenced by maybe_start_io_workers().

pgaio_worker_pm_test_grow()

bool pgaio_worker_pm_test_grow ( void  )

extern

Definition at line 339 of file method_worker.c.

{
    pg_memory_barrier();
    return io_worker_control && io_worker_control->grow;
}

References PgAioWorkerControl::grow, io_worker_control, and pg_memory_barrier.

Referenced by maybe_start_io_workers().

pgaio_worker_pm_test_grow_signal_sent()

bool pgaio_worker_pm_test_grow_signal_sent ( void  )

extern

Definition at line 328 of file method_worker.c.

{
    pg_memory_barrier();
    return io_worker_control && io_worker_control->grow_signal_sent;
}

References PgAioWorkerControl::grow_signal_sent, io_worker_control, and pg_memory_barrier.

Referenced by maybe_start_io_workers_scheduled_at().

Variable Documentation

io_max_workers

PGDLLIMPORT int io_max_workers

extern

Definition at line 132 of file method_worker.c.

Referenced by IoWorkerMain(), maybe_start_io_workers_scheduled_at(), and pgaio_worker_request_grow().

io_min_workers

PGDLLIMPORT int io_min_workers

extern

Definition at line 131 of file method_worker.c.

Referenced by maybe_start_io_workers(), maybe_start_io_workers_scheduled_at(), and pgaio_worker_can_timeout().

io_worker_idle_timeout

PGDLLIMPORT int io_worker_idle_timeout

extern

Definition at line 133 of file method_worker.c.

Referenced by IoWorkerMain().

io_worker_launch_interval

PGDLLIMPORT int io_worker_launch_interval

extern

Definition at line 134 of file method_worker.c.

Referenced by maybe_start_io_workers().