DatabaseWorkerPool.h

Go to the documentation of this file.

/*
* This file is part of Project SkyFire https://www.projectskyfire.org.
* See LICENSE.md file for Copyright information
*/
 
#ifndef _DATABASEWORKERPOOL_H
#define _DATABASEWORKERPOOL_H
 
#include "AdhocStatement.h"
#include "Callback.h"
#include "Common.h"
#include "DatabaseQueue.h"
#include "DatabaseWorker.h"
#include "Log.h"
#include "MySQLConnection.h"
#include "PreparedStatement.h"
#include "QueryHolder.h"
#include "QueryResult.h"
#include "Transaction.h"
 
#define MIN_MYSQL_SERVER_VERSION 50100u
#define MIN_MYSQL_CLIENT_VERSION 50100u
 
class PingOperation : public SQLOperation
{
    bool Execute()
    {
        m_conn->Ping();
        return true;
    }
};
 
template <class T>
class DatabaseWorkerPool
{
public:
    /* Activity state */
    DatabaseWorkerPool() : _queue(new Skyfire::DatabaseQueue()), _connectionInfo(NULL)
    {
        memset(_connectionCount, 0, sizeof(_connectionCount));
        _connections.resize(IDX_SIZE);
 
        WPFatal(mysql_thread_safe(), "Used MySQL library isn't thread-safe.");
        WPFatal(mysql_get_client_version() >= MIN_MYSQL_CLIENT_VERSION, "SkyFire does not support MySQL versions below 5.1");
    }
 
    ~DatabaseWorkerPool()
    {
    }
 
    bool Open(const char* host, const char* port, const char* user, const char* password, const char* database, uint8 async_threads, uint8 synch_threads)
    {
        bool res = true;
        _connectionInfo = new MySQLConnectionInfo(host, port, user, password, database);
 
        SF_LOG_INFO("sql.driver", "Opening DatabasePool '%s'. Asynchronous connections: %u, synchronous connections: %u.",
            GetDatabaseName(), async_threads, synch_threads);

        _connections[IDX_ASYNC].resize(async_threads);
        for (uint8 i = 0; i < async_threads; ++i)
        {
            T* t = new T(_queue, *_connectionInfo);
            res &= t->Open();
            if (res) // only check mysql version if connection is valid
                WPFatal(mysql_get_server_version(t->GetHandle()) >= MIN_MYSQL_SERVER_VERSION, "SkyFire does not support MySQL versions below 5.1");
            _connections[IDX_ASYNC][i] = t;
            ++_connectionCount[IDX_ASYNC];
        }

        _connections[IDX_SYNCH].resize(synch_threads);
        for (uint8 i = 0; i < synch_threads; ++i)
        {
            T* t = new T(*_connectionInfo);
            res &= t->Open();
            _connections[IDX_SYNCH][i] = t;
            ++_connectionCount[IDX_SYNCH];
        }
 
        if (res)
            SF_LOG_INFO("sql.driver", "DatabasePool '%s' opened successfully. %u total connections running.", GetDatabaseName(),
                (_connectionCount[IDX_SYNCH] + _connectionCount[IDX_ASYNC]));
        else
            SF_LOG_ERROR("sql.driver", "DatabasePool %s NOT opened. There were errors opening the MySQL connections. Check your SQLDriverLogFile "
                "for specific errors.", GetDatabaseName());
        return res;
    }
 
    bool Open(const std::string& infoString, uint8 async_threads, uint8 synch_threads)
    {
        bool res = true;
        _connectionInfo = new MySQLConnectionInfo(infoString);
 
        SF_LOG_INFO("sql.driver", "Opening DatabasePool '%s'. Asynchronous connections: %u, synchronous connections: %u.",
            GetDatabaseName(), async_threads, synch_threads);

        _connections[IDX_ASYNC].resize(async_threads);
        for (uint8 i = 0; i < async_threads; ++i)
        {
            T* t = new T(_queue, *_connectionInfo);
            res &= t->Open();
            if (res) // only check mysql version if connection is valid
                WPFatal(mysql_get_server_version(t->GetHandle()) >= MIN_MYSQL_SERVER_VERSION, "Skyfire does not support MySQL versions below 5.1");
            _connections[IDX_ASYNC][i] = t;
            ++_connectionCount[IDX_ASYNC];
        }

        _connections[IDX_SYNCH].resize(synch_threads);
        for (uint8 i = 0; i < synch_threads; ++i)
        {
            T* t = new T(*_connectionInfo);
            res &= t->Open();
            _connections[IDX_SYNCH][i] = t;
            ++_connectionCount[IDX_SYNCH];
        }
 
        if (res)
            SF_LOG_INFO("sql.driver", "DatabasePool '%s' opened successfully. %u total connections running.", GetDatabaseName(),
                (_connectionCount[IDX_SYNCH] + _connectionCount[IDX_ASYNC]));
        else
            SF_LOG_ERROR("sql.driver", "DatabasePool %s NOT opened. There were errors opening the MySQL connections. Check your SQLDriverLogFile "
                "for specific errors.", GetDatabaseName());
        return res;
    }
 
