Some changes to the lobject code to match changes in pqpath.[ch].

This commit is contained in:
James Henstridge 2008-05-05 12:07:24 +08:00
parent 3cfe438b74
commit ec20fa8912
5 changed files with 144 additions and 67 deletions

View File

@ -26,6 +26,7 @@
#include <libpq-fe.h> #include <libpq-fe.h>
#include <libpq/libpq-fs.h> #include <libpq/libpq-fs.h>
#include "psycopg/config.h"
#include "psycopg/connection.h" #include "psycopg/connection.h"
#ifdef __cplusplus #ifdef __cplusplus
@ -51,16 +52,16 @@ typedef struct {
/* functions exported from lobject_int.c */ /* functions exported from lobject_int.c */
extern int lobject_open(lobjectObject *self, connectionObject *conn, HIDDEN int lobject_open(lobjectObject *self, connectionObject *conn,
Oid oid, int mode, Oid new_oid, char *new_file); Oid oid, int mode, Oid new_oid, char *new_file);
extern int lobject_unlink(lobjectObject *self); HIDDEN int lobject_unlink(lobjectObject *self);
extern int lobject_export(lobjectObject *self, char *filename); HIDDEN int lobject_export(lobjectObject *self, char *filename);
extern size_t lobject_read(lobjectObject *self, char *buf, size_t len); HIDDEN Py_ssize_t lobject_read(lobjectObject *self, char *buf, size_t len);
extern size_t lobject_write(lobjectObject *self, char *buf, size_t len); HIDDEN Py_ssize_t lobject_write(lobjectObject *self, char *buf, size_t len);
extern int lobject_seek(lobjectObject *self, int pos, int whence); HIDDEN int lobject_seek(lobjectObject *self, int pos, int whence);
extern int lobject_tell(lobjectObject *self); HIDDEN int lobject_tell(lobjectObject *self);
extern void lobject_close(lobjectObject *self); HIDDEN int lobject_close(lobjectObject *self);
/* exception-raising macros */ /* exception-raising macros */

View File

@ -31,16 +31,31 @@
#ifdef PSYCOPG_EXTENSIONS #ifdef PSYCOPG_EXTENSIONS
static void
collect_error(connectionObject *conn, char **error)
{
const char *msg = PQerrorMessage(conn->pgconn);
if (msg)
*error = strdup(msg);
}
/* lobject_open - create a new/open an existing lo */ /* lobject_open - create a new/open an existing lo */
int int
lobject_open(lobjectObject *self, connectionObject *conn, lobject_open(lobjectObject *self, connectionObject *conn,
Oid oid, int mode, Oid new_oid, char *new_file) Oid oid, int mode, Oid new_oid, char *new_file)
{ {
int retvalue = -1;
PGresult *pgres = NULL;
char *error = NULL;
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
pq_begin(self->conn); retvalue = pq_begin_locked(self->conn, &pgres, &error);
if (retvalue < 0)
goto end;
/* if the oid is InvalidOid we create a new lob before opening it /* if the oid is InvalidOid we create a new lob before opening it
or we import a file from the FS, depending on the value of or we import a file from the FS, depending on the value of
@ -54,7 +69,11 @@ lobject_open(lobjectObject *self, connectionObject *conn,
Dprintf("lobject_open: large object created with oid = %d", Dprintf("lobject_open: large object created with oid = %d",
self->oid); self->oid);
if (self->oid == InvalidOid) goto end; if (self->oid == InvalidOid) {
collect_error(self->conn, &error);
retvalue = -1;
goto end;
}
mode = INV_WRITE; mode = INV_WRITE;
} }
@ -69,24 +88,19 @@ lobject_open(lobjectObject *self, connectionObject *conn,
self->fd = lo_open(self->conn->pgconn, self->oid, mode); self->fd = lo_open(self->conn->pgconn, self->oid, mode);
Dprintf("lobject_open: large object opened with fd = %d", Dprintf("lobject_open: large object opened with fd = %d",
self->fd); self->fd);
if (self->fd == -1) {
collect_error(self->conn, &error);
retvalue = -1;
goto end;
}
} }
else { else {
/* this is necessary to make sure no function that needs and /* this is necessary to make sure no function that needs and
fd is called on unopened lobjects */ fd is called on unopened lobjects */
self->closed = 1; self->closed = 1;
} }
/* set the mode for future reference */
end:
pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS;
/* here we check for errors before returning 0 */
if ((self->fd == -1 && mode != -1) || self->oid == InvalidOid) {
pq_raise(conn, NULL, NULL, NULL);
return -1;
}
else {
/* set the mode for future reference and return */
self->mode = mode; self->mode = mode;
switch (mode) { switch (mode) {
case -1: case -1:
@ -98,8 +112,54 @@ lobject_open(lobjectObject *self, connectionObject *conn,
case INV_READ+INV_WRITE: case INV_READ+INV_WRITE:
self->smode = "rw"; break; self->smode = "rw"; break;
} }
return 0; retvalue = 0;
end:
pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS;
if (retvalue < 0)
pq_complete_error(self->conn, &pgres, &error);
return retvalue;
} }
/* lobject_close - close an existing lo */
static int
lobject_close_locked(lobjectObject *self, char **error)
{
int retvalue;
if (self->conn->isolation_level == 0 ||
self->conn->mark != self->mark ||
self->fd == -1)
return 0;
retvalue = lo_close(self->conn->pgconn, self->fd);
self->fd = -1;
if (retvalue < 0)
collect_error(self->conn, error);
return retvalue;
}
int
lobject_close(lobjectObject *self)
{
char *error = NULL;
int retvalue;
Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock));
retvalue = lobject_close_locked(self, &error);
pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS;
if (retvalue < 0)
pq_complete_error(self->conn, NULL, &error);
return retvalue;
} }
/* lobject_unlink - remove an lo from database */ /* lobject_unlink - remove an lo from database */
@ -107,76 +167,79 @@ lobject_open(lobjectObject *self, connectionObject *conn,
int int
lobject_unlink(lobjectObject *self) lobject_unlink(lobjectObject *self)
{ {
int res; PGresult *pgres = NULL;
char *error = NULL;
/* first we make sure the lobject is closed and then we unlink */ int retvalue = -1;
lobject_close(self);
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
pq_begin(self->conn); retvalue = pq_begin_locked(self->conn, &pgres, &error);
if (retvalue < 0)
goto end;
res = lo_unlink(self->conn->pgconn, self->oid); /* first we make sure the lobject is closed and then we unlink */
retvalue = lobject_close_locked(self, &error);
if (retvalue < 0)
goto end;
retvalue = lo_unlink(self->conn->pgconn, self->oid);
if (retvalue < 0)
collect_error(self->conn, &error);
end:
pthread_mutex_unlock(&(self->conn->lock)); pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS; Py_END_ALLOW_THREADS;
if (res == -1) if (retvalue < 0)
pq_raise(self->conn, NULL, NULL, NULL); pq_complete_error(self->conn, &pgres, &error);
return res; return retvalue;
}
/* lobject_close - close an existing lo */
void
lobject_close(lobjectObject *self)
{
if (self->conn->isolation_level > 0
&& self->conn->mark == self->mark) {
if (self->fd != -1)
lo_close(self->conn->pgconn, self->fd);
}
} }
/* lobject_write - write bytes to a lo */ /* lobject_write - write bytes to a lo */
size_t Py_ssize_t
lobject_write(lobjectObject *self, char *buf, size_t len) lobject_write(lobjectObject *self, char *buf, size_t len)
{ {
size_t written; Py_ssize_t written;
char *error = NULL;
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
written = lo_write(self->conn->pgconn, self->fd, buf, len); written = lo_write(self->conn->pgconn, self->fd, buf, len);
if (written < 0)
collect_error(self->conn, &error);
pthread_mutex_unlock(&(self->conn->lock)); pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS; Py_END_ALLOW_THREADS;
if (written < 0) if (written < 0)
pq_raise(self->conn, NULL, NULL, NULL); pq_complete_error(self->conn, NULL, &error);
return written; return written;
} }
/* lobject_read - read bytes from a lo */ /* lobject_read - read bytes from a lo */
size_t Py_ssize_t
lobject_read(lobjectObject *self, char *buf, size_t len) lobject_read(lobjectObject *self, char *buf, size_t len)
{ {
size_t readed; Py_ssize_t n_read;
char *error = NULL;
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
readed = lo_read(self->conn->pgconn, self->fd, buf, len); n_read = lo_read(self->conn->pgconn, self->fd, buf, len);
if (n_read < 0)
collect_error(self->conn, &error);
pthread_mutex_unlock(&(self->conn->lock)); pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS; Py_END_ALLOW_THREADS;
if (readed < 0) if (n_read < 0)
pq_raise(self->conn, NULL, NULL, NULL); pq_complete_error(self->conn, NULL, &error);
return readed; return n_read;
} }
/* lobject_seek - move the current position in the lo */ /* lobject_seek - move the current position in the lo */
@ -185,17 +248,20 @@ int
lobject_seek(lobjectObject *self, int pos, int whence) lobject_seek(lobjectObject *self, int pos, int whence)
{ {
int where; int where;
char *error = NULL;
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
where = lo_lseek(self->conn->pgconn, self->fd, pos, whence); where = lo_lseek(self->conn->pgconn, self->fd, pos, whence);
if (where < 0)
collect_error(self->conn, &error);
pthread_mutex_unlock(&(self->conn->lock)); pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS; Py_END_ALLOW_THREADS;
if (where < 0) if (where < 0)
pq_raise(self->conn, NULL, NULL, NULL); pq_complete_error(self->conn, NULL, &error);
return where; return where;
} }
@ -205,17 +271,20 @@ int
lobject_tell(lobjectObject *self) lobject_tell(lobjectObject *self)
{ {
int where; int where;
char *error = NULL;
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
where = lo_tell(self->conn->pgconn, self->fd); where = lo_tell(self->conn->pgconn, self->fd);
if (where < 0)
collect_error(self->conn, &error);
pthread_mutex_unlock(&(self->conn->lock)); pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS; Py_END_ALLOW_THREADS;
if (where < 0) if (where < 0)
pq_raise(self->conn, NULL, NULL, NULL); pq_complete_error(self->conn, NULL, &error);
return where; return where;
} }
@ -225,17 +294,20 @@ int
lobject_export(lobjectObject *self, char *filename) lobject_export(lobjectObject *self, char *filename)
{ {
int res; int res;
char *error = NULL;
Py_BEGIN_ALLOW_THREADS; Py_BEGIN_ALLOW_THREADS;
pthread_mutex_lock(&(self->conn->lock)); pthread_mutex_lock(&(self->conn->lock));
res = lo_export(self->conn->pgconn, self->oid, filename); res = lo_export(self->conn->pgconn, self->oid, filename);
if (res < 0)
collect_error(self->conn, &error);
pthread_mutex_unlock(&(self->conn->lock)); pthread_mutex_unlock(&(self->conn->lock));
Py_END_ALLOW_THREADS; Py_END_ALLOW_THREADS;
if (res < 0) if (res < 0)
pq_raise(self->conn, NULL, NULL, NULL); pq_complete_error(self->conn, NULL, &error);
return res; return res;
} }

View File

@ -57,7 +57,8 @@ psyco_lobj_close(lobjectObject *self, PyObject *args)
&& self->conn->mark == self->mark) && self->conn->mark == self->mark)
{ {
self->closed = 1; self->closed = 1;
lobject_close(self); if (lobject_close(self) < 0)
return NULL;
Dprintf("psyco_lobj_close: lobject at %p closed", self); Dprintf("psyco_lobj_close: lobject at %p closed", self);
} }
@ -276,7 +277,8 @@ lobject_dealloc(PyObject* obj)
{ {
lobjectObject *self = (lobjectObject *)obj; lobjectObject *self = (lobjectObject *)obj;
lobject_close(self); if (lobject_close(self) < 0)
PyErr_Print();
Py_XDECREF((PyObject*)self->conn); Py_XDECREF((PyObject*)self->conn);
Dprintf("lobject_dealloc: deleted lobject object at %p, refcnt = %d", Dprintf("lobject_dealloc: deleted lobject object at %p, refcnt = %d",

View File

@ -368,7 +368,7 @@ pq_complete_error(connectionObject *conn, PGresult **pgres, char **error)
On error, -1 is returned, and the pgres argument will hold the On error, -1 is returned, and the pgres argument will hold the
relevant result structure. relevant result structure.
*/ */
static int int
pq_begin_locked(connectionObject *conn, PGresult **pgres, char **error) pq_begin_locked(connectionObject *conn, PGresult **pgres, char **error)
{ {
const char *query[] = { const char *query[] = {

View File

@ -33,6 +33,8 @@
/* exported functions */ /* exported functions */
HIDDEN int pq_fetch(cursorObject *curs); HIDDEN int pq_fetch(cursorObject *curs);
HIDDEN int pq_execute(cursorObject *curs, const char *query, int async); HIDDEN int pq_execute(cursorObject *curs, const char *query, int async);
HIDDEN int pq_begin_locked(connectionObject *conn, PGresult **pgres,
char **error);
HIDDEN int pq_commit(connectionObject *conn); HIDDEN int pq_commit(connectionObject *conn);
HIDDEN int pq_abort_locked(connectionObject *conn, PGresult **pgres, HIDDEN int pq_abort_locked(connectionObject *conn, PGresult **pgres,
char **error); char **error);