HPCC-16751 Prevent potential join crash on stop()

If an upstream activity called Join::stop() twice (which it shouldn't), it
caused a crash.
The particular case where this was spotted, was where a lightweight selfjoin
called it's input stop() twice.
It had to be a lightweight variety, where the code called through to the
input stop() directly and then again in the slave activity base class.

Signed-off-by: Jake Smith <jake.smith@lexisnexis.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;