Merge pull request #4839 from richardkchapman/checkingHeap

HPCC-9489 Add support to roxie for a checking memory allocator

Reviewed-by: Gavin Halliday <ghalliday@hpccsystems.com>

common/thorhelper/roxierow.cpp

@@ -36,7 +36,7 @@ public:
 };
 
 //NOTE: If a row requires checking then the row will also have the bit set to indicate it requires a destructor
-//so that rows are checked on destructon.
+//so that rows are checked on destruction.
 //Therefore checking if the destructor is set for a row in isValid() to protect us from uninitialised crcs.
 class Crc16CheckingHelper
 {
@@ -483,7 +483,7 @@ public:
         }
         if (!RoxieRowCheckValid(cacheId, row))
         {
-            //MORE: Give an error, but don't throw an exception!
+            throw MakeStringException(0, "ERROR: crc check failure destroying row!");
         }
         allocator->queryOutputMeta()->destruct((byte *) row);
     }
@@ -503,7 +503,7 @@ public:
         }
         if (!RoxieRowCheckValid(cacheId, row))
         {
-            //MORE: Give an error, but don't throw an exception!
+            throw MakeStringException(0, "ERROR: crc check failure cloning row!");
         }
         //This should only be called if the destructor needs to be called - so don't bother checking
         ChildRowLinkerWalker walker;
@@ -513,7 +513,7 @@ public:
     {
         if (!RoxieRowCheckValid(cacheId, row))
         {
-            //MORE: Throw an exception?
+            throw MakeStringException(0, "ERROR: crc check failure checking row!");
         }
     }
 };

roxie/ccd/ccd.hpp

@@ -423,6 +423,7 @@ extern unsigned defaultFetchPreload;
 extern unsigned defaultFullKeyedJoinPreload;
 extern unsigned defaultKeyedJoinPreload;
 extern unsigned defaultPrefetchProjectPreload;
+extern bool defaultCheckingHeap;
 
 extern StringBuffer logDirectory;
 extern StringBuffer pluginDirectory;
@@ -456,6 +457,7 @@ extern void saveTopology();
 #define LOGGING_DEBUGGERACTIVE  0x04
 #define LOGGING_BLIND           0x08
 #define LOGGING_TRACELEVELSET   0x10
+#define LOGGING_CHECKINGHEAP    0x20
 #define LOGGING_FLAGSPRESENT    0x40
 
 class LogItem : public CInterface
@@ -956,6 +958,7 @@ class SlaveContextLogger : public StringContextLogger
     mutable bool anyOutput;
     bool traceActivityTimes;
     bool debuggerActive;
+    bool checkingHeap;
     IpAddress ip;
     StringAttr wuid;
 public:
@@ -965,6 +968,7 @@ public:
     virtual void flush(bool closing, bool aborted) const;
     inline bool queryTraceActivityTimes() const { return traceActivityTimes; }
     inline bool queryDebuggerActive() const { return debuggerActive; }
+    inline bool queryCheckingHeap() const { return checkingHeap; }
     inline void setDebuggerActive(bool _active) { debuggerActive = _active; }
     inline const StatsCollector &queryStats() const 
     {

roxie/ccd/ccdcontext.cpp

@@ -604,6 +604,7 @@ protected:
     unsigned ctxFetchPreload;
     unsigned ctxPrefetchProjectPreload;
     bool traceActivityTimes;
+    bool checkingHeap;
 
     Owned<IConstWorkUnit> workUnit;
     Owned<IRoxieDaliHelper> daliHelperLink;
@@ -669,7 +670,7 @@ protected:
 
 public:
     IMPLEMENT_IINTERFACE;
-    CSlaveContext(const IQueryFactory *_factory, const IRoxieContextLogger &_logctx, unsigned _timeLimit, memsize_t _memoryLimit, IRoxieQueryPacket *_packet, bool _traceActivityTimes, bool _debuggerActive)
+    CSlaveContext(const IQueryFactory *_factory, const IRoxieContextLogger &_logctx, unsigned _timeLimit, memsize_t _memoryLimit, IRoxieQueryPacket *_packet, bool _traceActivityTimes, bool _debuggerActive, bool _checkingHeap)
         : factory(_factory), logctx(_logctx)
     {
         if (_packet)
@@ -696,6 +697,7 @@ public:
             debugContext.setown(slaveDebugContext);
             probeManager.setown(createDebugManager(debugContext, "slaveDebugger"));
         }
+        checkingHeap = _checkingHeap;
 
         aborted = false;
 
@@ -838,6 +840,11 @@ public:
         return traceActivityTimes;
     }
 
