Changeset: c11f5e34d57b for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=c11f5e34d57b
Modified Files:
monetdb5/optimizer/opt_iot.c
sql/backends/monet5/iot/50_iot.sql
sql/backends/monet5/iot/Tests/All
sql/backends/monet5/iot/Tests/bug01.stable.out
sql/backends/monet5/iot/Tests/iot02.sql
sql/backends/monet5/iot/Tests/iot10.sql
sql/backends/monet5/iot/Tests/iot10.stable.out
sql/backends/monet5/iot/Tests/iot12.sql
sql/backends/monet5/iot/Tests/iot13.sql
sql/backends/monet5/iot/Tests/iot13.stable.out
sql/backends/monet5/iot/Tests/webtest.stable.out
sql/backends/monet5/iot/basket.c
sql/backends/monet5/iot/basket.h
sql/backends/monet5/iot/iot.mal
sql/backends/monet5/iot/petrinet.c
Branch: iot
Log Message:
Various improvements
- unlock before you start another transaction
- added keep()/releas() routines to avoid tumbling
- removed basket status
- fixed control in pnscheduler
diffs (truncated from 600 to 300 lines):
diff --git a/monetdb5/optimizer/opt_iot.c b/monetdb5/optimizer/opt_iot.c
--- a/monetdb5/optimizer/opt_iot.c
+++ b/monetdb5/optimizer/opt_iot.c
@@ -52,6 +52,7 @@ OPTiotImplementation(Client cntxt, MalBl
int output[MAXBSKT]= {0};
int btop=0, lastmvc=0;
int noerror=0;
+ int mvcseen = 0;
int cq= strncmp(getFunctionId(getInstrPtr(mb,0)),"cq",2) == 0;
char buf[256];
lng usec = GDKusec();
@@ -167,6 +168,17 @@ OPTiotImplementation(Client cntxt, MalBl
if(getModuleId(p) == sqlRef && getFunctionId(p)==
mvcRef){
pushInstruction(mb,p);
lastmvc = getArg(p,0);
+ // watch out for second transaction in same
block
+ if( mvcseen){
+ // unlock the tables
+ for( j=btop-1; j>= 0; j--){
+ r=
newStmt(mb,basketRef,unlockRef);
+ r= pushArgument(mb,r,lastmvc);
+ r= pushStr(mb,r, schemas[j]);
+ r= pushStr(mb,r, tables[j]);
+ lastmvc= getArg(r,0);
+ }
+ }
// register and lock all baskets used
for( j=0; j<btop; j++){
p= newStmt(mb,basketRef,registerRef);
@@ -183,6 +195,7 @@ OPTiotImplementation(Client cntxt, MalBl
p= pushStr(mb,p, tables[j]);
lastmvc = getArg(p,0);
}
+ mvcseen=1;
continue;
}
// register all baskets used after the mvc had been
determined
@@ -262,7 +275,6 @@ OPTiotImplementation(Client cntxt, MalBl
r= pushStr(mb,r, schemas[j]);
r= pushStr(mb,r, tables[j]);
lastmvc= getArg(r,0);
-
}
//p= newStmt(mb,basketRef,commitRef);
//p= pushArgument(mb,p, lastmvc);
diff --git a/sql/backends/monet5/iot/50_iot.sql
b/sql/backends/monet5/iot/50_iot.sql
--- a/sql/backends/monet5/iot/50_iot.sql
+++ b/sql/backends/monet5/iot/50_iot.sql
@@ -39,6 +39,12 @@ create procedure iot.resume()
create procedure iot.pause("schema" string, name string)
external name iot.pause;
+create procedure iot.keep("schema" string, name string)
+ external name iot.keep;
+
+create procedure iot.release("schema" string, name string)
+ external name iot.release;
+
create procedure iot.pause()
external name iot.pause;
@@ -99,7 +105,7 @@ create function iot.getheartbeat("schema
external name iot.getheartbeat;
create function iot.baskets()
-returns table( "schema" string, "table" string, "status" string, winsize int,
winstride int, timeslice int, timestride int, heartbeat int, seen timestamp,
events int)
+returns table( "schema" string, "table" string, "status" string, winsize int,
winstride int, timeslice int, timestride int, heartbeat int, seen timestamp,
"count" bigint, events bigint)
external name iot.baskets;
create procedure iot.show("schema" string, "query" string)
diff --git a/sql/backends/monet5/iot/Tests/All
b/sql/backends/monet5/iot/Tests/All
--- a/sql/backends/monet5/iot/Tests/All
+++ b/sql/backends/monet5/iot/Tests/All
@@ -4,6 +4,7 @@ bug03
bug04
bug05
iot00
+iot01
iot02
iot03
iot04
diff --git a/sql/backends/monet5/iot/Tests/bug01.stable.out
b/sql/backends/monet5/iot/Tests/bug01.stable.out
--- a/sql/backends/monet5/iot/Tests/bug01.stable.out
+++ b/sql/backends/monet5/iot/Tests/bug01.stable.out
@@ -68,10 +68,10 @@ Ready.
# 15:59:58 >
#select * from iot.baskets();
-% ., ., ., ., ., ., ., ., ., . #
table_name
-% schema, table, status, winsize, winstride, timeslice,
timestride, heartbeat, seen, events # name
-% clob, clob, clob, int, int, int, int, int,
timestamp, int # type
-% 0, 0, 0, 1, 1, 1, 1, 1, 26, 1 #
length
+% ., ., ., ., ., ., ., ., ., .,
. # table_name
+% schema, table, status, winsize, winstride, timeslice,
timestride, heartbeat, seen, count, events # name
+% clob, clob, clob, int, int, int, int, int,
timestamp, bigint, bigint # type
+% 0, 0, 0, 1, 1, 1, 1, 1, 26, 1,
1 # length
# 15:59:58 >
# 15:59:58 > "Done."
diff --git a/sql/backends/monet5/iot/Tests/iot02.sql
b/sql/backends/monet5/iot/Tests/iot02.sql
--- a/sql/backends/monet5/iot/Tests/iot02.sql
+++ b/sql/backends/monet5/iot/Tests/iot02.sql
@@ -23,9 +23,9 @@ insert into stmp2 values('2005-09-23 12:
call iot.resume('iot','cq02');
--- wait for 5 seconds for handler
+-- wait for a few seconds for handler
-call iot.wait(5000);
+call iot.wait(2000);
call iot.pause();
select 'RESULT';
select * from stmp2;
diff --git a/sql/backends/monet5/iot/Tests/iot10.sql
b/sql/backends/monet5/iot/Tests/iot10.sql
--- a/sql/backends/monet5/iot/Tests/iot10.sql
+++ b/sql/backends/monet5/iot/Tests/iot10.sql
@@ -18,17 +18,20 @@ insert into stmp10 values('2005-09-23 12
insert into stmp10 values('2005-09-23 12:34:27.000',1,11.0);
insert into stmp10 values('2005-09-23 12:34:28.000',1,13.0);
insert into stmp10 values('2005-09-23 12:34:28.000',1,15.0);
+select "count" from iot.baskets();
+
+-- avoid emptying the stmp10
+call iot.keep('iot','stmp10');
select * from stmp10;
+call iot.release('iot','stmp10');
-call iot.window('iot','stmp10',2);
call iot.query('iot','collector');
--select * from iot.baskets();
--select * from iot.queries();
-call iot.resume();
-- wait a few seconds
-call iot.wait(5000);
+call iot.wait(2000);
select * from tmp_aggregate;
diff --git a/sql/backends/monet5/iot/Tests/iot10.stable.out
b/sql/backends/monet5/iot/Tests/iot10.stable.out
--- a/sql/backends/monet5/iot/Tests/iot10.stable.out
+++ b/sql/backends/monet5/iot/Tests/iot10.stable.out
@@ -46,6 +46,12 @@ Ready.
[ 1 ]
#insert into stmp10 values('2005-09-23 12:34:28.000',1,15.0);
[ 1 ]
+#select "count" from iot.baskets();
+% .L # table_name
+% count # name
+% bigint # type
+% 1 # length
+[ 4 ]
#select * from stmp10;
% iot.stmp10, iot.stmp10, iot.stmp10 # table_name
% t, sensor, val # name
@@ -53,6 +59,8 @@ Ready.
% 26, 1, 10 # length
[ 2005-09-23 12:34:26.000000, 1, 9.00 ]
[ 2005-09-23 12:34:27.000000, 1, 11.00 ]
+[ 2005-09-23 12:34:28.000000, 1, 13.00 ]
+[ 2005-09-23 12:34:28.000000, 1, 15.00 ]
#select * from tmp_aggregate;
% iot.tmp_aggregate, iot.tmp_aggregate # table_name
% tmp_total, tmp_count # name
diff --git a/sql/backends/monet5/iot/Tests/iot12.sql
b/sql/backends/monet5/iot/Tests/iot12.sql
--- a/sql/backends/monet5/iot/Tests/iot12.sql
+++ b/sql/backends/monet5/iot/Tests/iot12.sql
@@ -9,6 +9,8 @@ insert into clocks values(0,0);
create table clocklog( t timestamp, clk integer);
call iot.heartbeat('iot','clocks',2000);
+-- avoid loosing the tuples in the stream table
+call iot.window('iot','clocks',-1);
create procedure clk()
begin
diff --git a/sql/backends/monet5/iot/Tests/iot13.sql
b/sql/backends/monet5/iot/Tests/iot13.sql
--- a/sql/backends/monet5/iot/Tests/iot13.sql
+++ b/sql/backends/monet5/iot/Tests/iot13.sql
@@ -13,7 +13,12 @@ call iot.heartbeat('iot','tmp13',1000);
create procedure cq13()
begin
- insert into agenda13 select count(*), 'full batch' from tmp13;
+ declare b boolean;
+ set b = (select count(*) > 0 from tmp13);
+ if (b)
+ then
+ insert into agenda13 select count(*), 'full batch' from tmp13;
+ end if;
end;
call iot.query('iot','cq13');
@@ -23,7 +28,6 @@ insert into tmp13 values('2005-09-23 12:
insert into tmp13 values('2005-09-23 12:34:26.736',1,12.35);
insert into tmp13 values('2005-09-23 12:34:26.736',1,12.36);
--insert into tmp13 values('2005-09-23 12:34:26.736',1,12.37);
-select * from tmp13;
-- reactivate all continuous queries
call iot.resume();
diff --git a/sql/backends/monet5/iot/Tests/iot13.stable.out
b/sql/backends/monet5/iot/Tests/iot13.stable.out
--- a/sql/backends/monet5/iot/Tests/iot13.stable.out
+++ b/sql/backends/monet5/iot/Tests/iot13.stable.out
@@ -45,13 +45,6 @@ Ready.
[ 1 ]
#insert into tmp13 values('2005-09-23 12:34:26.736',1,12.36);
[ 1 ]
-#select * from tmp13;
-% iot.tmp13, iot.tmp13, iot.tmp13 # table_name
-% t, sensor, val # name
-% timestamp, int, decimal # type
-% 26, 1, 10 # length
-[ 2005-09-23 12:34:26.736000, 1, 12.34 ]
-[ 2005-09-23 12:34:26.736000, 1, 12.35 ]
#select 'RESULT';
% .L # table_name
% single_value # name
diff --git a/sql/backends/monet5/iot/Tests/webtest.stable.out
b/sql/backends/monet5/iot/Tests/webtest.stable.out
--- a/sql/backends/monet5/iot/Tests/webtest.stable.out
+++ b/sql/backends/monet5/iot/Tests/webtest.stable.out
@@ -62,12 +62,13 @@ end user.petri_0;
unsafe function user.iot_web00():void;
X_0 := sql.mvc();
X_28 := basket.register(X_0,"iot","ntemps",0);
-barrier X_56 := language.dataflow();
+ X_32 := basket.lock(X_28,"iot","ntemps");
+barrier X_57 := language.dataflow();
C_1:bat[:oid] := basket.tid(X_0,"iot","ntemps");
X_4:bat[:timestamp] := basket.bind(X_28,"iot","ntemps","iotclk");
X_8 := aggr.min(X_4);
-exit X_56;
- X_10 := sql.append(X_28,"iot","atemps","iotclk",X_8);
+exit X_57;
+ X_10 := sql.append(X_32,"iot","atemps","iotclk",X_8);
X_12 := aggr.count(X_4);
X_13 := calc.int(X_12);
X_15 := sql.append(X_10,"iot","atemps","cnt",X_13);
@@ -76,14 +77,14 @@ exit X_56;
X_21:dbl := aggr.avg(X_20);
X_22 := calc.flt(X_21);
X_24 := sql.append(X_15,"iot","atemps","temperature",X_22);
- X_32 := basket.tumble(X_24,"iot","ntemps");
- basket.commit(X_32,"iot","ntemps");
+ X_33 := basket.tumble(X_24,"iot","ntemps");
catch SQLexception:str;
iot.error("user","web00",SQLexception);
exit SQLexception:str;
catch MALexception:str;
iot.error("user","web00",MALexception);
exit MALexception:str;
+ basket.unlock(X_33,"iot","ntemps");
end user.iot_web00;
#inline actions= 0 time=0 usec
#candidates actions= 1 time=1 usec
diff --git a/sql/backends/monet5/iot/basket.c b/sql/backends/monet5/iot/basket.c
--- a/sql/backends/monet5/iot/basket.c
+++ b/sql/backends/monet5/iot/basket.c
@@ -139,7 +139,6 @@ BSKTnewbasket(mvc *m, sql_schema *s, sql
baskets[idx].table_name = GDKstrdup(t->base.name);
(void) MTIMEcurrent_timestamp(&baskets[idx].seen);
- baskets[idx].status = BSKTWAIT;
baskets[idx].count = 0;
baskets[idx].winstride = -1; /* all tuples are removed */
@@ -294,14 +293,13 @@ BSKTwindow(Client cntxt, MalBlkPtr mb, M
{
str sch = *getArgReference_str(stk,pci,1);
str tbl = *getArgReference_str(stk,pci,2);
- int elm = *getArgReference_int(stk,pci,3);
+ int winsize = *getArgReference_int(stk,pci,3);
+ int winstride;
int idx;
str msg;
(void) cntxt;
(void) mb;
- if( elm <= 0)
- throw(SQL,"basket.window","Positive slice expected\n");
idx = BSKTlocate(sch, tbl);
if( idx == 0){
msg= BSKTregisterInternal(cntxt, mb, sch, tbl);
@@ -311,8 +309,57 @@ BSKTwindow(Client cntxt, MalBlkPtr mb, M
if( idx ==0)
throw(SQL,"basket.window","Stream table %s.%s not
accessible\n",sch,tbl);
}
- baskets[idx].winsize = elm;
- baskets[idx].winstride = elm;
+ if( pci->argc == 5)
+ winstride = *getArgReference_int(stk,pci,4);
+ else winstride = winsize;
+ baskets[idx].winsize = winsize;
+ baskets[idx].winstride = winstride;
+ return MAL_SUCCEED;
+}
+
+str
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
