1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
|
/*--------------------------------------------------------------------------
*
* xid_wraparound.c
* Utilities for testing XID wraparound
*
*
* Portions Copyright (c) 1996-2024, PostgreSQL Global Development Group
* Portions Copyright (c) 1994, Regents of the University of California
*
* IDENTIFICATION
* src/test/modules/xid_wraparound/xid_wraparound.c
*
* -------------------------------------------------------------------------
*/
#include "postgres.h"
#include "access/xact.h"
#include "miscadmin.h"
#include "storage/proc.h"
#include "utils/xid8.h"
PG_MODULE_MAGIC;
static int64 consume_xids_shortcut(void);
static FullTransactionId consume_xids_common(FullTransactionId untilxid, uint64 nxids);
/*
* Consume the specified number of XIDs.
*/
PG_FUNCTION_INFO_V1(consume_xids);
Datum
consume_xids(PG_FUNCTION_ARGS)
{
int64 nxids = PG_GETARG_INT64(0);
FullTransactionId lastxid;
if (nxids < 0)
elog(ERROR, "invalid nxids argument: %lld", (long long) nxids);
if (nxids == 0)
lastxid = ReadNextFullTransactionId();
else
lastxid = consume_xids_common(InvalidFullTransactionId, (uint64) nxids);
PG_RETURN_FULLTRANSACTIONID(lastxid);
}
/*
* Consume XIDs, up to the given XID.
*/
PG_FUNCTION_INFO_V1(consume_xids_until);
Datum
consume_xids_until(PG_FUNCTION_ARGS)
{
FullTransactionId targetxid = PG_GETARG_FULLTRANSACTIONID(0);
FullTransactionId lastxid;
if (!FullTransactionIdIsNormal(targetxid))
elog(ERROR, "targetxid %llu is not normal",
(unsigned long long) U64FromFullTransactionId(targetxid));
lastxid = consume_xids_common(targetxid, 0);
PG_RETURN_FULLTRANSACTIONID(lastxid);
}
/*
* Common functionality between the two public functions.
*/
static FullTransactionId
consume_xids_common(FullTransactionId untilxid, uint64 nxids)
{
FullTransactionId lastxid;
uint64 last_reported_at = 0;
uint64 consumed = 0;
/* Print a NOTICE every REPORT_INTERVAL xids */
#define REPORT_INTERVAL (10 * 1000000)
/* initialize 'lastxid' with the system's current next XID */
lastxid = ReadNextFullTransactionId();
/*
* We consume XIDs by calling GetNewTransactionId(true), which marks the
* consumed XIDs as subtransactions of the current top-level transaction.
* For that to work, this transaction must have a top-level XID.
*
* GetNewTransactionId registers them in the subxid cache in PGPROC, until
* the cache overflows, but beyond that, we don't keep track of the
* consumed XIDs.
*/
(void) GetTopTransactionId();
for (;;)
{
uint64 xids_left;
CHECK_FOR_INTERRUPTS();
/* How many XIDs do we have left to consume? */
if (nxids > 0)
{
if (consumed >= nxids)
break;
xids_left = nxids - consumed;
}
else
{
if (FullTransactionIdFollowsOrEquals(lastxid, untilxid))
break;
xids_left = U64FromFullTransactionId(untilxid) - U64FromFullTransactionId(lastxid);
}
/*
* If we still have plenty of XIDs to consume, try to take a shortcut
* and bump up the nextXid counter directly.
*/
if (xids_left > 2000 &&
consumed - last_reported_at < REPORT_INTERVAL &&
MyProc->subxidStatus.overflowed)
{
int64 consumed_by_shortcut = consume_xids_shortcut();
if (consumed_by_shortcut > 0)
{
consumed += consumed_by_shortcut;
continue;
}
}
/* Slow path: Call GetNewTransactionId to allocate a new XID. */
lastxid = GetNewTransactionId(true);
consumed++;
/* Report progress */
if (consumed - last_reported_at >= REPORT_INTERVAL)
{
if (nxids > 0)
elog(NOTICE, "consumed %llu / %llu XIDs, latest %u:%u",
(unsigned long long) consumed, (unsigned long long) nxids,
EpochFromFullTransactionId(lastxid),
XidFromFullTransactionId(lastxid));
else
elog(NOTICE, "consumed up to %u:%u / %u:%u",
EpochFromFullTransactionId(lastxid),
XidFromFullTransactionId(lastxid),
EpochFromFullTransactionId(untilxid),
XidFromFullTransactionId(untilxid));
last_reported_at = consumed;
}
}
return lastxid;
}
/*
* These constants copied from .c files, because they're private.
*/
#define COMMIT_TS_XACTS_PER_PAGE (BLCKSZ / 10)
#define SUBTRANS_XACTS_PER_PAGE (BLCKSZ / sizeof(TransactionId))
#define CLOG_XACTS_PER_BYTE 4
#define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
/*
* All the interesting action in GetNewTransactionId happens when we extend
* the SLRUs, or at the uint32 wraparound. If the nextXid counter is not close
* to any of those interesting values, take a shortcut and bump nextXID
* directly, close to the next "interesting" value.
*/
static inline uint32
XidSkip(FullTransactionId fullxid)
{
uint32 low = XidFromFullTransactionId(fullxid);
uint32 rem;
uint32 distance;
if (low < 5 || low >= UINT32_MAX - 5)
return 0;
distance = UINT32_MAX - 5 - low;
rem = low % COMMIT_TS_XACTS_PER_PAGE;
if (rem == 0)
return 0;
distance = Min(distance, COMMIT_TS_XACTS_PER_PAGE - rem);
rem = low % SUBTRANS_XACTS_PER_PAGE;
if (rem == 0)
return 0;
distance = Min(distance, SUBTRANS_XACTS_PER_PAGE - rem);
rem = low % CLOG_XACTS_PER_PAGE;
if (rem == 0)
return 0;
distance = Min(distance, CLOG_XACTS_PER_PAGE - rem);
return distance;
}
static int64
consume_xids_shortcut(void)
{
FullTransactionId nextXid;
uint32 consumed;
LWLockAcquire(XidGenLock, LW_EXCLUSIVE);
nextXid = TransamVariables->nextXid;
/*
* Go slow near the "interesting values". The interesting zones include 5
* transactions before and after SLRU page switches.
*/
consumed = XidSkip(nextXid);
if (consumed > 0)
TransamVariables->nextXid.value += (uint64) consumed;
LWLockRelease(XidGenLock);
return consumed;
}
|