+    virtual bool queryCheckingHeap() const
+    {
+        return checkingHeap;
+    }
+
     virtual void checkAbort()
     {
         // MORE - really should try to apply limits at slave end too
@@ -1244,7 +1251,10 @@ public:
 // roxiemem::IRowAllocatorMetaActIdCacheCallback
     virtual IEngineRowAllocator *createAllocator(IOutputMetaData *meta, unsigned activityId, unsigned id, roxiemem::RoxieHeapFlags flags) const
     {
-        return createRoxieRowAllocator(*rowManager, meta, activityId, id, flags);
+        if (checkingHeap)
+            return createCrcRoxieRowAllocator(*rowManager, meta, activityId, id, flags);
+        else
+            return createRoxieRowAllocator(*rowManager, meta, activityId, id, flags);
     }
 
     virtual void getResultRowset(size32_t & tcount, byte * * & tgt, const char * stepname, unsigned sequence, IEngineRowAllocator * _rowAllocator, bool isGrouped, IXmlToRowTransformer * xmlTransformer, ICsvToRowTransformer * csvTransformer)
@@ -1663,7 +1673,7 @@ protected:
 
 IRoxieSlaveContext *createSlaveContext(const IQueryFactory *_factory, const SlaveContextLogger &_logctx, unsigned _timeLimit, memsize_t _memoryLimit, IRoxieQueryPacket *packet)
 {
-    return new CSlaveContext(_factory, _logctx, _timeLimit, _memoryLimit, packet, _logctx.queryTraceActivityTimes(), _logctx.queryDebuggerActive());
+    return new CSlaveContext(_factory, _logctx, _timeLimit, _memoryLimit, packet, _logctx.queryTraceActivityTimes(), _logctx.queryDebuggerActive(), _logctx.queryCheckingHeap());
 }
 
 class CRoxieServerDebugContext : extends CBaseServerDebugContext
@@ -1943,7 +1953,7 @@ public:
     IMPLEMENT_IINTERFACE;
 
     CRoxieServerContext(const IQueryFactory *_factory, const IRoxieContextLogger &_logctx)
-        : CSlaveContext(_factory, _logctx, 0, 0, NULL, false, false), serverQueryFactory(_factory)
+        : CSlaveContext(_factory, _logctx, 0, 0, NULL, false, false, false), serverQueryFactory(_factory)
     {
         init();
         rowManager->setMemoryLimit(serverQueryFactory->getMemoryLimit());
@@ -1952,7 +1962,7 @@ public:
     }
 
     CRoxieServerContext(IConstWorkUnit *_workUnit, const IQueryFactory *_factory, const IRoxieContextLogger &_logctx)
-        : CSlaveContext(_factory, _logctx, 0, 0, NULL, false, false), serverQueryFactory(_factory)
+        : CSlaveContext(_factory, _logctx, 0, 0, NULL, false, false, false), serverQueryFactory(_factory)
     {
         init();
         workUnit.set(_workUnit);
@@ -1963,7 +1973,7 @@ public:
     }
 
     CRoxieServerContext(IPropertyTree *_context, const IQueryFactory *_factory, SafeSocket &_client, TextMarkupFormat _mlFmt, bool _isRaw, bool _isBlocked, HttpHelper &httpHelper, bool _trim, unsigned _priority, const IRoxieContextLogger &_logctx, PTreeReaderOptions _xmlReadFlags)
-        : CSlaveContext(_factory, _logctx, 0, 0, NULL, false, false), serverQueryFactory(_factory)
+        : CSlaveContext(_factory, _logctx, 0, 0, NULL, false, false, false), serverQueryFactory(_factory)
     {
         init();
         context.set(_context);
@@ -2013,6 +2023,7 @@ public:
         ctxPrefetchProjectPreload = context->getPropInt("_PrefetchProjectPreload", defaultPrefetchProjectPreload);
 
         traceActivityTimes = context->getPropBool("_TraceActivityTimes", false) || context->getPropBool("@timing", false);
+        checkingHeap = context->getPropBool("_CheckingHeap", defaultCheckingHeap) || context->getPropBool("@checkingHeap", defaultCheckingHeap);
     }
 
     virtual roxiemem::IRowManager &queryRowManager()

roxie/ccd/ccdcontext.hpp

@@ -65,6 +65,7 @@ interface IRoxieSlaveContext : extends IRoxieContextLogger
     virtual void noteProcessed(const IRoxieContextLogger &_activityContext, const IRoxieServerActivity *activity, unsigned _idx, unsigned _processed, unsigned __int64 _totalCycles, unsigned __int64 _localCycles) const = 0;
     virtual IProbeManager *queryProbeManager() const = 0;
     virtual bool queryTimeActivities() const = 0;
+    virtual bool queryCheckingHeap() const = 0;
     virtual IDebuggableContext *queryDebugContext() const = 0;
     virtual void printResults(IXmlWriter *output, const char *name, unsigned sequence) = 0;
     virtual void setWUState(WUState state) = 0;

roxie/ccd/ccdmain.cpp

@@ -125,6 +125,7 @@ unsigned defaultFetchPreload = 0;
 unsigned defaultFullKeyedJoinPreload = 0;
 unsigned defaultKeyedJoinPreload = 0;
 unsigned dafilesrvLookupTimeout = 10000;
+bool defaultCheckingHeap = false;
 
 unsigned logQueueLen;
 unsigned logQueueDrop;
@@ -712,6 +713,7 @@ int STARTQUERY_API start_query(int argc, const char *argv[])
         defaultFullKeyedJoinPreload = topology->getPropInt("@defaultFullKeyedJoinPreload", 0);
         defaultKeyedJoinPreload = topology->getPropInt("@defaultKeyedJoinPreload", 0);
         defaultPrefetchProjectPreload = topology->getPropInt("@defaultPrefetchProjectPreload", 10);
+        defaultCheckingHeap = topology->getPropInt("@checkingHeap", false);  // NOTE - not in configmgr - too dangerous!
         diskReadBufferSize = topology->getPropInt("@diskReadBufferSize", 0x10000);
         fieldTranslationEnabled = topology->getPropBool("@fieldTranslationEnabled", false);
 

roxie/ccd/ccdqueue.cpp

@@ -465,6 +465,7 @@ void SlaveContextLogger::set(IRoxieQueryPacket *packet)
     anyOutput = false;
     intercept = false;
     debuggerActive = false;
+    checkingHeap = false;
     traceActivityTimes = false;
     start = msTick();
     if (packet)
@@ -489,6 +490,8 @@ void SlaveContextLogger::set(IRoxieQueryPacket *packet)
                 traceActivityTimes = true;
             if (loggingFlags & LOGGING_BLIND)
                 blind = true;
+            if (loggingFlags & LOGGING_CHECKINGHEAP)
+                checkingHeap = true;
             if (loggingFlags & LOGGING_DEBUGGERACTIVE)
             {
                 assertex(traceLength > sizeof(unsigned short));

roxie/ccd/ccdserver.cpp

@@ -363,6 +363,10 @@ public:
     {
         return ctx->queryTimeActivities();
     }
+    virtual bool queryCheckingHeap() const
+    {
+        return ctx->queryCheckingHeap();
+    }
     virtual void printResults(IXmlWriter *output, const char *name, unsigned sequence)
     {
         ctx->printResults(output, name, sequence);
@@ -3405,6 +3409,8 @@ private:
                     loggingFlags |= LOGGING_TIMEACTIVITIES; 
                 if (activity.queryLogCtx().isBlind())
                     loggingFlags |= LOGGING_BLIND;
+                if (ctx->queryCheckingHeap())
+                    loggingFlags |= LOGGING_CHECKINGHEAP;
                 if (debugContext)
                 {
                     loggingFlags |= LOGGING_DEBUGGERACTIVE;

roxie/ccd/ccdstate.cpp

@@ -1580,6 +1580,11 @@ private:
                 checkCompleted = control->getPropBool("@val", true);
                 topology->setPropBool("@checkCompleted", checkCompleted );
             }
+            else if (stricmp(queryName, "control:checkingHeap")==0)
+            {
+                defaultCheckingHeap = control->getPropBool("@val", true);
+                topology->setPropInt("@checkingHeap", defaultCheckingHeap);
+            }
             else if (stricmp(queryName, "control:clearIndexCache")==0)
             {
                 bool clearAll = control->getPropBool("@clearAll", true);