Changeset f91f47

Timestamp:

Feb 2, 2023, 8:14:03 PM (3 years ago)

Author:

Frederik Heber <frederik.heber@…>

Branches:

Candidate_v1.7.0, stable

Children:

1b84b8

Parents:

75a0cb

Message:

FIX: JobMarket's PoolGuard did not check for existing result.

• when a worker re-enrolls, the PoolGuard notices that it's no longer on its busy list. However, it now also checks whether the result the worker was working on is actually present. If missing, then the job is resubmitted to the queue.

Location:

ThirdParty/JobMarket/src/JobMarket

Files:

• FragmentScheduler.cpp (modified) (1 diff)
• Pool/PoolGuard.cpp (modified) (3 diffs)
• Pool/PoolGuard.hpp (modified) (3 diffs)

ThirdParty/JobMarket/src/JobMarket/FragmentScheduler.cpp

@@ -69 +69 @@
       boost::bind(&FragmentScheduler::removeWorker, boost::ref(*this), _1),
       boost::bind(&FragmentQueue::resubmitJob, boost::ref(JobsQueue), _1),
+      boost::bind(&FragmentQueue::isResultPresent, boost::cref(JobsQueue), _1),
       OpQueue)
 {

ThirdParty/JobMarket/src/JobMarket/Pool/PoolGuard.cpp

@@ -42 +42 @@
     const boost::function<void (const WorkerAddress)> _removeWorkerfunction,
     const boost::function<void (const JobId_t)> _resubmitJobfunction,
+    const boost::function<bool (const JobId_t)> _checkResultPresentfunction,
     OperationQueue &_OpQueue) :
   CheckAtNextInterval(false),
@@ -48 +49 @@
   removeWorkerfunction(_removeWorkerfunction),
   resubmitJobfunction(_resubmitJobfunction),
+  checkResultPresentfunction(_checkResultPresentfunction),
   OpQueue(_OpQueue),
   connection(_connection),
@@ -84 +86 @@
               << currentiter->second << " than last time "
               << iter->second << ", scheduling for checkalive.");
+        }
+      } else {
+        // check whether the result is not present
+        if (!checkResultPresentfunction(iter->second)) {
+          LOG(1, "INFO: Worker " << address << " is no longer busy but the job #"
+              << iter->second << "'s result is not present, resubmitting.");
+          resubmitJobfunction(iter->second);
         }
       }

ThirdParty/JobMarket/src/JobMarket/Pool/PoolGuard.hpp

@@ -44 +44 @@
    * @param _removeWorkerfunction bound function to remove worker by its WorkerAddress
    * @param _resubmitJobfunction bound function to resubmit job by its JobId_t
+   * @param _checkResultPresentfunction bound function to check whether a job's result by its JobId_t is present
    * @param _OpQueue access to operation queue for placing CheckAliveWorkerOperations
    */
@@ -52 +53 @@
       const boost::function<void (const WorkerAddress)> _removeWorkerfunction,
       const boost::function<void (const JobId_t)> _resubmitJobfunction,
+      const boost::function<bool (const JobId_t)> _checkResultPresentfunction,
       OperationQueue &_OpQueue
       );
@@ -131 +133 @@
   const boost::function<void (const JobId_t)> resubmitJobfunction;
 
+  //!> bound function for checking for a job result's presence
+  const boost::function<bool (const JobId_t)> checkResultPresentfunction;
+
   //!> typedef for the internal list of workers
   typedef std::map<WorkerAddress, JobId_t> WorkerList_t;