Changeset: fa2ab64bd2c2 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=fa2ab64bd2c2
Modified Files:
MonetDB5/conf/monetdb5.conf.in
MonetDB5/src/optimizer/opt_octopus.mx
MonetDB5/src/scheduler/run_octopus.mx
Branch: default
Log Message:
Octopus extension assuming full replication;
Fixed type resolution in tentacles;
Provision for tentacles with different signatures.
diffs (truncated from 326 to 300 lines):
diff -r 02d41779b83b -r fa2ab64bd2c2 MonetDB5/conf/monetdb5.conf.in
--- a/MonetDB5/conf/monetdb5.conf.in Wed Sep 22 17:38:05 2010 +0200
+++ b/MonetDB5/conf/monetdb5.conf.in Wed Sep 22 17:45:20 2010 +0200
@@ -239,7 +239,7 @@
#
# The Octopus pipeline for distributed processing (Merovingian enabled
# platforms only)
-octopus_pipe=inline,remap,evaluate,costModel,coercions,emptySet,aliases,mitosis,mergetable,deadcode,constants,commonTerms,joinPath,octopus,deadcode,reduce,dataflow,history,multiplex,garbageCollector
+octopus_pipe=inline,remap,evaluate,costModel,coercions,emptySet,aliases,mitosis,mergetable,deadcode,constants,commonTerms,joinPath,deadcode,octopus,reduce,dataflow,history,multiplex,garbageCollector
tarantula_pipe=inline,remap,evaluate,costModel,coercions,emptySet,aliases,mitosis,mergetable,deadcode,constants,commonTerms,joinPath,tarantula,deadcode,reduce,dataflow,history,multiplex,garbageCollector
#
# The MapReduce pipeline for a different flavour of distributed
diff -r 02d41779b83b -r fa2ab64bd2c2 MonetDB5/src/optimizer/opt_octopus.mx
--- a/MonetDB5/src/optimizer/opt_octopus.mx Wed Sep 22 17:38:05 2010 +0200
+++ b/MonetDB5/src/optimizer/opt_octopus.mx Wed Sep 22 17:45:20 2010 +0200
@@ -165,6 +165,7 @@
#define _OPT_OCTOPUS_
#include "opt_prelude.h"
#include "opt_support.h"
+#include "opt_mitosis.h"
opt_export str OPTlegAdvice(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr pci);
opt_export int OPTlegAdviceInternal(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci);
@@ -205,6 +206,8 @@
static MalPartPtr octCluster = NULL;
static int octClCnt = 0;
+static int octClResSize = 32; /* maximum number of results per tentacle */
+static int octFullRepl = 1;
static void
OCTinitMalPart()
@@ -218,7 +221,7 @@
}
GDKfree(octCluster);
}
- octCluster = (MalPartPtr) GDKzalloc( sizeof(MalPart) * 1000);
+ octCluster = (MalPartPtr) GDKzalloc( sizeof(MalPart) * MAXSLICES);
octClCnt = 0;
}
@@ -237,7 +240,7 @@
octCluster[i].tab = GDKstrdup(tab);
octCluster[i].low = low;
octCluster[i].hgh = hgh;
- octCluster[i].ret = (int*) GDKzalloc( sizeof(int) * 20);
+ octCluster[i].ret = (int*) GDKzalloc( sizeof(int) * octClResSize);
octCluster[i].retcnt = 0;
octClCnt++;
return i;
@@ -247,19 +250,23 @@
OCTaddResult(int cl, int residx)
{
MalPartPtr c;
+ int i, found = 0;
+
if ( cl <= 0 || cl >= octClCnt ) {
mnstr_printf(GDKout, "Illegal instruction partition index \n");
return;
}
c = &octCluster[cl];
- if ( c->retcnt < 20 ){
- c->ret[c->retcnt++] = residx;
+ for (i = 0; i < c->retcnt; i++)
+ if ( c->ret[i] == residx )
+ found = 1;
+ if (!found){
+ if ( c->retcnt < octClResSize ){
+ c->ret[c->retcnt++] = residx;
+ }
+ else
+ mnstr_printf(GDKout, "No room for more results \n");
}
- else {
- mnstr_printf(GDKout, "No room for more results \n");
- return;
- }
-
}
static int
@@ -316,7 +323,7 @@
InstrPtr p, tp, sig = mb->stmt[0], tsig;
char buf[BUFSIZ];
int *alias= (int*) GDKzalloc(mb->vtop * sizeof(int));
- int i, j, k, conn, last, tpe;
+ int i, j, k, conn= 0, last, tpe;
str nm;
ocl = &octCluster[tidx];
@@ -353,7 +360,8 @@
}
tsig = p;
- conn = OCTinitcode(cntxt, tmb);
+ if ( !octFullRepl )
+ conn = OCTinitcode(cntxt, tmb);
for ( i = 1; i < mb->stop; i++ ){ /* copy all instr of cluster
tidx and 0 */
p = mb->stmt[i];
@@ -363,7 +371,8 @@
}
if ( cl[getArg(p,0)] && cl[getArg(p,0)] != tidx )
continue;
- if (getModuleId(p) == sqlRef && getFunctionId(p) == mvcRef )
+ if (getModuleId(p) == sqlRef && getFunctionId(p) == mvcRef
+ && !octFullRepl)
continue;
tp = copyInstruction(p);
for ( j = 0; j< p->argc; j++){
@@ -371,7 +380,9 @@
if ( alias[a]==0 )
alias[a] = cloneVariable(tmb,mb,a);
getArg(tp,j) = alias[a];
+ setVarUDFtype(tmb, alias[a]);
}
+ if ( !octFullRepl ) {
if (getModuleId(p) == sqlRef &&
strcmp(getFunctionId(p), "getVariable") == 0 ){
getModuleId(tp) = octopusRef;
@@ -388,6 +399,7 @@
getArg(tp,1) = conn;
tp = pushArgument(tmb, tp, getArg(tsig,tsig->retc));
}
+ }
if (getModuleId(p) == algebraRef &&
( getFunctionId(p) == joinPathRef ||
getFunctionId(p) == leftjoinPathRef )){
@@ -397,6 +409,7 @@
pushInstruction(tmb, tp);
}
+ if ( !octFullRepl ) {
/* exeption block */
newCatchStmt(tmb, "ANYexception");
tp = newStmt(tmb, remoteRef, disconnectRef);
@@ -406,6 +419,7 @@
tp = newStmt(tmb, remoteRef, disconnectRef);
pushArgument(tmb, tp, conn);
+ }
/* return stmt */
tp = newAssignment(tmb);
@@ -445,7 +459,9 @@
insertSymbol(findModule(cntxt->nspace,octopusRef),s);
clrDeclarations(tmb);
chkProgram(cntxt->nspace,tmb);
-/* printFunction(cntxt->fdout, tmb, 0, LIST_MAL_STMT | LIST_MAL_UDF |
LIST_MAL_PROPS); */
+ OPTDEBUGoctopus{
+ printFunction(cntxt->fdout, tmb, 0, LIST_MAL_STMT |
LIST_MAL_UDF | LIST_MAL_PROPS);
+ }
GDKfree(alias);
return tmb;
@@ -592,9 +608,9 @@
q->barrier = BARRIERsymbol;
rexit = getArg(q,0);
- snprintf(buf,BUFSIZ,"exec_"LLFMT, mb->legid);
- name2 = putName(buf,strlen(buf));
for (j = 0; j < tcnt; j++){ /* generate tentacle CALLs:
every tentacle is called once */
+ snprintf(buf,BUFSIZ,"exec_"LLFMT"_%d", mb->legid, j+1);
+ name2 = putName(buf,strlen(buf));
q= newStmt2(mb,octopusRef, name2);
getArg(q,0) = res[j][0];
setVarUDFtype(mb,getArg(q,0));
@@ -764,12 +780,14 @@
insertSymbol(findModule(cntxt->nspace,octopusRef),s);
clrDeclarations(sm);
chkProgram(cntxt->nspace,sm);
-/* printFunction(cntxt->fdout, sm, 0, LIST_MAL_STMT | LIST_MAL_UDF |
LIST_MAL_PROPS ); */
+ OPTDEBUGoctopus{
+ printFunction(cntxt->fdout, sm, 0, LIST_MAL_STMT | LIST_MAL_UDF
| LIST_MAL_PROPS );
+ }
}
/* create the exec function for the tentacles */
static void
-OCTnewExec(Client cntxt, MalBlkPtr mb, MalBlkPtr t)
+OCTnewExec(Client cntxt, MalBlkPtr mb, MalBlkPtr t, int tno)
{
InstrPtr q, tsig = t->stmt[0], sig;
MalBlkPtr sm;
@@ -778,7 +796,7 @@
int i, ai, l, tpe, arg[1024], res[1024], rres[1024], lres[1024];
int conn, dbvar,qvar;
- snprintf(buf,BUFSIZ,"exec_"LLFMT, mb->legid);
+ snprintf(buf,BUFSIZ,"exec_"LLFMT"_%d", mb->legid,tno);
s = newFunction(octopusRef, putName(buf,strlen(buf)), FUNCTIONsymbol);
sm = s->def;
@@ -888,7 +906,9 @@
insertSymbol(findModule(cntxt->nspace,octopusRef),s);
clrDeclarations(sm);
chkProgram(cntxt->nspace,sm);
-/* printFunction(cntxt->fdout, sm, 0, LIST_MAL_STMT | LIST_MAL_UDF |
LIST_MAL_PROPS ); */
+ OPTDEBUGoctopus{
+ printFunction(cntxt->fdout, sm, 0, LIST_MAL_STMT | LIST_MAL_UDF
| LIST_MAL_PROPS );
+ }
}
static int
@@ -901,13 +921,14 @@
oid l,h;
str tnm;
char rname[BUFSIZ];
- MalBlkPtr *tentacle= (MalBlkPtr *) GDKmalloc(sizeof(MalBlkPtr) * 20);
+ MalBlkPtr *tentacle= (MalBlkPtr *) GDKmalloc(sizeof(MalBlkPtr) *
MAXSLICES);
(void) stk;
if ( tentacle == NULL)
return 0;
+/* optDebug |= 1 << DEBUG_OPT_OCTOPUS; */
OPTDEBUGoctopus{
mnstr_printf(cntxt->fdout, "#Octopus optimizer called\n");
printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT | LIST_MAL_UDF
| LIST_MAL_PROPS);
@@ -929,6 +950,7 @@
/* we do not support yet update operations in the octopus */
if ( update || autocommit==0 ) {
GDKfree(tentacle);
+ GDKfree(malPart);
return 0;
}
@@ -977,31 +999,40 @@
} /* for */
/* print mal block annotated with partitions */
+ OPTDEBUGoctopus{
+ for (i = 0; i < limit; i++) {
+ p = old[i];
+ mnstr_printf(cntxt->fdout, "%3d\t",
malPart[getArg(p,0)]);
+ printInstruction(cntxt->fdout,mb,0,p, LIST_MAL_STMT);
+ }
+ for (i = 0; i < octClCnt; i++){
+ mnstr_printf(cntxt->fdout, "Cluster %3d\t", i);
+ if (octCluster[i].retcnt > 0 )
+ for ( j = 0; j< octCluster[i].retcnt; j++ )
+ mnstr_printf(cntxt->fdout, "%3d\t",
octCluster[i].ret[j]);
+ mnstr_printf(cntxt->fdout, "\n");
+ }
+ }
-/* for (i = 0; i < limit; i++) {
- p = old[i];
- mnstr_printf(cntxt->fdout, "%3d\t", malPart[getArg(p,0)]);
- printInstruction(cntxt->fdout,mb,0,p, LIST_MAL_STMT);
+ /* check cluster results */
+ for (i = 1; i < octClCnt; i++){
+ if ( octCluster[i].retcnt == 0 ){
+ mnstr_printf(cntxt->fdout, "Tentacle %d without
result\n", i);
+ GDKfree(tentacle);
+ GDKfree(malPart);
+ return 0;
+ }
}
- for (i = 0; i < octClCnt; i++){
- mnstr_printf(cntxt->fdout, "Cluster %3d\t", i);
- if (octCluster[i].retcnt > 0 )
- for ( j = 0; j< octCluster[i].retcnt; j++ )
- mnstr_printf(cntxt->fdout, "%3d\t",
octCluster[i].ret[j]);
- mnstr_printf(cntxt->fdout, "\n");
- }
-*/
+
/* create tentacles */
v2 = newVariable(mb, GDKstrdup("version"), TYPE_int);
for (i = 1; i < octClCnt; i++){
tentacle[i] = OCTnewTentacle(cntxt, mb,i, v2, malPart);
-/* chkProgram(cntxt->nspace, tentacle[i]);
- printFunction(cntxt->fdout, tentacle[i], 0, LIST_MAL_STMT |
LIST_MAPI); */
+ OCTnewExec(cntxt, mb, tentacle[i], i);
actions++;
}
if ( actions ){
- OCTnewExec(cntxt, mb, tentacle[1]);
OCTnewBidding(cntxt, octClCnt-1);
}
@@ -1009,6 +1040,8 @@
sig = old[0];
if ( newMalBlkStmt(mb, mb->ssize) < 0){
+ GDKfree(tentacle);
+ GDKfree(malPart);
return 0;
}
pushInstruction(mb, sig);
@@ -1059,7 +1092,9 @@
clrDeclarations(mb);
chkProgram(cntxt->nspace,mb);
-/* printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT | LIST_MAPI); */
+ OPTDEBUGoctopus{
+ printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT | LIST_MAPI);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
