Clean up prepared statements on thread exit (#20211)
* Clean pending sql prepared statements on thread exit Do not use pthread keys * Revert thread changes * Fix stack-buffer-overflow * Finalize statements before closing the databases
Stelios Fragkakis committed
May 2, 2025 at 20:15 UTC
b641ae720cf962bd6fcade35ad7cc010005b666f
4 files changed
+90
-45
src/database/sqlite/sqlite_functions.c
+84
-45
@@ -2,8 +2,19 @@
2
3
#include "sqlite_functions.h"
4
5
-#define MAX_PREPARED_STATEMENTS (32)
6
-pthread_key_t key_pool[MAX_PREPARED_STATEMENTS];
5
+#define MAX_PREPARED_THREAD_STATEMENTS (32)
6
+
7
+SPINLOCK JudyL_thread_stmt_lock = SPINLOCK_INITIALIZER;
8
+Pvoid_t JudyL_thread_stmt_pool = NULL;
9
+
10
+struct stmt_pool_s {
11
+ int count;
12
+ pid_t thread_id;
13
+ char *name;
14
+ void *stmt[MAX_PREPARED_THREAD_STATEMENTS];
15
+};
16
+
17
+__thread struct stmt_pool_s *thread_stmt_pool = NULL;
18
19
long long def_journal_size_limit = 16777216;
20
@@ -157,64 +168,88 @@ int configure_sqlite_database(sqlite3 *database, int target_version, const char
168
return 0;
169
}
170
160
-#define MAX_OPEN_STATEMENTS (512)
161
-
162
-static void add_stmt_to_list(sqlite3_stmt *res)
171
+static void finalize_and_free_stmt_list(struct stmt_pool_s *stmt_list)
172
{
164
- static int idx = 0;
165
- static sqlite3_stmt *statements[MAX_OPEN_STATEMENTS];
166
-
167
- if (unlikely(!res)) {
168
- if (idx)
169
- netdata_log_info("Finilizing %d statements", idx);
170
- else
171
- netdata_log_info("No statements pending to finalize");
172
- while (idx > 0) {
173
- int rc;
174
- rc = sqlite3_finalize(statements[--idx]);
175
- if (unlikely(rc != SQLITE_OK))
176
- error_report("Failed to finalize statement during shutdown, rc = %d", rc);
177
- }
173
+ if (!stmt_list)
174
return;
175
+
176
+ int max_keys = stmt_list->count;
177
+ for (int i = 0; i < max_keys; i++) {
178
+ int rc = sqlite3_finalize((sqlite3_stmt *)stmt_list->stmt[i]);
179
+ if (unlikely(rc != SQLITE_OK))
180
+ error_report("Failed to finalize statement, rc = %d", rc);
181
}
182
+ freez(stmt_list->name);
183
+ freez(stmt_list);
184
+}
185
181
- if (unlikely(idx == MAX_OPEN_STATEMENTS))
186
+// This must be called when the thread terminates
187
+void finalize_self_prepared_sql_statements()
188
+{
189
+ if (!thread_stmt_pool)
190
return;
191
+
192
+ Word_t thread_id = thread_stmt_pool->thread_id;
193
+ finalize_and_free_stmt_list(thread_stmt_pool);
194
+ thread_stmt_pool = NULL;
195
+ spinlock_lock(&JudyL_thread_stmt_lock);
196
+ (void) JudyLDel(&JudyL_thread_stmt_pool, thread_id, PJE0);
197
+ spinlock_unlock(&JudyL_thread_stmt_lock);
198
}
199
185
-static void release_statement(void *statement)
200
+void finalize_all_prepared_sql_statements()
201
{
187
- int rc;
188
- spinlock_lock(&sqlite_spinlock);
189
- if (sqlite_online) {
190
- if (unlikely(rc = sqlite3_finalize((sqlite3_stmt *)statement) != SQLITE_OK))
191
- error_report("Failed to finalize statement, rc = %d", rc);
202
+ spinlock_lock(&JudyL_thread_stmt_lock);
203
+ bool first_then_next = true;
204
+ Pvoid_t *Pvalue = NULL;
205
+ Word_t thread_id = 0;
206
+ if (JudyL_thread_stmt_pool) {
207
+ while ((Pvalue = JudyLFirstThenNext(JudyL_thread_stmt_pool, &thread_id, &first_then_next))) {
208
+ struct stmt_pool_s *local_stmt_pool = (struct stmt_pool_s *) *Pvalue;
209
+ if (!local_stmt_pool)
210
+ continue;
211
+ nd_log_daemon(
212
+ NDLP_WARNING,
213
+ "SQL: Pending SQL statements for thread %lu (%s), make sure thread does a proper cleanup",
214
+ thread_id,
215
+ local_stmt_pool->name);
216
+ finalize_and_free_stmt_list(local_stmt_pool);
217
+ }
218
+ (void)JudyLFreeArray(&JudyL_thread_stmt_pool, PJE0);
219
}
193
- spinlock_unlock(&sqlite_spinlock);
220
+ spinlock_unlock(&JudyL_thread_stmt_lock);
221
}
222
196
-static void initialize_thread_key_pool(void)
197
-{
198
- for (int i = 0; i < MAX_PREPARED_STATEMENTS; i++)
199
- (void)pthread_key_create(&key_pool[i], release_statement);
223
+static void init_thread_stmt_pool(void) {
224
+ thread_stmt_pool = (struct stmt_pool_s *)mallocz(sizeof(struct stmt_pool_s));
225
+ if (!thread_stmt_pool)
226
+ fatal("Failed to allocate memory for statement pool");
227
+
228
+ thread_stmt_pool->count = 0;
229
+ thread_stmt_pool->thread_id = gettid_cached();
230
+ thread_stmt_pool->name = strdupz(nd_thread_tag());
231
+ memset(thread_stmt_pool->stmt, 0, sizeof(void *) * MAX_PREPARED_THREAD_STATEMENTS);
232
+
233
+ // Add it to the JudyL array
234
+ spinlock_lock(&JudyL_thread_stmt_lock);
235
+ Pvoid_t *Pvalue = JudyLIns(&JudyL_thread_stmt_pool, (Word_t)thread_stmt_pool->thread_id, PJE0);
236
+ if (!Pvalue || Pvalue == PJERR)
237
+ fatal("Failed to allocate memory for JudyL thread statement pool");
238
+ struct stmt_pool_s *old_pool = *Pvalue;
239
+ fatal_assert(old_pool == NULL);
240
+ *Pvalue = thread_stmt_pool;
241
+ spinlock_unlock(&JudyL_thread_stmt_lock);
242
}
243
244
int prepare_statement(sqlite3 *database, const char *query, sqlite3_stmt **statement)
245
{
204
- static __thread uint32_t keys_used = 0;
205
-
206
- pthread_key_t *key = NULL;
207
- int ret = 1;
208
-
209
- if (likely(keys_used < MAX_PREPARED_STATEMENTS))
210
- key = &key_pool[keys_used++];
211
-
246
int rc = sqlite3_prepare_v2(database, query, -1, statement, 0);
247
if (rc == SQLITE_OK) {
214
- if (key)
215
- ret = pthread_setspecific(*key, *statement);
216
- if (ret)
217
- add_stmt_to_list(*statement);
248
+ if (!thread_stmt_pool)
249
+ init_thread_stmt_pool();
250
+ int stmt_key = __atomic_fetch_add(&thread_stmt_pool->count, 1, __ATOMIC_RELAXED);
251
+ if (stmt_key < MAX_PREPARED_THREAD_STATEMENTS)
252
+ thread_stmt_pool->stmt[stmt_key] = *statement;
253
}
254
return rc;
255
}
@@ -385,7 +420,12 @@ extern sqlite3 *db_context_meta;
420
421
void sqlite_close_databases(void)
422
{
388
- add_stmt_to_list(NULL);
423
+ // In case we have statements in the main thread
424
+ finalize_self_prepared_sql_statements();
425
+
426
+ // Finalize pending statements and report any thread that failed
427
+ // to do it properly
428
+ finalize_all_prepared_sql_statements();
429
430
spinlock_lock(&sqlite_spinlock);
431
sqlite_online = false;
@@ -417,7 +457,6 @@ uint64_t get_total_database_space(void)
457
int sqlite_library_init(void)
458
{
459
spinlock_lock(&sqlite_spinlock);
420
- initialize_thread_key_pool();
460
461
int rc = sqlite3_initialize();
462
if (rc == SQLITE_OK) {
src/database/sqlite/sqlite_functions.h
+3
@@ -101,6 +101,9 @@ int configure_sqlite_database(sqlite3 *database, int target_version, const char
101
// Helpers
102
int bind_text_null(sqlite3_stmt *res, int position, const char *text, bool can_be_null);
103
int prepare_statement(sqlite3 *database, const char *query, sqlite3_stmt **statement);
104
+void finalize_self_prepared_sql_statements();
105
+void finalize_all_prepared_sql_statements();
106
+
107
int execute_insert(sqlite3_stmt *res);
108
int db_execute(sqlite3 *database, const char *cmd);
109
char *get_database_extented_error(sqlite3 *database, int i, const char *description);
src/health/health_event_loop.c
+1
@@ -704,6 +704,7 @@ static void health_main_cleanup(void *pptr) {
704
static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;
705
static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;
706
707
+ finalize_self_prepared_sql_statements();
708
nd_log(NDLS_DAEMON, NDLP_DEBUG, "Health thread ended.");
709
}
710
src/ml/ml.cc
+2
@@ -983,6 +983,7 @@ ml_detect_main(void *arg)
983
}
984
}
985
Cfg.training_stop = true;
986
+ finalize_self_prepared_sql_statements();
987
988
return NULL;
989
}
@@ -1211,6 +1212,7 @@ void *ml_train_main(void *arg) {
1212
worker_is_idle();
1213
std::this_thread::sleep_for(std::chrono::microseconds{remaining_ut});
1214
}
1215
+ finalize_self_prepared_sql_statements();
1216
1217
return NULL;
1218
}