Changeset 0542e2 for ThirdParty/JobMarket/src

Timestamp:

Apr 5, 2018, 1:19:33 PM (7 years ago)

Author:

Frederik Heber <frederik.heber@…>

Branches:

AutomationFragmentation_failures, Candidate_v1.6.1, ChemicalSpaceEvaluator, Enhanced_StructuralOptimization_continued, Exclude_Hydrogens_annealWithBondGraph, ForceAnnealing_with_BondGraph, ForceAnnealing_with_BondGraph_contraction-expansion, Gui_displays_atomic_force_velocity, JobMarket_RobustOnKillsSegFaults, JobMarket_StableWorkerPool, PythonUI_with_named_parameters, StoppableMakroAction, TremoloParser_IncreasedPrecision

Children:

456f19

Parents:

1181a3

git-author:

Frederik Heber <frederik.heber@…> (04/05/18 12:24:11)

git-committer:

Frederik Heber <frederik.heber@…> (04/05/18 13:19:33)

Message:

Added GetNumberWorkersOperation, Controller returns number of workers.

• TEST: Added regression test on this new functionality.

Location:

ThirdParty/JobMarket/src/JobMarket

Files:

• Controller/FragmentController.cpp (modified) (4 diffs)
• Controller/FragmentController.hpp (modified) (1 diff)
• Controller/controller_main.cpp (modified) (2 diffs)
• ControllerChoices.hpp (modified) (1 diff)
• FragmentScheduler.hpp (modified) (1 diff)
• FragmentScheduler_ControllerListener.cpp (modified) (4 diffs)
• Makefile.am (modified) (2 diffs)
• Operations/Controllers/GetNumberWorkersOperation.cpp (added)
• Operations/Controllers/GetNumberWorkersOperation.hpp (added)

ThirdParty/JobMarket/src/JobMarket/Controller/FragmentController.cpp

@@ -31 +31 @@
 #include "Operations/Controllers/CheckResultsOperation.hpp"
 #include "Operations/Controllers/GetNextJobIdOperation.hpp"
+#include "Operations/Controllers/GetNumberWorkersOperation.hpp"
 #include "Operations/Controllers/ReceiveResultsOperation.hpp"
 #include "Operations/Controllers/RemoveAllJobsOperation.hpp"
@@ -56 +57 @@
   Commands.registerInstance(new CheckResultsOperation(connection_, AsyncOperation::NoOpCallback, failed));
   Commands.registerInstance(new GetNextJobIdOperation(connection_, AsyncOperation::NoOpCallback, failed));
+  Commands.registerInstance(new GetNumberWorkersOperation(connection_, AsyncOperation::NoOpCallback, failed));
   Commands.registerInstance(new ReceiveResultsOperation(connection_, AsyncOperation::NoOpCallback, failed));
   Commands.registerInstance(new RemoveAllJobsOperation(connection_));
@@ -162 +164 @@
 }
 
+/** Obtains number of workers enrolled at server
+ *
+ * @param host address of server
+ * @param service port/service of server
+ */
+void FragmentController::checkEnrolledWorkers(
+    const std::string &host,
+    const std::string &service)
+{
+  GetNumberWorkersOperation *checkres = static_cast<GetNumberWorkersOperation *>(
+      Commands.getByName("numberworkers"));
+  (*checkres)(host, service);
+}
+
 /** Return scheduled and done jobs.
  *
@@ -173 +189 @@
   const size_t presentJobs = checkres->getPresentJobs();
   return make_pair(presentJobs, doneJobs);
+}
+
+/** Return scheduled and done jobs.
+ *
+ * @return number of workers in certain states enrolled in server
+ */
+std::vector<size_t> FragmentController::getNumberOfWorkers()
+{
+  const GetNumberWorkersOperation * const checkres = static_cast<const GetNumberWorkersOperation *>(
+      Commands.getByName("numberworkers"));
+  return checkres->get();
 }
 

ThirdParty/JobMarket/src/JobMarket/Controller/FragmentController.hpp

@@ -49 +49 @@
   void checkResults(const std::string &host, const std::string &service);
   std::pair<size_t, size_t> getJobStatus() const;
+  void checkEnrolledWorkers(const std::string &host, const std::string &service);
+  std::vector<size_t> getNumberOfWorkers();
   void removeWaitingResults(const std::string &host, const std::string &service);
   void removeWaitingJobs(const std::string &host, const std::string &service);

