Changeset: 83ecc90cf2d2 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=83ecc90cf2d2
Modified Files:
sql/backends/monet5/iot/50_iot.sql
sql/backends/monet5/iot/Tests/iot00.sql
sql/backends/monet5/iot/Tests/iot05.sql
sql/backends/monet5/iot/Tests/iot10.sql
sql/backends/monet5/iot/basket.c
sql/backends/monet5/iot/basket.h
sql/backends/monet5/iot/basket.mal
sql/backends/monet5/iot/iot.c
sql/backends/monet5/iot/iot.h
sql/backends/monet5/iot/iot.mal
sql/backends/monet5/iot/petrinet.c
sql/backends/monet5/iot/petrinet.h
sql/backends/monet5/iot/petrinet.mal
Branch: iot
Log Message:
Intermittent commit
Still some way to go before Datacell 2.0 works.
diffs (truncated from 563 to 300 lines):
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
@@ -49,8 +49,11 @@ create procedure iot.stop ("schema" stri
create procedure iot.stop ()
external name iot.stop;
+-- deliver a new basket with tuples
+create procedure iot.push("schema" string, "table" string, dirpath string)
+ external name iot.push;
+
-- Inspection tables
-
create function iot.baskets()
returns table( "schema" string, "table" string, threshold int, winsize int,
winstride int, timeslice int, timestride int, beat int, seen timestamp, events
int)
external name iot.baskets;
@@ -59,14 +62,13 @@ create function iot.queries()
returns table( "schema" string, "function" string, status string, lastrun
timestamp, cycles int, events int, time bigint, error string)
external name iot.queries;
--- next causes crash
---create function iot.places2()
--- returns table( "schema" string, "table" string, "schema" string, "query"
string)
--- external name iot.places;
+create function iot.inputplaces()
+ returns table( "s" string, "t" string, "sch" string, "qry" string)
+ external name iot.inputplaces;
-create function iot.places()
+create function iot.outputplaces()
returns table( "s" string, "t" string, "sch" string, "qry" string)
- external name iot.places;
+ external name iot.outputplaces;
-- create function iot.errors()
-- returns table( "schema" string, "table" string, error string)
diff --git a/sql/backends/monet5/iot/Tests/iot00.sql
b/sql/backends/monet5/iot/Tests/iot00.sql
--- a/sql/backends/monet5/iot/Tests/iot00.sql
+++ b/sql/backends/monet5/iot/Tests/iot00.sql
@@ -12,7 +12,7 @@ end;
call iot.query('iot','cq00');
call iot.query('insert into result select min(t), count(*), avg(val) from
stream_tmp;');
-call iot.baskets();
-call iot.petrinet();
-
-call iot.stop();
+select * from iot.baskets();
+select * from iot.queries();
+select * from iot.inputplaces();
+select * from iot.outputplaces();
diff --git a/sql/backends/monet5/iot/Tests/iot05.sql
b/sql/backends/monet5/iot/Tests/iot05.sql
--- a/sql/backends/monet5/iot/Tests/iot05.sql
+++ b/sql/backends/monet5/iot/Tests/iot05.sql
@@ -31,7 +31,5 @@ call iot.query('iot','clk1');
call iot.query('iot','clk3');
call iot.query('select 1;');
-call iot.baskets();
-call iot.scheduler();
-
-call iot.stop();
+select * from iot.baskets();
+select * from iot.queries();
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
@@ -14,6 +14,6 @@ begin
end;
iot.query('iot','collector');
-iot.baskets();
-iot.queries();
-iot.stop();
+
+select * from iot.baskets();
+select * from iot.queries();
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
@@ -33,6 +33,9 @@
#include "mal_builder.h"
#include "opt_prelude.h"
+//#define _DEBUG_BASKET_ if(0)
+#define _DEBUG_BASKET_
+
str statusname[6] = { "<unknown>", "init", "paused", "running", "stop",
"error" };
BasketRec *baskets; /* the global iot catalog */
@@ -230,13 +233,9 @@ str BSKTlock(void *ret, str *sch, str *t
bskt = BSKTlocate(*sch, *tbl);
if (bskt <= 0)
throw(SQL, "basket.lock", "Could not find the basket
%s.%s",*sch,*tbl);
-#ifdef _DEBUG_BASKET
- stream_printf(BSKTout, "lock group %s.%s\n", *sch, *tbl);
-#endif
+ _DEBUG_BASKET_ mnstr_printf(BSKTout, "lock group %s.%s\n", *sch, *tbl);
MT_lock_set(&baskets[bskt].lock);
-#ifdef _DEBUG_BASKET
- stream_printf(BSKTout, "got group locked %s.%s\n", *sch, *tbl);
-#endif
+ _DEBUG_BASKET_ mnstr_printf(BSKTout, "got group locked %s.%s\n", *sch,
*tbl);
(void) delay; /* control spinlock */
(void) ret;
return MAL_SUCCEED;
@@ -289,6 +288,35 @@ BSKTreset(void *ret)
return MAL_SUCCEED;
}
+/* collect the binary files and append them to what we have */
+str
+BSKTpush(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+{
+ str sch = *getArgReference_str(stk, pci, 1);
+ str tbl = *getArgReference_str(stk, pci, 2);
+ str dir = *getArgReference_str(stk, pci, 3);
+ int bskt,i;
+ char buf[BUFSIZ];
+
+ bskt = BSKTlocate(sch,tbl);
+ if (bskt == 0)
+ throw(SQL, "iot.push", "Could not find the basket
%s.%s",sch,tbl);
+
+ // check access permission to directory first
+ if( access (dir , F_OK | R_OK)){
+ throw(SQL, "iot.push", "Could not access the basket directory
%s. error %d",dir,errno);
+ }
+
+ for(i=0; i < baskets[bskt].count ; i++){
+ snprintf(buf,BUFSIZ, "%s%c%s",dir,DIR_SEP,
baskets[bskt].cols[i]);
+ _DEBUG_BASKET_ mnstr_printf(BSKTout,"Attach the file %s\n",buf);
+ }
+ (void) cntxt;
+ (void) mb;
+ (void) stk;
+ (void) pci;
+ return MAL_SUCCEED;
+}
str
BSKTdump(void *ret)
{
@@ -324,7 +352,7 @@ BSKTdump(void *ret)
}
str
-BSKTupdate(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+BSKTappend(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
(void) cntxt;
(void) mb;
@@ -334,31 +362,6 @@ BSKTupdate(Client cntxt, MalBlkPtr mb, M
}
InstrPtr
-BSKTgrabInstruction(MalBlkPtr mb, str sch, str tbl)
-{
- int i, j, bskt;
- InstrPtr p;
- BAT *b;
-
- bskt = BSKTlocate(sch,tbl);
- if (bskt == 0)
- return 0;
- p = newFcnCall(mb, basketRef, grabRef);
- p->argc = 0;
- for (i = 0; i < baskets[bskt].count; i++) {
- b = BBPquickdesc(baskets[bskt].bats[i], FALSE);
- j = newTmpVariable(mb, newBatType(TYPE_oid, b->ttype));
- setVarUDFtype(mb, j);
- setVarFixed(mb, j);
- p = pushArgument(mb, p, j);
- }
- p->retc = p->argc;
- p = pushStr(mb, p, sch);
- p = pushStr(mb, p, tbl);
- return p;
-}
-
-InstrPtr
BSKTupdateInstruction(MalBlkPtr mb, str sch, str tbl)
{
int i, j, bskt;
diff --git a/sql/backends/monet5/iot/basket.h b/sql/backends/monet5/iot/basket.h
--- a/sql/backends/monet5/iot/basket.h
+++ b/sql/backends/monet5/iot/basket.h
@@ -90,9 +90,8 @@ iot_export str BSKTdrop(Client cntxt, Ma
iot_export int BSKTlocate(str sch, str tbl);
iot_export int BSKTunlocate(str sch, str tbl);
iot_export int BSKTlocate(str sch, str tbl);
-iot_export str BSKTupdate(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
-iot_export InstrPtr BSKTgrabInstruction(MalBlkPtr mb, str sch, str tbl);
-iot_export str BSKTgrab(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
+iot_export str BSKTappend(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
+iot_export str BSKTpush(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
iot_export InstrPtr BSKTupdateInstruction(MalBlkPtr mb, str sch, str tbl);
iot_export str BSKTlock(void *ret, str *sch, str *tbl, int *delay);
diff --git a/sql/backends/monet5/iot/basket.mal
b/sql/backends/monet5/iot/basket.mal
--- a/sql/backends/monet5/iot/basket.mal
+++ b/sql/backends/monet5/iot/basket.mal
@@ -37,6 +37,10 @@ command unlock(sch:str,tbl:str):void
address BSKTunlock
comment "Unlock the basket";
+pattern append(mvc:int, sch:str, tbl:str, col:str, val:any):int
+address BSKTappend
+comment "Append new tuples to a basket";
+
pattern update(sch:str,tbl:str, cols:any...)
address BSKTupdate
comment "Dump the new tuples into the basket";
@@ -69,6 +73,10 @@ command reset():void
address BSKTreset
comment "Remove all baskets";
+pattern iot.push(sch:str, tbl:str, dir:str):void
+address BSKTpush
+comment "Push a directory with the binary files";
+
command iot.baskets()(sch:bat[:str],nme:bat[:str], threshold:bat[:int],
winsize:bat[:int], winstride:bat[:int], timeslice:bat[:int],
timestride:bat[:int], beat:bat[:int], seen:bat[:timestamp], events:bat[:int])
address BSKTtable
diff --git a/sql/backends/monet5/iot/iot.c b/sql/backends/monet5/iot/iot.c
--- a/sql/backends/monet5/iot/iot.c
+++ b/sql/backends/monet5/iot/iot.c
@@ -163,16 +163,6 @@ IOTcycles(Client cntxt, MalBlkPtr mb, Ma
{
return PNcycles(cntxt,mb,stk,pci);
}
-str
-IOTgrab(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
-{
- (void) cntxt;
- (void) mb;
- (void) stk;
- (void) pci;
- return MAL_SUCCEED;
-}
-
str
BSKTerror(int *ret, str *sch, str *fcn, str *msg)
diff --git a/sql/backends/monet5/iot/iot.h b/sql/backends/monet5/iot/iot.h
--- a/sql/backends/monet5/iot/iot.h
+++ b/sql/backends/monet5/iot/iot.h
@@ -45,6 +45,6 @@ iot_export str IOTpause(Client cntxt, Ma
iot_export str IOTresume(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
iot_export str IOTstop(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
iot_export str IOTcycles(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
-iot_export str IOTgrab(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
+iot_export str IOTpush(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
iot_export str IOTdump(void *ret);
#endif
diff --git a/sql/backends/monet5/iot/iot.mal b/sql/backends/monet5/iot/iot.mal
--- a/sql/backends/monet5/iot/iot.mal
+++ b/sql/backends/monet5/iot/iot.mal
@@ -53,10 +53,6 @@ pattern step(cnt:int):void
address IOTstop
comment "Make a limited series of transition steps";
-pattern grab(dir:str):void
-address IOTgrab
-comment "Grab a directory with the binary files";
-
command errors()(nme:bat[:str],error:bat[:str])
address BSKTtableerrors
comment "Return a table the erroneous events found during query processing";
diff --git a/sql/backends/monet5/iot/petrinet.c
b/sql/backends/monet5/iot/petrinet.c
--- a/sql/backends/monet5/iot/petrinet.c
+++ b/sql/backends/monet5/iot/petrinet.c
@@ -98,7 +98,7 @@ str PNanalyseWrapper(Client cntxt, MalBl
if (s == NULL)
throw(MAL, "petrinet.analysis", "Could not find function");
- return PNanalysis(cntxt, s->def);
+ return PNanalysis(cntxt, s->def,0);
}
@@ -133,10 +133,11 @@ PNlocate(str modname, str fcnname)
str
PNregisterInternal(Client cntxt, MalBlkPtr mb)
{
- int i, init= pnet == 0;
+ int i, init= pnettop == 0;
InstrPtr sig;
str msg = MAL_SUCCEED;
+ _DEBUG_PETRINET_ mnstr_printf(PNout, "#registerInternal status %d\n",
init);
if (pnettop == MAXPN)
GDKerror("petrinet.register:Too many transitions");
@@ -153,11 +154,12 @@ PNregisterInternal(Client cntxt, MalBlkP
pnet[pnettop].seen = *timestamp_nil;
/* all the rest is zero */
- pnettop++;
- msg = PNanalysis(cntxt, mb);
+ msg = PNanalysis(cntxt, mb, pnettop);
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
