Change memory size defaults + allow configurable

Allow the global pool to be configurable (per job) and allow it to be unbound,
IOW, allow it consume all available OS memory. Set to 0 to be unbound.
Allow 'largeMem' size to be configurable per job, and make it default to 75%
of global mem.
If global mem unset it will default to 75% of 2GB - this will change to
be automatically based on physical RAM and take into account slavesPerNode
in a future commit.

NB: largeMemSize setting was being completely ignored, due to a typo.

Signed-off-by: Jake Smith <jake.smith@lexisnexis.com>

ecl/hqlcpp/hqlresource.cpp

@@ -41,6 +41,7 @@
 //#define VERIFY_RESOURCING
 //#define SPOT_UNCONDITIONAL_CONDITIONS
 
+#define DEFAULT_LARGEMEM_BUFFER_SIZE (0x58000000) // ~ 1.4GB
 #define DEFAULT_MAX_SOCKETS 2000 // configurable by setting max_sockets in .ini
 #define DEFAULT_TOTAL_MEMORY ((1024*1024*1800)-DEFAULT_LARGEMEM_BUFFER_SIZE)
 #define FIXED_CLUSTER_SIZE 400

system/jlib/jmalloc.cpp

@@ -1440,7 +1440,7 @@ public:
         HeadOsBlockList.Prev = &HeadOsBlockList;
         HeadOsBlockList.Size = 0;
         OsTotal = 0;
-        OsMax = maxtotal;
+        OsMax = maxtotal ? maxtotal : ((memsize_t)-1); // unbound if not set
         OsMin = mintotal;
         if (_maxsubrecsize)
             suballocator.setown(new CSubAllocator(*this,_maxsubrecsize));

thorlcr/graph/thgraph.cpp

@@ -2338,6 +2338,21 @@ void CJobBase::init()
     maxActivityCores = (unsigned)getWorkUnitValueInt("maxActivityCores", globals->getPropInt("@maxActivityCores", 0)); // NB: 0 means system decides
     pausing = false;
     resumed = false;
