Changeset: 04fe188596d5 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=04fe188596d5
Modified Files:
MonetDB5/src/modules/mal/trader.mx
MonetDB5/src/optimizer/opt_octopus.mx
MonetDB5/src/scheduler/run_octopus.mx
sql/src/backends/monet5/Makefile.ag
sql/src/backends/monet5/sql.mx
Branch: default
Log Message:
Merging of octopus code. Main changes:
1) octopus head uses 1 connection per query/ per tentacle and keeps internally
track of connections to peers. Hence remote.connect is replaced by
octopus.connect thas uses the internal info;
2) sql.mvc calls are cut off the tentacle plan and put inside octopus.bind()
function in order to be executed remotely on the site where the query and data
come from.
diffs (truncated from 893 to 300 lines):
diff -r 9f6632051030 -r 04fe188596d5 MonetDB5/src/modules/mal/trader.mx
--- a/MonetDB5/src/modules/mal/trader.mx Tue Jun 15 10:38:44 2010 +0200
+++ b/MonetDB5/src/modules/mal/trader.mx Tue Jun 15 12:58:29 2010 +0200
@@ -69,15 +69,16 @@
#include <mal_recycle.h>
#include "trader.h"
-/*
-static lng estimateSavings0(MalBlkPtr mb)
+
+static lng estimateSavings0(MalBlkPtr mb, sht bidtype)
{
(void) mb;
+ (void) bidtype;
return (lng)100;
}
-*/
-static lng estimateSavings(MalBlkPtr mb, sht bidtype)
+
+lng estimateSavings(MalBlkPtr mb, sht bidtype)
{
MalStkPtr stk = NULL;
int i, j, k, marked = 0;
@@ -105,7 +106,7 @@
for(i= 1; i< mb->vtop; i++) {
lhs = &stk->stk[i];
if( isVarConstant(mb,i) > 0 ){
- assert(!isVarCleanup(mb,i));
+/* assert(!isVarCleanup(mb,i)); */
if( !isVarDisabled(mb,i)){
rhs = &getVarConstant(mb,i);
*lhs = *rhs;
@@ -194,7 +195,7 @@
if ( sym == NULL)
throw(MAL,"trader.makeBid", RUNTIME_SIGNATURE_MISSING "%s", fnname);
- *c = estimateSavings(sym->def, bidtype);
+ *c = estimateSavings0(sym->def, bidtype);
return MAL_SUCCEED;
}
diff -r 9f6632051030 -r 04fe188596d5 MonetDB5/src/optimizer/opt_octopus.mx
--- a/MonetDB5/src/optimizer/opt_octopus.mx Tue Jun 15 10:38:44 2010 +0200
+++ b/MonetDB5/src/optimizer/opt_octopus.mx Tue Jun 15 12:58:29 2010 +0200
@@ -300,6 +300,10 @@
for (; i > 0; i--){
/* p= getInstrPtr(mb,i); */
p = old[i];
+
+ if (getModuleId(p) == sqlRef && getFunctionId(p) == mvcRef )
+ continue;
+
fnd = 0;
if ( p->barrier ){
for (j = 0; j < p->argc; j++)
@@ -315,22 +319,9 @@
}
}
-/* snprintf(buf,BUFSIZ,"tentacle_%s_%d",
- getFunctionId(getInstrPtr(mb,0)),varid);
-
- snprintf(buf,BUFSIZ,"tentacle_"LLFMT"_%d", mb->octid,varid); */
- name = (str) getVarValue(mb, tnm[tidx]);
- putName(name,strlen(name));
-
snprintf(buf,BUFSIZ,"exec_"LLFMT, mb->octid);
name2 = putName(buf,strlen(buf));
-/* cst.vtype = TYPE_str;
- snprintf(buf,BUFSIZ,"tbd_%d",varid);
- cst.val.sval= GDKstrdup(buf);
- cst.len= (int) strlen(cst.val.sval);
- tbd = defConstant(mb, TYPE_str, &cst); */
-
/* generate tentacle CALLS: every tentacle is called once on each worker: tcnt
times */
for ( k = 0; k < tcnt; k++ ){
q= newStmt2(mb,octopusRef, name2);
@@ -345,7 +336,10 @@
q = pushArgument(mb,q,getArg(sig, j));
}
- /* create the TENTACLE procedure */
+ /* create the TENTACLE function */
+ name = (str) getVarValue(mb, tnm[tidx]);
+ putName(name,strlen(name));
+
s= newFunction(octopusRef,name,FUNCTIONsymbol);
tm= s->def;
tm->keephistory= mb->keephistory;
@@ -378,14 +372,16 @@
setModuleId(p, octopusRef);
setVarUDFtype(tm,getArg(p,0));
setArgType(tm,p,0, getArgType(mb,list[top],0));
- p= setArgument(tm,p,1,conn);
+/* p= setArgument(tm,p,1,conn); */
+ getArg(p,1) = conn;
p = pushArgument(tm, p, getArg(getInstrPtr(tm,0),1));
}
if (getModuleId(p) == sqlRef && getFunctionId(p) == binddbatRef
){
setModuleId(p, octopusRef);
setVarUDFtype(tm,getArg(p,0));
setArgType(tm,p,0, getArgType(mb,list[top],0));
- p= setArgument(tm,p,1,conn);
+/* p= setArgument(tm,p,1,conn); */
+ getArg(p,1) = conn;
p = pushArgument(tm, p, getArg(getInstrPtr(tm,0),1));
}
pushInstruction(tm, p);
@@ -655,7 +651,7 @@
}
/* freeInstruction(p); */
- /* now finalize the exec operation */
+ /* finalize the exec function (mal block sm) */
if (isaBatType(mpArgType)) { /*
exec_qry:= bat.new(:htp,:ttp); */
q = newFcnCall(sm, batRef, newRef);
@@ -675,14 +671,14 @@
r = pushArgument(sm,r,schvar);
r = pushBte(sm,r,0);
- q = newStmt(sm, remoteRef,connectRef);
+ q = newStmt(sm, octopusRef,connectRef);
conn= getArg(q,0);
q = pushArgument(sm, q, dbvar);
+/* q= pushStr(sm,q,"monetdb");
q= pushStr(sm,q,"monetdb");
- q= pushStr(sm,q,"monetdb");
- q= pushStr(sm,q,"msql");
+ q= pushStr(sm,q,"msql"); */
- /* v:= remote.put(version) */
+ /* v:= remote.put(conn,version) */
q= newFcnCall(sm,remoteRef,putRef);
q= pushArgument(sm,q,conn);
q= pushArgument(sm,q,version);
@@ -727,14 +723,14 @@
/* catch and propagate errors */
newCatchStmt(sm, "ANYexception");
- q = newStmt(sm, remoteRef,disconnectRef);
- pushArgument(sm,q,conn);
+/* q = newStmt(sm, remoteRef,disconnectRef);
+ pushArgument(sm,q,conn); */
newRaiseStmt(sm,"ANYexception");
newExitStmt(sm, "ANYexception");
- /* close connection */
+ /* close connection
q = newStmt(sm, remoteRef,disconnectRef);
- pushArgument(sm,q,conn);
+ pushArgument(sm,q,conn); */
/* exec_qry:= l; */
q= newAssignment(sm);
@@ -794,7 +790,7 @@
mb->errors++;
}
chkDeclarations(mb);
-/* chkProgram(cntxt->nspace,mb); */
+ chkProgram(cntxt->nspace,mb);
#ifdef DEBUG_OPT_OCTOPUS
if (actions) {
stream_printf(cntxt->fdout, "DONE octopus \n");
diff -r 9f6632051030 -r 04fe188596d5 MonetDB5/src/scheduler/run_octopus.mx
--- a/MonetDB5/src/scheduler/run_octopus.mx Tue Jun 15 10:38:44 2010 +0200
+++ b/MonetDB5/src/scheduler/run_octopus.mx Tue Jun 15 12:58:29 2010 +0200
@@ -63,6 +63,14 @@
address OCTOPUSdiscoverRegister
comment "Discover workers and register tentacles.";
+pattern octopus.register(dbname:str,fname:str):void
+address OCTOPUSregister
+comment "Register mal function fname at server dbname.";
+
+command octopus.connect(dbname:str):str
+address OCTOPUSconnect
+comment "Returns the connection to the server dbname.";
+
pattern scheduler.bidding():bit
address OCTOPUSbidding
comment "Collect bids from workers.";
@@ -78,7 +86,7 @@
#include "mal_instruction.h"
#include "mal_client.h"
-/* #define DEBUG_RUN_OCTOPUS to trace processing */
+#define DEBUG_RUN_OCTOPUS /* to trace processing */
#ifdef WIN32
#ifndef LIBRUN_OCTOPUS
@@ -92,8 +100,10 @@
octopus_export str OCTOPUSrun(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr p);
octopus_export str OCTOPUSdiscoverRegister(Client cntxt, MalBlkPtr mb,
MalStkPtr stk, InstrPtr p);
+octopus_export str OCTOPUSregister(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr pci);
octopus_export str OCTOPUSbidding(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr pci);
octopus_export str OCTOPUSmakeSchedule(Client cntxt, MalBlkPtr mb, MalStkPtr
stk, InstrPtr p);
+octopus_export str OCTOPUSconnect(str *c, str *dbname);
#endif /* MAL_RUN_OCTOPUS */
@+ Octopus scheduling implementation
@@ -115,23 +125,108 @@
} *Registry;
typedef struct {
- str name;
str uri;
str usr;
str pwd;
+ Registry nxt; /* list of registered mal functions */
+ bte active;
+ str conn;
+} Peer;
+
+typedef struct {
+ int pnum;
+ str name;
+/* str uri;
+ str usr;
+ str pwd;
lng *bids;
- Registry nxt; /* list of registered queries */
+ Registry nxt;*/ /* list of registered queries */
} Worker;
#define MAXSITES 2048 /* should become dynamic at some point */
-static Worker workers[MAXSITES];
-static int nroctopus = 0;
+static Peer peers[MAXSITES]; /* registry of peer servers */
+static Worker workers[MAXSITES]; /* registry of workers for the current query
*/
+static int nrworkers = 0;
+static int nrpeers=0;
bte optTarget = 1;
/*sht bidStrategy = 1;
#define BID_TRANS 1
#define BID_COVER 2
*/
+static int
+OCTOPUSfindPeer(str uri)
+{
+ int i;
+ for (i=0; i<nrpeers; i++){
+ if ( strcmp(uri, peers[i].uri) == 0 ){
+ return i;
+ }
+ }
+ return -1;
+}
+
+/* Look for and add a peer with uri in the registry.
+Return index in registry */
+static int
+OCTOPUSgetPeer(str uri)
+{
+ int i;
+
+ i = OCTOPUSfindPeer(uri);
+ if ( i >=0 ) {
+ peers[i].active = 1;
+ return i;
+ }
+ i = nrpeers;
+ peers[i].usr = GDKstrdup("monetdb");
+ peers[i].uri = GDKstrdup(uri);
+ peers[i].pwd = GDKstrdup("monetdb");
+ peers[i].active = 1;
+ peers[i].nxt = NULL;
+ nrpeers++;
+ return i;
+}
+
+/* Clean function registry of non-active peers */
+
+void OCTOPUScleanFunReg(int i)
+{
+ Registry r, q;
+ r = peers[i].nxt;
+ while ( r ) {
+ q = r->nxt;
+ GDKfree(r->fcn);
+ GDKfree(r);
+ r = q;
+ }
+ peers[i].nxt = NULL;
+}
+
+str
+OCTOPUSconnect(str *c, str *dburi)
+{
+ int i;
+ str msg = MAL_SUCCEED;
+ str conn = NULL;
+
+ i = OCTOPUSfindPeer(*dburi);
+ if ( i < 0 ){
+ msg = createException(MAL, "octopus.connect", "Server %s is not
registered", dburi);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
