summaryrefslogtreecommitdiff
path: root/src/common/Database/DatabaseWorkerPool.h
blob: 322644bf7a1e7d59a2dfd9df2d712415ef87c711 (plain)
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
/*
 * Copyright (C) 2016+     AzerothCore <www.azerothcore.org>
 * Copyright (C) 2008-2016 TrinityCore <http://www.trinitycore.org/>
 * Copyright (C) 2005-2009 MaNGOS <http://getmangos.com/>
 */

#ifndef _DATABASEWORKERPOOL_H
#define _DATABASEWORKERPOOL_H

#include <ace/Thread_Mutex.h>

#include "Common.h"
#include "Callback.h"
#include "MySQLConnection.h"
#include "Transaction.h"
#include "DatabaseWorker.h"
#include "PreparedStatement.h"
#include "Log.h"
#include "QueryResult.h"
#include "QueryHolder.h"
#include "AdhocStatement.h"
#include "StringFormat.h"

class PingOperation : public SQLOperation
{
    //! Operation for idle delaythreads
    bool Execute() override
    {
        m_conn->Ping();
        return true;
    }
};

template <class T>
class DatabaseWorkerPool
{
public:
    /* Activity state */
    DatabaseWorkerPool();

    ~DatabaseWorkerPool() = default;

    bool Open(const std::string& infoString, uint8 async_threads, uint8 synch_threads);

    void Close();

    /**
        Delayed one-way statement methods.
    */

    //! Enqueues a one-way SQL operation in string format that will be executed asynchronously.
    //! This method should only be used for queries that are only executed once, e.g during startup.
    void Execute(const char* sql);

