Changeset: 92bb5c749bd0 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=92bb5c749bd0
Modified Files:
sql/src/backends/monet5/merovingian/daemon/client.c
sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c
sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.c
sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.h
Branch: default
Log Message:
multiplex-funnel: add on-the-fly connection switchovers
Multiplex funnels are now hooked into the neighbour discovery system of
merovingian. Unresolved targets of the multiplex will be resolved as
soon as a new database becomes available that satisfies the pattern.
Databases that disappear will be removed, and their connection will be
cycled to the next provider, and if there is none, the connection is
left unsatisfied. The funnel will return an error about its
incompleteness in such case.
diffs (truncated from 501 to 300 lines):
diff -r 49729eb8f31f -r 92bb5c749bd0
sql/src/backends/monet5/merovingian/daemon/client.c
--- a/sql/src/backends/monet5/merovingian/daemon/client.c Tue Jan 04
11:16:31 2011 +0100
+++ b/sql/src/backends/monet5/merovingian/daemon/client.c Tue Jan 04
15:16:55 2011 +0100
@@ -240,16 +240,30 @@
break;
}
if (w == NULL) {
+ char *err;
w = malloc(sizeof(mplist));
w->next = mero_multiplex_funnel;
- w->mpf = multiplexInit(database);
+ if ((err = multiplexInit(&w->mpf, database)) != NULL) {
+ free(w);
+ mnstr_printf(fout, "!merovingian: failed to
create "
+ "multiplex-funnel: %s\n", err);
+ mnstr_flush(fout);
+ close_stream(fout);
+ close_stream(fdin);
+ return(err);
+ }
mero_multiplex_funnel = w;
if (pthread_create(&w->mpf->tid,
NULL, (void *(*)(void
*))multiplexThread,
(void *)w->mpf) < 0)
{
- Mfprintf(stderr, "starting multiplex-funnel %s
failed\n",
- database);
+ mnstr_printf(fout, "!merovingian: internal
failure while "
+ "creating multiplex-funnel:
unable to start thread\n");
+ mnstr_flush(fout);
+ close_stream(fout);
+ close_stream(fdin);
+ return(newErr("starting thread for
multiplex-funnel %s failed",
+ database));
}
}
multiplexAddClient(w->mpf, sock, fout, fdin, host);
diff -r 49729eb8f31f -r 92bb5c749bd0
sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c
--- a/sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c Tue Jan
04 11:16:31 2011 +0100
+++ b/sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c Tue Jan
04 15:16:55 2011 +0100
@@ -34,6 +34,7 @@
#include <utils/properties.h>
#include "merovingian.h"
+#include "multiplex-funnel.h"
#include "discoveryrunner.h"
@@ -59,19 +60,14 @@
remotedb rdb;
remotedb prv;
char hadmatch = 0;
- /* look for the database, and verify that its "conn"
- * (merovingian) is the same */
-
- /* technically, we could use Diffie-Hellman (without Debian
- * modifications) to negotiate a shared secret key, such
- * that only the original registrant can unregister a
- * database, however... do we really care that much? */
pthread_mutex_lock(&_mero_remotedb_lock);
prv = NULL;
rdb = _mero_remotedbs;
while (rdb != NULL) {
+ /* look for the database, and verify that its "conn"
+ * (merovingian) is the same */
if (strcmp(dbname, rdb->dbname) == 0 &&
strcmp(conn, rdb->conn) == 0)
{
@@ -81,6 +77,10 @@
} else {
prv->next = rdb->next;
}
+
+ /* inform multiplex-funnels about this removal */
+ multiplexNotifyRemovedDB(rdb->fullname);
+
Mfprintf(_mero_discout,
"removed neighbour database %s%s\n",
conn, rdb->fullname);
@@ -89,7 +89,7 @@
free(rdb->fullname);
free(rdb);
hadmatch = 1;
- /* there may be more, keep looking */
+ /* in the future, there may be more, so keep looking */
}
prv = rdb;
rdb = rdb->next;
@@ -142,6 +142,9 @@
pthread_mutex_unlock(&_mero_remotedb_lock);
+ /* inform multiplex-funnels about this addition */
+ multiplexNotifyAddedDB(rdb->fullname);
+
return(1);
}
diff -r 49729eb8f31f -r 92bb5c749bd0
sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.c
--- a/sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.c Tue Jan
04 11:16:31 2011 +0100
+++ b/sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.c Tue Jan
04 15:16:55 2011 +0100
@@ -22,22 +22,228 @@
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
+#include <pthread.h>
#include <gdk.h>
#include <mal_sabaoth.h>
#include <Mapi.h>
+#include "utils/glob.h"
+
#include "merovingian.h"
#include "discoveryrunner.h"
#include "multiplex-funnel.h"
-multiplex *
-multiplexInit(char *database)
+typedef struct _multiplexlist {
+ multiplex *m;
+ struct _multiplexlist *next;
+} multiplexlist;
+
+static multiplexlist *multiplexes = NULL;
+static pthread_t mfmanager = 0;
+static int mfpipe[2];
+/**
+ * Connections from all multiplex funnels are maintained by a single
+ * thread that resolves and creates connections upon updates on the
+ * discovery space. Connections aren't made/checked/updated upon their
+ * usage, because this introduces delays for the clients. This is in
+ * particular an issue when a target is updated to point to another
+ * database. To maintain a stable query performance, the connection
+ * creation must happen in the background and set life once established.
+ */
+void
+MFconnectionManager(void *d)
+{
+ int i;
+ multiplex *m;
+ multiplexlist *w;
+ char buf[1024];
+ size_t len;
+ ptr p;
+ char *msg;
+
+ (void)d;
+
+ while (_mero_keep_listening) {
+ /* FIXME: use select for timeout */
+ if (read(mfpipe[0], &p, sizeof(ptr)) < 0) {
+ Mfprintf(stderr, "failed reading from notification
pipe: %s\n",
+ strerror(errno));
+ break;
+ }
+ msg = (char *)p;
+
+ /* intended behaviour:
+ * - additions don't change any connection targets, they only
+ * fill in gaps (conn == NULL)
+ * - removals of targets in use, cause a re-lookup of the
+ * original pattern, on failure, conn is left NULL
+ */
+ if (msg[0] == '+') { /* addition */
+ for (w = multiplexes; w != NULL; w = w->next) {
+ m = w->m;
+ for (i = 0; i < m->dbcc; i++) {
+ if (m->dbcv[i]->conn == NULL) {
+ len = snprintf(buf,
sizeof(buf), "%s/*",
+
m->dbcv[i]->database);
+ if (len >= sizeof(buf)) {
+ Mfprintf(stderr,
"buffer buf too small, "
+
"increase size in %s:%d\n",
+
__FILE__, __LINE__);
+ continue;
+ }
+ /* avoid double /'*'/'* (no ')
*/
+ if (len >= 4 &&
+ buf[len - 3] ==
'*' && buf[len - 4] == '/')
+ buf[len - 2] = '\0';
+ if (glob(buf, msg + 1) == 1) {
+ sabdb *stats;
+ Mapi tm = NULL;
+ /* match! eat away
trailing / (for matching) */
+ msg[strlen(msg) - 1] =
'\0';
+ stats = getRemoteDB(msg
+ 1);
+ if (stats == NULL) {
+
Mfprintf(stderr, "target %s cannot be resolved "
+
"despite being just discovered as %s\n",
+
m->dbcv[i]->database, msg + 1);
+ continue;
+ }
+ snprintf(buf,
sizeof(buf), "%s%s",
+
stats->conns->val, stats->dbname);
+
SABAOTHfreeStatus(&stats);
+ Mfprintf(stdout,
"setting up multiplexer "
+ "target
%s->%s\n",
+
m->dbcv[i]->database, buf);
+ tm = mapi_mapiuri(buf,
+
m->dbcv[i]->user, m->dbcv[i]->pass, "sql");
+ if (mapi_reconnect(tm)
== MOK) {
+
m->dbcv[i]->conn = tm;
+ } else {
+
Mfprintf(stdout, "failed to connect to %s: %s\n",
+
buf, mapi_error_str(tm));
+
mapi_destroy(tm);
+ }
+ }
+ }
+ }
+ }
+ } else { /* removal */
+ for (w = multiplexes; w != NULL; w = w->next) {
+ m = w->m;
+ for (i = 0; i < m->dbcc; i++) {
+ if (m->dbcv[i]->conn != NULL) {
+ len = snprintf(buf,
sizeof(buf), "%s/*",
+
m->dbcv[i]->database);
+ if (len >= sizeof(buf)) {
+ Mfprintf(stderr,
"buffer buf too small, "
+
"increase size in %s:%d\n",
+
__FILE__, __LINE__);
+ continue;
+ }
+ /* avoid double /'*'/'* (no ')
*/
+ if (len >= 4 &&
+ buf[len - 3] ==
'*' && buf[len - 4] == '/')
+ buf[len - 2] = '\0';
+ if (glob(buf, msg + 1) == 1) {
+ /* reevaluate, to see
if connection is still
+ * available */
+ sabdb *walk;
+ sabdb *stats =
getRemoteDB(m->dbcv[i]->database);
+ Mapi tm =
m->dbcv[i]->conn;
+ char *uri =
mapi_get_uri(tm);
+ if (stats == NULL) {
+
Mfprintf(stderr, "target %s can no longer "
+
"be resolved\n",
+
m->dbcv[i]->database);
+
m->dbcv[i]->conn = NULL;
+
mapi_disconnect(tm);
+
mapi_destroy(tm);
+ continue;
+ }
+ /* walk all
connections, in an attempt to
+ * see if the original
connection is still
+ * available, despite
the removal of the
+ * server we got a
message for */
+ for (walk = stats; walk
!= NULL; walk = walk->next) {
+ snprintf(buf,
sizeof(buf), "%s%s",
+
walk->conns->val, walk->dbname);
+ if (strcmp(uri,
buf) == 0)
+ break;
+ }
+ if (walk == NULL) {
+ Mapi ttm;
+ snprintf(buf,
sizeof(buf), "%s%s",
+
stats->conns->val, stats->dbname);
+
Mfprintf(stdout, "changing multiplexer target %s: %s->%s\n",
+
m->dbcv[i]->database, uri, buf);
+ tm =
mapi_mapiuri(buf,
+
m->dbcv[i]->user, m->dbcv[i]->pass,
+
"sql");
+ if
(mapi_reconnect(tm) != MOK) {
+
Mfprintf(stderr, "mapi_reconnect for %s "
+
"failed: %s\n",
+
m->dbcv[i]->database,
+
mapi_error_str(tm));
+
mapi_destroy(tm);
+ tm =
m->dbcv[i]->conn;
+
m->dbcv[i]->conn = NULL;
+
mapi_disconnect(tm);
+
mapi_destroy(tm);
+
SABAOTHfreeStatus(&stats);
+
continue;
+ }
+
+ /* set the new
connection live */
+ ttm =
m->dbcv[i]->conn;
+
m->dbcv[i]->conn = tm;
+
mapi_disconnect(ttm);
+
mapi_destroy(ttm);
+ }
+
SABAOTHfreeStatus(&stats);
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
+void
+multiplexNotifyAddedDB(const char *database)
+{
+ char dbslash[256];
+ ptr p;
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