ThirdParty/JobMarket/src/JobMarket/Controller/controller_main.cpp

@@ -57 +57 @@
 }
 
+/** Print the status of workers
+ *
+ * @param number_workers vector of workers in certain states (busy, idle, ...)
+ */
+void printNumberWorkers(const std::vector<size_t> &number_workers)
+{
+  ASSERT(number_workers.size() == 2,
+      "printNumberWorkers() - vector of worker numbers is not of size 2.");
+  LOG(1, "INFO: " << number_workers[0] << " are busy and "
+      << number_workers[1] << " are idle.");
+}
+
 inline std::vector<std::string> getListOfCommands(const ControllerCommandRegistry &ControllerCommands)
 {
@@ -96 +108 @@
         (boost::bind(&printJobStatus,
             boost::bind(&FragmentController::getJobStatus, boost::ref(controller))))
+      ));
+  registrator(new ControllerCommand("numberworkers",
+      boost::assign::list_of< ControllerCommand::commands_t >
+        (boost::bind(&FragmentController::checkEnrolledWorkers,
+            boost::ref(controller), boost::cref(ControllerInfo->server), boost::cref(ControllerInfo->serverport)))
+        (boost::bind(&printNumberWorkers,
+            boost::bind(&FragmentController::getNumberOfWorkers, boost::ref(controller))))
       ));
   registrator(new ControllerCommand("removeall",

ThirdParty/JobMarket/src/JobMarket/ControllerChoices.hpp

@@ -24 +24 @@
   EmptyJobQueue,
   RemoveAll,
-  ShutdownControllerSocket
+  ShutdownControllerSocket,
+  NumberOfWorkers
 };
 

ThirdParty/JobMarket/src/JobMarket/FragmentScheduler.hpp

@@ -205 +205 @@
     void handle_SendResults(const boost::system::error_code& e, connection_ptr conn);
 
+    /// Controller callback function when result are to be sent.
+    void handle_SendNumberWorkers(const boost::system::error_code& e, connection_ptr conn);
+
     /// Controller callback function when results has been sent.
     void handle_finishSendResults(const boost::system::error_code& e, connection_ptr conn);

ThirdParty/JobMarket/src/JobMarket/FragmentScheduler_ControllerListener.cpp

@@ -25 +25 @@
 
 #include <algorithm>
+#include <boost/assign.hpp>
 #include <boost/bind.hpp>
 #include <sstream>
@@ -43 +44 @@
 
 #include "JobMarket/FragmentScheduler.hpp"
+
+using namespace boost::assign;
 
 /** Handle a new controller connection.
@@ -138 +141 @@
     {
       removeallWorkers();
+      break;
+    }
+    case NumberOfWorkers:
+    {
+      LOG(1, "INFO: Sending total number of workers to a controller ...");
+      std::vector<size_t> number_workers;
+      number_workers += pool.getNoBusyWorkers();
+      number_workers += pool.getNoIdleWorkers();
+      conn->async_write(number_workers,
+        boost::bind(&FragmentScheduler::ControllerListener_t::handle_SendNumberWorkers, this,
+        boost::asio::placeholders::error, conn));
       break;
     }
@@ -346 +360 @@
   LOG(1, "INFO: Results have been sent.");
 }
+
+/** Controller callback function when number of workers has been sent.
+ *
+ * \param e error code if something went wrong
+ * \param conn reference with the connection
+ */
+void FragmentScheduler::ControllerListener_t::handle_SendNumberWorkers(const boost::system::error_code& e, connection_ptr conn)
+{
+  DEBUG_FUNCTION_ENTRYEXIT
+  // do nothing
+  LOG(1, "INFO: Number of workers have been sent.");
+}

ThirdParty/JobMarket/src/JobMarket/Makefile.am

@@ -40 +40 @@
         Operations/Controllers/CheckResultsOperation.cpp \
         Operations/Controllers/GetNextJobIdOperation.cpp \
+        Operations/Controllers/GetNumberWorkersOperation.cpp \
         Operations/Controllers/ReceiveResultsOperation.cpp \
         Operations/Controllers/RemoveAllJobsOperation.cpp \
@@ -63 +64 @@
         Operations/Controllers/CheckResultsOperation.hpp \
         Operations/Controllers/GetNextJobIdOperation.hpp \
+        Operations/Controllers/GetNumberWorkersOperation.hpp \
         Operations/Controllers/ReceiveResultsOperation.hpp \
         Operations/Controllers/RemoveAllJobsOperation.hpp \