Merge pull request #9413 from jakesmith/hpcc-16751

HPCC-16751 Prevent potential join crash on stop()

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

thorlcr/activities/fetch/thfetchslave.cpp

@@ -463,7 +463,7 @@ public:
     }
     virtual void stop() override
     {
-        if (queryInputStarted(0))
+        if (hasStarted())
             fetchStreamOut->stop();
         dataLinkStop();
     }

thorlcr/activities/join/thjoinslave.cpp

@@ -373,7 +373,7 @@ public:
          * stop()'s are chained like this, so that upstream splitters can be stopped as quickly as possible
          * in order to reduce buffering.
          */
-        if (queryInputStarted(0))
+        if (hasStarted())
         {
             lhsProgressCount = joinhelper->getLhsProgress();
             rhsProgressCount = joinhelper->getRhsProgress();

thorlcr/activities/msort/thmsortslave.cpp

@@ -155,14 +155,11 @@ public:
             output->stop();
             output.clear();
         }
-        if (queryInputStarted(0))
+        if (hasStarted())
         {
             ActPrintLog("SORT waiting barrier.2");
             barrier->wait(false);
             ActPrintLog("SORT barrier.2 raised");
-        }
-        if (queryInputStarted(0))
-        {
             ActPrintLog("SORT waiting for merge");
             sorter->stopMerge();
         }

thorlcr/activities/selfjoin/thselfjoinslave.cpp

@@ -203,7 +203,8 @@ public:
             }
             if (strm)
             {
-                strm->stop();
+                if (!isLightweight) // if lightweight strm=input and PARENT::stop handles input stop
+                    strm->stop();
                 strm.clear();
             }
         }

thorlcr/activities/when/thwhenslave.cpp

@@ -84,8 +84,9 @@ public:
     }
     virtual void stop() override
     {
+        bool started = hasStarted();
         PARENT::stop();
-        if (queryInputStarted(0))
+        if (started)
         {
             if (!executeDependencies(abortSoon ? WhenFailureId : WhenSuccessId))
                 abortSoon = true;