geos_one/bongo
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases 1 Wiki Activity

Fix Store transaction deadlock under concurrency

Browse Source
This commit is contained in:
Mario Fetka
2026-07-18 13:31:40 +02:00
parent 35e9ed7c1f
commit bac7eac97d
3 changed files with 179 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/libs/msgapi/CMakeLists.txt
+20
View File
@@ -21,4 +21,24 @@ target_link_libraries(bongomsgapi
install(TARGETS bongomsgapi DESTINATION ${LIB_INSTALL_DIR})
if(BUILD_TESTING)
add_executable(msgapi-sqldb-transaction-test
tests/sqldb-transaction-test.c)
if(CMAKE_SYSTEM_NAME STREQUAL "Linux")
target_link_options(msgapi-sqldb-transaction-test PRIVATE
"LINKER:--no-as-needed")
endif()
target_link_libraries(msgapi-sqldb-transaction-test PRIVATE
bongomsgapi
bongocal
bongojson
bongoutil
bongoxpl
SQLite3::SQLite3
Threads::Threads)
add_test(NAME msgapi-sqldb-transactions
COMMAND msgapi-sqldb-transaction-test)
set_tests_properties(msgapi-sqldb-transactions PROPERTIES TIMEOUT 30)
endif()
add_subdirectory(auth-backends)
src/libs/msgapi/sqldb.c
+43 -19
View File
@@ -138,38 +138,48 @@ MsgSQLSetLockTimeout(MsgSQLHandle *handle, int timeoutms)
int
MsgSQLBeginTransaction(MsgSQLHandle *handle)
{
MsgSQLStatement statement;
MsgSQLStatement *stmt;
int result;
int count = 1 + handle->lockTimeoutMs / MSGSQL_STMT_SLEEP_MS;
BOOL locked = FALSE;
memset(&statement, 0, sizeof(statement));
// acquire the transaction lock to prevent other people doing stuff
XplMutexLock(handle->transactionLock);
if (locked) {
Log(LOG_TRACE, "sql3: Acquired lock successfully");
}
stmt = MsgSQLPrepare(handle, "BEGIN TRANSACTION;", &handle->stmts.begin);
/*
* Store connections move between worker threads. Keeping transaction
* statements cached on the shared handle made SQLite reject a later BEGIN
* even though the previous transaction had completed. Transaction control
* statements are small, so prepare and finalize them within the lock.
*/
stmt = MsgSQLPrepare(handle, "BEGIN TRANSACTION;", &statement);
if (!stmt) {
Log(LOG_ERROR, "sql3: Unable to begin transaction");
XplMutexUnlock(handle->transactionLock);
return -1;
}
do {
result = sqlite3_step(stmt->stmt);
Log(LOG_TRACE, "sql3: BEGIN TRAN stmt %ld", stmt->stmt);
sqlite3_reset(stmt->stmt);
Log(LOG_TRACE, "sql: - bt reset stmt %ld", stmt->stmt);
} while (SQLITE_BUSY == result && --count && (XplDelay(MSGSQL_STMT_SLEEP_MS), 1));
switch (result) {
case SQLITE_DONE:
MsgSQLFinalize(&statement);
++(handle->transactionDepth);
return 0;
case SQLITE_BUSY:
Log(LOG_ERROR, "sql3: Database remained busy while beginning transaction");
MsgSQLFinalize(&statement);
XplMutexUnlock(handle->transactionLock);
return -2;
default:
Log(LOG_ERROR, "sql3: Database error %d : %s", result, sqlite3_errmsg(handle->db));
MsgSQLFinalize(&statement);
XplMutexUnlock(handle->transactionLock);
return -1;
}
}
@@ -178,11 +188,18 @@ MsgSQLBeginTransaction(MsgSQLHandle *handle)
int
MsgSQLCommitTransaction(MsgSQLHandle *handle)
{
MsgSQLStatement statement;
MsgSQLStatement *stmt;
int result;
int count = 1 + handle->lockTimeoutMs / MSGSQL_STMT_SLEEP_MS;
stmt = MsgSQLPrepare(handle, "END TRANSACTION;", &handle->stmts.end);
if (handle->transactionDepth <= 0) {
Log(LOG_ERROR, "sql3: Attempted to commit without an active transaction");
return -1;
}
memset(&statement, 0, sizeof(statement));
stmt = MsgSQLPrepare(handle, "END TRANSACTION;", &statement);
if (!stmt) {
Log(LOG_ERROR, "sql3: Unable to prepare end of transaction statement");
return -1;
@@ -191,15 +208,15 @@ MsgSQLCommitTransaction(MsgSQLHandle *handle)
do {
result = sqlite3_step(stmt->stmt);
Log(LOG_TRACE, "sql3: COMMIT stmt %ld", stmt->stmt);
sqlite3_reset(stmt->stmt);
Log(LOG_TRACE, "sql3: - ct reset stmt %ld", stmt->stmt);
} while (SQLITE_BUSY == result && --count && (XplDelay(MSGSQL_STMT_SLEEP_MS), 1));
if (SQLITE_DONE != result) {
Log(LOG_ERROR, "sql3: Database commit error %d: %s", result, sqlite3_errmsg(handle->db));
MsgSQLFinalize(&statement);
return -1;
}
MsgSQLFinalize(&statement);
--handle->transactionDepth;
XplMutexUnlock(handle->transactionLock);
@@ -210,22 +227,29 @@ MsgSQLCommitTransaction(MsgSQLHandle *handle)
int
MsgSQLAbortTransaction(MsgSQLHandle *handle)
{
MsgSQLStatement statement;
MsgSQLStatement *stmt;
int result;
stmt = MsgSQLPrepare(handle, "ROLLBACK TRANSACTION;", &handle->stmts.abort);
if (!stmt) {
Log(LOG_ERROR, "sql3: Unable to rollback transaction");
if (handle->transactionDepth <= 0) {
Log(LOG_ERROR, "sql3: Attempted to rollback without an active transaction");
return -1;
}
result = sqlite3_step(stmt->stmt);
if (SQLITE_DONE != result) {
Log(LOG_ERROR, "sql3: Transaction rollback failed");
memset(&statement, 0, sizeof(statement));
stmt = MsgSQLPrepare(handle, "ROLLBACK TRANSACTION;", &statement);
if (!stmt) {
Log(LOG_ERROR, "sql3: Unable to rollback transaction");
result = SQLITE_ERROR;
} else {
result = sqlite3_step(stmt->stmt);
if (SQLITE_DONE != result) {
Log(LOG_ERROR, "sql3: Transaction rollback failed: %s", sqlite3_errmsg(handle->db));
}
Log(LOG_TRACE, "sql3: ABORT stmt %ld\n", stmt->stmt);
MsgSQLFinalize(&statement);
}
Log(LOG_TRACE, "sql3: ABORT stmt %ld\n", stmt->stmt);
sqlite3_reset(stmt->stmt);
Log(LOG_TRACE, "sql3: - at reset stmt %ld\n", stmt->stmt);
--handle->transactionDepth;
XplMutexUnlock(handle->transactionLock);
src/libs/msgapi/tests/sqldb-transaction-test.c
+116
View File
@@ -0,0 +1,116 @@
/****************************************************************************
* <Novell-copyright>
* Copyright (c) 2001 Novell, Inc. All Rights Reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of version 2 of the GNU General Public License
* as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, contact Novell, Inc.
* </Novell-copyright>
****************************************************************************/
#include <assert.h>
#include <pthread.h>
#include <sqlite3.h>
#include <string.h>
#include <xpl.h>
#include <msgapi.h>
#define THREAD_COUNT 12
#define TRANSACTIONS_PER_THREAD 100
typedef struct {
MsgSQLHandle *handle;
int failed;
} TransactionThread;
static void *
RunTransactions(void *argument)
{
TransactionThread *thread = argument;
int i;
for (i = 0; i < TRANSACTIONS_PER_THREAD; i++) {
if (MsgSQLBeginTransaction(thread->handle) != 0) {
thread->failed = 1;
return NULL;
}
if (MsgSQLQuickExecute(thread->handle, "SELECT 1;") != 0) {
MsgSQLAbortTransaction(thread->handle);
thread->failed = 1;
return NULL;
}
if (MsgSQLCommitTransaction(thread->handle) != 0) {
MsgSQLAbortTransaction(thread->handle);
thread->failed = 1;
return NULL;
}
}
return NULL;
}
static void
CheckBeginFailureReleasesLock(void)
{
MsgSQLHandle *handle;
char *error = NULL;
int lockResult;
handle = MsgSQLOpen(":memory:", NULL, 0);
assert(handle != NULL);
assert(sqlite3_exec(handle->db, "BEGIN TRANSACTION;", NULL, NULL,
&error) == SQLITE_OK);
assert(error == NULL);
assert(MsgSQLBeginTransaction(handle) == -1);
lockResult = XplMutexTryLock(handle->transactionLock);
assert(lockResult == 0);
XplMutexUnlock(handle->transactionLock);
assert(sqlite3_exec(handle->db, "ROLLBACK TRANSACTION;", NULL, NULL,
&error) == SQLITE_OK);
assert(error == NULL);
assert(MsgSQLBeginTransaction(handle) == 0);
assert(MsgSQLCommitTransaction(handle) == 0);
MsgSQLClose(handle);
}
static void
CheckConcurrentTransactions(void)
{
MsgSQLHandle *handle;
TransactionThread threads[THREAD_COUNT];
pthread_t ids[THREAD_COUNT];
int i;
handle = MsgSQLOpen(":memory:", NULL, 1000);
assert(handle != NULL);
memset(threads, 0, sizeof(threads));
for (i = 0; i < THREAD_COUNT; i++) {
threads[i].handle = handle;
assert(pthread_create(&ids[i], NULL, RunTransactions,
&threads[i]) == 0);
}
for (i = 0; i < THREAD_COUNT; i++) {
assert(pthread_join(ids[i], NULL) == 0);
assert(threads[i].failed == 0);
}
assert(handle->transactionDepth == 0);
MsgSQLClose(handle);
}
int
main(void)
{
CheckBeginFailureReleasesLock();
CheckConcurrentTransactions();
return 0;
}