Merge branch 'candidate-7.8.x' into candidate-7.10.x

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

testing/unittests/unittests.cpp

@@ -259,7 +259,7 @@ int main(int argc, char* argv[])
 
     objects.kill();
     ExitModuleObjects();
-    return wasSuccessful;
+    return wasSuccessful ? 0 : 1; // 0 == exit code success
 }
 
 

thorlcr/activities/diskread/thdiskreadslave.cpp

@@ -1175,7 +1175,8 @@ public:
             {
                 distributor->disconnect(true);
                 distributor->join();
-            }            
+            }
+            aggregateStream.clear();
         }
         PARENT::stop();
     }

thorlcr/activities/hashdistrib/thhashdistribslave.cpp

@@ -4316,6 +4316,7 @@ IRowStream *mergeLocalAggs(Owned<IHashDistributor> &distributor, CSlaveActivity
         IHThorRowAggregator &helper;
         IHashDistributor &distributor;
         CSlaveActivity &activity;
+        bool stopped = false;
     public:
         CAggregatingStream(IHThorRowAggregator &_helper, IEngineRowAllocator &_rowAllocator, ICompare &_cmp, IHashDistributor &_distributor, CSlaveActivity &_activity)
             : helper(_helper), rowAllocator(_rowAllocator), cmp(_cmp), distributor(_distributor), rowBuilder(_rowAllocator), activity(_activity)
@@ -4323,6 +4324,7 @@ IRowStream *mergeLocalAggs(Owned<IHashDistributor> &distributor, CSlaveActivity
         }
         void start(IRowStream *_input)
         {
+            stopped = false;
             input.setown(_input);
         }
         // IRowStream
@@ -4363,6 +4365,9 @@ IRowStream *mergeLocalAggs(Owned<IHashDistributor> &distributor, CSlaveActivity
         }
         virtual void stop() override
         {
+            if (stopped)
+                return;
+            stopped = true;
             sz = 0;
             rowBuilder.clear();
             input->stop();