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
|
/* ----------
* backend_progress.c
*
* Command progress reporting infrastructure.
*
* Copyright (c) 2001-2025, PostgreSQL Global Development Group
*
* src/backend/utils/activity/backend_progress.c
* ----------
*/
#include "postgres.h"
#include "access/parallel.h"
#include "libpq/pqformat.h"
#include "utils/backend_progress.h"
#include "utils/backend_status.h"
/*-----------
* pgstat_progress_start_command() -
*
* Set st_progress_command (and st_progress_command_target) in own backend
* entry. Also, zero-initialize st_progress_param array.
*-----------
*/
void
pgstat_progress_start_command(ProgressCommandType cmdtype, Oid relid)
{
volatile PgBackendStatus *beentry = MyBEEntry;
if (!beentry || !pgstat_track_activities)
return;
PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
beentry->st_progress_command = cmdtype;
beentry->st_progress_command_target = relid;
MemSet(&beentry->st_progress_param, 0, sizeof(beentry->st_progress_param));
PGSTAT_END_WRITE_ACTIVITY(beentry);
}
/*-----------
* pgstat_progress_update_param() -
*
* Update index'th member in st_progress_param[] of own backend entry.
*-----------
*/
void
pgstat_progress_update_param(int index, int64 val)
{
volatile PgBackendStatus *beentry = MyBEEntry;
Assert(index >= 0 && index < PGSTAT_NUM_PROGRESS_PARAM);
if (!beentry || !pgstat_track_activities)
return;
PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
beentry->st_progress_param[index] = val;
PGSTAT_END_WRITE_ACTIVITY(beentry);
}
/*-----------
* pgstat_progress_incr_param() -
*
* Increment index'th member in st_progress_param[] of own backend entry.
*-----------
*/
void
pgstat_progress_incr_param(int index, int64 incr)
{
volatile PgBackendStatus *beentry = MyBEEntry;
Assert(index >= 0 && index < PGSTAT_NUM_PROGRESS_PARAM);
if (!beentry || !pgstat_track_activities)
return;
PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
beentry->st_progress_param[index] += incr;
PGSTAT_END_WRITE_ACTIVITY(beentry);
}
/*-----------
* pgstat_progress_parallel_incr_param() -
*
* A variant of pgstat_progress_incr_param to allow a worker to poke at
* a leader to do an incremental progress update.
*-----------
*/
void
pgstat_progress_parallel_incr_param(int index, int64 incr)
{
/*
* Parallel workers notify a leader through a PqMsg_Progress message to
* update progress, passing the progress index and incremented value.
* Leaders can just call pgstat_progress_incr_param directly.
*/
if (IsParallelWorker())
{
static StringInfoData progress_message;
initStringInfo(&progress_message);
pq_beginmessage(&progress_message, PqMsg_Progress);
pq_sendint32(&progress_message, index);
pq_sendint64(&progress_message, incr);
pq_endmessage(&progress_message);
}
else
pgstat_progress_incr_param(index, incr);
}
/*-----------
* pgstat_progress_update_multi_param() -
*
* Update multiple members in st_progress_param[] of own backend entry.
* This is atomic; readers won't see intermediate states.
*-----------
*/
void
pgstat_progress_update_multi_param(int nparam, const int *index,
const int64 *val)
{
volatile PgBackendStatus *beentry = MyBEEntry;
int i;
if (!beentry || !pgstat_track_activities || nparam == 0)
return;
PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
for (i = 0; i < nparam; ++i)
{
Assert(index[i] >= 0 && index[i] < PGSTAT_NUM_PROGRESS_PARAM);
beentry->st_progress_param[index[i]] = val[i];
}
PGSTAT_END_WRITE_ACTIVITY(beentry);
}
/*-----------
* pgstat_progress_end_command() -
*
* Reset st_progress_command (and st_progress_command_target) in own backend
* entry. This signals the end of the command.
*-----------
*/
void
pgstat_progress_end_command(void)
{
volatile PgBackendStatus *beentry = MyBEEntry;
if (!beentry || !pgstat_track_activities)
return;
if (beentry->st_progress_command == PROGRESS_COMMAND_INVALID)
return;
PGSTAT_BEGIN_WRITE_ACTIVITY(beentry);
beentry->st_progress_command = PROGRESS_COMMAND_INVALID;
beentry->st_progress_command_target = InvalidOid;
PGSTAT_END_WRITE_ACTIVITY(beentry);
}
|
