Changeset: a3881091108d for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=a3881091108d
Modified Files:
clients/Tests/exports.stable.out
monetdb5/mal/mal_dataflow.c
monetdb5/mal/mal_resource.c
monetdb5/mal/mal_resource.h
monetdb5/modules/mal/clients.c
monetdb5/optimizer/opt_mitosis.c
monetdb5/optimizer/opt_mitosis.h
Branch: default
Log Message:
Fixed the workerlimit stuff and cleanup code.
diffs (truncated from 354 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
@@ -1355,7 +1355,8 @@ str LIKEjoin(bat *r1, bat *r2, const bat
str LIKEjoin1(bat *r1, bat *r2, const bat *lid, const bat *rid, const bat
*slid, const bat *srid, const bit *nil_matches, const lng *estimate);
str MACROprocessor(Client cntxt, MalBlkPtr mb, Symbol t);
int MAL_MAXCLIENTS;
-int MALadmission(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, lng
argclaim);
+int MALadmission_claim(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci, lng argclaim);
+void MALadmission_release(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr
pci, lng argclaim);
str MALassertBit(void *ret, bit *val, str *msg);
str MALassertHge(void *ret, hge *val, str *msg);
str MALassertInt(void *ret, int *val, str *msg);
diff --git a/monetdb5/mal/mal_dataflow.c b/monetdb5/mal/mal_dataflow.c
--- a/monetdb5/mal/mal_dataflow.c
+++ b/monetdb5/mal/mal_dataflow.c
@@ -327,6 +327,7 @@ DFLOWworker(void *T)
int tid = THRgettid();
str error = 0;
int i,last;
+ lng claim;
Client cntxt;
InstrPtr p;
@@ -381,7 +382,8 @@ DFLOWworker(void *T)
}
p= getInstrPtr(flow->mb,fe->pc);
- if (MALadmission(flow->cntxt, flow->mb, flow->stk, p,
fe->argclaim)) {
+ claim = fe->argclaim;
+ if (MALadmission_claim(flow->cntxt, flow->mb, flow->stk, p,
claim)) {
// never block on deblockdataflow()
if( p->fcn != (MALfcn) deblockdataflow){
fe->hotclaim = 0; /* don't assume priority
anymore */
@@ -394,9 +396,9 @@ DFLOWworker(void *T)
}
error = runMALsequence(flow->cntxt, flow->mb, fe->pc, fe->pc +
1, flow->stk, 0, 0);
PARDEBUG fprintf(stderr, "#executed pc= %d wrk= %d claim= "
LLFMT "," LLFMT "," LLFMT " %s\n",
- fe->pc, id, fe->argclaim,
fe->hotclaim, fe->maxclaim, error ? error : "");
+ fe->pc, id, claim,
fe->hotclaim, fe->maxclaim, error ? error : "");
/* release the memory claim */
- MALadmission(flow->cntxt, flow->mb, flow->stk, p,
-fe->argclaim);
+ MALadmission_release(flow->cntxt, flow->mb, flow->stk, p,
claim);
/* update the numa information. keep the thread-id producing
the value */
p= getInstrPtr(flow->mb,fe->pc);
for( i = 0; i < p->argc; i++)
diff --git a/monetdb5/mal/mal_resource.c b/monetdb5/mal/mal_resource.c
--- a/monetdb5/mal/mal_resource.c
+++ b/monetdb5/mal/mal_resource.c
@@ -12,15 +12,13 @@
#include "mal_resource.h"
#include "mal_private.h"
-/* MEMORY admission does not seem to have a major impact so far. */
+/* Memory based admission does not seem to have a major impact so far. */
static lng memorypool = 0; /* memory claimed by concurrent threads */
-static int memoryclaims = 0;
void
mal_resource_reset(void)
{
- memorypool = 0;
- memoryclaims = 0;
+ memorypool = (lng) MEMORY_THRESHOLD;
}
/*
* Running all eligible instructions in parallel creates
@@ -47,13 +45,7 @@ mal_resource_reset(void)
* 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.
+ * claim of the memory.
*/
/*
@@ -106,94 +98,88 @@ getMemoryClaim(MalBlkPtr mb, MalStkPtr s
*/
static MT_Lock admissionLock = MT_LOCK_INITIALIZER("admissionLock");
-/* experiments on sf-100 on small machine showed no real improvement */
-
int
-MALadmission(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, lng
argclaim)
+MALadmission_claim(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci,
lng argclaim)
{
- int workers;
- lng mbytes;
-
(void) mb;
(void) pci;
- /* optimistically set memory */
if (argclaim == 0)
return 0;
- /* if we are dealing with a check instruction, just continue */
- /* TOBEDONE */
MT_lock_set(&admissionLock);
/* Check if we are allowed to allocate another worker thread for this
client */
/* It is somewhat tricky, because we may be in a dataflow recursion,
each of which should be counted for.
- * A way out is to attach the thread count to the MAL stacks instead,
which just limits the level
+ * A way out is to attach the thread count to the MAL stacks, which
just limits the level
* of parallism for a single dataflow graph.
*/
- workers = stk->workers;
- if( cntxt->workerlimit && cntxt->workerlimit <= workers){
+ if(cntxt->workerlimit && cntxt->workerlimit < stk->workers){
PARDEBUG
- fprintf(stderr, "#DFLOWadmit worker limit reached, %d
<= %d\n", cntxt->workerlimit, workers);
+ fprintf(stderr, "#DFLOWadmit worker limit reached, %d
<= %d\n", cntxt->workerlimit, stk->workers);
MT_lock_unset(&admissionLock);
return -1;
}
- /* Determine if the total memory resource is exhausted, because it is
overall limitation.
- */
- if ( memoryclaims < 0){
- PARDEBUG
- fprintf(stderr, "#DFLOWadmit memoryclaim reset ");
- memoryclaims = 0;
- }
- if ( memorypool <= 0 && memoryclaims == 0){
+ /* Determine if the total memory resource is exhausted, because it is
overall limitation. */
+ if ( memorypool <= 0){
+ // we accidently released too much memory or need to initialize
PARDEBUG
fprintf(stderr, "#DFLOWadmit memorypool reset ");
memorypool = (lng) MEMORY_THRESHOLD;
}
/* the argument claim is based on the input for an instruction */
- if (argclaim > 0) {
- if ( memoryclaims == 0 || memorypool > argclaim ) {
- /* If we are low on memory resources, limit the user if
he exceeds his memory budget
- * but make sure there is at least one thread active */
- if ( cntxt->memorylimit) {
- mbytes = (lng) cntxt->memorylimit *
LL_CONSTANT(1048576);
- if (argclaim + stk->memory > mbytes){
- MT_lock_unset(&admissionLock);
- PARDEBUG
- fprintf(stderr, "#Delayed due
to lack of session memory " LLFMT " requested "LLFMT"\n",
- stk->memory,
argclaim);
- return -1;
- }
+ if ( memorypool > argclaim || stk->workers == 0 ) {
+ /* If we are low on memory resources, limit the user if he
exceeds his memory budget
+ * but make sure there is at least one worker thread active */
+ if ( 0 && cntxt->memorylimit) {
+ if (argclaim + stk->memory > (lng) cntxt->memorylimit *
LL_CONSTANT(1048576)){
+ MT_lock_unset(&admissionLock);
+ PARDEBUG
+ fprintf(stderr, "#Delayed due to lack
of session memory " LLFMT " requested "LLFMT"\n",
+ stk->memory, argclaim);
+ return -1;
}
- memorypool -= argclaim;
stk->memory += argclaim;
- memoryclaims++;
- PARDEBUG
- fprintf(stderr, "#DFLOWadmit %3d thread %d pool
" LLFMT "claims " LLFMT "\n",
- memoryclaims, THRgettid(),
memorypool, argclaim);
- stk->workers++;
- MT_lock_unset(&admissionLock);
- return 0;
}
+ memorypool -= argclaim;
PARDEBUG
- fprintf(stderr, "#Delayed due to lack of memory " LLFMT
" requested " LLFMT " memoryclaims %d\n",
- memorypool, argclaim, memoryclaims);
+ fprintf(stderr, "#DFLOWadmit thread %d pool " LLFMT
"claims " LLFMT "\n",
+ THRgettid(), memorypool, argclaim);
+ stk->workers++;
MT_lock_unset(&admissionLock);
- return -1;
+ return 0;
}
-
- /* return the session budget */
- if (cntxt->memorylimit) {
+ PARDEBUG
+ fprintf(stderr, "#Delayed due to lack of memory " LLFMT "
requested " LLFMT "\n",
+ memorypool, argclaim);
+ MT_lock_unset(&admissionLock);
+ return -1;
+}
+
+void
+MALadmission_release(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci,
lng argclaim)
+{
+ /* release memory claimed before */
+ (void) mb;
+ (void) pci;
+ if (argclaim == 0 )
+ return;
+
+ MT_lock_set(&admissionLock);
+ if ( 0 && cntxt->memorylimit) {
PARDEBUG
fprintf(stderr, "#Return memory to session budget "
LLFMT "\n", stk->memory);
stk->memory -= argclaim;
}
- /* release memory claimed before */
- memorypool -= argclaim;
- memoryclaims--;
+ memorypool += argclaim;
+ if ( memorypool > (lng) MEMORY_THRESHOLD ){
+ PARDEBUG
+ fprintf(stderr, "#DFLOWadmit memorypool reset ");
+ memorypool = (lng) MEMORY_THRESHOLD;
+ }
stk->workers--;
-
PARDEBUG
- fprintf(stderr, "#DFLOWadmit %3d thread %d pool " LLFMT "
claims " LLFMT "\n",
- memoryclaims, THRgettid(), memorypool,
argclaim);
+ fprintf(stderr, "#DFLOWadmit thread %d pool " LLFMT " claims "
LLFMT "\n",
+ THRgettid(), memorypool, argclaim);
MT_lock_unset(&admissionLock);
- return 0;
+ return;
}
diff --git a/monetdb5/mal/mal_resource.h b/monetdb5/mal/mal_resource.h
--- a/monetdb5/mal/mal_resource.h
+++ b/monetdb5/mal/mal_resource.h
@@ -21,7 +21,8 @@
#define heapinfo(X,Id) (((X) && (X)->base ) ? (X)->free : 0)
#define hashinfo(X,Id) ((X) && (X) != (Hash *) 1 ? heapinfo(&(X)->heap, Id) :
0)
-mal_export int MALadmission(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr pci, lng argclaim);
+mal_export int MALadmission_claim(Client cntxt, MalBlkPtr mb, MalStkPtr stk,
InstrPtr pci, lng argclaim);
+mal_export void MALadmission_release(Client cntxt, MalBlkPtr mb, MalStkPtr
stk, InstrPtr pci, lng argclaim);
#define FAIRNESS_THRESHOLD (MAX_DELAYS * DELAYUNIT)
diff --git a/monetdb5/modules/mal/clients.c b/monetdb5/modules/mal/clients.c
--- a/monetdb5/modules/mal/clients.c
+++ b/monetdb5/modules/mal/clients.c
@@ -306,8 +306,6 @@ CLTsetmemorylimit(Client cntxt, MalBlkPt
throw(MAL,"clients.setmemorylimit","Illegal session id");
if (mal_clients[idx].mode == FREECLIENT)
throw(MAL,"clients.setmemorylimit","Session not active
anymore");
- if( limit < 100)
- throw(MAL, "clients.setmemorylimit", "At least 100 MB needed");
if (cntxt->user == mal_clients[idx].user || cntxt->user == MAL_ADMIN){
mal_clients[idx].memorylimit = limit;
}
diff --git a/monetdb5/optimizer/opt_mitosis.c b/monetdb5/optimizer/opt_mitosis.c
--- a/monetdb5/optimizer/opt_mitosis.c
+++ b/monetdb5/optimizer/opt_mitosis.c
@@ -32,7 +32,6 @@ OPTmitosisImplementation(Client cntxt, M
(void) cntxt;
(void) stk;
- activeClients = mb->activeClients = MCactiveClients();
old = mb->stmt;
for (i = 1; i < mb->stop; i++) {
InstrPtr p = old[i];
@@ -123,25 +122,36 @@ OPTmitosisImplementation(Client cntxt, M
* Take into account the number of client connections,
* because all user together are responsible for resource contentions
*/
- m = GDK_mem_maxsize / argsize;
- /* if data exceeds memory size,
- * i.e., (rowcnt*argsize > GDK_mem_maxsize),
- * i.e., (rowcnt > GDK_mem_maxsize/argsize = m) */
- assert(threads > 0);
- assert(activeClients > 0);
- if (rowcnt > m && m / threads / activeClients > 0) {
- /* create |pieces| > |threads| partitions such that
- * |threads| partitions at a time fit in memory,
- * i.e., (threads*(rowcnt/pieces) <= m),
- * i.e., (rowcnt/pieces <= m/threads),
- * i.e., (pieces => rowcnt/(m/threads))
- * (assuming that (m > threads*MINPARTCNT)) */
- pieces = (int) (rowcnt / (m / threads / activeClients)) + 1;
- } else if (rowcnt > MINPARTCNT) {
- /* exploit parallelism, but ensure minimal partition size to
- * limit overhead */
- pieces = (int) MIN(rowcnt / MINPARTCNT, (BUN) threads);
+ activeClients = mb->activeClients = MCactiveClients();
+
+ if( cntxt->memorylimit){
+ /* the new mitosis scheme uses a maximum chunck size in MB from
the client context */
+ m = (cntxt->memorylimit * 1024 *1024) / row_size;
+ pieces = rowcnt / m + (rowcnt - m * pieces > 0);
}
+ if( cntxt->memorylimit == 0 || pieces <= 1){
+ /* the old allocation scheme */
+ m = GDK_mem_maxsize / argsize;
+ /* if data exceeds memory size,
+ * i.e., (rowcnt*argsize > GDK_mem_maxsize),
+ * i.e., (rowcnt > GDK_mem_maxsize/argsize = m) */
+ assert(threads > 0);
+ assert(activeClients > 0);
+ if (rowcnt > m && m / threads / activeClients > 0) {
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
