Changeset f67dfb for ThirdParty

Timestamp:

Mar 9, 2017, 10:16:58 PM (8 years ago)

Author:

Frederik Heber <heber@…>

Branches:

Action_Thermostats, Add_AtomRandomPerturbation, Add_RotateAroundBondAction, Add_SelectAtomByNameAction, Adding_Graph_to_ChangeBondActions, Adding_MD_integration_tests, Adding_StructOpt_integration_tests, AutomationFragmentation_failures, Candidate_v1.6.0, Candidate_v1.6.1, ChangeBugEmailaddress, ChangingTestPorts, ChemicalSpaceEvaluator, Debian_Package_split, Debian_package_split_molecuildergui_only, Disabling_MemDebug, Docu_Python_wait, EmpiricalPotential_contain_HomologyGraph_documentation, Enhance_userguide, Enhanced_StructuralOptimization, Enhanced_StructuralOptimization_continued, Example_ManyWaysToTranslateAtom, Exclude_Hydrogens_annealWithBondGraph, FitPartialCharges_GlobalError, Fix_ChronosMutex, Fix_StatusMsg, Fix_StepWorldTime_single_argument, Fix_Verbose_Codepatterns, ForceAnnealing_goodresults, ForceAnnealing_oldresults, ForceAnnealing_tocheck, ForceAnnealing_with_BondGraph, ForceAnnealing_with_BondGraph_continued, ForceAnnealing_with_BondGraph_continued_betteresults, ForceAnnealing_with_BondGraph_contraction-expansion, GeometryObjects, Gui_displays_atomic_force_velocity, IndependentFragmentGrids_IntegrationTest, JobMarket_RobustOnKillsSegFaults, JobMarket_StableWorkerPool, JobMarket_unresolvable_hostname_fix, PartialCharges_OrthogonalSummation, PythonUI_with_named_parameters, QtGui_reactivate_TimeChanged_changes, Recreated_GuiChecks, RotateToPrincipalAxisSystem_UndoRedo, StoppableMakroAction, TremoloParser_IncreasedPrecision, TremoloParser_MultipleTimesteps, Ubuntu_1604_changes, stable

Children:

4e8108, cc85e4

Parents:

8e7a1b

git-author:

Frederik Heber <heber@…> (02/23/17 15:17:19)

git-committer:

Frederik Heber <heber@…> (03/09/17 22:16:58)

Message:

FIX: FragmentScheduler and PoolWorker now handle unresolvable addresses properly.

• OperationQueue now checks whether last operation has succeeded of failed and calls a failure callback in error case which is a NoOp by default.
• if PoolWorker's Ops fail, he will automatically shutdown now.
• if FragmentScheduler's Ops fail, he will automatically remove the worker from its pool.

Location:

ThirdParty/JobMarket

Files:

• src/JobMarket/FragmentScheduler.cpp (modified) (1 diff)
• src/JobMarket/Operations/OperationQueue.cpp (modified) (3 diffs)
• src/JobMarket/Operations/OperationQueue.hpp (modified) (5 diffs)
• src/JobMarket/Pool/PoolWorker.cpp (modified) (1 diff)
• tests/regression/testsuite-falsehost.at (modified) (1 diff)

ThirdParty/JobMarket/src/JobMarket/FragmentScheduler.cpp ¶

@@ -65 +65 @@
       boost::bind(&FragmentScheduler::shutdown, boost::ref(*this))),
   connection(_io_service),
