sysmocom/asterisk
11
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge "func_odbc: Use one connection per DSN."

Browse source
This commit is contained in:
Joshua Colp 2016-04-25 05:14:17 -05:00 committed by Gerrit Code Review
commit 7f8d83fef4
1 changed files with 229 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

314
funcs/func_odbc.c
View file

@ -137,6 +137,163 @@ struct odbc_datastore {
char names[0];
};
/* \brief Data source name
*
* This holds data that pertains to a DSN
*/
struct dsn {
/*! A connection to the database */
struct odbc_obj *connection;
/*! The name of the DSN as defined in res_odbc.conf */
char name[0];
};
#define DSN_BUCKETS 37
struct ao2_container *dsns;
static int dsn_hash(const void *obj, const int flags)
{
const struct dsn *object;
const char *key;
switch (flags & OBJ_SEARCH_MASK) {
case OBJ_SEARCH_KEY:
key = obj;
break;
case OBJ_SEARCH_OBJECT:
object = obj;
key = object->name;
break;
default:
ast_assert(0);
return 0;
}
return ast_str_hash(key);
}
static int dsn_cmp(void *obj, void *arg, int flags)
{
const struct dsn *object_left = obj;
const struct dsn *object_right = arg;
const char *right_key = arg;
int cmp;
switch (flags & OBJ_SEARCH_MASK) {
case OBJ_SEARCH_OBJECT:
right_key = object_right->name;
/* Fall through */
case OBJ_SEARCH_KEY:
cmp = strcmp(object_left->name, right_key);
break;
case OBJ_SEARCH_PARTIAL_KEY:
cmp = strncmp(object_left->name, right_key, strlen(right_key));
break;
default:
cmp = 0;
break;
}
if (cmp) {
return 0;
}
return CMP_MATCH;
}
static void dsn_destructor(void *obj)
{
struct dsn *dsn = obj;
if (dsn->connection) {
ast_odbc_release_obj(dsn->connection);
}
}
/*!
* \brief Create a DSN and connect to the database
*
* \param name The name of the DSN as found in res_odbc.conf
* \retval NULL Fail
* \retval non-NULL The newly-created structure
*/
static struct dsn *create_dsn(const char *name)
{
struct dsn *dsn;
dsn = ao2_alloc(sizeof(*dsn) + strlen(name) + 1, dsn_destructor);
if (!dsn) {
return NULL;
}
/* Safe */
strcpy(dsn->name, name);
dsn->connection = ast_odbc_request_obj(name, 0);
if (!dsn->connection) {
ao2_ref(dsn, -1);
return NULL;
}
if (!ao2_link_flags(dsns, dsn, OBJ_NOLOCK)) {
ao2_ref(dsn, -1);
return NULL;
}
return dsn;
}
/*!
* \brief Retrieve a DSN, or create it if it does not exist.
*
* The created DSN is returned locked. This should be inconsequential
* to callers in most cases.
*
* When finished with the returned structure, the caller must call
* \ref release_dsn
*
* \param name Name of the DSN as found in res_odbc.conf
* \retval NULL Unable to retrieve or create the DSN
* \retval non-NULL The retrieved/created locked DSN
*/
static struct dsn *get_dsn(const char *name)
{
struct dsn *dsn;
ao2_lock(dsns);
dsn = ao2_find(dsns, name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
if (!dsn) {
dsn = create_dsn(name);
}
ao2_unlock(dsns);
if (!dsn) {
return NULL;
}
ao2_lock(dsn->connection);
return dsn;
}
/*!
* \brief Unlock and unreference a DSN
*
* \param dsn The dsn to unlock and unreference
* \return NULL
*/
static void *release_dsn(struct dsn *dsn)
{
if (!dsn) {
return NULL;
}
ao2_unlock(dsn->connection);
ao2_ref(dsn, -1);
return NULL;
}
static AST_RWLIST_HEAD_STATIC(queries, acf_odbc_query);
static int resultcount = 0;
@ -214,7 +371,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
struct odbc_obj *obj = NULL;
struct acf_odbc_query *query;
char *t, varname[15];
int i, dsn, bogus_chan = 0;
int i, dsn_num, bogus_chan = 0;
int transactional = 0;
AST_DECLARE_APP_ARGS(values,
AST_APP_ARG(field)[100];
@ -227,6 +384,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
struct ast_str *buf = ast_str_thread_get(&sql_buf, 16);
struct ast_str *insertbuf = ast_str_thread_get(&sql2_buf, 16);
const char *status = "FAILURE";
struct dsn *dsn = NULL;
if (!buf || !insertbuf) {
return -1;
@ -324,17 +482,21 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
* 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 (!ast_strlen_zero(query->writehandle[dsn])) {
for (dsn_num = 0; dsn_num < 5; dsn_num++) {
if (!ast_strlen_zero(query->writehandle[dsn_num])) {
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 ((obj = ast_odbc_retrieve_transaction_obj(chan, query->writehandle[dsn]))) {
if ((obj = ast_odbc_retrieve_transaction_obj(chan, query->writehandle[dsn_num]))) {
transactional = 1;
} else {
obj = ast_odbc_request_obj(query->writehandle[dsn], 0);
dsn = get_dsn(query->writehandle[dsn_num]);
if (!dsn) {
continue;
}
obj = dsn->connection;
transactional = 0;
}
@ -342,10 +504,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
break;
}
if (obj && !transactional) {
ast_odbc_release_obj(obj);
obj = NULL;
}
dsn = release_dsn(dsn);
}
}
@ -358,25 +517,25 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
status = "SUCCESS";
} else if (query->sql_insert) {
if (obj && !transactional) {
ast_odbc_release_obj(obj);
obj = NULL;
}
dsn = release_dsn(dsn);
for (transactional = 0, dsn = 0; dsn < 5; dsn++) {
if (!ast_strlen_zero(query->writehandle[dsn])) {
for (transactional = 0, dsn_num = 0; dsn_num < 5; dsn_num++) {
if (!ast_strlen_zero(query->writehandle[dsn_num])) {
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");
} else if (obj) {
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
}
if ((obj = ast_odbc_retrieve_transaction_obj(chan, query->writehandle[dsn]))) {
if ((obj = ast_odbc_retrieve_transaction_obj(chan, query->writehandle[dsn_num]))) {
transactional = 1;
} else {
obj = ast_odbc_request_obj(query->writehandle[dsn], 0);
dsn = get_dsn(query->writehandle[dsn_num]);
if (!dsn) {
continue;
}
obj = dsn->connection;
transactional = 0;
}
if (obj) {
@ -406,10 +565,7 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
}
if (obj && !transactional) {
ast_odbc_release_obj(obj);
obj = NULL;
}
dsn = release_dsn(dsn);
if (!bogus_chan) {
ast_autoservice_stop(chan);
@ -420,11 +576,10 @@ static int acf_odbc_write(struct ast_channel *chan, const char *cmd, char *s, co
static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, char *buf, size_t len)
{
struct odbc_obj *obj = NULL;
struct acf_odbc_query *query;
char varname[15], rowcount[12] = "-1";
struct ast_str *colnames = ast_str_thread_get(&colnames_buf, 16);
int res, x, y, buflen = 0, escapecommas, rowlimit = 1, multirow = 0, dsn, bogus_chan = 0;
int res, x, y, buflen = 0, escapecommas, rowlimit = 1, multirow = 0, dsn_num, bogus_chan = 0;
AST_DECLARE_APP_ARGS(args,
AST_APP_ARG(field)[100];
);
@ -436,6 +591,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
struct odbc_datastore_row *row = NULL;
struct ast_str *sql = ast_str_thread_get(&sql_buf, 16);
const char *status = "FAILURE";
struct dsn *dsn = NULL;
if (!sql || !colnames) {
if (chan) {
@ -523,28 +679,23 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
}
AST_RWLIST_UNLOCK(&queries);
for (dsn = 0; dsn < 5; dsn++) {
if (!ast_strlen_zero(query->readhandle[dsn])) {
obj = ast_odbc_request_obj(query->readhandle[dsn], 0);
if (obj) {
stmt = ast_odbc_direct_execute(obj, generic_execute, ast_str_buffer(sql));
for (dsn_num = 0; dsn_num < 5; dsn_num++) {
if (!ast_strlen_zero(query->readhandle[dsn_num])) {
dsn = get_dsn(query->readhandle[dsn_num]);
if (!dsn) {
continue;
}
stmt = ast_odbc_direct_execute(dsn->connection, generic_execute, ast_str_buffer(sql));
}
if (stmt) {
break;
}
if (obj) {
ast_odbc_release_obj(obj);
obj = NULL;
}
dsn = release_dsn(dsn);
}
if (!stmt) {
ast_log(LOG_ERROR, "Unable to execute query [%s]\n", ast_str_buffer(sql));
if (obj) {
ast_odbc_release_obj(obj);
obj = NULL;
}
dsn = release_dsn(dsn);
if (!bogus_chan) {
pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
ast_autoservice_stop(chan);
@ -558,8 +709,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
ast_log(LOG_WARNING, "SQL Column Count error!\n[%s]\n\n", ast_str_buffer(sql));
SQLCloseCursor(stmt);
SQLFreeHandle (SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
if (!bogus_chan) {
pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
ast_autoservice_stop(chan);
@ -583,8 +733,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
}
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
if (!bogus_chan) {
pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
pbx_builtin_setvar_helper(chan, "ODBCSTATUS", status);
@ -607,8 +756,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
odbc_datastore_free(resultset);
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
if (!bogus_chan) {
pbx_builtin_setvar_helper(chan, "ODBCSTATUS", "MEMERROR");
ast_autoservice_stop(chan);
@ -640,8 +788,7 @@ static int acf_odbc_read(struct ast_channel *chan, const char *cmd, char *s, cha
odbc_datastore_free(resultset);
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
if (!bogus_chan) {
pbx_builtin_setvar_helper(chan, "ODBCROWS", rowcount);
pbx_builtin_setvar_helper(chan, "ODBCSTATUS", "MEMERROR");
@ -750,8 +897,7 @@ end_acf_read:
odbc_datastore_free(resultset);
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
pbx_builtin_setvar_helper(chan, "ODBCSTATUS", "MEMERROR");
ast_autoservice_stop(chan);
return -1;
@ -764,8 +910,7 @@ end_acf_read:
}
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
if (resultset && !multirow) {
/* Fetch the first resultset */
if (!acf_fetch(chan, "", buf, buf, len)) {
@ -1192,8 +1337,8 @@ static char *cli_odbc_read(struct ast_cli_entry *e, int cmd, struct ast_cli_args
if (a->argc == 5 && !strcmp(a->argv[4], "exec")) {
/* Execute the query */
struct odbc_obj *obj = NULL;
int dsn, executed = 0;
struct dsn *dsn = NULL;
int dsn_num, executed = 0;
SQLHSTMT stmt;
int rows = 0, res, x;
SQLSMALLINT colcount = 0, collength;
@ -1207,19 +1352,18 @@ static char *cli_odbc_read(struct ast_cli_entry *e, int cmd, struct ast_cli_args
return CLI_SUCCESS;
}
for (dsn = 0; dsn < 5; dsn++) {
if (ast_strlen_zero(query->readhandle[dsn])) {
for (dsn_num = 0; dsn_num < 5; dsn_num++) {
if (ast_strlen_zero(query->readhandle[dsn_num])) {
continue;
}
ast_debug(1, "Found handle %s\n", query->readhandle[dsn]);
if (!(obj = ast_odbc_request_obj(query->readhandle[dsn], 0))) {
dsn = get_dsn(query->readhandle[dsn_num]);
if (!dsn) {
continue;
}
ast_debug(1, "Found handle %s\n", query->readhandle[dsn_num]);
ast_debug(1, "Got obj\n");
if (!(stmt = ast_odbc_direct_execute(obj, generic_execute, ast_str_buffer(sql)))) {
ast_odbc_release_obj(obj);
obj = NULL;
if (!(stmt = ast_odbc_direct_execute(dsn->connection, generic_execute, ast_str_buffer(sql)))) {
dsn = release_dsn(dsn);
continue;
}
@ -1230,8 +1374,7 @@ static char *cli_odbc_read(struct ast_cli_entry *e, int cmd, struct ast_cli_args
ast_cli(a->fd, "SQL Column Count error!\n[%s]\n\n", ast_str_buffer(sql));
SQLCloseCursor(stmt);
SQLFreeHandle (SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
AST_RWLIST_UNLOCK(&queries);
return CLI_SUCCESS;
}
@ -1240,10 +1383,9 @@ static char *cli_odbc_read(struct ast_cli_entry *e, int cmd, struct ast_cli_args
if ((res != SQL_SUCCESS) && (res != SQL_SUCCESS_WITH_INFO)) {
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
if (res == SQL_NO_DATA) {
ast_cli(a->fd, "Returned %d rows. Query executed on handle %d:%s [%s]\n", rows, dsn, query->readhandle[dsn], ast_str_buffer(sql));
ast_cli(a->fd, "Returned %d rows. Query executed on handle %d:%s [%s]\n", rows, dsn_num, query->readhandle[dsn_num], ast_str_buffer(sql));
break;
} else {
ast_cli(a->fd, "Error %d in FETCH [%s]\n", res, ast_str_buffer(sql));
@ -1270,8 +1412,7 @@ static char *cli_odbc_read(struct ast_cli_entry *e, int cmd, struct ast_cli_args
ast_cli(a->fd, "SQL Get Data error %d!\n[%s]\n\n", res, ast_str_buffer(sql));
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
dsn = release_dsn(dsn);
AST_RWLIST_UNLOCK(&queries);
return CLI_SUCCESS;
}
@ -1289,15 +1430,11 @@ static char *cli_odbc_read(struct ast_cli_entry *e, int cmd, struct ast_cli_args
}
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
ast_cli(a->fd, "Returned %d row%s. Query executed on handle %d [%s]\n", rows, rows == 1 ? "" : "s", dsn, query->readhandle[dsn]);
dsn = release_dsn(dsn);
ast_cli(a->fd, "Returned %d row%s. Query executed on handle %d [%s]\n", rows, rows == 1 ? "" : "s", dsn_num, query->readhandle[dsn_num]);
break;
}
if (obj) {
ast_odbc_release_obj(obj);
obj = NULL;
}
dsn = release_dsn(dsn);
if (!executed) {
ast_cli(a->fd, "Failed to execute query. [%s]\n", ast_str_buffer(sql));
@ -1420,30 +1557,29 @@ static char *cli_odbc_write(struct ast_cli_entry *e, int cmd, struct ast_cli_arg
if (a->argc == 6 && !strcmp(a->argv[5], "exec")) {
/* Execute the query */
struct odbc_obj *obj = NULL;
int dsn, executed = 0;
struct dsn *dsn;
int dsn_num, executed = 0;
SQLHSTMT stmt;
SQLLEN rows = -1;
for (dsn = 0; dsn < 5; dsn++) {
if (ast_strlen_zero(query->writehandle[dsn])) {
for (dsn_num = 0; dsn_num < 5; dsn_num++) {
if (ast_strlen_zero(query->writehandle[dsn_num])) {
continue;
}
if (!(obj = ast_odbc_request_obj(query->writehandle[dsn], 0))) {
dsn = get_dsn(query->writehandle[dsn_num]);
if (!dsn) {
continue;
}
if (!(stmt = ast_odbc_direct_execute(obj, generic_execute, ast_str_buffer(sql)))) {
ast_odbc_release_obj(obj);
obj = NULL;
if (!(stmt = ast_odbc_direct_execute(dsn->connection, generic_execute, ast_str_buffer(sql)))) {
dsn = release_dsn(dsn);
continue;
}
SQLRowCount(stmt, &rows);
SQLCloseCursor(stmt);
SQLFreeHandle(SQL_HANDLE_STMT, stmt);
ast_odbc_release_obj(obj);
obj = NULL;
ast_cli(a->fd, "Affected %d rows. Query executed on handle %d [%s]\n", (int)rows, dsn, query->writehandle[dsn]);
dsn = release_dsn(dsn);
ast_cli(a->fd, "Affected %d rows. Query executed on handle %d [%s]\n", (int)rows, dsn_num, query->writehandle[dsn_num]);
executed = 1;
break;
}
@ -1470,6 +1606,11 @@ static int load_module(void)
char *catg;
struct ast_flags config_flags = { 0 };
dsns = ao2_container_alloc(DSN_BUCKETS, dsn_hash, dsn_cmp);
if (!dsns) {
return AST_MODULE_LOAD_DECLINE;
}
res |= ast_custom_function_register(&fetch_function);
res |= ast_register_application_xml(app_odbcfinish, exec_odbcfinish);
AST_RWLIST_WRLOCK(&queries);
@ -1478,6 +1619,7 @@ static int load_module(void)
if (!cfg || cfg == CONFIG_STATUS_FILEINVALID) {
ast_log(LOG_NOTICE, "Unable to load config for func_odbc: %s\n", config);
AST_RWLIST_UNLOCK(&queries);
ao2_ref(dsns, -1);
return AST_MODULE_LOAD_DECLINE;
}
@ -1531,6 +1673,8 @@ static int unload_module(void)
AST_RWLIST_WRLOCK(&queries);
AST_RWLIST_UNLOCK(&queries);
ao2_ref(dsns, -1);
return res;
}