Changeset 778abb for src/Fragmentation/Automation/FragmentScheduler.cpp

Timestamp:

May 4, 2012, 2:19:07 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:

e70b9d

Parents:

b9c486

git-author:

Frederik Heber <heber@…> (12/09/11 18:18:14)

git-committer:

Frederik Heber <heber@…> (05/04/12 14:19:07)

Message:

Added ResultGetter and capabilities to receive calculated results to FragmentController.

• Added enum (and file) ControllerChoices that defines the state of FragmentScheduler.
• depending on what is desired the Scheduler switches between these states and either receives or sends information. Requires new member variable choice because receival is of course asynchronous (see note in previous commit).
• FragmentController has additional functions connect_get() and handle_connect_get() to receive results.
• connect_calc() and connect_check() now just the choice whereas the actual sending and receiving is done in handle_... functions.
• handle_FinishOperation() is the common final callback function for all three of these functions.
• FragmentScheduler contains an internal list of delivered results.
• FragmentScheduler only initiates worker socket when jobs are present.
• FIX: FragmentScheduler does only send results that are done and only once.
• TESTFIX: Removed third Worker that receives NoJob as socket is powered down before because queue has run empty and we haven't add new jobs.

File:

• src/Fragmentation/Automation/FragmentScheduler.cpp (modified) (9 diffs)

src/Fragmentation/Automation/FragmentScheduler.cpp

@@ -57 +57 @@
   ),
   result(JobId::NoJob),
+  choice(NoOperation),
   Exitflag(OkFlag)
 {
   Info info(__FUNCTION__);
 
-  initiateWorkerSocket();
+  // only initiate socket if jobs are already present
+  if (JobsQueue.isJobPresent()) {
+    LOG(1, "Listening for workers on port " << workerport << ".");
+    initiateWorkerSocket();
+  }
 
   initiateControllerSocket();
+  LOG(1, "Listening for controller on port " << controllerport << ".");
 }
 
@@ -117 +123 @@
         boost::asio::placeholders::error, conn));
 
-      // Start an accept operation for a new Connection only when there
-      // are still jobs present otherwise we quit.
-      initiateWorkerSocket();
     } else {
       // send the static NoJob
@@ -139 +142 @@
     ELOG(0, e.message());
   }
+
+  // Start an accept operation for a new Connection only when there
+  // are still jobs present
+  if (JobsQueue.isJobPresent())
+    initiateWorkerSocket();
 }
 
@@ -169 +177 @@
 {
   Info info(__FUNCTION__);
-  // nothing to do
   LOG(1, "INFO: Received result for job #" << result.getId() << " ...");
-  // and push into queue
   ASSERT(result.getId() != (JobId_t)JobId::NoJob,
       "FragmentScheduler::handle_ReceiveResultFromWorker() - result received has NoJob id.");
   ASSERT(result.getId() != (JobId_t)JobId::IllegalJob,
       "FragmentScheduler::handle_ReceiveResultFromWorker() - result received has IllegalJob id.");
+  // place id into expected
   if ((result.getId() != (JobId_t)JobId::NoJob) && (result.getId() != (JobId_t)JobId::IllegalJob))
     JobsQueue.pushResult(result);
   // erase result
   result = FragmentResult(JobId::NoJob);
+  LOG(1, "INFO: JobsQueue has " << JobsQueue.getDoneJobs() << " results.");
 }
 
@@ -196 +204 @@
   if (!e)
   {
-    if (JobsQueue.isJobPresent() || !JobsQueue.getDoneJobs()) {
-      // The connection::async_write() function will automatically
-      // serialize the data structure for us.
-      LOG(1, "INFO: Receiving bunch of jobs from a controller ...");
-      conn->async_read(jobs,
-        boost::bind(&FragmentScheduler::handle_ReceiveJobs, this,
-        boost::asio::placeholders::error, conn));
-    } else {
-      // we just give a status report ...
-      // first update number
-      doneJobs = JobsQueue.getDoneJobs();
-      // now we accept connections to check for state of calculations
-      LOG(1, "INFO: Sending state of results to controller ...");
-      conn->async_write(doneJobs,
-        boost::bind(&FragmentScheduler::handle_CheckResultState, this,
-        boost::asio::placeholders::error, conn));
-    }
+    conn->async_read(choice,
+      boost::bind(&FragmentScheduler::handle_ReadChoice, this,
+      boost::asio::placeholders::error, conn));
   }
   else
