Changeset: 007d6fd7e063 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=007d6fd7e063
Modified Files:
sql/src/backends/monet5/merovingian/daemon/Makefile.ag
sql/src/backends/monet5/merovingian/daemon/client.c
sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c
sql/src/backends/monet5/merovingian/daemon/discoveryrunner.h
sql/src/backends/monet5/merovingian/daemon/forkmserver.c
sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.c
sql/src/backends/monet5/merovingian/daemon/multiplex-funnel.h
Branch: default
Log Message:
Implemented a multiplex funnel.
This functionality has very rough edges currently, and there is no
single client that can use it (mclient doesn't accept language
multiplex-funnel, jdbcclient doesn't render results), but so far it
seems to attach and detach clients fairly well, thereby answering
queries reasonably fast as well.
Committing this to give a heads up on the NOZHUP deliverables status.
diffs (truncated from 882 to 300 lines):
diff -r 6dc5334b7273 -r 007d6fd7e063
sql/src/backends/monet5/merovingian/daemon/Makefile.ag
--- a/sql/src/backends/monet5/merovingian/daemon/Makefile.ag Thu Dec 23
15:23:19 2010 +0100
+++ b/sql/src/backends/monet5/merovingian/daemon/Makefile.ag Thu Dec 23
17:17:43 2010 +0100
@@ -22,6 +22,7 @@
INCLUDES = \
$(MONETDB5_INCS) \
$(MONETDB_INCS) \
+ $(CLIENTS_INCS) \
-I..
EXTRA_DIST = $(man_MANS) \
@@ -32,6 +33,7 @@
discoveryrunner.h \
forkmserver.h \
handlers.h \
+ multiplex-funnel.h \
peering.h \
proxy.h
@@ -45,12 +47,14 @@
discoveryrunner.c \
forkmserver.c \
handlers.c \
+ multiplex-funnel.c \
peering.c \
proxy.c
LIBS = \
../utils/libmeroutil \
$(MONETDB5_LIBS) -lmonetdb5 \
$(MONETDB_LIBS) -lbat -lstream \
+ $(CLIENTS_LIBS) -lMapi \
$(SOCKET_LIBS) \
$(MALLOC_LIBS) \
$(PTHREAD_LIBS)
diff -r 6dc5334b7273 -r 007d6fd7e063
sql/src/backends/monet5/merovingian/daemon/client.c
--- a/sql/src/backends/monet5/merovingian/daemon/client.c Thu Dec 23
15:23:19 2010 +0100
+++ b/sql/src/backends/monet5/merovingian/daemon/client.c Thu Dec 23
17:17:43 2010 +0100
@@ -40,7 +40,9 @@
#include "merovingian.h"
#include "forkmserver.h"
#include "proxy.h"
+#include "multiplex-funnel.h"
+static multiplex *mero_multiplex_funnel = NULL;
static err
handleClient(int sock, char isusock)
@@ -72,6 +74,33 @@
}
fout = block_stream(fout);
+ if (isusock) {
+ host = "(local)";
+ } else if (getpeername(sock, (struct sockaddr *)&saddr, &saddrlen) ==
-1) {
+ Mfprintf(stderr, "couldn't get peername of client: %s\n",
+ strerror(errno));
+ host = "(unknown)";
+ } else {
+ size_t len;
+ struct hostent *hoste =
+ gethostbyaddr(&saddr.sin_addr.s_addr, 4,
saddr.sin_family);
+ if (hoste == NULL) {
+ len = (3 + 1 + 3 + 1 + 3 + 1 + 3 + 1 + 5) + 1;
+ host = alloca(sizeof(char) * len);
+ snprintf(host, len, "%u.%u.%u.%u:%u",
+ (unsigned)
((ntohl(saddr.sin_addr.s_addr) >> 24) & 0xff),
+ (unsigned)
((ntohl(saddr.sin_addr.s_addr) >> 16) & 0xff),
+ (unsigned)
((ntohl(saddr.sin_addr.s_addr) >> 8) & 0xff),
+ (unsigned)
(ntohl(saddr.sin_addr.s_addr) & 0xff),
+ (unsigned) (ntohs(saddr.sin_port)));
+ } else {
+ len = strlen(hoste->h_name) + 1 + 5 + 1;
+ host = alloca(sizeof(char) * len);
+ snprintf(host, len, "%s:%u",
+ hoste->h_name, (unsigned)
(ntohs(saddr.sin_port)));
+ }
+ }
+
/* note that we claim to speak proto 8 here */
mnstr_printf(fout, "%s:merovingian:8:%s:%s:",
"void", /* some bs */
@@ -88,8 +117,10 @@
buf[0] = '\0';
if (mnstr_read_block(fdin, buf, 8095, 1) < 0) {
/* we didn't get a terminated block :/ */
- e = newErr("client sent challenge in incomplete block: %s",
buf);
- mnstr_printf(fout, "!merovingian: client sent something this
server could not understand, sorry\n", user);
+ e = newErr("client %s sent challenge in incomplete block: %s",
+ host, buf);
+ mnstr_printf(fout, "!merovingian: client sent something this "
+ "server could not understand, sorry\n", user);
mnstr_flush(fout);
close_stream(fout);
close_stream(fdin);
@@ -108,7 +139,7 @@
/* mnstr_set_byteorder(fin->s, strcmp(user, "BIG") == 0); */
user = s + 1;
} else {
- e = newErr("client challenge error: %s", buf);
+ e = newErr("client %s challenge error: %s", host, buf);
mnstr_printf(fout, "!merovingian: incomplete challenge '%s'\n",
user);
mnstr_flush(fout);
close_stream(fout);
@@ -123,7 +154,7 @@
passwd = s + 1;
/* decode algorithm, i.e. {plain}mypasswordchallenge */
if (*passwd != '{') {
- e = newErr("client challenge error: %s", buf);
+ e = newErr("client %s challenge error: %s", host, buf);
mnstr_printf(fout, "!merovingian: invalid password
entry\n");
mnstr_flush(fout);
close_stream(fout);
@@ -133,7 +164,7 @@
algo = passwd + 1;
s = strchr(algo, '}');
if (!s) {
- e = newErr("client challenge error: %s", buf);
+ e = newErr("client %s challenge error: %s", host, buf);
mnstr_printf(fout, "!merovingian: invalid password
entry\n");
mnstr_flush(fout);
close_stream(fout);
@@ -143,7 +174,7 @@
*s = 0;
passwd = s + 1;
} else {
- e = newErr("client challenge error: %s", buf);
+ e = newErr("client %s challenge error: %s", host, buf);
mnstr_printf(fout, "!merovingian: incomplete challenge '%s'\n",
user);
mnstr_flush(fout);
close_stream(fout);
@@ -157,7 +188,7 @@
*s = 0;
lang = s + 1;
} else {
- e = newErr("client challenge error: %s", buf);
+ e = newErr("client %s challenge error: %s", host, buf);
mnstr_printf(fout, "!merovingian: incomplete challenge, missing
language\n");
mnstr_flush(fout);
close_stream(fout);
@@ -174,7 +205,7 @@
* for another : */
s = strchr(database, ':');
if (s == NULL) {
- e = newErr("client challenge error: %s", buf);
+ e = newErr("client %s challenge error: %s", host, buf);
mnstr_printf(fout, "!merovingian: incomplete challenge,
missing trailing colon\n");
mnstr_flush(fout);
close_stream(fout);
@@ -192,38 +223,60 @@
mnstr_flush(fout);
close_stream(fout);
close_stream(fdin);
- return(newErr("no database specified"));
- } else {
- if (strcmp(lang, "resolve") == 0) {
- /* ensure the pattern ends with '/\*' such that we
force a
- * remote entry, including those for local databases,
this
- * way we will get a redirect back to merovingian for
such
- * database if it is proxied and hence not remotely
- * available */
- size_t len = strlen(database);
- if (len > 2 &&
- database[len - 2] != '/' &&
- database[len - 1] != '*')
+ return(newErr("client %s specified no database", host));
+ }
+
+ if (strcmp(lang, "multiplex-funnel") == 0) {
+ /* SQL multiplexer with funnelling capabilities */
+ /* find/start/attach funnel */
+ /* FIXME: maintain a list, search, match name, etc.
+ if (strcmp(_multiplex_funnel->name, database) == 0)
+ */
+ if (mero_multiplex_funnel == NULL) {
+ mero_multiplex_funnel = multiplexInit(database);
+ if (pthread_create(&mero_multiplex_funnel->tid,
+ NULL, (void *(*)(void
*))multiplexThread,
+ (void *)mero_multiplex_funnel) < 0)
{
- char *n = alloca(sizeof(char) * len + 2 + 1);
- snprintf(n, len + 2 + 1, "%s/*", database);
- database = n;
+ Mfprintf(stderr, "starting multiplex-funnel %s
failed\n",
+ database);
}
}
- if ((e = forkMserver(database, &top, 0)) != NO_ERR) {
- if (top == NULL) {
- mnstr_printf(fout, "!merovingian: no such
database '%s', please create it first\n", database);
- } else {
- mnstr_printf(fout, "!merovingian: internal
error while starting mserver, please refer to the logs\n");
- }
- mnstr_flush(fout);
- close_stream(fout);
- close_stream(fdin);
- return(e);
+ multiplexAddClient(mero_multiplex_funnel, sock, fout, fdin,
host);
+
+ return(NO_ERR);
+ }
+
+ if (strcmp(lang, "resolve") == 0) {
+ /* ensure the pattern ends with '/\*' such that we force a
+ * remote entry, including those for local databases, this
+ * way we will get a redirect back to merovingian for such
+ * database if it is proxied and hence not remotely
+ * available */
+ size_t len = strlen(database);
+ if (len > 2 &&
+ database[len - 2] != '/' &&
+ database[len - 1] != '*')
+ {
+ char *n = alloca(sizeof(char) * len + 2 + 1);
+ snprintf(n, len + 2 + 1, "%s/*", database);
+ database = n;
}
- stat = top;
}
+ if ((e = forkMserver(database, &top, 0)) != NO_ERR) {
+ if (top == NULL) {
+ mnstr_printf(fout, "!merovingian: no such database
'%s', please create it first\n", database);
+ } else {
+ mnstr_printf(fout, "!merovingian: internal error while
starting mserver, please refer to the logs\n");
+ }
+ mnstr_flush(fout);
+ close_stream(fout);
+ close_stream(fdin);
+ return(e);
+ }
+ stat = top;
+
/* collect possible redirects */
for (stat = top; stat != NULL; stat = stat->next) {
if (stat->conns == NULL || stat->conns->val == NULL) {
@@ -252,33 +305,6 @@
return(e);
}
- if (isusock) {
- host = "(local)";
- } else if (getpeername(sock, (struct sockaddr *)&saddr, &saddrlen) ==
-1) {
- Mfprintf(stderr, "couldn't get peername of client: %s\n",
- strerror(errno));
- host = "(unknown)";
- } else {
- size_t len;
- struct hostent *hoste =
- gethostbyaddr(&saddr.sin_addr.s_addr, 4,
saddr.sin_family);
- if (hoste == NULL) {
- len = (3 + 1 + 3 + 1 + 3 + 1 + 3 + 1 + 5) + 1;
- host = alloca(sizeof(char) * len);
- snprintf(host, len, "%u.%u.%u.%u:%u",
- (unsigned)
((ntohl(saddr.sin_addr.s_addr) >> 24) & 0xff),
- (unsigned)
((ntohl(saddr.sin_addr.s_addr) >> 16) & 0xff),
- (unsigned)
((ntohl(saddr.sin_addr.s_addr) >> 8) & 0xff),
- (unsigned)
(ntohl(saddr.sin_addr.s_addr) & 0xff),
- (unsigned) (ntohs(saddr.sin_port)));
- } else {
- len = strlen(hoste->h_name) + 1 + 5 + 1;
- host = alloca(sizeof(char) * len);
- snprintf(host, len, "%s:%u",
- hoste->h_name, (unsigned)
(ntohs(saddr.sin_port)));
- }
- }
-
/* need to send a response, either we are going to proxy, or we send
* a redirect, if we have multiple options, a redirect is our only
* option, but if the redir is a single remote we need to stick to
diff -r 6dc5334b7273 -r 007d6fd7e063
sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c
--- a/sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c Thu Dec
23 15:23:19 2010 +0100
+++ b/sql/src/backends/monet5/merovingian/daemon/discoveryrunner.c Thu Dec
23 17:17:43 2010 +0100
@@ -29,6 +29,7 @@
#include <gdk.h>
#include <mal_sabaoth.h>
+#include <utils/glob.h>
#include <utils/utils.h>
#include <utils/properties.h>
@@ -144,6 +145,88 @@
return(1);
}
+sabdb *
+getRemoteDB(char *database)
+{
+ struct _remotedb dummy = { NULL, NULL, NULL, NULL, 0, NULL };
+ remotedb rdb = NULL;
+ remotedb pdb = NULL;
+ remotedb down = NULL;
+ sabdb *walk = NULL;
+ sabdb *stats = NULL;
+ size_t dbsize = strlen(database);
+ char *mdatabase = GDKmalloc(sizeof(char) * (dbsize + 2 + 1));
+ char mfullname[8096]; /* should be enough for everyone... */
+
+ /* each request has an implicit /'* (without ') added to match
+ * all sub-levels to the request, such that a request for e.g. X
+ * will return X/level1/level2/... */
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