    void Close()
    {
        SF_LOG_INFO("sql.driver", "Closing down DatabasePool '%s'.", GetDatabaseName());

        _queue->close();
 
        for (uint8 i = 0; i < _connectionCount[IDX_ASYNC]; ++i)
        {
            T* t = _connections[IDX_ASYNC][i];
            DatabaseWorker* worker = t->m_worker;
            worker->wait();     
            delete worker;
            t->Close();         
        }
 
        SF_LOG_INFO("sql.driver", "Asynchronous connections on DatabasePool '%s' terminated. Proceeding with synchronous connections.",
            GetDatabaseName());

        for (uint8 i = 0; i < _connectionCount[IDX_SYNCH]; ++i)
            _connections[IDX_SYNCH][i]->Close();
 
        delete _queue;
 
        SF_LOG_INFO("sql.driver", "All connections on DatabasePool '%s' closed.", GetDatabaseName());
        delete _connectionInfo;
        _connectionInfo = NULL;
    }


    void Execute(const char* sql)
    {
        if (!sql)
            return;
 
        BasicStatementTask* task = new BasicStatementTask(sql);
        Enqueue(task);
    }

    void PExecute(const char* sql, ...)
    {
        if (!sql)
            return;
 
        va_list ap;
        char szQuery[MAX_QUERY_LEN];
        va_start(ap, sql);
        vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
        va_end(ap);
 
        Execute(szQuery);
    }

    void Execute(PreparedStatement* stmt)
    {
        PreparedStatementTask* task = new PreparedStatementTask(stmt);
        Enqueue(task);
    }


    void DirectExecute(const char* sql)
    {
        if (!sql)
            return;
 
        T* t = GetFreeConnection();
        t->Execute(sql);
        t->Unlock();
    }

    void DirectPExecute(const char* sql, ...)
    {
        if (!sql)
            return;
 
        va_list ap;
        char szQuery[MAX_QUERY_LEN];
        va_start(ap, sql);
        vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
        va_end(ap);
 
        return DirectExecute(szQuery);
    }

    void DirectExecute(PreparedStatement* stmt)
    {
        T* t = GetFreeConnection();
        t->Execute(stmt);
        t->Unlock();

        delete stmt;
    }


    QueryResult Query(const char* sql, T* conn = NULL)
    {
        if (!conn)
            conn = GetFreeConnection();
 
        ResultSet* result = conn->Query(sql);
        conn->Unlock();
        if (!result || !result->GetRowCount())
        {
            delete result;
            return QueryResult(NULL);
        }
 
        result->NextRow();
        return QueryResult(result);
    }

    QueryResult PQuery(const char* sql, T* conn, ...)
    {
        if (!sql)
            return QueryResult(NULL);
 
        va_list ap;
        char szQuery[MAX_QUERY_LEN];
        va_start(ap, conn);
        vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
        va_end(ap);
 
        return Query(szQuery, conn);
    }

    QueryResult PQuery(const char* sql, ...)
    {
        if (!sql)
            return QueryResult(NULL);
 
        va_list ap;
        char szQuery[MAX_QUERY_LEN];
        va_start(ap, sql);
        vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
        va_end(ap);
 
        return Query(szQuery);
    }

    PreparedQueryResult Query(PreparedStatement* stmt)
    {
        T* t = GetFreeConnection();
        PreparedResultSet* ret = t->Query(stmt);
        t->Unlock();

        delete stmt;
 
        if (!ret || !ret->GetRowCount())
        {
            delete ret;
            return PreparedQueryResult(NULL);
        }
 
        return PreparedQueryResult(ret);
    }


