Changeset: 30e52b978e67 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=30e52b978e67
Modified Files:
monetdb5/scheduler/run_centipede.mx
monetdb5/scheduler/run_multipede.mx
Branch: default
Log Message:
Set code aside for later
The centipede scheduler is postponed until it works
on a single node system first.
diffs (truncated from 772 to 300 lines):
diff --git a/monetdb5/scheduler/run_multipede.mx
b/monetdb5/scheduler/run_centipede.mx
rename from monetdb5/scheduler/run_multipede.mx
rename to monetdb5/scheduler/run_centipede.mx
--- a/monetdb5/scheduler/run_multipede.mx
+++ b/monetdb5/scheduler/run_centipede.mx
@@ -17,38 +17,23 @@
All Rights Reserved.
@
Use a single gauge structure to split the largest table.
-@f opt_slicing
+@f centipede
@a M. Kersten
-@- Slicing
+@- Centipede scheduling.
+TBD when the partition manager works in single node.
@{
@mal
-pattern optimizer.slicing():str
-address OPTslicing;
-pattern optimizer.slicing(mod:str, fcn:str):str
-address OPTslicing
-comment "Modify the plan to exploit parallel processing on multiple cores";
-
-pattern sql.gauges(b:bat[:oid,:any_1]) :any_1...
-address OPTgauges
-comment "Derive a series of gauge values based on sampling";
@h
-#ifndef _OPT_SLICING_
-#define _OPT_SLICING_
+#ifndef _CENTIPEDE_
+#define _CENTIPEDE_
#include "opt_prelude.h"
#include "opt_support.h"
#define MAXSITES 4 /* to be refined */
-@:exportOptimizer(slicing)@
-opt_export str OPTgauges(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
-
-#define _DEBUG_OPT_SLICE_
-#define OPTDEBUGslicing if ( optDebug & ((lng)1 <<DEBUG_OPT_SLICING) )
-#endif
@c
#include "monetdb_config.h"
-#include "opt_slicing.h"
-#include "opt_deadcode.h"
+#include "centipede.h"
#include "mal_builder.h"
#include <mapi.h>
#include "remote.h"
@@ -188,721 +173,3 @@
return MAL_SUCCEED;
}
-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 */
- 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
-OPTslicingStub(Client cntxt, MalBlkPtr mb, int pc, MalBlkPtr pmb, Gauge gauge)
-{
- MalBlkPtr smb = 0;
- Symbol s;
- InstrPtr sig, q, ret;
- int j,k,conn, *arg;
- char nme[BUFSIZ];
- (void) gauge;
-
- /* define the sub query stub for remote processing */
- snprintf(nme,BUFSIZ,"%s_stub%d",getFunctionId( getInstrPtr(mb,0)), pc);
- s = newFunction(userRef, putName(nme, strlen(nme)),FUNCTIONsymbol);
- if ( s == NULL)
- return 0;
- freeMalBlk(s->def);
- s->def = copyMalBlk(pmb); /* get variables */
- smb = s->def;
- if ( newMalBlkStmt(smb,smb->ssize) < 0 )
- return 0;
- pushInstruction(smb, copyInstruction(pmb->stmt[0]));
- getFunctionId( getInstrPtr(smb,0)) = putName(nme,strlen(nme));
- insertSymbol(cntxt->nspace,s);
-
- conn = OPTinitcode(cntxt,smb);
- sig = getInstrPtr(smb,0);
- arg = (int*) GDKzalloc(sizeof(int) * sig->argc);
- /* k:= remote.put(conn,kvar) */
- for (j= sig->retc; 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);
- }
-
- /* (k1,...kn):= remote.exec(conn,slicing,qry,version....) */
- snprintf(nme,BUFSIZ,"%s_sub",getFunctionId( getInstrPtr(mb,0)));
- q= newFcnCall(smb,remoteRef,execRef);
- q->retc= q->argc= 0;
- for (j=0; j < sig->retc; j++){
- arg[j]= newTmpVariable(smb,TYPE_str);
- 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++)
- q = pushArgument(smb,q,arg[j]);
-
- /* return exec_qry; */
- ret = newInstruction(smb, ASSIGNsymbol);
- ret->barrier= RETURNsymbol;
- ret->argc = ret->retc = 0;
- /* l:=remote.get(conn,k) */
- for ( j=0; j< sig->retc; j++){
- q= newFcnCall(smb,remoteRef,getRef);
- q= pushArgument(smb,q,conn);
- q= pushArgument(smb,q,arg[j]);
- k= getArg(q,0);
- setVarType(smb,k, getArgType(smb,sig,j));
- setVarUDFtype(smb, k);
- ret = pushArgument(smb,ret,k);
- 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 */
- newExitStmt(smb, "ANYexception");
-
- q = newStmt(smb, remoteRef, disconnectRef);
- pushArgument(smb, q, conn);
- pushInstruction(smb,ret);
- pushEndInstruction(smb);
-
- GDKfree(arg);
- return smb;
-}
-
-static MalBlkPtr
-OPTslicingCntrl(Client cntxt, MalBlkPtr mb, int pc, MalBlkPtr pmb, Gauge gauge)
-{
- MalBlkPtr cmb;
- Symbol s;
- char nme[BUFSIZ];
- int /*tpe,*/ x, i, j, k, *alias, nrpack;
- InstrPtr call, ret, q, *pack;
-
- (void) gauge;
- /* define the query controller */
- snprintf(nme,BUFSIZ,"%s_cntrl%d",getFunctionId( getInstrPtr(mb,0)),pc);
- s = newFunction(userRef, putName(nme, strlen(nme)),FUNCTIONsymbol);
- if ( s == NULL)
- return 0;
- freeMalBlk(s->def);
- s->def = copyMalBlk(pmb); /* get variables */
- cmb = s->def;
- if ( newMalBlkStmt(cmb,cmb->ssize) < 0 )
- return 0;
- q= getInstrPtr(mb,pc);
- if ( getModuleId(q) == groupRef && getFunctionId(q) == doneRef)
- nrpack = 1;
- else
- nrpack = q->retc;
- pack = (InstrPtr *) GDKzalloc(sizeof(InstrPtr) * nrpack);
- pushInstruction(cmb, copyInstruction(pmb->stmt[0]));
- call = copyInstruction(getInstrPtr(pmb,0));
- cmb->stmt[0]->argc -= 2; /* no bounds */
- getFunctionId( getInstrPtr(cmb,0)) = putName(nme,strlen(nme));
- insertSymbol(cntxt->nspace,s);
-
- /* initialize binds and histograms */
- q= newFcnCall(cmb, sqlRef, mvcRef);
- x= getArg(q,0);
- alias = (int*) GDKzalloc(nrpeers * sizeof(int));
- if( gauge.column) {
- q= newInstruction(cmb, ASSIGNsymbol);
- getModuleId(q) = sqlRef;
- getFunctionId(q) = bindRef;
- q = pushArgument(cmb,q,x);
- j = getArg(q,0) = newTmpVariable(cmb,newBatType(TYPE_oid,
gauge.type));
- setVarUDFtype(cmb,j);
- q= pushStr(cmb,q, gauge.schema);
- q= pushStr(cmb,q, gauge.table);
- q= pushStr(cmb,q, gauge.column);
- q= pushInt(cmb,q, 0);
- pushInstruction(cmb,q);
-
- q= newInstruction(cmb, ASSIGNsymbol);
- getModuleId(q) = sqlRef;
- getFunctionId(q) = putName("gauges",6);
- q = pushArgument(cmb, q, j);
- for ( j = 0; j < nrpeers +1; j++) {
- k = alias[j] = newTmpVariable(cmb, gauge.type);
- q= pushReturn(cmb,q, k);
- }
- pushInstruction(cmb,q);
- }
- /* dataflow */
- if ( gauge.column) {
- for ( k = 0; k < nrpack ; k++){
- pack[k] = newInstruction(cmb,ASSIGNsymbol);
- getModuleId(pack[k]) = matRef;
- getFunctionId(pack[k]) = packRef;
- getArg(pack[k],0) = newTmpVariable(cmb,
newBatType(TYPE_oid, getTailType(getArgType(cmb,getInstrPtr(pmb,0),k))) );
- }
- }
-
- if ( slicingLocal == 0){
- for ( k = 0; k < nrpack ; k++){
- q = newInstruction(cmb,ASSIGNsymbol);
- getArg(q,0) = getArg(pack[k],0);
- pushNil(cmb,q, getArgType(cmb,pack[k],0));
- pushInstruction(cmb,q);
- }
-
- q= newFcnCall(cmb,languageRef,dataflowRef);
- q->barrier= BARRIERsymbol;
- x = getArg(q,0);
- setVarType(cmb,x,TYPE_int);
- }
-
- /* tpe = getArgType( pmb, getInstrPtr(pmb,0),0); */
-
- for ( i = 0; i < nrpeers ; i++) {
- q= copyInstruction(call);
- q->token = ASSIGNsymbol;
- q->barrier = 0;
- for ( k = 0; k < q->retc; k++) {
- getArg(q,k) = newTmpVariable(cmb, getArgType(pmb,q,k));
- pack[k] = pushArgument(cmb,pack[k], getArg(q,k));
- }
-
- if ( gauge.column){
- getArg(q, q->argc-2 ) = alias[i];
- getArg(q, q->argc-1) = alias[i+1];
- }
- pushInstruction(cmb,q);
- }
- if ( gauge.column)
- for ( k = 0; k < nrpack; k++)
- pushInstruction(cmb, pack[k]);
-
-
- if ( slicingLocal == 0){
- q= newAssignment(cmb);
- q->barrier = EXITsymbol;
- getArg(q,0) = x;
- }
-
- q = copyInstruction(getInstrPtr(mb,pc));
- if ( getFunctionId(q) == countRef) {
- getFunctionId(q) = sumRef;
- getArg(q,1) = getArg(pack[0],0);
- } else
- if ( getFunctionId(q) == putName("exportValue",11))
- getArg(q,8)= getArg(pack[0],0);
- else
- if ( getModuleId(q) == groupRef && getFunctionId(q) == doneRef) {
- /* just return the argument */
- clrFunction(q);
- q->argc = q->retc = 1;
- q = pushArgument(cmb,q, getArg(pack[0],0));
- getArg(q,0) = getArg(q,1);
- } else
- if ( getFunctionId(q) == NULL){
- /* simple assignment */
- assert(nrpack == q->retc);
- for( k = 0; k < nrpack; k++)
- getArg(q, q->retc + k) = getArg(pack[k],0);
- }
-
- pushInstruction(cmb,q);
-
- ret = newInstruction(NULL,ASSIGNsymbol);
- ret->barrier = RETURNsymbol;
- ret->argc = ret->retc =0;
- for( i = 0 ; i< q->retc; i++)
- ret = pushReturn(cmb,ret,getArg(q,i));
- pushInstruction(cmb,ret);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
