Changeset: d15b9eef4bf0 for MonetDB URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=d15b9eef4bf0 Modified Files: monetdb5/scheduler/run_multipede.mx Branch: default Log Message:
Hook for the multipeder scheduler. diffs (truncated from 913 to 300 lines): diff --git a/monetdb5/scheduler/run_multipede.mx b/monetdb5/scheduler/run_multipede.mx new file mode 100644 --- /dev/null +++ b/monetdb5/scheduler/run_multipede.mx @@ -0,0 +1,908 @@ +@/ +The contents of this file are subject to the MonetDB Public License +Version 1.1 (the "License"); you may not use this file except in +compliance with the License. You may obtain a copy of the License at +http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html + +Software distributed under the License is distributed on an "AS IS" +basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the +License for the specific language governing rights and limitations +under the License. + +The Original Code is the MonetDB Database System. + +The Initial Developer of the Original Code is CWI. +Portions created by CWI are Copyright (C) 1997-July 2008 CWI. +Copyright August 2008-2011 MonetDB B.V. +All Rights Reserved. +@ +Use a single gauge structure to split the largest table. +@f opt_slicing +@a M. Kersten +@- Slicing +@{ +@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_ +#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 "mal_builder.h" +#include <mapi.h> +#include "remote.h" +#include "mal_sabaoth.h" +#include "mal_recycle.h" + +#include "mal_interpreter.h" + +typedef struct REGMAL{ + str fcn; + struct REGMAL *nxt; +} *Registry; + +typedef struct { + str uri; + str usr; + str pwd; + Registry nxt; /* list of registered mal functions */ + bte active; + str conn; + int inuse; +} Peer; + +static Peer peers[MAXSITES]; /* registry of peer servers */ +static int nrpeers; /* peers active in sliced processing */ +static bte slicingLocal; /* only use local node without remote calls*/ + +typedef struct{ + InstrPtr target; + str schema, table, column; + int type, slice; + int lgauge, hgauge; + ValRecord bounds[MAXSITES]; +} Gauge; + +static int +OPTfindPeer(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 */ +int +OPTgetPeer(str uri) +{ + int i; + + i = OPTfindPeer(uri); + if ( i >=0 ) { + peers[i].active = 1; + return i; + } + if ( nrpeers == MAXSITES) + return -1; + 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; + peers[i].inuse = 0; + nrpeers++; + return i; +} + +/* Clean function registry of non-active peers */ + +void OPTcleanFunReg(int i) +{ + Registry r, q; + mal_set_lock(mal_contextLock,"slicing.cleanFunReg"); + r = peers[i].nxt; + peers[i].nxt = NULL; + mal_unset_lock(mal_contextLock,"slicing.cleanFunReg"); + while ( r ) { + q = r->nxt; + GDKfree(r->fcn); + GDKfree(r); + r = q; + } +} + +str +OPTdiscover(Client cntxt) +{ + bat bid = 0; + BAT *b; + BUN p,q; + str msg = MAL_SUCCEED; + BATiter bi; + char buf[BUFSIZ]= "*/slicing", *s= buf; + int i, nrworkers = 0; + + slicingLocal = 0; + + /* we have a new list of candidate peers */ + for (i=0; i<nrpeers; i++) + peers[i].active = 0; + + msg = RMTresolve(&bid,&s); + if ( msg == MAL_SUCCEED) { + b = BATdescriptor(bid); + if ( b != NULL && BATcount(b) > 0 ) { + bi = bat_iterator(b); + BATloop(b,p,q){ + str t= (str) BUNtail(bi,p); + nrworkers += OPTgetPeer(t) >= 0; + } + } + BBPreleaseref(bid); + } else + GDKfree(msg); + + if ( !nrworkers ) { + /* there is a last resort, local execution */ + SABAOTHgetLocalConnection(&s); + + nrworkers += OPTgetPeer(s) >= 0; + slicingLocal = 1; + } + +#ifdef DEBUG_RUN_OPT + mnstr_printf(cntxt->fdout,"Active peers discovered %d\n",nrworkers); + for (i=0; i<nrpeers; i++) + if ( peers[i].uri ) + mnstr_printf(cntxt->fdout,"%s\n", peers[i].uri); +#else + (void) cntxt; +#endif + + for (i=0; i<nrpeers; i++) + if ( !peers[i].active ) + OPTcleanFunReg(i); + + 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; +} + _______________________________________________ Checkin-list mailing list [email protected] http://mail.monetdb.org/mailman/listinfo/checkin-list
