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

Timestamp:

May 4, 2012, 2:19:07 PM (14 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, Candidate_v1.7.0, Candidate_v1.7.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/FragmentController.cpp (modified) (9 diffs)

src/Fragmentation/Automation/FragmentController.cpp

@@ -24 +24 @@
 
 #include <boost/bind.hpp>
+#include <boost/foreach.hpp>
 #include <iostream>
 #include <vector>
@@ -32 +33 @@
 #include "FragmentJob.hpp"
 #include "FragmentResult.hpp"
+#include "ControllerChoices.hpp"
 
 #include "FragmentController.hpp"
@@ -70 +72 @@
   if (!e)
   {
-    // Successfully established connection. Start operation to read the vector
-    // of jobs. The connection::async_write() function will automatically
-    // encode the data that is written to the underlying socket.
-    LOG(1, "INFO: Sending "+toString(jobs.size())+" jobs ...");
-    connection_.async_write(jobs,
+    // Successfully established connection. Give choice.
+    enum ControllerChoices choice = ReceiveJobs;
+    connection_.async_write(choice,
       boost::bind(&FragmentController::handle_SendJobs, this,
       boost::asio::placeholders::error));
@@ -104 +104 @@
   if (!e)
   {
-    // Successfully established connection. Start operation to read the vector
-    // of jobs. The connection::async_write() function will automatically
-    // encode the data that is written to the underlying socket.
-    LOG(1, "INFO: Checking number of done jobs ...");
-    connection_.async_read(doneJobs,
+    // Successfully established connection. Give choice.
+    enum ControllerChoices choice = CheckState;
+    connection_.async_write(choice,
       boost::bind(&FragmentController::handle_ReceiveDoneJobs, this,
       boost::asio::placeholders::error));
@@ -127 +125 @@
 }
 
-/** Callback function when jobs have been sent.
- *
- * \param e error code if something went wrong
- */
-void FragmentController::handle_SendJobs(const boost::system::error_code& e)
-{
-  Info info(__FUNCTION__);
-  if (!e)
-  {
-    LOG(1, "INFO: Sent jobs ...");
+/** Handle completion of a connect operation.
+ *
+ * \param e error code if something went wrong
+ * \param endpoint_iterator endpoint of the connection
+ */
+void FragmentController::handle_connect_get(const boost::system::error_code& e,
+    boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    // Successfully established connection. Give choice.
+    enum ControllerChoices choice = SendResults;
+    connection_.async_write(choice,
+      boost::bind(&FragmentController::handle_ReceivingResults, this,
+      boost::asio::placeholders::error));
+  } else if (endpoint_iterator != boost::asio::ip::tcp::resolver::iterator()) {
+    // Try the next endpoint.
+    connection_.socket().close();
+    boost::asio::ip::tcp::endpoint endpoint = *endpoint_iterator;
+    connection_.socket().async_connect(endpoint,
+      boost::bind(&FragmentController::handle_connect_check, this,
+      boost::asio::placeholders::error, ++endpoint_iterator));
+  } 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.
+    Exitflag = ErrorFlag;
+    ELOG(1, e.message());
+  }
+}
+
+/** Callback function when operation has been completed.
+ *
+ * \param e error code if something went wrong
+ */
+void FragmentController::handle_FinishOperation(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    LOG(1, "INFO: Operation completed.");
   }
   else
@@ -149 +179 @@
 }
 
+/** Callback function when jobs have been sent.
+ *
+ * \param e error code if something went wrong
+ */
+void FragmentController::handle_SendJobs(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    // Successfully established connection. Start operation to read the vector
+    // of jobs. The connection::async_write() function will automatically
+    // encode the data that is written to the underlying socket.
+    LOG(1, "INFO: Sending "+toString(jobs.size())+" jobs ...");
+    connection_.async_write(jobs,
+      boost::bind(&FragmentController::handle_FinishOperation, this,
+      boost::asio::placeholders::error));
+  }
+  else
+  {
+    // An error occurred.
+    Exitflag = ErrorFlag;
+    ELOG(1, e.message());
+  }
+
+  // Since we are not starting a new operation the io_service will run out of
+  // work to do and the client will exit.
+}
+
+/** Callback function when results have been received.
+ *
+ * \param e error code if something went wrong
+ */
+void FragmentController::handle_ReceivingResults(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    // The connection::async_read() function will automatically
+    // decode the data that is written to the underlying socket.
+    connection_.async_read(results,
+      boost::bind(&FragmentController::handle_ReceivedResults, this,
+      boost::asio::placeholders::error));
+  }
+  else
+  {
+    // An error occurred.
+    Exitflag = ErrorFlag;
+    ELOG(1, e.message());
+  }
+
+  // Since we are not starting a new operation the io_service will run out of
+  // work to do and the client will exit.
+}
+
 /** Callback function when doneJobs have been received.
  *
  * \param e error code if something went wrong
  */
+void FragmentController::handle_ReceivedResults(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+
+  LOG(1, "INFO: Received "+toString(results.size())+" results ...");
+
+  handle_FinishOperation(e);
+}
+
+/** Callback function when doneJobs have been received.
+ *
+ * \param e error code if something went wrong
+ */
 void FragmentController::handle_ReceiveDoneJobs(const boost::system::error_code& e)
 {
   Info info(__FUNCTION__);
-  // Nothing to do.
-  LOG(1, "INFO: "+toString(doneJobs)+" jobs are currently done.");
+  if (!e)
+  {
+    // The connection::async_read() function will automatically
+    // decode the data that is written to the underlying socket.
+    LOG(1, "INFO: Checking number of done jobs ...");
+    connection_.async_read(doneJobs,
+      boost::bind(&FragmentController::handle_FinishOperation, this,
+      boost::asio::placeholders::error));
+  }
+  else
+  {
+    // An error occurred.
+    Exitflag = ErrorFlag;
+    ELOG(1, e.message());
+  }
 }
 
@@ -211 +321 @@
 }
 
