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