SVN Commit by dpage: r4219 - in trunk/pgadmin3/xtra/pgagent: . include

Author: dpage
Date: 2005-05-19 15:53:32 +0100 (Thu, 19 May 2005)
New Revision: 4219

Modified:
trunk/pgadmin3/xtra/pgagent/connection.cpp
trunk/pgadmin3/xtra/pgagent/include/connection.h
trunk/pgadmin3/xtra/pgagent/include/job.h
trunk/pgadmin3/xtra/pgagent/include/pgAgent.h
trunk/pgadmin3/xtra/pgagent/job.cpp
trunk/pgadmin3/xtra/pgagent/misc.cpp
trunk/pgadmin3/xtra/pgagent/pgAgent.cpp
trunk/pgadmin3/xtra/pgagent/pgAgent.dsp
trunk/pgadmin3/xtra/pgagent/unix.cpp
trunk/pgadmin3/xtra/pgagent/win32.cpp
Log:
Dynamic connection pooling in preparation for threaded operation.

Modified: trunk/pgadmin3/xtra/pgagent/connection.cpp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/connection.cpp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/connection.cpp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -13,18 +13,15 @@
#include <libpq-fe.h>
#include <time.h>

-
-// entries in the connection pool
-int connPoolCount=5;
-
-
-DBconn **DBconn::pool=0;
+DBconn *DBconn::primaryConn;
string DBconn::basicConnectString;

-
DBconn::DBconn(const string &name)
{
dbname = name;
+ inUse = false;
+ next=0;
+ prev=0;
Connect(basicConnectString + " dbname=" + dbname);
}

@@ -32,6 +29,9 @@
DBconn::DBconn(const string &connectString, const string &name)
{
dbname = name;
+ inUse = false;
+ next=0;
+ prev=0;
Connect(connectString);
}

@@ -65,21 +65,12 @@

DBconn *DBconn::InitConnection(const string &connectString)
{
- if (!pool)
- {
- pool = new DBconn*[connPoolCount];
- if (pool)
- memset(pool, 0, sizeof(DBconn*) * connPoolCount);
- }
- if (!pool)
- LogMessage("Out of memory for connection pool", LOG_ERROR);
-
basicConnectString=connectString;
string dbname;

int pos=basicConnectString.find("dbname=");
if (pos == -1)
- dbname = "dba";
+ dbname = "pgadmin";
else
{
dbname = basicConnectString.substr(pos+7);
@@ -93,89 +84,84 @@
dbname = dbname.substr(0, pos);
}
}
- pool[0] = new DBconn(connectString, dbname);
- pool[0]->primary = true;
+ primaryConn = new DBconn(connectString, dbname);

- return pool[0];
+ if (!primaryConn)
+ LogMessage("Failed to create primary connection!", LOG_ERROR);
+
+ primaryConn->inUse = true;
+
+ return primaryConn;
}


-DBconn *DBconn::Get(const string &dbname, bool asPrimary)
+DBconn *DBconn::Get(const string &dbname)
{
- if (!pool)
- {
- pool = new DBconn*[connPoolCount];
- if (pool)
- memset(pool, 0, sizeof(DBconn*) * connPoolCount);
- }
- if (!pool)
- LogMessage("Out of memory for connection pool", LOG_ERROR);
+ DBconn *thisConn = primaryConn, *testConn;

- int i;
- DBconn **emptyConn=0, **oldestConn=0;
-
// find an existing connection
- for (i=0 ; i < connPoolCount ; i++)
+ do
{
- if (pool[i])
- {
- if (dbname == pool[i]->dbname)
- return pool[i];
+ if (dbname == thisConn->dbname && !thisConn->inUse)
+ {
+ LogMessage("Allocating existing connection to database " + thisConn->dbname, LOG_DEBUG);
+ thisConn->inUse = true;
+ return thisConn;
+ }

- // while searching, also mark the oldest non-primary connection
- if (!pool[i]->primary)
- {
- if (!oldestConn || pool[i]->timestamp < (*oldestConn)->timestamp)
- oldestConn=pool+i;
- }
- }
- else
- {
- // while searching, mark the first empty slot
- if (!emptyConn)
- emptyConn=pool+i;
- }
- }
- if (!emptyConn)
- {
- delete *oldestConn;
- emptyConn=oldestConn;
- *emptyConn=0;
- }
+ testConn = thisConn;
+ if (thisConn->next != 0)
+ thisConn = thisConn->next;

- DBconn *conn=new DBconn(dbname);
- if (conn->conn)
+ } while (testConn->next != 0);
+
+
+ // No suitable connection was found, so create a new one.
+ DBconn *newConn=new DBconn(dbname);
+ if (newConn->conn)
{
- *emptyConn=conn;
+ LogMessage("Allocating new connection to database " + newConn->dbname, LOG_DEBUG);
+ newConn->inUse = true;
+ newConn->prev = thisConn;
+ thisConn->next = newConn;
}
- return *emptyConn;
+ else
+ LogMessage("Failed to create new connection to database: " + dbname, LOG_ERROR);
+
+ return newConn;
}

