jthread.hpp 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. ############################################################################## */
  13. #ifndef __JTHREAD__
  14. #define __JTHREAD__
  15. #include "jiface.hpp"
  16. #include "jmutex.hpp"
  17. #include "jexcept.hpp"
  18. #include "jhash.hpp"
  19. #include <functional>
  20. #ifdef _WIN32
  21. #define DEFAULT_THREAD_PRIORITY THREAD_PRIORITY_NORMAL
  22. #else
  23. // no thread priority handling?
  24. #endif
  25. interface jlib_decl IThread : public IInterface
  26. {
  27. virtual void start() = 0;
  28. virtual int run() = 0;
  29. };
  30. interface jlib_decl IThreadName
  31. {
  32. virtual const char *get()=0;
  33. };
  34. extern jlib_decl void addThreadExceptionHandler(IExceptionHandler *handler);
  35. extern jlib_decl void removeThreadExceptionHandler(IExceptionHandler *handler);
  36. extern jlib_decl void enableThreadSEH();
  37. extern jlib_decl void disableThreadSEH();
  38. extern jlib_decl unsigned threadLogID(); // for use in logging
  39. // A function registered via addThreadTermFunc will be called when the thread that registered that function
  40. // terminates. Such a function should call on to the previously registered function (if any) - generally you
  41. // would expect to store that value in thread-local storage.
  42. // This can be used to ensure that thread-specific objects can be properly destructed.
  43. // Note that threadpools also call the thread termination hook when each thread's threadmain function terminates,
  44. // so the hook function should clear any variables if necessary rather than assuming that they will be cleared
  45. // at thread startup time.
  46. typedef bool (*ThreadTermFunc)(bool isPooled);
  47. extern jlib_decl void addThreadTermFunc(ThreadTermFunc onTerm);
  48. extern jlib_decl void callThreadTerminationHooks(bool isPooled);
  49. //An exception safe way of ensuring that the thread termination hooks are called.
  50. class jlib_decl QueryTerminationCleanup
  51. {
  52. bool isPooled;
  53. public:
  54. inline QueryTerminationCleanup(bool _isPooled) : isPooled(_isPooled) { }
  55. inline ~QueryTerminationCleanup() { callThreadTerminationHooks(isPooled); }
  56. };
  57. class jlib_decl Thread : public CInterface, public IThread
  58. {
  59. private:
  60. ThreadId threadid;
  61. unsigned short stacksize; // in 4K blocks
  62. int prioritydelta;
  63. int nicelevel;
  64. bool alive;
  65. unsigned tidlog;
  66. #ifdef _WIN32
  67. HANDLE hThread;
  68. static unsigned WINAPI _threadmain(LPVOID v);
  69. #else
  70. static void *_threadmain(void *v);
  71. #endif
  72. virtual int begin();
  73. void init(const char *name);
  74. void handleException(IException *e);
  75. void adjustNiceLevel();
  76. protected:
  77. struct cThreadName: implements IThreadName
  78. {
  79. char *threadname;
  80. const char *get() { return threadname; }
  81. } cthreadname;
  82. IThreadName *ithreadname;
  83. public:
  84. #ifndef _WIN32
  85. Semaphore suspend;
  86. Semaphore starting;
  87. #endif
  88. Semaphore stopped;
  89. IMPLEMENT_IINTERFACE;
  90. Thread(const char *_name) { init(_name); }
  91. Thread() { init(NULL); }
  92. ~Thread();
  93. void adjustPriority(int delta);
  94. bool isCurrentThread() const;
  95. void setNice(int nicelevel);
  96. void setStackSize(size32_t size); // required stack size in bytes - called before start() (obviously)
  97. const char *getName() { const char *ret = ithreadname?ithreadname->get():NULL; return ret?ret:"unknown"; }
  98. bool isAlive() { return alive; }
  99. bool join(unsigned timeout=INFINITE);
  100. virtual void start();
  101. virtual void startRelease();
  102. StringBuffer &getInfo(StringBuffer &str) { str.appendf("%8" I64F "X %6" I64F "d %u: %s",(__int64)threadid,(__int64)threadid,tidlog,getName()); return str; }
  103. const char *getLogInfo(int &thandle,unsigned &tid) {
  104. #ifdef _WIN32
  105. thandle = (int)(memsize_t)hThread;
  106. #elif defined __FreeBSD__ || defined __APPLE__
  107. thandle = (int)(memsize_t)threadid;
  108. #else
  109. thandle = (int)threadid;
  110. #endif
  111. tid = tidlog;
  112. return getName();
  113. }
  114. // run method not implemented - concrete derived classes must do so
  115. static void setDefaultStackSize(size32_t size); // NB under windows requires linker setting (/stack:)
  116. IThreadName *queryThreadName() { return ithreadname; }
  117. void setThreadName(IThreadName *name) { ithreadname = name; }
  118. };
  119. interface IThreaded
  120. {
  121. virtual void threadmain() = 0;
  122. protected:
  123. virtual ~IThreaded() {}
  124. };
  125. // utility class, useful for containing a thread
  126. class CThreaded : public Thread
  127. {
  128. IThreaded *owner;
  129. public:
  130. inline CThreaded(const char *name, IThreaded *_owner) : Thread(name), owner(_owner) { }
  131. inline CThreaded(const char *name) : Thread(name) { owner = NULL; }
  132. inline void init(IThreaded *_owner) { owner = _owner; start(); }
  133. virtual int run() { owner->threadmain(); return 1; }
  134. };
  135. extern jlib_decl void asyncStart(IThreaded & threaded);
  136. extern jlib_decl void asyncStart(const char * name, IThreaded & threaded);
  137. #if defined(__cplusplus) && __cplusplus >= 201100
  138. extern jlib_decl void asyncStart(std::function<void()> func);
  139. #endif
  140. // Similar to above, but the underlying thread always remains running. This can make repeated start + join's significantly quicker
  141. class jlib_decl CThreadedPersistent
  142. {
  143. class CAThread : public Thread
  144. {
  145. CThreadedPersistent &owner;
  146. public:
  147. CAThread(CThreadedPersistent &_owner, const char *name) : Thread(name), owner(_owner) { }
  148. virtual int run() { owner.threadmain(); return 1; }
  149. } athread;
  150. Owned<IException> exception;
  151. IThreaded *owner;
  152. Semaphore sem, joinSem;
  153. std::atomic_uint state;
  154. bool halt;
  155. enum ThreadStates { s_ready, s_running, s_joining };
  156. void threadmain();
  157. public:
  158. CThreadedPersistent(const char *name, IThreaded *_owner);
  159. ~CThreadedPersistent();
  160. void start();
  161. bool join(unsigned timeout, bool throwException = true);
  162. };
  163. // Asynchronous 'for' utility class
  164. // see HRPCUTIL.CPP for example of usage
  165. class jlib_decl CAsyncFor
  166. {
  167. public:
  168. void For(unsigned num,unsigned maxatonce,bool abortFollowingException=false,bool shuffled=false);
  169. virtual void Do(unsigned idx=0)=0;
  170. };
  171. // ---------------------------------------------------------------------------
  172. // Thread Pools
  173. // ---------------------------------------------------------------------------
  174. interface IPooledThread: extends IInterface // base class for deriving pooled thread (alternative to Thread)
  175. {
  176. public:
  177. virtual void init(void *param) = 0; // called before threadmain started (from within start)
  178. virtual void threadmain() = 0; // where threads code goes (param is passed from start)
  179. virtual bool stop() = 0; // called to cause threadmain to return, returns false if request rejected
  180. virtual bool canReuse() const = 0; // return true if object can be re-used (after stopped), otherwise released
  181. };
  182. interface IThreadFactory: extends IInterface // factory for creating new pooled instances (called when pool empty)
  183. {
  184. virtual IPooledThread *createNew()=0;
  185. };
  186. typedef IIteratorOf<IPooledThread> IPooledThreadIterator;
  187. typedef unsigned PooledThreadHandle;
  188. interface IThreadPool : extends IInterface
  189. {
  190. virtual PooledThreadHandle start(void *param)=0; // starts a new thread reuses stopped pool entries
  191. virtual PooledThreadHandle start(void *param,const char *name)=0; // starts a new thread reuses stopped pool entries
  192. virtual PooledThreadHandle start(void *param,const char *name,unsigned timeout)=0; // starts a new thread reuses stopped pool entries, throws exception if can't start within timeout
  193. virtual bool stop(PooledThreadHandle handle)=0; // initiates stop on specified thread (may return false)
  194. virtual bool stopAll(bool tryall=false)=0; // initiates stop on all threads, if tryall continues even if one or more fails
  195. virtual bool join(PooledThreadHandle handle,unsigned timeout=INFINITE)=0;
  196. // waits for a single thread to terminate
  197. virtual bool joinAll(bool del=true,unsigned timeout=INFINITE)=0; // waits for all threads in thread pool to terminate
  198. // if del true frees all pooled threads
  199. virtual IPooledThreadIterator *running()=0; // return an iterator for all currently running threads
  200. virtual unsigned runningCount()=0; // number of currently running threads
  201. virtual PooledThreadHandle startNoBlock(void *param)=0; // starts a new thread if it can do so without blocking, else throws exception
  202. virtual PooledThreadHandle startNoBlock(void *param,const char *name)=0; // starts a new thread if it can do so without blocking, else throws exception
  203. virtual void setStartDelayTracing(unsigned secs) = 0; // set start delay tracing period
  204. };
  205. extern jlib_decl IThreadPool *createThreadPool(
  206. const char *poolname, // trace name of pool
  207. IThreadFactory *factory, // factory for creating new thread instances
  208. IExceptionHandler *exceptionHandler=NULL, // optional exception handler
  209. unsigned defaultmax=50, // maximum number of threads before starts blocking
  210. unsigned delay=1000, // maximum delay on each block
  211. unsigned stacksize=0, // stack size (bytes) 0 is default
  212. unsigned timeoutOnRelease=INFINITE, // maximum time waited for thread to terminate on releasing pool
  213. unsigned targetpoolsize=0 // target maximum size of pool (default same as defaultmax)
  214. );
  215. extern jlib_decl StringBuffer &getThreadList(StringBuffer &str);
  216. extern jlib_decl unsigned getThreadCount();
  217. extern jlib_decl StringBuffer &getThreadName(int thandle,unsigned logtid,StringBuffer &name); // either thandle or tid should be 0
  218. // Simple pipe process support
  219. interface ISimpleReadStream;
  220. #define START_FAILURE (199) // return code if program cannot be started
  221. interface IPipeProcessException : extends IException
  222. {
  223. };
  224. extern jlib_decl IPipeProcessException *createPipeErrnoException(int code, const char *msg);
  225. extern jlib_decl IPipeProcessException *createPipeErrnoExceptionV(int code, const char *msg, ...) __attribute__((format(printf, 2, 3)));
  226. interface IPipeProcess: extends IInterface
  227. {
  228. virtual bool run(const char *title,const char *prog, const char *dir,
  229. bool hasinput,bool hasoutput,bool haserror=false,
  230. size32_t stderrbufsize=0, // set to non-zero to automatically buffer stderror output
  231. bool newProcessGroup=false) __attribute__ ((warn_unused_result)) = 0;
  232. virtual bool hasInput() = 0; // i.e. can write to pipe
  233. virtual size32_t write(size32_t sz, const void *buffer) = 0; // write pipe process standard output
  234. virtual bool hasOutput() = 0; // i.e. can read from pipe
  235. virtual size32_t read(size32_t sz, void *buffer) = 0; // read from pipe process standard output
  236. virtual ISimpleReadStream *getOutputStream() = 0; // read from pipe process standard output
  237. virtual bool hasError() = 0; // i.e. can read from pipe stderr
  238. virtual size32_t readError(size32_t sz, void *buffer) = 0; // read from pipe process standard error
  239. virtual ISimpleReadStream *getErrorStream() = 0; // read from pipe process standard error
  240. virtual unsigned wait() = 0; // returns return code
  241. virtual unsigned wait(unsigned timeoutms, bool &timedout) = 0; // sets timedout to true if times out
  242. virtual void closeInput() = 0; // indicate finished input to pipe
  243. virtual void closeOutput() = 0; // indicate finished reading from pipe (generally called automatically)
  244. virtual void closeError() = 0; // indicate finished reading from pipe stderr
  245. virtual void abort() = 0;
  246. virtual void notifyTerminated(HANDLE pid,unsigned retcode) = 0; // internal
  247. virtual HANDLE getProcessHandle() = 0; // used to auto kill
  248. virtual void setenv(const char *var, const char *value) = 0; // Set a value to be passed in the called process environment
  249. };
  250. extern jlib_decl IPipeProcess *createPipeProcess(const char *allowedprograms=NULL);
  251. //--------------------------------------------------------
  252. interface IWorkQueueItem: extends IInterface
  253. {
  254. virtual void execute()=0;
  255. };
  256. interface IWorkQueueThread: extends IInterface
  257. {
  258. virtual void post(IWorkQueueItem *item)=0; // takes ownership of item
  259. virtual void wait()=0;
  260. virtual unsigned pending()=0;
  261. };
  262. // Simple lightweight async worker queue
  263. // internally thread persists for specified time waiting before self destroying
  264. extern jlib_decl IWorkQueueThread *createWorkQueueThread(unsigned persisttime=1000*60);
  265. #endif