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
|
/*
* Copyright (C) 2016+ AzerothCore <www.azerothcore.org>, released under GNU GPL v2 license, you may redistribute it and/or modify it under version 2 of the License, or (at your option), any later version.
* Copyright (C) 2021+ WarheadCore <https://github.com/WarheadCore>
*/
#include "Transaction.h"
#include "Log.h"
#include "MySQLConnection.h"
#include "PreparedStatement.h"
#include "Timer.h"
#include <mysqld_error.h>
#include <sstream>
#include <thread>
std::mutex TransactionTask::_deadlockLock;
#define DEADLOCK_MAX_RETRY_TIME_MS 60000
//- Append a raw ad-hoc query to the transaction
void TransactionBase::Append(char const* sql)
{
SQLElementData data;
data.type = SQL_ELEMENT_RAW;
data.element.query = strdup(sql);
m_queries.push_back(data);
}
//- Append a prepared statement to the transaction
void TransactionBase::AppendPreparedStatement(PreparedStatementBase* stmt)
{
SQLElementData data;
data.type = SQL_ELEMENT_PREPARED;
data.element.stmt = stmt;
m_queries.push_back(data);
}
void TransactionBase::Cleanup()
{
// This might be called by explicit calls to Cleanup or by the auto-destructor
if (_cleanedUp)
return;
for (SQLElementData const& data : m_queries)
{
switch (data.type)
{
case SQL_ELEMENT_PREPARED:
delete data.element.stmt;
break;
case SQL_ELEMENT_RAW:
free((void*)(data.element.query));
break;
}
}
m_queries.clear();
_cleanedUp = true;
}
bool TransactionTask::Execute()
{
int errorCode = TryExecute();
if (!errorCode)
return true;
if (errorCode == ER_LOCK_DEADLOCK)
{
std::ostringstream threadIdStream;
threadIdStream << std::this_thread::get_id();
std::string threadId = threadIdStream.str();
// Make sure only 1 async thread retries a transaction so they don't keep dead-locking each other
std::lock_guard<std::mutex> lock(_deadlockLock);
for (uint32 loopDuration = 0, startMSTime = getMSTime(); loopDuration <= DEADLOCK_MAX_RETRY_TIME_MS; loopDuration = GetMSTimeDiffToNow(startMSTime))
{
if (!TryExecute())
return true;
LOG_WARN("sql.sql", "Deadlocked SQL Transaction, retrying. Loop timer: %u ms, Thread Id: %s", loopDuration, threadId.c_str());
}
LOG_ERROR("sql.sql", "Fatal deadlocked SQL Transaction, it will not be retried anymore. Thread Id: %s", threadId.c_str());
}
// Clean up now.
CleanupOnFailure();
return false;
}
int TransactionTask::TryExecute()
{
return m_conn->ExecuteTransaction(m_trans);
}
void TransactionTask::CleanupOnFailure()
{
m_trans->Cleanup();
}
bool TransactionWithResultTask::Execute()
{
int errorCode = TryExecute();
if (!errorCode)
{
m_result.set_value(true);
return true;
}
if (errorCode == ER_LOCK_DEADLOCK)
{
std::ostringstream threadIdStream;
threadIdStream << std::this_thread::get_id();
std::string threadId = threadIdStream.str();
// Make sure only 1 async thread retries a transaction so they don't keep dead-locking each other
std::lock_guard<std::mutex> lock(_deadlockLock);
for (uint32 loopDuration = 0, startMSTime = getMSTime(); loopDuration <= DEADLOCK_MAX_RETRY_TIME_MS; loopDuration = GetMSTimeDiffToNow(startMSTime))
{
if (!TryExecute())
{
m_result.set_value(true);
return true;
}
LOG_WARN("sql.sql", "Deadlocked SQL Transaction, retrying. Loop timer: %u ms, Thread Id: %s", loopDuration, threadId.c_str());
}
LOG_ERROR("sql.sql", "Fatal deadlocked SQL Transaction, it will not be retried anymore. Thread Id: %s", threadId.c_str());
}
// Clean up now.
CleanupOnFailure();
m_result.set_value(false);
return false;
}
bool TransactionCallback::InvokeIfReady()
{
if (m_future.valid() && m_future.wait_for(std::chrono::seconds(0)) == std::future_status::ready)
{
m_callback(m_future.get());
return true;
}
return false;
}
|
