NCBI C++ ToolKit
app_job_dispatcher.hpp
Go to the documentation of this file.

Go to the SVN repository for this file.

1 #ifndef GUI_OBJUTILS___APP_JOB_DISPATCHER__HPP
2 #define GUI_OBJUTILS___APP_JOB_DISPATCHER__HPP
3 
4 /* $Id: app_job_dispatcher.hpp 34478 2016-01-13 16:19:10Z katargir $
5  * ===========================================================================
6  *
7  * PUBLIC DOMAIN NOTICE
8  * National Center for Biotechnology Information
9  *
10  * This software/database is a "United States Government Work" under the
11  * terms of the United States Copyright Act. It was written as part of
12  * the author's official duties as a United States Government employee and
13  * thus cannot be copyrighted. This software/database is freely available
14  * to the public for use. The National Library of Medicine and the U.S.
15  * Government have not placed any restriction on its use or reproduction.
16  *
17  * Although all reasonable efforts have been taken to ensure the accuracy
18  * and reliability of the software and data, the NLM and the U.S.
19  * Government do not and cannot warrant the performance or results that
20  * may be obtained by using this software or data. The NLM and the U.S.
21  * Government disclaim all warranties, express or implied, including
22  * warranties of performance, merchantability or fitness for any particular
23  * purpose.
24  *
25  * Please cite the author in any work or product based on this material.
26  *
27  * ===========================================================================
28  *
29  * Authors: Andrey Yazhuk
30  *
31  * File Description:
32  *
33  */
34 
35 /** @addtogroup GUI_UTILS
36 *
37 * @{
38 */
39 
41 #include <gui/utils/app_job.hpp>
42 
44 
45 
47 
48 class IAppJobListener;
49 class IAppJobEngine;
50 class IEngineParams;
51 
52 ///////////////////////////////////////////////////////////////////////////////
53 /// IAppJobEngineListener - receives notifications about changes in job state.
54 /// Job Engines use this interface to notify external objects.
56 {
57 public:
59 
61 
62  virtual void OnEngineJobStateChanged(IAppJob& job, TJobState new_state) = 0;
63 };
64 
65 
66 ///////////////////////////////////////////////////////////////////////////////
67 /// IAppJobListener
68 /// Interface for components that need to be notified about changes in Jobs.
69 /*
70 class NCBI_GUIOBJUTILS_EXPORT IAppJobListener
71 {
72 public:
73  typedef CAppJobDispatcher::TJobID TJobID;
74  typedef IAppJob::EJobState TJobState;
75 
76  virtual ~IAppJobListener() {};
77 
78  virtual void SetDispatcher(CAppJobDispatcher& disp) = 0;
79 
80  virtual void OnJobStateChanged(TJobID job_id, TJobState new_state) = 0;
81  virtual void OnJobProgress(TJobID job_id) = 0;
82 };
83 */
84 
85 ///////////////////////////////////////////////////////////////////////////////
86 /// CAppJobException - Exception thrown by Job Dispatcher and Job Engines.
89 {
90 public:
91  enum EErrCode {
92  eInvalidOperation, ///< Dispatcher - operation is invalid
93  eUnknownJob, ///< Job record lost
94  eEngine_IncompatibleJob, ///< the Job is incompatible with the Engine
95  eEngine_InvalidParams, ///< Invalid parameters provided to the Engine
96  eEngine_InvalidOperation,///< Engine - operation is invalid
97  eEngine_UnknownJob, ///< the job is not registered in the Engine
98  eEngineFailed, ///< Engine failed to perforn an operation
99  eEngineBusy, ///< Engine is busy, caller needs to re-try the operation
100  eFatalError ///< unknown tragic error
101  };
103 };
104 
106 
107 ///////////////////////////////////////////////////////////////////////////////
108 /// CAppJobDispatcher
110  public CObject,
111  public IAppJobEngineListener
112 {
113 public:
115  enum EConsts {
116  eInvalidJobID = -1
117  };
118 
119  typedef int TJobID;
121 
122 public:
123 
124  static CAppJobDispatcher& GetInstance(); /// get the Singleton Dispatcher
125  static void ReleaseInstance(); /// Release the singleton
126 
128  virtual ~CAppJobDispatcher();
129 
130  /// Terminates all jobs and releases Engines
131  void ShutDown();
132 
133  /// this function shall be called in the the application idle function.
134  /// Calling this function gives Dispatcher a change to effectively poll
135  /// passive Engines.
136  bool IdleCallback();
137 
138  /// Registers a new Engine, returns true if successful. Dispatcher does not
139  /// assume ownership of the engine.
140  bool RegisterEngine(const string& name, IAppJobEngine& engine);
141 
142  /// Starts a Job on the specified engine in "passive mode" - no notifications
143  /// or progress reports will be sent.
144  TJobID StartJob(IAppJob& job, const string& engine_name,
145  IEngineParams* params = NULL);
146 
147  /// Starts a Job on the specified engine; the provided listener will
148  /// receive notification events when the Job's state changes.
149  TJobID StartJob(IAppJob& job, const string& engine_name,
150  CEventHandler& listener, int report_period = -1,
151  bool auto_delete = false,
152  IEngineParams* params = NULL);
153 
154  /// Starts a Job on the specified engine, Dispatcher will call methods of
155  /// the provide listener when the Job's state changes.
156  TJobID StartJob(IAppJob& job, const string& engine_name,
157  CAppJobEventTranslator& listener, int report_period = -1,
158  bool auto_delete = false,
159  IEngineParams* params = NULL);
160 
161  /// Runs jon synchronously sending job notifications synchronously
162  /// Returns when job is finished
163  void RunSync(IAppJob& job, TJobID& jobId, CEventHandler& listener);
164 
165  /// Mute all notifications
166  void Mute(bool bMute=true);
167 
168  void CancelJob(TJobID job_id);
169  void SuspendJob(TJobID job_id);
170  void ResumeJob(TJobID job_id);
171  /// Request to cancel all jobs (func returns without waiting)
172  void CancelAllJobs();
173 
174  /// when a Job is deleted the listener is not notified
175  ///
176  /// @return true - if job was deleted, false if job not found
177  bool DeleteJob(TJobID job_id);
178 
179  TJobState GetJobState(TJobID job_id);
180 
181  CConstIRef<IAppJobProgress> GetJobProgress(TJobID job_id);
182  CRef<CObject> GetJobResult(TJobID job_id);
183  CConstIRef<IAppJobError> GetJobError(TJobID job_id);
184 
185  /// @name IAppJobEngineListener
186  /// @{
187  virtual void OnEngineJobStateChanged(IAppJob& job, TJobState new_state);
188  /// @}
189 
190  static string StateToStr(TJobState state);
191  static bool IsTerminal(TJobState state);
192 
193  // CGuard related
194  void Lock() const;
195  void Unlock() const;
196 
197  /// Debugging method for status strings
198  static
199  string GetStatusString(TJobState job_state);
200 
201 protected:
202  /// SJobRecord is a Job Descriptor
203  struct SJobRecord;
204  struct SQueueItem;
205 
206  friend struct SJobRecord;
207  friend struct SQueueItem;
208 
209  /// SJobRecord describes a Job registered in Dispatcher
210  struct SJobRecord
211  {
216  CIRef<CAppJobEventTranslator> m_Listener; ///< if not null - "active" mode
217  int m_ReportPeriod; ///< if > 0, active progress reporting is required
218 
220  bool m_AutoDelete; ///< delete the record when job finishes
221 
223  IAppJobEngine& engine, CAppJobEventTranslator* listener,
224  int report_period, bool auto_delete);
225 
226  inline bool ActiveProgress() const { return m_ReportPeriod > 0; }
227  };
228 
229  /// SQueueItem - element of the Polling Queue
230  struct SQueueItem
231  {
232  int m_JobId;
234 
235  SQueueItem() : m_JobId(0), m_Progress(false) {}
236  SQueueItem(int job_id, bool progress)
237  : m_JobId(job_id), m_Progress(progress) {}
238  };
239 
241  {
244 
245  SJobStateEvent() : m_NewState(IAppJob::eInvalid) {}
247  : m_Job(&job), m_NewState(new_state) {}
248  };
249 
250 protected:
251  void x_AddJobRecord(SJobRecord& rec);
252  void x_RemoveJobRecord(SJobRecord& rec);
253 
254  TJobID x_StartJob(IAppJob& job, const string& engine_name,
255  CAppJobEventTranslator* listener, int report_period,
256  bool auto_delete, IEngineParams* params);
257  void x_OnJobStarted(IAppJob& job, IAppJobEngine& /*engine*/,
258  CAppJobEventTranslator* listener, int report_period,
259  bool /*auto_delete*/);
260 
261  IAppJobEngine* x_GetRegisteredEngine(const string& engine_name);
262  SJobRecord* x_GetJobRecord(TJobID job_id);
263  SJobRecord* x_GetJobRecord(IAppJob& job);
264 
265  /// Update job record, throws an exception if new state change is incorrect
266  ///
267  static
268  void x_OnJobStateChanged(SJobRecord& rec, TJobState new_state);
269 
270  void x_OnJobStateChangedNotify(SJobRecord& rec);
271  void x_OnJobProgressNotify(SJobRecord& rec);
272 
273  bool x_PollEngines();
274 
275  //void x_OnJobStateChangedEvent(IAppJob& job, TJobState new_state);
276 
277  void x_FlushStateEventQueue();
278 
279  void x_VerifyProgressNotNull(CAppJobDispatcher::SJobRecord& rec);
280 
281  bool x_IsCanceled(int job_id) const;
282 
283 protected:
288  typedef list<SJobStateEvent> TStateEventQueue;
289 
291 protected:
292  DECLARE_CLASS_STATIC_MUTEX(sm_Mutex); // guards sm_Dispatcher
293 
294  /// global dispatcher, this instance is used by default in most cases,
295  /// however it is possible to create another instance if needed
297 
298  /// guards this instance of the Dispatcher
300 
301 
302  TNameToEngine m_NameToEngine; ///< Engines Registry
303  CFastMutex m_EngineMutex; ///< Engines registry mutex
304 
306  TIDToRec m_JobRecs; /// Job Registry (index by JobID)
307  TPtrToRec m_PtrToRec; /// Job Index (by pointer)
308 
309  TTimeToItem m_PollQueue; /// priority queue for Dispatcher to poll on
310 
311  /// a Queue that holds state change events posted by Engines
314 
315  bool m_Mute;
316  bm::bvector<> m_CancelVect; ///< Canceled jobs vector
317  bool m_ShutDownInProgress; ///< Shutdown flag
318 };
319 
320 ///////////////////////////////////////////////////////////////////////////////
321 /// CAppJobNotification
322 /// Notification send by CAppJobEventTranslator
324  public CEvent
325 {
326 public:
329 
330  enum EType {
332  eProgress
333  };
334 
335  CAppJobNotification(TJobID job_id, TJobState state);
336  CAppJobNotification(TJobID job_id, CObject* result = NULL);
337  CAppJobNotification(TJobID job_id, const IAppJobError& error);
338  CAppJobNotification(TJobID job_id, const IAppJobProgress& progress);
339 
340  TJobID GetJobID() const { return m_JobID; }
341  TJobState GetState() const { return m_State; }
342 
343  /// returns non-null pointer only if Completed or Running
344  /// and has temporary results available
345  CRef<CObject> GetResult() const { return m_Result; }
346 
347  /// returns non-null pointer only if job Failed
348  CConstIRef<IAppJobError> GetError() const { return m_Error; }
349 
350  /// returns non-null pointer only if notification type is eProgress
351  CConstIRef<IAppJobProgress> GetProgress() const { return m_Progress; }
352 
353 protected:
359 };
360 
361 
362 ///////////////////////////////////////////////////////////////////////////////
363 /// CAppJobEventTranslator
364 /// Standard Listener that generates notification events
366  public CObject//,
367  //public IAppJobListener
368 {
369 public:
372 
373 public:
375 
376  /// @name IAppJobListener implementation
377  /// @{
378  void SetDispatcher(CAppJobDispatcher& disp);
379 
380  //void OnJobStateChanged(TJobID job_id, TJobState new_state);
381  void OnJobStateChanged(CAppJobDispatcher::SJobRecord* job_rec, TJobState new_state);
382  //void OnJobProgress(TJobID job_id);
383  void OnJobProgress(CAppJobDispatcher::SJobRecord* job_rec);
384  /// @}
385 
386 protected:
387  virtual void x_NotifyObservers(CRef<CEvent> & evt);
388 private:
393 };
394 
395 
396 
397 
399 
400 /* @} */
401 
402 #endif // GUI_OBJUTILS___APP_JOB_DISPATCHER__HPP
CAppJobDispatcher.
CAppJobEventTranslator Standard Listener that generates notification events.
IAppJobListener Interface for components that need to be notified about changes in Jobs.
CAppJobNotification Notification send by CAppJobEventTranslator.
CEventHandler.
CEvent - generic event implementation TODO TODO - Attachments.
Definition: event.hpp:86
CFastMutex –.
Definition: ncbimtx.hpp:667
CMutex –.
Definition: ncbimtx.hpp:749
CObject –.
Definition: ncbiobj.hpp:180
IAppJobEngineListener - receives notifications about changes in job state.
IAppJobEngine.
IAppJobError.
Definition: app_job.hpp:65
IAppJobProgress.
Definition: app_job.hpp:50
IAppJob.
Definition: app_job.hpp:82
Definition: map.hpp:338
#define false
Definition: bool.h:36
#define NULL
Definition: ncbistd.hpp:225
#define EXCEPTION_VIRTUAL_BASE
Do not use virtual base classes in exception declaration at all, because in this case derived class s...
Definition: ncbiexpt.hpp:1388
NCBI_EXCEPTION_DEFAULT(CAppJobException, CException)
CConstIRef< IAppJobError > GetError() const
returns non-null pointer only if job Failed
CConstIRef< IAppJobProgress > m_Progress
IAppJob::EJobState TJobState
TTimeToItem m_PollQueue
Job Index (by pointer)
SQueueItem(int job_id, bool progress)
multimap< time_t, SQueueItem > TTimeToItem
static CRef< CAppJobDispatcher > sm_Dispatcher
global dispatcher, this instance is used by default in most cases, however it is possible to create a...
CRef< CObject > GetResult() const
returns non-null pointer only if Completed or Running and has temporary results available
IAppJob::EJobState TJobState
CGuard< CAppJobDispatcher > TDispatcherGuard
CEventHandler & m_TargetEventHandler
map< string, CIRef< IAppJobEngine > > TNameToEngine
map< IAppJob *, SJobRecord * > TPtrToRec
CConstIRef< IAppJobError > m_Error
TStateEventQueue m_StateEventQueue
priority queue for Dispatcher to poll on
SJobStateEvent(IAppJob &job, TJobState new_state)
bool m_AutoDelete
delete the record when job finishes
int m_ReportPeriod
if > 0, active progress reporting is required
DECLARE_CLASS_STATIC_MUTEX(sm_Mutex)
CWeakIRef< CEventHandler > m_TargetWeakPtr
CFastMutex m_EngineMutex
Engines registry mutex.
CIRef< CAppJobEventTranslator > m_Listener
if not null - "active" mode
EJobState
Job states (describe FSM)
Definition: app_job.hpp:86
bool m_ShutDownInProgress
Shutdown flag.
TPtrToRec m_PtrToRec
Job Registry (index by JobID)
IAppJob::EJobState TJobState
CConstIRef< IAppJobProgress > GetProgress() const
returns non-null pointer only if notification type is eProgress
CAppJobDispatcher::TJobID TJobID
IAppJob::EJobState TJobState
list< SJobStateEvent > TStateEventQueue
TJobState GetState() const
TNameToEngine m_NameToEngine
Engines Registry.
map< TJobID, SJobRecord * > TIDToRec
CConstIRef< IAppJobProgress > m_Progress
CAppJobDispatcher::TJobID TJobID
CMutex m_MainMutex
guards this instance of the Dispatcher
virtual void OnEngineJobStateChanged(IAppJob &job, TJobState new_state)=0
CAppJobDispatcher * m_Dispatcher
bm::bvector m_CancelVect
Canceled jobs vector.
@ eEngine_UnknownJob
the job is not registered in the Engine
@ eUnknownJob
Job record lost.
@ eEngine_InvalidParams
Invalid parameters provided to the Engine.
@ eEngineFailed
Engine failed to perforn an operation.
@ eEngine_IncompatibleJob
the Job is incompatible with the Engine
@ eEngineBusy
Engine is busy, caller needs to re-try the operation.
@ eInvalidOperation
Dispatcher - operation is invalid.
@ eEngine_InvalidOperation
Engine - operation is invalid.
#define END_NCBI_SCOPE
End previously defined NCBI scope.
Definition: ncbistl.hpp:103
#define BEGIN_NCBI_SCOPE
Define ncbi namespace.
Definition: ncbistl.hpp:100
#define NCBI_GUIUTILS_EXPORT
Definition: gui_export.h:518
#define NCBI_GUIOBJUTILS_EXPORT
Definition: gui_export.h:512
Compressed bitset (entry point to bm.h)
SJobRecord describes a Job registered in Dispatcher.
SQueueItem - element of the Polling Queue.
else result
Definition: token2.c:20
Modified on Wed Jun 19 17:01:01 2024 by modify_doxy.py rev. 669887