Changeset ef2767 for src

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.

Location:

src/Fragmentation/Automation

Files:

• FragmentScheduler.cpp (modified) (6 diffs)
• FragmentScheduler.hpp (modified) (2 diffs)
• FragmentWorker.cpp (modified) (5 diffs)
• FragmentWorker.hpp (modified) (3 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);
+}
+

src/Fragmentation/Automation/FragmentScheduler.hpp

@@ -20 +20 @@
 #include "FragmentJob.hpp"
 #include "FragmentQueue.hpp"
+#include "FragmentResult.hpp"
 
 /** FragmentScheduler serves FragmentJobs to Workers.
@@ -37 +38 @@
   void handle_write(const boost::system::error_code& e, connection_ptr conn);
 
+  /// Handle completion of a read operation.
+  void handle_read(const boost::system::error_code& e, connection_ptr conn);
+
 private:
   /// The acceptor object used to accept incoming socket connections.
   boost::asio::ip::tcp::acceptor acceptor_;
+
+  /// result that is received from the client.
+  FragmentResult result;
 
   /// Queue with data to be sent to each client.

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();
+}
+

src/Fragmentation/Automation/FragmentWorker.hpp

@@ -18 +18 @@
 #include "Connection.hpp"
 #include "FragmentJob.hpp"
-#include "FragmentResult.hpp"
 
 /** Receives a job from Server to execute and return FragmentResult.
@@ -37 +36 @@
   void handle_read(const boost::system::error_code& e);
 
+  /// Handle completion of a write operation.
+  void handle_write(const boost::system::error_code& e);
+
   enum Exitflag_t {
     OkFlag = 0,
@@ -52 +54 @@
 
 private:
+  // static entity of an empty result
+  static FragmentResult EmptyResult;
+
   /// The Connection to the server.
   Connection connection_;