You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

threading.hpp 24KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485
  1. // threading.hpp
  2. //
  3. // (C) 2006 - 2009 MicroNeil Research Corporation.
  4. //
  5. // This program is part of the MicroNeil Research Open Library Project. For
  6. // more information go to http://www.microneil.com/OpenLibrary/index.html
  7. //
  8. // This program is free software; you can redistribute it and/or modify it
  9. // under the terms of the GNU General Public License as published by the
  10. // Free Software Foundation; either version 2 of the License, or (at your
  11. // option) any later version.
  12. //
  13. // This program is distributed in the hope that it will be useful, but WITHOUT
  14. // ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  15. // FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
  16. // more details.
  17. //
  18. // You should have received a copy of the GNU General Public License along with
  19. // this program; if not, write to the Free Software Foundation, Inc., 59 Temple
  20. // Place, Suite 330, Boston, MA 02111-1307 USA
  21. // The "Threading" module is a basic, cross-platform, multi-threading tool kit.
  22. // The differences between posix compatible systems and win32 based systems are
  23. // abstracted. On win32 systems, native win32 primatives are used to construct.
  24. // efficient, lightweight objects.
  25. // On others we assume we can use pthreads. In either case the objects we have
  26. // here are designed to cover all of the basics efficiently while hiding the
  27. // required under-cover work.
  28. // A lot of this module is coded here in the header with the inline keyword
  29. // because it is likely that the more basic objects can be efficiently compiled
  30. // as inline abstractions to native calls. Really basic systems won't need
  31. // anything beyond what is in this file.
  32. // 20070202.1601 _M Further research has suggested that using a Semaphore in
  33. // WIN32 environments in place of a CRITICAL_SECTION may provide the best
  34. // performance and stability on all platforms. Specifically, SMP platforms may
  35. // race and waste resources with CRITICAL_SECTIONs and in those cases it is
  36. // recommended that the CRITICAL_SECTIONs may be "throttled" using Semaphores
  37. // to limit the number of threads that may contend for a critical section. It
  38. // is also suggested that if the Semaphore has an initialization value of 1
  39. // the CRITICAL_SECTION is redundant. So this code has been modified to do
  40. // precisely that!
  41. //
  42. // This new version also includes a ProductionGateway object that simplifies
  43. // the producer/consumer model. The object keeps track of the number of calls
  44. // to produce() and consume() and ensures that threads will block on consume()
  45. // until a sufficient number of calls to produce() are made. That is, for every
  46. // one call to produce(), a call to consume() will be allowed to proceed. The
  47. // object also allows for the potentially asynchronous nature of these calls.
  48. // 20070530.1751 _M Added top level exception handling in threads along with
  49. // isRunning() and isBad() methods.
  50. // 20060528.1647 _M All of the basics are complete and tested on both WIN32 and
  51. // RHEL4 single and multiple processors.
  52. // Include MNR_threading Once Only =============================================
  53. #ifndef MNR_threading
  54. #define MNR_threading
  55. #include <set>
  56. #include <vector>
  57. #include <string>
  58. #include <queue>
  59. #include "faults.hpp"
  60. using namespace std;
  61. class ThreadManager; // ThreadManager does exist.
  62. extern ThreadManager Threads; // Master thread manager.
  63. ////////////////////////////////////////////////////////////////////////////////
  64. // Thread Status & Type
  65. //
  66. // ThreadState objects are constant static objects defined for each Thread class
  67. // so that the thread can update it's state by changing a pointer. The state
  68. // can then be compared between threads of the same type and can be read-out
  69. // as text for debugging purposes.
  70. class ThreadState { // Thread State Object.
  71. public:
  72. const string Name; // Text name of thread descriptor.
  73. ThreadState(string N) : Name(N) {} // Constructor requires text name.
  74. };
  75. // ThreadType objects are constant static objects defined for each Thread class
  76. // so that classes can be identified by type using a pointer to the constant.
  77. class ThreadType {
  78. public:
  79. const string Name;
  80. ThreadType(string N) : Name(N) {}
  81. };
  82. class Thread; // There is such thing as a Thread.
  83. class ThreadStatusRecord { // Describes a Thread's condition.
  84. private:
  85. Thread* Pointer; // A pointer to the thread.
  86. ThreadType* Type; // A descriptor of it's type.
  87. ThreadState* State; // A descriptor of it's state.
  88. string Name; // Name of the thread if any.
  89. bool isRunning; // True if the thread is running.
  90. bool isBad; // True if the thread is bad.
  91. string Fault; // Bad Thread's Fault if any.
  92. public:
  93. ThreadStatusRecord( // Initialize all items.
  94. Thread* P,
  95. ThreadType& T,
  96. ThreadState& S,
  97. bool R,
  98. bool B,
  99. string F,
  100. string N
  101. ) :
  102. Pointer(P),
  103. Type(&T),
  104. State(&S),
  105. Name(N),
  106. isRunning(R),
  107. isBad(B),
  108. Fault(F)
  109. {}
  110. ThreadStatusRecord& operator=(const ThreadStatusRecord& Right) { // Minimal Assignment Operator
  111. Pointer = Right.Pointer;
  112. Type = Right.Type;
  113. State = Right.State;
  114. isRunning = Right.isRunning;
  115. isBad = Right.isBad;
  116. Fault = Right.Fault;
  117. Name = Right.Name;
  118. return *this;
  119. }
  120. bool operator<(const ThreadStatusRecord& Right) { // Minimal Comparison Operator.
  121. return (Pointer < Right.Pointer);
  122. }
  123. // How to get the details of the report.
  124. const Thread* getPointer() { return Pointer; }
  125. const ThreadType& getType() { return *Type; }
  126. const ThreadState& getState() { return *State; }
  127. bool getRunning() { return isRunning; }
  128. bool getBad() { return isBad; }
  129. string getFault() { return Fault; }
  130. string getName() { return Name; }
  131. };
  132. typedef vector<ThreadStatusRecord> ThreadStatusReport; // Status report type.
  133. // End ThreadDescriptor
  134. ////////////////////////////////////////////////////////////////////////////////
  135. ////////////////////////////////////////////////////////////////////////////////
  136. // Win32 / POSIX abstractions
  137. #ifdef WIN32
  138. // When in WIN32 land...
  139. // Remember to compile (on GNU anyway) with -mthreads
  140. #include <windows.h>
  141. #include <process.h>
  142. typedef HANDLE thread_primative; // The WIN32 thread primative abstracts
  143. // HANDLE
  144. typedef HANDLE mutex_primative; // The WIN32 mutex primative abstracts
  145. // a HANDLE to a Semaphore.
  146. inline void threading_yield() { // When we want to yield time in WIN32
  147. SwitchToThread(); // we call SwitchToThread();
  148. }
  149. #else
  150. // When in POSIX land...
  151. // Remember to compile (on GMU anyway) with -pthread
  152. #include <pthread.h>
  153. #include <sched.h>
  154. typedef pthread_t thread_primative; // The POSIX thread primative abstracts
  155. // pthread_t
  156. typedef pthread_mutex_t mutex_primative; // The POSIX mutex primative abstracts
  157. // pthread_mutex_t
  158. inline void threading_yield() { // When we want to yield time in POSIX
  159. sched_yield(); // we call sched_yield();
  160. }
  161. #endif
  162. // End Win32 / POSIX abstractions
  163. ////////////////////////////////////////////////////////////////////////////////
  164. ////////////////////////////////////////////////////////////////////////////////
  165. // The Thread class gets extended to do any specific work. The pure virtual
  166. // function MyTask is overloaded by the derived class to define that work. It
  167. // is expected that the class will be initialized with any parameters that
  168. // will be used by the thread and that the thread will make available any
  169. // results through public interfaces either during and/or after the thread
  170. // has finished running.
  171. class Thread {
  172. private:
  173. ThreadState* MyThreadState; // Track current thread state.
  174. protected:
  175. const ThreadType& MyThreadType; // Identify thread type.
  176. const string MyThreadName; // Name string of this instance.
  177. thread_primative MyThread; // Abstracted thread.
  178. bool RunningFlag; // True when thread is in myTask()
  179. bool BadFlag; // True when myTask() throws!
  180. string BadWhat; // Bad exception what() if any.
  181. void CurrentThreadState(const ThreadState& TS); // Set thread state.
  182. public:
  183. Thread(); // Constructor (just in case)
  184. Thread(const ThreadType& T, string N); // Construct with specific Type/Name
  185. virtual ~Thread(); // Destructor (just in case)
  186. void run(); // Method to launch this thread.
  187. void join(); // Method to Join this thread.
  188. void launchTask(); // Launch and watch myTask().
  189. virtual void myTask() = 0; // The actual task must be overloaded.
  190. thread_primative getMyThread(); // Inspect my thread primative.
  191. bool isRunning(); // Return the Running flag state.
  192. bool isBad(); // Return the Bad flag state.
  193. const string MyFault(); // Return exception Bad fault if any.
  194. const string MyName(); // The thread's name.
  195. const ThreadType& MyType(); // Thread type for this thread.
  196. const ThreadState& MyState(); // Returns the current thread state.
  197. const ThreadState& CurrentThreadState(); // Returns the current thread state.
  198. ThreadStatusRecord StatusReport(); // Return's the thread's status reprt.
  199. // Constants for Thread...
  200. const static ThreadType Type; // The thread's type.
  201. const static ThreadState ThreadInitialized; // Constructed successfully.
  202. const static ThreadState ThreadStarted; // Started.
  203. const static ThreadState ThreadFailed; // Failed by unhandled exception.
  204. const static ThreadState ThreadStopped; // Stopped normally.
  205. const static ThreadState ThreadDestroyed; // Safety value for destructed Threads.
  206. };
  207. // End Thread
  208. ////////////////////////////////////////////////////////////////////////////////
  209. ////////////////////////////////////////////////////////////////////////////////
  210. // The Mutex class abstracts a lightweight, very basic mutex object.
  211. // As with the Thread object, more ellaborate forms can be built up from
  212. // this basic mechanism. An important design constraint for this basic
  213. // mutex object is that it work even if the thread that's running was not
  214. // created with the Thread object... that ensures that it can be used in
  215. // code that is destined to function in other applications.
  216. class Mutex {
  217. private:
  218. mutex_primative MyMutex; // Here is our primative mutex.
  219. volatile bool IAmLocked; // Here is our Lock Count.
  220. public:
  221. Mutex(); // Construct the mutex.
  222. ~Mutex(); // Destroy the mutex.
  223. void lock(); // Lock it.
  224. void unlock(); // Unlock it.
  225. bool tryLock(); // Try to lock it.
  226. bool isLocked(); // Check to see if it's locked.
  227. };
  228. // End of Mutex
  229. ////////////////////////////////////////////////////////////////////////////////
  230. ////////////////////////////////////////////////////////////////////////////////
  231. // ScopeMutex
  232. // A ScopeMutex is a nifty trick for locking a mutex during some segment of
  233. // code. On construction, it locks the Mutex that it is given and keeps it
  234. // locked until it is destroyed. Of course this also means that it will unlock
  235. // the mutex when it goes out of scope - which is precisely the point :-)
  236. //
  237. // The right way to use a ScopeMutex is to create it just before you need to
  238. // have control and then forget about it. From a design perspective, you might
  239. // want to make sure that whatever happens after the ScopeMutex has been
  240. // created is as short as possible and if it is not then you may want to
  241. // use the Mutex directly.
  242. //
  243. // The best place to use a ScopeMutex is where you might leave the controling
  244. // bit of code through a number of logical paths such as a logic tree or even
  245. // due to some exceptions. In this context it saves you having to track down
  246. // all of the possible cases and unlock the mutex in each of them.
  247. class ScopeMutex {
  248. private:
  249. Mutex& MyMutex; // ScopeMutex has an ordinary Mutex to use.
  250. public:
  251. ScopeMutex(Mutex& M); // Constructing a ScopeMutex requires a Mutex
  252. ~ScopeMutex(); // We do have special code for descrution.
  253. };
  254. // End ScopeMutex
  255. ////////////////////////////////////////////////////////////////////////////////
  256. ////////////////////////////////////////////////////////////////////////////////
  257. // ProductionGateway
  258. // A ProductionGateway encapsulates the thread synchronization required for a
  259. // producer / consumer relationship. For each call to the produce() method one
  260. // call to the consume() method can proceed. The object takes into account that
  261. // these methods may be called out of sequence and that, for example, produce()
  262. // might be called several times before any calls to consume.
  263. #ifdef WIN32
  264. // Win32 Implementation ////////////////////////////////////////////////////////
  265. class ProductionGateway {
  266. private:
  267. HANDLE MySemaphore; // WIN32 makes this one easy w/ a 0 semi.
  268. public:
  269. ProductionGateway(); // The constructor and destructor handle
  270. ~ProductionGateway(); // creating and destroying the semi.
  271. void produce(); // Produce "releases" the semi.
  272. void consume(); // Consume "waits" if needed.
  273. };
  274. #else
  275. // POSIX Implementation ////////////////////////////////////////////////////////
  276. class ProductionGateway { // Posix needs a few pieces for this.
  277. private:
  278. mutex_primative MyMutex; // Mutex to protect the data.
  279. pthread_cond_t MyConditionVariable; // A condition variable for signaling.
  280. int Product; // A count of unused calls to produce()
  281. int Waiting; // A count of waiting threads.
  282. int Signaled; // A count of signaled threads.
  283. public:
  284. ProductionGateway(); // The constructor and destructor handle
  285. ~ProductionGateway(); // creating and destroying the semi.
  286. void produce(); // Produce "releases" the semi.
  287. void consume(); // Consume "waits" if needed.
  288. };
  289. #endif
  290. // End ProductionGateway
  291. ////////////////////////////////////////////////////////////////////////////////
  292. ////////////////////////////////////////////////////////////////////////////////
  293. // The ThreadManager class provides a global thread management tool. All Thread
  294. // objects register themselves with the Threads object upon construction and
  295. // remove themselves from the registry upon destruction. The Threads object can
  296. // produce a status report for all of the known threads on the system and can
  297. // temporarily lock the existing thread so that it can be contacted reliably.
  298. // locking and unlocking the ThreadManager is intended only for short messages
  299. // that set flags in the thread or pass some small data packet. The lock only
  300. // prevents the thread from being destroyed before the message can be sent so
  301. // that the thread that owns the threadlock will not make any calls to a dead
  302. // pointer. Most apps should be designed so that the threadlock mechanism is
  303. // not required.
  304. class ThreadManager { // Central manager for threads.
  305. friend class Thread; // Threads are friends.
  306. private:
  307. Mutex MyMutex; // Protect our data with this.
  308. set<Thread*> KnownThreads; // Keep track of all threads.
  309. void rememberThread(Thread* T); // Threads register themselves.
  310. void forgetThread(Thread* T); // Threads remove themselves.
  311. Thread* LockedThread; // Pointer to locked thread if any.
  312. public:
  313. ThreadManager():LockedThread(0){} // Initialize nice and clean.
  314. ThreadStatusReport StatusReport(); // Get a status report.
  315. bool lockExistingThread(Thread* T); // Locks ThreadManager if T exists.
  316. void unlockExistingThread(Thread* T); // Unlocks ThreadManager if T locked.
  317. };
  318. class ScopeThreadLock { // This is like a ScopeMutex for
  319. private: // the ThreadManager.
  320. Thread* MyLockedThread; // It needs to know it's Thread.
  321. public:
  322. ScopeThreadLock(Thread* T); // Locks T in ThreadManager if it can.
  323. ~ScopeThreadLock(); // Unlocks T in ThreadManager if locked.
  324. bool isGood(); // True if T was locked.
  325. bool isBad(); // False if T was not locked.
  326. };
  327. // End Thread Manager
  328. ////////////////////////////////////////////////////////////////////////////////
  329. ////////////////////////////////////////////////////////////////////////////////
  330. // A ProductionQueue is a templated, thread safe mechanism for implementing
  331. // a producer/consumer relationship. The objects in the queue should be simple
  332. // data so that they can be created, destroyed, and copied without trouble. Put
  333. // another way - the objects in the ProductionQueue should be lightweight
  334. // handles for other things. Those things should be created and destroyed
  335. // elsewhere.
  336. template<typename T> // Templatized
  337. class ProductionQueue { // Production Queue Class
  338. private:
  339. Mutex myMutex; // Contains a mutex and
  340. volatile unsigned int LatestSize; // a volatile (blinking light) size
  341. ProductionGateway myGateway; // integrated with a production
  342. queue<T> myQueue; // gateway and a queue.
  343. public:
  344. ProductionQueue() : LatestSize(0) {} // The size always starts at zero.
  345. T take() { // To consume a queued object
  346. myGateway.consume(); // we wait on the production gateway
  347. ScopeMutex OneAtATimePlease(myMutex); // and when we get through we lock
  348. T O = myQueue.front(); // the mutext, take the object on the
  349. myQueue.pop(); // front of the queue, pop it out,
  350. LatestSize = myQueue.size(); // and rest our size (blinking light).
  351. return O; // Then return the object we got.
  352. }
  353. void give(T O) { // To produce a queued object
  354. ScopeMutex OneAtATimePlease(myMutex); // we wait on the mutex. When we
  355. myQueue.push(O); // get through we push our object
  356. LatestSize = myQueue.size(); // into the queue, reset our size
  357. myGateway.produce(); // indicator and tell the gateway.
  358. } // When we're done it can be grabbed.
  359. unsigned int size() { // To check the size we look at
  360. return LatestSize; // the blinking light.
  361. }
  362. };
  363. // End Production Queue
  364. ////////////////////////////////////////////////////////////////////////////////
  365. #endif
  366. // End Of Include MNR_threading Once Only ======================================