Changeset: 5c8b197ba592 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=5c8b197ba592
Modified Files:
clients/mapilib/mapi.mx
sql/backends/monet5/merovingian/daemon/multiplex-funnel.c
Branch: default
Log Message:
mapi: introduce mapi_send and mapi_read_response
Allow to send a query without waiting for its result, such that multiple
queries can be sent to multiple servers, and the results only collected
afterwards. E.g. the waiting for the query to be processed, is done
concurrently, as used by merovingian's multiplex-funnel now.
diffs (136 lines):
diff --git a/clients/mapilib/mapi.mx b/clients/mapilib/mapi.mx
--- a/clients/mapilib/mapi.mx
+++ b/clients/mapilib/mapi.mx
@@ -1246,6 +1246,8 @@
mapi_export MapiHdl mapi_quick_query(Mapi mid, const char *cmd, FILE *fd);
mapi_export MapiHdl mapi_query_array(Mapi mid, const char *cmd, char **val);
mapi_export MapiHdl mapi_quick_query_array(Mapi mid, const char *cmd, char
**val, FILE *fd);
+mapi_export MapiHdl mapi_send(Mapi mid, const char *cmd);
+mapi_export MapiMsg mapi_read_response(MapiHdl hdl);
mapi_export MapiHdl mapi_stream_query(Mapi mid, const char *cmd, int
windowsize);
mapi_export MapiMsg mapi_cache_limit(Mapi mid, int limit);
mapi_export MapiMsg mapi_cache_shuffle(MapiHdl hdl, int percentage);
@@ -4323,7 +4325,7 @@
mnstr_flush(mid->to);
check_stream(mid, mid->to, "write error on stream", "mapi_execute",
mid->error);
mid->active = hdl;
- return read_into_cache(hdl, 1);
+ return MOK;
}
MapiMsg
@@ -4332,7 +4334,9 @@
int ret;
mapi_hdl_check(hdl, "mapi_execute");
- ret = mapi_execute_internal(hdl);
+ if ((ret = mapi_execute_internal(hdl)) == MOK)
+ return read_into_cache(hdl, 1);
+
return ret;
}
@@ -4347,6 +4351,8 @@
ret = mapi_execute_internal(hdl);
if (ret == MOK)
ret = read_into_cache(hdl, 1);
+ if (ret == MOK)
+ ret = read_into_cache(hdl, 1);
return ret;
}
@@ -4361,9 +4367,32 @@
ret = mid->error;
if (ret == MOK)
ret = mapi_execute_internal(hdl);
+ if (ret == MOK)
+ ret = read_into_cache(hdl, 1);
return hdl;
}
+/* version of mapi_query that does not wait for a response */
+MapiHdl
+mapi_send(Mapi mid, const char *cmd)
+{
+ int ret;
+ MapiHdl hdl;
+
+ mapi_check0(mid, "mapi_send");
+ hdl = prepareQuery(mapi_new_handle(mid), cmd);
+ ret = mid->error;
+ if (ret == MOK)
+ ret = mapi_execute_internal(hdl);
+ return hdl;
+}
+
+MapiMsg
+mapi_read_response(MapiHdl hdl)
+{
+ return read_into_cache(hdl, 1);
+}
+
MapiMsg
mapi_query_handle(MapiHdl hdl, const char *cmd)
{
@@ -4376,6 +4405,8 @@
ret = hdl->mid->error;
if (ret == MOK)
ret = mapi_execute_internal(hdl);
+ if (ret == MOK)
+ ret = read_into_cache(hdl, 1);
return ret;
}
@@ -4394,6 +4425,8 @@
ret = mapi_prepare_array_internal(hdl, val);
if (ret == MOK)
ret = mapi_execute_internal(hdl);
+ if (ret == MOK)
+ ret = read_into_cache(hdl, 1);
return hdl;
}
@@ -4500,6 +4533,8 @@
if (ret == MOK)
ret = mapi_execute_internal(hdl);
if (ret == MOK)
+ ret = read_into_cache(hdl, 1);
+ if (ret == MOK)
ret = mapi_quick_response(hdl, fd);
if (mid->trace == MAPI_TRACE)
printf("mapi_quick_query return:%d\n", ret);
@@ -4521,6 +4556,8 @@
ret = mapi_prepare_array_internal(hdl, val);
if (ret == MOK)
ret = mapi_execute_internal(hdl);
+ if (ret == MOK)
+ ret = read_into_cache(hdl, 1);
if (ret == MOK) {
/* look ahead to detect errors */
ret = mapi_quick_response(hdl, fd);
diff --git a/sql/backends/monet5/merovingian/daemon/multiplex-funnel.c
b/sql/backends/monet5/merovingian/daemon/multiplex-funnel.c
--- a/sql/backends/monet5/merovingian/daemon/multiplex-funnel.c
+++ b/sql/backends/monet5/merovingian/daemon/multiplex-funnel.c
@@ -414,7 +414,7 @@
mapi_cache_limit(m->dbcv[i]->conn, -1); /* don't page */
}
- hdl[i] = mapi_query(m->dbcv[i]->conn, buf);
+ hdl[i] = mapi_send(m->dbcv[i]->conn, buf);
}
/* fail as soon as one of the servers fails */
t = NULL;
@@ -422,6 +422,13 @@
fcnt = -1;
/* only support Q_TABLE, because appending is easy */
for (i = 0; i < m->dbcc; i++) {
+ if (mapi_read_response(hdl[i]) != MOK) {
+ mnstr_printf(fout, "!node %s failed: no response\n",
+ m->dbcv[i]->database);
+ Mfprintf(stderr, "mapi_read_response for %s returned
0\n",
+ m->dbcv[i]->database);
+ break;
+ }
if ((t = mapi_result_error(hdl[i])) != NULL) {
mnstr_printf(fout, "!node %s failed: %s\n",
m->dbcv[i]->database, t);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
