Changeset: 9f0d67e80fb0 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=9f0d67e80fb0
Modified Files:
MonetDB5/src/mal/mal_interpreter.mx
Branch: Jun2010
Log Message:
Dataflow issues solved
The dataflow could not properly handle multiple assigments to the
same variable, but also could not handle assignment to one of
its arguments. In particular, the MT code produced statements like
_170 := algebra.kdifference(_170,_162)
The effect was that some instructions got never executed.
Another issue was to wait properly for instructions to finish
before starting a new dataflow block. This is now handled by
keeping track of todo/done counts.
diffs (143 lines):
diff -r 2037157ff132 -r 9f0d67e80fb0 MonetDB5/src/mal/mal_interpreter.mx
--- a/MonetDB5/src/mal/mal_interpreter.mx Wed Jul 28 14:29:57 2010 +0200
+++ b/MonetDB5/src/mal/mal_interpreter.mx Wed Jul 28 14:42:17 2010 +0200
@@ -209,15 +209,6 @@
return 0;
return 1;
}
-static int
-isOutputVar(InstrPtr p, int a)
-{
- int k;
- for (k=0; k <p->retc; k++)
- if( getArg(p,k)== a)
- return 1;
- return 0;
-}
MalStkPtr
prepareMALstack(MalBlkPtr mb, int size){
@@ -1135,10 +1126,15 @@
int i, n;
char *assign= (char*) GDKzalloc(sizeof(char) * flow->mb->vtop);
+ PARDEBUG
+ printf("Initialize dflow block\n");
for (n=0, i = flow->start; i<flow->stop; i++, n++) {
InstrPtr p = getInstrPtr(flow->mb, i);
int j, a;
+ PARDEBUG
+ printInstruction(GDKstdout, flow->mb, 0, p,
LIST_MAL_STMT);
+
/* initial state, ie everything can run */
fs[n].pc = i;
fs[n].status = DFLOWpending;
@@ -1148,28 +1144,28 @@
for (j=0; j<p->argc; j++){
a = getArg(p, j);
- if ( assign[a])
- continue;
- if (j<p->retc ){
+
+ if (j<p->retc && assign[a] == 0 ){
assign[a] = i;
- if (!flow->inuse[a])
+ if ( isNotUsedIn(p, p->retc,a))
flow->blocked[a] = 1;
}
+
if (j>=p->retc && !isVarConstant(flow->mb, a) )
flow->inuse[a] += isNotUsedIn(p,j+1,a);
}
}
+
for( i=0; i<flow->mb->vtop; i++)
flow->inuse[i] = 0;
-#ifdef DEBUG_FLOW2
+ PARDEBUG
for (i=0; i < flow->mb->vtop; i++)
if ( flow->blocked[i] || flow->inuse[i] || assign[i] ){
printf("%s %d [ %1d %2d ]\n", getVarName(flow->mb,i),
getEndOfLife(flow->mb,i),
flow->blocked[i],
flow->inuse[i]);
-}
-#endif
+ }
GDKfree(assign);
}
@@ -1202,9 +1198,8 @@
{
int j;
for(j=p->retc; j<p->argc; j++) {
- if (!isVarConstant(flow->mb, getArg(p,j)))
- flow->inuse[getArg(p,j)] += isNotUsedIn(p,j+1,getArg(p,j));
- flow->blocked[getArg(p,j)] |= isOutputVar(p,getArg(p,j));
+ if (!isVarConstant(flow->mb, getArg(p,j)))
+ flow->inuse[getArg(p,j)] +=
isNotUsedIn(p,j+1,getArg(p,j));
}
/* the target variables become blocked until this instruction is
finished */
@@ -1225,6 +1220,7 @@
DFLOWscheduler( DataFlow flow )
{
int queued = 0, candidates;
+ int todo = 0, done = 0;
int pc = 0, i, j, oa =0, ia;
int limit = flow->stop - flow->start;
str ret = MAL_SUCCEED;
@@ -1242,19 +1238,21 @@
if( limit)
goto firststep; /* confuse static code analysers, i want a jump
*/
- while(queued){
- PARDEBUG stream_printf(GDKstdout,"#waiting for results, queued
%d\n", queued);
+ while(queued || todo != done){
+ PARDEBUG stream_printf(GDKstdout,"#waiting for results, queued
%d todo %d done %d\n", queued,todo,done);
f = q_dequeue(flow->done);
f->status = DFLOWwrapup;
+ done++;
queued--;
if (f->pc < 0) {
- PARDEBUG {
+ PARDEBUG
stream_printf(GDKstdout,"#errors encountered %s
", f->error);
- }
/* we have to wait for all threads to report back */
/* dequeue the remainders in case of an error */
- while(queued-- > 0)
+ while(todo != done) {
(void)q_dequeue(flow->done);
+ done++;
+ }
return f->error;
} else {
p = getInstrPtr(flow->mb, f->pc);
@@ -1297,6 +1295,7 @@
for ( j= p->retc; j < p->argc; j++)
if ( getArg(p,j)== oa && DFLOWeligible(flow,fs,i,p,pc)){
queued++;
+ todo++;
candidates ++;
DFLOWactivate(flow,fs,i,p);
q_enqueue(flow->todo, fs+i);
@@ -1309,6 +1308,7 @@
p = getInstrPtr(flow->mb, fs[i].pc);
if ( DFLOWeligible(flow,fs,i,p,pc)){
queued++;
+ todo++;
DFLOWactivate(flow,fs,i,p);
q_enqueue(flow->todo, fs+i);
}
@@ -1316,7 +1316,7 @@
}
PARDEBUG {
int candidates = 0;
- stream_printf(GDKstdout,"#end of data flow %d %d\n",pc,limit);
+ stream_printf(GDKstdout,"#end of data flow %d %d todo %d done
%d\n",pc,limit,todo,done);
for ( i =0 ; i<limit; i++)
if (fs[i].status != DFLOWwrapup && fs[i].pc >=0) {
stream_printf(GDKstdout,"#missed %d %d %d ",
i, fs[i].status, fs[i].pc);
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
