Changeset: 7bd5ca0ce2f1 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=7bd5ca0ce2f1
Modified Files:
embedded/monetdb_embedded.c
embedded/monetdb_embedded.h
Branch: cmake-monetdblite
Log Message:
Made MoneDBLite API thread safe.
diffs (truncated from 962 to 300 lines):
diff --git a/embedded/monetdb_embedded.c b/embedded/monetdb_embedded.c
--- a/embedded/monetdb_embedded.c
+++ b/embedded/monetdb_embedded.c
@@ -30,27 +30,53 @@
#include "rel_updates.h"
#include "monet_options.h"
-static int monetdb_embedded_initialized = 0;
+static MT_Lock embedded_lock = MT_LOCK_INITIALIZER("embedded_lock");
+static bool monetdb_embedded_initialized = false;
-int
+bool
monetdb_is_initialized(void)
{
- return monetdb_embedded_initialized > 0;
+ MT_lock_set(&embedded_lock);
+ bool res = monetdb_embedded_initialized;
+ MT_lock_unset(&embedded_lock);
+ return res;
}
static char*
-validate_connection(monetdb_connection conn, const char* call)
+validate_connection(monetdb_connection conn, const char* call) // Call this
function always inside the embedded_lock
{
- if (!monetdb_is_initialized())
+ if (!monetdb_embedded_initialized)
return createException(MAL, call, SQLSTATE(HY001) "Embedded
MonetDB is not started");
if (!MCvalid((Client) conn))
return createException(MAL, call, SQLSTATE(HY001) "Invalid
connection");
return MAL_SUCCEED;
}
-static MT_Lock embedded_lock = MT_LOCK_INITIALIZER("embedded_lock");
+static void
+monetdb_destroy_column(monetdb_column* column)
+{
+ size_t j;
+
+ if (!column)
+ return;
-static void monetdb_destroy_column(monetdb_column* column);
+ if (column->type == monetdb_str) {
+ // FIXME: clean up individual strings
+ char** data = (char**)column->data;
+ for(j = 0; j < column->count; j++) {
+ if (data[j])
+ GDKfree(data[j]);
+ }
+ } else if (column->type == monetdb_blob) {
+ monetdb_data_blob* data = (monetdb_data_blob*)column->data;
+ for(j = 0; j < column->count; j++) {
+ if (data[j].data)
+ GDKfree(data[j].data);
+ }
+ }
+ GDKfree(column->data);
+ GDKfree(column);
+}
typedef struct {
monetdb_result res;
@@ -58,153 +84,26 @@ typedef struct {
monetdb_column **converted_columns;
} monetdb_result_internal;
-char*
-monetdb_connect(monetdb_connection *conn)
-{
- mvc *m;
- char* msg = MAL_SUCCEED;
- Client mc = NULL;
-
- if (!monetdb_is_initialized()) {
- msg = createException(MAL, "embedded.monetdb_connect",
SQLSTATE(42000) "Embedded MonetDB is not started");
- goto cleanup;
- }
- if (!conn) {
- msg = createException(MAL, "embedded.monetdb_connect",
SQLSTATE(42000) "monetdb_connection parameter is NULL");
- goto cleanup;
- }
- mc = MCinitClient((oid) 0, 0, 0);
- if (!MCvalid(mc)) {
- msg = createException(MAL, "embedded.monetdb_connect",
SQLSTATE(HY001) "Failed to initialize client");
- goto cleanup;
- }
- mc->curmodule = mc->usermodule = userModule();
- if (mc->usermodule == NULL) {
- msg = createException(MAL, "embedded.monetdb_connect",
SQLSTATE(HY001) "Failed to initialize client MAL module");
- goto cleanup;
- }
- if ((msg = SQLinitClient(mc)) != MAL_SUCCEED)
- goto cleanup;
- if ((msg = getSQLContext(mc, NULL, &m, NULL)) != MAL_SUCCEED)
- goto cleanup;
- m->session->auto_commit = 1;
-
-cleanup:
- if (msg && mc) {
- char* other = monetdb_disconnect(mc);
- if (other)
- GDKfree(other);
- *conn = NULL;
- } else if(conn)
- *conn = mc;
- return msg;
-}
-
-char*
-monetdb_disconnect(monetdb_connection conn)
-{
- char* msg = MAL_SUCCEED;
-
- if ((msg = validate_connection(conn, "embedded.monetdb_disconnect")) !=
MAL_SUCCEED)
- return msg;
- SQLexitClient((Client) conn);
- MCcloseClient((Client) conn);
- return MAL_SUCCEED;
-}
-
static char*
-monetdb_shutdown_internal(void)
-{
- if (monetdb_embedded_initialized) {
- malEmbeddedReset();
- monetdb_embedded_initialized = 0;
- }
- return MAL_SUCCEED;
-}
-
-char*
-monetdb_startup(char* dbdir, char silent, char sequential)
+monetdb_cleanup_result_internal(monetdb_connection conn, monetdb_result*
result)
{
char* msg = MAL_SUCCEED;
- monetdb_result* res = NULL;
- void* c;
- opt *set = NULL;
- int setlen;
- gdk_return gdk_res;
+ monetdb_result_internal* res = (monetdb_result_internal *) result;
- MT_lock_set(&embedded_lock);
- GDKfataljumpenable = 1;
- if(setjmp(GDKfataljump) != 0) {
- msg = GDKfatalmsg;
- // we will get here if GDKfatal was called.
- if (msg == NULL)
- msg = createException(MAL, "embedded.monetdb_startup",
SQLSTATE(HY002) "GDKfatal() with unspecified error");
- goto cleanup;
- }
-
- if (monetdb_embedded_initialized)
- goto cleanup;
+ if ((msg = validate_connection(conn,
"embedded.monetdb_cleanup_result_internal")) != MAL_SUCCEED)
+ return msg;
- if (silent)
- MT_fprintf_silent(true);
+ if (res->monetdb_resultset)
+ res_tables_destroy(res->monetdb_resultset);
- if ((setlen = mo_builtin_settings(&set)) == 0) {
- msg = createException(MAL, "embedded.monetdb_startup",
SQLSTATE(HY001) MAL_MALLOC_FAIL);
- goto cleanup;
- }
- if (dbdir && (setlen = mo_add_option(&set, setlen, opt_cmdline,
"gdk_dbpath", dbdir)) == 0) {
- mo_free_options(set, setlen);
- msg = createException(MAL, "embedded.monetdb_startup",
SQLSTATE(HY001) MAL_MALLOC_FAIL);
- goto cleanup;
- }
- if (sequential)
- setlen = mo_add_option(&set, setlen, opt_cmdline,
"sql_optimizer", "sequential_pipe");
- else
- setlen = mo_add_option(&set, setlen, opt_cmdline,
"sql_optimizer", "default_pipe");
- if (setlen == 0) {
- mo_free_options(set, setlen);
- msg = createException(MAL, "embedded.monetdb_startup",
SQLSTATE(HY001) MAL_MALLOC_FAIL);
- goto cleanup;
+ if (res->converted_columns) {
+ size_t i;
+ for (i = 0; i < res->res.ncols; i++)
+ monetdb_destroy_column(res->converted_columns[i]);
+ GDKfree(res->converted_columns);
}
- gdk_res = GDKinit(set, setlen);
- mo_free_options(set, setlen);
- if (gdk_res == GDK_FAIL) {
- msg = createException(MAL, "embedded.monetdb_startup",
SQLSTATE(HY002) "GDKinit() failed");
- goto cleanup;
- }
-
- if ((msg = malEmbeddedBoot()) != MAL_SUCCEED)
- goto cleanup;
- if (!SQLisInitialized()) {
- msg = createException(MAL, "embedded.monetdb_startup",
SQLSTATE(HY002) "SQL initialization failed");
- goto cleanup;
- }
-
- monetdb_embedded_initialized = true;
-
- if ((msg = monetdb_connect(&c)) != MAL_SUCCEED)
- goto cleanup;
- GDKfataljumpenable = 0;
+ GDKfree(res);
- // we do not want to jump after this point, since we cannot do so
between threads
- // sanity check, run a SQL query
- if ((msg = monetdb_query(c, "SELECT id FROM _tables LIMIT 1;", &res,
NULL, NULL)) != MAL_SUCCEED) {
- monetdb_embedded_initialized = false;
- goto cleanup;
- }
- if ((msg = monetdb_cleanup_result(c, res)) != MAL_SUCCEED) {
- monetdb_embedded_initialized = false;
- goto cleanup;
- }
- if ((msg = monetdb_disconnect(c)) != MAL_SUCCEED) {
- monetdb_embedded_initialized = false;
- goto cleanup;
- }
-
-cleanup:
- if (msg)
- monetdb_shutdown_internal();
- MT_lock_unset(&embedded_lock);
return msg;
}
@@ -212,16 +111,16 @@ static char*
monetdb_query_internal(monetdb_connection conn, char* query, monetdb_result**
result, int64_t* affected_rows,
int64_t* prepare_id, char language)
{
- char* msg = MAL_SUCCEED, *commit_msg;
+ char* msg = MAL_SUCCEED, *commit_msg = MAL_SUCCEED;
Client c = (Client) conn;
- mvc* m;
+ mvc* m = NULL;
backend *b;
- char *nq, *qname = "somequery";
+ char *nq = NULL, *qname = "somequery";
size_t query_len;
buffer query_buf;
stream *query_stream;
monetdb_result_internal *res_internal = NULL;
- bstream *old_bstream;
+ bstream *old_bstream = NULL;
if ((msg = validate_connection(conn,
"embedded.monetdb_query_internal")) != MAL_SUCCEED)
return msg;
@@ -232,12 +131,15 @@ monetdb_query_internal(monetdb_connectio
if (!query)
return createException(MAL, "embedded.monetdb_query_internal",
SQLSTATE(42000) "Query missing");
- if (!(query_stream = buffer_rastream(&query_buf, qname)))
- return createException(MAL, "embedded.monetdb_query_internal",
SQLSTATE(HY001) "WARNING: could not setup query stream.");
+ if (!(query_stream = buffer_rastream(&query_buf, qname))) {
+ msg = createException(MAL, "embedded.monetdb_query_internal",
SQLSTATE(HY001) "WARNING: could not setup query stream.");
+ goto cleanup;
+ }
query_len = strlen(query) + 3;
- nq = GDKmalloc(query_len);
- if (!nq)
- return createException(MAL, "embedded.monetdb_query_internal",
SQLSTATE(HY001) "WARNING: could not setup query stream.");
+ if (!(nq = GDKmalloc(query_len))) {
+ msg = createException(MAL, "embedded.monetdb_query_internal",
SQLSTATE(HY001) "WARNING: could not setup query stream.");
+ goto cleanup;
+ }
sprintf(nq, "%s\n;", query);
query_buf.pos = 0;
@@ -272,8 +174,7 @@ monetdb_query_internal(monetdb_connectio
*affected_rows = m->rowcnt;
if (result) {
- res_internal = GDKzalloc(sizeof(monetdb_result_internal));
- if (!res_internal) {
+ if (!(res_internal =
GDKzalloc(sizeof(monetdb_result_internal)))) {
msg = createException(MAL,
"embedded.monetdb_query_internal", SQLSTATE(HY001) MAL_MALLOC_FAIL);
goto cleanup;
}
@@ -309,15 +210,19 @@ monetdb_query_internal(monetdb_connectio
}
cleanup:
- GDKfree(nq);
+ if (nq)
+ GDKfree(nq);
MSresetInstructions(c->curprg->def, 1);
- bstream_destroy(c->fdin);
- c->fdin = old_bstream;
+ if (old_bstream) { //c->fdin was set
+ bstream_destroy(c->fdin);
+ c->fdin = old_bstream;
+ }
- commit_msg = SQLautocommit(m); //need always to commit even if msg is
set
+ if (m) //the connection is valid and a query was attempted
+ commit_msg = SQLautocommit(m); //need always to commit even if
msg is set
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
