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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
|
/*-
* Copyright (c) 2014-2018 MongoDB, Inc.
* Copyright (c) 2008-2014 WiredTiger, Inc.
* All rights reserved.
*
* See the file LICENSE for redistribution information.
*/
#include "wt_internal.h"
static int __ckpt_server_start(WT_CONNECTION_IMPL *);
/*
* __ckpt_server_config --
* Parse and setup the checkpoint server options.
*/
static int
__ckpt_server_config(WT_SESSION_IMPL *session, const char **cfg, bool *startp)
{
WT_CONFIG_ITEM cval;
WT_CONNECTION_IMPL *conn;
*startp = false;
conn = S2C(session);
WT_RET(__wt_config_gets(session, cfg, "checkpoint.wait", &cval));
conn->ckpt_usecs = (uint64_t)cval.val * WT_MILLION;
WT_RET(__wt_config_gets(session, cfg, "checkpoint.log_size", &cval));
conn->ckpt_logsize = (wt_off_t)cval.val;
/*
* The checkpoint configuration requires a wait time and/or a log size,
* if neither is set, we're not running at all. Checkpoints based on log
* size also require logging be enabled.
*/
if (conn->ckpt_usecs != 0 ||
(conn->ckpt_logsize != 0 &&
FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))) {
/*
* If checkpointing based on log data, use a minimum of the
* log file size. The logging subsystem has already been
* initialized.
*/
if (conn->ckpt_logsize != 0 &&
FLD_ISSET(conn->log_flags, WT_CONN_LOG_ENABLED))
conn->ckpt_logsize = WT_MAX(
conn->ckpt_logsize, conn->log_file_max);
/* Checkpoints are incompatible with in-memory configuration */
WT_RET(__wt_config_gets(session, cfg, "in_memory", &cval));
if (cval.val != 0)
WT_RET_MSG(session, EINVAL,
"checkpoint configuration incompatible with "
"in-memory configuration");
__wt_log_written_reset(session);
*startp = true;
}
return (0);
}
/*
* __ckpt_server_run_chk --
* Check to decide if the checkpoint server should continue running.
*/
static bool
__ckpt_server_run_chk(WT_SESSION_IMPL *session)
{
return (F_ISSET(S2C(session), WT_CONN_SERVER_CHECKPOINT));
}
/*
* __ckpt_server --
* The checkpoint server thread.
*/
static WT_THREAD_RET
__ckpt_server(void *arg)
{
WT_CONNECTION_IMPL *conn;
WT_DECL_RET;
WT_SESSION *wt_session;
WT_SESSION_IMPL *session;
uint64_t checkpoint_gen;
session = arg;
conn = S2C(session);
wt_session = (WT_SESSION *)session;
for (;;) {
/*
* Wait...
* NOTE: If the user only configured logsize, then usecs
* will be 0 and this wait won't return until signalled.
*/
__wt_cond_wait(session,
conn->ckpt_cond, conn->ckpt_usecs, __ckpt_server_run_chk);
/* Check if we're quitting or being reconfigured. */
if (!__ckpt_server_run_chk(session))
break;
checkpoint_gen = __wt_gen(session, WT_GEN_CHECKPOINT);
WT_ERR(wt_session->checkpoint(wt_session, NULL));
/*
* Reset the log file size counters if the checkpoint wasn't
* skipped.
*/
if (checkpoint_gen != __wt_gen(session, WT_GEN_CHECKPOINT) &&
conn->ckpt_logsize) {
__wt_log_written_reset(session);
conn->ckpt_signalled = false;
/*
* In case we crossed the log limit during the
* checkpoint and the condition variable was
* already signalled, do a tiny wait to clear
* it so we don't do another checkpoint
* immediately.
*/
__wt_cond_wait(session, conn->ckpt_cond, 1, NULL);
}
}
if (0) {
err: WT_PANIC_MSG(session, ret, "checkpoint server error");
}
return (WT_THREAD_RET_VALUE);
}
/*
* __ckpt_server_start --
* Start the checkpoint server thread.
*/
static int
__ckpt_server_start(WT_CONNECTION_IMPL *conn)
{
WT_SESSION_IMPL *session;
uint32_t session_flags;
/* Nothing to do if the server is already running. */
if (conn->ckpt_session != NULL)
return (0);
F_SET(conn, WT_CONN_SERVER_CHECKPOINT);
/*
* The checkpoint server gets its own session.
*
* Checkpoint does enough I/O it may be called upon to perform slow
* operations for the block manager.
*/
session_flags = WT_SESSION_CAN_WAIT;
WT_RET(__wt_open_internal_session(conn,
"checkpoint-server", true, session_flags, &conn->ckpt_session));
session = conn->ckpt_session;
WT_RET(__wt_cond_alloc(session, "checkpoint server", &conn->ckpt_cond));
/*
* Start the thread.
*/
WT_RET(__wt_thread_create(
session, &conn->ckpt_tid, __ckpt_server, session));
conn->ckpt_tid_set = true;
return (0);
}
/*
* __wt_checkpoint_server_create --
* Configure and start the checkpoint server.
*/
int
__wt_checkpoint_server_create(WT_SESSION_IMPL *session, const char *cfg[])
{
WT_CONNECTION_IMPL *conn;
bool start;
conn = S2C(session);
start = false;
/*
* Stop any server that is already running. This means that each time
* reconfigure is called we'll bounce the server even if there are no
* configuration changes. This makes our life easier as the underlying
* configuration routine doesn't have to worry about freeing objects
* in the connection structure (it's guaranteed to always start with a
* blank slate), and we don't have to worry about races where a running
* server is reading configuration information that we're updating, and
* it's not expected that reconfiguration will happen a lot.
*/
if (conn->ckpt_session != NULL)
WT_RET(__wt_checkpoint_server_destroy(session));
WT_RET(__ckpt_server_config(session, cfg, &start));
if (start)
WT_RET(__ckpt_server_start(conn));
return (0);
}
/*
* __wt_checkpoint_server_destroy --
* Destroy the checkpoint server thread.
*/
int
__wt_checkpoint_server_destroy(WT_SESSION_IMPL *session)
{
WT_CONNECTION_IMPL *conn;
WT_DECL_RET;
WT_SESSION *wt_session;
conn = S2C(session);
F_CLR(conn, WT_CONN_SERVER_CHECKPOINT);
if (conn->ckpt_tid_set) {
__wt_cond_signal(session, conn->ckpt_cond);
WT_TRET(__wt_thread_join(session, &conn->ckpt_tid));
conn->ckpt_tid_set = false;
}
__wt_cond_destroy(session, &conn->ckpt_cond);
/* Close the server thread's session. */
if (conn->ckpt_session != NULL) {
wt_session = &conn->ckpt_session->iface;
WT_TRET(wt_session->close(wt_session, NULL));
}
/*
* Ensure checkpoint settings are cleared - so that reconfigure doesn't
* get confused.
*/
conn->ckpt_session = NULL;
conn->ckpt_tid_set = false;
conn->ckpt_cond = NULL;
conn->ckpt_usecs = 0;
return (ret);
}
/*
* __wt_checkpoint_signal --
* Signal the checkpoint thread if sufficient log has been written.
*/
void
__wt_checkpoint_signal(WT_SESSION_IMPL *session, wt_off_t logsize)
{
WT_CONNECTION_IMPL *conn;
conn = S2C(session);
WT_ASSERT(session, WT_CKPT_LOGSIZE(conn));
if (logsize >= conn->ckpt_logsize && !conn->ckpt_signalled) {
__wt_cond_signal(session, conn->ckpt_cond);
conn->ckpt_signalled = true;
}
}
|