dovecot-1.1: pgsql: Don't break when using multiple transactions.
dovecot at dovecot.org
dovecot at dovecot.org
Mon Jan 26 03:27:15 EET 2009
details: http://hg.dovecot.org/dovecot-1.1/rev/b6a79afd18e9
changeset: 8130:b6a79afd18e9
user: Timo Sirainen <tss at iki.fi>
date: Sun Jan 25 20:27:09 2009 -0500
description:
pgsql: Don't break when using multiple transactions.
diffstat:
2 files changed, 122 insertions(+), 64 deletions(-)
src/lib-sql/driver-mysql.c | 4
src/lib-sql/driver-pgsql.c | 182 +++++++++++++++++++++++++++++---------------
diffs (295 lines):
diff -r 2e4623d6a09c -r b6a79afd18e9 src/lib-sql/driver-mysql.c
--- a/src/lib-sql/driver-mysql.c Sun Jan 25 20:13:17 2009 -0500
+++ b/src/lib-sql/driver-mysql.c Sun Jan 25 20:27:09 2009 -0500
@@ -654,10 +654,6 @@ driver_mysql_update(struct sql_transacti
(struct mysql_transaction_context *)_ctx;
struct mysql_query_list *list;
- /* FIXME: with mysql we're just appending everything into one big
- linked list which gets committed in sql_transaction_commit().
- we could avoid this if we knew for sure that transactions actually
- worked, but I don't know how to do that.. */
list = p_new(ctx->query_pool, struct mysql_query_list, 1);
list->query = p_strdup(ctx->query_pool, query);
diff -r 2e4623d6a09c -r b6a79afd18e9 src/lib-sql/driver-pgsql.c
--- a/src/lib-sql/driver-pgsql.c Sun Jan 25 20:13:17 2009 -0500
+++ b/src/lib-sql/driver-pgsql.c Sun Jan 25 20:27:09 2009 -0500
@@ -70,19 +70,29 @@ struct pgsql_queue {
struct pgsql_transaction_context {
struct sql_transaction_context ctx;
+ int refcount;
sql_commit_callback_t *callback;
void *context;
+ pool_t query_pool;
+ struct pgsql_query_list *head, *tail;
const char *error;
- unsigned int opened:1;
+ unsigned int begin_succeeded:1;
+ unsigned int begin_failed:1;
unsigned int failed:1;
};
+struct pgsql_query_list {
+ struct pgsql_query_list *next;
+ const char *query;
+};
extern struct sql_db driver_pgsql_db;
extern struct sql_result driver_pgsql_result;
+static void
+transaction_update_callback(struct sql_result *result, void *context);
static void
driver_pgsql_query_full(struct sql_db *db, const char *query,
sql_query_callback_t *callback, void *context,
@@ -312,7 +322,9 @@ static void result_finish(struct pgsql_r
} T_END;
result->api.callback = FALSE;
free_result = db->sync_result != &result->api;
- }
+ if (db->queue == NULL && db->ioloop != NULL)
+ io_loop_stop(db->ioloop);
+ }
if (disconnected) {
/* disconnected */
@@ -442,6 +454,9 @@ static void queue_abort_next(struct pgsq
i_free(queue->result);
i_free(queue->query);
i_free(queue);
+
+ if (db->queue == NULL && db->ioloop != NULL)
+ io_loop_stop(db->ioloop);
}
static void queue_drop_timed_out_queries(struct pgsql_db *db)
@@ -598,7 +613,6 @@ static void pgsql_query_s_callback(struc
db->query_finished = TRUE;
db->sync_result = result;
- io_loop_stop(db->ioloop);
}
static struct sql_result *
@@ -628,7 +642,8 @@ driver_pgsql_query_s(struct sql_db *_db,
}
db->query_finished = FALSE;
- driver_pgsql_query(_db, query, pgsql_query_s_callback, db);
+ if (query != NULL)
+ driver_pgsql_query(_db, query, pgsql_query_s_callback, db);
if (!db->query_finished) {
if ((db->connected || db->connecting) && db->io != NULL)
@@ -842,7 +857,22 @@ driver_pgsql_transaction_begin(struct sq
ctx = i_new(struct pgsql_transaction_context, 1);
ctx->ctx.db = db;
+ ctx->refcount = 1;
+ /* we need to be able to handle multiple open transactions, so at least
+ for now just keep them in memory until commit time. */
+ ctx->query_pool = pool_alloconly_create("pgsql transaction", 1024);
return &ctx->ctx;
+}
+
+static void
+driver_pgsql_transaction_unref(struct pgsql_transaction_context *ctx)
+{
+ i_assert(ctx->refcount > 0);
+ if (--ctx->refcount > 0)
+ return;
+
+ pool_unref(&ctx->query_pool);
+ i_free(ctx);
}
static void
@@ -855,6 +885,23 @@ transaction_commit_callback(struct sql_r
ctx->callback(sql_result_get_error(result), ctx->context);
else
ctx->callback(NULL, ctx->context);
+ driver_pgsql_transaction_unref(ctx);
+}
+
+static void
+transaction_update_callback(struct sql_result *result, void *context)
+{
+ struct pgsql_transaction_context *ctx = context;
+
+ if (sql_result_next_row(result) < 0) {
+ if (!ctx->begin_succeeded)
+ ctx->begin_failed = TRUE;
+ ctx->failed = TRUE;
+ ctx->error = sql_result_get_error(result);
+ } else {
+ ctx->begin_succeeded = TRUE;
+ }
+ driver_pgsql_transaction_unref(ctx);
}
static void
@@ -864,25 +911,28 @@ driver_pgsql_transaction_commit(struct s
struct pgsql_transaction_context *ctx =
(struct pgsql_transaction_context *)_ctx;
- if (ctx->failed) {
- callback(ctx->error, context);
- if (ctx->opened)
- sql_exec(_ctx->db, "ROLLBACK");
- i_free(ctx);
- return;
- }
-
- if (!ctx->opened) {
- /* nothing done */
- ctx->callback(NULL, ctx->context);
- return;
- }
-
ctx->callback = callback;
ctx->context = context;
- driver_pgsql_query_full(_ctx->db, "COMMIT",
- transaction_commit_callback, ctx, FALSE);
+ if (ctx->failed || ctx->head == NULL) {
+ callback(ctx->failed ? ctx->error : NULL, context);
+ driver_pgsql_transaction_unref(ctx);
+ } else if (ctx->head->next == NULL) {
+ /* just a single query, send it */
+ sql_query(_ctx->db, ctx->head->query,
+ transaction_commit_callback, ctx);
+ } else {
+ /* multiple queries, use a transaction */
+ ctx->refcount++;
+ sql_query(_ctx->db, "BEGIN", transaction_update_callback, ctx);
+ while (ctx->head != NULL) {
+ ctx->refcount++;
+ sql_query(_ctx->db, ctx->head->query,
+ transaction_update_callback, ctx);
+ ctx->head = ctx->head->next;
+ }
+ sql_query(_ctx->db, "COMMIT", transaction_commit_callback, ctx);
+ }
}
static int
@@ -893,24 +943,49 @@ driver_pgsql_transaction_commit_s(struct
(struct pgsql_transaction_context *)_ctx;
struct sql_result *result;
- if (ctx->failed) {
+ *error_r = NULL;
+
+ if (ctx->failed || ctx->head == NULL) {
+ /* nothing to be done */
+ result = NULL;
+ } else if (ctx->head->next == NULL) {
+ /* just a single query, send it */
+ result = sql_query_s(_ctx->db, ctx->head->query);
+ } else {
+ /* multiple queries, use a transaction */
+ ctx->refcount++;
+ sql_query(_ctx->db, "BEGIN", transaction_update_callback, ctx);
+ while (ctx->head != NULL) {
+ ctx->refcount++;
+ sql_query(_ctx->db, ctx->head->query,
+ transaction_update_callback, ctx);
+ ctx->head = ctx->head->next;
+ }
+ if (ctx->refcount > 1) {
+ /* flush the previous queries */
+ (void)driver_pgsql_query_s(_ctx->db, NULL);
+ }
+
+ if (ctx->begin_failed) {
+ result = NULL;
+ } else if (ctx->failed) {
+ result = sql_query_s(_ctx->db, "ROLLBACK");
+ } else {
+ result = sql_query_s(_ctx->db, "COMMIT");
+ }
+ }
+
+ if (ctx->failed)
*error_r = ctx->error;
- if (ctx->opened)
- sql_exec(_ctx->db, "ROLLBACK");
- } else if (!ctx->opened)
- *error_r = NULL;
- else {
- result = sql_query_s(_ctx->db, "COMMIT");
- if (ctx->failed)
- *error_r = ctx->error;
- else if (sql_result_next_row(result) < 0)
+ else if (result != NULL) {
+ if (sql_result_next_row(result) < 0)
*error_r = sql_result_get_error(result);
- else
- *error_r = NULL;
+ }
+ if (result != NULL)
sql_result_free(result);
- }
-
- i_free(ctx);
+
+ i_assert(ctx->refcount == 1);
+ driver_pgsql_transaction_unref(ctx);
return *error_r == NULL ? 0 : -1;
}
@@ -920,20 +995,8 @@ driver_pgsql_transaction_rollback(struct
struct pgsql_transaction_context *ctx =
(struct pgsql_transaction_context *)_ctx;
- if (ctx->opened)
- sql_exec(_ctx->db, "ROLLBACK");
- i_free(ctx);
-}
-
-static void
-transaction_update_callback(struct sql_result *result, void *context)
-{
- struct pgsql_transaction_context *ctx = context;
-
- if (sql_result_next_row(result) < 0) {
- ctx->failed = TRUE;
- ctx->error = sql_result_get_error(result);
- }
+ i_assert(ctx->refcount == 1);
+ driver_pgsql_transaction_unref(ctx);
}
static void
@@ -941,17 +1004,16 @@ driver_pgsql_update(struct sql_transacti
{
struct pgsql_transaction_context *ctx =
(struct pgsql_transaction_context *)_ctx;
-
- if (ctx->failed)
- return;
-
- if (!ctx->opened) {
- ctx->opened = TRUE;
- sql_query(_ctx->db, "BEGIN", transaction_update_callback, ctx);
- }
-
- driver_pgsql_query_full(_ctx->db, query,
- transaction_update_callback, ctx, FALSE);
+ struct pgsql_query_list *list;
+
+ list = p_new(ctx->query_pool, struct pgsql_query_list, 1);
+ list->query = p_strdup(ctx->query_pool, query);
+
+ if (ctx->head == NULL)
+ ctx->head = list;
+ else
+ ctx->tail->next = list;
+ ctx->tail = list;
}
struct sql_db driver_pgsql_db = {
More information about the dovecot-cvs
mailing list