+  OpQueue(boost::bind(&FragmentScheduler::removeWorker, boost::ref(*this), _1)),
   guard(_io_service, timeout, connection,
       boost::bind(&FragmentScheduler::removeWorker, boost::ref(*this), _1),

ThirdParty/JobMarket/src/JobMarket/Operations/OperationQueue.cpp ¶

@@ -33 +33 @@
 #include "JobMarket/Operations/AsyncOperation.hpp"
 #include "JobMarket/Operations/OperationQueue.hpp"
-#include "JobMarket/WorkerAddress.hpp"
+
+static void NoOp(const WorkerAddress) {}
+
+// static instances
+const boost::function<void (const WorkerAddress)> OperationQueue::NoOpCallback = boost::bind(&NoOp, _1);
 
 size_t OperationQueue::max_connections = 1;
@@ -74 +78 @@
             boost::bind(&AddressMap_t::count, boost::ref(AddressMap), boost::lambda::_1) );
     if (queueiter != queue.end()) {
-      AddressMap_t::iterator mapiter = AddressMap.find(*queueiter);
+      const AddressMap_t::iterator mapiter = AddressMap.find(*queueiter);
       ASSERT( mapiter != AddressMap.end(),
           "OperationQueue::LaunchNextOp() - cannot find connection "+toString((*queueiter)->getName())+" in AddressMap.");
-      const WorkerAddress address = mapiter->second;
-      AsyncOp_ptr ptr = mapiter->first;
+      currentOpsAddress = mapiter->second;
+      const AsyncOp_ptr ptr = mapiter->first;
       // always erase the op from the list of ones pending for launch
       AddressMap.erase(mapiter);
       // only launch when not a debug op
-      if ((!address.host.empty()) && (!address.service.empty())) {
+      if ((!currentOpsAddress.host.empty()) && (!currentOpsAddress.service.empty())) {
         LOG(2, "DEBUG: Launching next operation " << ptr->getName() << ".");
-       (*ptr)(address.host, address.service);
+       (*ptr)(currentOpsAddress.host, currentOpsAddress.service);
       } else {
         LOG(3, "DEBUG: Skipping debug operation " << ptr->getName() << " with empty address.");
@@ -117 +121 @@
   AsyncOperation *op = static_cast<AsyncOperation *>(publisher);
   if (op != NULL) {
-    LOG(1, "INFO: We are notified that " << op->getName() << " is done, removing ...");
+    // check for error code of operation
+    if (op->getStatus() != Operation::success) {
+      // remove the worker from the queue
+      failure_callback(currentOpsAddress);
+      LOG(1, "INFO: We are notified that " << op->getName() << " has failed, removing ...");
+    } else {
+      LOG(1, "INFO: We are notified that " << op->getName() << " is done, removing ...");
+    }
     // remove from queue
     remove(op, this);

ThirdParty/JobMarket/src/JobMarket/Operations/OperationQueue.hpp ¶

@@ -15 +15 @@
 #endif
 
+#include <boost/function.hpp>
 #include <boost/shared_ptr.hpp>
 #include <deque>
@@ -20 +21 @@
 #include "CodePatterns/Observer/Observer.hpp"
 
+#include "JobMarket/WorkerAddress.hpp"
+
 class AsyncOperation;
 class Observer;
 class OperationQueueTest;
-class WorkerAddress;
 
 /** This class is a container for \ref AsyncOperation's that are kept as shared_ptr
@@ -32 +34 @@
   //!> grant unit test access to private part
   friend class OperationQueueTest;
+
+  //!> static function as default callback on failure of operation that does nothing
+  static const boost::function<void (const WorkerAddress)> NoOpCallback;
 public:
   /** Default constructor for class OperationQueue.
@@ -39 +44 @@
     Observer("OperationQueue"),
     RunningOps(0),
-    IsBlockedFlag(false)
+    currentOpsAddress("localhost", "default"),
+    IsBlockedFlag(false),
+    failure_callback(NoOpCallback)
   {}
+
+  /** Default constructor for class OperationQueue.
+   *
+   */
+  OperationQueue(const boost::function<void (const WorkerAddress)> &_failure_callback) :
+    Observer("OperationQueue"),
+    RunningOps(0),
+    currentOpsAddress("localhost", "default"),
+    IsBlockedFlag(false),
+    failure_callback(_failure_callback)
+  {}
+
   /** Default destructor for class OperationQueue.
    *
@@ -140 +159 @@
   AddressMap_t AddressMap;
 
+  //!> address of current operation in case of failure
+  WorkerAddress currentOpsAddress;
+
   //!> status flag whether queue is blocked or operations may be pushed.
   bool IsBlockedFlag;
+
+  //!> failure callback to call when operation's address could not be resolved
+  const boost::function<void (const WorkerAddress)> failure_callback;
 };
 

ThirdParty/JobMarket/src/JobMarket/Pool/PoolWorker.cpp ¶

@@ -63 +63 @@
   connection_(_io_service),
   failed(boost::bind(&ExitflagContainer::setExitflag, this, ExitflagContainer::ErrorFlag)),
+  OpQueue(boost::bind(&PoolWorker::finish, boost::ref(*this))),
   thread_remover(boost::bind(&PoolWorker::PoolListener_t::handle_RemoveThread, boost::ref(PoolListener)))
 {

ThirdParty/JobMarket/tests/regression/testsuite-falsehost.at ¶

@@ -3 +3 @@
 AT_SETUP([Fragmentation - Automation: Worker sets false hostname])
 AT_KEYWORDS([fragmentation automation controller poolworker server falsehost])
-AT_XFAIL_IF([/bin/true])
 
 WORKERPORT=1080