NCBI C++ ToolKit
prefetch_manager_impl.cpp
Go to the documentation of this file.

Go to the SVN repository for this file.

1 /* $Id: prefetch_manager_impl.cpp 99890 2023-05-18 18:07:50Z vasilche $
2 * ===========================================================================
3 *
4 * PUBLIC DOMAIN NOTICE
5 * National Center for Biotechnology Information
6 *
7 * This software/database is a "United States Government Work" under the
8 * terms of the United States Copyright Act. It was written as part of
9 * the author's official duties as a United States Government employee and
10 * thus cannot be copyrighted. This software/database is freely available
11 * to the public for use. The National Library of Medicine and the U.S.
12 * Government have not placed any restriction on its use or reproduction.
13 *
14 * Although all reasonable efforts have been taken to ensure the accuracy
15 * and reliability of the software and data, the NLM and the U.S.
16 * Government do not and cannot warrant the performance or results that
17 * may be obtained by using this software or data. The NLM and the U.S.
18 * Government disclaim all warranties, express or implied, including
19 * warranties of performance, merchantability or fitness for any particular
20 * purpose.
21 *
22 * Please cite the author in any work or product based on this material.
23 *
24 * ===========================================================================
25 *
26 * Author: Eugene Vasilchenko
27 *
28 * File Description:
29 * Prefetch implementation
30 *
31 */
32 
33 #include <ncbi_pch.hpp>
37 #include <corelib/ncbithr.hpp>
38 #include <corelib/ncbi_system.hpp>
40 
41 
44 
45 
47 
48 
49 BEGIN_SCOPE(prefetch)
50 
51 /////////////////////////////////////////////////////////////////////////////
52 // CCancelRequest::
53 //
54 // Exception used to cancel requests safely, cleaning up
55 // all the resources allocated.
56 //
57 
59 {
60 public:
61  // Create new exception object, initialize counter.
63 
64  // Create a copy of exception object, increase counter.
66 
67  // Destroy the object, decrease counter. If the counter is
68  // zero outside of CThread::Wrapper(), rethrow exception.
70 
71  // Inform the object it has reached normal catch.
72  void SetFinished(void)
73  {
74  m_Data->m_Finished = true;
75  }
76 private:
77  struct SData {
78  SData(void)
79  : m_RefCounter(1),
80  m_Finished(false)
81  {
82  }
84  bool m_Finished;
85  };
87 };
88 
89 
91  : m_Data(new SData())
92 {
93 }
94 
95 
97  : m_Data(prev.m_Data)
98 {
100 }
101 
102 
104 {
105  if ( --m_Data->m_RefCounter > 0 ) {
106  // Not the last object - continue to handle exceptions
107  return;
108  }
109 
110  bool finished = m_Data->m_Finished; // save the flag
111  delete m_Data;
112 
113  if ( !finished ) {
114  ERR_POST(Critical<<"CancelRequest() failed due to catch(...) in "<<
115  CStackTrace());
116  }
117 }
118 
119 END_SCOPE(prefetch)
120 
122  IPrefetchAction* action,
123  IPrefetchListener* listener,
124  unsigned int priority)
125  : CThreadPool_Task(priority),
126  m_StateMutex(state_mutex),
127  m_Action(action),
128  m_Listener(listener),
129  m_Progress(0)
130 {
131 }
132 
133 
135 {
136 }
137 
138 
140 {
141  switch (GetStatus()) {
143  return eInvalid;
147  return eStarted;
154  }
155 
156  return eInvalid;
157 }
158 
159 
161 {
162  CMutexGuard guard(m_StateMutex->GetData());
163  if ( m_Listener ) {
164  NCBI_THROW(CObjMgrException, eOtherError,
165  "CPrefetchToken::SetListener: listener already set");
166  }
167  m_Listener = listener;
168 }
169 
170 
172 {
173  CMutexGuard guard(m_StateMutex->GetData());
174  if (m_Listener) {
175  m_Listener->PrefetchNotify(Ref(this), GetState());
176  }
177 }
178 
181 {
182  CMutexGuard guard(m_StateMutex->GetData());
183  if ( GetStatus() != eExecuting ) {
184  NCBI_THROW(CObjMgrException, eOtherError,
185  "CPrefetchToken::SetProgress: not processing");
186  }
187  TProgress old_progress = m_Progress;
188  if ( progress != old_progress ) {
189  m_Progress = progress;
190  if ( m_Listener ) {
191  m_Listener->PrefetchNotify(Ref(this), eAdvanced);
192  }
193  }
194  return old_progress;
195 }
196 
197 
199 {
200  try {
202  if (m_Action.NotNull()) {
203  if (! GetAction()->Execute(Ref(this))) {
204  if ( IsCancelRequested() )
206  else
208  }
209  }
210  return result;
211  }
212  catch ( CPrefetchCanceled& /* ignored */ ) {
214  }
215  catch ( prefetch::CCancelRequestException& exc ) {
216  exc.SetFinished();
218  }
219 }
220 
221 
223  CThread::TRunMode threads_mode)
224  : m_StateMutex(new CObjectFor<CMutex>()),
225  m_ThreadPool(kMax_Int, max_threads, 2, threads_mode)
226 {
227 }
228 
229 
231 {
232 }
233 
234 
236  IPrefetchAction* action,
237  IPrefetchListener* listener)
238 {
239  //CMutexGuard guard0(m_ThreadPool.GetMainPoolMutex());
240  if ( action && m_ThreadPool.IsAborted() ) {
241  throw prefetch::CCancelRequestException();
242  }
243  CMutexGuard guard(m_StateMutex->GetData());
245  action,
246  listener,
247  priority));
248  m_ThreadPool.AddTask(req);
249  return req;
250 }
251 
252 
254 {
255  CThreadPool_Thread* thread = dynamic_cast<CThreadPool_Thread*>(
257  if ( !thread ) {
258  return false;
259  }
260 
261  CRef<CThreadPool_Task> req = thread->GetCurrentTask();
262  if ( !req ) {
263  return false;
264  }
265 
266  if ( req->IsCancelRequested() && dynamic_cast<CPrefetchRequest*>(&*req) ) {
267  throw prefetch::CCancelRequestException();
268  }
269 
270  return true;
271 }
272 
273 
CMutex –.
Definition: ncbimtx.hpp:749
Base class for all object manager exceptions.
CObjectFor –.
Definition: ncbiobj.hpp:2335
This exception is used to interrupt actions canceled by user.
CRef –.
Definition: ncbiobj.hpp:618
Abstract class for representing single task executing in pool of threads To use this class in applica...
Definition: thread_pool.hpp:76
Base class for a thread running inside CThreadPool and executing tasks.
#define false
Definition: bool.h:36
static DLIST_TYPE *DLIST_NAME() prev(DLIST_LIST_TYPE *list, DLIST_TYPE *item)
Definition: dlist.tmpl.h:61
#define ERR_POST(message)
Error posting with file, line number information but without error codes.
Definition: ncbidiag.hpp:186
void Critical(CExceptionArgs_Base &args)
Definition: ncbiexpt.hpp:1203
#define NCBI_THROW(exception_class, err_code, message)
Generic macro to throw an exception, given the exception class, error code and message string.
Definition: ncbiexpt.hpp:704
static bool IsActive(void)
CIRef< IPrefetchAction > m_Action
CRef< CObjectFor< CMutex > > m_StateMutex
CRef< CObjectFor< CMutex > > m_StateMutex
virtual void OnStatusChange(EStatus)
Callback to notify on changes in the task status.
virtual EStatus Execute(void)
Do the actual job.
CIRef< IPrefetchListener > m_Listener
CPrefetchManager_Impl(unsigned max_threads, CThread::TRunMode threads_mode)
TProgress SetProgress(TProgress progress)
void SetListener(IPrefetchListener *listener)
IPrefetchAction * GetAction(void) const
CRef< CPrefetchRequest > AddAction(TPriority priority, IPrefetchAction *action, IPrefetchListener *listener)
EState GetState(void) const
bool NotNull(void) const THROWS_NONE
Check if pointer is not null – same effect as NotEmpty().
Definition: ncbiobj.hpp:744
CRef< C > Ref(C *object)
Helper functions to get CRef<> and CConstRef<> objects.
Definition: ncbiobj.hpp:2015
T & GetData(void)
Get data as a reference.
Definition: ncbiobj.hpp:2346
#define kMax_Int
Definition: ncbi_limits.h:184
#define END_NCBI_SCOPE
End previously defined NCBI scope.
Definition: ncbistl.hpp:103
#define END_SCOPE(ns)
End the previously defined scope.
Definition: ncbistl.hpp:75
#define BEGIN_NCBI_SCOPE
Define ncbi namespace.
Definition: ncbistl.hpp:100
#define BEGIN_SCOPE(ns)
Define a new scope.
Definition: ncbistl.hpp:72
EStatus GetStatus(void) const
Get status of the task.
CRef< CThreadPool_Task > GetCurrentTask(void) const
Get the task currently executing in the thread.
EStatus
Status of the task.
Definition: thread_pool.hpp:79
bool IsAborted(void) const
Does method Abort() was already called for this ThreadPool.
bool IsCancelRequested(void) const
Check if cancellation of the task was requested.
void AddTask(CThreadPool_Task *task, const CTimeSpan *timeout=NULL)
Add task to the pool for execution.
@ eIdle
has not been placed in queue yet
Definition: thread_pool.hpp:80
@ eQueued
in the queue, awaiting execution
Definition: thread_pool.hpp:81
@ eExecuting
being executed
Definition: thread_pool.hpp:82
@ eFailed
failure during execution
Definition: thread_pool.hpp:84
@ eCompleted
executed successfully
Definition: thread_pool.hpp:83
@ eCanceled
canceled - possible only if canceled before processing was started or if method Execute() returns res...
Definition: thread_pool.hpp:85
int TRunMode
Bitwise OR'd flags for thread creation passed to Run().
Definition: ncbithr.hpp:558
static CThread * GetCurrentThread(void)
Get current CThread object (or NULL, if main thread)
Definition: ncbithr.cpp:508
Static variables safety - create on demand, destroy on application termination.
Multi-threading – classes, functions, and features.
else result
Definition: token2.c:20
Modified on Wed Apr 17 13:09:40 2024 by modify_doxy.py rev. 669887