Line data    Source code

       1             : /*
       2             :  * SPDX-License-Identifier: MPL-2.0
       3             :  *
       4             :  * This Source Code Form is subject to the terms of the Mozilla Public
       5             :  * License, v. 2.0.  If a copy of the MPL was not distributed with this
       6             :  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
       7             :  *
       8             :  * Copyright 2024 MonetDB Foundation;
       9             :  * Copyright August 2008 - 2023 MonetDB B.V.;
      10             :  * Copyright 1997 - July 2008 CWI.
      11             :  */
      12             : 
      13             : /* (author) M.L. Kersten
      14             :  */
      15             : #include "monetdb_config.h"
      16             : #include "mal_exception.h"
      17             : #include "mal_resource.h"
      18             : #include "mal_private.h"
      19             : #include "mal_internal.h"
      20             : #include "mal_instruction.h"
      21             : 
      22             : /* Memory based admission does not seem to have a major impact so far. */
      23             : static lng memorypool = 0;              /* memory claimed by concurrent threads */
      24             : 
      25             : static MT_Lock admissionLock = MT_LOCK_INITIALIZER(admissionLock);
      26             : 
      27             : void
      28         334 : mal_resource_reset(void)
      29             : {
      30         334 :         MT_lock_set(&admissionLock);
      31         334 :         memorypool = (lng) MEMORY_THRESHOLD;
      32         334 :         MT_lock_unset(&admissionLock);
      33         334 : }
      34             : 
      35             : /*
      36             :  * Running all eligible instructions in parallel creates
      37             :  * resource contention. This means we should implement
      38             :  * an admission control scheme where threads are temporarily
      39             :  * postponed if the claim for memory exceeds a threshold
      40             :  * In general such contentions will be hard to predict,
      41             :  * because they depend on the algorithm, the input sizes,
      42             :  * concurrent use of the same variables, and the output produced.
      43             :  *
      44             :  * One heuristic is based on calculating the storage footprint
      45             :  * of the operands and assuming it preferrably should fit in memory.
      46             :  * Ofcourse, there may be intermediate structures being
      47             :  * used and the size of the result is not a priori known.
      48             :  * For this, we use a high watermark on the amount of
      49             :  * physical memory we pre-allocate for the claims.
      50             :  *
      51             :  * Instructions are eligible to be executed when the
      52             :  * total footprint of all concurrent executions stays below
      53             :  * the high-watermark or it is the single expensive
      54             :  * instruction being started.
      55             :  *
      56             :  * When we run out of memory, the instruction is delayed.
      57             :  * How long depends on the other instructions to free up
      58             :  * resources. The current policy simple takes a local
      59             :  * decision by delaying the instruction based on its
      60             :  * claim of the memory.
      61             :  */
      62             : 
      63             : /*
      64             :  * The memory claim is the estimate for the amount of memory hold.
      65             :  * Views are consider cheap and ignored.
      66             :  * Given that auxiliary structures are important for performance,
      67             :  * we use their maximum as an indication of the memory footprint.
      68             :  * An alternative would be to focus solely on the base table cost.
      69             :  * (Good for a MSc study)
      70             :  */
      71             : lng
      72    12767338 : getMemoryClaim(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, int i, int flag)
      73             : {
      74    12767338 :         lng total = 0, itotal = 0, t;
      75    12767338 :         BAT *b;
      76             : 
      77    12767338 :         (void) mb;
      78    12767338 :         if (stk->stk[getArg(pci, i)].vtype == TYPE_bat) {
      79     6852593 :                 bat bid = stk->stk[getArg(pci, i)].val.bval;
      80     6852593 :                 if (!BBPcheck(bid))
      81             :                         return 0;
      82     6409525 :                 b = BBP_desc(bid);
      83     6409525 :                 if (b == NULL)
      84             :                         return 0;
      85     6409525 :                 MT_lock_set(&b->theaplock);
      86     6409240 :                 if (flag && isVIEW(b)) {
      87           0 :                         MT_lock_unset(&b->theaplock);
      88           0 :                         return 0;
      89             :                 }
      90             : 
      91             :                 /* calculate the basic scan size */
      92     6409240 :                 total += BATcount(b) << b->tshift;
      93     6409240 :                 total += heapinfo(b->tvheap, b->batCacheid);
      94     6409240 :                 MT_lock_unset(&b->theaplock);
      95             : 
      96             :                 /* indices should help, find their maximum footprint */
      97     6410349 :                 MT_rwlock_rdlock(&b->thashlock);
      98     6410506 :                 itotal = hashinfo(b->thash, d->batCacheid);
      99     6410506 :                 MT_rwlock_rdunlock(&b->thashlock);
     100     6410122 :                 t = IMPSimprintsize(b);
     101     6410084 :                 if (t > itotal)
     102             :                         itotal = t;
     103             :                 /* We should also consider the ordered index size */
     104    12820168 :                 t = b->torderidx
     105     6410084 :                                 && b->torderidx != (Heap *) 1 ? (lng) b->torderidx->free : 0;
     106     6410084 :                 if (t > itotal)
     107             :                         itotal = t;
     108             :                 //total = total > (lng)(MEMORY_THRESHOLD ) ? (lng)(MEMORY_THRESHOLD ) : total;
     109     6410084 :                 if (total < itotal)
     110             :                         total = itotal;
     111             :         }
     112             :         return total;
     113             : }
     114             : 
     115             : /*
     116             :  * The argclaim provides a hint on how much we actually may need to execute
     117             :  *
     118             :  * The client context also keeps bounds on the memory claim/client.
     119             :  * Surpassing this bound may be a reason to not admit the instruction to proceed.
     120             :  */
     121             : bool
     122     7779670 : MALadmission_claim(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci,
     123             :                                    lng argclaim)
     124             : {
     125     7779670 :         (void) pci;
     126             : 
     127             :         /* Check if we are allowed to allocate another worker thread for this client */
     128             :         /* It is somewhat tricky, because we may be in a dataflow recursion, each of which should be counted for.
     129             :          * A way out is to attach the thread count to the MAL stacks, which just limits the level
     130             :          * of parallism for a single dataflow graph.
     131             :          */
     132     7779670 :         if (cntxt->workerlimit > 0
     133           0 :                 && (int) ATOMIC_GET(&cntxt->workers) >= cntxt->workerlimit)
     134             :                 return false;
     135             : 
     136     7779670 :         if (argclaim == 0)
     137             :                 return true;
     138             : 
     139     3925111 :         MT_lock_set(&admissionLock);
     140             :         /* Determine if the total memory resource is exhausted, because it is overall limitation.  */
     141     3926766 :         if (memorypool <= 0) {
     142             :                 // we accidently released too much memory or need to initialize
     143         325 :                 memorypool = (lng) MEMORY_THRESHOLD;
     144             :         }
     145             : 
     146             :         /* the argument claim is based on the input for an instruction */
     147     3926766 :         if (memorypool > argclaim || ATOMIC_GET(&cntxt->workers) == 0) {
     148             :                 /* If we are low on memory resources, limit the user if he exceeds his memory budget
     149             :                  * but make sure there is at least one worker thread active */
     150     3926766 :                 if (cntxt->memorylimit) {
     151           0 :                         if (argclaim + stk->memory >
     152           0 :                                 (lng) cntxt->memorylimit * LL_CONSTANT(1048576)
     153           0 :                                 && ATOMIC_GET(&cntxt->workers) > 0) {
     154           0 :                                 MT_lock_unset(&admissionLock);
     155           0 :                                 return false;
     156             :                         }
     157           0 :                         stk->memory += argclaim;
     158             :                 }
     159     3926766 :                 memorypool -= argclaim;
     160     3926766 :                 stk->memory += argclaim;
     161     3926766 :                 MT_lock_set(&mal_delayLock);
     162     3926766 :                 if (mb->memory < stk->memory)
     163       36470 :                         mb->memory = stk->memory;
     164     3926766 :                 MT_lock_unset(&mal_delayLock);
     165     3926766 :                 MT_lock_unset(&admissionLock);
     166     3926766 :                 return true;
     167             :         }
     168           0 :         MT_lock_unset(&admissionLock);
     169           0 :         return false;
     170             : }
     171             : 
     172             : void
     173     7782130 : MALadmission_release(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci,
     174             :                                          lng argclaim)
     175             : {
     176             :         /* release memory claimed before */
     177     7782130 :         (void) cntxt;
     178     7782130 :         (void) mb;
     179     7782130 :         (void) pci;
     180     7782130 :         if (argclaim == 0)
     181             :                 return;
     182             : 
     183     3926571 :         MT_lock_set(&admissionLock);
     184     3926766 :         if (cntxt->memorylimit) {
     185           0 :                 stk->memory -= argclaim;
     186             :         }
     187     3926766 :         memorypool += argclaim;
     188     3926766 :         if (memorypool > (lng) MEMORY_THRESHOLD) {
     189           0 :                 memorypool = (lng) MEMORY_THRESHOLD;
     190             :         }
     191     3926766 :         stk->memory -= argclaim;
     192     3926766 :         MT_lock_unset(&admissionLock);
     193     3926766 :         return;
     194             : }