Changeset d1dbfc for src


Ignore:
Timestamp:
Jun 27, 2012, 4:07:20 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:
554809
Parents:
6f2bc7
git-author:
Frederik Heber <heber@…> (02/16/12 15:09:52)
git-committer:
Frederik Heber <heber@…> (06/27/12 16:07:20)
Message:

New GetNextJobIdOperation for obtaining next available JobId from server.

  • as JobId is required to create the job, we now need two, separate communication phases: gathering info (ids) and sending info (jobs).
  • new SchedulderState GetNextJobId along with handlers.
  • new GetNextJobIdOperation that requests another id which is internally stored into a list, along with a getter that extracts them one by one.
  • controller's createjobs() and parsejobs() each take an nextid parameter now.
Location:
src/Fragmentation/Automation
Files:
2 added
8 edited

Legend:

Unmodified
Added
Removed

  • src/Fragmentation/Automation/Controller/FragmentController.cpp

    r6f2bc7 rd1dbfc  
    3030#include "Controller/Commands/Operation.hpp"
    3131#include "Controller/Commands/CheckResultsOperation.hpp"
     32#include "Controller/Commands/GetNextJobIdOperation.hpp"
    3233#include "Controller/Commands/ReceiveJobsOperation.hpp"
    3334#include "Controller/Commands/SendResultsOperation.hpp"
    … …  
    4849  // insert commands into registry
    4950  Commands.registerInstance(new CheckResultsOperation(connection_));
     51  Commands.registerInstance(new GetNextJobIdOperation(connection_));
    5052  Commands.registerInstance(new ReceiveJobsOperation(connection_));
    5153  Commands.registerInstance(new SendResultsOperation(connection_));

  • src/Fragmentation/Automation/ControllerChoices.hpp

    r6f2bc7 rd1dbfc  
    1818enum ControllerChoices {
    1919  NoOperation,
     20  GetNextJobId,
    2021  ReceiveJobs,
    2122  CheckState,

  • src/Fragmentation/Automation/FragmentScheduler.cpp

    r6f2bc7 rd1dbfc  
    6868  jobInfo((size_t)2, 0),
    6969  choice(NoOperation),
     70  globalId(0),
    7071  Exitflag(OkFlag)
    7172{
    … …  
    240241  {
    241242    bool LaunchNewAcceptor = true;
     243    LOG(1, "INFO: Received request for operation " << choice << ".");
    242244    // switch over the desired choice read previously
    243245    switch(choice) {
    … …  
    247249      break;
    248250    }
     251    case GetNextJobId:
     252    {
     253      const JobId_t nextid = globalId.getNextId();
     254      LOG(1, "INFO: Sending next available job id " << nextid << " to controller ...");
     255      conn->async_write(nextid,
     256        boost::bind(&FragmentScheduler::handle_GetNextJobIdState, this,
     257        boost::asio::placeholders::error, conn));
     258      break;
     259    }
    249260    case ReceiveJobs:
    250       {
    251         // The connection::async_write() function will automatically
    252         // serialize the data structure for us.
    253         LOG(1, "INFO: Receiving bunch of jobs from a controller ...");
    254         conn->async_read(jobs,
    255           boost::bind(&FragmentScheduler::handle_ReceiveJobs, this,
    256           boost::asio::placeholders::error, conn));
    257         break;
    258       }
     261    {
     262      // The connection::async_write() function will automatically
     263      // serialize the data structure for us.
     264      LOG(1, "INFO: Receiving bunch of jobs from a controller ...");
     265      conn->async_read(jobs,
     266        boost::bind(&FragmentScheduler::handle_ReceiveJobs, this,
     267        boost::asio::placeholders::error, conn));
     268      break;
     269    }
    259270    case CheckState:
    260271    {
    … …  
    350361}
    351362
     363/** Controller callback function when checking on state of results.
     364 *
     365 * \param e error code if something went wrong
     366 * \param conn reference with the connection
     367 */
     368void FragmentScheduler::handle_GetNextJobIdState(const boost::system::error_code& e, connection_ptr conn)
     369{
     370  Info info(__FUNCTION__);
     371  // do nothing
     372  LOG(1, "INFO: Sent next available job id.");
     373}
     374
    352375/** Controller callback function when result has been received.
    353376 *

  • src/Fragmentation/Automation/FragmentScheduler.hpp

    r6f2bc7 rd1dbfc  
    1919#include "Connection.hpp"
    2020#include "ControllerChoices.hpp"
    21 #include "ControllerChoices.hpp"
    2221#include "FragmentQueue.hpp"
     22#include "GlobalJobId.hpp"
    2323#include "Jobs/FragmentJob.hpp"
    2424#include "Results/FragmentResult.hpp"
    … …  
    7575  void handle_CheckResultState(const boost::system::error_code& e, connection_ptr conn);
    7676
     77  /// Controller callback function when checking on state of results.
     78  void handle_GetNextJobIdState(const boost::system::error_code& e, connection_ptr conn);
     79
    7780  /// Controller callback function when result has been received.
    7881  void handle_SendResults(const boost::system::error_code& e, connection_ptr conn);
    … …  
    109112  static FragmentJob::ptr NoJob;
    110113
     114  // TODO: replace this instance by a IdPool.
     115  //!> global id to give next available job id
     116  GlobalJobId globalId;
     117
    111118  // Exit flag on program exit
    112119  enum Exitflag_t Exitflag;

  • src/Fragmentation/Automation/GlobalJobId.cpp

    r6f2bc7 rd1dbfc  
    2222#include "GlobalJobId.hpp"
    2323
    24 GlobalJobId::GlobalJobId() :
    25     JobId(0)
     24GlobalJobId::GlobalJobId(const JobId_t _jobid) :
     25    JobId(_jobid)
    2626{}
    2727
    … …  
    3131JobId_t GlobalJobId::getNextId()
    3232{
    33   setId(getId()+1);
     33  setId(getId()+(JobId_t)1);
    3434  return getId();
    3535}

  • src/Fragmentation/Automation/GlobalJobId.hpp

    r6f2bc7 rd1dbfc  
    1919{
    2020public:
    21   GlobalJobId();
     21  explicit GlobalJobId(const JobId_t _jobid);
    2222  ~GlobalJobId();
    2323

  • src/Fragmentation/Automation/Makefile.am

    r6f2bc7 rd1dbfc  
    4747  Controller/FragmentController.cpp \
    4848  Controller/Commands/CheckResultsOperation.cpp \
     49  Controller/Commands/GetNextJobIdOperation.cpp \
    4950  Controller/Commands/Operation.cpp \
    5051  Controller/Commands/ReceiveJobsOperation.cpp \
    … …  
    6061  Controller/FragmentController.hpp \
    6162  Controller/Commands/CheckResultsOperation.hpp \
     63  Controller/Commands/GetNextJobIdOperation.hpp \
    6264  Controller/Commands/Operation.hpp \
    6365  Controller/Commands/ReceiveJobsOperation.hpp \

  • src/Fragmentation/Automation/controller.cpp

    r6f2bc7 rd1dbfc  
    3939#include "Controller/FragmentController.hpp"
    4040#include "Controller/Commands/CheckResultsOperation.hpp"
     41#include "Controller/Commands/GetNextJobIdOperation.hpp"
    4142#include "Controller/Commands/ReceiveJobsOperation.hpp"
    4243#include "Controller/Commands/SendResultsOperation.hpp"
    4344#include "Controller/Commands/ShutdownOperation.hpp"
    44 #include "GlobalJobId.hpp"
    4545#include "Jobs/MPQCCommandJob.hpp"
    4646#include "Jobs/SystemCommandJob.hpp"
    … …  
    5656};
    5757
    58 // TODO: replace this instance by a IdPool owned by controller.
    59 GlobalJobId globalId;
    60 
    61 void createjobs(std::vector<FragmentJob::ptr> &jobs)
    62 {
    63   FragmentJob::ptr testJob( new SystemCommandJob( std::string("cat"), std::string("Nothing"), globalId.getNextId()) );
    64   FragmentJob::ptr othertestJob( new SystemCommandJob( std::string("cat"), std::string("Nothing"), globalId.getNextId()) );
     58/** Requests an available id from server
     59 *
     60 * @param controller FragmentController with CommandRegistry
     61 * @param host address of server
     62 * @param service port/service of server
     63 */
     64void requestid(
     65    FragmentController &controller,
     66    const std::string &host,
     67    const std::string &service)
     68{
     69  GetNextJobIdOperation *getnextid = static_cast<GetNextJobIdOperation *>(
     70      controller.Commands.getByName("getnextjobid"));
     71  (*getnextid)(host,service);
     72}
     73
     74/** Returns another available id from a finished GetNextJobIdOperation.
     75 *
     76 * @param controller FragmentController with CommandRegistry
     77 * @return next available id
     78 */
     79JobId_t getavailableid(FragmentController &controller)
     80{
     81  GetNextJobIdOperation *getnextid = static_cast<GetNextJobIdOperation *>(
     82      controller.Commands.getByName("getnextjobid"));
     83  const JobId_t nextid = getnextid->getNextId();
     84  LOG(1, "INFO: Next available id is " << nextid << ".");
     85  return nextid;
     86}
     87
     88void createjobs(std::vector<FragmentJob::ptr> &jobs, const JobId_t nextid)
     89{
     90  FragmentJob::ptr testJob( new SystemCommandJob( std::string("cat"), std::string("Nothing"), nextid) );
    6591  jobs.push_back(testJob);
    66   jobs.push_back(othertestJob);
    67 }
    68 
    69 void parsejob(std::vector<FragmentJob::ptr> &jobs, const std::string &filename)
     92  LOG(1, "INFO: Added one empty SystemCommandJob's.");
     93}
     94
     95/** Creates a MPQCCommandJob with argument \a filename.
     96 *
     97 * @param jobs created job is added to this vector
     98 * @param filename filename being argument to job
     99 * @param nextid id for this job
     100 */
     101void parsejob(
     102    std::vector<FragmentJob::ptr> &jobs,
     103    const std::string &filename,
     104    const JobId_t nextid)
    70105{
    71106  std::ifstream file;
    … …  
    74109  std::string output((std::istreambuf_iterator<char>(file)),
    75110      std::istreambuf_iterator<char>());
    76   FragmentJob::ptr testJob( new MPQCCommandJob(output, globalId.getNextId()) );
     111  FragmentJob::ptr testJob( new MPQCCommandJob(output, nextid) );
    77112  jobs.push_back(testJob);
    78113  file.close();
    … …  
    80115}
    81116
    82 void addjobs(FragmentController &controller, std::vector<FragmentJob::ptr> &jobs)
     117/** Adds a vector of jobs to the send operation.
     118 *
     119 * @param controller FragmentController with CommandRegistry
     120 * @param jobs jobs to add
     121 */
     122void addjobs(
     123    FragmentController &controller,
     124    std::vector<FragmentJob::ptr> &jobs)
    83125{
    84126  ReceiveJobsOperation *recjobs = static_cast<ReceiveJobsOperation *>(
    … …  
    87129}
    88130
    89 void sendjobs(FragmentController &controller, const std::string &host, const std::string &service)
     131/** Sends contained jobs in operation to server
     132 *
     133 * @param controller FragmentController with CommandRegistry
     134 * @param host address of server
     135 * @param service port/service of server
     136 */
     137void sendjobs(
     138    FragmentController &controller,
     139    const std::string &host,
     140    const std::string &service)
    90141{
    91142  ReceiveJobsOperation *recjobs = static_cast<ReceiveJobsOperation *>(
    … …  
    94145}
    95146
    96 void checkresults(FragmentController &controller, const std::string &host, const std::string &service)
     147/** Obtains scheduled and done jobs from server
     148 *
     149 * @param controller FragmentController with CommandRegistry
     150 * @param host address of server
     151 * @param service port/service of server
     152 */
     153void checkresults(
     154    FragmentController &controller,
     155    const std::string &host,
     156    const std::string &service)
    97157{
    98158  CheckResultsOperation *checkres = static_cast<CheckResultsOperation *>(
    … …  
    101161}
    102162
     163/** Prints scheduled and done jobs.
     164 *
     165 * @param controller FragmentController with CommandRegistry
     166 */
    103167void printdonejobs(FragmentController &controller)
    104168{
    … …  
    110174}
    111175
    112 void receiveresults(FragmentController &controller, const std::string &host, const std::string &service)
     176/** Obtains results from done jobs from server.
     177 *
     178 * @param controller FragmentController with CommandRegistry
     179 * @param host address of server
     180 * @param service port/service of server
     181 */
     182void receiveresults(
     183    FragmentController &controller,
     184    const std::string &host,
     185    const std::string &service)
    113186{
    114187  SendResultsOperation *sendres = static_cast<SendResultsOperation *>(
    … …  
    117190}
    118191
     192/** Print received results.
     193 *
     194 * @param controller FragmentController with CommandRegistry
     195 */
    119196void printreceivedresults(FragmentController &controller)
    120197{
    … …  
    127204}
    128205
    129 void shutdown(FragmentController &controller, const std::string &host, const std::string &service)
     206/** Sends shutdown signal to server
     207 *
     208 * @param controller FragmentController with CommandRegistry
     209 * @param host address of server
     210 * @param service port/service of server
     211 */
     212void shutdown(
     213    FragmentController &controller,
     214    const std::string &host,
     215    const std::string &service)
    130216{
    131217  ShutdownOperation *shutdown = static_cast<ShutdownOperation *>(
    … …  
    183269    FragmentController controller(io_service);
    184270
     271    // Initial phase: information gathering from server
     272
    185273    switch(getCommandIndex(CommandsMap, argv[3])) {
    186274      case AddJobsIndex:
    187275      {
     276        if (argc < 5) {
     277          ELOG(1, "Please add a filename for the MPQCCommandJob.");
     278        } else {
     279          // get an id for every filename
     280          for (int argcount = 4; argcount < argc; ++argcount) {
     281            requestid(controller, argv[1], argv[2]);
     282          }
     283        }
     284        break;
     285      }
     286      case CreateJobsIndex:
     287      {
     288        requestid(controller, argv[1], argv[2]);
     289        break;
     290      }
     291      case CheckResultsIndex:
     292        break;
     293      case ReceiveResultsIndex:
     294        break;
     295      case ShutdownIndex:
     296        break;
     297      case UnknownCommandIndex:
     298      default:
     299        ELOG(1, "Unrecognized command '"+toString(argv[3])+"'.");
     300        break;
     301    }
     302
     303    {
     304      io_service.reset();
     305      Info info("io_service: Phase One");
     306      io_service.run();
     307    }
     308
     309    // Second phase: Building jobs and sending information to server
     310
     311    switch(getCommandIndex(CommandsMap, argv[3])) {
     312      case AddJobsIndex:
     313      {
    188314        std::vector<FragmentJob::ptr> jobs;
    189         if (argc == 4) {
     315        if (argc < 5) {
    190316          ELOG(1, "Please add a filename for the MPQCCommandJob.");
    191317        } else {
    192318          for (int argcount = 4; argcount < argc; ++argcount) {
    193             LOG(1, "INFO: Parsing job for file " << argv[argcount] << ".");
    194             parsejob(jobs, argv[argcount]);
     319            const JobId_t next_id = getavailableid(controller);
     320            const std::string filename(argv[argcount]);
     321            LOG(1, "INFO: Creating MPQCCommandJob with filename'"
     322                +filename+"', and id "+toString(next_id)+".");
     323            parsejob(jobs, filename, next_id);
    195324          }
    196325          addjobs(controller, jobs);
    … …  
    202331      {
    203332        std::vector<FragmentJob::ptr> jobs;
    204         createjobs(jobs);
     333          const JobId_t next_id = getavailableid(controller);
     334          createjobs(jobs, next_id);
    205335        addjobs(controller, jobs);
    206336        sendjobs(controller, argv[1], argv[2]);
    … …  
    229359
    230360    {
    231       Info info("io_service");
     361      io_service.reset();
     362      Info info("io_service: Phase Two");
    232363      io_service.run();
    233364    }
     365
     366    // Final phase: Print result of command
    234367
    235368    switch(getCommandIndex(CommandsMap, argv[3])) {
Note: See TracChangeset for help on using the changeset viewer.