Changeset ef2767 for src/Fragmentation/Automation/FragmentScheduler.cpp

Timestamp:

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

083490

Parents:

630a6e

git-author:

Frederik Heber <heber@…> (11/27/11 18:38:26)

git-committer:

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

Message:

Server and Worker now also correctly exchange result.

• Server uses FragmentQueue which also stores the obtained result.
• Worker calls received FragmentJob::Work() function and returns the thereby calculated FragmentResult.
• Regression test for Server/Worker now checks whether job #1 is listed twice, this basically checks whether the result has been exchanged.

There have been some concepts to understand to get this working and these

shall be briefly recorded here:

• asynchronous communication can only work on objects that live beyond the scope of where they have been called, e.g. therefore FragmentScheduler contains a FragmentResult and FragmentWorker a FragmentJob instance. They receive this object and need the write access in the scope of the aync. comm. and not of the caller's scope. This is probably because the initial argument is only used to set up a buffer of correct length. However, the received instance is created/deserialized first when the communication is completed. And this may well be after the caller's scope has been left.
• This is different to read operations as probably there the object to send is immediately serialized and placed into an internal buffer such that later access is only to this buffer and not to the original instance which therefore does not need to exist anymore. That's why the above Schedulder and Worker do not have the "other" instance as class members as well.
• chaining asynchronous communications, e.g. a write after a read has been performed, can only be done by using the callback functions that the async_write/read gets as parameters. They are called when the one operation has finished and therein the next operation can then be launched. This way a successful chain is executed.

File:

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

src/Fragmentation/Automation/FragmentScheduler.cpp

@@ -35 +35 @@
 #include "CodePatterns/Log.hpp"
 #include "FragmentJob.hpp"
+#include "JobId.hpp"
 
 #include "FragmentScheduler.hpp"
@@ -43 +44 @@
   acceptor_(io_service,
       boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)
-  )
+  ),
+  result(JobId::NoJob)
 {
   Info info(__FUNCTION__);
@@ -60 +62 @@
 {
   Info info(__FUNCTION__);
-  std::cout << "handle_accept called." << std::endl;
   if (!e)
   {
@@ -67 +68 @@
     if (JobsQueue.isJobPresent()) {
       // pop a job and send it to the client.
-      FragmentJob s(JobsQueue.popJob());
+      FragmentJob job(JobsQueue.popJob());
       // The connection::async_write() function will automatically
       // serialize the data structure for us.
-      conn->async_write(s,
+      LOG(1, "INFO: Sending job #" << job.getId() << ".");
+      conn->async_write(job,
         boost::bind(&FragmentScheduler::handle_write, this,
         boost::asio::placeholders::error, conn));
@@ -85 +87 @@
         boost::bind(&FragmentScheduler::handle_write, this,
         boost::asio::placeholders::error, conn));
+
+      // then there must be no read necesary
 
       ELOG(2, "There is currently no job present in the queue.");
@@ -102 +106 @@
 {
     Info info(__FUNCTION__);
-  // Nothing to do. The socket will be closed automatically when the last
-  // reference to the connection object goes away.
+    LOG(1, "INFO: Job sent.");
+    // obtain result
+    LOG(1, "INFO: Receiving result for a job ...");
+    conn->async_read(result,
+      boost::bind(&FragmentScheduler::handle_read, this,
+      boost::asio::placeholders::error, conn));
 }
 
+/// Handle completion of a read operation.
+void FragmentScheduler::handle_read(const boost::system::error_code& e, connection_ptr conn)
+{
+    Info info(__FUNCTION__);
+    // nothing to do
+    LOG(1, "INFO: Received result for job #" << result.getId() << " ...");
+    // and push into queue
+    ASSERT(result.getId() != JobId::NoJob,
+        "FragmentScheduler::handle_write() - result received has NoJob id.");
+    ASSERT(result.getId() != JobId::IllegalJob,
+        "FragmentScheduler::handle_write() - result received has IllegalJob id.");
+    if ((result.getId() != JobId::NoJob) && (result.getId() != JobId::IllegalJob))
+      JobsQueue.pushResult(result);
+    // erase result
+    result = FragmentResult(JobId::NoJob);
+}
+