+
+    unsigned gmemSize = getOptInt("@globalMemorySize"); // in MB
+    // NB: gmemSize is permitted to be unset, meaning unbound
+    initThorMemoryManager(gmemSize, getOptInt("@memTraceLevel", 1), getOptInt("@memoryStatsInterval", 60));
+
+    unsigned defaultMemMB = gmemSize;
+    if (!defaultMemMB)
+        defaultMemMB = 2048; // JCSMORE - should really be based on physical ram and take into account slavesPerNode.
+    defaultMemMB = defaultMemMB*3/4;
+    unsigned largeMemSize = getOptInt("@largeMemSize", defaultMemMB);
+    if (gmemSize && largeMemSize >= gmemSize)
+        throw MakeStringException(0, "largeMemSize(%d) can not exceed globalMemorySize(%d)", largeMemSize, gmemSize);
+    PROGLOG("Global memory size = %d MB, large mem size = %d MB", gmemSize, largeMemSize);
+    setLargeMemSize(largeMemSize);
+
     setLCRrowCRCchecking(0 != getWorkUnitValueInt("THOR_ROWCRC", globals->getPropBool("@THOR_ROWCRC", 
 #ifdef _DEBUG
         true
@@ -2539,6 +2554,18 @@ mptag_t CJobBase::deserializeMPTag(MemoryBuffer &mb)
     return tag;
 }
 
+unsigned CJobBase::getOptInt(const char *opt, unsigned dft)
+{
+    const char *wOpt = (opt&&(*opt)=='@') ? opt+1 : opt; // strip @ for options in workunit
+    return (unsigned)getWorkUnitValueInt(wOpt, globals->getPropInt(opt, dft));
+}
+
+__int64 CJobBase::getOptInt64(const char *opt, __int64 dft)
+{
+    const char *wOpt = (opt&&(*opt)=='@') ? opt+1 : opt; // strip @ for options in workunit
+    return getWorkUnitValueInt(opt, globals->getPropInt64(opt, dft));
+}
+
 // IGraphCallback
 void CJobBase::runSubgraph(CGraphBase &graph, size32_t parentExtractSz, const byte *parentExtract)
 {

thorlcr/graph/thgraph.hpp

@@ -888,6 +888,8 @@ public:
     mptag_t allocateMPTag();
     void freeMPTag(mptag_t tag);
     mptag_t deserializeMPTag(MemoryBuffer &mb);
+    unsigned getOptInt(const char *opt, unsigned dft=0);
+    __int64 getOptInt64(const char *opt, __int64 dft=0);
 
     virtual void abort(IException *e);
     virtual IBarrier *createBarrier(mptag_t tag) { UNIMPLEMENTED; return NULL; }

thorlcr/master/thmastermain.cpp

@@ -684,15 +684,6 @@ int main( int argc, char *argv[]  )
         Owned<CRegistryServer> registry = new CRegistryServer();
         StringBuffer thorEpStr;
         LOG(MCdebugProgress, thorJob, "ThorMaster version %d.%d, Started on %s", THOR_VERSION_MAJOR,THOR_VERSION_MINOR,thorEp.getUrlStr(thorEpStr).toCharArray());
-
-        unsigned gmemsize = globals->getPropInt("@masterGlobalMemorySize"); // in MB
-        if (gmemsize==0) {
-            gmemsize = globals->getPropInt("@globalMemorySize"); // in MB
-            if (gmemsize==0)
-                gmemsize = 2048;
-        }
-        initThorMemoryManager(gmemsize,globals->getPropInt("@memTraceLevel", 1),globals->getPropInt("@memoryStatsInterval", 60));
-
         LOG(MCdebugProgress, thorJob, "Thor name = %s, queue = %s, nodeGroup = %s",thorname,queueName.str(),nodeGroup.str());
 
         serverStatus.queryProperties()->setProp("@thorname", thorname);

thorlcr/slave/thslavemain.cpp

@@ -356,10 +356,6 @@ int main( int argc, char *argv[]  )
                 tempdir = tempdirstr.str();
             SetTempDir(tempdir,true);
 
-            unsigned gmemsize = globals->getPropInt("@globalMemorySize"); // in MB
-            if (gmemsize==0) 
-                gmemsize = 2048;
-            initThorMemoryManager(gmemsize,globals->getPropInt("@memTraceLevel", 1),globals->getPropInt("@memoryStatsInterval", 60));
             useMemoryMappedRead(globals->getPropBool("@useMemoryMappedRead"));
 
             LOG(MCdebugProgress, thorJob, "ThorSlave Version LCR - %d.%d started",THOR_VERSION_MAJOR,THOR_VERSION_MINOR);

thorlcr/thorutil/thbufdef.hpp

@@ -20,7 +20,6 @@
 #define __THBUFDEF__
 
 
-#define DEFAULT_LARGEMEM_BUFFER_SIZE                    (0x58000000)            // ~ 1.4GB  
 #define SMALL_SMART_BUFFER_SIZE                 (0x100000)              // 1MB
 #define PULL_SMART_BUFFER_SIZE                  (0x100000*8*3)          // 24MB
 #define CHOOSESETS_SMART_BUFFER_SIZE            (0x100000*8*3)          // 12MB

thorlcr/thorutil/thmem.cpp

@@ -147,8 +147,39 @@ extern graph_decl void setMultiThorMemoryNotify(size32_t size,ILargeMemLimitNoti
 }
 
 
+static memsize_t largeMemSize = 0;
+memsize_t setLargeMemSize(unsigned limitMB)
+{
+    memsize_t prevLargeMemSize = largeMemSize;
+    largeMemSize = 1024*1024*(memsize_t)limitMB;
+    return prevLargeMemSize;
+}
+
+memsize_t queryLargeMemSize()
+{
+    if (0 == largeMemSize)
+        throwUnexpected();
+    return largeMemSize;
+}
 
 
+CThorRowArray::CThorRowArray()
+{
+    numelem = 0;
+    totalsize = 0;
+    overhead = 0;
+    sizing = false;
+    raiseexceptions = false;
+    memsize_t tmp = queryLargeMemSize();
+
+    if (tmp>0xffffffff)
+        maxtotal = 0xffffffff;
+    else
+        maxtotal = (unsigned)tmp;
+    if (maxtotal<0x100000)
+        maxtotal = 0x100000;
+}
+
 void CThorRowArray::adjSize(const void *row, bool inc)
 {
     if (!row)
@@ -721,7 +752,7 @@ void resetThorMemoryManager()
     ThorAllocatorCache.reset(); // clears cached rows
     if (ThorMemoryManager) {
         ThorMemoryManager->Release();
-        ThorMemoryManager = createThorRowManager(ThorMemoryManagerMaxSize, &ThorAllocatorCache, false);
+        ThorMemoryManager = NULL;
     }
     ThorAllocatorCache.clear(); // do after so that act ids still around
 }

thorlcr/thorutil/thmem.hpp

@@ -211,6 +211,8 @@ interface IThorRowArrayException: extends IException
 extern graph_decl void checkMultiThorMemoryThreshold(bool inc);
 extern graph_decl void setMultiThorMemoryNotify(size32_t size,ILargeMemLimitNotify *notify);
 
+extern graph_decl memsize_t setLargeMemSize(unsigned limit);
+
 class graph_decl CThorRowArray
 {
     MemoryBuffer ptrbuf;
@@ -227,23 +229,7 @@ class graph_decl CThorRowArray
 
 
 public:
-    CThorRowArray()
-    {
-        numelem = 0;
-        totalsize = 0;
-        overhead = 0;
-        sizing = false;
-        raiseexceptions = false;
-        memsize_t tmp = ((unsigned __int64)ThorRowMemoryAvailable())*7/8;   // don't fill up completely
-
-        if (tmp>0xffffffff)
-            maxtotal = 0xffffffff;
-        else
-            maxtotal = (unsigned)tmp;
-        if (maxtotal<0x100000)
-            maxtotal = 0x100000;
-    }
-
+    CThorRowArray();
 
     ~CThorRowArray()
     {
@@ -302,7 +288,7 @@ public:
             return NULL;
         byte ** rp = ((byte **)ptrbuf.toByteArray())+idx;
         const byte *ret = *rp;
-        if (sizing) 
+        if (sizing)
             adjSize(ret,false);
         *rp = NULL;
         return ret;

thorlcr/thorutil/thormisc.cpp

@@ -812,20 +812,6 @@ void reportExceptionToWorkunit(IConstWorkUnit &workunit,IException *e, WUExcepti
     }
 } 
 
-static memsize_t largeMemSize = 0;
-memsize_t queryLargeMemSize()
-{
-    if (!largeMemSize)
-    {
-        if (globals->hasProp("largeMemSize"))
-            largeMemSize = globals->getPropInt("@largeMemSize") * 0x100000;
-        else
-            largeMemSize = DEFAULT_LARGEMEM_BUFFER_SIZE;
-        PROGLOG("Setting largemem to %"I64F"d", (unsigned __int64) largeMemSize);
-    }
-    return largeMemSize;
-}
-
 StringBuffer &getCompoundQueryName(StringBuffer &compoundName, const char *queryName, unsigned version)
 {
     return compoundName.append('V').append(version).append('_').append(queryName);