Changeset: 6ca60818140b for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=6ca60818140b
Modified Files:
tools/embedded/Tests/lowlevel.R
tools/embedded/embedded.c
tools/embedded/embeddedr.c
tools/embedded/rpackage/R/monetdb.R
Branch: embedded
Log Message:
Bug 3885
diffs (truncated from 316 to 300 lines):
diff --git a/tools/embedded/Tests/lowlevel.R b/tools/embedded/Tests/lowlevel.R
--- a/tools/embedded/Tests/lowlevel.R
+++ b/tools/embedded/Tests/lowlevel.R
@@ -20,8 +20,10 @@ test_that("one can connect", {
con <- monetdb_embedded_connect()
expect_that(con, is_a("monetdb_embedded_connection"))
monetdb_embedded_disconnect(con)
+ # closed connections can be closed again
monetdb_embedded_disconnect(con)
expect_error(monetdb_embedded_disconnect(NULL))
+ expect_error(monetdb_embedded_disconnect(42))
})
test_that("db runs queries and returns results", {
@@ -37,6 +39,14 @@ test_that("db runs queries and returns r
monetdb_embedded_disconnect(con)
})
+test_that("a disconnected connection cannot be used", {
+ con <- monetdb_embedded_connect()
+ monetdb_embedded_query(con, "SELECT 42")
+ monetdb_embedded_disconnect(con)
+ res <- monetdb_embedded_query(con, "SELECT 42")
+ expect_equal(res$type, "!")
+})
+
test_that("commit", {
con <- monetdb_embedded_connect()
monetdb_embedded_query(con, "START TRANSACTION")
@@ -116,6 +126,21 @@ test_that("inserting data", {
monetdb_embedded_disconnect(con)
})
+test_that("the garbage collector closes connections", {
+ # there are 64 connections max. if gc() does not close them, the last
line will provoke a crash
+ conns <- lapply(1:64, function(x) monetdb_embedded_connect())
+ expect_error(monetdb_embedded_connect())
+
+ rm(conns)
+ gc()
+
+ conns <- lapply(1:64, function(x) monetdb_embedded_connect())
+ rm(conns)
+ gc()
+ monetdb_embedded_connect()
+})
+
+
test_that("the logger does not misbehave", {
con <- monetdb_embedded_connect()
monetdb_embedded_query(con, "CREATE TABLE foo(i INTEGER, j INTEGER)")
@@ -124,11 +149,3 @@ test_that("the logger does not misbehave
monetdb_embedded_disconnect(con)
})
-test_that("the garbage collector closes connections", {
- # there are 64 connections max. if gc() does not close them, the last
line will provoke a crash
- conns <- lapply(1:60, function(x) monetdb_embedded_connect())
- rm(conns)
- gc()
- conns <- lapply(1:60, function(x) monetdb_embedded_connect())
-})
-
diff --git a/tools/embedded/embedded.c b/tools/embedded/embedded.c
--- a/tools/embedded/embedded.c
+++ b/tools/embedded/embedded.c
@@ -46,8 +46,9 @@ sqlcleanup_ptr_tpe sqlcleanup_ptr = NULL
typedef void (*mvc_trans_ptr_tpe)(mvc*);
mvc_trans_ptr_tpe mvc_trans_ptr = NULL;
-static MT_Lock monetdb_embedded_lock;
int monetdb_embedded_initialized = 0;
+#define EMBEDDED_MAX_CONNS 64
+static void *monetdb_embedded_connections[EMBEDDED_MAX_CONNS];
static void* lookup_function(char* func) {
void *dl, *fun;
@@ -60,26 +61,62 @@ static void* lookup_function(char* func)
return fun;
}
+static int monetdb_valid_conn(void* conn) {
+ int i;
+ if (conn == NULL) {
+ return 0;
+ }
+ for (i = 0; i < EMBEDDED_MAX_CONNS; i++) {
+ if (conn == monetdb_embedded_connections[i]) {
+ return 1;
+ }
+ }
+ return 0;
+}
+
void* monetdb_connect() {
+ void **conn = NULL;
+ int i;
+
+ if (!monetdb_embedded_initialized) {
+ return NULL;
+ }
+
+ for (i = 0; i < EMBEDDED_MAX_CONNS; i++) {
+ if (monetdb_embedded_connections[i] == NULL) {
+ conn = &monetdb_embedded_connections[i];
+ break;
+ }
+ }
+ if (conn == NULL) {
+ return NULL;
+ }
Client c = MCforkClient(&mal_clients[0]);
if ((*SQLinitClient_ptr)(c) != MAL_SUCCEED) {
return NULL;
}
((backend *) c->sqlcontext)->mvc->session->auto_commit = 1;
- // TODO: keep track of pointers returned
+ *conn = c;
return c;
}
void monetdb_disconnect(void* conn) {
- if (conn == NULL) {
+ int i;
+ if (!monetdb_valid_conn(conn)) {
return;
}
- MCcloseClient((Client) conn);
+ for (i = 0; i < EMBEDDED_MAX_CONNS; i++) {
+ if (conn == monetdb_embedded_connections[i]) {
+ MCcloseClient((Client) conn);
+ monetdb_embedded_connections[i] = NULL;
+ return;
+ }
+ }
}
char* monetdb_startup(char* dbdir, char silent) {
opt *set = NULL;
- int setlen = 0;
+ int setlen = 0, i;
str retval = MAL_SUCCEED;
char* sqres = NULL;
void* res = NULL;
@@ -98,8 +135,7 @@ char* monetdb_startup(char* dbdir, char
}
goto cleanup;
}
- MT_lock_init(&monetdb_embedded_lock, "monetdb_embedded_lock");
- MT_lock_set(&monetdb_embedded_lock);
+
if (monetdb_embedded_initialized) goto cleanup;
setlen = mo_builtin_settings(&set);
@@ -115,6 +151,9 @@ char* monetdb_startup(char* dbdir, char
GDKsetenv("mapi_disable", "true");
GDKsetenv("sql_optimizer", "sequential_pipe");
+ for (i = 0; i < EMBEDDED_MAX_CONNS; i++) {
+ monetdb_embedded_connections[i] = NULL;
+ }
if (silent) THRdata[0] = stream_blackhole_create();
msab_dbpathinit(dbdir);
@@ -148,14 +187,16 @@ char* monetdb_startup(char* dbdir, char
goto cleanup;
}
+ monetdb_embedded_initialized = true;
c = monetdb_connect();
if (c == NULL) {
+ monetdb_embedded_initialized = false;
retval = GDKstrdup("Failed to initialize client");
goto cleanup;
}
- monetdb_embedded_initialized = true;
+ GDKfataljumpenable = 0;
+
// we do not want to jump after this point, since we cannot do so
between threads
- GDKfataljumpenable = 0;
// sanity check, run a SQL query
sqres = monetdb_query(c, "SELECT * FROM tables;", &res);
if (sqres != NULL) {
@@ -167,18 +208,20 @@ char* monetdb_startup(char* dbdir, char
monetdb_disconnect(c);
cleanup:
mo_free_options(set, setlen);
- MT_lock_unset(&monetdb_embedded_lock);
return retval;
}
char* monetdb_query(void* conn, char* query, void** result) {
- // TODO: check client pointer
str res = MAL_SUCCEED;
Client c = (Client) conn;
- mvc* m = ((backend *) c->sqlcontext)->mvc;
+ mvc* m;
if (!monetdb_embedded_initialized) {
return GDKstrdup("Embedded MonetDB is not started");
}
+ if (!monetdb_valid_conn(conn)) {
+ return GDKstrdup("Invalid connection");
+ }
+ m = ((backend *) c->sqlcontext)->mvc;
while (*query == ' ' || *query == '\t') query++;
if (strncasecmp(query, "START", 5) == 0) { // START TRANSACTION
@@ -216,7 +259,17 @@ char* monetdb_append(void* conn, const c
Client c = (Client) conn;
mvc* m = ((backend *) c->sqlcontext)->mvc;
- assert(table != NULL && data != NULL && ncols > 0);
+ // TODO: check client pointer
+
+ if (!monetdb_embedded_initialized) {
+ return GDKstrdup("Embedded MonetDB is not started");
+ }
+ if(table == NULL || data == NULL || ncols < 1) {
+ return GDKstrdup("Invalid parameters");
+ }
+ if (!monetdb_valid_conn(conn)) {
+ return GDKstrdup("Invalid connection");
+ }
// very black MAL magic below
mb.var = GDKmalloc(nvar * sizeof(VarRecord*));
@@ -302,7 +355,6 @@ str monetdb_get_columns(void* conn, cons
// TODO: fix this, it is not working correctly
void monetdb_shutdown() {
- MT_lock_set(&monetdb_embedded_lock);
// kill SQL
(*SQLepilogue_ptr)(NULL);
// kill MAL & GDK
@@ -310,5 +362,4 @@ void monetdb_shutdown() {
// clean up global state
BBPresetfarms();
monetdb_embedded_initialized = 0;
- MT_lock_unset(&monetdb_embedded_lock);
}
diff --git a/tools/embedded/embeddedr.c b/tools/embedded/embeddedr.c
--- a/tools/embedded/embeddedr.c
+++ b/tools/embedded/embeddedr.c
@@ -132,8 +132,11 @@ SEXP monetdb_append_R(SEXP connsexp, SEX
SEXP monetdb_connect_R() {
- SEXP conn = PROTECT(R_MakeExternalPtr(
- monetdb_connect(), R_NilValue, R_NilValue));
+ void* llconn = monetdb_connect();
+ if (!llconn) {
+ error("Could not create connection.");
+ }
+ SEXP conn = PROTECT(R_MakeExternalPtr(llconn, R_NilValue, R_NilValue));
R_RegisterCFinalizer(conn, (void (*)(SEXP)) monetdb_disconnect_R);
UNPROTECT(1);
return conn;
diff --git a/tools/embedded/rpackage/R/monetdb.R
b/tools/embedded/rpackage/R/monetdb.R
--- a/tools/embedded/rpackage/R/monetdb.R
+++ b/tools/embedded/rpackage/R/monetdb.R
@@ -36,7 +36,6 @@ monetdb_embedded_startup <- function(dir
if (is.character(res)) {
stop("Failed to initialize embedded MonetDB ", res)
}
-
monetdb_embedded_env$is_started <- TRUE
monetdb_embedded_env$started_dir <- dir
invisible(TRUE)
@@ -52,7 +51,7 @@ monetdb_embedded_query <- function(conn,
stop("Need a single noreally flag as parameter.")
}
if (!inherits(conn, classname)) {
- stop("Need a embedded monetdb connection as parameter")
+ stop("Invalid connection")
}
# make sure the query is terminated
query <- paste(query, "\n;", sep="")
@@ -90,7 +89,7 @@ monetdb_embedded_append <- function(conn
stop("Need a data frame as tdata parameter.")
}
if (!inherits(conn, classname)) {
- stop("Need a embedded monetdb connection as parameter")
+ stop("Invalid connection")
}
.Call("monetdb_append_R", conn, schema, table, tdata,
PACKAGE=libfilename)
}
@@ -100,23 +99,23 @@ monetdb_embedded_connect <- function() {
if (!monetdb_embedded_env$is_started) {
stop("Call monetdb_embedded_startup() first")
}
- res <- .Call("monetdb_connect_R", PACKAGE=libfilename)
- class(res) <- classname
- return(res)
+ conn <- .Call("monetdb_connect_R", PACKAGE=libfilename)
+ class(conn) <- classname
+ return(conn)
}
monetdb_embedded_disconnect <- function(conn) {
if (!inherits(conn, classname)) {
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
