dataflow00.mal

一个内存数据库的源代码这是服务器端还有客户端

# baseline test for dataflow processing.
# cardinalities

function makeBat(limit:int)(:bat[:int,:int],:bat[:int,:int]);
	b:=bat.new(:int,:int);
barrier (go,i):= language.newRange(0:int);
	j:= mmath.rand();
	j:= j % limit;
	bat.insert(b,i,j);
redo (go,i):= language.nextElement(1:int,limit);
exit go;
	br:= bat.reverse(b);
	return (b,br);
end makeBat;

(b10,b10r):= makeBat(10);
io.print(b10);
(b10k,b10kr):= makeBat(10240);


#for each possible join plan calculate the total cost
function llinear(a:bat[:int,:int], b:bat[:int,:int],
		c:bat[:int,:int], d:bat[:int,:int]) (:lng,:int);
	t0:= alarm.usec();
	l := bat.new(:int,:int);
barrier x:= language.dataflow();
	j:= algebra.join(a,b);
	k:= algebra.join(j,c);
	l:= algebra.join(k,d);
exit x;
	t1:= alarm.usec();
	t1:= t1-t0;
	cnt:= aggr.count(l);
	return (t1,cnt);
end llinear;

function rlinear(a:bat[:int,:int], b:bat[:int,:int],
		c:bat[:int,:int], d:bat[:int,:int]) (:lng,:int);
	t0:= alarm.usec();
	l := bat.new(:int,:int);
barrier x:= language.dataflow();
	j:= algebra.join(c,d);
	k:= algebra.join(b,j);
	l:= algebra.join(a,k);
exit x;
	t1:= alarm.usec();
	t1:= t1-t0;
	cnt:= aggr.count(l);
	return (t1,cnt);
end rlinear;

function bushy(a:bat[:int,:int], b:bat[:int,:int],
		c:bat[:int,:int], d:bat[:int,:int]) (:lng,:int);
	t0:= alarm.usec();
	l := bat.new(:int,:int);
barrier x:= language.dataflow();
	j:= algebra.join(a,b);
	k:= algebra.join(c,d);
	l:= algebra.join(j,k);
	cnt:= aggr.count(l);
exit x;
	t1:= alarm.usec();
	t1:= t1-t0;
	cnt:= aggr.count(l);
	return (t1,cnt);
end bushy;


# mdb.setTimer(true);
#Lineare foreign key sequences
io.printf("b10 - b10 - b10 - b10\n");
(clk,sze):= llinear(b10, b10, b10, b10); io.print(clk);
#io.printf("llinear size %d\n",sze);
(clk,sze):= rlinear(b10, b10, b10, b10); io.print(clk);
#io.printf("rlinear size %d\n",sze);
(clk,sze):= bushy(b10, b10, b10, b10); io.print(clk);
#io.printf("bushy size %d\n",sze);

#....
io.printf("b10 - b10 - b10k - b10k\n");
(clk,sze):= llinear(b10, b10, b10k, b10k); io.print(clk);
#io.printf("llinear size %d\n",sze);
(clk,sze):= rlinear(b10, b10, b10k, b10k); io.print(clk);
#io.printf("rlinear size %d\n",sze);
(clk,sze):= bushy(b10, b10, b10k, b10k); io.print(clk);
#io.printf("bushy size %d\n",sze);

##....
#io.printf("b10k - b10 - b10 - b10k\n");
#(clk,sze):= llinear(b10k, b10, b10, b10k); io.print(clk);
##io.printf("llinear size %d\n",sze);
#(clk,sze):= rlinear(b10k, b10, b10, b10k); io.print(clk);
##io.printf("rlinear size %d\n",sze);
#(clk,sze):= bushy(b10k, b10, b10, b10k); io.print(clk);
##io.printf("bushy size %d\n",sze);
#
##....
#io.printf("b10k - b10k - b10 - b10k\n");
#(clk,sze):= llinear(b10k, b10k, b10, b10k); io.print(clk);
##io.printf("llinear size %d\n",sze);
#(clk,sze):= rlinear(b10k, b10k, b10, b10k); io.print(clk);
##io.printf("rlinear size %d\n",sze);
#(clk,sze):= bushy(b10k, b10k, b10, b10k); io.print(clk);
##io.printf("bushy size %d\n",sze);
#
##....
#io.printf("b10k - b10k - b10k - b10k\n");
#(clk,sze):= llinear(b10k, b10k, b10k, b10k); io.print(clk);
##io.printf("llinear size %d\n",sze);
#(clk,sze):= rlinear(b10k, b10k, b10k, b10k); io.print(clk);
##io.printf("rlinear size %d\n",sze);
#(clk,sze):= bushy(b10k, b10k, b10k, b10k); io.print(clk);
##io.printf("bushy size %d\n",sze);
#
##....
#io.printf("b10k - b10kr - b10k - b10kr\n");
#(clk,sze):= llinear(b10k, b10kr, b10k, b10kr); io.print(clk);
##io.printf("llinear size %d\n",sze);
#(clk,sze):= rlinear(b10k, b10kr, b10k, b10kr); io.print(clk);
##io.printf("rlinear size %d\n",sze);
#(clk,sze):= bushy(b10k, b10kr, b10k, b10kr); io.print(clk);
##io.printf("bushy size %d\n",sze);