Index: 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 tokens = StrSplit(infoString, ";"); + InitDelayThread(); - vector::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 query_connection_guard((ZThread::ThreadImpl::current()==tranThread?tranMutex:mMutex)); + // guarded block for thread-safe request + ZThread::Guard 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 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 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 -#include +#include #else #include #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 + * + * 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 + 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 -#include +#include +//#include #else #include +//#include #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