Changeset: 25c47143f4ec for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=25c47143f4ec
Modified Files:
monetdb5/optimizer/opt_partition.mx
Branch: default
Log Message:
Finetuning for remote execution.
diffs (truncated from 341 to 300 lines):
diff --git a/monetdb5/optimizer/opt_partition.mx
b/monetdb5/optimizer/opt_partition.mx
--- a/monetdb5/optimizer/opt_partition.mx
+++ b/monetdb5/optimizer/opt_partition.mx
@@ -42,6 +42,14 @@ pattern partition.vector(b:bat[:oid,:any
address OPTvector
comment "Derive a series of slices values based on sampling";
+pattern partition.peers( pat:str):int
+address OPTpeers
+comment "Activate the peers matching the pattern and return number found.";
+
+pattern partition.peers( N:int):int
+address OPTpeers
+comment "Consider N peers";
+
@h
#ifndef _OPT_SLICING_
#define _OPT_SLICING_
@@ -53,6 +61,7 @@ comment "Derive a series of slices value
opt_export str OPTvector(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
opt_export str OPTpartitionMaterialize(int *result, int *bid, ptr low, ptr
high);
+opt_export str OPTpeers(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
/* #define DEBUG_DETAIL*/
#define _DEBUG_OPT_PARTITION_
@@ -86,8 +95,8 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
{
MalBlkPtr cmb;
Symbol s;
- char nme[BUFSIZ];
- int x, i, j, k, *alias, nrpack;
+ char nme[BUFSIZ], *plan, *stub;
+ int base, barrier, x, i, j, k, *alias, nrpack;
InstrPtr ret, p, q, *pack;
#ifdef _DEBUG_OPT_PARTITION_
@@ -95,6 +104,11 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
printFunction(cntxt->fdout, mb, 0, LIST_MAL_STMT);
#endif
/* define the query controller */
+ snprintf(nme,BUFSIZ,"%s_plan",getFunctionId( getInstrPtr(mb,0)));
+ plan = putName(nme, strlen(nme));
+ snprintf(nme,BUFSIZ,"%s_stub",getFunctionId( getInstrPtr(mb,0)));
+ stub = putName(nme, strlen(nme));
+
snprintf(nme,BUFSIZ,"%s_cntrl",getFunctionId( getInstrPtr(mb,0)));
s = newFunction(userRef, putName(nme, strlen(nme)),FUNCTIONsymbol);
if ( s == NULL)
@@ -157,12 +171,22 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
pushInstruction(cmb,q);
}
- /* mark start of dataflow block */
- q= newFcnCall(cmb,languageRef,dataflowRef);
+ /* PHASE 1: dont run in distributed mode */
+ q= newFcnCall(cmb,schedulerRef,srvpoolRef);
q->barrier= BARRIERsymbol;
- x = getArg(q,0);
+ barrier = getArg(q,0);
+ q= pushStr(cmb,q, plan);
setVarType(cmb,x,TYPE_int);
+ /* get servers to execute the query */
+ q= newStmt(cmb,srvpoolRef,putName("server",6));
+ q->retc= q->argc = 0;
+ for( i = 0; i < nrpeers; i++)
+ q= pushReturn(cmb,q, newTmpVariable(cmb,TYPE_str));
+ q= pushStr(cmb,q,plan);
+ p= q;
+
+ base = cmb->stop;
for ( i = 0; i < nrpeers ; i++) {
q= copyInstruction(getInstrPtr(pmb,0));
q->token = ASSIGNsymbol;
@@ -178,6 +202,9 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
q= pushArgument(cmb,q,alias[i]);
q= pushArgument(cmb,q,alias[i+1]);
}
+ /* for distributed execution we use the stub */
+ getFunctionId(q) = stub;
+ q= setArgument(cmb,q,q->retc, getArg(p,i));
pushInstruction(cmb,q);
}
/* put all mat.pack instructions into the program
@@ -193,7 +220,35 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
/* finalize the dataflow block */
q= newAssignment(cmb);
q->barrier = EXITsymbol;
- getArg(q,0) = x;
+ getArg(q,0) = barrier;
+
+ /* PHASE 2 The second block runs the query in local mode */
+ q= newStmt(cmb,calcRef,notRef);
+ q->barrier= BARRIERsymbol;
+ q = pushArgument(cmb,q, barrier);
+ barrier = getArg(q,0);
+ setVarType(cmb,x,TYPE_int);
+
+ /* re-use instructions, injecting missing server selections */
+ for ( i = 0; i < nrpeers ; i++) {
+ q= copyInstruction(getInstrPtr(cmb,base + i));
+ getFunctionId(q) = plan;
+ delArgument(q,q->retc);
+ pushInstruction(cmb,q);
+ }
+ /* put all mat.pack instructions into the program
+ make sure that they have contiguous void headed columns
+ */
+ p = getInstrPtr(pmb,0);
+ if ( slices->column)
+ for ( k=0 ; k < nrpack; k++) {
+ pushInstruction(cmb, pack[k]);
+ getArg(pack[k],0)= getArg(p,k);
+ }
+ /* finalize the second dataflow block */
+ q= newAssignment(cmb);
+ q->barrier = EXITsymbol;
+ getArg(q,0) = barrier;
/* consolidate the result */
ret = copyInstruction(getInstrPtr(cmb,0));
@@ -213,30 +268,6 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
return cmb;
}
-static int
-OPTinitcode(Client cntxt, MalBlkPtr mb)
-{
- InstrPtr p;
- str s;
- str l = NULL;
-
- (void) cntxt;
-
- /* _x := remote.connect(uri,"monetdb","monetdb","msql"); */
- p = newStmt(mb, remoteRef,connectRef);
- s = GDKgetenv("merovingian_uri");
- if (s == NULL) /* aparently not under Merovingian control, fall back to
local only */
- s= "dummyconnection";
- /* SABAOTHgetLocalConnection(&l);*/
- p= pushStr(mb,p, s == NULL ? l : s);
- p= pushStr(mb,p,"monetdb");
- p= pushStr(mb,p,"monetdb");
- p= pushStr(mb,p,"msql");
- if (l)
- GDKfree(l);
- return getArg(p,0);
-}
-
static MalBlkPtr
OPTpartitionStub(Client cntxt, MalBlkPtr mb, MalBlkPtr pmb)
{
@@ -260,11 +291,31 @@ OPTpartitionStub(Client cntxt, MalBlkPtr
getFunctionId( getInstrPtr(smb,0)) = putName(nme,strlen(nme));
insertSymbol(cntxt->nspace,s);
- conn = OPTinitcode(cntxt,smb);
+ /* _x := remote.connect(uri,"monetdb","monetdb","msql");
+ conn = newTmpVariable(smb, TYPE_str);
+ q = newStmt(smb, remoteRef,connectRef);
+ q = pushArgument(smb,q,conn);
+ q= pushStr(smb,q,"monetdb");
+ q= pushStr(smb,q,"monetdb");
+ q= pushStr(smb,q,"msql");
+ */
+
+ conn = newTmpVariable(smb, TYPE_str);
+ setArgument(smb,getInstrPtr(smb,0), getInstrPtr(smb,0)->retc, conn);
+ /* conn = getArg(q,0);*/
+
sig = getInstrPtr(smb,0);
arg = (int*) GDKzalloc(sizeof(int) * sig->argc);
/* k:= remote.put(conn,kvar) */
- for (j= sig->retc; j < sig->argc; j++) {
+ for (j= sig->retc+1; j < sig->argc; j++) {
+ q= newFcnCall(smb,remoteRef,putRef);
+ setVarType(smb, getArg(q,0), TYPE_str);
+ setVarUDFtype(smb, getArg(q,0));
+ q= pushArgument(smb,q,conn);
+ q= pushArgument(smb,q,getArg(sig,j));
+ arg[j]= getArg(q,0);
+ }
+ for (j= 0; j < sig->retc; j++) {
q= newFcnCall(smb,remoteRef,putRef);
setVarType(smb, getArg(q,0), TYPE_str);
setVarUDFtype(smb, getArg(q,0));
@@ -273,21 +324,19 @@ OPTpartitionStub(Client cntxt, MalBlkPtr
arg[j]= getArg(q,0);
}
- /* (k1,...kn):= remote.exec(conn,slicing,qry,version....) */
+ /* (k1,...kn):= remote.exec(conn,srvpool,qry,version....) */
snprintf(nme,BUFSIZ,"%s_plan",getFunctionId( getInstrPtr(mb,0)));
q = newInstruction(smb,ASSIGNsymbol);
getModuleId(q) = remoteRef;
getFunctionId(q) = execRef;
q->retc= q->argc= 0;
- for (j=0; j < sig->retc; j++){
- arg[j]= newTmpVariable(smb,TYPE_str);
+ for (j=0; j < sig->retc; j++)
q = pushReturn(smb,q,arg[j]);
- }
q= pushArgument(smb,q,conn);
q= pushStr(smb,q,userRef);
q= pushStr(smb,q,putName(nme,strlen(nme)));
/* deal with all arguments ! */
- for (j=sig->retc; j < sig->argc; j++)
+ for (j=sig->retc+1; j < sig->argc; j++)
q = pushArgument(smb,q,arg[j]);
pushInstruction(smb,q);
@@ -307,14 +356,16 @@ OPTpartitionStub(Client cntxt, MalBlkPtr
ret = pushReturn(smb,ret,getArg(sig,j));
}
+/*
newCatchStmt(smb, "ANYexception");
q = newStmt(smb, remoteRef, disconnectRef);
pushArgument(smb, q, conn);
- newRaiseStmt(smb, "ANYexception"); /* pass to caller */
+ newRaiseStmt(smb, "ANYexception");
newExitStmt(smb, "ANYexception");
q = newStmt(smb, remoteRef, disconnectRef);
pushArgument(smb, q, conn);
+*/
if ( sig->retc)
pushInstruction(smb,ret);
pushEndInstruction(smb);
@@ -457,7 +508,6 @@ OPTplanFragment(Client cntxt, MalBlkPtr
freeMalBlk(s->def);
s->def = copyMalBlk(mb);
nmb = s->def;
- getFunctionId( getInstrPtr(nmb,0)) = putName(nme,strlen(nme));
limit = nmb->stop;
old = nmb->stmt;
@@ -628,6 +678,7 @@ OPTplanFragment(Client cntxt, MalBlkPtr
ret= newInstruction(nmb,ASSIGNsymbol);
ret->barrier = RETURNsymbol;
ret->argc= ret->retc = 0;
+
for ( i = limit-1; i >= 0 ; i--)
if ( plan[i] == REQUIRED || plan[i] == SUPPORTIVE ){
p = old[i];
@@ -663,8 +714,12 @@ OPTplanFragment(Client cntxt, MalBlkPtr
}
/* Phase 4: Bake a new function that produces them */
- plan[0] = REQUIRED;
- for ( i = 0; i < limit ; i++)
+
+ p = copyInstruction(getInstrPtr(mb, 0));
+ getModuleId(p) = userRef;
+ pushInstruction(nmb,p);
+
+ for ( i = 1; i < limit ; i++)
if( plan[i] == REQUIRED || plan[i] == SUPPORTIVE ) {
p = copyInstruction(getInstrPtr(mb, i));
if ( old[i]->token == ENDsymbol) {
@@ -697,16 +752,6 @@ OPTplanFragment(Client cntxt, MalBlkPtr
printFunction(cntxt->fdout, nmb, 0, LIST_MAL_STMT);
#endif
- /* construct the control plan */
- cmb = OPTplanCntrl(cntxt,mb,nmb, slices);
- msg= optimizeMALBlock(cntxt, cmb);
- if ( cmb)
- chkProgram(cntxt->nspace, cmb);
-#ifdef _DEBUG_OPT_PARTITION_
- mnstr_printf(cntxt->fdout,"#control plan errors %d %s
\n",cmb->errors,msg?msg:"");
- printFunction(cntxt->fdout, cmb, 0, LIST_MAL_STMT);
-#endif
-
/* construct the remote stub plan */
smb = OPTpartitionStub(cntxt, mb, nmb);
msg= optimizeMALBlock(cntxt, smb);
@@ -717,6 +762,16 @@ OPTplanFragment(Client cntxt, MalBlkPtr
printFunction(cntxt->fdout, smb, 0, LIST_MAL_STMT);
#endif
+ /* construct the control plan for local/remote execution */
+ cmb = OPTplanCntrl(cntxt,mb,nmb, slices);
+ msg= optimizeMALBlock(cntxt, cmb);
+ if ( cmb)
+ chkProgram(cntxt->nspace, cmb);
+#ifdef _DEBUG_OPT_PARTITION_
+ mnstr_printf(cntxt->fdout,"#control plan errors %d %s
\n",cmb->errors,msg?msg:"");
+ printFunction(cntxt->fdout, cmb, 0, LIST_MAL_STMT);
+#endif
+
call = copyInstruction(getInstrPtr(nmb,0));
call->barrier = 0;
call->token = ASSIGNsymbol;
@@ -863,7 +918,35 @@ OPTpartitionImplementation(Client cntxt,
}
@
@c
-/* determine the initial slices values */
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