    QueryResultFuture AsyncQuery(const char* sql)
    {
        QueryResultFuture res;
        BasicStatementTask* task = new BasicStatementTask(sql, res);
        Enqueue(task);
        return res;         
    }

    QueryResultFuture AsyncPQuery(const char* sql, ...)
    {
        va_list ap;
        char szQuery[MAX_QUERY_LEN];
        va_start(ap, sql);
        vsnprintf(szQuery, MAX_QUERY_LEN, sql, ap);
        va_end(ap);
 
        return AsyncQuery(szQuery);
    }

    PreparedQueryResultFuture AsyncQuery(PreparedStatement* stmt)
    {
        PreparedQueryResultFuture res;
        PreparedStatementTask* task = new PreparedStatementTask(stmt, res);
        Enqueue(task);
        return res;
    }

    QueryResultHolderFuture DelayQueryHolder(SQLQueryHolder* holder)
    {
        QueryResultHolderFuture res;
        SQLQueryHolderTask* task = new SQLQueryHolderTask(holder, res);
        Enqueue(task);
        return res;     
    }


    SQLTransaction BeginTransaction()
    {
        return SQLTransaction(new Transaction);
    }

    void CommitTransaction(SQLTransaction transaction)
    {
#ifdef SKYFIRE_DEBUG
        switch (transaction->GetSize())
        {
            case 0:
                SF_LOG_DEBUG("sql.driver", "Transaction contains 0 queries. Not executing.");
                return;
            case 1:
                SF_LOG_DEBUG("sql.driver", "Warning: Transaction only holds 1 query, consider removing Transaction context in code.");
                break;
            default:
                break;
        }
#endif // SKYFIRE_DEBUG
 
        Enqueue(new TransactionTask(transaction));
    }

    void Wait()
    {
        _queue->wait();
    }

    void DirectCommitTransaction(SQLTransaction& transaction)
    {
        T* con = GetFreeConnection();
        if (con->ExecuteTransaction(transaction))
        {
            con->Unlock();      // OK, operation succesful
            return;
        }

        if (con->GetLastError() == 1213)
        {
            uint8 loopBreaker = 5;
            for (uint8 i = 0; i < loopBreaker; ++i)
            {
                if (con->ExecuteTransaction(transaction))
                    break;
            }
        }

        transaction->Cleanup();
 
        con->Unlock();
    }

    void ExecuteOrAppend(SQLTransaction& trans, PreparedStatement* stmt)
    {
        if (trans.null())
            Execute(stmt);
        else
            trans->Append(stmt);
    }

    void ExecuteOrAppend(SQLTransaction& trans, const char* sql)
    {
        if (trans.null())
            Execute(sql);
        else
            trans->Append(sql);
    }


    PreparedStatement* GetPreparedStatement(uint32 index)
    {
        return new PreparedStatement(index);
    }

    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;
    }

    void KeepAlive()
    {
        for (uint8 i = 0; i < _connectionCount[IDX_SYNCH]; ++i)
        {
            T* t = _connections[IDX_SYNCH][i];
            if (t->LockIfReady())
            {
                t->Ping();
                t->Unlock();
            }
        }

        for (size_t i = 0; i < _connections[IDX_ASYNC].size(); ++i)
            Enqueue(new PingOperation);
    }
 
private:
    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);
    }
 
    void Enqueue(SQLOperation* op)
    {
        _queue->enqueue(op);
    }

    T* GetFreeConnection()
    {
        uint8 i = 0;
        size_t num_cons = _connectionCount[IDX_SYNCH];
        T* t = NULL;
        for (;;)
        {
            t = _connections[IDX_SYNCH][++i % num_cons];
            if (t->LockIfReady())
                break;
        }
 
        return t;
    }
 
    char const* GetDatabaseName() const
    {
        return _connectionInfo->_database.c_str();
    }
 
private:
    enum _internalIndex
    {
        IDX_ASYNC,
        IDX_SYNCH,
        IDX_SIZE
    };
 
    Skyfire::DatabaseQueue* _queue;           
    std::vector< std::vector<T*> >  _connections;
    uint32                          _connectionCount[2];       
    MySQLConnectionInfo* _connectionInfo;
};
 
#endif