+/** Internal function to connect to the endpoint of the server asynchronuously.
+ *
+ * We require internal connetion_ and host and service to be set up for this.
+ */
+void FragmentController::connect_get()
+{
+  Info info(__FUNCTION__);
+  // Resolve the host name into an IP address.
+  boost::asio::ip::tcp::resolver::iterator endpoint_iterator = getEndpointIterator();
+  boost::asio::ip::tcp::endpoint endpoint = *endpoint_iterator;
+
+  // Start an asynchronous connect operation.
+  std::cout << "Connecting to endpoint " << endpoint << " to get results " << std::endl;
+  connection_.socket().async_connect(endpoint,
+    boost::bind(&FragmentController::handle_connect_get, this,
+      boost::asio::placeholders::error, ++endpoint_iterator));
+}
+
 /** Internal function to disconnect connection_ correctly.
  *
@@ -225 +353 @@
 void FragmentController::addJobs(const std::vector<FragmentJob> &_jobs)
 {
-  jobs = _jobs;
+  jobs.reserve(jobs.size()+_jobs.size());
+  BOOST_FOREACH(FragmentJob job, _jobs) {
+    jobs.push_back(job);
+  }
 }
 
@@ -250 +381 @@
 }
 
+/** Getter for results.
+ *
+ * \sa calculateResults()
+ * \return vector of results for the added jobs (\sa addJobs()).
+ */
+std::vector<FragmentResult> FragmentController::getResults()
+{
+  Info info(__FUNCTION__);
+  return results;
+}
+
+/** Function to initiate receival of results.
+ *
+ */
+void FragmentController::obtainResults()
+{
+  // connect
+  connect_get();
+  //disconnect
+  disconnect();
+}
+
 /** Getter for doneJobs.
  *