Changeset 2c3ae5

Timestamp:

Aug 11, 2025, 5:45:18 PM (3 months ago)

Author:

Frederik Heber <frederik.heber@…>

Branches:

Candidate_v1.7.0, stable

Children:

e2f31d3

Parents:

74d798

git-author:

Frederik Heber <frederik.heber@…> (07/13/25 21:35:10)

git-committer:

Frederik Heber <frederik.heber@…> (08/11/25 17:45:18)

Message:

JobMarket: Extended PoolGuard to also check idle workers.

Location:

ThirdParty/JobMarket/src/JobMarket/Pool

Files:

• PoolGuard.cpp (modified) (9 diffs)
• PoolGuard.hpp (modified) (2 diffs)
• WorkerPool.hpp (modified) (1 diff)

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

@@ -65 +65 @@
 {
   if (WaitingOps == 0) {
-    LOG(1, "INFO: There are " << LastWorkerList.size() << " busy workers from last time, "
-         << CurrentWorkerList.size() << " are currently busy.");
+    LOG(1, "INFO: There are " << LastBusyWorkerList.size() << " busy workers from last time, "
+         << CurrentBusyWorkerList.size() << " are currently busy.");
     // create a vector of workers to check
     typedef std::set<WorkerAddress> CheckList_t;
     CheckList_t currentworkers;
-    for (WorkerList_t::const_iterator iter = LastWorkerList.begin();
-        iter != LastWorkerList.end();
+    for (WorkerList_t::const_iterator iter = LastBusyWorkerList.begin();
+        iter != LastBusyWorkerList.end();
         ++iter) {
       const WorkerAddress &address = iter->first;
       LOG(2, "DEBUG: Checking whether worker " << address << " is still busy ...");
-      WorkerList_t::const_iterator currentiter = CurrentWorkerList.find(address);
+      WorkerList_t::const_iterator currentiter = CurrentBusyWorkerList.find(address);
       // check if worker was busy last time and on same job
-      if (currentiter != CurrentWorkerList.end()) {
+      if (currentiter != CurrentBusyWorkerList.end()) {
         LOG(2, "DEBUG: Worker " << address << " was busy last time on job " << iter->second << ".");
         if (currentiter->second == iter->second) {
@@ -101 +101 @@
     for(CheckList_t::const_iterator iter = currentworkers.begin();
         iter != currentworkers.end(); ++iter) {
-      const WorkerAddress &address = *iter;
-      LOG(1, "INFO: Checking whether " << address << " is alive.");
-      AsyncOperation *checkaliveWorkerOp = new CheckAliveWorkerOperation(connection,
-          boost::bind(&PoolGuard::checkAddress, this, address, _1),
-          boost::bind(&PoolGuard::printWorkerIsAlive, this, address),
-          boost::bind(&PoolGuard::removeFromPool, this, address));
-      OpQueue.push_back(checkaliveWorkerOp, address);
-      ++WaitingOps;
+      CheckWorker(*iter);
+    }
+    LOG(1, "INFO: Checking on " << CurrentIdleWorkerList.size() << " idle workers.");
+    for(std::set<WorkerAddress>::const_iterator iter = CurrentIdleWorkerList.begin();
+        iter != CurrentIdleWorkerList.end(); ++iter) {
+      CheckWorker(*iter);
     }
   } else {
@@ -114 +112 @@
   }
   // set old list to new list
-  LastWorkerList = CurrentWorkerList;
+  LastBusyWorkerList = CurrentBusyWorkerList;
 
   // set next check interval
@@ -126 +124 @@
 }
 
+void PoolGuard::CheckWorker(const WorkerAddress &address) 
+{
+  LOG(1, "INFO: Checking whether " << address << " is alive.");
+  AsyncOperation *checkaliveWorkerOp = new CheckAliveWorkerOperation(connection,
+      boost::bind(&PoolGuard::checkAddress, this, address, _1),
+      boost::bind(&PoolGuard::printWorkerIsAlive, this, address),
+      boost::bind(&PoolGuard::removeFromPool, this, address));
+  OpQueue.push_back(checkaliveWorkerOp, address);
+  ++WaitingOps;
+}
+
 void PoolGuard::checkAddress(
     const WorkerAddress trueaddress,
@@ -142 +151 @@
   --WaitingOps;
   // erase from our lists
-  LastWorkerList.erase(address);
-  WorkerList_t::iterator iter = CurrentWorkerList.find(address);
-  if ( iter != CurrentWorkerList.end()) {
+  LastBusyWorkerList.erase(address);
+  WorkerList_t::iterator iter = CurrentBusyWorkerList.find(address);
+  if ( iter != CurrentBusyWorkerList.end()) {
     const JobId_t jobid = iter->second;
-    CurrentWorkerList.erase(iter);
+    CurrentBusyWorkerList.erase(iter);
     // erase from pool
     removeWorkerfunction(address);
@@ -160 +169 @@
   std::pair<WorkerList_t::iterator, bool> inserter =
 #endif
-  CurrentWorkerList.insert( std::make_pair(address, id) );
+  CurrentBusyWorkerList.insert( std::make_pair(address, id) );
   ASSERT( inserter.second,
       "PoolGuard::addBusyWorker() - worker "+toString(address)+" "
@@ -169 +178 @@
 {
   LOG(1, "INFO: Removing worker " << address << " as busy.");
-  WorkerList_t::iterator iter = CurrentWorkerList.find(address);
-  if (iter != CurrentWorkerList.end()) {
-    CurrentWorkerList.erase(address);
-    LastWorkerList.erase(address);
+  WorkerList_t::iterator iter = CurrentBusyWorkerList.find(address);
+  if (iter != CurrentBusyWorkerList.end()) {
+    CurrentBusyWorkerList.erase(address);
+    LastBusyWorkerList.erase(address);
   }
 }
@@ -188 +197 @@
   timer.cancel();
   // clear internal list such that we may correctly be started again
-  LastWorkerList.clear();
-  CurrentWorkerList.clear();
+  LastBusyWorkerList.clear();
+  CurrentBusyWorkerList.clear();
 }
 
@@ -196 +205 @@
   // set flag to true
   CheckAtNextInterval = true;
-  // and check right away to fill LastWorkerList
+  // and check right away to fill LastBusyWorkerList
   checkWorkers();
 }

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

@@ -67 +67 @@
   void checkWorkers();
 
+  /** Schedules a checkalive operation for a single worker at \a address.
+   *
+   * \param address the address to check
+   */
+  void CheckWorker(const WorkerAddress &address);
+  
   /** Checks busy workers in pool whether react to CheckAlive.
    *
@@ -142 +148 @@
 
   //!> last set of working workers
-  WorkerList_t LastWorkerList;
+  WorkerList_t LastBusyWorkerList;
 
   //!> last set of working workers
-  WorkerList_t CurrentWorkerList;
+  WorkerList_t CurrentBusyWorkerList;
+
+  //!> set of idle working workers
+  std::set<WorkerAddress> CurrentIdleWorkerList;
 
   //!> reference to OperationQueue for pushing CheckAliveWorkerOperation

ThirdParty/JobMarket/src/JobMarket/Pool/WorkerPool.hpp

@@ -69 +69 @@
   typedef std::vector<std::pair<std::string, std::string> > WorkerList_t;
   WorkerList_t getListOfIdleWorkers() const;
-  const Pool_t& getPoolOfWorkers() const
-  {
-    return pool;
-  }
 
   /** Return the number of busy workers.