2014-03-30 23:54:35 +02:00
|
|
|
/*
|
|
|
|
* Copyright (c) 2014 Joris Vink <joris@coders.se>
|
|
|
|
*
|
|
|
|
* Permission to use, copy, modify, and distribute this software for any
|
|
|
|
* purpose with or without fee is hereby granted, provided that the above
|
|
|
|
* copyright notice and this permission notice appear in all copies.
|
|
|
|
*
|
|
|
|
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
|
|
|
|
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
|
|
|
|
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
|
|
|
|
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
|
|
|
|
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
|
|
|
|
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
|
|
|
|
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
|
|
|
|
*/
|
|
|
|
|
|
|
|
#include <sys/param.h>
|
|
|
|
#include <sys/queue.h>
|
|
|
|
|
|
|
|
#include <libpq-fe.h>
|
2014-07-18 15:59:07 +02:00
|
|
|
#include <pg_config.h>
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
#include "kore.h"
|
|
|
|
#include "http.h"
|
2014-07-03 22:14:46 +02:00
|
|
|
#include "pgsql.h"
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
struct pgsql_job {
|
|
|
|
u_int8_t idx;
|
|
|
|
struct http_request *req;
|
|
|
|
u_int64_t start;
|
|
|
|
char *query;
|
|
|
|
|
|
|
|
TAILQ_ENTRY(pgsql_job) list;
|
|
|
|
};
|
|
|
|
|
|
|
|
#define PGSQL_CONN_MAX 2
|
|
|
|
#define PGSQL_CONN_FREE 0x01
|
|
|
|
|
|
|
|
struct pgsql_conn {
|
|
|
|
u_int8_t type;
|
|
|
|
u_int8_t flags;
|
|
|
|
|
|
|
|
PGconn *db;
|
|
|
|
struct pgsql_job *job;
|
|
|
|
TAILQ_ENTRY(pgsql_conn) list;
|
|
|
|
};
|
|
|
|
|
2014-03-31 01:04:22 +02:00
|
|
|
static void pgsql_conn_cleanup(struct pgsql_conn *);
|
|
|
|
static int pgsql_conn_create(struct http_request *, int);
|
|
|
|
static void pgsql_read_result(struct http_request *, int,
|
|
|
|
struct pgsql_conn *);
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
static TAILQ_HEAD(, pgsql_conn) pgsql_conn_free;
|
|
|
|
static u_int16_t pgsql_conn_count;
|
2014-04-14 08:41:41 +02:00
|
|
|
char *pgsql_conn_string = NULL;
|
2014-07-04 09:14:05 +02:00
|
|
|
u_int16_t pgsql_conn_max = PGSQL_CONN_MAX;
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
void
|
|
|
|
kore_pgsql_init(void)
|
|
|
|
{
|
|
|
|
pgsql_conn_count = 0;
|
|
|
|
TAILQ_INIT(&pgsql_conn_free);
|
|
|
|
}
|
|
|
|
|
|
|
|
int
|
|
|
|
kore_pgsql_query(struct http_request *req, char *query, int idx)
|
|
|
|
{
|
|
|
|
int fd;
|
|
|
|
struct pgsql_conn *conn;
|
|
|
|
|
|
|
|
if (idx >= HTTP_PGSQL_MAX)
|
|
|
|
fatal("kore_pgsql_query: %d > %d", idx, HTTP_PGSQL_MAX);
|
|
|
|
if (req->pgsql[idx] != NULL)
|
|
|
|
fatal("kore_pgsql_query: %d already exists", idx);
|
|
|
|
|
|
|
|
if (TAILQ_EMPTY(&pgsql_conn_free)) {
|
2014-07-04 09:14:05 +02:00
|
|
|
if (pgsql_conn_count >= pgsql_conn_max)
|
2014-03-30 23:54:35 +02:00
|
|
|
return (KORE_RESULT_ERROR);
|
|
|
|
}
|
|
|
|
|
|
|
|
req->pgsql[idx] = kore_malloc(sizeof(struct kore_pgsql));
|
|
|
|
req->pgsql[idx]->state = KORE_PGSQL_STATE_INIT;
|
|
|
|
req->pgsql[idx]->result = NULL;
|
|
|
|
req->pgsql[idx]->error = NULL;
|
2014-04-01 21:43:32 +02:00
|
|
|
req->pgsql[idx]->conn = NULL;
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
if (TAILQ_EMPTY(&pgsql_conn_free)) {
|
|
|
|
if (pgsql_conn_create(req, idx) == KORE_RESULT_ERROR)
|
|
|
|
return (KORE_RESULT_ERROR);
|
|
|
|
}
|
|
|
|
|
2014-07-04 16:51:19 +02:00
|
|
|
http_request_sleep(req);
|
2014-03-30 23:54:35 +02:00
|
|
|
conn = TAILQ_FIRST(&pgsql_conn_free);
|
|
|
|
if (!(conn->flags & PGSQL_CONN_FREE))
|
|
|
|
fatal("received a pgsql conn that was not free?");
|
|
|
|
|
|
|
|
conn->flags &= ~PGSQL_CONN_FREE;
|
|
|
|
TAILQ_REMOVE(&pgsql_conn_free, conn, list);
|
|
|
|
|
2014-03-31 00:57:00 +02:00
|
|
|
req->pgsql[idx]->conn = conn;
|
2014-03-30 23:54:35 +02:00
|
|
|
conn->job = kore_malloc(sizeof(struct pgsql_job));
|
|
|
|
conn->job->query = kore_strdup(query);
|
|
|
|
conn->job->start = kore_time_ms();
|
|
|
|
conn->job->req = req;
|
|
|
|
conn->job->idx = idx;
|
|
|
|
|
|
|
|
if (!PQsendQuery(conn->db, query)) {
|
|
|
|
pgsql_conn_cleanup(conn);
|
|
|
|
return (KORE_RESULT_ERROR);
|
|
|
|
}
|
|
|
|
|
|
|
|
fd = PQsocket(conn->db);
|
|
|
|
if (fd < 0)
|
|
|
|
fatal("PQsocket returned < 0 fd on open connection");
|
|
|
|
|
|
|
|
kore_platform_schedule_read(fd, conn);
|
2014-03-31 00:57:00 +02:00
|
|
|
req->pgsql[idx]->state = KORE_PGSQL_STATE_WAIT;
|
2014-03-30 23:54:35 +02:00
|
|
|
kore_debug("query '%s' for %p sent on %p", query, req, conn);
|
|
|
|
|
|
|
|
return (KORE_RESULT_OK);
|
|
|
|
}
|
|
|
|
|
|
|
|
void
|
|
|
|
kore_pgsql_handle(void *c, int err)
|
|
|
|
{
|
2014-03-31 00:57:00 +02:00
|
|
|
int i;
|
2014-03-30 23:54:35 +02:00
|
|
|
struct http_request *req;
|
|
|
|
struct pgsql_conn *conn = (struct pgsql_conn *)c;
|
2014-03-31 00:57:00 +02:00
|
|
|
|
|
|
|
if (err) {
|
|
|
|
pgsql_conn_cleanup(conn);
|
|
|
|
return;
|
|
|
|
}
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
i = conn->job->idx;
|
|
|
|
req = conn->job->req;
|
2014-04-17 10:49:48 +02:00
|
|
|
kore_debug("kore_pgsql_handle: %p (%d) (%d)",
|
|
|
|
req, i, req->pgsql[i]->state);
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
if (!PQconsumeInput(conn->db)) {
|
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_ERROR;
|
2014-03-31 00:57:00 +02:00
|
|
|
req->pgsql[i]->error = kore_strdup(PQerrorMessage(conn->db));
|
2014-03-30 23:54:35 +02:00
|
|
|
} else {
|
2014-03-31 01:04:22 +02:00
|
|
|
pgsql_read_result(req, i, conn);
|
2014-03-30 23:54:35 +02:00
|
|
|
}
|
|
|
|
|
2014-04-17 10:49:48 +02:00
|
|
|
if (req->pgsql[i]->state == KORE_PGSQL_STATE_WAIT) {
|
2014-07-04 16:51:19 +02:00
|
|
|
http_request_sleep(req);
|
2014-04-17 10:49:48 +02:00
|
|
|
} else {
|
2014-07-04 16:51:19 +02:00
|
|
|
http_request_wakeup(req);
|
2014-04-17 10:49:48 +02:00
|
|
|
http_process_request(req, 1);
|
|
|
|
}
|
2014-03-31 00:57:00 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
void
|
|
|
|
kore_pgsql_continue(struct http_request *req, int i)
|
|
|
|
{
|
|
|
|
int fd;
|
|
|
|
struct pgsql_conn *conn;
|
|
|
|
|
2014-04-17 10:49:48 +02:00
|
|
|
kore_debug("kore_pgsql_continue: %p->%p (%d) (%d)",
|
|
|
|
req->owner, req, i, req->pgsql[i]->state);
|
|
|
|
|
2014-03-31 00:57:00 +02:00
|
|
|
if (req->pgsql[i]->error) {
|
|
|
|
kore_mem_free(req->pgsql[i]->error);
|
|
|
|
req->pgsql[i]->error = NULL;
|
2014-03-30 23:54:35 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
if (req->pgsql[i]->result)
|
|
|
|
PQclear(req->pgsql[i]->result);
|
|
|
|
|
2014-03-31 00:57:00 +02:00
|
|
|
conn = req->pgsql[i]->conn;
|
2014-03-30 23:54:35 +02:00
|
|
|
switch (req->pgsql[i]->state) {
|
|
|
|
case KORE_PGSQL_STATE_INIT:
|
|
|
|
case KORE_PGSQL_STATE_WAIT:
|
|
|
|
break;
|
|
|
|
case KORE_PGSQL_STATE_DONE:
|
2014-07-04 16:51:19 +02:00
|
|
|
http_request_wakeup(req);
|
2014-03-31 01:09:08 +02:00
|
|
|
req->pgsql[i]->conn = NULL;
|
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_COMPLETE;
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
kore_mem_free(conn->job->query);
|
|
|
|
kore_mem_free(conn->job);
|
|
|
|
|
|
|
|
conn->job = NULL;
|
|
|
|
conn->flags |= PGSQL_CONN_FREE;
|
|
|
|
TAILQ_INSERT_TAIL(&pgsql_conn_free, conn, list);
|
|
|
|
|
|
|
|
fd = PQsocket(conn->db);
|
|
|
|
kore_platform_disable_read(fd);
|
2014-04-17 10:49:48 +02:00
|
|
|
|
|
|
|
http_process_request(req, 0);
|
2014-03-30 23:54:35 +02:00
|
|
|
break;
|
|
|
|
case KORE_PGSQL_STATE_ERROR:
|
|
|
|
case KORE_PGSQL_STATE_RESULT:
|
|
|
|
kore_pgsql_handle(conn, 0);
|
|
|
|
break;
|
|
|
|
default:
|
|
|
|
fatal("unknown pgsql state");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
void
|
|
|
|
kore_pgsql_cleanup(struct http_request *req)
|
|
|
|
{
|
2014-03-31 01:09:08 +02:00
|
|
|
int i;
|
|
|
|
struct pgsql_conn *conn;
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
for (i = 0; i < HTTP_PGSQL_MAX; i++) {
|
|
|
|
if (req->pgsql[i] == NULL)
|
|
|
|
continue;
|
|
|
|
|
|
|
|
if (req->pgsql[i]->result != NULL) {
|
|
|
|
kore_log(LOG_NOTICE, "cleaning up leaked pgsql result");
|
|
|
|
PQclear(req->pgsql[i]->result);
|
|
|
|
}
|
|
|
|
|
2014-03-31 00:57:00 +02:00
|
|
|
if (req->pgsql[i]->error != NULL)
|
|
|
|
kore_mem_free(req->pgsql[i]->error);
|
|
|
|
|
2014-03-31 01:09:08 +02:00
|
|
|
if (req->pgsql[i]->conn != NULL) {
|
|
|
|
conn = req->pgsql[i]->conn;
|
|
|
|
while (PQgetResult(conn->db) != NULL)
|
|
|
|
;
|
|
|
|
}
|
|
|
|
|
2014-04-01 21:43:32 +02:00
|
|
|
req->pgsql[i]->conn = NULL;
|
2014-03-30 23:54:35 +02:00
|
|
|
kore_mem_free(req->pgsql[i]);
|
|
|
|
req->pgsql[i] = NULL;
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2014-04-02 23:01:47 +02:00
|
|
|
void
|
|
|
|
kore_pgsql_logerror(struct kore_pgsql *pgsql)
|
|
|
|
{
|
|
|
|
kore_log(LOG_NOTICE, "pgsql error: %s",
|
|
|
|
(pgsql->error) ? pgsql->error : "unknown");
|
|
|
|
}
|
|
|
|
|
2014-03-30 23:54:35 +02:00
|
|
|
int
|
2014-04-01 21:43:32 +02:00
|
|
|
kore_pgsql_ntuples(struct kore_pgsql *pgsql)
|
2014-03-30 23:54:35 +02:00
|
|
|
{
|
2014-04-01 21:43:32 +02:00
|
|
|
return (PQntuples(pgsql->result));
|
|
|
|
}
|
|
|
|
|
|
|
|
char *
|
|
|
|
kore_pgsql_getvalue(struct kore_pgsql *pgsql, int row, int col)
|
|
|
|
{
|
|
|
|
return (PQgetvalue(pgsql->result, row, col));
|
2014-03-30 23:54:35 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
static int
|
|
|
|
pgsql_conn_create(struct http_request *req, int idx)
|
|
|
|
{
|
|
|
|
struct pgsql_conn *conn;
|
|
|
|
|
2014-04-14 08:45:31 +02:00
|
|
|
if (pgsql_conn_string == NULL)
|
|
|
|
fatal("pgsql_conn_create: no connection string");
|
|
|
|
|
2014-03-30 23:54:35 +02:00
|
|
|
pgsql_conn_count++;
|
|
|
|
conn = kore_malloc(sizeof(*conn));
|
|
|
|
kore_debug("pgsql_conn_create(): %p", conn);
|
|
|
|
memset(conn, 0, sizeof(*conn));
|
|
|
|
|
2014-04-14 08:41:41 +02:00
|
|
|
conn->db = PQconnectdb(pgsql_conn_string);
|
2014-03-30 23:54:35 +02:00
|
|
|
if (conn->db == NULL || (PQstatus(conn->db) != CONNECTION_OK)) {
|
2014-03-31 00:57:00 +02:00
|
|
|
req->pgsql[idx]->state = KORE_PGSQL_STATE_ERROR;
|
|
|
|
req->pgsql[idx]->error = kore_strdup(PQerrorMessage(conn->db));
|
2014-03-30 23:54:35 +02:00
|
|
|
pgsql_conn_cleanup(conn);
|
|
|
|
return (KORE_RESULT_ERROR);
|
|
|
|
}
|
|
|
|
|
|
|
|
conn->job = NULL;
|
|
|
|
conn->flags = PGSQL_CONN_FREE;
|
|
|
|
conn->type = KORE_TYPE_PGSQL_CONN;
|
|
|
|
TAILQ_INSERT_TAIL(&pgsql_conn_free, conn, list);
|
|
|
|
|
|
|
|
return (KORE_RESULT_OK);
|
|
|
|
}
|
|
|
|
|
|
|
|
static void
|
|
|
|
pgsql_conn_cleanup(struct pgsql_conn *conn)
|
|
|
|
{
|
2014-03-31 00:57:00 +02:00
|
|
|
int i;
|
2014-03-30 23:54:35 +02:00
|
|
|
struct http_request *req;
|
|
|
|
|
|
|
|
kore_debug("pgsql_conn_cleanup(): %p", conn);
|
|
|
|
|
|
|
|
if (conn->flags & PGSQL_CONN_FREE)
|
|
|
|
TAILQ_REMOVE(&pgsql_conn_free, conn, list);
|
|
|
|
|
|
|
|
if (conn->job) {
|
|
|
|
i = conn->job->idx;
|
|
|
|
req = conn->job->req;
|
2014-07-04 16:51:19 +02:00
|
|
|
http_request_wakeup(req);
|
2014-03-30 23:54:35 +02:00
|
|
|
|
2014-04-01 21:43:32 +02:00
|
|
|
req->pgsql[i]->conn = NULL;
|
2014-03-30 23:54:35 +02:00
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_ERROR;
|
2014-03-31 00:57:00 +02:00
|
|
|
req->pgsql[i]->error = kore_strdup(PQerrorMessage(conn->db));
|
2014-03-30 23:54:35 +02:00
|
|
|
|
|
|
|
kore_mem_free(conn->job->query);
|
|
|
|
kore_mem_free(conn->job);
|
|
|
|
conn->job = NULL;
|
|
|
|
}
|
|
|
|
|
|
|
|
if (conn->db != NULL)
|
|
|
|
PQfinish(conn->db);
|
|
|
|
|
|
|
|
pgsql_conn_count--;
|
|
|
|
kore_mem_free(conn);
|
|
|
|
}
|
2014-03-31 01:04:22 +02:00
|
|
|
|
|
|
|
static void
|
|
|
|
pgsql_read_result(struct http_request *req, int i, struct pgsql_conn *conn)
|
|
|
|
{
|
|
|
|
if (PQisBusy(conn->db)) {
|
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_WAIT;
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
req->pgsql[i]->result = PQgetResult(conn->db);
|
|
|
|
if (req->pgsql[i]->result == NULL) {
|
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_DONE;
|
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
switch (PQresultStatus(req->pgsql[i]->result)) {
|
|
|
|
case PGRES_COPY_OUT:
|
|
|
|
case PGRES_COPY_IN:
|
|
|
|
case PGRES_NONFATAL_ERROR:
|
|
|
|
case PGRES_COPY_BOTH:
|
|
|
|
break;
|
|
|
|
case PGRES_COMMAND_OK:
|
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_DONE;
|
|
|
|
break;
|
|
|
|
case PGRES_TUPLES_OK:
|
2014-07-03 22:58:04 +02:00
|
|
|
#if PG_VERSION_NUM >= 90200
|
2014-03-31 01:04:22 +02:00
|
|
|
case PGRES_SINGLE_TUPLE:
|
2014-07-03 22:58:04 +02:00
|
|
|
#endif
|
2014-03-31 01:04:22 +02:00
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_RESULT;
|
|
|
|
break;
|
|
|
|
case PGRES_EMPTY_QUERY:
|
|
|
|
case PGRES_BAD_RESPONSE:
|
|
|
|
case PGRES_FATAL_ERROR:
|
|
|
|
req->pgsql[i]->state = KORE_PGSQL_STATE_ERROR;
|
|
|
|
req->pgsql[i]->error =
|
|
|
|
kore_strdup(PQresultErrorMessage(req->pgsql[i]->result));
|
|
|
|
break;
|
|
|
|
}
|
|
|
|
}
|