Changeset: 82e33e93d5a3 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=82e33e93d5a3
Modified Files:
monetdb5/mal/Makefile.ag
monetdb5/mal/mal.mx
monetdb5/mal/mal_client.mx
monetdb5/mal/mal_dataflow.mx
monetdb5/mal/mal_debugger.mx
monetdb5/mal/mal_function.mx
monetdb5/mal/mal_instruction.mx
monetdb5/mal/mal_interpreter.mx
monetdb5/mal/mal_listing.mx
monetdb5/mal/mal_module.mx
monetdb5/mal/mal_profiler.mx
monetdb5/mal/mal_recycle.mx
monetdb5/mal/mal_resolve.mx
monetdb5/mal/mal_runtime.mx
monetdb5/modules/atoms/mtime.mx
monetdb5/modules/kernel/colcast.mx
monetdb5/modules/kernel/colifthen.mx
monetdb5/modules/kernel/colmmath.mx
monetdb5/modules/kernel/colmtime.mx
Branch: headless
Log Message:
More preparations for and clean up
Splitting some files, removing Mx macros, etc.
Found one design bug, returning exceptions within a dataflow
is dangerous and leads to undefined behavior.
diffs (truncated from 6695 to 300 lines):
diff --git a/monetdb5/mal/Makefile.ag b/monetdb5/mal/Makefile.ag
--- a/monetdb5/mal/Makefile.ag
+++ b/monetdb5/mal/Makefile.ag
@@ -40,11 +40,14 @@
mal_function.mx \
mal_import.mx \
mal_instruction.mx \
+ mal_listing.mx \
mal_interpreter.mx \
+ mal_dataflow.mx \
mal_linker.mx \
mal_module.mx \
mal_namespace.mx \
mal_parser.mx \
+ mal_runtime.mx \
mal_profiler.mx \
mal_properties.mx \
mal_readline.mx \
diff --git a/monetdb5/mal/mal.mx b/monetdb5/mal/mal.mx
--- a/monetdb5/mal/mal.mx
+++ b/monetdb5/mal/mal.mx
@@ -108,6 +108,27 @@
#define MAXPATHLEN 1024
#endif
+/* Recycler statistics per client */
+typedef struct RECSTAT {
+ int statements; /* total number of statements executed */
+ int recycled; /* total number of statements recycled */
+ int recycled0; /* recycled statements per query */
+ lng time0; /* time per query */
+ int curQ; /* index of current query in Qry Patterns array*/
+ int recent; /* the most recent entry in RP touched by current
query */
+ int recycleMiss; /* DBG:count of misses due to cache eviction */
+ int recycleRem; /* DBG:count of removed entries */
+ lng ccCalls; /* Number of calls to cleanCache */
+ lng ccInstr; /* Number of instructions evicted by eviction policy*/
+ lng crdInstr; /* Number of instructions not admited in RP by CRD */
+ int trans; /* Number of data transfer instructions */
+ lng transKB; /* Size in KB of transferred data */
+ int recTrans; /* Number of recycled data transfer instructions */
+ lng recTransKB; /* Size in KB of recycled transferred data */
+ int RPadded0; /* Number of instructions added to RP per query */
+ int RPreset0; /* Number of instructions evicted from RP by reset()
due to updates*/
+} *RecPtr, RecStat;
+
#endif /* _MAL_H*/
@c
/* Author(s) M.L. Kersten
@@ -313,6 +334,7 @@
#include "mal_linker.h"
#include "mal_session.h"
#include "mal_parser.h"
+#include "mal_runtime.h"
#include "mal_interpreter.h"
#include "mal_namespace.h" /* for initNamespace() */
#include "mal_client.h"
@@ -412,7 +434,7 @@
*/
MCexitPending();
RECYCLEshutdown(mal_clients); /* remove any left over intermediates */
- stopProfiling();
+ /* stopProfiling(); assume circumventional close */
#if 0
{
int reruns=0, goon;
diff --git a/monetdb5/mal/mal_client.mx b/monetdb5/mal/mal_client.mx
--- a/monetdb5/mal/mal_client.mx
+++ b/monetdb5/mal/mal_client.mx
@@ -26,7 +26,6 @@
/*#define MAL_CLIENT_DEBUG */
#include "mal_resolve.h"
-#include "mal_profiler.h"
#include "mal.h"
#define MAL_MAXCLIENTS 64
@@ -448,7 +447,7 @@
#ifdef MAL_CLIENT_DEBUG
printf("# Exit client %d\n", c->idx);
#endif
- MPresetProfiler(c->fdout);
+ /* MPresetProfiler(c->fdout);*/
if (c->father == NULL) { /* normal client */
if( c->fdout && c->fdout != GDKstdout){
(void) mnstr_close(c->fdout);
@@ -557,7 +556,7 @@
t = THRnew(c->mypid,cname);
if ( t==0) {
showException(MAL, "initClientThread", "Failed to initialize
client");
- MPresetProfiler(c->fdout);
+ /* MPresetProfiler(c->fdout);*/
return -1;
}
/*
diff --git a/monetdb5/mal/mal_dataflow.mx b/monetdb5/mal/mal_dataflow.mx
new file mode 100644
--- /dev/null
+++ b/monetdb5/mal/mal_dataflow.mx
@@ -0,0 +1,1178 @@
+@/
+The contents of this file are subject to the MonetDB Public License
+Version 1.1 (the "License"); you may not use this file except in
+compliance with the License. You may obtain a copy of the License at
+http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
+
+Software distributed under the License is distributed on an "AS IS"
+basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+License for the specific language governing rights and limitations
+under the License.
+
+The Original Code is the MonetDB Database System.
+
+The Initial Developer of the Original Code is CWI.
+Portions created by CWI are Copyright (C) 1997-July 2008 CWI.
+Copyright August 2008-2011 MonetDB B.V.
+All Rights Reserved.
+@
+
+@h
+#ifndef _MAL_INTERPRET_H
+#define _MAL_INTERPRET_H
+
+#include "mal_interpreter.h"
+
+/*#define DEBUG_MAL_INTERPRETER*/
+/* #define DEBUG_FLOW */
+/* #define DEBUG_FLOW2 */
+/* #define DEBUG_MEMORY_CLAIM*/
+
+mal_export str runMALdataflow( Client cntxt, MalBlkPtr mb, int startpc,
+ int stoppc, MalStkPtr stk, MalStkPtr env, InstrPtr pcicaller);
+
+#endif /* _MAL_INTERPRET_H*/
+@c
+/* Author(s) M.L. Kersten, N. Nes
+ * The MAL Dataflow Driven Interpreter
+ * Dataflow processing only works on a code
+ * sequence that does not include additional (implicit) flow of control
+ * statements and, ideally, it consist of expensive columns operations.
+ * The dataflow interpreter selects cheap instructions
+ * using a simple cost function based on the size of the columns involved.
+ *
+ * The dataflow portion is identified as a guarded block,
+ * whose entry is controlled by the function language.dataflow();
+ * This way the function can inform the caller to skip the block
+ * when dataflow execution was performed.
+ *
+ * The flow graphs should be organized such that parallel threads can
+ * access it mostly without expensive locking.
+ * Furthermore, per flow block we keep an administration when
+ * variables are blocked or not. An instruction can fire if all
+ * its input variables are not blocked.
+ *
+ * After the expression has been evaluated we should check for a
+ * possible change in the control flow.
+ * NOT the case in dataflow
+*/
+
+#include "monetdb_config.h"
+#include "mal_profiler.h"
+#include "mal_interpreter.h"
+#include "mal_debugger.h"
+#include "mal_recycle.h"
+#include "mal_linker.h"
+#include "mal_listing.h"
+#include "mal_type.h"
+
+#define MEMORY_THRESHOLD 0.8
+#define MAXHOT 16
+
+static lng memorypool; /* memory claimed by concurrent threads */
+static lng memoryused; /* memory used for intermediates */
+static int memoryclaims = 0; /* number of threads active with expensive
operations */
+static struct{
+ lng claim; /* actual claim on memory*/
+ int bid;
+} hotpotatoes[MAXHOT];
+static int hottop = 0;
+
+#define DFLOWpending 0 /* runnable */
+#define DFLOWrunning 1 /* currently in progress */
+#define DFLOWwrapup 2 /* done! */
+#define DFLOWretry 3 /* reschedule */
+
+typedef struct queue {
+ int size; /* size of queue */
+ int last; /* last element in the queue */
+ void **data;
+ MT_Lock l; /* its a shared resource, ie we need locks */
+ MT_Sema s; /* threads wait on empty queues */
+} queue;
+
+
+typedef struct {
+ MT_Id tid;
+ int id;
+ queue *todo; /* pending actions for this client */
+ lng clk;
+} FlowTask;
+
+typedef struct DataFlow {
+ Client cntxt; /* for debugging and client resolution */
+ MalBlkPtr mb; /* carry the context */
+ MalStkPtr stk;
+ int start, stop; /* guarded block under consideration*/
+ char *status; /* statements can be blocked on other
statements */
+ char *blocked; /* blocked, should be created first */
+ int *assign; /* first assignment of variable */
+ int *inuse; /* inuse in parallel threads reference count */
+ queue *done; /* work finished */
+ queue *todo; /* pending actions for this client */
+ int nway; /* number of workers */
+ FlowTask *worker; /* worker threads for the client */
+ struct DataFlow *free; /* free list */
+} *DataFlow, DataFlowRec;
+
+typedef struct {
+ int pc; /* pc in underlying malblock */
+ sht status;
+ sht cost;
+ str error;
+ DataFlow flow;
+} *FlowStep, FlowStepRec;
+
+
+static DataFlow flows = NULL;
+static int workerid = 0;
+/*
+ * Running all eligible instructions in parallel creates
+ * resource contention. This means we should implement
+ * an admission control scheme where threads are temporarily
+ * postponed if the claim for memory exceeds a threshold
+ * In general such contentions will be hard to predict,
+ * because they depend on the algorithm, the input sizes,
+ * and the output produced.
+ *
+ * The heuristic is based on calculating the storage footprint
+ * of the operands and assuming it preferrably should fit in memory.
+ * Ofcourse, there may be intermediate structures being
+ * used and the size of the result is not a priori known.
+ * For this, we use a high watermark on the amount of
+ * physical memory we pre-allocate for the claims.
+ *
+ * Instructions are eligible to be executed when the
+ * total footprint of all concurrent executions stays below
+ * the high-watermark or it is the single expensive
+ * instruction being started.
+ *
+ * When we run out of memory, the instruction is delayed.
+ * How long depends on the other instructions to free up
+ * resources. The current policy simple takes a local
+ * decision by delaying the instruction based on its
+ * past and the size of the memory pool size.
+ * The waiting penalty decreases with each step to ensure
+ * it will ultimately taken into execution, with possibly
+ * all resource contention effects.
+ *
+ * Another option would be to maintain a priority queue of
+ * suspended instructions.
+*/
+
+#define heapinfo(X) if((X) && (X)->base) vol = (X)->free; else vol = 0;
+#define hashinfo(X) if((X) && (X)->mask) vol =
((X)->mask+(X)->lim+1)*sizeof(int) + sizeof(*(X)); else vol = 0;
+
+static lng
+calcclaim(MalStkPtr stk, InstrPtr pci, int i)
+{
+ lng total = 0,vol;
+ BAT *b;
+
+ assert(i < pci->argc);
+ if (stk->stk[getArg(pci,i)].vtype == TYPE_bat){
+ b = BBPquickdesc(stk->stk[getArg(pci,i)].val.bval,TRUE);
+ if (b==NULL)
+ return total;
+ heapinfo(&b->H->heap); total += vol;
+ heapinfo(b->H->vheap); total += vol;
+ hashinfo(b->H->hash); total += vol;
+ if ( b->htype == TYPE_oid && !BAThordered(b) && !BAThdense(b) ){
+ /* assume we may have to do random IO, punish it by
increasing the claim with hash elem size*/
+ total += 4 * BATcount(b) ;
+ total = total > (lng) (MEMORY_THRESHOLD *
monet_memory)? (lng) (MEMORY_THRESHOLD * monet_memory) : total;
+ }
+ }
+ return total;
+}
+
+static lng
+getHotClaim()
+{
+ lng total = 0;
+ int i;
+ for ( i = 0; i < hottop; i++)
+ total += hotpotatoes[i].claim;
+ return total;
+}
+
+/*
+ * Calculate the memory need of a single instruction and also determine how
much of the
_______________________________________________
Checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
