Merge pull request #5374 from richardkchapman/roxie-abort

HPCC-10351 Cannot abort roxie job

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

common/workunit/workunit.cpp

@@ -3523,78 +3523,6 @@ void CLocalWorkUnit::subscribe(WUSubscribeOptions options)
     }
 }
 
-#if 0 // I don't think this is used (I grepped the source), am leaving here just in case I've missed somewhere (PG)
-WUState CLocalWorkUnit::waitComplete(int timeout, bool returnOnWaitState)
-{
-    class WorkUnitWaiter : public CInterface, implements ISDSSubscription, implements IAbortHandler
-    {
-        Semaphore changed;
-        CLocalWorkUnit *parent;
-    public:
-        IMPLEMENT_IINTERFACE;
-
-        WorkUnitWaiter(CLocalWorkUnit *_parent) : parent(_parent) { aborted = false; };
-
-        void notify(SubscriptionId id, const char *xpath, SDSNotifyFlags flags, unsigned valueLen, const void *valueData)
-        {
-            parent->notify(id, xpath, flags, valueLen, valueData);
-            changed.signal();
-        }
-        bool wait(unsigned timeout)
-        {
-            return changed.wait(timeout) && !aborted;
-        }
-        bool onAbort()
-        {
-            aborted = true;
-            changed.signal();
-            return false;
-        }
-        bool aborted;
-
-    } waiter(this);
-    Owned<CWorkUnitWatcher> w = new CWorkUnitWatcher(&waiter, p->queryName(), false);
-    LocalIAbortHandler abortHandler(waiter);
-    forceReload(); // or may miss changes that already happened, between load of wu and now.
-    unsigned start = msTick();
-    WUState ret;
-    loop
-    {
-        ret = getState();
-        switch (ret)
-        {
-        case WUStateWait:
-            if(!returnOnWaitState)
-                break;
-            //fall thru
-        case WUStateCompleted:
-        case WUStateFailed:
-        case WUStateAborted:
-            w->unsubscribe();
-            return ret;
-        }
-        unsigned waited = msTick() - start;
-        if (timeout==-1)
-        {
-            waiter.wait(20000);
-            if (waiter.aborted)
-            {
-                ret = WUStateUnknown;  // MORE - throw an exception?
-                break;
-            }
-        }
-        else if (waited > timeout || !waiter.wait(timeout-waited))
-        {
-            ret = WUStateUnknown;  // MORE - throw an exception?
-            break;
-        }
-        reload();
-    }
-    w->unsubscribe();
-    return ret;
-}
-#endif
-
 void CLocalWorkUnit::forceReload()
 {
     dirty = true;

roxie/ccd/ccdcontext.cpp

@@ -2180,6 +2180,7 @@ protected:
     void startWorkUnit()
     {
         WorkunitUpdate wu(&workUnit->lock());
+        wu->subscribe(SubscribeOptionAbort);
         wu->addTimeStamp("Roxie", GetCachedHostName(), "Started");
         if (!context->getPropBool("@outputToSocket", false))
             client = NULL;