+void DBconn::Return()
+{
+ LogMessage("Returning connection to database " + this->dbname, LOG_DEBUG);
+ inUse = false;
+}

void DBconn::ClearConnections(bool all)
{
- // clears all connections, except for the primary one.
- // if all is true, even the primary connection will be killed.
- if (pool)
- {
- int i;
- for (i=0 ; i < connPoolCount ; i++)
- {
- if (pool[i])
- {
- if (all || !pool[i]->primary)
- {
- delete pool[i];
- pool[i]=0;
- }
- }
- }
- if (all)
- {
- delete[] pool;
- pool=0;
- }
- }
+ if (all)
+ LogMessage("Clearing all connections", LOG_DEBUG);
+ else
+ LogMessage("Clearing all connections except the primary", LOG_DEBUG);
+
+ DBconn *thisConn=primaryConn, *deleteConn;
+
+ // Find the last connection
+ while (thisConn->next != 0)
+ thisConn = thisConn->next;
+
+ // Delete connections as required
+ while (thisConn->prev != 0)
+ {
+ deleteConn = thisConn;
+ thisConn = deleteConn->prev;
+ delete deleteConn;
+ thisConn->next = 0;
+ }
+
+ if (all)
+ delete thisConn;
+
}


@@ -205,6 +191,18 @@
return rows;
}

+string DBconn::GetLastError()
+{
+ // Return the last error message, minus any trailing line ends
+ if (lastError.substr(lastError.length()-2, 2) == "\r\n") // DOS
+ return lastError.substr(0, lastError.length()-2);
+ else if (lastError.substr(lastError.length()-1, 1) == "\n") // Unix
+ return lastError.substr(0, lastError.length()-1);
+ else if (lastError.substr(lastError.length()-1, 1) == "\r") // Mac
+ return lastError.substr(0, lastError.length()-1);
+ else
+ return lastError;
+}

///////////////////////////////////////////////////////7

Modified: trunk/pgadmin3/xtra/pgagent/include/connection.h
===================================================================
--- trunk/pgadmin3/xtra/pgagent/include/connection.h 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/include/connection.h 2005-05-19 14:53:32 UTC (rev 4219)
@@ -16,7 +16,6 @@
#include <libpq-fe.h>

class DBresult;
-extern int connPoolCount;


class DBconn
@@ -27,29 +26,31 @@
~DBconn();

public:
- static DBconn *Get(const string &dbname, bool asPrimary=false);
+ static DBconn *Get(const string &dbname);
static DBconn *InitConnection(const string &connectString);

static void ClearConnections(bool allIncludingPrimary=false);
static void SetBasicConnectString(const string &bcs) { basicConnectString = bcs; }
- string GetLastError() { return lastError; }
+
+ string GetLastError();
string GetDBname() {return dbname; }
bool IsValid() { return conn != 0; }

DBresult *Execute(const string &query);
int ExecuteVoid(const string &query);
+ void Return();

private:
bool DBconn::Connect(const string &connectString);

protected:
static string basicConnectString;
- static DBconn **pool;
+ static DBconn *primaryConn;

string dbname, lastError;
PGconn *conn;
- long timestamp;
- bool primary;
+ DBconn *next, *prev;
+ bool inUse;

friend class DBresult;

