PostgreSQL Source Code: src/backend/backup/basebackup_throttle.c Source File

PostgreSQL Source Code git master
basebackup_throttle.c
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * basebackup_throttle.c
4 * Basebackup sink implementing throttling. Data is forwarded to the
5 * next base backup sink in the chain at a rate no greater than the
6 * configured maximum.
7 *
8 * Portions Copyright (c) 2010-2025, PostgreSQL Global Development Group
9 *
10 * IDENTIFICATION
11 * src/backend/backup/basebackup_throttle.c
12 *
13 *-------------------------------------------------------------------------
14 */
15#include "postgres.h"
16
17#include "backup/basebackup_sink.h"
18#include "miscadmin.h"
19#include "pgstat.h"
20#include "storage/latch.h"
21#include "utils/timestamp.h"
22
23 typedef struct bbsink_throttle
24{
25 /* Common information for all types of sink. */
26 bbsink base;
27
28 /* The actual number of bytes, transfer of which may cause sleep. */
29 uint64 throttling_sample;
30
31 /* Amount of data already transferred but not yet throttled. */
32 int64 throttling_counter;
33
34 /* The minimum time required to transfer throttling_sample bytes. */
35 TimeOffset elapsed_min_unit;
36
37 /* The last check of the transfer rate. */
38 TimestampTz throttled_last;
39 } bbsink_throttle;
40
41static void bbsink_throttle_begin_backup(bbsink *sink);
42static void bbsink_throttle_archive_contents(bbsink *sink, size_t len);
43static void bbsink_throttle_manifest_contents(bbsink *sink, size_t len);
44static void throttle(bbsink_throttle *sink, size_t increment);
45
46 static const bbsink_ops bbsink_throttle_ops = {
47 .begin_backup = bbsink_throttle_begin_backup,
48 .begin_archive = bbsink_forward_begin_archive,
49 .archive_contents = bbsink_throttle_archive_contents,
50 .end_archive = bbsink_forward_end_archive,
51 .begin_manifest = bbsink_forward_begin_manifest,
52 .manifest_contents = bbsink_throttle_manifest_contents,
53 .end_manifest = bbsink_forward_end_manifest,
54 .end_backup = bbsink_forward_end_backup,
55 .cleanup = bbsink_forward_cleanup
56};
57
58/*
59 * How frequently to throttle, as a fraction of the specified rate-second.
60 */
61 #define THROTTLING_FREQUENCY 8
62
63/*
64 * Create a new basebackup sink that performs throttling and forwards data
65 * to a successor sink.
66 */
67bbsink *
68 bbsink_throttle_new(bbsink *next, uint32 maxrate)
69{
70 bbsink_throttle *sink;
71
72 Assert(next != NULL);
73 Assert(maxrate > 0);
74
75 sink = palloc0(sizeof(bbsink_throttle));
76 *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_throttle_ops;
77 sink->base.bbs_next = next;
78
79 sink->throttling_sample =
80 (int64) maxrate * (int64) 1024 / THROTTLING_FREQUENCY;
81
82 /*
83 * The minimum amount of time for throttling_sample bytes to be
84 * transferred.
85 */
86 sink->elapsed_min_unit = USECS_PER_SEC / THROTTLING_FREQUENCY;
87
88 return &sink->base;
89}
90
91/*
92 * There's no real work to do here, but we need to record the current time so
93 * that it can be used for future calculations.
94 */
95static void
96 bbsink_throttle_begin_backup(bbsink *sink)
97{
98 bbsink_throttle *mysink = (bbsink_throttle *) sink;
99
100 bbsink_forward_begin_backup(sink);
101
102 /* The 'real data' starts now (header was ignored). */
103 mysink->throttled_last = GetCurrentTimestamp();
104}
105
106/*
107 * First throttle, and then pass archive contents to next sink.
108 */
109static void
110 bbsink_throttle_archive_contents(bbsink *sink, size_t len)
111{
112 throttle((bbsink_throttle *) sink, len);
113
114 bbsink_forward_archive_contents(sink, len);
115}
116
117/*
118 * First throttle, and then pass manifest contents to next sink.
119 */
120static void
121 bbsink_throttle_manifest_contents(bbsink *sink, size_t len)
122{
123 throttle((bbsink_throttle *) sink, len);
124
125 bbsink_forward_manifest_contents(sink, len);
126}
127
128/*
129 * Increment the network transfer counter by the given number of bytes,
130 * and sleep if necessary to comply with the requested network transfer
131 * rate.
132 */
133static void
134 throttle(bbsink_throttle *sink, size_t increment)
135{
136 TimeOffset elapsed_min;
137
138 Assert(sink->throttling_counter >= 0);
139
140 sink->throttling_counter += increment;
141 if (sink->throttling_counter < sink->throttling_sample)
142 return;
143
144 /* How much time should have elapsed at minimum? */
145 elapsed_min = sink->elapsed_min_unit *
146 (sink->throttling_counter / sink->throttling_sample);
147
148 /*
149 * Since the latch could be set repeatedly because of concurrently WAL
150 * activity, sleep in a loop to ensure enough time has passed.
151 */
152 for (;;)
153 {
154 TimeOffset elapsed,
155 sleep;
156 int wait_result;
157
158 /* Time elapsed since the last measurement (and possible wake up). */
159 elapsed = GetCurrentTimestamp() - sink->throttled_last;
160
161 /* sleep if the transfer is faster than it should be */
162 sleep = elapsed_min - elapsed;
163 if (sleep <= 0)
164 break;
165
166 ResetLatch(MyLatch);
167
168 /* We're eating a potentially set latch, so check for interrupts */
169 CHECK_FOR_INTERRUPTS();
170
171 /*
172 * (TAR_SEND_SIZE / throttling_sample * elapsed_min_unit) should be
173 * the maximum time to sleep. Thus the cast to long is safe.
174 */
175 wait_result = WaitLatch(MyLatch,
176 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
177 (long) (sleep / 1000),
178 WAIT_EVENT_BASE_BACKUP_THROTTLE);
179
180 if (wait_result & WL_LATCH_SET)
181 CHECK_FOR_INTERRUPTS();
182
183 /* Done waiting? */
184 if (wait_result & WL_TIMEOUT)
185 break;
186 }
187
188 /*
189 * As we work with integers, only whole multiple of throttling_sample was
190 * processed. The rest will be done during the next call of this function.
191 */
192 sink->throttling_counter %= sink->throttling_sample;
193
194 /*
195 * Time interval for the remaining amount and possible next increments
196 * starts now.
197 */
198 sink->throttled_last = GetCurrentTimestamp();
199}
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1645
void bbsink_forward_begin_backup(bbsink *sink)
void bbsink_forward_begin_manifest(bbsink *sink)
void bbsink_forward_end_backup(bbsink *sink, XLogRecPtr endptr, TimeLineID endtli)
void bbsink_forward_cleanup(bbsink *sink)
void bbsink_forward_manifest_contents(bbsink *sink, size_t len)
void bbsink_forward_end_archive(bbsink *sink)
void bbsink_forward_archive_contents(bbsink *sink, size_t len)
void bbsink_forward_begin_archive(bbsink *sink, const char *archive_name)
void bbsink_forward_end_manifest(bbsink *sink)
static void bbsink_throttle_archive_contents(bbsink *sink, size_t len)
static void throttle(bbsink_throttle *sink, size_t increment)
bbsink * bbsink_throttle_new(bbsink *next, uint32 maxrate)
static void bbsink_throttle_begin_backup(bbsink *sink)
struct bbsink_throttle bbsink_throttle
#define THROTTLING_FREQUENCY
static void bbsink_throttle_manifest_contents(bbsink *sink, size_t len)
static const bbsink_ops bbsink_throttle_ops
static int32 next
Definition: blutils.c:224
int64_t int64
Definition: c.h:535
uint64_t uint64
Definition: c.h:539
uint32_t uint32
Definition: c.h:538
int64 TimestampTz
Definition: timestamp.h:39
#define USECS_PER_SEC
Definition: timestamp.h:134
int64 TimeOffset
Definition: timestamp.h:40
struct Latch * MyLatch
Definition: globals.c:63
Assert(PointerIsAligned(start, uint64))
void ResetLatch(Latch *latch)
Definition: latch.c:374
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:172
void * palloc0(Size size)
Definition: mcxt.c:1395
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
static int32 maxrate
Definition: pg_basebackup.c:147
const void size_t len
void(* begin_backup)(bbsink *sink)
TimestampTz throttled_last
bbsink base
TimeOffset elapsed_min_unit
int64 throttling_counter
uint64 throttling_sample
bbsink * bbs_next
const bbsink_ops * bbs_ops
#define WL_TIMEOUT
Definition: waiteventset.h:37
#define WL_EXIT_ON_PM_DEATH
Definition: waiteventset.h:39
#define WL_LATCH_SET
Definition: waiteventset.h:34

AltStyle によって変換されたページ (->オリジナル) /