Changeset: bfd1ade40019 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=bfd1ade40019
Modified Files:
monetdb5/mal/mal.c
monetdb5/mal/mal.h
monetdb5/optimizer/opt_partition.mx
monetdb5/scheduler/srvpool.c
monetdb5/scheduler/srvpool.h
monetdb5/scheduler/srvpool.mal
Branch: default
Log Message:
Reshuffle nrservers administration.
diffs (truncated from 413 to 300 lines):
diff --git a/monetdb5/mal/mal.c b/monetdb5/mal/mal.c
--- a/monetdb5/mal/mal.c
+++ b/monetdb5/mal/mal.c
@@ -180,6 +180,8 @@ int monet_welcome = 1;
str *monet_script;
int monet_daemon=0;
size_t monet_memory;
+int nrservers = 0;
+
#include "mal_stack.h"
#include "mal_linker.h"
#include "mal_session.h"
diff --git a/monetdb5/mal/mal.h b/monetdb5/mal/mal.h
--- a/monetdb5/mal/mal.h
+++ b/monetdb5/mal/mal.h
@@ -51,6 +51,7 @@ mal_export int monet_welcome
mal_export str *monet_script;
mal_export int monet_daemon;
mal_export size_t monet_memory;
+mal_export int nrservers;
#define mal_set_lock(X,Y) if(GDKprotected) MT_lock_set(&X,Y)
#define mal_unset_lock(X,Y) if(GDKprotected) MT_lock_unset(&X,Y)
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
@@ -23,7 +23,6 @@ All Rights Reserved.
* @a M. Kersten
* @- Partitioning
* Documentation in accompanying paper.
- * Partition optimizer should be followed by the aliases optimizer.
*/
@mal
module partition;
@@ -42,14 +41,6 @@ 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_
@@ -75,8 +66,6 @@ opt_export str OPTpeers(Client cntxt, Ma
#include "mal_recycle.h"
#include "mal_interpreter.h"
-static int nrpeers; /* peers active in sliced
processing */
-
typedef struct{
InstrPtr target;
str schema, table, column;
@@ -127,7 +116,7 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
/* initialize binds and histograms */
q= newFcnCall(cmb, sqlRef, mvcRef);
x= getArg(q,0);
- alias = (int*) GDKzalloc(nrpeers * sizeof(int));
+ alias = (int*) GDKzalloc(nrservers * sizeof(int));
if( slices->column) {
q= newInstruction(cmb, ASSIGNsymbol);
getModuleId(q) = sqlRef;
@@ -146,7 +135,7 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
getModuleId(q) = partitionRef;
getFunctionId(q) = vectorRef;
q = pushArgument(cmb, q, j);
- for ( j = 0; j < nrpeers +1; j++) {
+ for ( j = 0; j < nrservers +1; j++) {
k = alias[j] = newTmpVariable(cmb, slices->type);
q= pushReturn(cmb,q, k);
}
@@ -163,13 +152,15 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
}
}
- /* under dataflow control, initialize the variables */
+ /* under dataflow control, initialize the variables
+ Arguments are considered defined already
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);
}
+ */
/* PHASE 1: dont run in distributed mode */
q= newFcnCall(cmb,schedulerRef,srvpoolRef);
@@ -180,13 +171,13 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
/* get servers to execute the query */
q= newStmt(cmb,srvpoolRef,putName("server",6));
q->retc= q->argc = 0;
- for( i = 0; i < nrpeers; i++)
+ for( i = 0; i < nrservers; 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++) {
+ for ( i = 0; i < nrservers ; i++) {
q= copyInstruction(getInstrPtr(pmb,0));
q->token = ASSIGNsymbol;
q->barrier = 0;
@@ -229,7 +220,7 @@ OPTplanCntrl(Client cntxt, MalBlkPtr mb,
setVarType(cmb,x,TYPE_int);
/* re-use instructions, injecting missing server selections */
- for ( i = 0; i < nrpeers ; i++) {
+ for ( i = 0; i < nrservers ; i++) {
q= copyInstruction(getInstrPtr(cmb,base + i));
getFunctionId(q) = plan;
delArgument(q,q->retc);
@@ -813,9 +804,8 @@ OPTpartitionImplementation(Client cntxt,
msg = GDKgetenv("gdk_readonly");
if( msg == 0 || strcmp(msg,"yes"))
mnstr_printf(cntxt->fdout,"#WARNING centipede only works for
readonly databases\n");
- /* nrpeers = GDKnr_threads ? GDKnr_threads:1;*/
- /* to ease debugging now */
- nrpeers = 2;
+ if ( nrservers == 0)
+ nrservers = 2; /* to ease debugging now */
/* modify the block as we go */
omb= copyMalBlk(mb);
@@ -851,10 +841,10 @@ OPTpartitionImplementation(Client cntxt,
OPTDEBUGpartition
mnstr_printf(cntxt->fdout,"#opt_partition: target is %s.%s "
- " with " SSZFMT " rows into %d peers\n",
+ " with " SSZFMT " rows into %d servers\n",
getVarConstant(mb,
getArg(slices.target,2)).val.sval,
getVarConstant(mb,
getArg(slices.target,3)).val.sval,
- rowcnt, nrpeers);
+ rowcnt, nrservers);
/* derive a local plan based on forward flow reasoning */
if( OPTplanFragment(cntxt, mb, &slices)== 0) {
@@ -917,34 +907,6 @@ OPTpartitionImplementation(Client cntxt,
}
@
@c
-/*
- * The partition optimizer requires the number of peers.
- * This information ideally comes from the centipede scheduler,
- * which maintains the collection of remote servers involved
- * in query processing.
- * A direct call is also provided for testing.
- * A local catalog of partitioned BATs is kept around to further
- * improve remote processing.
-*/
-str OPTpeers(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
-{
- int n;
- int *ret = (int*) getArgReference(stk,pci,0);
-
- (void) cntxt;
- if( getArgType(mb,pci,1) == TYPE_int) {
- n = *(int*) getArgReference(stk,pci,1);
- } else
- if( getArgType(mb,pci,1) == TYPE_str) {
- n = 2;
- }
- if ( n <= 0)
- throw(MAL,"optimizer.peers","Illegal number of peers");
- /* inform the centipede scheduler */
-
- *ret = n;
- return MAL_SUCCEED;
-}
/* the min/max bounds are null values */
/* It is the heart of the approach and requires experimentation */
diff --git a/monetdb5/scheduler/srvpool.c b/monetdb5/scheduler/srvpool.c
--- a/monetdb5/scheduler/srvpool.c
+++ b/monetdb5/scheduler/srvpool.c
@@ -80,14 +80,43 @@ typedef struct {
#define MAXSITES 2048 /* should become dynamic at some point */
static Server servers[MAXSITES]; /* registry of servers */
-static int nrservers=0;
+static int srvtop = 0;
+static int srvbaseline= 0;
+static str srvpattern = NULL;
static int localExecution= FALSE;
+/*
+ * The partition optimizer requires the number of peers.
+ * This information ideally comes from the centipede scheduler,
+ * which maintains the collection of remote servers involved
+ * in query processing.
+ * A direct call is also provided for testing.
+ * A local catalog of partitioned BATs is kept around to further
+ * improve remote processing.
+*/
+str SRVsetServers(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+{
+ (void) cntxt;
+ if( getArgType(mb,pci,1) == TYPE_int) {
+ srvbaseline = *(int*) getArgReference(stk,pci,1);
+ if ( srvbaseline <= 0) {
+ srvbaseline = 0 ;
+ throw(MAL,"scheduler.setServers","Illegal number of
servers");
+ }
+ } else
+ if( getArgType(mb,pci,1) == TYPE_str) {
+ if ( srvpattern)
+ GDKfree(srvpattern);
+ srvpattern = GDKstrdup(*(str*) getArgReference(stk,pci,1));
+ }
+ return MAL_SUCCEED;
+}
+
static int
SRVPOOLfindServer(str dbalias)
{
int i;
- for (i=0; i<nrservers; i++)
+ for (i=0; i<srvtop; i++)
if ( strcmp(dbalias, servers[i].name) == 0 )
return i;
return -1;
@@ -102,7 +131,7 @@ SRVPOOLnewServer(str uri)
int i;
char buf[BUFSIZ];
- i = nrservers;
+ i = srvtop;
/* use default settings */
snprintf(buf,BUFSIZ,"srv_%d",serverid++);
servers[i].name = GDKstrdup(buf);
@@ -111,7 +140,7 @@ SRVPOOLnewServer(str uri)
servers[i].pwd = GDKstrdup("monetdb");
servers[i].active = 1;
servers[i].nxt = NULL;
- nrservers++;
+ srvtop++;
return i;
}
@@ -120,7 +149,7 @@ SRVPOOLgetServer(str uri)
{
int i;
- for (i=0; i<nrservers; i++)
+ for (i=0; i<srvtop; i++)
if ( strcmp(uri, servers[i].name) == 0 || strcmp(uri, servers[i].uri)
== 0 )
return i;
@@ -149,7 +178,7 @@ SRVPOOLdisconnect(void)
int i, ret;
str msg = MAL_SUCCEED;
- for ( i=0; i< nrservers; i++)
+ for ( i=0; i< srvtop; i++)
if ( servers[i].active && servers[i].conn != NULL ) {
msg = RMTdisconnect(&ret,&servers[i].conn);
GDKfree(servers[i].conn);
@@ -165,12 +194,12 @@ str SRVPOOLreset(int *ret)
(void) ret;
msg = SRVPOOLdisconnect();
- for ( i=0; i< nrservers; i++){
+ for ( i=0; i< srvtop; i++){
SRVPOOLcleanup(i);
GDKfree(servers[i].name);
}
- memset((char*)servers, 0, sizeof(Server) * nrservers);
- nrservers = 0;
+ memset((char*)servers, 0, sizeof(Server) * srvtop);
+ srvtop = 0;
return msg;
}
@@ -185,10 +214,10 @@ SRVPOOLconnect(str *c, str *uri)
*c = NULL;
i = SRVPOOLfindServer(*uri);
if ( i < 0 ){
- for ( i =0; i < nrservers; i++)
+ for ( i =0; i < srvtop; i++)
if ( strcmp(*uri, servers[i].uri) == 0)
break;
- if ( i == nrservers)
+ if ( i == srvtop)
return createException(MAL, "srvpool.connect", "Server
%s is not registered", *uri);
}
@@ -206,7 +235,7 @@ SRVPOOLconnect(str *c, str *uri)
/* Look up the servers available for processing , guarantee a minimum number
of servers */
static str
-SRVPOOLdiscover(Client cntxt, str pattern, int minservers)
+SRVPOOLdiscover(Client cntxt)
{
bat bid = 0;
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
