Merge pull request #1270 from vogel/fix-work-notify

Fix work notify

Reviewed-by: jedmccaleb

src/historywork/GetAndUnzipRemoteFileWork.cpp

@@ -57,11 +57,6 @@ GetAndUnzipRemoteFileWork::onSuccess()
 {
     if (mGunzipFileWork)
     {
-        if (mGunzipFileWork->getState() != WORK_SUCCESS)
-        {
-            return WORK_PENDING;
-        }
-
         if (!fs::exists(mFt.localPath_nogz()))
         {
             CLOG(ERROR, "History") << "Downloading and unzipping "

src/work/Work.cpp

@@ -8,6 +8,7 @@
 #include "util/Logging.h"
 #include "util/Math.h"
 #include "util/make_unique.h"
+#include "work/WorkManager.h"
 #include "work/WorkParent.h"
 
 #include "medida/meter.h"
@@ -137,38 +138,57 @@ Work::callComplete()
 void
 Work::scheduleRun()
 {
+    if (mScheduled)
+    {
+        return;
+    }
+
     std::weak_ptr<Work> weak(
         std::static_pointer_cast<Work>(shared_from_this()));
     CLOG(DEBUG, "Work") << "scheduling run of " << getUniqueName();
+    mScheduled = true;
     mApp.getClock().getIOService().post([weak]() {
         auto self = weak.lock();
         if (!self)
         {
             return;
         }
+        self->mScheduled = false;
         self->run();
     });
 }
 
 void
 Work::scheduleComplete(CompleteResult result)
 {
+    if (mScheduled)
+    {
+        return;
+    }
+
     std::weak_ptr<Work> weak(
         std::static_pointer_cast<Work>(shared_from_this()));
     CLOG(DEBUG, "Work") << "scheduling completion of " << getUniqueName();
+    mScheduled = true;
     mApp.getClock().getIOService().post([weak, result]() {
         auto self = weak.lock();
         if (!self)
         {
             return;
         }
+        self->mScheduled = false;
         self->complete(result);
     });
 }
 
 void
 Work::scheduleRetry()
 {
+    if (mScheduled)
+    {
+        return;
+    }
+
     if (getState() != WORK_FAILURE_RETRY)
     {
         std::string msg = fmt::format("retrying {} in state {}",
@@ -190,13 +210,15 @@ Work::scheduleRetry()
         << "Scheduling retry #" << (mRetries + 1) << "/" << mMaxRetries
         << " in " << std::chrono::duration_cast<std::chrono::seconds>(t).count()
         << " sec, for " << getUniqueName();
+    mScheduled = true;
     mRetryTimer->async_wait(
         [weak]() {
             auto self = weak.lock();
             if (!self)
             {
                 return;
             }
+            self->mScheduled = false;
             self->mRetries++;
             self->reset();
             self->advance();

src/work/Work.h

@@ -104,6 +104,7 @@ class Work : public WorkParent
     size_t mMaxRetries{RETRY_A_FEW};
     size_t mRetries{0};
     State mState{WORK_PENDING};
+    bool mScheduled {false};
 
     std::unique_ptr<VirtualTimer> mRetryTimer;
 

src/work/WorkTests.cpp

@@ -96,3 +96,160 @@ TEST_CASE("work steps", "[work]")
         clock.crank();
     }
 }
+
+class FailingWork : public Work
+{
+  public:
+    FailingWork(Application& app, WorkParent& parent,
+                std::string const& uniqueName)
+        : Work(app, parent, uniqueName, 0)
+    {
+    }
+
+    virtual void
+    onRun() override
+    {
+        scheduleFailure();
+    }
+};
+
+class WorkCountingOnFailureRaise : public Work
+{
+  public:
+    int mCount{0};
+
+    WorkCountingOnFailureRaise(Application& app, WorkParent& parent)
+        : Work(app, parent, std::string("counting-on-complete"), 0)
+    {
+    }
+
+    virtual void
+    onRun() override
+    {
+        mCount++;
+        Work::onRun();
+    }
+
+    virtual void
+    onFailureRaise() override
+    {
+        mCount++;
+        Work::onFailureRaise();
+    }
+};
+
+TEST_CASE("subwork items fail at the same time", "[work]")
+{
+    VirtualClock clock;
+    auto const& cfg = getTestConfig();
+    auto app = Application::create(clock, cfg);
+    auto& wm = app->getWorkManager();
+    auto w = wm.addWork<WorkCountingOnFailureRaise>();
+    w->addWork<FailingWork>("work-1");
+    w->addWork<FailingWork>("work-2");
+    wm.advanceChildren();
+    while (!wm.allChildrenDone())
+    {
+        clock.crank();
+    }
+
+    REQUIRE(w->mCount == 1);
+}
+
+class WorkDoNothing : public Work
+{
+  public:
+    WorkDoNothing(Application& app, WorkParent& parent,
+                  std::string const& uniqueName)
+        : Work(app, parent, uniqueName, 0)
+    {
+    }
+
+    virtual void
+    onRun() override
+    {
+    }
+
+    void
+    forceSuccess()
+    {
+        callComplete()({});
+    }
+};
+
+class WorkWith2Subworks : public Work
+{
+  public:
+    WorkWith2Subworks(Application& app, WorkParent& parent,
+                      std::string const& uniqueName)
+        : Work(app, parent, uniqueName, 0)
+    {
+    }
+
+    virtual void
+    onReset() override
+    {
+        clearChildren();
+        mFirstSubwork.reset();
+        mSecondSubwork.reset();
+
+        mFirstSubwork =
+            addWork<WorkDoNothing>("first-subwork-of-" + getUniqueName());
+    }
+
+    virtual Work::State
+    onSuccess() override
+    {
+        CLOG(DEBUG, "History") << "onSuccess() for " << getUniqueName();
+
+        if (mSecondSubwork)
+        {
+            mCalledSuccessWithPendingSubwork =
+                mCalledSuccessWithPendingSubwork ||
+                mSecondSubwork->getState() == WORK_PENDING;
+            return WORK_SUCCESS;
+        }
+
+        mSecondSubwork =
+            addWork<WorkDoNothing>("second-subwork-of-" + getUniqueName());
+        return WORK_PENDING;
+    }
+
+    std::shared_ptr<WorkDoNothing> mFirstSubwork;
+    std::shared_ptr<WorkDoNothing> mSecondSubwork;
+    bool mCalledSuccessWithPendingSubwork{false};
+};
+
+TEST_CASE("sub-subwork items succed at the same time", "[work]")
+{
+    VirtualClock clock;
+    auto const& cfg = getTestConfig();
+    auto app = Application::create(clock, cfg);
+    auto& wm = app->getWorkManager();
+    auto w = wm.addWork<Work>("parent-of-many");
+    auto work1 = w->addWork<WorkWith2Subworks>("work-1");
+    auto work2 = w->addWork<WorkDoNothing>("work-2");
+    auto work3 = w->addWork<WorkDoNothing>("work-3");
+
+    auto i = 0;
+
+    wm.advanceChildren();
+    while (!wm.allChildrenDone())
+    {
+        clock.crank(false);
+
+        switch (i++)
+        {
+        case 1:
+            work2->forceSuccess();
+            work1->mFirstSubwork->forceSuccess();
+            work3->forceSuccess();
+            break;
+        case 2:
+            work1->mSecondSubwork->forceSuccess();
+            break;
+        }
+    }
+
+    REQUIRE(!work1->mCalledSuccessWithPendingSubwork);
+}