Changeset: 8bbfb5f0159d for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=8bbfb5f0159d
Modified Files:
clients/Tests/exports.stable.out
clients/Tests/malcheck.stable.out
sql/backends/monet5/iot/50_iot.sql
sql/backends/monet5/iot/basket.c
sql/backends/monet5/iot/basket.h
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/sql_scenario.c
sql/test/BugTracker-2016/Tests/stream_table_crash.Bug-3952.stable.out
Branch: iot
Log Message:
Various improvements
-API change iot.basket ->iot.import
-cleanup of all baskets
- property derivation for attached bats
diffs (truncated from 416 to 300 lines):
diff --git a/clients/Tests/exports.stable.out b/clients/Tests/exports.stable.out
--- a/clients/Tests/exports.stable.out
+++ b/clients/Tests/exports.stable.out
@@ -2547,6 +2547,7 @@ str transactionRef;
void trimMalBlk(MalBlkPtr mb);
void trimMalVariables(MalBlkPtr mb, MalStkPtr stk);
void trimMalVariables_(MalBlkPtr mb, bit *used, MalStkPtr glb);
+str tumbleRef;
void typeChecker(stream *out, Module scope, MalBlkPtr mb, InstrPtr p, int
silent);
int tzone_fromstr(const char *buf, int *len, tzone **d);
tzone tzone_local;
@@ -2560,6 +2561,7 @@ str updateRef;
void updateScenario(str scen, str nme, MALfcn fcn);
str userRef;
str vectorRef;
+str windowRef;
str zero_or_oneRef;
# stream
diff --git a/clients/Tests/malcheck.stable.out
b/clients/Tests/malcheck.stable.out
--- a/clients/Tests/malcheck.stable.out
+++ b/clients/Tests/malcheck.stable.out
@@ -6,12 +6,8 @@ stdout of test 'malcheck` in directory '
# 15:15:51 >
BSKTdrop: missing for MAL command drop in sql/backends/monet5/iot/basket.mal
-BSKTthreshold: missing for MAL command iot.threshold in
sql/backends/monet5/iot/basket.mal
-BSKTbeat: missing for MAL command iot.beat in
sql/backends/monet5/iot/basket.mal
-BSKTwindow: missing for MAL command iot.window in
sql/backends/monet5/iot/basket.mal
-BSKTwindow: missing for MAL command iot.window in
sql/backends/monet5/iot/basket.mal
BSKTerror: missing for MAL command error in sql/backends/monet5/iot/iot.mal
-IOTstop: missing for MAL pattern step in sql/backends/monet5/iot/iot.mal
+BSKTimportBasket: missing for MAL pattern iot.import in
sql/backends/monet5/iot/iot.mal
PNstep: missing for MAL pattern step in sql/backends/monet5/iot/petrinet.mal
PNsetCycleDelay: missing for MAL pattern period in
sql/backends/monet5/iot/petrinet.mal
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
@@ -57,8 +57,8 @@ create procedure iot.period(n integer)
external name iot.period;
-- deliver a new basket with tuples
-create procedure iot.basket("schema" string, "table" string, dirpath string)
- external name iot.basket;
+create procedure iot.import("schema" string, "table" string, dirpath string)
+ external name iot.import;
-- input/output places
create procedure iot.receptor("schema" string, "table" string, dir string)
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
@@ -64,18 +64,31 @@ static int BSKTnewEntry(void)
// free all malloced space
-static void
+void
BSKTclean(int idx)
{
- GDKfree(baskets[idx].schema_name);
- GDKfree(baskets[idx].table_name);
- baskets[idx].schema_name = NULL;
- baskets[idx].table_name = NULL;
+ if( idx){
+ GDKfree(baskets[idx].schema_name);
+ GDKfree(baskets[idx].table_name);
+ baskets[idx].schema_name = NULL;
+ baskets[idx].table_name = NULL;
- BBPreclaim(baskets[idx].errors);
- baskets[idx].winstride = -1;
- baskets[idx].errors = NULL;
- baskets[idx].count = 0;
+ BBPreclaim(baskets[idx].errors);
+ baskets[idx].winstride = -1;
+ baskets[idx].errors = NULL;
+ baskets[idx].count = 0;
+ }
+ for(idx = 1; idx < bsktTop; idx++){
+ GDKfree(baskets[idx].schema_name);
+ GDKfree(baskets[idx].table_name);
+ baskets[idx].schema_name = NULL;
+ baskets[idx].table_name = NULL;
+
+ BBPreclaim(baskets[idx].errors);
+ baskets[idx].winstride = -1;
+ baskets[idx].errors = NULL;
+ baskets[idx].count = 0;
+ }
}
// locate the basket in the catalog
@@ -181,12 +194,17 @@ str
BSKTregister(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
str sch, tbl;
+ str msg= MAL_SUCCEED;
(void) stk;
(void) pci;
sch = getVarConstant(mb, getArg(pci,2)).val.sval;
tbl = getVarConstant(mb, getArg(pci,3)).val.sval;
- return BSKTregisterInternal(cntxt,mb,sch,tbl);
+ msg = BSKTregisterInternal(cntxt,mb,sch,tbl);
+ // also lock the basket
+ if( msg == MAL_SUCCEED){
+ }
+ return msg;
}
str
@@ -233,6 +251,7 @@ BSKTwindow(Client cntxt, MalBlkPtr mb, M
throw(SQL,"basket.window","Stream table %s.%s not
accessible to deactivate\n",sch,tbl);
}
baskets[idx].winsize = elm;
+ baskets[idx].winstride = elm;
return MAL_SUCCEED;
}
@@ -407,15 +426,19 @@ BSKTimportInternal(Client cntxt, int bsk
}
}
BATsetcount(b, bcnt );
+ break;
+ default:
+ msg= createException(MAL,"iot.basket","Import type not
yet supported\n");
}
(void) fclose(f);
}
- /* check for mis-aligned columns */
+ /* check for mis-aligned columns and derive properties */
for( n = baskets[bskt].table->columns.set->h; msg == MAL_SUCCEED && n;
n= n->next){
sql_column *c = n->data;
b = store_funcs.bind_col(m->session->tr,c,RD_INS);
assert( b );
+ BATderiveProps(b, FALSE);
if( first){
first = 0;
cnt = BATcount(b);
@@ -431,8 +454,8 @@ BSKTimportInternal(Client cntxt, int bsk
assert( access (buf,R_OK) == 0);
//unlink(buf);
}
- baskets[bskt].status = BSKTWAIT;
baskets[bskt].count = cnt;
+ baskets[bskt].status = BSKTFILLED;
recover:
/* reset all BATs when they are misaligned or error occurred */
@@ -545,7 +568,7 @@ BSKTtumble(Client cntxt, MalBlkPtr mb, M
{
str sch;
str tbl;
- BUN elm = oid_nil;
+ int elm = -1;
int idx;
(void) cntxt;
@@ -560,8 +583,9 @@ BSKTtumble(Client cntxt, MalBlkPtr mb, M
idx = BSKTlocate(sch, tbl);
if( idx ==0)
throw(SQL,"basket.tumble","Stream table %s.%s not
accessible \n",sch,tbl);
- elm =(int) baskets[idx].winstride;
}
+ /* also take care of time-based tumbling */
+ elm =(int) baskets[idx].winstride;
return BSKTtumbleInternal(cntxt, sch, tbl, elm);
}
@@ -594,6 +618,7 @@ BSKTcommit(Client cntxt, MalBlkPtr mb, M
idx = BSKTlocate(sch, tbl);
if( idx ==0)
throw(SQL,"basket.commit","Stream table %s.%s not accessible to
empty\n",sch,tbl);
+ /* release the basket lock */
return MAL_SUCCEED;
}
@@ -657,6 +682,7 @@ BSKTappend(Client cntxt, MalBlkPtr mb, M
sql_column *c;
BAT *bn=0, *binsert = 0;
int bskt;
+ BUN cnt =0;
*res = 0;
if ((msg = getSQLContext(cntxt, mb, &m, NULL)) != NULL)
@@ -692,10 +718,15 @@ BSKTappend(Client cntxt, MalBlkPtr mb, M
BATappend(bn, binsert, TRUE);
else
BUNappend(bn, value, TRUE);
+ cnt = BATcount(bn);
+ BATderiveProps(bn, FALSE);
BBPunfix(bn->batCacheid);
- baskets[bskt].count = BATcount(bn);
+ } else throw(SQL, "basket.append", "Cannot access target descriptor");
+
+ if(cnt){
+ baskets[bskt].count = cnt;
baskets[bskt].status = BSKTFILLED;
- } else throw(SQL, "basket.append", "Cannot access target descriptor");
+ }
if (binsert )
BBPunfix(((BAT *) binsert)->batCacheid);
return MAL_SUCCEED;
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
@@ -36,6 +36,7 @@ typedef struct{
sql_schema *schema;
sql_table *table;
str *cols;
+ BAT *bats[MAXBSKT]; /* the bats comprising the basket */
int threshold ; /* bound to determine scheduling eligibility */
BUN winsize, winstride; /* sliding window operations */
@@ -94,5 +95,6 @@ iot_export str BSKTimport(Client cntxt,
iot_export str BSKTerror(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
iot_export str BSKTlock(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
iot_export str BSKTunlock(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci);
+iot_export void BSKTclean(int idx);
#endif
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
@@ -31,19 +31,19 @@
MT_Lock iotLock MT_LOCK_INITIALIZER("iotLock");
#define IOTout mal_clients[1].fdout
-
// locate the SQL procedure in the catalog
static str
IOTprocedureStmt(Client cntxt, MalBlkPtr mb, str schema, str nme)
{
mvc *m = NULL;
- str msg = getSQLContext(cntxt, mb, &m, NULL);
+ str msg = MAL_SUCCEED;
sql_schema *s;
backend *be;
node *o;
sql_func *f;
/*sql_trans *tr;*/
+ msg = getSQLContext(cntxt, mb, &m, NULL);
if ((msg = checkSQLContext(cntxt)) != MAL_SUCCEED)
return msg;
s = mvc_bind_schema(m, schema);
@@ -96,6 +96,9 @@ IOTquery(Client cntxt, MalBlkPtr mb, Mal
s = findSymbolInModule(cntxt->nspace, putName(nme));
if (s == NULL)
throw(SQL, "iot.query", "Definition missing");
+ //snprintf(name,
IDLENGTH,"cq_%s",getFunctionId(getInstrPtr(s->def,0)));
+ /* re-package the MAL block for the petrinet */
+ //nme = name;
qry = s->def;
} else if (pci->argc == 2){
// pre-create the new procedure
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
@@ -22,6 +22,7 @@
#include "mal.h"
#include "mal_interpreter.h"
+#include "mal_client.h"
#include "sql.h"
#include "sql_scenario.h"
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
@@ -74,7 +74,7 @@ address BSKTwindow
comment "Use a window of precisely N events";
pattern iot.import(sch:str, tbl:str, dir:str):void
-address BSKTimportBasket
+address BSKTimport
comment "Import a single directory with the binary files for a stream table";
pattern iot.baskets()(sch:bat[:str],nme:bat[:str], status:bat[:str],
threshold:bat[:int], winsize:bat[:int], winstride:bat[:int],
timeslice:bat[:int],
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
@@ -235,11 +235,18 @@ PNwait(Client cntxt, MalBlkPtr mb, MalSt
/* safely stop the engine by stopping all CQ firt */
str
PNstop(void){
- int i=0;
+ int i,cnt;
_DEBUG_PETRINET_ mnstr_printf(PNout, "#scheduler being stopped\n");
+
pnstatus = PNSTOP;
- while( i++ < 10000 && pnstatus != PNINIT)
+ do{
MT_sleep_ms(20);
+ for(cnt=0, i = 0; i < pnettop; i++){
+ cnt += pnet[i].status == PNRUNNING;
+ }
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
