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)

Variables
PGDLLIMPORT int	io_workers

Function Documentation

IoWorkerMain()

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

Definition at line 384 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];
 
    MyBackendType = B_IO_WORKER;
    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. */
        LWLockAcquire(AioWorkerSubmissionQueueLock, LW_EXCLUSIVE);
        if ((io_index = pgaio_worker_submission_queue_consume()) == UINT32_MAX)
        {
            /*
             * 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_choose_idle_worker()) < 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 != UINT32_MAX)
        {
            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();
    }
 
    error_context_stack = errcallback.previous;
    proc_exit(0);
}

References ErrorContextCallback::arg, Assert(), AuxiliaryProcessMainCommon(), B_IO_WORKER, ErrorContextCallback::callback, CHECK_FOR_INTERRUPTS, DEBUG4, die, EmitErrorReport(), END_CRIT_SECTION, error_context_stack, HOLD_INTERRUPTS, i, AioWorkerControl::idle_worker_mask, INJECTION_POINT, PgAioCtl::io_handles, io_worker_control, IO_WORKER_WAKEUP_FANOUT, AioWorkerSlot::latch, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), LWLockReleaseAll(), Min, MyBackendType, MyIoWorkerId, MyLatch, PG_exception_stack, pgaio_choose_idle_worker(), pgaio_ctl, pgaio_debug_io, pgaio_io_get_iovec_length(), pgaio_io_perform_synchronously(), pgaio_io_process_completion(), pgaio_io_reopen(), pgaio_worker_error_callback(), pgaio_worker_register(), pgaio_worker_submission_queue_consume(), pgaio_worker_submission_queue_depth(), pqsignal, ErrorContextCallback::previous, proc_exit(), 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 AioWorkerControl::workers.

Variable Documentation

io_workers

PGDLLIMPORT int io_workers

extern

Definition at line 94 of file method_worker.c.

Referenced by maybe_adjust_io_workers().