mirror of https://git.kore.io/kore.git
Several postgresql improvements.
- Make pgsql_conn_count count per database rather then globally. This means you now define the number of clients *per* database registered rather then the number of clients in total of all databases. - In case a connection is in failed transaction state Kore will now automatically rollback the transaction before placing that connection back in the connection pool.
This commit is contained in:
parent
6415670753
commit
7f1a9b8092
|
@ -43,6 +43,8 @@ struct pgsql_conn {
|
||||||
struct pgsql_db {
|
struct pgsql_db {
|
||||||
char *name;
|
char *name;
|
||||||
char *conn_string;
|
char *conn_string;
|
||||||
|
u_int16_t pgsql_conn_max;
|
||||||
|
u_int16_t pgsql_conn_count;
|
||||||
|
|
||||||
LIST_ENTRY(pgsql_db) rlist;
|
LIST_ENTRY(pgsql_db) rlist;
|
||||||
};
|
};
|
||||||
|
|
74
src/pgsql.c
74
src/pgsql.c
|
@ -51,6 +51,7 @@ static void pgsql_conn_release(struct kore_pgsql *);
|
||||||
static void pgsql_conn_cleanup(struct pgsql_conn *);
|
static void pgsql_conn_cleanup(struct pgsql_conn *);
|
||||||
static void pgsql_read_result(struct kore_pgsql *);
|
static void pgsql_read_result(struct kore_pgsql *);
|
||||||
static void pgsql_schedule(struct kore_pgsql *);
|
static void pgsql_schedule(struct kore_pgsql *);
|
||||||
|
static void pgsql_rollback_state(struct kore_pgsql *, void *);
|
||||||
|
|
||||||
static struct pgsql_conn *pgsql_conn_create(struct kore_pgsql *,
|
static struct pgsql_conn *pgsql_conn_create(struct kore_pgsql *,
|
||||||
struct pgsql_db *);
|
struct pgsql_db *);
|
||||||
|
@ -62,13 +63,12 @@ static struct kore_pool pgsql_wait_pool;
|
||||||
static TAILQ_HEAD(, pgsql_conn) pgsql_conn_free;
|
static TAILQ_HEAD(, pgsql_conn) pgsql_conn_free;
|
||||||
static TAILQ_HEAD(, pgsql_wait) pgsql_wait_queue;
|
static TAILQ_HEAD(, pgsql_wait) pgsql_wait_queue;
|
||||||
static LIST_HEAD(, pgsql_db) pgsql_db_conn_strings;
|
static LIST_HEAD(, pgsql_db) pgsql_db_conn_strings;
|
||||||
static u_int16_t pgsql_conn_count;
|
|
||||||
u_int16_t pgsql_conn_max = PGSQL_CONN_MAX;
|
u_int16_t pgsql_conn_max = PGSQL_CONN_MAX;
|
||||||
|
|
||||||
void
|
void
|
||||||
kore_pgsql_sys_init(void)
|
kore_pgsql_sys_init(void)
|
||||||
{
|
{
|
||||||
pgsql_conn_count = 0;
|
|
||||||
TAILQ_INIT(&pgsql_conn_free);
|
TAILQ_INIT(&pgsql_conn_free);
|
||||||
TAILQ_INIT(&pgsql_wait_queue);
|
TAILQ_INIT(&pgsql_wait_queue);
|
||||||
LIST_INIT(&pgsql_db_conn_strings);
|
LIST_INIT(&pgsql_db_conn_strings);
|
||||||
|
@ -179,7 +179,6 @@ kore_pgsql_query(struct kore_pgsql *pgsql, const char *query)
|
||||||
|
|
||||||
if (pgsql->flags & KORE_PGSQL_SYNC) {
|
if (pgsql->flags & KORE_PGSQL_SYNC) {
|
||||||
pgsql->result = PQexec(pgsql->conn->db, query);
|
pgsql->result = PQexec(pgsql->conn->db, query);
|
||||||
|
|
||||||
if ((PQresultStatus(pgsql->result) != PGRES_TUPLES_OK) &&
|
if ((PQresultStatus(pgsql->result) != PGRES_TUPLES_OK) &&
|
||||||
(PQresultStatus(pgsql->result) != PGRES_COMMAND_OK)) {
|
(PQresultStatus(pgsql->result) != PGRES_COMMAND_OK)) {
|
||||||
pgsql_set_error(pgsql, PQerrorMessage(pgsql->conn->db));
|
pgsql_set_error(pgsql, PQerrorMessage(pgsql->conn->db));
|
||||||
|
@ -290,6 +289,7 @@ kore_pgsql_register(const char *dbname, const char *connstring)
|
||||||
|
|
||||||
pgsqldb = kore_malloc(sizeof(*pgsqldb));
|
pgsqldb = kore_malloc(sizeof(*pgsqldb));
|
||||||
pgsqldb->name = kore_strdup(dbname);
|
pgsqldb->name = kore_strdup(dbname);
|
||||||
|
pgsqldb->pgsql_conn_max = pgsql_conn_max;
|
||||||
pgsqldb->conn_string = kore_strdup(connstring);
|
pgsqldb->conn_string = kore_strdup(connstring);
|
||||||
LIST_INSERT_HEAD(&pgsql_db_conn_strings, pgsqldb, rlist);
|
LIST_INSERT_HEAD(&pgsql_db_conn_strings, pgsqldb, rlist);
|
||||||
|
|
||||||
|
@ -432,8 +432,11 @@ kore_pgsql_getvalue(struct kore_pgsql *pgsql, int row, int col)
|
||||||
static struct pgsql_conn *
|
static struct pgsql_conn *
|
||||||
pgsql_conn_next(struct kore_pgsql *pgsql, struct pgsql_db *db)
|
pgsql_conn_next(struct kore_pgsql *pgsql, struct pgsql_db *db)
|
||||||
{
|
{
|
||||||
struct pgsql_conn *conn;
|
PGTransactionStatusType state;
|
||||||
|
struct pgsql_conn *conn;
|
||||||
|
struct kore_pgsql *rollback;
|
||||||
|
|
||||||
|
rescan:
|
||||||
conn = NULL;
|
conn = NULL;
|
||||||
|
|
||||||
TAILQ_FOREACH(conn, &pgsql_conn_free, list) {
|
TAILQ_FOREACH(conn, &pgsql_conn_free, list) {
|
||||||
|
@ -443,8 +446,36 @@ pgsql_conn_next(struct kore_pgsql *pgsql, struct pgsql_db *db)
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (conn != NULL) {
|
||||||
|
state = PQtransactionStatus(conn->db);
|
||||||
|
if (state == PQTRANS_INERROR) {
|
||||||
|
conn->flags &= ~PGSQL_CONN_FREE;
|
||||||
|
TAILQ_REMOVE(&pgsql_conn_free, conn, list);
|
||||||
|
|
||||||
|
rollback = kore_malloc(sizeof(*rollback));
|
||||||
|
kore_pgsql_init(rollback);
|
||||||
|
kore_pgsql_bind_callback(rollback,
|
||||||
|
pgsql_rollback_state, NULL);
|
||||||
|
rollback->flags |= KORE_PGSQL_ASYNC;
|
||||||
|
|
||||||
|
rollback->conn = conn;
|
||||||
|
rollback->conn->job = kore_pool_get(&pgsql_job_pool);
|
||||||
|
rollback->conn->job->pgsql = rollback;
|
||||||
|
|
||||||
|
if (!kore_pgsql_query(rollback, "ROLLBACK")) {
|
||||||
|
kore_pgsql_logerror(rollback);
|
||||||
|
kore_pgsql_cleanup(rollback);
|
||||||
|
kore_free(rollback);
|
||||||
|
pgsql_conn_cleanup(conn);
|
||||||
|
return (NULL);
|
||||||
|
}
|
||||||
|
|
||||||
|
goto rescan;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if (conn == NULL) {
|
if (conn == NULL) {
|
||||||
if (pgsql_conn_count >= pgsql_conn_max) {
|
if (db->pgsql_conn_count >= db->pgsql_conn_max) {
|
||||||
if (pgsql->flags & KORE_PGSQL_ASYNC) {
|
if (pgsql->flags & KORE_PGSQL_ASYNC) {
|
||||||
pgsql_queue_add(pgsql);
|
pgsql_queue_add(pgsql);
|
||||||
} else {
|
} else {
|
||||||
|
@ -563,7 +594,7 @@ pgsql_conn_create(struct kore_pgsql *pgsql, struct pgsql_db *db)
|
||||||
if (db == NULL || db->conn_string == NULL)
|
if (db == NULL || db->conn_string == NULL)
|
||||||
fatal("pgsql_conn_create: no connection string");
|
fatal("pgsql_conn_create: no connection string");
|
||||||
|
|
||||||
pgsql_conn_count++;
|
db->pgsql_conn_count++;
|
||||||
|
|
||||||
conn = kore_malloc(sizeof(*conn));
|
conn = kore_malloc(sizeof(*conn));
|
||||||
conn->job = NULL;
|
conn->job = NULL;
|
||||||
|
@ -588,6 +619,7 @@ static void
|
||||||
pgsql_conn_release(struct kore_pgsql *pgsql)
|
pgsql_conn_release(struct kore_pgsql *pgsql)
|
||||||
{
|
{
|
||||||
int fd;
|
int fd;
|
||||||
|
PGresult *result;
|
||||||
|
|
||||||
if (pgsql->conn == NULL)
|
if (pgsql->conn == NULL)
|
||||||
return;
|
return;
|
||||||
|
@ -605,8 +637,8 @@ pgsql_conn_release(struct kore_pgsql *pgsql)
|
||||||
}
|
}
|
||||||
|
|
||||||
/* Drain just in case. */
|
/* Drain just in case. */
|
||||||
while (PQgetResult(pgsql->conn->db) != NULL)
|
while ((result = PQgetResult(pgsql->conn->db)) != NULL)
|
||||||
;
|
PQclear(result);
|
||||||
|
|
||||||
pgsql->conn->job = NULL;
|
pgsql->conn->job = NULL;
|
||||||
pgsql->conn->flags |= PGSQL_CONN_FREE;
|
pgsql->conn->flags |= PGSQL_CONN_FREE;
|
||||||
|
@ -625,6 +657,7 @@ static void
|
||||||
pgsql_conn_cleanup(struct pgsql_conn *conn)
|
pgsql_conn_cleanup(struct pgsql_conn *conn)
|
||||||
{
|
{
|
||||||
struct kore_pgsql *pgsql;
|
struct kore_pgsql *pgsql;
|
||||||
|
struct pgsql_db *pgsqldb;
|
||||||
|
|
||||||
kore_debug("pgsql_conn_cleanup(): %p", conn);
|
kore_debug("pgsql_conn_cleanup(): %p", conn);
|
||||||
|
|
||||||
|
@ -647,7 +680,13 @@ pgsql_conn_cleanup(struct pgsql_conn *conn)
|
||||||
if (conn->db != NULL)
|
if (conn->db != NULL)
|
||||||
PQfinish(conn->db);
|
PQfinish(conn->db);
|
||||||
|
|
||||||
pgsql_conn_count--;
|
LIST_FOREACH(pgsqldb, &pgsql_db_conn_strings, rlist) {
|
||||||
|
if (strcmp(pgsqldb->name, conn->name)) {
|
||||||
|
pgsqldb->pgsql_conn_count--;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
kore_free(conn->name);
|
kore_free(conn->name);
|
||||||
kore_free(conn);
|
kore_free(conn);
|
||||||
}
|
}
|
||||||
|
@ -701,3 +740,20 @@ pgsql_cancel(struct kore_pgsql *pgsql)
|
||||||
PQfreeCancel(cancel);
|
PQfreeCancel(cancel);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static void
|
||||||
|
pgsql_rollback_state(struct kore_pgsql *pgsql, void *arg)
|
||||||
|
{
|
||||||
|
switch (pgsql->state) {
|
||||||
|
case KORE_PGSQL_STATE_ERROR:
|
||||||
|
kore_pgsql_logerror(pgsql);
|
||||||
|
kore_pgsql_cleanup(pgsql);
|
||||||
|
break;
|
||||||
|
case KORE_PGSQL_STATE_COMPLETE:
|
||||||
|
kore_pgsql_cleanup(pgsql);
|
||||||
|
break;
|
||||||
|
default:
|
||||||
|
kore_pgsql_continue(pgsql);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
Loading…
Reference in New Issue