Changeset: 559f57c2691f for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=559f57c2691f
Modified Files:
gdk/gdk_system.c
gdk/gdk_system.h
sql/backends/monet5/cquery.mal
sql/backends/monet5/sql_cquery.c
sql/backends/monet5/sql_cquery.h
sql/backends/monet5/sql_timestamps.c
sql/backends/monet5/sql_timestamps.h
Branch: trails
Log Message:
Delay of the continuous query beginning working
diffs (truncated from 386 to 300 lines):
diff --git a/gdk/gdk_system.c b/gdk/gdk_system.c
--- a/gdk/gdk_system.c
+++ b/gdk/gdk_system.c
@@ -947,3 +947,55 @@ GDKms(void)
{
return (int) (GDKusec() / 1000);
}
+
+lng
+GDKunix_timestamp_usec(void)
+{
+ /* Return the time in microseconds since an epoch. */
+#ifdef _MSC_VER
+ {
+ FILETIME ft;
+ lng tt;
+
+ GetSystemTimeAsFileTime(&ft);
+ tt = ft.dwHighDateTime;
+ tt <<= 32;
+ tt |= ft.dwLowDateTime;
+ tt /= 10;
+ tt -= 11644473600000000L;
+
+ return (lng) tt;
+ }
+#endif
+#ifdef HAVE_CLOCK_GETTIME
+ {
+ struct timespec ts;
+
+ if (clock_gettime(CLOCK_REALTIME, &ts) == 0)
+ return ts.tv_sec * 1000000 + ts.tv_nsec / 1000;
+ }
+#endif
+#ifdef HAVE_GETTIMEOFDAY
+ {
+ struct timeval tp;
+
+ gettimeofday(&tp, NULL);
+ return (lng) tp.tv_sec * 1000000 + (lng) tp.tv_usec;
+ }
+#else
+#ifdef HAVE_FTIME
+ {
+ struct timeb tb;
+
+ ftime(&tb);
+ return (lng) tb.time * 1000000 + (lng) tb.millitm * 1000;
+ }
+#endif
+#endif
+}
+
+int
+GDKunix_timestamp_msec(void)
+{
+ return (int) (GDKunix_timestamp_usec() / 1000);
+}
diff --git a/gdk/gdk_system.h b/gdk/gdk_system.h
--- a/gdk/gdk_system.h
+++ b/gdk/gdk_system.h
@@ -356,4 +356,7 @@ gdk_export int MT_check_nr_cores(void);
gdk_export lng GDKusec(void);
gdk_export int GDKms(void);
+gdk_export lng GDKunix_timestamp_usec(void);
+gdk_export int GDKunix_timestamp_msec(void);
+
#endif /*_GDK_SYSTEM_H_*/
diff --git a/sql/backends/monet5/cquery.mal b/sql/backends/monet5/cquery.mal
--- a/sql/backends/monet5/cquery.mal
+++ b/sql/backends/monet5/cquery.mal
@@ -25,9 +25,6 @@ the MAL block to determine the input/out
pattern resume(mod:str, fcn:str)
address CQresume
comment "Activate a specific continuous query with no changes";
-pattern resumenoalter(mod:str, fcn:str)
-address CQresumeNoAlter
-comment "Activate a specific continuous query with changes to the cycles and
heartbeat values";
pattern resume()
address CQresumeAll
@@ -56,15 +53,25 @@ comment "Sleep for some time";
pattern cycles(mod:str, fcn:str,cnt:int)
address CQcycles
comment "Limit number of petrinet steps";
+
pattern cycles(cnt:int)
address CQcycles
comment "Limit number of petrinet steps";
+pattern beginat(mod:str, fcn:str,beginat:lng)
+address CQbeginAt
+comment "Delay beginning of a continuous query, where beginat is an UNIX
timestamp with microsecond precision";
+
+pattern beginat(beginat:lng)
+address CQbeginAt
+comment "Delay beginning of all continuous queries, where beginat is an UNIX
timestamp with microsecond precision";
+
# scheduler heart beat for each CQ
-pattern heartbeat(mod:str, fcn:str,cnt:int)
+pattern heartbeat(mod:str, fcn:str,cnt:lng)
address CQheartbeat
comment "Awake the query after cnt milliseconds";
-pattern heartbeat(cnt:int)
+
+pattern heartbeat(cnt:lng)
address CQheartbeat
comment "Awake all queries after cnt milliseconds";
diff --git a/sql/backends/monet5/sql_cquery.c b/sql/backends/monet5/sql_cquery.c
--- a/sql/backends/monet5/sql_cquery.c
+++ b/sql/backends/monet5/sql_cquery.c
@@ -375,7 +375,7 @@ CQerror(Client cntxt, MalBlkPtr mb, MalS
idx = CQlocate(sch, fcn);
if( idx == pnettop)
- throw(SQL,"cquery.error","The continuous procedure %s.%s is not
accessible\n",sch,fcn);
+ throw(SQL,"cquery.error","The continuous query %s.%s is not
accessible\n",sch,fcn);
pnet[idx].error = GDKstrdup(error);
if(pnet[idx].error == NULL)
@@ -396,9 +396,9 @@ CQshow(Client cntxt, MalBlkPtr mb, MalSt
idx = CQlocate(sch, fcn);
if( idx == pnettop)
- throw(SQL,"cquery.show","The continuous procedure %s.%s is not
accessible\n",sch,fcn);
+ throw(SQL,"cquery.show","The continuous query %s.%s is not
accessible\n",sch,fcn);
- printFunction(cntxt->fdout, pnet[idx].mb, 0, LIST_MAL_NAME |
LIST_MAL_VALUE | LIST_MAL_MAPI);
+ printFunction(cntxt->fdout, pnet[idx].mb, 0, LIST_MAL_NAME |
LIST_MAL_VALUE | LIST_MAL_MAPI);
return MAL_SUCCEED;
}
@@ -502,11 +502,12 @@ CQregister(Client cntxt, MalBlkPtr mb, M
msg = createException(SQL,"cquery.register",SQLSTATE(42000)
"The heartbeats value must be non negative\n");
goto finish;
}
- if(start_atom && (msg = convert_atom_into_unix_timestamp(start_atom,
&start_at_parsed)) != MAL_SUCCEED){
+ if(start_atom && (msg = convert_atom_into_unix_timestamp(start_atom->a,
&start_at_parsed)) != MAL_SUCCEED){
goto finish;
}
/**
- * We are using GDKusec() to check if the query is enable to fire, so
we have to convert into microseconds
+ * We are using GDKunix_timestamp_usec() to check if the query is
enable to fire, so we have to convert into
+ * microseconds
*/
start_at_parsed *= 1000;
@@ -635,7 +636,8 @@ CQregister(Client cntxt, MalBlkPtr mb, M
}
pnet[pnettop].cycles = cycles;
pnet[pnettop].beats = SET_HEARTBEATS(heartbeats);
- pnet[pnettop].run = start_at_parsed;
+ //subtract the beats value so the CQ will start at the precise moment
+ pnet[pnettop].run = start_at_parsed - (pnet[pnettop].beats > 0 ?
pnet[pnettop].beats : 0);
pnet[pnettop].seen = *timestamp_nil;
pnet[pnettop].status = CQWAIT;
pnettop++;
@@ -675,7 +677,7 @@ CQresumeInternal(Client cntxt, MalBlkPtr
msg =
createException(SQL,"cquery.resume",SQLSTATE(42000) "The heartbeats value must
be non negative\n");
goto finish;
}
- if(start_atom && (msg =
convert_atom_into_unix_timestamp(start_atom, &start_at_parsed)) != MAL_SUCCEED){
+ if(start_atom && (msg =
convert_atom_into_unix_timestamp(start_atom->a, &start_at_parsed)) !=
MAL_SUCCEED){
goto finish;
}
start_at_parsed *= 1000;
@@ -709,9 +711,8 @@ CQresumeInternal(Client cntxt, MalBlkPtr
pnet[idx].status = CQWAIT;
if(with_alter) {
pnet[idx].cycles = cycles;
- if(start_at_parsed > 0)
- pnet[idx].run = start_at_parsed;
pnet[idx].beats = SET_HEARTBEATS(heartbeats);
+ pnet[idx].run = start_at_parsed - (pnet[idx].beats > 0 ?
pnet[idx].beats : 0);
}
/* start the scheduler if needed */
@@ -907,6 +908,45 @@ CQpauseAll(Client cntxt, MalBlkPtr mb, M
}
str
+CQbeginAt(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
+{
+ str sch, fcn, msg = MAL_SUCCEED;
+ int idx=0, last;
+ lng delay;
+ (void) cntxt;
+ (void) mb;
+
+ MT_lock_set(&ttrLock);
+ last = pnettop;
+ if( pci->argc >2){
+ sch = *getArgReference_str(stk,pci,1);
+ fcn = *getArgReference_str(stk,pci,2);
+ idx = CQlocate(sch, fcn);
+ if( idx == pnettop) {
+ msg = createException(SQL,"cquery.begintat",
+
SQLSTATE(3F000) "The continuous query %s.%s not accessible\n",sch,fcn);
+ goto finish;
+ }
+ last = idx+1;
+ delay = *getArgReference_lng(stk,pci,3);
+ } else
+ delay = *getArgReference_lng(stk,pci,1);
+#ifdef DEBUG_CQUERY
+ fprintf(stderr, "#set begin at \n");
+#endif
+ if(delay < 0){
+ msg = createException(SQL,"cquery.begintat",SQLSTATE(42000)
"The delay value must be non negative\n");
+ goto finish;
+ }
+ for( ; idx < last; idx++)
+ pnet[idx].run = delay - (pnet[idx].beats > 0 ? pnet[idx].beats
: 0);
+
+ finish:
+ MT_lock_unset(&ttrLock);
+ return msg;
+}
+
+str
CQcycles(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
str sch, fcn, msg = MAL_SUCCEED;
@@ -922,7 +962,7 @@ CQcycles(Client cntxt, MalBlkPtr mb, Mal
idx = CQlocate(sch, fcn);
if( idx == pnettop) {
msg = createException(SQL,"cquery.cycles",
-
SQLSTATE(3F000) "The continuous procedure %s.%s not accessible\n",sch,fcn);
+
SQLSTATE(3F000) "The continuous query %s.%s not accessible\n",sch,fcn);
goto finish;
}
last = idx+1;
@@ -948,7 +988,8 @@ str
CQheartbeat(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
str sch, fcn, msg = MAL_SUCCEED;
- int heartbeats, j, there_is_window_constraint, idx=0, last= pnettop;
+ int j, there_is_window_constraint, idx=0, last= pnettop;
+ lng heartbeats;
(void) cntxt;
(void) mb;
@@ -959,16 +1000,16 @@ CQheartbeat(Client cntxt, MalBlkPtr mb,
idx = CQlocate(sch, fcn);
if( idx == pnettop) {
msg = createException(SQL,"cquery.heartbeat",
-
SQLSTATE(3F000) "The continuous procedure %s.%s not accessible\n",sch,fcn);
+
SQLSTATE(3F000) "The continuous query %s.%s not accessible\n",sch,fcn);
goto finish;
}
last = idx+1;
- heartbeats = *getArgReference_int(stk,pci,3);
+ heartbeats = *getArgReference_lng(stk,pci,3);
#ifdef DEBUG_CQUERY
fprintf(stderr, "#set the heartbeat of %s.%s to
%d\n",sch,fcn,beats);
#endif
} else{
- heartbeats = *getArgReference_int(stk,pci,1);
+ heartbeats = *getArgReference_lng(stk,pci,1);
#ifdef DEBUG_CQUERY
fprintf(stderr, "#set the heartbeat %d ms\n",beats);
#endif
@@ -993,7 +1034,13 @@ CQheartbeat(Client cntxt, MalBlkPtr mb,
}
for( ; idx < last; idx++){
- pnet[idx].beats = SET_HEARTBEATS(heartbeats);
+ int new_hearbeats = SET_HEARTBEATS(heartbeats);
+ if(new_hearbeats > pnet[idx].beats) { //has to do the alignment
of the starting point
+ pnet[idx].run -= (new_hearbeats - pnet[idx].beats);
+ } else {
+ pnet[idx].run += (pnet[idx].beats - new_hearbeats);
+ }
+ pnet[idx].beats = new_hearbeats;
}
finish:
@@ -1162,7 +1209,7 @@ CQexecute( Client cntxt, int idx)
#ifdef DEBUG_CQUERY
fprintf(stderr, "#cquery.execute %s.%s locked\n",node->mod, node->fcn);
- fprintFunction(stderr, node->mb, 0, LIST_MAL_NAME | LIST_MAL_VALUE |
LIST_MAL_MAPI);
+ fprintFunction(stderr, node->mb, 0, LIST_MAL_NAME | LIST_MAL_VALUE |
LIST_MAL_MAPI);
#endif
msg = runMALsequence(cntxt, node->mb, 1, 0, node->stk, 0, 0);
@@ -1205,7 +1252,7 @@ CQscheduler(void *dummy)
come back. We also only have to check the places that are
marked
non empty. You can only trigger on empty baskets using a
heartbeat */
memset((void*) claimed, 0, sizeof(claimed));
- now = GDKusec();
+ now = GDKunix_timestamp_usec();
pntasks=0;
MT_lock_set(&ttrLock); // analysis should be done with
exclusive access
for (k = i = 0; i < pnettop; i++)
@@ -1402,7 +1449,7 @@ CQstartScheduler(void)
throw(MAL, "cquery.startScheduler",SQLSTATE(HY001) "Could not
initialize CQscheduler\n");
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
