Merge remote-tracking branch 'origin/candidate-3.8.x' into candidate-3.10.0

Conflicts:
	thorlcr/graph/thgraph.cpp

Signed-off-by: Richard Chapman <rchapman@hpccsystems.com>

common/fileview2/fvtransform.cpp

@@ -296,6 +296,7 @@ void ViewTransformerRegistry::addPlugins(const char * name)
     {
         IHqlScope * scope = &scopes.item(i);
         HqlExprArray symbols;
+        scope->ensureSymbolsDefined(ctx);
         scope->getSymbols(symbols);
 
         ForEachItemIn(j, symbols)

dali/base/dadfs.cpp

@@ -51,7 +51,11 @@
 #define DFSSERVER_THROTTLE_COUNT 20
 #define DFSSERVER_THROTTLE_TIME 1000
 
+#if _INTERNAL_EDITION == 1
+#warning Disabling Sub-file compatibility checking
+#else
 #define SUBFILE_COMPATIBILITY_CHECKING
+#endif
 
 //#define PACK_ECL
 

thorlcr/activities/csvread/thcsvread.cpp

@@ -52,8 +52,12 @@ public:
         if (headerLines)
         {
             dst.append((int)mpTag);
-            ISuperFileDescriptor *superFDesc = fileDesc->querySuperFileDescriptor();
-            unsigned subFiles = superFDesc ? superFDesc->querySubFiles() : 1;
+            unsigned subFiles = 0;
+            if (fileDesc)
+            {
+                ISuperFileDescriptor *superFDesc = fileDesc->querySuperFileDescriptor();
+                subFiles = superFDesc ? superFDesc->querySubFiles() : 1;
+            }
             dst.append(subFiles);
         }
     }

thorlcr/activities/enth/thenthslave.cpp

@@ -196,6 +196,23 @@ class CEnthSlaveActivity : public BaseEnthActivity
         msg.append(count);
         container.queryJob().queryJobComm().send(msg, container.queryJob().queryMyRank()+1, mpTag);
     }
+    bool getPrev()
+    {
+        if (!firstNode()) // no need if 1st node
+        {
+            CMessageBuffer msg;
+            if (!receiveMsg(msg, container.queryJob().queryMyRank()-1, mpTag))
+                return false;
+            msg.read(prevRecCount);
+        }
+        setInitialCounter(prevRecCount);
+        if (haveLocalCount()) // if local total count known, send total now
+            sendCount(prevRecCount + localRecCount);
+        else
+            prevRecCountSem.signal();
+        return true;
+    }
+
 public:
     CEnthSlaveActivity(CGraphElementBase *container) : BaseEnthActivity(container)
     { 
@@ -225,18 +242,8 @@ public:
         if (first)
         {
             first = false;
-            if (!firstNode()) // no need if 1st node
-            {
-                CMessageBuffer msg;
-                if (!receiveMsg(msg, container.queryJob().queryMyRank()-1, mpTag))
-                    return NULL;
-                msg.read(prevRecCount);
-            }
-            setInitialCounter(prevRecCount);
-            if (haveLocalCount()) // if local total count known, send total now
-                sendCount(prevRecCount + localRecCount);
-            else
-                prevRecCountSem.signal();
+            if (!getPrev())
+                return NULL;
         }
         while (!abortSoon)
         {
@@ -251,6 +258,16 @@ public:
         }
         return NULL;        
     }
+    virtual void stop()
+    {
+        // Need to ensure sequence continues, in nextRow has never been called.
+        if (first)
+        {
+            first = false;
+            getPrev();
+        }
+        BaseEnthActivity::stop();
+    }
     virtual void onInputFinished(rowcount_t localRecCount)
     {
         if (!haveLocalCount())

thorlcr/activities/thactivityutil.cpp

@@ -87,11 +87,16 @@ class ThorLookaheadCache: public IThorDataLink, public CSimpleInterface
     } thread;
 
 public:
-    
-
-
     IMPLEMENT_IINTERFACE_USING(CSimpleInterface);
 
+    void doNotify()
+    {
+        if (notify)
+            notify->onInputFinished(count);
+        if (smartbuf)
+            smartbuf->queryWriter()->flush();
+    }
+
     int run()
     {
         if (!started) {
@@ -161,13 +166,42 @@ public:
         {
             ActPrintLog(&activity, e, "ThorLookaheadCache get exception");
             getexception.setown(e);
-        }   
-        if (notify)
-            notify->onInputFinished(count);
-        if (smartbuf)
-            smartbuf->queryWriter()->flush();
+        }
+
+        // notify and flush async, as these can block, but we do not want to block in->stop()
+        // especially if this is a spilling read ahead, where use case scenarios include not wanting to
+        // block the upstream input.
+        // An example is a firstn which if stop() it not called, it may block
+        // other nodes from pulling because it is blocked upstream on full buffers (which can be discarded
+        // on stop()), and those in turn are blocking other arms of the graph.
+        class CNotifyThread : implements IThreaded
+        {
+            CThreaded threaded;
+            ThorLookaheadCache &owner;
+        public:
+            CNotifyThread(ThorLookaheadCache &_owner) : threaded("Lookahead-CNotifyThread"), owner(_owner)
+            {
+                threaded.init(this);
+            }
+            ~CNotifyThread()
+            {
+                loop
+                {
+                    if (threaded.join(60000))
+                        break;
+                    PROGLOG("Still waiting on lookahead CNotifyThread thread to complete");
+                }
+            }
+        // IThreaded impl.
+            virtual void main()
+            {
+                owner.doNotify();
+            }
+        } notifyThread(*this);
+
         running = false;
-        try {
+        try
+        {
             if (in)
                 in->stop();
         }
@@ -176,7 +210,8 @@ public:
             ActPrintLog(&activity, e, "ThorLookaheadCache stop exception");
             if (!getexception.get())
                 getexception.setown(e);
-        }   
+        }
+        // NB: Will wait on CNotifyThread to finish before returning
         return 0;
     }
         
@@ -249,8 +284,7 @@ public:
         }
         return row.getClear();
     }
-                    
-    
+
     bool isGrouped() { return false; }
             
     void getMetaInfo(ThorDataLinkMetaInfo &info)