Changeset 9a3f84 for src

Timestamp:

Jun 29, 2012, 1:23:14 PM (13 years ago)

Author:

Frederik Heber <heber@…>

Branches:

Action_Thermostats, Add_AtomRandomPerturbation, Add_FitFragmentPartialChargesAction, Add_RotateAroundBondAction, Add_SelectAtomByNameAction, Added_ParseSaveFragmentResults, AddingActions_SaveParseParticleParameters, Adding_Graph_to_ChangeBondActions, Adding_MD_integration_tests, Adding_ParticleName_to_Atom, Adding_StructOpt_integration_tests, AtomFragments, Automaking_mpqc_open, AutomationFragmentation_failures, Candidate_v1.5.4, Candidate_v1.6.0, Candidate_v1.6.1, ChangeBugEmailaddress, ChangingTestPorts, ChemicalSpaceEvaluator, CombiningParticlePotentialParsing, Combining_Subpackages, Debian_Package_split, Debian_package_split_molecuildergui_only, Disabling_MemDebug, Docu_Python_wait, EmpiricalPotential_contain_HomologyGraph, EmpiricalPotential_contain_HomologyGraph_documentation, Enable_parallel_make_install, Enhance_userguide, Enhanced_StructuralOptimization, Enhanced_StructuralOptimization_continued, Example_ManyWaysToTranslateAtom, Exclude_Hydrogens_annealWithBondGraph, FitPartialCharges_GlobalError, Fix_BoundInBox_CenterInBox_MoleculeActions, Fix_ChargeSampling_PBC, Fix_ChronosMutex, Fix_FitPartialCharges, Fix_FitPotential_needs_atomicnumbers, Fix_ForceAnnealing, Fix_IndependentFragmentGrids, Fix_ParseParticles, Fix_ParseParticles_split_forward_backward_Actions, Fix_PopActions, Fix_QtFragmentList_sorted_selection, Fix_Restrictedkeyset_FragmentMolecule, Fix_StatusMsg, Fix_StepWorldTime_single_argument, Fix_Verbose_Codepatterns, Fix_fitting_potentials, Fixes, ForceAnnealing_goodresults, ForceAnnealing_oldresults, ForceAnnealing_tocheck, ForceAnnealing_with_BondGraph, ForceAnnealing_with_BondGraph_continued, ForceAnnealing_with_BondGraph_continued_betteresults, ForceAnnealing_with_BondGraph_contraction-expansion, FragmentAction_writes_AtomFragments, FragmentMolecule_checks_bonddegrees, GeometryObjects, Gui_Fixes, Gui_displays_atomic_force_velocity, ImplicitCharges, IndependentFragmentGrids, IndependentFragmentGrids_IndividualZeroInstances, IndependentFragmentGrids_IntegrationTest, IndependentFragmentGrids_Sole_NN_Calculation, JobMarket_RobustOnKillsSegFaults, JobMarket_StableWorkerPool, JobMarket_unresolvable_hostname_fix, MoreRobust_FragmentAutomation, ODR_violation_mpqc_open, PartialCharges_OrthogonalSummation, PdbParser_setsAtomName, PythonUI_with_named_parameters, QtGui_reactivate_TimeChanged_changes, Recreated_GuiChecks, Rewrite_FitPartialCharges, RotateToPrincipalAxisSystem_UndoRedo, SaturateAtoms_findBestMatching, SaturateAtoms_singleDegree, StoppableMakroAction, Subpackage_CodePatterns, Subpackage_JobMarket, Subpackage_LinearAlgebra, Subpackage_levmar, Subpackage_mpqc_open, Subpackage_vmg, Switchable_LogView, ThirdParty_MPQC_rebuilt_buildsystem, TrajectoryDependenant_MaxOrder, TremoloParser_IncreasedPrecision, TremoloParser_MultipleTimesteps, TremoloParser_setsAtomName, Ubuntu_1604_changes, stable

Children:

aec098

Parents:

a8f54b6

git-author:

Frederik Heber <heber@…> (05/21/12 07:14:59)

git-committer:

