Inna/CypherCore
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Core/DB: Some DB improvements

Browse Source
This commit is contained in:
hondacrx
2018-05-22 11:05:16 -04:00
parent 0be2dc08e2
commit dbd62e5c6f
23 changed files with 360 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Source/Framework/Database/DatabaseWorker.cs
+62
View File
@@ -0,0 +1,62 @@
/*
* Copyright (C) 2012-2018 CypherCore <http://github.com/CypherCore>
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* 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, see <http://www.gnu.org/licenses/>.
*/
using Framework.Threading;
using System.Threading;
namespace Framework.Database
{
public interface ISqlOperation
{
bool Execute<T>(MySqlBase<T> mySqlBase);
}
class DatabaseWorker<T>
{
Thread _workerThread;
volatile bool _cancelationToken;
ProducerConsumerQueue<ISqlOperation> _queue;
MySqlBase<T> _mySqlBase;
public DatabaseWorker(ProducerConsumerQueue<ISqlOperation> newQueue, MySqlBase<T> mySqlBase)
{
_queue = newQueue;
_mySqlBase = mySqlBase;
_cancelationToken = false;
_workerThread = new Thread(WorkerThread);
_workerThread.Start();
}
void WorkerThread()
{
if (_queue == null)
return;
for (; ; )
{
ISqlOperation operation;
_queue.WaitAndPop(out operation);
if (_cancelationToken || operation == null)
return;
operation.Execute(_mySqlBase);
}
}
}
}
Source/Framework/Database/MySqlBase.cs
+64 -139
View File
@@ -15,6 +15,7 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
using Framework.Threading;
using MySql.Data.MySqlClient;
using System;
using System.Collections.Generic;
@@ -27,19 +28,14 @@ namespace Framework.Database
{
public class MySqlConnectionInfo
{
public MySqlConnectionInfo(int poolSize = 10)
{
Poolsize = poolSize;
}
public MySqlConnection GetConnection()
{
return new MySqlConnection($"Server={Host};Port={Port};User Id={Username};Password={Password};Database={Database};Allow Zero Datetime=True;Allow User Variables=True;Pooling=true;MaximumPoolSize={Poolsize};");
return new MySqlConnection($"Server={Host};Port={Port};User Id={Username};Password={Password};Database={Database};Allow User Variables=True;Pooling=true;");
}
public MySqlConnection GetConnectionNoDatabase()
{
return new MySqlConnection($"Server={Host};Port={Port};User Id={Username};Password={Password};Allow Zero Datetime=True;Allow User Variables=True;Pooling=true;MaximumPoolSize={Poolsize};");
return new MySqlConnection($"Server={Host};Port={Port};User Id={Username};Password={Password};Allow User Variables=True;Pooling=true;");
}
public string Host;
@@ -52,10 +48,18 @@ namespace Framework.Database
public abstract class MySqlBase<T>
{
Dictionary<T, string> _preparedQueries = new Dictionary<T, string>();
ProducerConsumerQueue<ISqlOperation> _queue = new ProducerConsumerQueue<ISqlOperation>();
MySqlConnectionInfo _connectionInfo;
DatabaseUpdater<T> _updater;
DatabaseWorker<T> _worker;
public MySqlErrorCode Initialize(MySqlConnectionInfo connectionInfo)
{
_connectionInfo = connectionInfo;
_updater = new DatabaseUpdater<T>(this);
_worker = new DatabaseWorker<T>(_queue, this);
try
{
@@ -67,16 +71,17 @@ namespace Framework.Database
}
}
catch (MySqlException ex)
{
{
return HandleMySQLException(ex);
}
}
public void Execute(string sql, params object[] args)
public bool DirectExecute(string sql, params object[] args)
{
Execute(new PreparedStatement(string.Format(sql, args)));
return DirectExecute(new PreparedStatement(string.Format(sql, args)));
}
public void Execute(PreparedStatement stmt)
public bool DirectExecute(PreparedStatement stmt)
{
try
{
@@ -90,15 +95,28 @@ namespace Framework.Database
cmd.Parameters.AddWithValue("@" + parameter.Key, parameter.Value);
cmd.ExecuteNonQuery();
return true;
}
}
}
catch (MySqlException ex)
{
HandleMySQLException(ex, stmt.CommandText);
return false;
}
}
public void Execute(string sql, params object[] args)
{
Execute(new PreparedStatement(string.Format(sql, args)));
}
public void Execute(PreparedStatement stmt)
{
PreparedStatementTask task = new PreparedStatementTask(stmt);
_queue.Push(task);
}
public void ExecuteOrAppend(SQLTransaction trans, PreparedStatement stmt)
{
if (trans == null)
@@ -114,141 +132,41 @@ namespace Framework.Database
public SQLResult Query(PreparedStatement stmt)
{
List<object[]> rows = new List<object[]>();
try
{
using (var Connection = _connectionInfo.GetConnection())
{
Connection.Open();
using (MySqlCommand cmd = Connection.CreateCommand())
{
MySqlConnection Connection = _connectionInfo.GetConnection();
Connection.Open();
cmd.CommandText = stmt.CommandText;
foreach (var parameter in stmt.Parameters)
cmd.Parameters.AddWithValue("@" + parameter.Key, parameter.Value);
MySqlCommand cmd = Connection.CreateCommand();
cmd.CommandText = stmt.CommandText;
foreach (var parameter in stmt.Parameters)
cmd.Parameters.AddWithValue("@" + parameter.Key, parameter.Value);
using (var reader = cmd.ExecuteReader())
{
if (reader.HasRows)
{
while(reader.Read())
{
var row = new object[reader.FieldCount];
reader.GetValues(row);
rows.Add(row);
}
}
}
}
}
return new SQLResult(cmd.ExecuteReader(System.Data.CommandBehavior.CloseConnection));
}
catch (MySqlException ex)
{
HandleMySQLException(ex, stmt.CommandText);
return new SQLResult();
}
return new SQLResult(rows);
}
public QueryCallback AsyncQuery(string sql, params object[] args)
{
return AsyncQuery(new PreparedStatement(string.Format(sql, args)));
}
public QueryCallback AsyncQuery(PreparedStatement stmt)
{
return new QueryCallback(_AsyncQuery(stmt));
PreparedStatementTask task = new PreparedStatementTask(stmt, true);
// Store future result before enqueueing - task might get already processed and deleted before returning from this method
Task<SQLResult> result = task.GetFuture();
_queue.Push(task);
return new QueryCallback(result);
}
async Task<SQLResult> _AsyncQuery(PreparedStatement stmt)
public Task<SQLQueryHolder<R>> DelayQueryHolder<R>(SQLQueryHolder<R> holder)
{
List<object[]> rows = new List<object[]>();
try
{
using (var Connection = _connectionInfo.GetConnection())
{
await Connection.OpenAsync();
using (MySqlCommand cmd = Connection.CreateCommand())
{
cmd.CommandText = stmt.CommandText;
foreach (var parameter in stmt.Parameters)
cmd.Parameters.AddWithValue("@" + parameter.Key, parameter.Value);
using (var reader = await cmd.ExecuteReaderAsync())
{
if (reader.HasRows)
{
while (await reader.ReadAsync())
{
var row = new object[reader.FieldCount];
reader.GetValues(row);
rows.Add(row);
}
}
}
}
}
}
catch (MySqlException ex)
{
HandleMySQLException(ex, stmt.CommandText);
}
return new SQLResult(rows);
}
public async Task<SQLQueryHolder<R>> DelayQueryHolder<R>(SQLQueryHolder<R> holder)
{
return await Task.Run(() =>
{
string query = "";
try
{
using (var Connection = _connectionInfo.GetConnection())
{
Connection.OpenAsync();
foreach (var pair in holder.m_queries)
{
List<object[]> rows = new List<object[]>();
using (MySqlCommand cmd = Connection.CreateCommand())
{
cmd.CommandText = pair.Value.stmt.CommandText;
foreach (var parameter in pair.Value.stmt.Parameters)
cmd.Parameters.AddWithValue("@" + parameter.Key, parameter.Value);
query = cmd.CommandText;
using (var reader = cmd.ExecuteReader())
{
if (reader.HasRows)
{
while (reader.Read())
{
var row = new object[reader.FieldCount];
reader.GetValues(row);
rows.Add(row);
}
}
}
}
holder.SetResult(pair.Key, new SQLResult(rows));
}
}
return holder;
}
catch (MySqlException ex)
{
HandleMySQLException(ex, query);
return holder;
}
});
SQLQueryHolderTask<R> task = new SQLQueryHolderTask<R>(holder);
// Store future result before enqueueing - task might get already processed and deleted before returning from this method
Task<SQLQueryHolder<R>> result = task.GetFuture();
_queue.Push(task);
return result;
}
public void LoadPreparedStatements()
@@ -268,12 +186,12 @@ namespace Framework.Database
sb.Append(sql[i]);
}
_queries[statement] = sb.ToString();
_preparedQueries[statement] = sb.ToString();
}
public PreparedStatement GetPreparedStatement(T statement)
{
return new PreparedStatement(_queries[statement]);
return new PreparedStatement(_preparedQueries[statement]);
}
public bool Apply(string sql)
@@ -282,9 +200,9 @@ namespace Framework.Database
{
using (var Connection = _connectionInfo.GetConnectionNoDatabase())
{
Connection.Open();
using (MySqlCommand cmd = Connection.CreateCommand())
{
Connection.Open();
cmd.CommandText = sql;
cmd.ExecuteNonQuery();
return true;
@@ -302,12 +220,16 @@ namespace Framework.Database
{
try
{
string query = File.ReadAllText(path);
if (query.Length > 1048576) //Default size limit of querys
Apply("SET GLOBAL max_allowed_packet=1073741824;");
using (var connection = _connectionInfo.GetConnection())
{
connection.Open();
using (MySqlCommand cmd = connection.CreateCommand())
{
connection.Open();
cmd.CommandText = File.ReadAllText(path);
cmd.CommandText = query;
cmd.ExecuteNonQuery();
return true;
}
@@ -326,6 +248,11 @@ namespace Framework.Database
}
public void CommitTransaction(SQLTransaction transaction)
{
_queue.Push(new TransactionTask(transaction));
}
public bool DirectCommitTransaction(SQLTransaction transaction)
{
using (var Connection = _connectionInfo.GetConnection())
{
@@ -349,11 +276,13 @@ namespace Framework.Database
trans.Commit();
scope.Complete();
}
return true;
}
catch (MySqlException ex) //error occurred
{
HandleMySQLException(ex, query);
trans.Rollback();
return false;
}
}
}
@@ -407,9 +336,5 @@ namespace Framework.Database
}
public abstract void PreparedStatements();
Dictionary<T, string> _queries = new Dictionary<T, string>();
MySqlConnectionInfo _connectionInfo;
DatabaseUpdater<T> _updater;
}
}
Source/Framework/Database/PreparedStatement.cs
+38 -2
View File
@@ -16,11 +16,15 @@
*/
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Framework.Database
{
public class PreparedStatement
{
public string CommandText;
public Dictionary<int, object> Parameters = new Dictionary<int, object>();
public PreparedStatement(string commandText)
{
CommandText = commandText;
@@ -35,8 +39,40 @@ namespace Framework.Database
{
Parameters.Clear();
}
}
public string CommandText;
public Dictionary<int, object> Parameters = new Dictionary<int, object>();
public class PreparedStatementTask : ISqlOperation
{
PreparedStatement m_stmt;
bool _needsResult;
TaskCompletionSource<SQLResult> m_result;
public PreparedStatementTask(PreparedStatement stmt, bool needsResult = false)
{
m_stmt = stmt;
_needsResult = needsResult;
if (needsResult)
m_result = new TaskCompletionSource<SQLResult>();
}
public bool Execute<T>(MySqlBase<T> mySqlBase)
{
if (_needsResult)
{
SQLResult result = mySqlBase.Query(m_stmt);
if (result == null)
{
m_result.SetResult(new SQLResult());
return false;
}
m_result.SetResult(result);
return true;
}
return mySqlBase.DirectExecute(m_stmt);
}
public Task<SQLResult> GetFuture() { return m_result.Task; }
}
}
Source/Framework/Database/SQLQueryHolder.cs
+38 -21
View File
@@ -16,45 +16,62 @@
*/
using System.Collections.Generic;
using System.Threading.Tasks;
namespace Framework.Database
{
public class SQLQueryHolder<T>
{
public void SetQuery(T index, PreparedStatement _stmt)
{
m_queries[index] = new SQLResultPair(_stmt);
}
public Dictionary<T, PreparedStatement> m_queries = new Dictionary<T, PreparedStatement>();
Dictionary<T, SQLResult> _results = new Dictionary<T, SQLResult>();
public void SetQuery(T index, string sql, params object[] args)
{
SetQuery(index, new PreparedStatement(string.Format(sql, args)));
}
public void SetResult(T index, SQLResult _result)
public void SetQuery(T index, PreparedStatement stmt)
{
m_queries[index].result = _result;
m_queries[index] = stmt;
}
public void SetResult(T index, SQLResult result)
{
_results[index] = result;
}
public SQLResult GetResult(T index)
{
if (!m_queries.ContainsKey(index))
if (!_results.ContainsKey(index))
return new SQLResult();
return m_queries[index].result;
}
public Dictionary<T, SQLResultPair> m_queries = new Dictionary<T, SQLResultPair>();
public class SQLResultPair
{
public SQLResultPair(PreparedStatement _stmt)
{
stmt = _stmt;
}
public PreparedStatement stmt;
public SQLResult result;
return _results[index];
}
}
class SQLQueryHolderTask<R> : ISqlOperation
{
SQLQueryHolder<R> m_holder;
TaskCompletionSource<SQLQueryHolder<R>> m_result;
public SQLQueryHolderTask(SQLQueryHolder<R> holder)
{
m_holder = holder;
m_result = new TaskCompletionSource<SQLQueryHolder<R>>();
}
public bool Execute<T>(MySqlBase<T> mySqlBase)
{
if (m_holder == null)
return false;
// execute all queries in the holder and pass the results
foreach (var pair in m_holder.m_queries)
m_holder.SetResult(pair.Key, mySqlBase.Query(pair.Value));
return m_result.TrySetResult(m_holder);
}
public Task<SQLQueryHolder<R>> GetFuture() { return m_result.Task; }
}
}
Source/Framework/Database/SQLResult.cs
+37 -43
View File
@@ -15,81 +15,77 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
using MySql.Data.MySqlClient;
using System;
using System.Collections.Generic;
namespace Framework.Database
{
public class SQLResult
{
public SQLResult()
MySqlDataReader _reader;
public SQLResult() { }
public SQLResult(MySqlDataReader reader)
{
_rows = new List<object[]>();
_reader = reader;
NextRow();
}
public SQLResult(List<object[]> values)
~SQLResult()
{
_rows = values;
_reader = null;
}
public T Read<T>(int column)
{
var value = _rows[_rowIndex][column];
var value = _reader[column];
if (value.GetType() == typeof(T))
return (T)value;
if (value == DBNull.Value)
return default(T);
if (value != DBNull.Value)
return (T)Convert.ChangeType(value, typeof(T));
if (value.GetType() != typeof(T))
return (T)Convert.ChangeType(value, typeof(T));//todo remove me when all fields are the right type this is super slow
if (typeof(T).Name == "String")
return (T)Convert.ChangeType("", typeof(T));
return default(T);
}
public T[] ReadValues<T>(int startIndex, int numColumns)
{
T[] values = new T[numColumns];
for (var c = 0; c < numColumns; ++c)
values[c] = Read<T>(startIndex + c);
return values;
return (T)value;
}
public bool IsNull(int column)
{
return _rows[_rowIndex][column] == DBNull.Value;
return _reader.IsDBNull(column);
}
public int GetRowCount() { return _rows.Count; }
public int GetFieldCount() { return _reader.FieldCount; }
public bool IsEmpty() { return GetRowCount() == 0; }
public bool IsEmpty()
{
return _reader == null || !_reader.HasRows;
}
public SQLFields GetFields() { return new SQLFields(_rows[_rowIndex]); }
public SQLFields GetFields()
{
object[] values = new object[_reader.FieldCount];
_reader.GetValues(values);
return new SQLFields(values);
}
public bool NextRow()
{
if (_rowIndex >= GetRowCount() - 1)
{
_rowIndex = 0;
if (_reader == null)
return false;
}
_rowIndex++;
return true;
if (_reader.Read())
return true;
_reader.Close();
return false;
}
public void ResetRowIndex()
{
_rowIndex = 0;
}
private int _rowIndex;
List<object[]> _rows;
}
public class SQLFields
{
object[] _currentRow;
public SQLFields(object[] row) { _currentRow = row; }
public T Read<T>(int column)
@@ -116,7 +112,5 @@ namespace Framework.Database
return values;
}
object[] _currentRow;
}
}
Source/Framework/Database/SQLTransaction.cs
+15
View File
@@ -43,4 +43,19 @@ namespace Framework.Database
commands.Add(new MySqlCommand(string.Format(sql, args)));
}
}
class TransactionTask : ISqlOperation
{
public TransactionTask(SQLTransaction trans)
{
m_trans = trans;
}
public bool Execute<T>(MySqlBase<T> mySqlBase)
{
return mySqlBase.DirectCommitTransaction(m_trans);
}
SQLTransaction m_trans;
}
}
Source/Framework/Realm/RealmManager.cs
+1 -2
View File
@@ -59,7 +59,6 @@ public class RealmManager : Singleton<RealmManager>
{
PreparedStatement stmt = DB.Login.GetPreparedStatement(LoginStatements.SEL_REALMLIST);
SQLResult result = DB.Login.Query(stmt);
Dictionary<RealmHandle, string> existingRealms = new Dictionary<RealmHandle, string>();
foreach (var p in _realms)
existingRealms[p.Key] = p.Value.Name;
@@ -285,7 +284,7 @@ public class RealmManager : Singleton<RealmManager>
stmt.AddValue(2, locale);
stmt.AddValue(3, os);
stmt.AddValue(4, accountName);
DB.Login.Execute(stmt);
DB.Login.DirectExecute(stmt);
Bgs.Protocol.Attribute attribute = new Bgs.Protocol.Attribute();
attribute.Name = "Param_RealmJoinTicket";