[asterisk-commits] tilghman: branch tilghman/realtime_failover r177349 - in /team/tilghman/realt...
SVN commits to the Asterisk project
asterisk-commits at lists.digium.com
Wed Feb 18 20:35:44 CST 2009
Author: tilghman
Date: Wed Feb 18 20:35:43 2009
New Revision: 177349
URL: http://svn.digium.com/svn-view/asterisk?view=rev&rev=177349
Log:
Merged revisions 177320 via svnmerge from
https://origsvn.digium.com/svn/asterisk/trunk
........
r177320 | tilghman | 2009-02-18 18:26:01 -0600 (Wed, 18 Feb 2009) | 2 lines
ODBC transaction support
........
Modified:
team/tilghman/realtime_failover/ (props changed)
team/tilghman/realtime_failover/CHANGES
team/tilghman/realtime_failover/configs/res_odbc.conf.sample
team/tilghman/realtime_failover/funcs/func_odbc.c
team/tilghman/realtime_failover/include/asterisk/res_odbc.h
team/tilghman/realtime_failover/res/res_config_odbc.c
team/tilghman/realtime_failover/res/res_odbc.c
Propchange: team/tilghman/realtime_failover/
------------------------------------------------------------------------------
automerge = *
Propchange: team/tilghman/realtime_failover/
------------------------------------------------------------------------------
--- svnmerge-integrated (original)
+++ svnmerge-integrated Wed Feb 18 20:35:43 2009
@@ -1,1 +1,1 @@
-/trunk:1-177316
+/trunk:1-177348
Modified: team/tilghman/realtime_failover/CHANGES
URL: http://svn.digium.com/svn-view/asterisk/team/tilghman/realtime_failover/CHANGES?view=diff&rev=177349&r1=177348&r2=177349
==============================================================================
--- team/tilghman/realtime_failover/CHANGES (original)
+++ team/tilghman/realtime_failover/CHANGES Wed Feb 18 20:35:43 2009
@@ -94,6 +94,7 @@
of "core show function AES_ENCRYPT" from the CLI
* Added AES_DECRYPT. For information on its use, please see the output
of "core show function AES_DECRYPT" from the CLI
+ * func_odbc now supports database transactions across multiple queries.
Applications
------------
Modified: team/tilghman/realtime_failover/configs/res_odbc.conf.sample
URL: http://svn.digium.com/svn-view/asterisk/team/tilghman/realtime_failover/configs/res_odbc.conf.sample?view=diff&rev=177349&r1=177348&r2=177349
==============================================================================
--- team/tilghman/realtime_failover/configs/res_odbc.conf.sample (original)
+++ team/tilghman/realtime_failover/configs/res_odbc.conf.sample Wed Feb 18 20:35:43 2009
@@ -47,6 +47,13 @@
password => thegrouch
pre-connect => yes
sanitysql => select count(*) from systables
+; forcecommit => no ; Default to committing uncommitted transactions?
+; isolation => read_committed ; Isolation level; supported levels are:
+ ; read_uncommitted, read_committed, repeatable_read,
+ ; serializable. Note that not all databases support
+ ; all isolation levels (e.g. Postgres only supports
+ ; repeatable_read and serializable). See database
+ ; documentation for further information.
;
; Many databases have a default of '\' to escape special characters. MS SQL
; Server does not.
Modified: team/tilghman/realtime_failover/funcs/func_odbc.c
URL: http://svn.digium.com/svn-view/asterisk/team/tilghman/realtime_failover/funcs/func_odbc.c?view=diff&rev=177349&r1=177348&r2=177349
==============================================================================
--- team/tilghman/realtime_failover/funcs/func_odbc.c (original)
+++ team/tilghman/realtime_failover/funcs/func_odbc.c Wed Feb 18 20:35:43 2009
@@ -2,6 +2,7 @@
* Asterisk -- An open source telephony toolkit.
*
* Copyright (c) 2005, 2006 Tilghman Lesher
+ * Copyright (c) 2008 Digium, Inc.
*
* Tilghman Lesher <func_odbc__200508 at the-tilghman.com>
*
@@ -205,6 +206,7 @@
struct acf_odbc_query *query;
char *t, varname[15];
int i, dsn, bogus_chan = 0;
+ int transactional = 0;
AST_DECLARE_APP_ARGS(values,
AST_APP_ARG(field)[100];
);
@@ -293,16 +295,32 @@
}
pbx_builtin_setvar_helper(chan, "VALUE", NULL);
+ /*!\note
+ * Okay, this part is confusing. Transactions belong to a single database
+ * handle. Therefore, when working with transactions, we CANNOT failover
+ * to multiple DSNs. We MUST have a single handle all the way through the
+ * transaction, or else we CANNOT enforce atomicity.
+ */
for (dsn = 0; dsn < 5; dsn++) {
+ if (transactional) {
+ /* This can only happen second time through or greater. */
+ ast_log(LOG_WARNING, "Transactions do not work well with multiple DSNs for 'writehandle'\n");
+ }
+
if (!ast_strlen_zero(query->writehandle[dsn])) {
- obj = ast_odbc_request_obj(query->writehandle[dsn], 0);
- if (obj)
- stmt = ast_odbc_direct_execute(obj, generic_execute, ast_str_buffer(buf));
- }
- if (stmt) {
- status = "SUCCESS";
- SQLRowCount(stmt, &rows);
- break;
+ if ((obj = ast_odbc_retrieve_transaction_obj(chan, query->writehandle[dsn]))) {
+ transactional = 1;
+ } else {
+ obj = ast_odbc_request_obj(query->writehandle[dsn], 0);
+ transactional = 0;
+ }
+ if (obj && (stmt = ast_odbc_direct_execute(obj, generic_execute, ast_str_buffer(buf)))) {
+ break;
+ }
+ }
+
+ if (obj && !transactional) {
+ ast_odbc_release_obj(obj);
}
}
@@ -322,6 +340,9 @@
break;
}
}
+ } else if (stmt) {
+ status = "SUCCESS";
+ SQLRowCount(stmt, &rows);
}
AST_RWLIST_UNLOCK(&queries);
@@ -338,7 +359,7 @@
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
}
- if (obj) {
+ if (obj && !transactional) {
ast_odbc_release_obj(obj);
obj = NULL;
}
Modified: team/tilghman/realtime_failover/include/asterisk/res_odbc.h
URL: http://svn.digium.com/svn-view/asterisk/team/tilghman/realtime_failover/include/asterisk/res_odbc.h?view=diff&rev=177349&r1=177348&r2=177349
==============================================================================
--- team/tilghman/realtime_failover/include/asterisk/res_odbc.h (original)
+++ team/tilghman/realtime_failover/include/asterisk/res_odbc.h Wed Feb 18 20:35:43 2009
@@ -35,19 +35,28 @@
typedef enum { ODBC_SUCCESS=0, ODBC_FAIL=-1} odbc_status;
+/*! \brief Flags for use with ast_odbc_request_obj2 */
+enum {
+ RES_ODBC_SANITY_CHECK = (1 << 0),
+ RES_ODBC_INDEPENDENT_CONNECTION = (1 << 1),
+ RES_ODBC_CONNECTED = (1 << 2),
+};
+
/*! \brief ODBC container */
struct odbc_obj {
ast_mutex_t lock;
- SQLHDBC con; /* ODBC Connection Handle */
- struct odbc_class *parent; /* Information about the connection is protected */
- struct timeval last_used;
+ SQLHDBC con; /*!< ODBC Connection Handle */
+ struct odbc_class *parent; /*!< Information about the connection is protected */
+ struct timeval last_used; /*!< Used by idlecheck to determine if the connection should be renegotiated */
#ifdef DEBUG_THREADS
char file[80];
char function[80];
int lineno;
#endif
- unsigned int used:1;
+ unsigned int used:1; /*!< Is this connection currently in use? */
unsigned int up:1;
+ unsigned int tx:1; /*!< Should this connection be unshared, regardless of the class setting? */
+ struct odbc_txn_frame *txf; /*!< Reference back to the transaction frame, if applicable */
AST_LIST_ENTRY(odbc_obj) list;
};
@@ -96,9 +105,10 @@
* \brief Retrieves a connected ODBC object
* \param name The name of the ODBC class for which a connection is needed.
* \param flags One or more of the following flags:
- * ODBC_SANITY_CHECK Whether to ensure that a connection is valid before returning the handle. Usually unnecessary.
- * ODBC_CONNECTED Only return a connected handle. Intended for use with peers which use idlecheck, which are checked periodically for reachability.
- * \retval ODBC object
+ * \li RES_ODBC_SANITY_CHECK Whether to ensure that a connection is valid before returning the handle. Usually unnecessary.
+ * \li RES_ODBC_INDEPENDENT_CONNECTION Return a handle which is independent from all others. Usually used when starting a transaction.
+ * \li RES_ODBC_CONNECTED Only return a connected handle. Intended for use with peers which use idlecheck, which are checked periodically for reachability.
+ * \return ODBC object
* \retval NULL if there is no connection available with the requested name.
*
* Connection classes may, in fact, contain multiple connection handles. If
@@ -107,14 +117,30 @@
* when the thread is done by calling ast_odbc_release_obj(), below.
*/
#ifdef DEBUG_THREADS
-struct odbc_obj *_ast_odbc_request_obj(const char *name, int flags, const char *file, const char *function, int lineno);
+struct odbc_obj *_ast_odbc_request_obj2(const char *name, struct ast_flags flags, const char *file, const char *function, int lineno);
+#define ast_odbc_request_obj2(a, b) _ast_odbc_request_obj2(a, b, __FILE__, __PRETTY_FUNCTION__, __LINE__)
+#else
+struct odbc_obj *ast_odbc_request_obj2(const char *name, struct ast_flags flags);
+#endif
+
+#ifdef DEBUG_THREADS
+struct odbc_obj *_ast_odbc_request_obj(const char *name, int check, const char *file, const char *function, int lineno);
#define ast_odbc_request_obj(a, b) _ast_odbc_request_obj(a, b, __FILE__, __PRETTY_FUNCTION__, __LINE__)
#else
struct odbc_obj *ast_odbc_request_obj(const char *name, int flags);
#endif
-#define ODBC_SANITY_CHECK (1 << 0)
-#define ODBC_CONNECTED (1 << 1)
+/*!
+ * \brief Retrieve a stored ODBC object, if a transaction has been started.
+ * \param chan Channel associated with the transaction.
+ * \param objname Name of the database handle. This name corresponds to the name passed
+ * to ast_odbc_request_obj2 (or formerly, to ast_odbc_request_obj). Note that the
+ * existence of this parameter name explicitly allows for multiple transactions to be open
+ * at once, albeit to different databases.
+ * \retval A stored ODBC object, if a transaction was already started.
+ * \retval NULL, if no transaction yet exists.
+ */
+struct odbc_obj *ast_odbc_retrieve_transaction_obj(struct ast_channel *chan, const char *objname);
/*!
* \brief Releases an ODBC object previously allocated by ast_odbc_request_obj()
Modified: team/tilghman/realtime_failover/res/res_config_odbc.c
URL: http://svn.digium.com/svn-view/asterisk/team/tilghman/realtime_failover/res/res_config_odbc.c?view=diff&rev=177349&r1=177348&r2=177349
==============================================================================
--- team/tilghman/realtime_failover/res/res_config_odbc.c (original)
+++ team/tilghman/realtime_failover/res/res_config_odbc.c Wed Feb 18 20:35:43 2009
@@ -138,7 +138,7 @@
if (!table)
return NULL;
- obj = ast_odbc_request_obj(database, ODBC_CONNECTED);
+ obj = ast_odbc_request_obj(database, RES_ODBC_CONNECTED);
if (!obj) {
ast_log(LOG_ERROR, "No database handle available with the name of '%s' (check res_odbc.conf)\n", database);
@@ -286,7 +286,7 @@
if (!table)
return NULL;
- obj = ast_odbc_request_obj(database, ODBC_CONNECTED);
+ obj = ast_odbc_request_obj(database, RES_ODBC_CONNECTED);
if (!obj)
return NULL;
@@ -422,7 +422,7 @@
return -1;
}
- obj = ast_odbc_request_obj(database, ODBC_CONNECTED);
+ obj = ast_odbc_request_obj(database, RES_ODBC_CONNECTED);
if (!obj) {
ast_odbc_release_table(tableptr);
return -1;
@@ -649,7 +649,7 @@
if (!table)
return -1;
- obj = ast_odbc_request_obj(database, ODBC_CONNECTED);
+ obj = ast_odbc_request_obj(database, RES_ODBC_CONNECTED);
if (!obj)
return -1;
@@ -723,7 +723,7 @@
if (!table)
return -1;
- obj = ast_odbc_request_obj(database, ODBC_CONNECTED);
+ obj = ast_odbc_request_obj(database, RES_ODBC_CONNECTED);
if (!obj)
return -1;
@@ -815,7 +815,7 @@
if (!file || !strcmp (file, "res_config_odbc.conf"))
return NULL; /* cant configure myself with myself ! */
- obj = ast_odbc_request_obj(database, ODBC_CONNECTED);
+ obj = ast_odbc_request_obj(database, RES_ODBC_CONNECTED);
if (!obj)
return NULL;
Modified: team/tilghman/realtime_failover/res/res_odbc.c
URL: http://svn.digium.com/svn-view/asterisk/team/tilghman/realtime_failover/res/res_odbc.c?view=diff&rev=177349&r1=177348&r2=177349
==============================================================================
--- team/tilghman/realtime_failover/res/res_odbc.c (original)
+++ team/tilghman/realtime_failover/res/res_odbc.c Wed Feb 18 20:35:43 2009
@@ -1,7 +1,7 @@
/*
* Asterisk -- An open source telephony toolkit.
*
- * Copyright (C) 1999 - 2005, Digium, Inc.
+ * Copyright (C) 1999 - 2008, Digium, Inc.
*
* Mark Spencer <markster at digium.com>
*
@@ -25,6 +25,7 @@
*
* \author Mark Spencer <markster at digium.com>
* \author Anthony Minessale II <anthmct at yahoo.com>
+ * \author Tilghman Lesher <tilghman at digium.com>
*
* \arg See also: \ref cdr_odbc
*/
@@ -48,7 +49,69 @@
#include "asterisk/res_odbc.h"
#include "asterisk/time.h"
#include "asterisk/astobj2.h"
+#include "asterisk/app.h"
#include "asterisk/strings.h"
+#include "asterisk/threadstorage.h"
+
+/*** DOCUMENTATION
+ <function name="ODBC" language="en_US">
+ <synopsis>
+ Controls ODBC transaction properties.
+ </synopsis>
+ <syntax>
+ <parameter name="property" required="true">
+ <enumlist>
+ <enum name="transaction">
+ <para>Gets or sets the active transaction ID. If set, and the transaction ID does not
+ exist and a <replaceable>database name</replaceable> is specified as an argument, it will be created.</para>
+ </enum>
+ <enum name="forcecommit">
+ <para>Controls whether a transaction will be automatically committed when the channel
+ hangs up. Defaults to false. If a <replaceable>transaction ID</replaceable> is specified in the optional argument,
+ the property will be applied to that ID, otherwise to the current active ID.</para>
+ </enum>
+ <enum name="isolation">
+ <para>Controls the data isolation on uncommitted transactions. May be one of the
+ following: <literal>read_committed</literal>, <literal>read_uncommitted</literal>,
+ <literal>repeatable_read</literal>, or <literal>serializable</literal>. Defaults to the
+ database setting in <filename>res_odbc.conf</filename> or <literal>read_committed</literal>
+ if not specified. If a <replaceable>transaction ID</replaceable> is specified as an optional argument, it will be
+ applied to that ID, otherwise the current active ID.</para>
+ </enum>
+ </enumlist>
+ </parameter>
+ <parameter name="argument" required="false" />
+ </syntax>
+ <description>
+ <para>The ODBC() function allows setting several properties to influence how a connected
+ database processes transactions.</para>
+ </description>
+ </function>
+ <application name="ODBC_Commit" language="en_US">
+ <synopsis>
+ Commits a currently open database transaction.
+ </synopsis>
+ <syntax>
+ <parameter name="transaction ID" required="no" />
+ </syntax>
+ <description>
+ <para>Commits the database transaction specified by <replaceable>transaction ID</replaceable>
+ or the current active transaction, if not specified.</para>
+ </description>
+ </application>
+ <application name="ODBC_Rollback" language="en_US">
+ <synopsis>
+ Rollback a currently open database transaction.
+ </synopsis>
+ <syntax>
+ <parameter name="transaction ID" required="no" />
+ </syntax>
+ <description>
+ <para>Rolls back the database transaction specified by <replaceable>transaction ID</replaceable>
+ or the current active transaction, if not specified.</para>
+ </description>
+ </application>
+ ***/
struct odbc_class
{
@@ -59,12 +122,14 @@
char *password;
char *sanitysql;
SQLHENV env;
- unsigned int haspool:1; /* Boolean - TDS databases need this */
- unsigned int delme:1; /* Purge the class */
- unsigned int backslash_is_escape:1; /* On this database, the backslash is a native escape sequence */
- unsigned int limit; /* 1023 wasn't enough for some people */
- unsigned int count; /* Running count of pooled connections */
- unsigned int idlecheck; /* Recheck the connection if it is idle for this long */
+ unsigned int haspool:1; /*!< Boolean - TDS databases need this */
+ unsigned int delme:1; /*!< Purge the class */
+ unsigned int backslash_is_escape:1; /*!< On this database, the backslash is a native escape sequence */
+ unsigned int forcecommit:1; /*!< Should uncommitted transactions be auto-committed on handle release? */
+ unsigned int isolation; /*!< Flags for how the DB should deal with data in other, uncommitted transactions */
+ unsigned int limit; /*!< Maximum number of database handles we will allow */
+ int count; /*!< Running count of pooled connections */
+ unsigned int idlecheck; /*!< Recheck the connection if it is idle for this long (in seconds) */
unsigned int conntimeout; /* Maximum time the connection process should take */
/* When a connection fails, cache that failure for how long? */
struct timeval negative_connection_cache;
@@ -81,6 +146,232 @@
static odbc_status odbc_obj_connect(struct odbc_obj *obj);
static odbc_status odbc_obj_disconnect(struct odbc_obj *obj);
static int odbc_register_class(struct odbc_class *class, int connect);
+static void odbc_txn_free(void *data);
+static void odbc_release_obj2(struct odbc_obj *obj, struct odbc_txn_frame *tx);
+
+AST_THREADSTORAGE(errors_buf);
+
+static struct ast_datastore_info txn_info = {
+ .type = "ODBC_Transaction",
+ .destroy = odbc_txn_free,
+};
+
+struct odbc_txn_frame {
+ AST_LIST_ENTRY(odbc_txn_frame) list;
+ struct ast_channel *owner;
+ struct odbc_obj *obj; /*!< Database handle within which transacted statements are run */
+ /*!\brief Is this record the current active transaction within the channel?
+ * Note that the active flag is really only necessary for statements which
+ * are triggered from the dialplan, as there isn't a direct correlation
+ * between multiple statements. Applications wishing to use transactions
+ * may simply perform each statement on the same odbc_obj, which keeps the
+ * transaction persistent.
+ */
+ unsigned int active:1;
+ unsigned int forcecommit:1; /*!< Should uncommitted transactions be auto-committed on handle release? */
+ unsigned int isolation; /*!< Flags for how the DB should deal with data in other, uncommitted transactions */
+ char name[0]; /*!< Name of this transaction ID */
+};
+
+static const char *isolation2text(int iso)
+{
+ if (iso == SQL_TXN_READ_COMMITTED) {
+ return "read_committed";
+ } else if (iso == SQL_TXN_READ_UNCOMMITTED) {
+ return "read_uncommitted";
+ } else if (iso == SQL_TXN_SERIALIZABLE) {
+ return "serializable";
+ } else if (iso == SQL_TXN_REPEATABLE_READ) {
+ return "repeatable_read";
+ } else {
+ return "unknown";
+ }
+}
+
+static int text2isolation(const char *txt)
+{
+ if (strncasecmp(txt, "read_", 5) == 0) {
+ if (strncasecmp(txt + 5, "c", 1) == 0) {
+ return SQL_TXN_READ_COMMITTED;
+ } else if (strncasecmp(txt + 5, "u", 1) == 0) {
+ return SQL_TXN_READ_UNCOMMITTED;
+ } else {
+ return 0;
+ }
+ } else if (strncasecmp(txt, "ser", 3) == 0) {
+ return SQL_TXN_SERIALIZABLE;
+ } else if (strncasecmp(txt, "rep", 3) == 0) {
+ return SQL_TXN_REPEATABLE_READ;
+ } else {
+ return 0;
+ }
+}
+
+static struct odbc_txn_frame *find_transaction(struct ast_channel *chan, struct odbc_obj *obj, const char *name, int active)
+{
+ struct ast_datastore *txn_store;
+ AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
+ struct odbc_txn_frame *txn = NULL;
+
+ if (!chan && obj && obj->txf && obj->txf->owner) {
+ chan = obj->txf->owner;
+ } else if (!chan) {
+ /* No channel == no transaction */
+ return NULL;
+ }
+
+ ast_channel_lock(chan);
+ if ((txn_store = ast_channel_datastore_find(chan, &txn_info, NULL))) {
+ oldlist = txn_store->data;
+ } else {
+ /* Need to create a new datastore */
+ if (!(txn_store = ast_datastore_alloc(&txn_info, NULL))) {
+ ast_log(LOG_ERROR, "Unable to allocate a new datastore. Cannot create a new transaction.\n");
+ ast_channel_unlock(chan);
+ return NULL;
+ }
+
+ if (!(oldlist = ast_calloc(1, sizeof(*oldlist)))) {
+ ast_log(LOG_ERROR, "Unable to allocate datastore list head. Cannot create a new transaction.\n");
+ ast_datastore_free(txn_store);
+ ast_channel_unlock(chan);
+ return NULL;
+ }
+
+ txn_store->data = oldlist;
+ AST_LIST_HEAD_INIT(oldlist);
+ ast_channel_datastore_add(chan, txn_store);
+ }
+
+ AST_LIST_LOCK(oldlist);
+ ast_channel_unlock(chan);
+
+ /* Scanning for an object is *fast*. Scanning for a name is much slower. */
+ if (obj != NULL || active == 1) {
+ AST_LIST_TRAVERSE(oldlist, txn, list) {
+ if (txn->obj == obj || txn->active) {
+ AST_LIST_UNLOCK(oldlist);
+ return txn;
+ }
+ }
+ }
+
+ if (name != NULL) {
+ AST_LIST_TRAVERSE(oldlist, txn, list) {
+ if (!strcasecmp(txn->name, name)) {
+ AST_LIST_UNLOCK(oldlist);
+ return txn;
+ }
+ }
+ }
+
+ /* Nothing found, create one */
+ if (name && obj && (txn = ast_calloc(1, sizeof(*txn) + strlen(name) + 1))) {
+ struct odbc_txn_frame *otxn;
+
+ strcpy(txn->name, name); /* SAFE */
+ txn->obj = obj;
+ txn->isolation = obj->parent->isolation;
+ txn->forcecommit = obj->parent->forcecommit;
+ txn->owner = chan;
+ txn->active = 1;
+
+ /* On creation, the txn becomes active, and all others inactive */
+ AST_LIST_TRAVERSE(oldlist, otxn, list) {
+ otxn->active = 0;
+ }
+ AST_LIST_INSERT_TAIL(oldlist, txn, list);
+
+ obj->txf = txn;
+ obj->tx = 1;
+ }
+ AST_LIST_UNLOCK(oldlist);
+
+ return txn;
+}
+
+static struct odbc_txn_frame *release_transaction(struct odbc_txn_frame *tx)
+{
+ if (!tx) {
+ return NULL;
+ }
+
+ ast_debug(2, "release_transaction(%p) called (tx->obj = %p, tx->obj->txf = %p)\n", tx, tx->obj, tx->obj ? tx->obj->txf : NULL);
+
+ /* If we have an owner, disassociate */
+ if (tx->owner) {
+ struct ast_datastore *txn_store;
+ AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
+
+ ast_channel_lock(tx->owner);
+ if ((txn_store = ast_channel_datastore_find(tx->owner, &txn_info, NULL))) {
+ oldlist = txn_store->data;
+ AST_LIST_LOCK(oldlist);
+ AST_LIST_REMOVE(oldlist, tx, list);
+ AST_LIST_UNLOCK(oldlist);
+ }
+ ast_channel_unlock(tx->owner);
+ tx->owner = NULL;
+ }
+
+ if (tx->obj) {
+ /* If we have any uncommitted transactions, they are handled when we release the object */
+ struct odbc_obj *obj = tx->obj;
+ /* Prevent recursion during destruction */
+ tx->obj->txf = NULL;
+ tx->obj = NULL;
+ odbc_release_obj2(obj, tx);
+ }
+ ast_free(tx);
+ return NULL;
+}
+
+static void odbc_txn_free(void *vdata)
+{
+ struct odbc_txn_frame *tx;
+ AST_LIST_HEAD(, odbc_txn_frame) *oldlist = vdata;
+
+ ast_debug(2, "odbc_txn_free(%p) called\n", vdata);
+
+ AST_LIST_LOCK(oldlist);
+ while ((tx = AST_LIST_REMOVE_HEAD(oldlist, list))) {
+ release_transaction(tx);
+ }
+ AST_LIST_UNLOCK(oldlist);
+ AST_LIST_HEAD_DESTROY(oldlist);
+ ast_free(oldlist);
+}
+
+static int mark_transaction_active(struct ast_channel *chan, struct odbc_txn_frame *tx)
+{
+ struct ast_datastore *txn_store;
+ AST_LIST_HEAD(, odbc_txn_frame) *oldlist;
+ struct odbc_txn_frame *active = NULL, *txn;
+
+ if (!chan && tx && tx->owner) {
+ chan = tx->owner;
+ }
+
+ ast_channel_lock(chan);
+ if (!(txn_store = ast_channel_datastore_find(chan, &txn_info, NULL))) {
+ ast_channel_unlock(chan);
+ return -1;
+ }
+
+ oldlist = txn_store->data;
+ AST_LIST_LOCK(oldlist);
+ AST_LIST_TRAVERSE(oldlist, txn, list) {
+ if (txn == tx) {
+ txn->active = 1;
+ active = txn;
+ } else {
+ txn->active = 0;
+ }
+ }
+ AST_LIST_UNLOCK(oldlist);
+ ast_channel_unlock(chan);
+ return active ? 0 : -1;
+}
static void odbc_class_destructor(void *data)
{
@@ -88,12 +379,15 @@
/* Due to refcounts, we can safely assume that any objects with a reference
* to us will prevent our destruction, so we don't need to worry about them.
*/
- if (class->username)
+ if (class->username) {
ast_free(class->username);
- if (class->password)
+ }
+ if (class->password) {
ast_free(class->password);
- if (class->sanitysql)
+ }
+ if (class->sanitysql) {
ast_free(class->sanitysql);
+ }
ao2_ref(class->obj_container, -1);
SQLFreeHandle(SQL_HANDLE_ENV, class->env);
}
@@ -160,18 +454,18 @@
if (!obj) {
ast_log(LOG_WARNING, "Unable to retrieve database handle for table description '%s@%s'\n", tablename, database);
+ AST_RWLIST_UNLOCK(&odbc_tables);
return NULL;
}
/* Table structure not already cached; build it now. */
do {
-retry:
res = SQLAllocHandle(SQL_HANDLE_STMT, obj->con, &stmt);
if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO)) {
if (try == 0) {
try = 1;
ast_odbc_sanity_check(obj);
- goto retry;
+ continue;
}
ast_log(LOG_WARNING, "SQL Alloc Handle failed on connection '%s'!\n", database);
break;
@@ -183,7 +477,7 @@
try = 1;
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_sanity_check(obj);
- goto retry;
+ continue;
}
ast_log(LOG_ERROR, "Unable to query database columns on connection '%s'.\n", database);
break;
@@ -233,7 +527,8 @@
AST_RWLIST_INSERT_TAIL(&odbc_tables, tableptr, list);
AST_RWLIST_RDLOCK(&(tableptr->columns));
- } while (0);
+ break;
+ } while (1);
AST_RWLIST_UNLOCK(&odbc_tables);
@@ -284,6 +579,9 @@
stmt = exec_cb(obj, data);
if (stmt) {
+ break;
+ } else if (obj->tx) {
+ ast_log(LOG_WARNING, "Failed to execute, but unable to reconnect, as we're transactional.\n");
break;
} else {
obj->up = 0;
@@ -328,22 +626,29 @@
}
}
- ast_log(LOG_WARNING, "SQL Execute error %d! Attempting a reconnect...\n", res);
- SQLFreeHandle(SQL_HANDLE_STMT, stmt);
- stmt = NULL;
-
- obj->up = 0;
- /*
- * While this isn't the best way to try to correct an error, this won't automatically
- * fail when the statement handle invalidates.
- */
- ast_odbc_sanity_check(obj);
- continue;
- } else
+ if (obj->tx) {
+ ast_log(LOG_WARNING, "SQL Execute error, but unable to reconnect, as we're transactional.\n");
+ break;
+ } else {
+ ast_log(LOG_WARNING, "SQL Execute error %d! Attempting a reconnect...\n", res);
+ SQLFreeHandle(SQL_HANDLE_STMT, stmt);
+ stmt = NULL;
+
+ obj->up = 0;
+ /*
+ * While this isn't the best way to try to correct an error, this won't automatically
+ * fail when the statement handle invalidates.
+ */
+ ast_odbc_sanity_check(obj);
+ continue;
+ }
+ } else {
obj->last_used = ast_tvnow();
+ }
break;
- } else if (attempt == 0)
+ } else if (attempt == 0) {
ast_odbc_sanity_check(obj);
+ }
}
return stmt;
@@ -419,7 +724,7 @@
SQLFreeHandle (SQL_HANDLE_STMT, stmt);
}
- if (!obj->up) { /* Try to reconnect! */
+ if (!obj->up && !obj->tx) { /* Try to reconnect! */
ast_log(LOG_WARNING, "Connection is down attempting to reconnect...\n");
odbc_obj_disconnect(obj);
odbc_obj_connect(obj);
@@ -434,7 +739,7 @@
struct ast_variable *v;
char *cat;
const char *dsn, *username, *password, *sanitysql;
- int enabled, pooling, limit, bse, conntimeout;
+ int enabled, pooling, limit, bse, conntimeout, forcecommit, isolation;
struct timeval ncache = { 0, 0 };
unsigned int idlecheck;
int preconnect = 0, res = 0;
@@ -461,6 +766,8 @@
pooling = 0;
limit = 0;
bse = 1;
+ forcecommit = 0;
+ isolation = SQL_TXN_READ_COMMITTED;
for (v = ast_variable_browse(config, cat); v; v = v->next) {
if (!strcasecmp(v->name, "pooling")) {
if (ast_true(v->value))
@@ -511,6 +818,13 @@
ncache.tv_sec = (int)dncache;
ncache.tv_usec = (dncache - ncache.tv_sec) * 1000000;
}
+ } else if (!strcasecmp(v->name, "forcecommit")) {
+ forcecommit = ast_true(v->value);
+ } else if (!strcasecmp(v->name, "isolation")) {
+ if ((isolation = text2isolation(v->value)) == 0) {
+ ast_log(LOG_ERROR, "Unrecognized value for 'isolation': '%s' in section '%s'\n", v->value, cat);
+ isolation = SQL_TXN_READ_COMMITTED;
+ }
}
}
@@ -544,6 +858,8 @@
}
new->backslash_is_escape = bse ? 1 : 0;
+ new->forcecommit = forcecommit ? 1 : 0;
+ new->isolation = isolation;
new->idlecheck = idlecheck;
new->conntimeout = conntimeout;
new->negative_connection_cache = ncache;
@@ -641,10 +957,11 @@
ao2_ref(current, -1);
}
} else {
- /* Should only ever be one of these */
+ /* Should only ever be one of these (unless there are transactions) */
struct ao2_iterator aoi2 = ao2_iterator_init(class->obj_container, 0);
while ((current = ao2_iterator_next(&aoi2))) {
- ast_cli(a->fd, " Pooled: No\n Connected: %s\n", current->up && ast_odbc_sanity_check(current) ? "Yes" : "No");
+ ast_cli(a->fd, " Pooled: No\n Connected: %s\n", current->used ? "In use" :
+ current->up && ast_odbc_sanity_check(current) ? "Yes" : "No");
ao2_ref(current, -1);
}
}
@@ -670,8 +987,9 @@
if (preconnect) {
/* Request and release builds a connection */
obj = ast_odbc_request_obj(class->name, 0);
- if (obj)
+ if (obj) {
ast_odbc_release_obj(obj);
+ }
}
return 0;
@@ -681,65 +999,204 @@
}
}
-void ast_odbc_release_obj(struct odbc_obj *obj)
-{
- /* For pooled connections, this frees the connection to be
- * reused. For non-pooled connections, it does nothing. */
- obj->used = 0;
+static void odbc_release_obj2(struct odbc_obj *obj, struct odbc_txn_frame *tx)
+{
+ SQLINTEGER nativeerror=0, numfields=0;
+ SQLSMALLINT diagbytes=0, i;
+ unsigned char state[10], diagnostic[256];
+
+ ast_debug(2, "odbc_release_obj2(%p) called (obj->txf = %p)\n", obj, obj->txf);
+ if (tx) {
+ ast_debug(1, "called on a transactional handle with %s\n", tx->forcecommit ? "COMMIT" : "ROLLBACK");
+ if (SQLEndTran(SQL_HANDLE_DBC, obj->con, tx->forcecommit ? SQL_COMMIT : SQL_ROLLBACK) == SQL_ERROR) {
+ /* Handle possible transaction commit failure */
+ SQLGetDiagField(SQL_HANDLE_DBC, obj->con, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+ for (i = 0; i < numfields; i++) {
+ SQLGetDiagRec(SQL_HANDLE_DBC, obj->con, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
+ ast_log(LOG_WARNING, "SQLEndTran returned an error: %s: %s\n", state, diagnostic);
+ if (!strcmp((char *)state, "25S02") || !strcmp((char *)state, "08007")) {
+ /* These codes mean that a commit failed and a transaction
+ * is still active. We must rollback, or things will get
+ * very, very weird for anybody using the handle next. */
+ SQLEndTran(SQL_HANDLE_DBC, obj->con, SQL_ROLLBACK);
+ }
+ if (i > 10) {
+ ast_log(LOG_WARNING, "Oh, that was good. There are really %d diagnostics?\n", (int)numfields);
+ break;
+ }
+ }
+ }
+
+ /* Transaction is done, reset autocommit */
+ if (SQLSetConnectAttr(obj->con, SQL_ATTR_AUTOCOMMIT, (void *)SQL_AUTOCOMMIT_ON, 0) == SQL_ERROR) {
+ SQLGetDiagField(SQL_HANDLE_DBC, obj->con, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+ for (i = 0; i < numfields; i++) {
+ SQLGetDiagRec(SQL_HANDLE_DBC, obj->con, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
+ ast_log(LOG_WARNING, "SetConnectAttr (Autocommit) returned an error: %s: %s\n", state, diagnostic);
+ if (i > 10) {
+ ast_log(LOG_WARNING, "Oh, that was good. There are really %d diagnostics?\n", (int)numfields);
+ break;
+ }
+ }
+ }
+ }
+
#ifdef DEBUG_THREADS
obj->file[0] = '\0';
obj->function[0] = '\0';
obj->lineno = 0;
#endif
+
+ /* For pooled connections, this frees the connection to be
+ * reused. For non-pooled connections, it does nothing. */
+ obj->used = 0;
+ if (obj->txf) {
+ /* Prevent recursion -- transaction is already closed out. */
+ obj->txf->obj = NULL;
+ obj->txf = release_transaction(obj->txf);
+ }
ao2_ref(obj, -1);
}
+void ast_odbc_release_obj(struct odbc_obj *obj)
+{
+ struct odbc_txn_frame *tx = find_transaction(NULL, obj, NULL, 0);
+ odbc_release_obj2(obj, tx);
+}
+
int ast_odbc_backslash_is_escape(struct odbc_obj *obj)
{
return obj->parent->backslash_is_escape;
}
+static int commit_exec(struct ast_channel *chan, void *data)
+{
+ struct odbc_txn_frame *tx;
+ SQLINTEGER nativeerror=0, numfields=0;
+ SQLSMALLINT diagbytes=0, i;
+ unsigned char state[10], diagnostic[256];
+
+ if (ast_strlen_zero(data)) {
+ tx = find_transaction(chan, NULL, NULL, 1);
+ } else {
+ tx = find_transaction(chan, NULL, data, 0);
+ }
+
+ pbx_builtin_setvar_helper(chan, "COMMIT_RESULT", "OK");
+
+ if (tx) {
+ if (SQLEndTran(SQL_HANDLE_DBC, tx->obj->con, SQL_COMMIT) == SQL_ERROR) {
+ struct ast_str *errors = ast_str_thread_get(&errors_buf, 16);
+ ast_str_reset(errors);
+
+ /* Handle possible transaction commit failure */
+ SQLGetDiagField(SQL_HANDLE_DBC, tx->obj->con, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+ for (i = 0; i < numfields; i++) {
+ SQLGetDiagRec(SQL_HANDLE_DBC, tx->obj->con, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
+ ast_str_append(&errors, 0, "%s%s", ast_str_strlen(errors) ? "," : "", state);
+ ast_log(LOG_WARNING, "SQLEndTran returned an error: %s: %s\n", state, diagnostic);
+ if (i > 10) {
+ ast_log(LOG_WARNING, "Oh, that was good. There are really %d diagnostics?\n", (int)numfields);
+ break;
+ }
+ }
+ pbx_builtin_setvar_helper(chan, "COMMIT_RESULT", ast_str_buffer(errors));
+ }
+ }
+ return 0;
+}
+
+static int rollback_exec(struct ast_channel *chan, void *data)
+{
+ struct odbc_txn_frame *tx;
+ SQLINTEGER nativeerror=0, numfields=0;
+ SQLSMALLINT diagbytes=0, i;
+ unsigned char state[10], diagnostic[256];
+
+ if (ast_strlen_zero(data)) {
+ tx = find_transaction(chan, NULL, NULL, 1);
+ } else {
+ tx = find_transaction(chan, NULL, data, 0);
+ }
+
+ pbx_builtin_setvar_helper(chan, "ROLLBACK_RESULT", "OK");
+
+ if (tx) {
+ if (SQLEndTran(SQL_HANDLE_DBC, tx->obj->con, SQL_ROLLBACK) == SQL_ERROR) {
+ struct ast_str *errors = ast_str_thread_get(&errors_buf, 16);
+ ast_str_reset(errors);
+
+ /* Handle possible transaction commit failure */
+ SQLGetDiagField(SQL_HANDLE_DBC, tx->obj->con, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+ for (i = 0; i < numfields; i++) {
+ SQLGetDiagRec(SQL_HANDLE_DBC, tx->obj->con, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
+ ast_str_append(&errors, 0, "%s%s", ast_str_strlen(errors) ? "," : "", state);
+ ast_log(LOG_WARNING, "SQLEndTran returned an error: %s: %s\n", state, diagnostic);
+ if (i > 10) {
+ ast_log(LOG_WARNING, "Oh, that was good. There are really %d diagnostics?\n", (int)numfields);
+ break;
+ }
+ }
+ pbx_builtin_setvar_helper(chan, "ROLLBACK_RESULT", ast_str_buffer(errors));
+ }
+ }
+ return 0;
+}
+
+static int aoro2_class_cb(void *obj, void *arg, int flags)
+{
+ struct odbc_class *class = obj;
+ char *name = arg;
+ if (!strcmp(class->name, name) && !class->delme) {
+ return CMP_MATCH | CMP_STOP;
+ }
+ return 0;
+}
+
+#define USE_TX (void *)(long)1
+#define NO_TX (void *)(long)2
+#define EOR_TX (void *)(long)3
+
+static int aoro2_obj_cb(void *vobj, void *arg, int flags)
+{
+ struct odbc_obj *obj = vobj;
+ ast_mutex_lock(&obj->lock);
+ if ((arg == NO_TX && !obj->tx) || (arg == EOR_TX && !obj->used) || (arg == USE_TX && obj->tx && !obj->used)) {
+ obj->used = 1;
+ ast_mutex_unlock(&obj->lock);
+ return CMP_MATCH | CMP_STOP;
+ }
+ ast_mutex_unlock(&obj->lock);
+ return 0;
+}
+
#ifdef DEBUG_THREADS
-struct odbc_obj *_ast_odbc_request_obj(const char *name, int flags, const char *file, const char *function, int lineno)
+struct odbc_obj *_ast_odbc_request_obj2(const char *name, struct ast_flags flags, const char *file, const char *function, int lineno)
#else
-struct odbc_obj *ast_odbc_request_obj(const char *name, int flags)
+struct odbc_obj *ast_odbc_request_obj2(const char *name, struct ast_flags flags)
#endif
{
struct odbc_obj *obj = NULL;
struct odbc_class *class;
- struct ao2_iterator aoi = ao2_iterator_init(class_container, 0);
-
- while ((class = ao2_iterator_next(&aoi))) {
- if (!strcmp(class->name, name) && !class->delme) {
- break;
- }
- ao2_ref(class, -1);
- }
-
- if (!class)
+ SQLINTEGER nativeerror=0, numfields=0;
+ SQLSMALLINT diagbytes=0, i;
+ unsigned char state[10], diagnostic[256];
+
+ if (!(class = ao2_callback(class_container, 0, aoro2_class_cb, (char *) name))) {
return NULL;
+ }
ast_assert(ao2_ref(class, 0) > 1);
if (class->haspool) {
/* Recycle connections before building another */
- aoi = ao2_iterator_init(class->obj_container, 0);
- while ((obj = ao2_iterator_next(&aoi))) {
- if (! obj->used) {
- ast_mutex_lock(&obj->lock);
- obj->used = 1;
- ast_mutex_unlock(&obj->lock);
- break;
- }
- ao2_ref(obj, -1);
- }
+ obj = ao2_callback(class->obj_container, 0, aoro2_obj_cb, EOR_TX);
if (obj) {
ast_assert(ao2_ref(obj, 0) > 1);
}
if (!obj && (class->count < class->limit) &&
ast_tvdiff_ms(ast_tvnow(), ast_tvadd(class->last_negative_connect, class->negative_connection_cache)) > 0) {
- class->count++;
obj = ao2_alloc(sizeof(*obj), odbc_obj_destructor);
if (!obj) {
class->count--;
@@ -750,6 +1207,7 @@
ast_mutex_init(&obj->lock);
/* obj inherits the outstanding reference to class */
obj->parent = class;
+ class = NULL;
if (odbc_obj_connect(obj) == ODBC_FAIL) {
ast_log(LOG_WARNING, "Failed to connect to %s\n", name);
ao2_ref(obj, -1);
@@ -757,31 +1215,73 @@
obj = NULL;
} else {
obj->used = 1;
- ao2_link(class->obj_container, obj);
- }
- class = NULL;
+ ao2_link(obj->parent->obj_container, obj);
+ ast_atomic_fetchadd_int(&obj->parent->count, +1);
+ }
} else {
/* Object is not constructed, so delete outstanding reference to class. */
ao2_ref(class, -1);
class = NULL;
}
+
+ if (obj && ast_test_flag(&flags, RES_ODBC_INDEPENDENT_CONNECTION)) {
+ /* Ensure this connection has autocommit turned off. */
+ if (SQLSetConnectAttr(obj->con, SQL_ATTR_AUTOCOMMIT, (void *)SQL_AUTOCOMMIT_OFF, 0) == SQL_ERROR) {
+ SQLGetDiagField(SQL_HANDLE_DBC, obj->con, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+ for (i = 0; i < numfields; i++) {
+ SQLGetDiagRec(SQL_HANDLE_DBC, obj->con, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
+ ast_log(LOG_WARNING, "SQLSetConnectAttr (Autocommit) returned an error: %s: %s\n", state, diagnostic);
+ if (i > 10) {
+ ast_log(LOG_WARNING, "Oh, that was good. There are really %d diagnostics?\n", (int)numfields);
+ break;
+ }
+ }
+ }
+ }
+ } else if (ast_test_flag(&flags, RES_ODBC_INDEPENDENT_CONNECTION)) {
+ /* Non-pooled connections -- but must use a separate connection handle */
+ if (!(obj = ao2_callback(class->obj_container, 0, aoro2_obj_cb, USE_TX))) {
+ obj = ao2_alloc(sizeof(*obj), odbc_obj_destructor);
+ if (!obj) {
+ ao2_ref(class, -1);
+ return NULL;
+ }
+ ast_mutex_init(&obj->lock);
+ /* obj inherits the outstanding reference to class */
+ obj->parent = class;
+ class = NULL;
+ if (odbc_obj_connect(obj) == ODBC_FAIL) {
+ ast_log(LOG_WARNING, "Failed to connect to %s\n", name);
+ ao2_ref(obj, -1);
+ obj = NULL;
+ } else {
+ obj->used = 1;
+ ao2_link(obj->parent->obj_container, obj);
+ ast_atomic_fetchadd_int(&obj->parent->count, +1);
+ }
+ }
+
+ if (obj && SQLSetConnectAttr(obj->con, SQL_ATTR_AUTOCOMMIT, (void *)SQL_AUTOCOMMIT_OFF, 0) == SQL_ERROR) {
+ SQLGetDiagField(SQL_HANDLE_DBC, obj->con, 1, SQL_DIAG_NUMBER, &numfields, SQL_IS_INTEGER, &diagbytes);
+ for (i = 0; i < numfields; i++) {
+ SQLGetDiagRec(SQL_HANDLE_DBC, obj->con, i + 1, state, &nativeerror, diagnostic, sizeof(diagnostic), &diagbytes);
[... 297 lines stripped ...]
More information about the asterisk-commits
mailing list