Frederik Heber <heber@…> (06/29/12 13:23:14)

Message:

FragmentScheduler now uses WorkerChoices for handling connection workers.

• Workers always first send address, then their choice and depending on this we branch into various handlers.
• SubmitResultOperation and EnrollInPoolOperation now correctly give choice after sending address.
• handle_enrolled uses getNextIdleWorker() to mark indirectly as busy. This is temporary.
• we do not resubmit if Max_Attempts is 1.

Location:

src/Fragmentation/Automation

Files:

• Controller/Commands/EnrollInPoolOperation.cpp (modified) (4 diffs)
• Controller/Commands/EnrollInPoolOperation.hpp (modified) (1 diff)
• Controller/Commands/ObtainJobOperation.cpp (modified) (1 diff)
• Controller/Commands/SubmitResultOperation.cpp (modified) (6 diffs)
• Controller/Commands/SubmitResultOperation.hpp (modified) (1 diff)
• FragmentQueue.cpp (modified) (2 diffs)
• FragmentScheduler.cpp (modified) (10 diffs)
• FragmentScheduler.hpp (modified) (8 diffs)
• Pool/WorkerPool.hpp (modified) (1 diff)

src/Fragmentation/Automation/Controller/Commands/EnrollInPoolOperation.cpp

@@ -30 +30 @@
 #include "Jobs/FragmentJob.hpp"
 #include "WorkerAddress.hpp"
+#include "WorkerChoices.hpp"
 
 /// Handle completion of a connect operation.
@@ -41 +42 @@
     // of jobs. The connection::async_read() function will automatically
     // decode the data that is read from the underlying socket.
-    LOG(1, "INFO: Enrolling in pool with " << address << "...");
     connection_.async_write(address,
-      boost::bind(&EnrollInPoolOperation::handle_FinishOperation, this,
+      boost::bind(&EnrollInPoolOperation::handle_SendChoice, this,
       boost::asio::placeholders::error));
   } else if (endpoint_iterator != boost::asio::ip::tcp::resolver::iterator()) {
@@ -61 +61 @@
 
 
-/// Handle sent of address
-void EnrollInPoolOperation::handle_sendAddress(const boost::system::error_code& e)
+/// Handle sending choice
+void EnrollInPoolOperation::handle_SendChoice(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    // Successfully established connection. Start operation to read the list
+    // of jobs. The connection::async_read() function will automatically
+    // decode the data that is read from the underlying socket.
+    LOG(1, "INFO: Enrolling in pool with " << address << " ...");
+    enum WorkerChoices choice = EnrollInPool;
+    connection_.async_write(choice,
+      boost::bind(&EnrollInPoolOperation::handle_ReceiveFlag, this,
+      boost::asio::placeholders::error));
+  } else {
+    // An error occurred. Log it and return. Since we are not starting a new
+    // operation the io_service will run out of work to do and the client will
+    // exit.
+    ELOG(1, e.message());
+  }
+}
+
+
+/// Handle receiving flag
+void EnrollInPoolOperation::handle_ReceiveFlag(const boost::system::error_code& e)
 {
   Info info(__FUNCTION__);
@@ -81 +104 @@
 }
 
-/// Handle sent of address
+/// Handle received flag
 void EnrollInPoolOperation::handle_FinishOperation(const boost::system::error_code& e)
 {

src/Fragmentation/Automation/Controller/Commands/EnrollInPoolOperation.hpp

@@ -47 +47 @@
       boost::asio::ip::tcp::resolver::iterator endpoint_iterator);
 
-  /// Handle sent of address
-  void handle_sendAddress(const boost::system::error_code& e);
+  /// Handle sending choice
+  void handle_SendChoice(const boost::system::error_code& e);
 
-  /// Handle sent of address
+  /// Handle receiving flag
+  void handle_ReceiveFlag(const boost::system::error_code& e);
+
+  /// Handle received flag
   void handle_FinishOperation(const boost::system::error_code& e);
 

src/Fragmentation/Automation/Controller/Commands/ObtainJobOperation.cpp