    //! Enqueues a one-way SQL operation in string format -with variable args- that will be executed asynchronously.
    //! This method should only be used for queries that are only executed once, e.g during startup.
    template<typename Format, typename... Args>
    void PExecute(Format&& sql, Args&& ... args)
    {
        if (acore::IsFormatEmptyOrNull(sql))
            return;

        Execute(acore::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
    }

    //! Enqueues a one-way SQL operation in prepared statement format that will be executed asynchronously.
    //! Statement must be prepared with CONNECTION_ASYNC flag.
    void Execute(PreparedStatement* stmt);

    /**
        Direct synchronous one-way statement methods.
    */

    //! Directly executes a one-way SQL operation in string format, that will block the calling thread until finished.
    //! This method should only be used for queries that are only executed once, e.g during startup.
    void DirectExecute(const char* sql);

    //! Directly executes a one-way SQL operation in string format -with variable args-, that will block the calling thread until finished.
    //! This method should only be used for queries that are only executed once, e.g during startup.
    template<typename Format, typename... Args>
    void DirectPExecute(Format&& sql, Args&& ... args)
    {
        if (acore::IsFormatEmptyOrNull(sql))
            return;

        DirectExecute(acore::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
    }

    //! Directly executes a one-way SQL operation in prepared statement format, that will block the calling thread until finished.
    //! Statement must be prepared with the CONNECTION_SYNCH flag.
    void DirectExecute(PreparedStatement* stmt);

    /**
        Synchronous query (with resultset) methods.
    */

    //! Directly executes an SQL query in string format that will block the calling thread until finished.
    //! Returns reference counted auto pointer, no need for manual memory management in upper level code.
    QueryResult Query(const char* sql, T* conn = nullptr);

    //! Directly executes an SQL query in string format -with variable args- that will block the calling thread until finished.
    //! Returns reference counted auto pointer, no need for manual memory management in upper level code.
    template<typename Format, typename... Args>
    QueryResult PQuery(Format&& sql, T* conn, Args&& ... args)
    {
        if (acore::IsFormatEmptyOrNull(sql))
            return QueryResult(nullptr);

        return Query(acore::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str(), conn);
    }

    //! Directly executes an SQL query in string format -with variable args- that will block the calling thread until finished.
    //! Returns reference counted auto pointer, no need for manual memory management in upper level code.
    template<typename Format, typename... Args>
    QueryResult PQuery(Format&& sql, Args&& ... args)
    {
        if (acore::IsFormatEmptyOrNull(sql))
            return QueryResult(nullptr);

        return Query(acore::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
    }

    //! Directly executes an SQL query in prepared format that will block the calling thread until finished.
    //! Returns reference counted auto pointer, no need for manual memory management in upper level code.
    //! Statement must be prepared with CONNECTION_SYNCH flag.
    PreparedQueryResult Query(PreparedStatement* stmt);

    /**
        Asynchronous query (with resultset) methods.
    */

    //! Enqueues a query in string format that will set the value of the QueryResultFuture return object as soon as the query is executed.
    //! The return value is then processed in ProcessQueryCallback methods.
    QueryResultFuture AsyncQuery(const char* sql);

    //! Enqueues a query in string format -with variable args- that will set the value of the QueryResultFuture return object as soon as the query is executed.
    //! The return value is then processed in ProcessQueryCallback methods.
    template<typename Format, typename... Args>
    QueryResultFuture AsyncPQuery(Format&& sql, Args&& ... args)
    {
        if (acore::IsFormatEmptyOrNull(sql))
            return QueryResult(nullptr);

        return AsyncQuery(acore::StringFormat(std::forward<Format>(sql), std::forward<Args>(args)...).c_str());
    }

    //! Enqueues a query in prepared format that will set the value of the PreparedQueryResultFuture return object as soon as the query is executed.
    //! The return value is then processed in ProcessQueryCallback methods.
    //! Statement must be prepared with CONNECTION_ASYNC flag.
    PreparedQueryResultFuture AsyncQuery(PreparedStatement* stmt);

    //! Enqueues a vector of SQL operations (can be both adhoc and prepared) that will set the value of the QueryResultHolderFuture
    //! return object as soon as the query is executed.
    //! The return value is then processed in ProcessQueryCallback methods.
    //! Any prepared statements added to this holder need to be prepared with the CONNECTION_ASYNC flag.
    QueryResultHolderFuture DelayQueryHolder(SQLQueryHolder* holder);

    /**
        Transaction context methods.
    */

    //! Begins an automanaged transaction pointer that will automatically rollback if not commited. (Autocommit=0)
    SQLTransaction BeginTransaction();

    //! Enqueues a collection of one-way SQL operations (can be both adhoc and prepared). The order in which these operations
    //! were appended to the transaction will be respected during execution.
    void CommitTransaction(SQLTransaction transaction);

    //! Directly executes a collection of one-way SQL operations (can be both adhoc and prepared). The order in which these operations
    //! were appended to the transaction will be respected during execution.
    void DirectCommitTransaction(SQLTransaction& transaction);

    //! Method used to execute prepared statements in a diverse context.
    //! Will be wrapped in a transaction if valid object is present, otherwise executed standalone.
    void ExecuteOrAppend(SQLTransaction& trans, PreparedStatement* stmt);

    //! Method used to execute ad-hoc statements in a diverse context.
    //! Will be wrapped in a transaction if valid object is present, otherwise executed standalone.
    void ExecuteOrAppend(SQLTransaction& trans, const char* sql);

    /**
        Other
    */

    //! Automanaged (internally) pointer to a prepared statement object for usage in upper level code.
    //! Pointer is deleted in this->DirectExecute(PreparedStatement*), this->Query(PreparedStatement*) or PreparedStatementTask::~PreparedStatementTask.
    //! This object is not tied to the prepared statement on the MySQL context yet until execution.
    PreparedStatement* GetPreparedStatement(uint32 index);

    //! Apply escape string'ing for current collation. (utf8)
    unsigned long EscapeString(char* to, const char* from, unsigned long length)
    {
        if (!to || !from || !length)
            return 0;

        return mysql_real_escape_string(_connections[IDX_SYNCH][0]->GetHandle(), to, from, length);
    }

    //! Keeps all our MySQL connections alive, prevent the server from disconnecting us.
    void KeepAlive();

    [[nodiscard]] char const* GetDatabaseName() const
    {
        return _connectionInfo.database.c_str();
    }

    void EscapeString(std::string& str)
    {
        if (str.empty())
            return;

        char* buf = new char[str.size() * 2 + 1];
        EscapeString(buf, str.c_str(), str.size());
        str = buf;
        delete[] buf;
    }

private:

    void Enqueue(SQLOperation* op)
    {
        _queue->enqueue(op);
    }

    //! Gets a free connection in the synchronous connection pool.
    //! Caller MUST call t->Unlock() after touching the MySQL context to prevent deadlocks.
    T* GetFreeConnection();

private:
    enum _internalIndex
    {
        IDX_ASYNC,
        IDX_SYNCH,
        IDX_SIZE
    };

    ACE_Message_Queue<ACE_SYNCH>*   _mqueue;
    ACE_Activation_Queue*           _queue;             //! Queue shared by async worker threads.
    std::vector<std::vector<T*>>    _connections;
    uint32                          _connectionCount[2];       //! Counter of MySQL connections;
    MySQLConnectionInfo             _connectionInfo;
};

#endif