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         339 : mal_resource_reset(void)
      29             : {
      30         339 :         MT_lock_set(&admissionLock);
      31         339 :         memorypool = (lng) MEMORY_THRESHOLD;
      32         339 :         MT_lock_unset(&admissionLock);
      33         339 : }
      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    17308673 : getMemoryClaim(MalBlkPtr mb, MalStkPtr stk, InstrPtr pci, int i, int flag)
      73             : {
      74    17308673 :         lng total = 0, itotal = 0, t;
      75    17308673 :         BAT *b;
      76             : 
      77    17308673 :         (void) mb;
      78    17308673 :         if (stk->stk[getArg(pci, i)].bat) {
      79    10338553 :                 bat bid = stk->stk[getArg(pci, i)].val.bval;
      80    10338553 :                 if (!BBPcheck(bid))
      81             :                         return 0;
      82     9857215 :                 b = BBP_desc(bid);
      83     9857215 :                 MT_lock_set(&b->theaplock);
      84     9854263 :                 if (flag && isVIEW(b)) {
      85           0 :                         MT_lock_unset(&b->theaplock);
      86           0 :                         return 0;
      87             :                 }
      88             : 
      89             :                 /* calculate the basic scan size */
      90     9854263 :                 total += BATcount(b) << b->tshift;
      91     9854263 :                 total += heapinfo(b->tvheap, b->batCacheid);
      92     9854263 :                 MT_lock_unset(&b->theaplock);
      93             : 
      94             :                 /* indices should help, find their maximum footprint */
      95     9860998 :                 MT_rwlock_rdlock(&b->thashlock);
      96     9861548 :                 itotal = hashinfo(b->thash, d->batCacheid);
      97     9861548 :                 MT_rwlock_rdunlock(&b->thashlock);
      98     9860776 :                 t = IMPSimprintsize(b);
      99     9858799 :                 if (t > itotal)
     100             :                         itotal = t;
     101             :                 /* We should also consider the ordered index size */
     102    19717598 :                 t = b->torderidx
     103     9858799 :                                 && b->torderidx != (Heap *) 1 ? (lng) b->torderidx->free : 0;
     104     9858799 :                 if (t > itotal)
     105             :                         itotal = t;
     106             :                 //total = total > (lng)(MEMORY_THRESHOLD ) ? (lng)(MEMORY_THRESHOLD ) : total;
     107     9858799 :                 if (total < itotal)
     108             :                         total = itotal;
     109             :         }
     110             :         return total;
     111             : }
     112             : 
     113             : /*
     114             :  * The argclaim provides a hint on how much we actually may need to execute
     115             :  *
     116             :  * The client context also keeps bounds on the memory claim/client.
     117             :  * Surpassing this bound may be a reason to not admit the instruction to proceed.
     118             :  */
     119             : bool
     120    11580452 : MALadmission_claim(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci,
     121             :                                    lng argclaim)
     122             : {
     123    11580452 :         (void) pci;
     124             : 
     125             :         /* Check if we are allowed to allocate another worker thread for this client */
     126             :         /* It is somewhat tricky, because we may be in a dataflow recursion, each of which should be counted for.
     127             :          * A way out is to attach the thread count to the MAL stacks, which just limits the level
     128             :          * of parallism for a single dataflow graph.
     129             :          */
     130    11580452 :         if (cntxt->workerlimit > 0
     131           0 :                 && (int) ATOMIC_GET(&cntxt->workers) >= cntxt->workerlimit)
     132             :                 return false;
     133             : 
     134    11580452 :         if (argclaim == 0)
     135             :                 return true;
     136             : 
     137     5610872 :         MT_lock_set(&admissionLock);
     138             :         /* Determine if the total memory resource is exhausted, because it is overall limitation.  */
     139     5625000 :         if (memorypool <= 0) {
     140             :                 // we accidently released too much memory or need to initialize
     141         330 :                 memorypool = (lng) MEMORY_THRESHOLD;
     142             :         }
     143             : 
     144             :         /* the argument claim is based on the input for an instruction */
     145     5625000 :         if (memorypool > argclaim || ATOMIC_GET(&cntxt->workers) == 0) {
     146             :                 /* If we are low on memory resources, limit the user if he exceeds his memory budget
     147             :                  * but make sure there is at least one worker thread active */
     148     5625000 :                 if (cntxt->memorylimit) {
     149           0 :                         if (argclaim + stk->memory >
     150           0 :                                 (lng) cntxt->memorylimit * LL_CONSTANT(1048576)
     151           0 :                                 && ATOMIC_GET(&cntxt->workers) > 0) {
     152           0 :                                 MT_lock_unset(&admissionLock);
     153           0 :                                 return false;
     154             :                         }
     155           0 :                         stk->memory += argclaim;
     156             :                 }
     157     5625000 :                 memorypool -= argclaim;
     158     5625000 :                 stk->memory += argclaim;
     159     5625000 :                 MT_lock_set(&mal_delayLock);
     160     5625000 :                 if (mb->memory < stk->memory)
     161       46768 :                         mb->memory = stk->memory;
     162     5625000 :                 MT_lock_unset(&mal_delayLock);
     163     5625000 :                 MT_lock_unset(&admissionLock);
     164     5625000 :                 return true;
     165             :         }
     166           0 :         MT_lock_unset(&admissionLock);
     167           0 :         return false;
     168             : }
     169             : 
     170             : void
     171    11534188 : MALadmission_release(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci,
     172             :                                          lng argclaim)
     173             : {
     174             :         /* release memory claimed before */
     175    11534188 :         (void) cntxt;
     176    11534188 :         (void) mb;
     177    11534188 :         (void) pci;
     178    11534188 :         if (argclaim == 0)
     179             :                 return;
     180             : 
     181     5615472 :         MT_lock_set(&admissionLock);
     182     5625000 :         if (cntxt->memorylimit) {
     183           0 :                 stk->memory -= argclaim;
     184             :         }
     185     5625000 :         memorypool += argclaim;
     186     5625000 :         if (memorypool > (lng) MEMORY_THRESHOLD) {
     187           0 :                 memorypool = (lng) MEMORY_THRESHOLD;
     188             :         }
     189     5625000 :         stk->memory -= argclaim;
     190     5625000 :         MT_lock_unset(&admissionLock);
     191     5625000 :         return;
     192             : }