@@ -29 +29 @@
 #include "CodePatterns/Log.hpp"
 #include "Jobs/FragmentJob.hpp"
+#include "WorkerChoices.hpp"
 
 /// Handle completion of a connect operation.

src/Fragmentation/Automation/Controller/Commands/SubmitResultOperation.cpp

@@ -29 +29 @@
 #include "CodePatterns/Log.hpp"
 #include "Results/FragmentResult.hpp"
+#include "WorkerChoices.hpp"
 
 
@@ -43 +44 @@
     LOG(1, "INFO: Sending address " << address << " ...");
     connection_.async_write(address,
-      boost::bind(&SubmitResultOperation::handle_SendAddress, this,
+      boost::bind(&SubmitResultOperation::handle_SendChoice, this,
       boost::asio::placeholders::error));
   } else if (endpoint_iterator != boost::asio::ip::tcp::resolver::iterator()) {
@@ -62 +63 @@
 
 /// Callback function when address has been sent and result is about to
-void SubmitResultOperation::handle_SendAddress(const boost::system::error_code& e)
+void SubmitResultOperation::handle_SendChoice(const boost::system::error_code& e)
 {
   Info info(__FUNCTION__);
@@ -70 +71 @@
     // of jobs. The connection::async_write() function will automatically
     // decode the data that is read from the underlying socket.
-    LOG(1, "INFO: Sending result #" << result->getId() << " ...");
-    connection_.async_write(result,
+    enum WorkerChoices choice = SendResult;
+    connection_.async_write(choice,
       boost::bind(&SubmitResultOperation::handle_SendResult, this,
       boost::asio::placeholders::error));
@@ -83 +84 @@
 }
 
-/// Callback function when result has been sent.
+/// Callback function when address has been sent and result is about to
 void SubmitResultOperation::handle_SendResult(const boost::system::error_code& e)
 {
   Info info(__FUNCTION__);
-  if (!e) {
-    LOG(1, "INFO: result #" << result->getId() << " sent.");
-    AsyncOperation::handle_FinishOperation(e);
+  if (!e)
+  {
+    // Successfully established connection. Start operation to read the list
+    // of jobs. The connection::async_write() function will automatically
+    // decode the data that is read from the underlying socket.
+    LOG(1, "INFO: Sending result #" << result->getId() << " ...");
+    connection_.async_write(result,
+      boost::bind(&SubmitResultOperation::handle_FinishOperation, this,
+      boost::asio::placeholders::error));
   } else {
     // An error occurred. Log it and return. Since we are not starting a new
@@ -98 +105 @@
   }
 }
+
+/// Callback function when result has been sent.
+void SubmitResultOperation::handle_FinishOperation(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+  LOG(1, "INFO: result #" << result->getId() << " sent.");
+  AsyncOperation::handle_FinishOperation(e);
+}

src/Fragmentation/Automation/Controller/Commands/SubmitResultOperation.hpp

@@ -68 +68 @@
 
   /// Callback function when address has been sent and result is about to.
-  void handle_SendAddress(const boost::system::error_code& e);
+  void handle_SendChoice(const boost::system::error_code& e);
 
   /// Callback function when result has been sent.
   void handle_SendResult(const boost::system::error_code& e);
+
+  /// Callback function when result has been sent.
+  void handle_FinishOperation(const boost::system::error_code& e);
 
 private:

src/Fragmentation/Automation/FragmentQueue.cpp

@@ -216 +216 @@
   AttemptsMap::iterator attemptiter = attempts.find(id);
   // check whether succeeded or (finally) failed
-  if ((_result->exitflag == 0) || ((attemptiter != attempts.end()) && (attemptiter->second >= Max_Attempts))) {
+  if ((_result->exitflag == 0) 
+       || ((attemptiter != attempts.end()) && (attemptiter->second >= Max_Attempts))
+       || (Max_Attempts == 1)) {
     // give notice if it is resubmitted job
     if (attemptiter != attempts.end()) {
@@ -258 +260 @@
       +" not stored in backup.");
   if (iter != backup.end()) {
-    // remove result
+    // remove present result
     ResultMap::iterator resiter = results.find(jobid);
     ASSERT( resiter != results.end(),
-        "FragmentQueue::resubmitJob() - resubmitting job "+toString(jobid)
-        +" for which no result is present.");
+        "FragmentQueue::resubmitJob() - job "+toString(jobid)
+        +" to resubmit has no result present.");
     results.erase(resiter);
     pushJob(iter->second);

src/Fragmentation/Automation/FragmentScheduler.cpp

@@ -93 +93 @@
     // read address
     conn->async_read(address,
-      boost::bind(&FragmentScheduler::WorkerListener_t::handle_checkAddress, this,
+      boost::bind(&FragmentScheduler::WorkerListener_t::handle_ReadAddress, this,
       boost::asio::placeholders::error, conn));
-  }  else {
+  }
+  else
+  {
     // An error occurred. Log it and return. Since we are not starting a new
     // accept operation the io_service will run out of work to do and the
@@ -104 +106 @@
 }
 
-/** Callback function when worker address has been received.
- *
- * \param e error code if something went wrong
- * \param conn reference with the connection
- */
-void FragmentScheduler::WorkerListener_t::handle_checkAddress(const boost::system::error_code& e, connection_ptr conn)
+/** Handle having received Worker's address
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::WorkerListener_t::handle_ReadAddress(const boost::system::error_code& e, connection_ptr conn)
 {
   Info info(__FUNCTION__);
   if (!e)
   {
-    if (pool.presentInPool(address)) {
-      // check whether its priority is busy_priority
-      if (pool.isWorkerBusy(address)) {
-        conn->async_read(result,
-          boost::bind(&FragmentScheduler::WorkerListener_t::handle_ReceiveResultFromWorker, this,
-          boost::asio::placeholders::error, conn));
-      } else {
-        ELOG(1, "INFO: Idle worker "+toString(address)+" has logged in again.");
-      }
-    } else {
-      // insert as its new worker
-      pool.addWorker(address);
-      enum EnrollInPoolOperation::EnrollFlag flag = EnrollInPoolOperation::Success;
-      conn->async_write(flag,
-        boost::bind(&FragmentScheduler::WorkerListener_t::handle_enrolled, this,
-        boost::asio::placeholders::error, conn));
-    }
+    // Successfully accepted a new connection.
+    // read address
+    conn->async_read(choice,
+      boost::bind(&FragmentScheduler::WorkerListener_t::handle_ReadChoice, this,
+      boost::asio::placeholders::error, conn));
   }
   else
@@ -140 +130 @@
     ELOG(0, e.message());
   }
+}
+
+/** Controller callback function to read the choice for next operation.
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::WorkerListener_t::handle_ReadChoice(const boost::system::error_code& e, connection_ptr conn)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    LOG(1, "INFO: Received request for operation " << choice << ".");
+    // switch over the desired choice read previously
+    switch(choice) {
+      case NoWorkerOperation:
+      {
+        ELOG(1, "WorkerListener_t::handle_ReadChoice() - called with NoOperation.");
+        break;
+      }
+      case EnrollInPool:
+      {
+        if (pool.presentInPool(address)) {
+          ELOG(1, "INFO: worker "+toString(address)+" is already contained in pool.");
+          enum EnrollInPoolOperation::EnrollFlag flag = EnrollInPoolOperation::Fail;
+          conn->async_write(flag,
+            boost::bind(&FragmentScheduler::WorkerListener_t::handle_enrolled, this,
+            boost::asio::placeholders::error, conn));
+        } else {
+          // insert as its new worker
+          LOG(1, "INFO: Adding " << address << " to pool ...");
+          pool.addWorker(address);
+          enum EnrollInPoolOperation::EnrollFlag flag = EnrollInPoolOperation::Success;
+          conn->async_write(flag,
+            boost::bind(&FragmentScheduler::WorkerListener_t::handle_enrolled, this,
+            boost::asio::placeholders::error, conn));
+        break;
+      }
+      case SendResult:
+      {
+        if (pool.presentInPool(address)) {
+          // check whether its priority is busy_priority
+          if (pool.isWorkerBusy(address)) {
+            conn->async_read(result,
+              boost::bind(&FragmentScheduler::WorkerListener_t::handle_ReceiveResultFromWorker, this,
+              boost::asio::placeholders::error, conn));
+          } else {
+            ELOG(1, "Worker " << address << " trying to send result who is not marked as busy.");
+            conn->async_read(result,
+              boost::bind(&FragmentScheduler::WorkerListener_t::handle_RejectResultFromWorker, this,
+              boost::asio::placeholders::error, conn));
+          }
+        } else {
+          ELOG(1, "Worker " << address << " trying to send result who is not in pool.");
+          conn->async_read(result,
+            boost::bind(&FragmentScheduler::WorkerListener_t::handle_RejectResultFromWorker, this,
+            boost::asio::placeholders::error, conn));
+        }
+        break;
+      }
+      case RemoveFromPool:
+      {
+        if (pool.presentInPool(address)) {
+          // removing present worker
+          pool.removeWorker(address);
+        } else {
+          ELOG(1, "Shutting down Worker " << address << " not contained in pool.");
+        }
+        break;
+      }
+      default:
+        Exitflag = ErrorFlag;
+        ELOG(1, "WorkerListener_t::handle_ReadChoice() - called with no valid choice.");
+        break;
+      }
+    }
+    // restore NoOperation choice such that choice is not read twice
+    choice = NoWorkerOperation;
+  }
+  else
+  {
+    // An error occurred. Log it and return. Since we are not starting a new
+    // accept operation the io_service will run out of work to do and the
+    // server will exit.
+    Exitflag = ErrorFlag;
+    ELOG(0, e.message());
+  }
 
   if (JobsQueue.isJobPresent()) {
@@ -147 +224 @@
   }
 }
+
 
 /** Callback function when new worker has enrolled.
@@ -164 +242 @@
       job = NoJob;
     }
-    callback_sendJobToWorker(address, job);
+    callback_sendJobToWorker(pool.getNextIdleWorker(), job);
   }
   else
@@ -205 +283 @@
 }
 
+/** Callback function when result has been received.
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::WorkerListener_t::handle_RejectResultFromWorker(const boost::system::error_code& e, connection_ptr conn)
+{
+  Info info(__FUNCTION__);
+  // nothing to do
+  LOG(1, "INFO: Rejecting result for job #" << result->getId() << ", placing back into queue.");
+
+  JobsQueue.resubmitJob(result->getId());
+
+  LOG(1, "INFO: JobsQueue has " << JobsQueue.getDoneJobs() << " results.");
+}
+
 
 /** Handle a new controller connection.
@@ -250 +344 @@
     case NoControllerOperation:
     {
-      ELOG(1, "FragmentScheduler::handle_ReadChoice() - called with NoOperation.");
+      ELOG(1, "ControllerListener_t::handle_ReadChoice() - called with NoOperation.");
       break;
     }
@@ -296 +390 @@
     case ShutdownControllerSocket:
     {
-      LaunchNewAcceptor = false;
+      LOG(1, "INFO: Received shutdown from controller ...");
+      // only allow for shutdown when there are no more jobs in the queue
+      if (!JobsQueue.isJobPresent()) {
+        LaunchNewAcceptor = false;
+      } else {
+        ELOG(2, "There are still jobs waiting in the queue.");
+      }
       break;
     }
     default:
       Exitflag = ErrorFlag;
-      ELOG(1, "FragmentScheduler::handle_ReadChoice() - called with no valid choice.");
+      ELOG(1, "ControllerListener_t::handle_ReadChoice() - called with no valid choice.");
       break;
     }
@@ -389 +489 @@
 /** Helper function to send a job to worker.
  *
+ * Note that we do not set the worker as busy. We simply send it the job.
+ *
  * @param address address of worker
  * @param job job to send
@@ -394 +496 @@
 void FragmentScheduler::sendJobToWorker(const WorkerAddress &address, FragmentJob::ptr &job)
 {
-  ASSERT( !pool.isWorkerBusy(address),
-      "FragmentScheduler::sendJobToWorker() - Worker "+toString(address)+" is already busy.");
+  ASSERT( pool.isWorkerBusy(address),
+      "FragmentScheduler::sendJobToWorker() - Worker "+toString(address)+" is not marked as busy.");
   LOG(1, "INFO: Sending job " << job->getId() << " to worker " << address << ".");
   sendJobOp.setJob(job);
   sendJobOp(address.host, address.service);
-  // set worker as busy (assuming it has been removed from idle queue already)
-  WorkerPool::Idle_Queue_t::iterator iter = pool.getIdleWorker(address);
-  ASSERT( iter != pool.getIdleEnd(),
-      "FragmentScheduler::sendJobToWorker() - cannot find worker "
-      +toString(address)+" in idle queue.");
-  pool.markWorkerBusy(iter);
 }
 

src/Fragmentation/Automation/FragmentScheduler.hpp

@@ -29 +29 @@
 #include "Pool/WorkerPool.hpp"
 #include "WorkerAddress.hpp"
+#include "WorkerChoices.hpp"
 
 /** FragmentScheduler serves FragmentJobs to Workers and accepts commands from
@@ -60 +61 @@
       pool(_pool),
       result( new FragmentResult(JobId::NoJob) ),
-      callback_sendJobToWorker(_callback)
+      callback_sendJobToWorker(_callback),
+      choice(NoWorkerOperation)
     {}
     virtual ~WorkerListener_t() {}
@@ -68 +70 @@
     void handle_Accept(const boost::system::error_code& e, connection_ptr conn);
 
+    /// Handle completion of Worker operation to read choice
+    void handle_ReadChoice(const boost::system::error_code& e, connection_ptr conn);
+
     /// Worker callback function when job has been sent.
-    void handle_checkAddress(const boost::system::error_code& e, connection_ptr conn);
+    void handle_ReadAddress(const boost::system::error_code& e, connection_ptr conn);
 
     /// Worker callback function when new worker has enrolled.
@@ -76 +81 @@
     /// Worker callback function when result has been received.
     void handle_ReceiveResultFromWorker(const boost::system::error_code& e, connection_ptr conn);
+
+    /// Worker callback function when invalid result has been received.
+    void handle_RejectResultFromWorker(const boost::system::error_code& e, connection_ptr conn);
   private:
+    //!> static entity to indicate to clients that the queue is empty.
+    static FragmentJob::ptr NoJob;
+
     //!> address of new Worker
     WorkerAddress address;
@@ -86 +97 @@
     WorkerPool &pool;
 
-    /// result that is received from the client.
+    //!> result that is received from the client.
     FragmentResult::ptr result;
 
@@ -92 +103 @@
     boost::function<void (const WorkerAddress&, FragmentJob::ptr&)> callback_sendJobToWorker;
 
-    // static entity to indicate to clients that the queue is empty.
-    static FragmentJob::ptr NoJob;
+    //!> choice
+    enum WorkerChoices choice;
   };
 
@@ -136 +147 @@
     FragmentQueue & JobsQueue;
 
-    /// bunch of jobs received from controller before placed in JobsQueue
+    //!> bunch of jobs received from controller before placed in JobsQueue
     std::vector<FragmentJob::ptr> jobs;
 
-    /// number of jobs that are waiting to be and are calculated, required for returning status
+    //!> number of jobs that are waiting to be and are calculated, required for returning status
     std::vector<size_t> jobInfo;
 
-    // choice
+    //!> choice
     enum ControllerChoices choice;
 
@@ -154 +165 @@
 
 private:
-  /// Queue with data to be sent to each client.
+  //!> Queue with data to be sent to each client.
   FragmentQueue JobsQueue;
 

src/Fragmentation/Automation/Pool/WorkerPool.hpp

@@ -55 +55 @@
 
   Idle_Queue_t::iterator getIdleWorker(const WorkerAddress &address);
-  Idle_Queue_t::iterator getIdleEnd()
-  {
-    return idle_queue.end();
-  }
   void markWorkerBusy(Idle_Queue_t::iterator &iter);