/* * Copyright (c) 2012-2013 Los Alamos National Security, Inc. All rights reserved. * $COPYRIGHT$ * * Additional copyrights may follow * * $HEADER$ * */ #include "opal_config.h" #include "opal/constants.h" #include #include #ifdef HAVE_LIMITS_H #include #endif #include #ifdef HAVE_FCNTL_H #include #endif #ifdef HAVE_UNISTD_H #include #endif #include #include "opal/dss/dss_types.h" #include "opal/util/argv.h" #include "opal/util/basename.h" #include "opal/util/os_dirpath.h" #include "opal/util/os_path.h" #include "opal/util/output.h" #include "opal/util/malloc.h" #include "opal/util/basename.h" #include "opal/mca/pstat/base/base.h" #include "opal/util/show_help.h" #include "opal/mca/db/base/base.h" #include "db_sqlite.h" static int init(void); static void finalize(void); static int add_log(const char *table, const opal_value_t *kvs, int nkvs); opal_db_base_module_t opal_db_sqlite_module = { init, finalize, NULL, NULL, NULL, NULL, NULL, NULL, NULL, NULL, add_log }; /* local variables */ static sqlite3 **dbhandles=NULL; static int nthreads = 1; static int active = 0; static int init(void) { int i; /* initialize sqlite3 */ if (SQLITE_OK != sqlite3_initialize()) { return OPAL_ERR_UNREACH; } /* check if sqlite was built thread-safe - if not, we won't * use worker threads for thruput */ if (0 == mca_db_sqlite_component.num_worker_threads || 0 != sqlite3_threadsafe()) { nthreads = 1; } else { nthreads = mca_db_sqlite_component.num_worker_threads; } /* get the required number of database handles */ dbhandles = (sqlite3**)malloc(nthreads * sizeof(sqlite3*)); /* open the database - this will create the database file if * it doesn't already exist */ for (i=0; i < nthreads; i++) { if (SQLITE_OK != sqlite3_open(mca_db_sqlite_component.db_file, &dbhandles[i])) { opal_show_help("help-db-sqlite.txt", "cannot-create-sqlite", true, mca_db_sqlite_component.db_file); return OPAL_ERR_FILE_OPEN_FAILURE; } } return OPAL_SUCCESS; } static void finalize(void) { int i; /* if we are normally terminating, remove the recovery file */ if (NULL != dbhandles) { for (i=0; i < nthreads; i++) { if (SQLITE_OK != sqlite3_close(dbhandles[i])) { opal_output(0, "sqlite failed to close"); } } } } static int add_log(const char *table, const opal_value_t *kvs, int nkvs) { int i, rc; char *sql, **cmd = NULL, *tmp; sqlite3_stmt *stmt; opal_output_verbose(2, opal_db_base_framework.framework_output, "Logging data for table %s", table); /* setup the insert statement */ for (i=0; i < nkvs; i++) { opal_argv_append_nosize(&cmd, "?"); } tmp = opal_argv_join(cmd, ','); asprintf(&sql, "INSERT INTO %s VALUES (%s)", table, tmp); free(tmp); opal_argv_free(cmd); /* use the next worker thread */ OPAL_SQLITE_CMD(prepare_v2(dbhandles[active], sql, strlen(sql)+1, &stmt, NULL), dbhandles[active], &rc); if (SQLITE_OK != rc) { return OPAL_ERROR; } /* cycle through the provided values and construct * an insert command for them - note that the values * MUST be in column-order for the database! */ for (i=0; i < nkvs; i++) { switch (kvs[i].type) { case OPAL_STRING: OPAL_SQLITE_CMD(bind_text(stmt, i, kvs[i].data.string, strlen(kvs[i].data.string), NULL), dbhandles[active], &rc); break; case OPAL_INT32: OPAL_SQLITE_CMD(bind_int(stmt, i, kvs[i].data.int32), dbhandles[active], &rc); break; case OPAL_INT16: OPAL_SQLITE_CMD(bind_int(stmt, i, kvs[i].data.int16), dbhandles[active], &rc); break; case OPAL_PID: OPAL_SQLITE_CMD(bind_int64(stmt, i, kvs[i].data.pid), dbhandles[active], &rc); break; case OPAL_INT64: OPAL_SQLITE_CMD(bind_int64(stmt, i, kvs[i].data.int64), dbhandles[active], &rc); break; case OPAL_FLOAT: OPAL_SQLITE_CMD(bind_double(stmt, i, kvs[i].data.fval), dbhandles[active], &rc); break; case OPAL_TIMEVAL: asprintf(&tmp, "%d.%06d", (int)kvs[i].data.tv.tv_sec, (int)kvs[i].data.tv.tv_usec); OPAL_SQLITE_CMD(bind_text(stmt, i, tmp, strlen(tmp), NULL), dbhandles[active], &rc); free(tmp); break; } if (SQLITE_OK != rc) { return OPAL_ERROR; } } OPAL_SQLITE_OP(step(stmt), DONE, dbhandles[active], &rc); if (SQLITE_OK != rc) { return OPAL_ERROR; } opal_output_verbose(2, opal_db_base_framework.framework_output, "INSERTED ROW %d", (int)sqlite3_last_insert_rowid(dbhandles[active])); /* cycle to the next worker thread */ active++; if (nthreads < active) { active = 0; } return OPAL_SUCCESS; }