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

Timestamp:

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

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/FragmentWorker.cpp (modified) (5 diffs)

src/Fragmentation/Automation/FragmentWorker.cpp

@@ -34 +34 @@
 #include "CodePatterns/Log.hpp"
 #include "FragmentJob.hpp"
+#include "FragmentResult.hpp"
 #include "FragmentWorker.hpp"
+
+FragmentResult FragmentWorker::EmptyResult(JobId::NoJob, std::string("EmptyResult"));
 
 /// Constructor starts the asynchronous connect operation.
@@ -64 +67 @@
     boost::asio::ip::tcp::resolver::iterator endpoint_iterator)
 {
-    Info info(__FUNCTION__);
+  Info info(__FUNCTION__);
   if (!e)
   {
@@ -70 +73 @@
     // of jobs. The connection::async_read() function will automatically
     // decode the data that is read from the underlying socket.
+    LOG(1, "INFO: Receiving a job ...");
     connection_.async_read(job,
       boost::bind(&FragmentWorker::handle_read, this,
@@ -95 +99 @@
   if (!e)
   {
+    LOG(1, "INFO: Received job #" << job.getId() << ".");
     if (job.getId() != JobId::NoJob) {
       // Print out the data that was received.
       std::cout << "Job output: " << job.outputfile << "\n";
       std::cout << "Job id: " << job.getId() << "\n";
+
       // do something .. right now: wait
-      result = job.Work();
+      LOG(1, "INFO: Calculating job #" << job.getId() << " ...");
+      FragmentResult result(job.Work());
+
+      // write the result to the server
+      LOG(1, "INFO: Sending result for job #" << job.getId() << " ...");
+      connection_.async_write(result,
+        boost::bind(&FragmentWorker::handle_write, this,
+        boost::asio::placeholders::error));
+
     } else {
       std::cout << "The server has no job for me." << std::endl;
+      // send out empty result
+      LOG(1, "INFO: Sending empty result ...");
+      connection_.async_write(EmptyResult,
+        boost::bind(&FragmentWorker::handle_write, this,
+        boost::asio::placeholders::error));
       Exitflag = NoJobFlag;
     }
@@ -117 +136 @@
 }
 
+/// Handle completion of a write operation.
+void FragmentWorker::handle_write(const boost::system::error_code& e)
+{
+  Info info(__FUNCTION__);
+  // Nothing to do.
+  LOG(1, "INFO: Job #" << job.getId() << " calculated and sent.");
+  // erase job
+  job = FragmentJob();
+}
+