mirror of
git://git.psyced.org/git/psyclpc
synced 2024-08-15 03:20:16 +00:00
1232 lines
31 KiB
C
1232 lines
31 KiB
C
|
/*---------------------------------------------------------------------------
|
||
|
* PostgreSQL Database package.
|
||
|
*
|
||
|
* Based on code written and donated 2001 by Florian Heinz.
|
||
|
*---------------------------------------------------------------------------
|
||
|
* The interface to the PostgreSQL database is implemented through the
|
||
|
* concept of a controlling object: when opening a database connection,
|
||
|
* the LPC code has to provide a callback function. The object this function
|
||
|
* is bound to is the controlling object: all queries to the database
|
||
|
* will be issued by this object, and the responses will be sent to the
|
||
|
* callback function.
|
||
|
*
|
||
|
* The interface is also asynchronous: the pg_query() efun just queues
|
||
|
* the query with the database connection, and returns immediately. When
|
||
|
* the database has finished working the query, the callback function is
|
||
|
* called with the results.
|
||
|
*
|
||
|
* The callback function can be defined by name or by closure, and can
|
||
|
* be defined with extra parameters:
|
||
|
*
|
||
|
* void <callback>(int type, mixed ret, int id [, mixed extra...])
|
||
|
*
|
||
|
* <type> is the type of the call, <id> identifies the query for which
|
||
|
* this call is executed:
|
||
|
*
|
||
|
* PGRES_TUPLES_OK: <ret> is the result from a query.
|
||
|
* It is either a mapping (field name as key, indexing
|
||
|
* <n> values for n returned tuples), or an array
|
||
|
* of arrays (one per row).
|
||
|
*
|
||
|
* PGRES_COMMAND_OK: <ret> is a string which contains the
|
||
|
* server response (e.g. on INSERT or DELETE)
|
||
|
*
|
||
|
* PGRES_BAD_RESPONSE,
|
||
|
* PGRES_NONFATAL_ERROR,
|
||
|
* PGRES_FATAL_ERROR: ret is the error-string
|
||
|
*
|
||
|
*
|
||
|
* void <callback>(int type, mixed ret [, mixed extra...])
|
||
|
*
|
||
|
* <type> is the type of the call, which is not related a specific query:
|
||
|
*
|
||
|
* PGCONN_SUCCESS: The database-connection was established, <ret> is
|
||
|
* a dummy string.
|
||
|
* PGCONN_FAILED: The database-connection failed, <ret> is the error
|
||
|
* message.
|
||
|
* The first message to the callback after a call to pg_connect()
|
||
|
* is always one of these two.
|
||
|
*
|
||
|
* PGRES_NOTICE: <ret> is a informational text.
|
||
|
*
|
||
|
* PGCONN_ABORTED: If the connection to the backend fails
|
||
|
* we try to re-establish (reset) it. If the
|
||
|
* reset fails, the connection is closed and
|
||
|
* this value is returned. Consider the
|
||
|
* connection gone and don't try to close or
|
||
|
* otherwise operate further on it.
|
||
|
* <ret> is a dummy string.
|
||
|
*---------------------------------------------------------------------------
|
||
|
*/
|
||
|
|
||
|
#include "driver.h"
|
||
|
|
||
|
#ifdef USE_PGSQL
|
||
|
|
||
|
#include "typedefs.h"
|
||
|
|
||
|
#include "my-alloca.h"
|
||
|
#include <errno.h>
|
||
|
#include <stddef.h>
|
||
|
#include <time.h>
|
||
|
#include <unistd.h>
|
||
|
|
||
|
#ifdef HAVE_POLL
|
||
|
|
||
|
#include <sys/poll.h>
|
||
|
|
||
|
#else
|
||
|
|
||
|
# if defined(_AIX) || defined(__EMX__) || defined(OS2)
|
||
|
# include <sys/select.h>
|
||
|
# endif
|
||
|
|
||
|
#endif /* HAVE_POLL */
|
||
|
|
||
|
#include <libpq-fe.h>
|
||
|
|
||
|
#include "pkg-pgsql.h"
|
||
|
|
||
|
#include "actions.h"
|
||
|
#include "array.h"
|
||
|
#include "gcollect.h"
|
||
|
#include "instrs.h"
|
||
|
#include "interpret.h"
|
||
|
#include "main.h"
|
||
|
#include "mapping.h"
|
||
|
#include "mstrings.h"
|
||
|
#include "simulate.h"
|
||
|
#include "stdstrings.h"
|
||
|
#include "xalloc.h"
|
||
|
|
||
|
#include "../mudlib/sys/pgsql.h"
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
|
||
|
#define MAX_RESETS 5 /* Number of reset tries, at max. one per second */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
/* Types */
|
||
|
|
||
|
/* --- struct query_queue_s: one entry in the query queue
|
||
|
* The queue is organized as single-linked list, one for each database
|
||
|
* connection.
|
||
|
*/
|
||
|
struct query_queue_s
|
||
|
{
|
||
|
long id;
|
||
|
char * str;
|
||
|
int flags;
|
||
|
struct query_queue_s * next;
|
||
|
};
|
||
|
|
||
|
/* --- struct dbconn_s: one active database connection.
|
||
|
* The connections are held in a singly linked list.
|
||
|
* There can be only one connection per object.
|
||
|
*/
|
||
|
|
||
|
struct dbconn_s
|
||
|
{
|
||
|
callback_t callback; /* The callback */
|
||
|
|
||
|
PGconn * conn;
|
||
|
PostgresPollingStatusType pgstate;
|
||
|
|
||
|
int state;
|
||
|
int fd;
|
||
|
int resets;
|
||
|
|
||
|
time_t lastreset;
|
||
|
time_t lastreply;
|
||
|
|
||
|
struct query_queue_s * queue;
|
||
|
|
||
|
struct dbconn_s * next;
|
||
|
};
|
||
|
|
||
|
|
||
|
/* Possible struct dbconn_s.states */
|
||
|
|
||
|
#define PG_UNCONNECTED 0
|
||
|
#define PG_CONNECTING 1
|
||
|
#define PG_RESETTING 2
|
||
|
#define PG_RESET_NEXT 3
|
||
|
#define PG_IDLE 4
|
||
|
#define PG_SENDQUERY 5
|
||
|
#define PG_WAITREPLY 6
|
||
|
#define PG_REPLYREADY 7
|
||
|
|
||
|
|
||
|
typedef struct dbconn_s dbconn_t;
|
||
|
typedef struct query_queue_s query_queue_t;
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
|
||
|
static dbconn_t *head = NULL;
|
||
|
/* The list of database connections.
|
||
|
*/
|
||
|
|
||
|
static long query_id = 1;
|
||
|
/* The query ID counter, used to generate unique IDs.
|
||
|
*/
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
/* Forward Declarations */
|
||
|
|
||
|
static void pgnotice (dbconn_t *pgconn, const char *msg);
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
void
|
||
|
pg_setfds (fd_set *readfds, fd_set *writefds, int *nfds)
|
||
|
|
||
|
/* Called from the get_message() loop in comm.c, this function has to add
|
||
|
* the fds of the database connections to the fd sets.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *ptr;
|
||
|
|
||
|
for (ptr = head; ptr != NULL; ptr = ptr->next)
|
||
|
{
|
||
|
if (ptr->fd < 0)
|
||
|
continue;
|
||
|
if ((ptr->pgstate == PGRES_POLLING_WRITING)
|
||
|
|| (ptr->state == PG_SENDQUERY)
|
||
|
)
|
||
|
FD_SET(ptr->fd, writefds);
|
||
|
FD_SET(ptr->fd, readfds);
|
||
|
if (*nfds <= ptr->fd)
|
||
|
*nfds = ptr->fd + 1;
|
||
|
}
|
||
|
} /* pg_setfds() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static dbconn_t *
|
||
|
find_current_connection (object_t * obj)
|
||
|
|
||
|
/* Find the <dbconn> which has a callback in the object <obj>.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *ptr = head;
|
||
|
|
||
|
while (ptr && callback_object(&ptr->callback) != obj)
|
||
|
ptr = ptr->next;
|
||
|
|
||
|
return ptr;
|
||
|
} /* find_current_connection() */
|
||
|
|
||
|
/*=========================================================================*/
|
||
|
|
||
|
/* Queue management */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static query_queue_t *
|
||
|
queue (dbconn_t *db, const char *query)
|
||
|
|
||
|
/* Create a new query_queue entry for <query>, link it into the list
|
||
|
* for database connection <db>, and return it.
|
||
|
* The query string is duplicated.
|
||
|
*
|
||
|
* Throw an error when out of memory.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
query_queue_t *tmp;
|
||
|
|
||
|
tmp = db->queue;
|
||
|
if (!tmp)
|
||
|
tmp = db->queue = xalloc(sizeof(*tmp));
|
||
|
else
|
||
|
{
|
||
|
while (tmp->next)
|
||
|
tmp = tmp->next;
|
||
|
tmp->next = xalloc(sizeof(*tmp));
|
||
|
tmp = tmp->next;
|
||
|
}
|
||
|
|
||
|
if (!tmp)
|
||
|
{
|
||
|
outofmemory("new Postgres query");
|
||
|
/* NOTREACHED */
|
||
|
return NULL;
|
||
|
}
|
||
|
|
||
|
memset(tmp, 0, sizeof(*tmp));
|
||
|
tmp->id = query_id++;
|
||
|
tmp->str = string_copy(query);
|
||
|
tmp->flags = 0;
|
||
|
|
||
|
return tmp;
|
||
|
} /* queue() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
dequeue (dbconn_t *db)
|
||
|
|
||
|
/* Unqueue the first query from connection <db> and deallocate it.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
query_queue_t *tmp;
|
||
|
|
||
|
tmp = db->queue;
|
||
|
db->queue = tmp->next;
|
||
|
if (tmp->str)
|
||
|
xfree(tmp->str);
|
||
|
xfree(tmp);
|
||
|
} /* dequeue() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static dbconn_t *
|
||
|
alloc_dbconn (void)
|
||
|
|
||
|
/* Allocate a new database connection structure, link it into the global
|
||
|
* list and return it.
|
||
|
*
|
||
|
* Throw an error when out of memory.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *ret;
|
||
|
|
||
|
memsafe(ret = xalloc(sizeof(*ret)), sizeof(*ret), "new DB connection");
|
||
|
|
||
|
memset(ret, 0, sizeof(*ret));
|
||
|
ret->next = head;
|
||
|
ret->fd = -1;
|
||
|
ret->state = PG_UNCONNECTED;
|
||
|
head = ret;
|
||
|
|
||
|
return ret;
|
||
|
} /* alloc_dbconn() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
dealloc_dbconn (dbconn_t *del)
|
||
|
|
||
|
/* Unlink the database connection <del> from the list and deallocate it
|
||
|
* and all resources held by it.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *ptr;
|
||
|
|
||
|
if (!del)
|
||
|
return;
|
||
|
|
||
|
free_callback(&del->callback);
|
||
|
|
||
|
if (del == head)
|
||
|
head = head->next;
|
||
|
else
|
||
|
{
|
||
|
ptr = head;
|
||
|
while (ptr->next && (ptr->next != del))
|
||
|
ptr = ptr->next;
|
||
|
if (ptr->next)
|
||
|
ptr->next = ptr->next->next;
|
||
|
}
|
||
|
while (del->queue)
|
||
|
dequeue(del);
|
||
|
xfree(del);
|
||
|
} /* dealloc_dbconn() */
|
||
|
|
||
|
/*=========================================================================*/
|
||
|
|
||
|
/* Connection management */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static int
|
||
|
pgconnect (dbconn_t *db, char *connstr)
|
||
|
|
||
|
/* Connect <db> to a database, using <connstr> for the connection parameters.
|
||
|
* Return 0 on success, and -1 on failure.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
db->conn = PQconnectStart(connstr);
|
||
|
if (!db->conn)
|
||
|
return -1;
|
||
|
|
||
|
if (PQstatus(db->conn) == CONNECTION_BAD)
|
||
|
return -1;
|
||
|
|
||
|
PQsetNoticeProcessor(db->conn, (void*) pgnotice, db);
|
||
|
db->fd = PQsocket(db->conn);
|
||
|
db->state = PG_CONNECTING;
|
||
|
db->pgstate = PGRES_POLLING_WRITING;
|
||
|
return 0;
|
||
|
} /* pgconnect() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pgclose (dbconn_t *pgconn)
|
||
|
|
||
|
/* Close the database connection of <pgconn>.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
pgconn->state = PG_UNCONNECTED;
|
||
|
if (pgconn->conn)
|
||
|
PQfinish(pgconn->conn);
|
||
|
pgconn->conn = NULL;
|
||
|
pgconn->fd = -1;
|
||
|
} /* pgclose() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pgreset (dbconn_t *pgconn)
|
||
|
|
||
|
/* Reset the connection to <pgconn>.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
if (!PQresetStart(pgconn->conn))
|
||
|
{
|
||
|
pgclose(pgconn);
|
||
|
|
||
|
if (callback_object(&pgconn->callback))
|
||
|
{
|
||
|
push_number(inter_sp, PGCONN_ABORTED);
|
||
|
push_ref_string(inter_sp, STR_PG_RESET_FAILED);
|
||
|
(void)apply_callback(&pgconn->callback, 2);
|
||
|
}
|
||
|
return;
|
||
|
}
|
||
|
|
||
|
pgconn->state = PG_RESETTING;
|
||
|
pgconn->pgstate = PGRES_POLLING_WRITING;
|
||
|
} /* pgreset() */
|
||
|
|
||
|
|
||
|
/*=========================================================================*/
|
||
|
|
||
|
/* Database Result Handling */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pgnotice (dbconn_t *pgconn, const char *msg)
|
||
|
|
||
|
/* Database connection <pgconn> wishes to send <msg> to the controlling
|
||
|
* object.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
current_object = callback_object(&pgconn->callback);
|
||
|
command_giver = 0;
|
||
|
current_interactive = 0;
|
||
|
|
||
|
if (current_object != NULL)
|
||
|
{
|
||
|
push_number(inter_sp, PGRES_NOTICE);
|
||
|
push_c_string(inter_sp, msg);
|
||
|
(void)apply_callback(&pgconn->callback, 2);
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
pgreset(pgconn);
|
||
|
}
|
||
|
} /* pgnotice() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pgresult (dbconn_t *pgconn, PGresult *res)
|
||
|
|
||
|
/* The most recent query on <pgconn> returned the result <res>. Encode it into
|
||
|
* a nice LPC data package and send it to the controlling object.
|
||
|
* The query is removed from <pgconn>.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
int type;
|
||
|
|
||
|
current_object = callback_object(&pgconn->callback);
|
||
|
command_giver = 0;
|
||
|
current_interactive = 0;
|
||
|
|
||
|
type = PQresultStatus(res);
|
||
|
|
||
|
push_number(inter_sp, type);
|
||
|
|
||
|
switch (type)
|
||
|
{
|
||
|
case PGRES_TUPLES_OK:
|
||
|
{
|
||
|
int nfields, ntuples, i, j;
|
||
|
|
||
|
nfields = PQnfields(res);
|
||
|
ntuples = PQntuples(res);
|
||
|
|
||
|
if (pgconn->queue->flags & PG_RESULT_MAP)
|
||
|
{
|
||
|
/* Return the result as mapping */
|
||
|
|
||
|
mapping_t *map;
|
||
|
|
||
|
if (max_mapping_size
|
||
|
&& (nfields * (ntuples + 1)) > (p_int)max_mapping_size)
|
||
|
{
|
||
|
PQclear(res);
|
||
|
dequeue(pgconn);
|
||
|
errorf("Query result exceeded mappingsize limit.\n");
|
||
|
}
|
||
|
|
||
|
if (max_mapping_keys && nfields > (p_int)max_mapping_keys)
|
||
|
{
|
||
|
PQclear(res);
|
||
|
dequeue(pgconn);
|
||
|
errorf("Query result exceeded mappingsize limit.\n");
|
||
|
}
|
||
|
|
||
|
map = allocate_mapping(nfields, ntuples);
|
||
|
if (!map)
|
||
|
{
|
||
|
push_number(inter_sp, 0);
|
||
|
break;
|
||
|
}
|
||
|
|
||
|
for (i = 0; i < nfields; i++)
|
||
|
{
|
||
|
svalue_t * entry, fname;
|
||
|
|
||
|
put_c_string(&fname, PQfname(res, i));
|
||
|
entry = get_map_lvalue(map, &fname);
|
||
|
free_svalue(&fname);
|
||
|
if (!entry)
|
||
|
break;
|
||
|
for (j = 0; j < ntuples; j++)
|
||
|
put_c_string(&entry[j], PQgetvalue(res, j, i));
|
||
|
}
|
||
|
|
||
|
push_mapping(inter_sp, map);
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
/* Return the result as array of arrays */
|
||
|
|
||
|
vector_t * array;
|
||
|
svalue_t * entry;
|
||
|
|
||
|
if (max_array_size
|
||
|
&& ( (ntuples >= (p_int)max_array_size)
|
||
|
|| (nfields >= (p_int)max_array_size))
|
||
|
)
|
||
|
{
|
||
|
PQclear(res);
|
||
|
dequeue(pgconn);
|
||
|
errorf("Query result exceeded array limit.\n");
|
||
|
}
|
||
|
|
||
|
array = allocate_array(ntuples+1);
|
||
|
if (!array)
|
||
|
{
|
||
|
push_number(inter_sp, 0);
|
||
|
break;
|
||
|
}
|
||
|
|
||
|
entry = &array->item[0];
|
||
|
put_array(entry, allocate_array(nfields));
|
||
|
for (j = 0; j < nfields; j++)
|
||
|
put_c_string(&entry->u.vec->item[j], PQfname(res, j));
|
||
|
|
||
|
for (i = 0; i < ntuples; i++)
|
||
|
{
|
||
|
entry = &array->item[i+1];
|
||
|
put_array(entry, allocate_array(nfields));
|
||
|
for (j = 0; j < nfields; j++)
|
||
|
put_c_string(&entry->u.vec->item[j], PQgetvalue(res, i, j));
|
||
|
}
|
||
|
push_array(inter_sp, array);
|
||
|
}
|
||
|
break;
|
||
|
}
|
||
|
|
||
|
case PGRES_COMMAND_OK:
|
||
|
push_c_string(inter_sp, PQcmdStatus(res));
|
||
|
break;
|
||
|
|
||
|
case PGRES_BAD_RESPONSE:
|
||
|
push_c_string(inter_sp, PQerrorMessage(pgconn->conn));
|
||
|
break;
|
||
|
|
||
|
case PGRES_FATAL_ERROR:
|
||
|
case PGRES_NONFATAL_ERROR:
|
||
|
push_c_string(inter_sp, PQerrorMessage(pgconn->conn));
|
||
|
break;
|
||
|
|
||
|
default:
|
||
|
PQclear(res);
|
||
|
return;
|
||
|
}
|
||
|
|
||
|
if (callback_object(&pgconn->callback))
|
||
|
{
|
||
|
push_number(inter_sp, pgconn->queue->id);
|
||
|
(void)apply_callback(&pgconn->callback, 3);
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
pgreset(pgconn);
|
||
|
}
|
||
|
|
||
|
dequeue(pgconn);
|
||
|
PQclear(res);
|
||
|
} /* pgresult() */
|
||
|
|
||
|
|
||
|
/*=========================================================================*/
|
||
|
|
||
|
/* Database Connection Handling */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pg_process_connect_reset (dbconn_t *pgconn)
|
||
|
|
||
|
/* Reset the connection <pgconn>.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
int reset;
|
||
|
PostgresPollingStatusType (*pqpollhandler)(PGconn *);
|
||
|
|
||
|
reset = (pgconn->state == PG_RESETTING);
|
||
|
|
||
|
if (reset)
|
||
|
debug_message("%s PGSQL Connection resetting.\n", time_stamp());
|
||
|
|
||
|
pqpollhandler = reset ? PQresetPoll : PQconnectPoll;
|
||
|
|
||
|
if (pgconn->pgstate != PGRES_POLLING_ACTIVE)
|
||
|
{
|
||
|
int rc;
|
||
|
|
||
|
#ifdef HAVE_POLL
|
||
|
struct pollfd ufd;
|
||
|
|
||
|
ufd.fd = pgconn->fd;
|
||
|
#else
|
||
|
fd_set readfds, writefds;
|
||
|
|
||
|
FD_ZERO(&readfds);
|
||
|
FD_ZERO(&writefds);
|
||
|
#endif /* HAVE_POLL */
|
||
|
|
||
|
switch (pgconn->pgstate)
|
||
|
{
|
||
|
case PGRES_POLLING_READING:
|
||
|
#ifdef HAVE_POLL
|
||
|
ufd.events = POLLIN;
|
||
|
#else
|
||
|
FD_SET(pgconn->fd, &readfds);
|
||
|
#endif /* HAVE_POLL */
|
||
|
break;
|
||
|
|
||
|
case PGRES_POLLING_WRITING:
|
||
|
#ifdef HAVE_POLL
|
||
|
ufd.events = POLLOUT;
|
||
|
#else
|
||
|
FD_SET(pgconn->fd, &writefds);
|
||
|
#endif /* HAVE_POLL */
|
||
|
break;
|
||
|
|
||
|
default:
|
||
|
/* Shouldn't happen */
|
||
|
break;
|
||
|
}
|
||
|
|
||
|
#ifdef HAVE_POLL
|
||
|
do {
|
||
|
rc = poll(&ufd, 1, 0);
|
||
|
} while (rc < 0 && errno == EINTR);
|
||
|
|
||
|
if (rc > 0)
|
||
|
{
|
||
|
pgconn->pgstate = PGRES_POLLING_ACTIVE;
|
||
|
if (ufd.revents & POLLIN)
|
||
|
pgconn->lastreply = time(NULL);
|
||
|
}
|
||
|
#else
|
||
|
do {
|
||
|
struct timeval timeout;
|
||
|
|
||
|
timeout.tv_sec = 0;
|
||
|
timeout.tv_usec = 0;
|
||
|
rc = select(pgconn->fd+1, &readfds, &writefds, NULL, &timeout);
|
||
|
if (rc >= 0 || errno != EINTR)
|
||
|
break;
|
||
|
} while (rc < 0 && errno == EINTR);
|
||
|
|
||
|
if (rc > 0)
|
||
|
{
|
||
|
pgconn->pgstate = PGRES_POLLING_ACTIVE;
|
||
|
if (FD_ISSET(pgconn->fd, &readfds))
|
||
|
pgconn->lastreply = time(NULL);
|
||
|
}
|
||
|
#endif /* HAVE_POLL */
|
||
|
}
|
||
|
|
||
|
if (pgconn->pgstate == PGRES_POLLING_ACTIVE)
|
||
|
pgconn->pgstate = pqpollhandler(pgconn->conn);
|
||
|
|
||
|
if (pgconn->pgstate == PGRES_POLLING_FAILED)
|
||
|
{
|
||
|
if (reset && (pgconn->resets < MAX_RESETS))
|
||
|
{
|
||
|
pgconn->resets++;
|
||
|
pgconn->state = PG_RESET_NEXT;
|
||
|
pgconn->lastreset = time(NULL);
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
if (callback_object(&pgconn->callback))
|
||
|
{
|
||
|
push_number(inter_sp, reset ? PGCONN_ABORTED : PGCONN_FAILED);
|
||
|
push_c_string(inter_sp, PQerrorMessage(pgconn->conn));
|
||
|
(void)apply_callback(&pgconn->callback, 2);
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
}
|
||
|
pgclose(pgconn);
|
||
|
}
|
||
|
}
|
||
|
else if (pgconn->pgstate == PGRES_POLLING_OK)
|
||
|
{
|
||
|
if (!reset)
|
||
|
{
|
||
|
/* The program should not notice a successful reset */
|
||
|
if (callback_object(&pgconn->callback))
|
||
|
{
|
||
|
push_number(inter_sp, PGCONN_SUCCESS);
|
||
|
push_ref_string(inter_sp, STR_SUCCESS);
|
||
|
(void)apply_callback(&pgconn->callback, 2);
|
||
|
}
|
||
|
else
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
pgreset(pgconn);
|
||
|
}
|
||
|
}
|
||
|
pgconn->resets = 0;
|
||
|
if (pgconn->queue)
|
||
|
pgconn->state = PG_SENDQUERY;
|
||
|
else
|
||
|
pgconn->state = PG_IDLE;
|
||
|
}
|
||
|
} /* pg_process_connect_reset() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pg_process_query (dbconn_t *pgconn)
|
||
|
|
||
|
/* Query the connection <pgconn> for data and act on it.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
int rc;
|
||
|
|
||
|
#ifdef HAVE_POLL
|
||
|
struct pollfd ufd;
|
||
|
|
||
|
ufd.fd = pgconn->fd;
|
||
|
ufd.events = POLLIN;
|
||
|
#else
|
||
|
struct fd_set readfds;
|
||
|
|
||
|
FD_ZERO(&readfds);
|
||
|
FD_SET(pgconn->fd, &readfds);
|
||
|
|
||
|
#endif /* HAVE_POLL */
|
||
|
|
||
|
PQflush(pgconn->conn);
|
||
|
|
||
|
#ifdef HAVE_POLL
|
||
|
|
||
|
do {
|
||
|
rc = poll(&ufd, 1, 0);
|
||
|
} while (rc < 0 && errno == EINTR);
|
||
|
|
||
|
#else
|
||
|
|
||
|
do {
|
||
|
struct timeval timeout;
|
||
|
|
||
|
timeout.tv_sec = 0;
|
||
|
timeout.tv_usec = 0;
|
||
|
rc = select(pgconn->fd+1, &readfds, NULL, NULL, &timeout);
|
||
|
} while (rc < 0 && errno == EINTR);
|
||
|
|
||
|
#endif /* HAVE_POLL */
|
||
|
|
||
|
if (rc > 0)
|
||
|
{
|
||
|
pgconn->lastreply = time(NULL);
|
||
|
PQconsumeInput(pgconn->conn);
|
||
|
if (!PQisBusy(pgconn->conn))
|
||
|
pgconn->state = PG_REPLYREADY;
|
||
|
}
|
||
|
} /* pg_process_query() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static void
|
||
|
pg_process_one (dbconn_t *pgconn)
|
||
|
|
||
|
/* Check the state of <pgconn> and take appropriate action.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
PGresult *res;
|
||
|
|
||
|
switch (pgconn->state)
|
||
|
{
|
||
|
case PG_CONNECTING:
|
||
|
case PG_RESETTING:
|
||
|
pg_process_connect_reset(pgconn);
|
||
|
break;
|
||
|
|
||
|
case PG_SENDQUERY:
|
||
|
case PG_IDLE:
|
||
|
if (pgconn->queue)
|
||
|
{
|
||
|
pgconn->lastreply = time(NULL);
|
||
|
PQsendQuery(pgconn->conn, pgconn->queue->str);
|
||
|
pgconn->state = PG_WAITREPLY;
|
||
|
}
|
||
|
break;
|
||
|
|
||
|
case PG_WAITREPLY:
|
||
|
pg_process_query(pgconn);
|
||
|
break;
|
||
|
|
||
|
case PG_UNCONNECTED:
|
||
|
dealloc_dbconn(pgconn);
|
||
|
break;
|
||
|
|
||
|
case PG_RESET_NEXT:
|
||
|
if (pgconn->lastreset != time(NULL))
|
||
|
pgreset(pgconn);
|
||
|
break;
|
||
|
} /* switch() */
|
||
|
|
||
|
/* Validate the connection */
|
||
|
if ((PQstatus(pgconn->conn) != CONNECTION_OK)
|
||
|
&& (pgconn->state >= PG_IDLE)
|
||
|
)
|
||
|
pgreset(pgconn);
|
||
|
|
||
|
/* If there is a result waiting, get it and forward
|
||
|
* it to the controlling object.
|
||
|
*/
|
||
|
if (pgconn->state == PG_REPLYREADY)
|
||
|
{
|
||
|
do
|
||
|
{
|
||
|
res = PQgetResult(pgconn->conn);
|
||
|
if (!res)
|
||
|
{
|
||
|
pgconn->state = PG_IDLE;
|
||
|
pg_process_one(pgconn);
|
||
|
break;
|
||
|
} else
|
||
|
pgresult(pgconn, res);
|
||
|
}
|
||
|
while (!PQisBusy(pgconn->conn));
|
||
|
}
|
||
|
} /* pg_process_one() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
void
|
||
|
pg_process_all (void)
|
||
|
|
||
|
/* Called from the get_message() loop in comm.c, this function checks
|
||
|
* all known database connections for their status and takes appropriate
|
||
|
* actions.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *ptr = head;
|
||
|
Bool got_dead = MY_FALSE;
|
||
|
|
||
|
while (ptr)
|
||
|
{
|
||
|
if (!callback_object(&ptr->callback)
|
||
|
&& ptr->state != PG_UNCONNECTED
|
||
|
)
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
pgclose(ptr);
|
||
|
got_dead = MY_TRUE;
|
||
|
}
|
||
|
else
|
||
|
pg_process_one(ptr);
|
||
|
ptr = ptr->next;
|
||
|
}
|
||
|
|
||
|
if (got_dead)
|
||
|
pg_purge_connections();
|
||
|
} /* pg_process_all() */
|
||
|
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
void
|
||
|
pg_purge_connections (void)
|
||
|
|
||
|
/* Check the list of database connections and purge all UNCONNECTED
|
||
|
* connections and those with destructed callback objects.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *prev;
|
||
|
|
||
|
while (head)
|
||
|
{
|
||
|
if (head->state != PG_UNCONNECTED
|
||
|
&& !callback_object(&head->callback)
|
||
|
)
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
pgclose(head);
|
||
|
}
|
||
|
if (head->state == PG_UNCONNECTED)
|
||
|
dealloc_dbconn(head);
|
||
|
else
|
||
|
break;
|
||
|
}
|
||
|
|
||
|
if (head)
|
||
|
{
|
||
|
prev = head;
|
||
|
while (prev->next)
|
||
|
{
|
||
|
if (prev->next->state != PG_UNCONNECTED
|
||
|
&& !callback_object(&prev->next->callback)
|
||
|
)
|
||
|
{
|
||
|
debug_message("%s PG connection object destructed.\n", time_stamp());
|
||
|
pgclose(prev->next);
|
||
|
}
|
||
|
if (prev->next->state == PG_UNCONNECTED)
|
||
|
dealloc_dbconn(prev->next);
|
||
|
else
|
||
|
prev = prev->next;
|
||
|
}
|
||
|
}
|
||
|
} /* pg_purge_connections() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
static
|
||
|
Bool check_privilege (const char * efun_name, Bool raise_error, svalue_t * sp)
|
||
|
|
||
|
/* Check if the user has the privileges to execute efun <efun_name>.
|
||
|
* The function executes a call to master->privilege_violation("mysql",
|
||
|
* efun_name) and evaluates the result.
|
||
|
* If the master result is TRUE, the function returns TRUE.
|
||
|
* If the master result is FALSE, the function returns FALSE if <raise_error>
|
||
|
* is FALSE, and raises an error if <raise_error> is true.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
Bool rc;
|
||
|
|
||
|
inter_sp = sp+1;
|
||
|
put_c_string(inter_sp, efun_name);
|
||
|
rc = privilege_violation(STR_PGSQL, inter_sp, inter_sp);
|
||
|
free_svalue(inter_sp);
|
||
|
inter_sp--;
|
||
|
|
||
|
if (rc)
|
||
|
return MY_TRUE;
|
||
|
|
||
|
if (raise_error)
|
||
|
{
|
||
|
errorf("%s(): Privilege violation.\n", efun_name);
|
||
|
/* NOTREACHED */
|
||
|
}
|
||
|
|
||
|
return MY_FALSE;
|
||
|
} /* check_privilege() */
|
||
|
|
||
|
|
||
|
|
||
|
/*=========================================================================*/
|
||
|
|
||
|
/* EFUNS */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
svalue_t *
|
||
|
v_pg_connect (svalue_t *sp, int num_arg)
|
||
|
|
||
|
/* EFUN pg_connect()
|
||
|
*
|
||
|
* int pg_connect (string conn, string fun)
|
||
|
* int pg_connect (string conn, string fun, string|object obj, mixed extra, ...)
|
||
|
* int pg_connect (string conn, closure cl, mixed extra, ...)
|
||
|
*
|
||
|
* Open a database connection as directed by <conn>, and assign the
|
||
|
* callback function <fun>/<cl> with the optional <extra> parameters to
|
||
|
* it.
|
||
|
*
|
||
|
* The object holding the callback function becomes the controlling object;
|
||
|
* obiously it is an error to assign more than one connection to the same
|
||
|
* controlling object.
|
||
|
*
|
||
|
* The <conn> string is in the format accepted by Postgres' PQconnectStart()
|
||
|
* API functions. Pass an empty string to use the default options, or
|
||
|
* a string holding the '<key>=<value>' options separated by whitespace.
|
||
|
* The most useful options are:
|
||
|
* dbname: The database name
|
||
|
* user: The user name to connect as.
|
||
|
* password: Password to be used.
|
||
|
*
|
||
|
* Return 0 on success, and -1 on failure.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *db;
|
||
|
int st;
|
||
|
int error_index;
|
||
|
callback_t cb;
|
||
|
object_t *cb_object;
|
||
|
svalue_t *arg = sp - num_arg + 1;
|
||
|
|
||
|
check_privilege(instrs[F_PG_CONNECT].name, MY_TRUE, sp);
|
||
|
|
||
|
/* Get the callback information */
|
||
|
|
||
|
error_index = setup_efun_callback(&cb, arg+1, num_arg-1);
|
||
|
|
||
|
if (error_index >= 0)
|
||
|
{
|
||
|
vefun_bad_arg(error_index+2, arg);
|
||
|
/* NOTREACHED */
|
||
|
return arg;
|
||
|
}
|
||
|
inter_sp = sp = arg+1;
|
||
|
put_callback(sp, &cb);
|
||
|
|
||
|
cb_object = callback_object(&cb);
|
||
|
if (!cb_object)
|
||
|
{
|
||
|
free_callback(&cb);
|
||
|
errorf("pgconnect(): Callback object is destructed.\n");
|
||
|
/* NOTREACHED */
|
||
|
return arg;
|
||
|
}
|
||
|
|
||
|
/* Check the callback object if it has a connection already */
|
||
|
|
||
|
db = find_current_connection(cb_object);
|
||
|
if (db)
|
||
|
{
|
||
|
if (db->state == PG_UNCONNECTED)
|
||
|
dealloc_dbconn(db);
|
||
|
else
|
||
|
{
|
||
|
free_callback(&cb);
|
||
|
errorf("pgconnect(): Already connected\n");
|
||
|
/* NOTREACHED */
|
||
|
return arg;
|
||
|
}
|
||
|
}
|
||
|
|
||
|
/* Connect to the database */
|
||
|
|
||
|
db = alloc_dbconn();
|
||
|
db->callback = cb;
|
||
|
|
||
|
st = pgconnect(db, get_txt(arg[0].u.str));
|
||
|
if (st < 0)
|
||
|
pgclose(db);
|
||
|
|
||
|
free_svalue(arg); /* the callback entries are gone already */
|
||
|
put_number(arg, st);
|
||
|
return arg;
|
||
|
} /* f_pg_connect() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
svalue_t *
|
||
|
f_pg_pending (svalue_t *sp)
|
||
|
|
||
|
/* EFUN pg_pending()
|
||
|
*
|
||
|
* int pg_pending ()
|
||
|
* int pg_pending (object obj)
|
||
|
*
|
||
|
* Return the number of pending queries for the connection on the given
|
||
|
* object <obj> (default is the current object). The object has no
|
||
|
* database connection, return -1.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *db;
|
||
|
int count = -1;
|
||
|
|
||
|
check_privilege(instrs[F_PG_PENDING].name, MY_TRUE, sp);
|
||
|
|
||
|
db = find_current_connection(sp->u.ob);
|
||
|
if (db)
|
||
|
{
|
||
|
query_queue_t * qu;
|
||
|
|
||
|
for (count = 0, qu = db->queue
|
||
|
; qu != NULL
|
||
|
; count++, qu = qu->next
|
||
|
) NOOP;
|
||
|
}
|
||
|
|
||
|
free_svalue(sp);
|
||
|
put_number(sp, count);
|
||
|
return sp;
|
||
|
} /* f_pg_pending() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
svalue_t *
|
||
|
v_pg_query (svalue_t *sp, int numarg)
|
||
|
|
||
|
/* EFUN pg_query()
|
||
|
*
|
||
|
* int pg_query (string query)
|
||
|
* int pg_query (string query, int flags)
|
||
|
*
|
||
|
* Queue a new query <query> to the database connection on the current
|
||
|
* object. Return the unique id of the query. The query result itself
|
||
|
* will be passed as argument to the callback function.
|
||
|
*
|
||
|
* <flags> can be one of these values:
|
||
|
* PG_RESULT_ARRAY: Pass the query result as array.
|
||
|
* PG_RESULT_MAP: Pass the query result as mapping.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *db;
|
||
|
query_queue_t *q;
|
||
|
int flags = PG_RESULT_ARRAY;
|
||
|
|
||
|
check_privilege(instrs[F_PG_QUERY].name, MY_TRUE, sp);
|
||
|
|
||
|
if (numarg == 2)
|
||
|
{
|
||
|
flags = sp->u.number;
|
||
|
sp--;
|
||
|
}
|
||
|
|
||
|
db = find_current_connection(current_object);
|
||
|
if (!db)
|
||
|
errorf("pgquery(): not connected\n");
|
||
|
|
||
|
q = queue(db, get_txt(sp->u.str));
|
||
|
q->flags = flags;
|
||
|
if (db->state == PG_IDLE)
|
||
|
db->state = PG_SENDQUERY;
|
||
|
|
||
|
free_svalue(sp);
|
||
|
put_number(sp, q->id);
|
||
|
return sp;
|
||
|
} /* f_pg_query() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
svalue_t *
|
||
|
f_pg_close (svalue_t *sp)
|
||
|
|
||
|
/* EFUN pg_close()
|
||
|
*
|
||
|
* void pg_close()
|
||
|
*
|
||
|
* Close the database connection for the current object, if there is one.
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *db;
|
||
|
|
||
|
check_privilege(instrs[F_PG_CLOSE].name, MY_TRUE, sp);
|
||
|
|
||
|
db = find_current_connection(current_object);
|
||
|
if (db)
|
||
|
pgclose(db);
|
||
|
|
||
|
return sp;
|
||
|
} /* f_pg_close() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
svalue_t *
|
||
|
f_pg_conv_string (svalue_t *sp)
|
||
|
|
||
|
/* EFUN pg_escapeString
|
||
|
*
|
||
|
* string pg_conv_string(string input)
|
||
|
*
|
||
|
* Escape a string for use within an SQL command.
|
||
|
*/
|
||
|
{
|
||
|
string_t *escaped;
|
||
|
int size = mstrsize(sp->u.str);
|
||
|
memsafe(escaped = alloc_mstring(2 * size), 2 * size
|
||
|
, "escaped sql string");
|
||
|
|
||
|
// PQescapeString(char *to, char *from, size_t length);
|
||
|
PQescapeString( (unsigned char *)get_txt(escaped)
|
||
|
, (unsigned char *)get_txt(sp->u.str), size);
|
||
|
free_string_svalue(sp);
|
||
|
put_string(sp, escaped);
|
||
|
return sp;
|
||
|
} /* pg_conv_string() */
|
||
|
|
||
|
/*=========================================================================*/
|
||
|
|
||
|
/* GC SUPPORT */
|
||
|
|
||
|
#ifdef GC_SUPPORT
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
void
|
||
|
pg_clear_refs (void)
|
||
|
|
||
|
/* GC Support: Clear all references from the database connections
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *dbconn;
|
||
|
|
||
|
for (dbconn = head; dbconn != NULL; dbconn = dbconn->next)
|
||
|
{
|
||
|
clear_ref_in_callback(&(dbconn->callback));
|
||
|
}
|
||
|
} /* pg_clear_refs() */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
void
|
||
|
pg_count_refs (void)
|
||
|
|
||
|
/* GC Support: Count all references from the database connections
|
||
|
*/
|
||
|
|
||
|
{
|
||
|
dbconn_t *dbconn;
|
||
|
|
||
|
for (dbconn = head; dbconn != NULL; dbconn = dbconn->next)
|
||
|
{
|
||
|
query_queue_t *qu;
|
||
|
|
||
|
note_malloced_block_ref(dbconn);
|
||
|
count_ref_in_callback(&(dbconn->callback));
|
||
|
|
||
|
for (qu = dbconn->queue; qu != NULL; qu = qu->next)
|
||
|
{
|
||
|
note_malloced_block_ref(qu);
|
||
|
note_malloced_block_ref(qu->str);
|
||
|
}
|
||
|
}
|
||
|
} /* pg_count_refs() */
|
||
|
|
||
|
#endif /* GC_SUPPORT */
|
||
|
|
||
|
/*-------------------------------------------------------------------------*/
|
||
|
|
||
|
#endif /* USE_PGSQL */
|
||
|
|
||
|
/*************************************************************************/
|