Changeset db03d9 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:

3c4a5e

Parents:

8ee5ac

git-author:

Frederik Heber <heber@…> (11/27/11 23:20:43)

git-committer:

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

Message:

Added possibility to add jobs via a FragmentController.

• Server listens on another port for a controller to connect, also now has an Exitflag.
• Controller can add a bunch of jobs.
• JobAdder can successfully add a number jobs to Server, if 0 are sent, the socket shuts down.
• TESTFIX: extended regression Fragmentation/Automation test to use Jobadder as by default Server starts with no jobs in the queue.

File:

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

src/Fragmentation/Automation/FragmentScheduler.cpp

@@ -41 +41 @@
 FragmentJob FragmentScheduler::NoJob(std::string("NoJob"), JobId::NoJob);
 
-FragmentScheduler::FragmentScheduler(boost::asio::io_service& io_service, unsigned short port) :
-  acceptor_(io_service,
-      boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)
+/** Constructor of class FragmentScheduler.
+ *
+ * We setup both acceptors to accept connections from workers and Controller.
+ *
+ * \param io_service io_service of the asynchronous communications
+ * \param workerport port to listen for worker connections
+ * \param controllerport port to listen for controller connections.
+ */
+FragmentScheduler::FragmentScheduler(boost::asio::io_service& io_service, unsigned short workerport, unsigned short controllerport) :
+  worker_acceptor_(io_service,
+      boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), workerport)
   ),
-  result(JobId::NoJob)
-{
-  Info info(__FUNCTION__);
-  FragmentJob s(std::string("test"), 1);
-  JobsQueue.pushJob(s);
-
-  // Start an accept operation for a new connection.
-  connection_ptr new_conn(new Connection(acceptor_.get_io_service()));
-  acceptor_.async_accept(new_conn->socket(),
-    boost::bind(&FragmentScheduler::handle_accept, this,
-      boost::asio::placeholders::error, new_conn));
-}
-
-/// Handle completion of a accept operation.
-void FragmentScheduler::handle_accept(const boost::system::error_code& e, connection_ptr conn)
+  controller_acceptor_(io_service,
+      boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), controllerport)
+  ),
+  result(JobId::NoJob),
+  Exitflag(OkFlag)
+{
+  Info info(__FUNCTION__);
+  {
+    // Start an accept operation for worker connections.
+    connection_ptr new_conn(new Connection(worker_acceptor_.get_io_service()));
+    worker_acceptor_.async_accept(new_conn->socket(),
+      boost::bind(&FragmentScheduler::handle_AcceptWorker, this,
+        boost::asio::placeholders::error, new_conn));
+  }
+
+  {
+    // Start an accept operation for controller connection.
+    connection_ptr new_conn(new Connection(controller_acceptor_.get_io_service()));
+    controller_acceptor_.async_accept(new_conn->socket(),
+      boost::bind(&FragmentScheduler::handle_AcceptController, this,
+        boost::asio::placeholders::error, new_conn));
+  }
+}
+
+/** Handle a new worker connection.
+ *
+ * We check whether jobs are in the JobsQueue. If present, job is sent.
+ *
+ * \sa handle_SendJobtoWorker()
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::handle_AcceptWorker(const boost::system::error_code& e, connection_ptr conn)
 {
   Info info(__FUNCTION__);
@@ -73 +100 @@
       LOG(1, "INFO: Sending job #" << job.getId() << ".");
       conn->async_write(job,
-        boost::bind(&FragmentScheduler::handle_SendJob, this,
+        boost::bind(&FragmentScheduler::handle_SendJobtoWorker, this,
         boost::asio::placeholders::error, conn));
 
       // Start an accept operation for a new Connection only when there
       // are still jobs present otherwise we quit.
-      connection_ptr new_conn(new Connection(acceptor_.get_io_service()));
-      acceptor_.async_accept(new_conn->socket(),
-        boost::bind(&FragmentScheduler::handle_accept, this,
+      connection_ptr new_conn(new Connection(worker_acceptor_.get_io_service()));
+      worker_acceptor_.async_accept(new_conn->socket(),
+        boost::bind(&FragmentScheduler::handle_AcceptWorker, this,
         boost::asio::placeholders::error, new_conn));
     } else {
       // send the static NoJob
       conn->async_write(NoJob,
-        boost::bind(&FragmentScheduler::handle_SendJob, this,
+        boost::bind(&FragmentScheduler::handle_SendJobtoWorker, this,
         boost::asio::placeholders::error, conn));
 
@@ -102 +129 @@
 }
 
-/// Callback function when job has been sent.
-void FragmentScheduler::handle_SendJob(const boost::system::error_code& e, connection_ptr conn)
+/** Callback function when job has been sent.
+ *
+ * After job has been sent we start async_read() for the result.
+ *
+ * \sa handle_ReceiveResultFromWorker()
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::handle_SendJobtoWorker(const boost::system::error_code& e, connection_ptr conn)
 {
     Info info(__FUNCTION__);
@@ -110 +145 @@
     LOG(1, "INFO: Receiving result for a job ...");
     conn->async_read(result,
-      boost::bind(&FragmentScheduler::handle_ReceiveResult, this,
+      boost::bind(&FragmentScheduler::handle_ReceiveResultFromWorker, this,
       boost::asio::placeholders::error, conn));
 }
 
-/// Callback function when result has been received.
-void FragmentScheduler::handle_ReceiveResult(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);
-}
-
+/** Callback function when result has been received.
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::handle_ReceiveResultFromWorker(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_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.");
+  if ((result.getId() != (JobId_t)JobId::NoJob) && (result.getId() != (JobId_t)JobId::IllegalJob))
+    JobsQueue.pushResult(result);
+  // erase result
+  result = FragmentResult(JobId::NoJob);
+}
+
+/** Handle a new controller connection.
+ *
+ * \sa handle_ReceiveJobs()
+ * \sa handle_CheckResultState()
+ * \sa handle_SendResults()
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::handle_AcceptController(const boost::system::error_code& e, connection_ptr conn)
+{
+  Info info(__FUNCTION__);
+  if (!e)
+  {
+    if (jobs.empty()) {
+      // 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
+  {
+    // 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 = ErrorFlag;
+    ELOG(0, e.message());
+  }
+}
+
+/** Controller callback function when job has been sent.
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::handle_ReceiveJobs(const boost::system::error_code& e, connection_ptr conn)
+{
+  Info info(__FUNCTION__);
+  // 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.
+    connection_ptr new_conn(new Connection(controller_acceptor_.get_io_service()));
+    controller_acceptor_.async_accept(new_conn->socket(),
+      boost::bind(&FragmentScheduler::handle_AcceptController, this,
+      boost::asio::placeholders::error, new_conn));
+  } else {
+    LOG(1, "INFO: Shutting down controller socket.");
+  }
+
+  jobs.clear();
+}
+