Modified: trunk/pgadmin3/xtra/pgagent/include/job.h
===================================================================
--- trunk/pgadmin3/xtra/pgagent/include/job.h 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/include/job.h 2005-05-19 14:53:32 UTC (rev 4219)
@@ -23,7 +23,7 @@
bool Runnable() { return status == "r"; }

protected:
- DBconn *serviceConn;
+ DBconn *threadConn;
string jobid, logid;
string status;
};

Modified: trunk/pgadmin3/xtra/pgagent/include/pgAgent.h
===================================================================
--- trunk/pgadmin3/xtra/pgagent/include/pgAgent.h 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/include/pgAgent.h 2005-05-19 14:53:32 UTC (rev 4219)
@@ -29,6 +29,7 @@
extern long minLogLevel;
extern string connectString;
extern string serviceDBname;
+extern string backendPid;

// Log levels
enum

Modified: trunk/pgadmin3/xtra/pgagent/job.cpp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/job.cpp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/job.cpp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -15,23 +15,23 @@

Job::Job(DBconn *conn, const string &jid)
{
- serviceConn=conn;
+ threadConn=conn;
jobid=jid;
status="";

- int rc=serviceConn->ExecuteVoid(
- "UPDATE pgagent.pga_job SET jobagentid=pg_backend_pid(), joblastrun=now() "
+ int rc=threadConn->ExecuteVoid(
+ "UPDATE pgagent.pga_job SET jobagentid=" + backendPid + ", joblastrun=now() "
" WHERE jobagentid IS NULL AND jobid=" + jobid);

if (rc == 1)
{
- DBresult *id=serviceConn->Execute(
+ DBresult *id=threadConn->Execute(
"SELECT nextval('pgagent.pga_joblog_jlgid_seq') AS id");
if (id)
{
logid=id->GetString("id");

- DBresult *res=serviceConn->Execute(
+ DBresult *res=threadConn->Execute(
"INSERT INTO pgagent.pga_joblog(jlgid, jlgjobid, jlgstatus) "
"VALUES (" + logid + ", " + jobid + ", 'r')");
if (res)
@@ -49,7 +49,7 @@
{
if (status != "")
{
- serviceConn->ExecuteVoid(
+ threadConn->ExecuteVoid(
"UPDATE pgagent.pga_joblog "
" SET jlgstatus='" + status + "', jlgduration=now() - jlgstart "
" WHERE jlgid=" + logid + ";\n"
@@ -59,13 +59,14 @@
" WHERE jobid=" + jobid
);
}
+ threadConn->Return();
}


int Job::Execute()
{
int rc=0;
- DBresult *steps=serviceConn->Execute(
+ DBresult *steps=threadConn->Execute(
"SELECT jstid, jstkind, jstdbname, jstcode, jstonerror "
" FROM pgagent.pga_jobstep "
" WHERE jstenabled "
@@ -80,17 +81,17 @@

while (steps->HasData())
{
- DBconn *conn;
+ DBconn *stepConn;
string jslid, stepid, jpecode;

stepid = steps->GetString("jstid");

- DBresult *id=serviceConn->Execute(
+ DBresult *id=threadConn->Execute(
"SELECT nextval('pgagent.pga_jobsteplog_jslid_seq') AS id");
if (id)
{
jslid=id->GetString("id");
- DBresult *res=serviceConn->Execute(
+ DBresult *res=threadConn->Execute(
"INSERT INTO pgagent.pga_jobsteplog(jslid, jsljlgid, jsljstid, jslstatus) "
"SELECT " + jslid + ", " + logid + ", " + stepid + ", 'r'"
" FROM pgagent.pga_jobstep WHERE jstid=" + stepid);
@@ -115,11 +116,12 @@
{
case 's':
{
- conn=DBconn::Get(steps->GetString("jstdbname"));
- if (conn)
+ stepConn=DBconn::Get(steps->GetString("jstdbname"));
+ if (stepConn)
{
LogMessage("Executing step " + stepid + " on database " + steps->GetString("jstdbname"), LOG_DEBUG);
- rc=conn->ExecuteVoid(steps->GetString("jstcode"));
+ rc=stepConn->ExecuteVoid(steps->GetString("jstcode"));
+ stepConn->Return();
}
else
rc=-1;
@@ -144,7 +146,7 @@
else
stepstatus = steps->GetString("jstonerror");

- rc=serviceConn->ExecuteVoid(
+ rc=threadConn->ExecuteVoid(
"UPDATE pgagent.pga_jobsteplog "
" SET jslduration = now() - jslstart, "
" jslresult = " + NumToStr(rc) + ", jslstatus = '" + stepstatus + "' "

Modified: trunk/pgadmin3/xtra/pgagent/misc.cpp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/misc.cpp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/misc.cpp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -64,13 +64,6 @@
longWait = val;
break;
}
- case 'c':
- {
- int val = atoi(getArg(argc, argv).c_str());
- if (val >= 5)
- connPoolCount = val;
- break;
- }
case 'l':
{
int val = atoi(getArg(argc, argv).c_str());

Modified: trunk/pgadmin3/xtra/pgagent/pgAgent.cpp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/pgAgent.cpp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/pgAgent.cpp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -11,6 +11,8 @@

#include "pgAgent.h"

+#include <wx/thread.h>
+
#ifdef WIN32
#include <winsock2.h>
#else
@@ -19,12 +21,12 @@

string connectString;
string serviceDBname;
+string backendPid;
long longWait=30;
long shortWait=10;
long minLogLevel=LOG_ERROR;


-
int MainRestartLoop(DBconn *serviceConn)
{
// clean up old jobs
@@ -65,8 +67,6 @@
);
}

-
-
char hostname[255];
gethostname(hostname, 255);

@@ -96,18 +96,20 @@

if (jobid != "")
{
- Job job(serviceConn, jobid);
+ DBconn *threadConn=DBconn::Get(serviceDBname);
+ Job job(threadConn, jobid);

if (job.Runnable())
{
foundJobToExecute=true;
LogMessage("Running job: " + jobid, LOG_DEBUG);
job.Execute();
+ LogMessage("Completed job: " + jobid, LOG_DEBUG);
}
}
else
{
- LogMessage("No jobs to run - time for a pint :-)", LOG_DEBUG);
+ LogMessage("No jobs to run - sleeping...", LOG_DEBUG);
WaitAWhile();
}
}
@@ -124,28 +126,34 @@

void MainLoop()
{
- // Basic sanity check
- LogMessage("Database sanity check", LOG_DEBUG);
- DBconn *sanityConn=DBconn::Get(serviceDBname, true);
- DBresult *res=sanityConn->Execute("SELECT count(*) As count FROM pg_class cl JOIN pg_namespace ns ON ns.oid=relnamespace WHERE relname='pga_job' AND nspname='pgagent'");
- if (res)
- {
- string val=res->GetString("count");
-
- if (val == "0")
- LogMessage("Could not find the table 'pgagent.pga_job'. Have you run pgagent.sql on this database?", LOG_ERROR);
- }

// OK, let's get down to business
do
{
- DBconn *serviceConn=DBconn::Get(serviceDBname, true);
+ LogMessage("Creating primary connection", LOG_DEBUG);
+ DBconn *serviceConn=DBconn::InitConnection(connectString);
+
+ if (serviceConn && serviceConn->IsValid())
+ {
+ serviceDBname = serviceConn->GetDBname();

- if (serviceConn)
- {
+ // Basic sanity check, and a chance to get the serviceConn's PID
+ LogMessage("Database sanity check", LOG_DEBUG);
+ DBresult *res=serviceConn->Execute("SELECT count(*) As count, pg_backend_pid() AS pid FROM pg_class cl JOIN pg_namespace ns ON ns.oid=relnamespace WHERE relname='pga_job' AND nspname='pgagent'");
+ if (res)
+ {
+ string val=res->GetString("count");
+
+ if (val == "0")
+ LogMessage("Could not find the table 'pgagent.pga_job'. Have you run pgagent.sql on this database?", LOG_ERROR);
+
+ backendPid=res->GetString("pid");
+ }
+
MainRestartLoop(serviceConn);
}

+ LogMessage("Couldn't create connection: " + serviceConn->GetLastError(), LOG_WARNING);
DBconn::ClearConnections(true);
WaitAWhile(true);
}

Modified: trunk/pgadmin3/xtra/pgagent/pgAgent.dsp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/pgAgent.dsp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/pgAgent.dsp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -70,7 +70,7 @@
# PROP Ignore_Export_Lib 0
# PROP Target_Dir ""
# ADD BASE CPP /nologo /W3 /Gm /GX /ZI /Od /I "include" /I "c:/program files/postgresql/8.0/include" /D "WIN32" /D "_DEBUG" /D "_WINDOWS" /D "_MBCS" /D "_MT" /Yu"pgAgent.h" /FD /GZ /c
-# ADD CPP /nologo /MT /W3 /GX /O2 /I "include" /I "c:/program files/postgresql/8.0/include" /I "c:/wxWidgets-2.6/include" /I "c:/wxWidgets-2.6/contrib/include" /D "WIN32" /D "_WINDOWS" /D "_MBCS" /D "_MT" /FR /Yu"pgamsgevent.h" /FD /c
+# ADD CPP /nologo /MT /W3 /GX /O2 /I "include" /I "c:/program files/postgresql/8.0/include" /I "c:/wxWidgets-2.6/include" /I "c:/wxWidgets-2.6/contrib/include" /D "WIN32" /D "_WINDOWS" /D "_MBCS" /D "_MT" /FR /Yu"pgAgent.h" /FD /c
# ADD BASE RSC /l 0x407 /d "_DEBUG"
# ADD RSC /l 0x809 /i "c:/wxWidgets-2.6/include"
BSC32=bscmake.exe

Modified: trunk/pgadmin3/xtra/pgagent/unix.cpp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/unix.cpp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/unix.cpp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -25,7 +25,6 @@
"options:\n"
"-t <poll time interval in seconds (default 10)>\n"
"-r <retry period after connection abort in seconds (>=10, default 30)>\n"
- "-c <connection pool size (>=5, default 5)>\n"
"-l <logging verbosity (ERROR=0, WARNING=1, DEBUG=2, default 0)>\n"
);
}
@@ -87,12 +86,6 @@

setOptions(argc, argv);

- DBconn *conn=DBconn::InitConnection(connectString);
- if (!conn->IsValid())
- LogMessage("Invalid connection: " + conn->GetLastError(), LOG_ERROR);
-
- serviceDBname = conn->GetDBname();
-
daemonize();

MainLoop();

Modified: trunk/pgadmin3/xtra/pgagent/win32.cpp
===================================================================
--- trunk/pgadmin3/xtra/pgagent/win32.cpp 2005-05-19 10:30:28 UTC (rev 4218)
+++ trunk/pgadmin3/xtra/pgagent/win32.cpp 2005-05-19 14:53:32 UTC (rev 4219)
@@ -328,7 +328,6 @@
"-d <displayname>\n"
"-t <poll time interval in seconds (default 10)>\n"
"-r <retry period after connection abort in seconds (>=10, default 30)>\n"
- "-c <connection pool size (>=5, default 5)>\n"
"-l <logging verbosity (ERROR=0, WARNING=1, DEBUG=2, default 0)>\n"
);
}
@@ -337,19 +336,16 @@

////////////////////////////////////////////////////////////

-void setupForRun(int argc, char **argv)
+void setupForRun(int argc, char **argv, bool debug=false)
{
- eventHandle = RegisterEventSource(0, serviceName.c_str());
- if (!eventHandle)
- LogMessage("Couldn't register event handle.", LOG_ERROR);
+ if (!debug)
+ {
+ eventHandle = RegisterEventSource(0, serviceName.c_str());
+ if (!eventHandle)
+ LogMessage("Couldn't register event handle.", LOG_ERROR);
+ }

setOptions(argc, argv);
-
- DBconn *conn=DBconn::InitConnection(connectString);
- if (!conn->IsValid())
- LogMessage("Invalid connection: " + conn->GetLastError(), LOG_ERROR);
-
- serviceDBname = conn->GetDBname();
}


@@ -368,7 +364,7 @@

if (command == "INSTALL")
{
- string displayname = "pgAgent " + serviceName;
+ string displayname = "PostgreSQL scheduling agent - " + serviceName;
string arg = executable + " RUN " + serviceName;

while (argc-- > 0)
@@ -416,7 +412,7 @@
}
else if (command == "DEBUG")
{
- setupForRun(argc, argv);
+ setupForRun(argc, argv, true);

initService();
#if START_SUSPENDED