rendered paste bodyIndex: DatabasePostgre.cpp
===================================================================
--- DatabasePostgre.cpp (revision 5278)
+++ DatabasePostgre.cpp (working copy)
@@ -23,11 +23,10 @@
#include "Platform/Define.h"
#include "../src/zthread/ThreadImpl.h"
#include "DatabaseEnv.h"
+#include "Database/PGSQLDelayThread.h"
#include "Database/SqlOperations.h"
#include "Timer.h"
-using namespace std;
-
void DatabasePostgre::ThreadStart()
{
}
@@ -38,12 +37,26 @@
size_t DatabasePostgre::db_count = 0;
-DatabasePostgre::DatabasePostgre() : Database()
+DatabasePostgre::DatabasePostgre() : Database(), mPGconn(NULL)
{
+ // before first connection
+ if( db_count++ == 0 )
+ {
+
+ if (!PQisthreadsafe())
+ {
+ sLog.outError("FATAL ERROR: PostgreSQL libpq isn't thread-safe.");
+ exit(1);
+ }
+ }
}
DatabasePostgre::~DatabasePostgre()
{
+
+ if (m_delayThread)
+ HaltDelayThread();
+
if( mPGconn )
{
PQfinish(mPGconn);
@@ -58,10 +71,12 @@
tranThread = NULL;
- vector<string> tokens = StrSplit(infoString, ";");
+ InitDelayThread();
- vector<string>::iterator iter;
+ Tokens tokens = StrSplit(infoString, ";");
+ Tokens::iterator iter;
+
std::string host, port_or_socket, user, password, database;
iter = tokens.begin();
@@ -91,6 +106,7 @@
{
sLog.outDetail( "Connected to Postgre database at %s\n",
host.c_str());
+ sLog.outDetail( "PostgreSQL server ver: %d",PQserverVersion(mPGconn));
return true;
}
@@ -104,18 +120,22 @@
uint64 rowCount = 0;
uint32 fieldCount = 0;
- // guarded block for thread-safe mySQL request
- ZThread::Guard<ZThread::FastMutex> query_connection_guard((ZThread::ThreadImpl::current()==tranThread?tranMutex:mMutex));
+ // guarded block for thread-safe request
+ ZThread::Guard<ZThread::FastMutex> query_connection_guard(mMutex);
#ifdef MANGOS_DEBUG
uint32 _s = getMSTime();
#endif
// Send the query
PGresult * result = PQexec(mPGconn, sql);
+ if (!result )
+ {
+ return NULL;
+ }
- if (PQresultStatus(result) != PGRES_COMMAND_OK)
+ if (PQresultStatus(result) != PGRES_TUPLES_OK)
{
- sLog.outErrorDb( "SQL: %s", sql );
- sLog.outErrorDb("query ERROR: %s", PQerrorMessage(mPGconn));
+ sLog.outErrorDb( "SQL : %s", sql );
+ sLog.outErrorDb( "SQL %s", PQerrorMessage(mPGconn));
PQclear(result);
return NULL;
}
@@ -130,21 +150,13 @@
fieldCount = PQnfields(result);
// end guarded block
- if (!result )
- {
- PQclear(result);
- return NULL;
- }
-
if (!rowCount)
{
PQclear(result);
return NULL;
}
- QueryResultPostgre * queryResult = new QueryResultPostgre(result, fieldCount, rowCount);
- return queryResult;
-
+ QueryResultPostgre * queryResult = new QueryResultPostgre(result, rowCount, fieldCount);
queryResult->NextRow();
return queryResult;
@@ -152,20 +164,43 @@
bool DatabasePostgre::Execute(const char *sql)
{
+
if (!mPGconn)
return false;
+ // don't use queued execution if it has not been initialized
+ if (!m_threadBody) return DirectExecute(sql);
+
+ tranThread = ZThread::ThreadImpl::current(); // owner of this transaction
+ TransactionQueues::iterator i = m_tranQueues.find(tranThread);
+ if (i != m_tranQueues.end() && i->second != NULL)
+ { // Statement for transaction
+ i->second->DelayExecute(sql);
+ }
+ else
{
- // guarded block for thread-safe mySQL request
- ZThread::Guard<ZThread::FastMutex> query_connection_guard((ZThread::ThreadImpl::current()==tranThread?tranMutex:mMutex));
+ // Simple sql statement
+ m_threadBody->Delay(new SqlStatement(sql));
+ }
+
+ return true;
+}
+
+bool DatabasePostgre::DirectExecute(const char* sql)
+{
+ if (!mPGconn)
+ return false;
+ {
+ // guarded block for thread-safe request
+ ZThread::Guard<ZThread::FastMutex> query_connection_guard(mMutex);
#ifdef MANGOS_DEBUG
uint32 _s = getMSTime();
#endif
PGresult *res = PQexec(mPGconn, sql);
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
- sLog.outErrorDb("SQL: %s", sql);
- sLog.outErrorDb("SQL ERROR: %s", PQerrorMessage(mPGconn));
+ sLog.outErrorDb( "SQL: %s", sql );
+ sLog.outErrorDb( "SQL %s", PQerrorMessage(mPGconn) );
return false;
}
else
@@ -178,15 +213,9 @@
// end guarded block
}
-
return true;
}
-bool DatabasePostgre::DirectExecute(const char* sql)
-{
- return true;
-}
-
bool DatabasePostgre::_TransactionCmd(const char *sql)
{
if (!mPGconn)
@@ -210,16 +239,30 @@
{
if (!mPGconn)
return false;
- if (tranThread==ZThread::ThreadImpl::current())
- return false; // huh? this thread already started transaction
- mMutex.acquire();
- if (!_TransactionCmd("START TRANSACTION"))
+ // don't use queued execution if it has not been initialized
+ if (!m_threadBody)
{
- mMutex.release(); // can't start transaction
- return false;
+
+ if (tranThread==ZThread::ThreadImpl::current())
+ return false; // huh? this thread already started transaction
+ mMutex.acquire();
+ if (!_TransactionCmd("START TRANSACTION"))
+ {
+ mMutex.release(); // can't start transaction
+ return false;
+ }
+ return true;
}
// transaction started
tranThread = ZThread::ThreadImpl::current(); // owner of this transaction
+ TransactionQueues::iterator i = m_tranQueues.find(tranThread);
+ if (i != m_tranQueues.end() && i->second != NULL)
+ // If for thread exists queue and also contains transaction
+ // delete that transaction (not allow trans in trans)
+ delete i->second;
+
+ m_tranQueues[tranThread] = new SqlTransaction();
+
return true;
}
@@ -227,24 +270,55 @@
{
if (!mPGconn)
return false;
- if (tranThread!=ZThread::ThreadImpl::current())
+
+ // don't use queued execution if it has not been initialized
+ if (!m_threadBody)
+ {
+
+ if (tranThread!=ZThread::ThreadImpl::current())
+ return false;
+ bool _res = _TransactionCmd("COMMIT");
+ tranThread = NULL;
+ mMutex.release();
+ return _res;
+ }
+ tranThread = ZThread::ThreadImpl::current();
+ TransactionQueues::iterator i = m_tranQueues.find(tranThread);
+ if (i != m_tranQueues.end() && i->second != NULL)
+ {
+ m_threadBody->Delay(i->second);
+ i->second = NULL;
+ return true;
+ }
+ else
return false;
- bool _res = _TransactionCmd("COMMIT");
- tranThread = NULL;
- mMutex.release();
- return _res;
+
}
bool DatabasePostgre::RollbackTransaction()
{
if (!mPGconn)
return false;
- if (tranThread!=ZThread::ThreadImpl::current())
- return false;
- bool _res = _TransactionCmd("ROLLBACK");
- tranThread = NULL;
- mMutex.release();
- return _res;
+ // don't use queued execution if it has not been initialized
+ if (!m_threadBody)
+ {
+
+ if (tranThread!=ZThread::ThreadImpl::current())
+ return false;
+ bool _res = _TransactionCmd("ROLLBACK");
+ tranThread = NULL;
+ mMutex.release();
+ return _res;
+ }
+ tranThread = ZThread::ThreadImpl::current();
+ TransactionQueues::iterator i = m_tranQueues.find(tranThread);
+ if (i != m_tranQueues.end() && i->second != NULL)
+ {
+ delete i->second;
+ i->second = NULL;
+ }
+ return true;
+
}
unsigned long DatabasePostgre::escape_string(char *to, const char *from, unsigned long length)
@@ -257,9 +331,21 @@
void DatabasePostgre::InitDelayThread()
{
+ assert(!m_delayThread);
+
+ //New delay thread for delay execute
+ m_delayThread = new ZThread::Thread(m_threadBody = new PGSQLDelayThread(this));
}
void DatabasePostgre::HaltDelayThread()
{
+ if (!m_threadBody || !m_delayThread) return;
+
+ m_threadBody->Stop(); //Stop event
+ m_delayThread->wait(); //Wait for flush to DB
+ delete m_delayThread; //This also deletes m_threadBody
+ m_delayThread = NULL;
+ m_threadBody = NULL;
}
+
#endif
Index: DatabasePostgre.h
===================================================================
--- DatabasePostgre.h (revision 5278)
+++ DatabasePostgre.h (working copy)
@@ -26,7 +26,7 @@
#ifdef WIN32
#define FD_SETSIZE 1024
#include <winsock2.h>
-#include <postgre/libpq/libpq-fe.h>
+#include <postgre/libpq-fe.h>
#else
#include <libpq-fe.h>
#endif
Index: Field.cpp
===================================================================
--- Field.cpp (revision 5278)
+++ Field.cpp (working copy)
@@ -67,7 +67,7 @@
{
if(mValue) delete [] mValue;
- if (value)
+ if (strlen(value))
{
mValue = new char[strlen(value) + 1];
strcpy(mValue, value);
@@ -80,7 +80,7 @@
{
if(mName) delete [] mName;
- if (name)
+ if (strlen(name))
{
mName = new char[strlen(name) + 1];
strcpy(mName, name);
Index: Makefile.am
===================================================================
--- Makefile.am (revision 5278)
+++ Makefile.am (working copy)
@@ -44,6 +44,7 @@
Field.cpp \
Field.h \
MySQLDelayThread.h \
+ PGSQLDelayThread.h \
QueryResult.h \
QueryResultMysql.cpp \
QueryResultMysql.h \
Index: PGSQLDelayThread.h
===================================================================
--- PGSQLDelayThread.h (revision 0)
+++ PGSQLDelayThread.h (revision 0)
@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2005-2008 MaNGOS <http://www.mangosproject.org/>
+ *
+ * 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 2 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, write to the Free Software
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+ */
+
+#ifndef __PGSQLDELAYTHREAD_H
+#define __PGSQLDELAYTHREAD_H
+
+#include "Database/SqlDelayThread.h"
+
+class PGSQLDelayThread : public SqlDelayThread
+{
+ public:
+ PGSQLDelayThread(Database* db) : SqlDelayThread(db) {}
+ void Stop() { SqlDelayThread::Stop(); }
+};
+#endif //__PGSQLDELAYTHREAD_H
Index: QueryResultPostgre.cpp
===================================================================
--- QueryResultPostgre.cpp (revision 5278)
+++ QueryResultPostgre.cpp (working copy)
@@ -21,7 +21,7 @@
#include "DatabaseEnv.h"
QueryResultPostgre::QueryResultPostgre(PGresult *result, uint64 rowCount, uint32 fieldCount) :
-QueryResult(rowCount, fieldCount), mResult(result)
+QueryResult(rowCount, fieldCount), mResult(result), mTableIndex(0)
{
mCurrentRow = new Field[mFieldCount];
@@ -30,8 +30,7 @@
for (uint32 i = 0; i < mFieldCount; i++)
{
mCurrentRow[i].SetName(PQfname(result, i));
- //mCurrentRow[i].SetType(ConvertNativeType(PQftype( result, i )));
- mCurrentRow[i].SetType(Field::DB_TYPE_UNKNOWN);
+ mCurrentRow[i].SetType(ConvertNativeType(PQftype( result, i )));
}
}
@@ -50,12 +49,11 @@
EndQuery();
return false;
}
-
for (int j = 0; j < mFieldCount; j++)
{
mCurrentRow[j].SetValue(PQgetvalue(mResult, mTableIndex, j));
}
-
+ mTableIndex++;
return true;
}
@@ -77,34 +75,37 @@
// dummy -> all is set to "uknown"
enum Field::DataTypes QueryResultPostgre::ConvertNativeType(Oid pOid ) const
{
+
+ /// TODO: need Fix This!!!
+ /*
switch (pOid)
{
-
- // temporarry commented out until proper resolution found
- // case FIELD_TYPE_DATE:
- // case FIELD_TYPE_TIME:
- // case FIELD_TYPE_DATETIME:
- // case FIELD_TYPE_YEAR:
- // case FIELD_TYPE_STRING:
- // case FIELD_TYPE_VAR_STRING:
- // case FIELD_TYPE_BLOB:
- // case FIELD_TYPE_SET:
- // case FIELD_TYPE_NULL:
- // return Field::DB_TYPE_STRING;
- // case FIELD_TYPE_TINY:
-
- // case FIELD_TYPE_SHORT:
- // case FIELD_TYPE_LONG:
- // case FIELD_TYPE_INT24:
- // case FIELD_TYPE_LONGLONG:
- // case FIELD_TYPE_ENUM:
- // return Field::DB_TYPE_INTEGER;
- // case FIELD_TYPE_DECIMAL:
- // case FIELD_TYPE_FLOAT:
- // case FIELD_TYPE_DOUBLE:
- // return Field::DB_TYPE_FLOAT;
+ // see types in #include <postgre/pg_type.h>
+ case BOOLOID:
+ case NUMERICOID:
+ case INT8OID:
+ case INT4OID:
+ case INT2OID:
+ case OIDOID:
+ case TEXTOID:
+ case BYTEAOID:
+ case CHAROID:
+ case NAMEOID:
+ case CSTRINGARRAYOID:
+ case BPCHAROID:
+ case VARCHAROID:
+ case FLOAT4OID:
+ case FLOAT8OID:
+ case DATEOID:
+ case TIMEOID:
+ case TIMESTAMPOID:
+ case TIMESTAMPTZOID:
+ case UNKNOWNOID:
default:
return Field::DB_TYPE_UNKNOWN;
+
}
+ */
+ return Field::DB_TYPE_UNKNOWN;
}
#endif
Index: QueryResultPostgre.h
===================================================================
--- QueryResultPostgre.h (revision 5278)
+++ QueryResultPostgre.h (working copy)
@@ -22,9 +22,11 @@
#ifdef WIN32
#define FD_SETSIZE 1024
#include <winsock2.h>
-#include <postgre/libpq/libpq-fe.h>
+#include <postgre/libpq-fe.h>
+//#include <postgre/pg_type.h>
#else
#include <libpq-fe.h>
+//#include <pg_type.h>
#endif
class QueryResultPostgre : public QueryResult
Index: SqlDelayThread.cpp
===================================================================
--- SqlDelayThread.cpp (revision 5278)
+++ SqlDelayThread.cpp (working copy)
@@ -16,8 +16,6 @@
* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
*/
-#ifndef DO_POSTGRESQL
-
#include "Database/SqlDelayThread.h"
#include "Database/SqlOperations.h"
#include "DatabaseEnv.h"
@@ -29,8 +27,9 @@
void SqlDelayThread::run()
{
SqlOperation* s;
-
+ #ifndef DO_POSTGRESQL
mysql_thread_init();
+ #endif
while (m_running)
{
@@ -45,11 +44,12 @@
}
}
+ #ifndef DO_POSTGRESQL
mysql_thread_end();
+ #endif
}
void SqlDelayThread::Stop()
{
m_running = false;
}
-#endif