basebackup_throttle.c File Reference

#include "postgres.h"
#include "backup/basebackup_sink.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "storage/latch.h"
#include "utils/timestamp.h"
#include "utils/wait_event.h"

Include dependency graph for basebackup_throttle.c:

Go to the source code of this file.

Data Structures
struct	bbsink_throttle

Macros
#define	THROTTLING_FREQUENCY   8

Typedefs
typedef struct bbsink_throttle	bbsink_throttle

Functions
static void	bbsink_throttle_begin_backup (bbsink *sink)
static void	bbsink_throttle_archive_contents (bbsink *sink, size_t len)
static void	bbsink_throttle_manifest_contents (bbsink *sink, size_t len)
static void	throttle (bbsink_throttle *sink, size_t increment)
bbsink *	bbsink_throttle_new (bbsink *next, uint32 maxrate)

Variables
static const bbsink_ops	bbsink_throttle_ops

Macro Definition Documentation

THROTTLING_FREQUENCY

#define THROTTLING_FREQUENCY   8

Definition at line 62 of file basebackup_throttle.c.

Typedef Documentation

bbsink_throttle

typedef struct bbsink_throttle bbsink_throttle

Function Documentation

bbsink_throttle_archive_contents()

static void bbsink_throttle_archive_contents ( bbsink *  sink, size_t  len  )

static

Definition at line 111 of file basebackup_throttle.c.

{
    throttle((bbsink_throttle *) sink, len);
 
    bbsink_forward_archive_contents(sink, len);
}

References bbsink_forward_archive_contents(), fb(), len, and throttle().

bbsink_throttle_begin_backup()

static void bbsink_throttle_begin_backup ( bbsink *  sink )

static

Definition at line 97 of file basebackup_throttle.c.

{
    bbsink_throttle *mysink = (bbsink_throttle *) sink;
 
    bbsink_forward_begin_backup(sink);
 
    /* The 'real data' starts now (header was ignored). */
    mysink->throttled_last = GetCurrentTimestamp();
}

References bbsink_forward_begin_backup(), fb(), and GetCurrentTimestamp().

bbsink_throttle_manifest_contents()

static void bbsink_throttle_manifest_contents ( bbsink *  sink, size_t  len  )

static

Definition at line 122 of file basebackup_throttle.c.

{
    throttle((bbsink_throttle *) sink, len);
 
    bbsink_forward_manifest_contents(sink, len);
}

References bbsink_forward_manifest_contents(), fb(), len, and throttle().

bbsink_throttle_new()

bbsink * bbsink_throttle_new	(	bbsink *	next,
		uint32	maxrate
	)

Definition at line 69 of file basebackup_throttle.c.

{
    bbsink_throttle *sink;
 
    Assert(next != NULL);
    Assert(maxrate > 0);
 
    sink = palloc0_object(bbsink_throttle);
    *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_throttle_ops;
    sink->base.bbs_next = next;
 
    sink->throttling_sample =
        (int64) maxrate * (int64) 1024 / THROTTLING_FREQUENCY;
 
    /*
     * The minimum amount of time for throttling_sample bytes to be
     * transferred.
     */
    sink->elapsed_min_unit = USECS_PER_SEC / THROTTLING_FREQUENCY;
 
    return &sink->base;
}

References Assert, bbsink_throttle_ops, fb(), maxrate, next, palloc0_object, THROTTLING_FREQUENCY, and USECS_PER_SEC.

Referenced by SendBaseBackup().

throttle()

static void throttle ( bbsink_throttle *  sink, size_t  increment  )

static

Definition at line 135 of file basebackup_throttle.c.

{
    TimeOffset  elapsed_min;
 
    Assert(sink->throttling_counter >= 0);
 
    sink->throttling_counter += increment;
    if (sink->throttling_counter < sink->throttling_sample)
        return;
 
    /* How much time should have elapsed at minimum? */
    elapsed_min = sink->elapsed_min_unit *
        (sink->throttling_counter / sink->throttling_sample);
 
    /*
     * Since the latch could be set repeatedly because of concurrently WAL
     * activity, sleep in a loop to ensure enough time has passed.
     */
    for (;;)
    {
        TimeOffset  elapsed,
                    sleep;
        int         wait_result;
 
        /* Time elapsed since the last measurement (and possible wake up). */
        elapsed = GetCurrentTimestamp() - sink->throttled_last;
 
        /* sleep if the transfer is faster than it should be */
        sleep = elapsed_min - elapsed;
        if (sleep <= 0)
            break;
 
        ResetLatch(MyLatch);
 
        /* We're eating a potentially set latch, so check for interrupts */
        CHECK_FOR_INTERRUPTS();
 
        /*
         * (TAR_SEND_SIZE / throttling_sample * elapsed_min_unit) should be
         * the maximum time to sleep. Thus the cast to long is safe.
         */
        wait_result = WaitLatch(MyLatch,
                                WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                                (long) (sleep / 1000),
                                WAIT_EVENT_BASE_BACKUP_THROTTLE);
 
        if (wait_result & WL_LATCH_SET)
            CHECK_FOR_INTERRUPTS();
 
        /* Done waiting? */
        if (wait_result & WL_TIMEOUT)
            break;
    }
 
    /*
     * As we work with integers, only whole multiple of throttling_sample was
     * processed. The rest will be done during the next call of this function.
     */
    sink->throttling_counter %= sink->throttling_sample;
 
    /*
     * Time interval for the remaining amount and possible next increments
     * starts now.
     */
    sink->throttled_last = GetCurrentTimestamp();
}

References Assert, CHECK_FOR_INTERRUPTS, fb(), GetCurrentTimestamp(), MyLatch, ResetLatch(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by bbsink_throttle_archive_contents(), and bbsink_throttle_manifest_contents().

Variable Documentation

bbsink_throttle_ops

const bbsink_ops bbsink_throttle_ops

static

Initial value:

= {
    .begin_backup = bbsink_throttle_begin_backup,
    .begin_archive = bbsink_forward_begin_archive,
    .archive_contents = bbsink_throttle_archive_contents,
    .end_archive = bbsink_forward_end_archive,
    .begin_manifest = bbsink_forward_begin_manifest,
    .manifest_contents = bbsink_throttle_manifest_contents,
    .end_manifest = bbsink_forward_end_manifest,
    .end_backup = bbsink_forward_end_backup,
    .cleanup = bbsink_forward_cleanup
}

Definition at line 47 of file basebackup_throttle.c.

                                              {
    .begin_backup = bbsink_throttle_begin_backup,
    .begin_archive = bbsink_forward_begin_archive,
    .archive_contents = bbsink_throttle_archive_contents,
    .end_archive = bbsink_forward_end_archive,
    .begin_manifest = bbsink_forward_begin_manifest,
    .manifest_contents = bbsink_throttle_manifest_contents,
    .end_manifest = bbsink_forward_end_manifest,
    .end_backup = bbsink_forward_end_backup,
    .cleanup = bbsink_forward_cleanup
};

Referenced by bbsink_throttle_new().