@@ -224 +218 @@
 }
 
+/** 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::handle_ReadChoice(const boost::system::error_code& e, connection_ptr conn)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    // switch over the desired choice read previously
+    switch(choice) {
+    case NoOperation:
+    {
+      ELOG(1, "FragmentScheduler::handle_ReadChoice() - called with NoOperation.");
+      break;
+    }
+    case ReceiveJobs:
+      {
+        // The connection::async_write() function will automatically
+        // serialize the data structure for us.
+        LOG(1, "INFO: Receiving bunch of jobs from a controller ...");
+        conn->async_read(jobs,
+          boost::bind(&FragmentScheduler::handle_ReceiveJobs, this,
+          boost::asio::placeholders::error, conn));
+        break;
+      }
+    case CheckState:
+    {
+      // first update number
+      doneJobs = JobsQueue.getDoneJobs();
+      // now we accept connections to check for state of calculations
+      LOG(1, "INFO: Sending state that "+toString(doneJobs)+" jobs are done to controller ...");
+      conn->async_write(doneJobs,
+        boost::bind(&FragmentScheduler::handle_CheckResultState, this,
+        boost::asio::placeholders::error, conn));
+
+      initiateControllerSocket();
+      break;
+    }
+    case SendResults:
+    {
+      const std::vector<FragmentResult> results = JobsQueue.getAllResults();
+      // ... or we give the results
+      LOG(1, "INFO: Sending "+toString(results.size())+" results to controller ...");
+      conn->async_write(results,
+        boost::bind(&FragmentScheduler::handle_SendResults, this,
+        boost::asio::placeholders::error, conn));
+
+      initiateControllerSocket();
+      break;
+    }
+    default:
+      Exitflag = ControllerErrorFlag;
+      ELOG(1, "FragmentScheduler::handle_ReadChoice() - called with no valid choice.");
+      break;
+    }
+    // restore NoOperation choice such that choice is not read twice
+    choice = NoOperation;
+
+  }
+  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 = ControllerErrorFlag;
+    ELOG(0, e.message());
+  }
+}
+
 /** Controller callback function when job has been sent.
  *
+ * We check here whether the worker socket is accepting, if there
+ * have been no jobs we re-activate it, as it is shut down after
+ * last job.
+ *
  * \param e error code if something went wrong
  * \param conn reference with the connection
@@ -232 +301 @@
 {
   Info info(__FUNCTION__);
+  bool initiateSocket = !JobsQueue.isJobPresent();
+
   // jobs are received, hence place in JobsQueue
   if (!jobs.empty()) {
     LOG(1, "INFO: Pushing " << jobs.size() << " jobs into queue.");
     JobsQueue.pushJobs(jobs);
-  }
-  // launch new acceptor of queue has been filled/is full
-  if (JobsQueue.isJobPresent()) {
-    // Start an accept operation for a new Connection.
+    // initiate socket if we had no jobs before
+    if (initiateSocket)
+      initiateWorkerSocket();
+    // launch new acceptor of queue has been filled/is full
     initiateControllerSocket();
   } else {
@@ -246 +317 @@
 
   jobs.clear();
+
 }
 
@@ -259 +331 @@
   LOG(1, "INFO: Sent that " << doneJobs << " jobs are done.");
 }
+
+/** Controller callback function when result has been received.
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::handle_SendResults(const boost::system::error_code& e, connection_ptr conn)
+{
+  Info info(__FUNCTION__);
+  // do nothing
+  LOG(1, "INFO: Results have been sent.");
+}
+