HPCC-16387 Fix issues with stopping arms of conditionals

Previous fix did not cover some cases.
If a condition was used (and branch selected), but then on
another call to the child query containing the conditional, it
was not used (because of e.g. another downstream conditional),
then the unused conditional would not stop all it's inputs.
This could cause issues when activities restarted on the next
child query execution, e.g. stranded projects could hit
assert(active==0), because they had not been stopped correctly
on the last child query execution.

This fix resolves by:
1) Clearing the branch in the conditional stop() method.

2) Clearing hasStarted() flag on stop()
NB: This change also effects other activities which
tested hasStarted() to perform conditional action in stop().
i.e. previously they might have been calling code that they
shouldn't have done on multiple calls to a child query,
because the hasStarted flag was not cleared in stop().

3) Allow dataLinkIncrement to be called regardless of stop/start
flag. Previously, it asserted that isStarted(), which is really
correct. However many activities call stop() early then return
records and call dataLinkIncrement.
This should changed in a future PR, so that activities do not
call stop() early (i.e. do not mark stopped flat) and instead
only call stopInput and ensure stop is called when all records
have been returned.

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

thorlcr/activities/loop/thloopslave.cpp

@@ -831,6 +831,7 @@ public:
             stopInput(branch);
         selectedInputStream = NULL;
         abortSoon = true;
+        branch = (unsigned)-1;
         PARENT::stop();
     }
     CATCH_NEXTROW()

thorlcr/graph/thgraphslave.hpp

@@ -37,6 +37,9 @@
 #include "traceslave.hpp"
 #include "thorstrand.hpp"
 
+// #define OUTPUT_RECORDSIZE // causes the record size to be logged for each activity on the 1st call to dataLinkIncrement
+
+
 interface IStartableEngineRowStream : extends IEngineRowStream
 {
     virtual void start() = 0;
@@ -73,13 +76,12 @@ public:
         assertex(!hasStarted() || hasStopped());      // ITDL started twice
 #endif
         icount = 0;
-        rowcount_t prevCount = count & THORDATALINK_COUNT_MASK;
-        count = prevCount | THORDATALINK_STARTED;
+        count = (count & THORDATALINK_COUNT_MASK) | THORDATALINK_STARTED;
     }
 
     inline void dataLinkStop()
     {
-        count |= THORDATALINK_STOPPED;
+        count = (count & THORDATALINK_COUNT_MASK) | THORDATALINK_STOPPED;
 #ifdef _TESTING
         owner.ActPrintLog("ITDL output %d stopped, count was %" RCPF "d", outputId, getDataLinkCount());
 #endif
@@ -87,8 +89,6 @@ public:
     inline void dataLinkIncrement() { dataLinkIncrement(1); }
     inline void dataLinkIncrement(rowcount_t v)
     {
-#ifdef _TESTING
-        assertex(hasStarted());
 #ifdef OUTPUT_RECORDSIZE
         if (count==THORDATALINK_STARTED)
         {
@@ -96,7 +96,6 @@ public:
             parent.ActPrintLog("Record size %s= %d", parent.queryRowMetaData(this)->isVariableSize()?"(min) ":"",rsz);
         }
 #endif
-#endif
         icount += v;
         count += v;
     }