source: src/Fragmentation/Automation/Operations/OperationQueue.hpp@ 92b1d3

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
Last change on this file since 92b1d3 was 92b1d3, checked in by Frederik Heber <heber@…>, 12 years ago

OperationQueue now has a max_connections and allows only this number of running operations.

  • OperationQueue::launchNextOp() is called by update() and push_back().
  • new AddressMap takes endpoint addresses and is used to know which operations still have to be executed.
  • Property mode set to 100644
File size: 2.9 KB
Line 
1/*
2 * OperationQueue.hpp
3 *
4 * Created on: Apr 24, 2012
5 * Author: heber
6 */
7
8#ifndef OPERATIONQUEUE_HPP_
9#define OPERATIONQUEUE_HPP_
10
11
12// include config.h
13#ifdef HAVE_CONFIG_H
14#include <config.h>
15#endif
16
17#include <boost/shared_ptr.hpp>
18#include <deque>
19
20#include "CodePatterns/Observer/Observer.hpp"
21
22class AsyncOperation;
23class Observer;
24class OperationQueueTest;
25class WorkerAddress;
26
27/** This class is a container for \ref AsyncOperation's that are kept as shared_ptr
28 * and removed when the operation is done.
29 */
30class OperationQueue : public Observer
31{
32 //!> grant unit test access to private part
33 friend class OperationQueueTest;
34public:
35 /** Default constructor for class OperationQueue.
36 *
37 */
38 OperationQueue() :
39 Observer("OperationQueue"),
40 RunningOps(0)
41 {}
42 /** Default destructor for class OperationQueue.
43 *
44 */
45 virtual ~OperationQueue()
46 {}
47
48 typedef boost::shared_ptr<AsyncOperation> AsyncOp_ptr;
49
50 /** Add an operation to the internal queue and hand over memory responsibility to it,
51 * also the operation is run.
52 *
53 * @param op operation to add, is NULL on return.
54 */
55 void push_back(AsyncOperation *&op, const WorkerAddress &address);
56
57 /** States whether the queue is empty.
58 *
59 * @return true - queue is empty, false - operations are pending
60 */
61 bool empty() const {
62 return queue.empty();
63 }
64
65 void update(Observable *publisher);
66 void recieveNotification(Observable *publisher, Notification_ptr notification);
67 void subjectKilled(Observable *publisher);
68
69private:
70 /** Removes an operation from the queue.
71 *
72 * @param op operation to remove from queue
73 * @param observer observer to sign off from operation, NULL if none to sign off
74 */
75 void remove(AsyncOperation *op, Observer *observer);
76
77 /** Returns the number of currently running operations.
78 *
79 * @return Gives the difference between the entries in the queue and in the AddressMap.
80 */
81 size_t getNumberOfRunningOps() const
82 {
83 return queue.size() - AddressMap.size();
84 }
85
86 /** Helper to launch the next pending operation.
87 *
88 */
89 void LaunchNextOp();
90
91 //!> internal operation to send jobs to workers
92 typedef std::deque<AsyncOp_ptr> OperationQueue_t;
93
94 /** Tiny Helper function to find an operation inside FragmentScheduler::OperationQueue.
95 *
96 * @param op operation to remove from queue
97 * @return iterator to element or to OperationQueue.end()
98 */
99 OperationQueue_t::iterator findOperation(AsyncOperation *op);
100
101 //!> internal number stating how many operations are running
102 size_t RunningOps;
103
104 //!> giving the maximum number of connections
105 static size_t max_connections;
106
107private:
108 //!> internal queue with operations
109 OperationQueue_t queue;
110
111 //!> typedef for the association for each operation to its address to connect to
112 typedef std::map<AsyncOp_ptr, WorkerAddress> AddressMap_t;
113 //!> Association for each operation to its address to connect to
114 AddressMap_t AddressMap;
115};
116
117#endif /* OPERATIONQUEUE_HPP_ */
Note: See TracBrowser for help on using the repository browser.