jthread.cpp 70 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490
  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. #include "jthread.hpp"
  14. #include "jlib.hpp"
  15. #include "jfile.hpp"
  16. #include "jmutex.hpp"
  17. #include "jexcept.hpp"
  18. #include "jmisc.hpp"
  19. #include "jqueue.tpp"
  20. #include "jregexp.hpp"
  21. #include <assert.h>
  22. #ifdef _WIN32
  23. #include <process.h>
  24. #else
  25. #include <unistd.h>
  26. #include <sys/wait.h>
  27. #include <sys/syscall.h>
  28. #include <sys/types.h>
  29. #include <sys/resource.h>
  30. #endif
  31. #define LINUX_STACKSIZE_CAP (0x200000)
  32. //#define NO_CATCHALL
  33. static __thread ThreadTermFunc threadTerminationHook;
  34. ThreadTermFunc addThreadTermFunc(ThreadTermFunc onTerm)
  35. {
  36. ThreadTermFunc old = threadTerminationHook;
  37. threadTerminationHook = onTerm;
  38. return old;
  39. }
  40. void callThreadTerminationHooks(bool isPooled)
  41. {
  42. if (threadTerminationHook)
  43. {
  44. (*threadTerminationHook)(isPooled);
  45. }
  46. }
  47. PointerArray *exceptionHandlers = NULL;
  48. MODULE_INIT(INIT_PRIORITY_JTHREAD)
  49. {
  50. exceptionHandlers = new PointerArray();
  51. return true;
  52. }
  53. MODULE_EXIT()
  54. {
  55. if (threadTerminationHook)
  56. (*threadTerminationHook)(false); // May be too late :(
  57. delete exceptionHandlers;
  58. }
  59. void addThreadExceptionHandler(IExceptionHandler *handler)
  60. {
  61. assertex(exceptionHandlers); // have to ensure MODULE_INIT has appropriate priority.
  62. exceptionHandlers->append(handler);
  63. }
  64. void removeThreadExceptionHandler(IExceptionHandler *handler)
  65. {
  66. exceptionHandlers->zap(handler);
  67. }
  68. static bool SEHHandling = false;
  69. void enableThreadSEH() { SEHHandling=true; }
  70. void disableThreadSEH() { SEHHandling=false; } // only prevents new threads from having SEH handler, no mech. for turning off existing threads SEH handling.
  71. static ICopyArrayOf<Thread> ThreadList;
  72. static CriticalSection ThreadListSem;
  73. static size32_t defaultThreadStackSize=0;
  74. static ICopyArrayOf<Thread> ThreadDestroyList;
  75. static SpinLock ThreadDestroyListLock;
  76. #ifdef _WIN32
  77. extern void *EnableSEHtranslation();
  78. unsigned WINAPI Thread::_threadmain(LPVOID v)
  79. #else
  80. void *Thread::_threadmain(void *v)
  81. #endif
  82. {
  83. Thread * t = (Thread *)v;
  84. #ifdef _WIN32
  85. if (SEHHandling)
  86. EnableSEHtranslation();
  87. #else
  88. t->tidlog = threadLogID();
  89. #endif
  90. int ret = t->begin();
  91. char *&threadname = t->cthreadname.threadname;
  92. if (threadname) {
  93. memsize_t l=strlen(threadname);
  94. char *newname = (char *)malloc(l+8+1);
  95. memcpy(newname,"Stopped ",8);
  96. memcpy(newname+8,threadname,l+1);
  97. char *oldname = threadname;
  98. threadname = newname;
  99. free(oldname);
  100. }
  101. {
  102. // need to ensure joining thread does not race with us to release
  103. t->Link(); // extra safety link
  104. {
  105. SpinBlock block(ThreadDestroyListLock);
  106. ThreadDestroyList.append(*t);
  107. }
  108. try {
  109. t->stopped.signal();
  110. if (t->Release()) {
  111. PROGLOG("extra unlinked thread");
  112. PrintStackReport();
  113. }
  114. else
  115. t->Release();
  116. }
  117. catch (...) {
  118. PROGLOG("thread release exception");
  119. throw;
  120. }
  121. {
  122. SpinBlock block(ThreadDestroyListLock);
  123. ThreadDestroyList.zap(*t); // hopefully won't get too big (i.e. one entry!)
  124. }
  125. }
  126. #if defined(_WIN32)
  127. return ret;
  128. #else
  129. return (void *) (memsize_t)ret;
  130. #endif
  131. }
  132. // JCSMORE - should have a setPriority(), unsupported under _WIN32
  133. void Thread::adjustPriority(int delta)
  134. {
  135. if (delta < -2)
  136. prioritydelta = -2;
  137. else if (delta > 2)
  138. prioritydelta = 2;
  139. else
  140. prioritydelta = delta;
  141. if (alive)
  142. {
  143. #if defined(_WIN32)
  144. int priority;
  145. switch (delta)
  146. {
  147. case -2: priority = THREAD_PRIORITY_LOWEST; break;
  148. case -1: priority = THREAD_PRIORITY_BELOW_NORMAL; break;
  149. case 0: priority = THREAD_PRIORITY_NORMAL; break;
  150. case +1: priority = THREAD_PRIORITY_ABOVE_NORMAL; break;
  151. case +2: priority = THREAD_PRIORITY_HIGHEST; break;
  152. }
  153. SetThreadPriority(hThread, priority);
  154. #else
  155. //MORE - What control is there?
  156. int policy;
  157. sched_param param;
  158. int rc;
  159. if (( rc = pthread_getschedparam(threadid, &policy, &param)) != 0)
  160. DBGLOG("pthread_getschedparam error: %d", rc);
  161. switch (delta)
  162. {
  163. // JCS - doubtful whether these good values...
  164. case -2: param.sched_priority = 0; policy =SCHED_OTHER; break;
  165. case -1: param.sched_priority = 0; policy =SCHED_OTHER; break;
  166. case 0: param.sched_priority = 0; policy =SCHED_OTHER; break;
  167. case +1: param.sched_priority = (sched_get_priority_max(SCHED_RR)-sched_get_priority_min(SCHED_RR))/2; policy =SCHED_RR; break;
  168. case +2: param.sched_priority = sched_get_priority_max(SCHED_RR); policy =SCHED_RR; break;
  169. }
  170. if(( rc = pthread_setschedparam(threadid, policy, &param)) != 0)
  171. DBGLOG("pthread_setschedparam error: %d policy=%i pr=%i id=%" I64F "u PID=%i", rc,policy,param.sched_priority,(unsigned __int64) threadid,getpid());
  172. else
  173. DBGLOG("priority set id=%" I64F "u policy=%i pri=%i PID=%i",(unsigned __int64) threadid,policy,param.sched_priority,getpid());
  174. #endif
  175. }
  176. }
  177. void Thread::adjustNiceLevel()
  178. {
  179. #if defined(_WIN32)
  180. int priority;
  181. if(nicelevel < -15)
  182. priority = THREAD_PRIORITY_TIME_CRITICAL;
  183. else if(nicelevel >= -15 && nicelevel < -10)
  184. priority = THREAD_PRIORITY_HIGHEST;
  185. else if(nicelevel >= -10 && nicelevel < 0)
  186. priority = THREAD_PRIORITY_ABOVE_NORMAL;
  187. else if(nicelevel == 0)
  188. priority = THREAD_PRIORITY_NORMAL;
  189. else if(nicelevel > 0 && nicelevel <= 10)
  190. priority = THREAD_PRIORITY_BELOW_NORMAL;
  191. else if(nicelevel > 10 && nicelevel <= 15)
  192. priority = THREAD_PRIORITY_LOWEST;
  193. else if(nicelevel >15)
  194. priority = THREAD_PRIORITY_IDLE;
  195. SetThreadPriority(hThread, priority);
  196. #elif defined(__linux__)
  197. setpriority(PRIO_PROCESS, 0, nicelevel);
  198. #else
  199. UNIMPLEMENTED;
  200. #endif
  201. }
  202. bool Thread::isCurrentThread() const
  203. {
  204. return GetCurrentThreadId() == threadid;
  205. }
  206. // _nicelevel ranges from -20 to 19, the higher the nice level, the less cpu time the thread will get.
  207. void Thread::setNice(int _nicelevel)
  208. {
  209. if (_nicelevel < -20 || _nicelevel > 19)
  210. throw MakeStringException(0, "nice level should be between -20 and 19");
  211. if(alive)
  212. throw MakeStringException(0, "nice can only be set before the thread is started.");
  213. nicelevel = _nicelevel;
  214. }
  215. void Thread::setStackSize(size32_t size)
  216. {
  217. stacksize = (unsigned short)(size/0x1000);
  218. }
  219. void Thread::setDefaultStackSize(size32_t size)
  220. {
  221. defaultThreadStackSize = size; // has no effect under windows (though may be used for calculations later)
  222. }
  223. int Thread::begin()
  224. {
  225. if(nicelevel)
  226. adjustNiceLevel();
  227. #ifndef _WIN32
  228. starting.signal();
  229. suspend.wait();
  230. #endif
  231. int ret=-1;
  232. try {
  233. ret = run();
  234. }
  235. catch (IException *e)
  236. {
  237. handleException(e);
  238. }
  239. #ifndef NO_CATCHALL
  240. catch (...)
  241. {
  242. handleException(MakeStringException(0, "Unknown exception in Thread %s", getName()));
  243. }
  244. #endif
  245. callThreadTerminationHooks(false);
  246. #ifdef _WIN32
  247. #ifndef _DEBUG
  248. CloseHandle(hThread); // leak handle when debugging,
  249. // fixes some lockups/crashes in the debugger when lots of threads being created
  250. #endif
  251. hThread = NULL;
  252. #endif
  253. //alive = false; // not safe here
  254. return ret;
  255. }
  256. void Thread::handleException(IException *e)
  257. {
  258. assertex(exceptionHandlers);
  259. if (exceptionHandlers->ordinality() == 0)
  260. {
  261. PrintExceptionLog(e,getName());
  262. //throw; // don't rethrow unhandled, preferable over alternative of causing process death
  263. e->Release();
  264. }
  265. else
  266. {
  267. PrintExceptionLog(e,getName());
  268. bool handled = false;
  269. ForEachItemIn(ie, *exceptionHandlers)
  270. {
  271. IExceptionHandler *handler = (IExceptionHandler *) exceptionHandlers->item(ie);
  272. handled = handler->fireException(e) || handled;
  273. }
  274. if (!handled)
  275. {
  276. // if nothing choose to handle it.
  277. EXCLOG(e, NULL);
  278. //throw e; // don't rethrow unhandled, preferable over alternative of causing process death
  279. }
  280. e->Release();
  281. }
  282. }
  283. void Thread::init(const char *_name)
  284. {
  285. #ifdef _WIN32
  286. hThread = NULL;
  287. #endif
  288. threadid = 0;
  289. tidlog = 0;
  290. alive = false;
  291. cthreadname.threadname = (NULL == _name) ? NULL : strdup(_name);
  292. ithreadname = &cthreadname;
  293. prioritydelta = 0;
  294. nicelevel = 0;
  295. stacksize = 0; // default is EXE default stack size (set by /STACK)
  296. }
  297. void Thread::start()
  298. {
  299. if (alive) {
  300. IWARNLOG("Thread::start(%s) - Thread already started!",getName());
  301. PrintStackReport();
  302. #ifdef _DEBUG
  303. throw MakeStringException(-1,"Thread::start(%s) - Thread already started!",getName());
  304. #endif
  305. return;
  306. }
  307. Link();
  308. startRelease();
  309. }
  310. void Thread::startRelease()
  311. {
  312. assertex(!alive);
  313. stopped.reinit(0); // just in case restarting
  314. #ifdef _WIN32
  315. hThread = (HANDLE)_beginthreadex(NULL, 0x1000*(unsigned)stacksize, Thread::_threadmain, this, CREATE_SUSPENDED, (unsigned *)&threadid);
  316. if (!hThread || !threadid)
  317. {
  318. Release();
  319. throw makeOsException(GetLastError());
  320. }
  321. #else
  322. int status;
  323. unsigned numretrys = 8;
  324. unsigned delay = 1000;
  325. for (;;) {
  326. pthread_attr_t attr;
  327. pthread_attr_init(&attr);
  328. pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);
  329. pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
  330. if (stacksize)
  331. pthread_attr_setstacksize(&attr, (unsigned)stacksize*0x1000);
  332. else if (defaultThreadStackSize)
  333. pthread_attr_setstacksize(&attr, defaultThreadStackSize);
  334. else {
  335. #ifndef __64BIT__ // no need to cap 64bit
  336. size_t defss=0;
  337. pthread_attr_getstacksize(&attr, &defss);
  338. if (defss>LINUX_STACKSIZE_CAP)
  339. pthread_attr_setstacksize(&attr, LINUX_STACKSIZE_CAP);
  340. #endif
  341. }
  342. status = pthread_create(&threadid, &attr, Thread::_threadmain, this);
  343. if ((status==EAGAIN)||(status==EINTR)) {
  344. if (numretrys--==0)
  345. break;
  346. IWARNLOG("pthread_create(%d): Out of threads, retrying...",status);
  347. Sleep(delay);
  348. delay *= 2;
  349. }
  350. else
  351. break;
  352. }
  353. if (status) {
  354. threadid = 0;
  355. Release();
  356. IERRLOG("pthread_create returns %d",status);
  357. PrintStackReport();
  358. PrintMemoryReport();
  359. StringBuffer s;
  360. getThreadList(s);
  361. IERRLOG("Running threads:\n %s",s.str());
  362. throw makeOsException(status);
  363. }
  364. unsigned retryCount = 10;
  365. for (;;)
  366. {
  367. if (starting.wait(1000*10))
  368. break;
  369. else if (0 == --retryCount)
  370. throw MakeStringException(-1, "Thread::start(%s) failed", getName());
  371. IWARNLOG("Thread::start(%s) stalled, waiting to start, retrying", getName());
  372. }
  373. #endif
  374. alive = true;
  375. if (prioritydelta)
  376. adjustPriority(prioritydelta);
  377. {
  378. CriticalBlock block(ThreadListSem);
  379. ThreadList.zap(*this); // just in case restarting
  380. ThreadList.append(*this);
  381. }
  382. #ifdef _WIN32
  383. DWORD count = ResumeThread(hThread);
  384. assertex(count == 1);
  385. #else
  386. suspend.signal();
  387. #endif
  388. }
  389. bool Thread::join(unsigned timeout)
  390. {
  391. if (!alive&&!threadid) {
  392. #ifdef _DEBUG
  393. PROGLOG("join on unstarted thread!");
  394. PrintStackReport();
  395. #endif
  396. return true;
  397. }
  398. if (!stopped.wait(timeout))
  399. return false;
  400. if (!alive) // already joined
  401. {
  402. stopped.signal();
  403. return true;
  404. }
  405. unsigned st = 0;
  406. for (;;) { // this is to prevent race with destroy
  407. // (because Thread objects are not always link counted!)
  408. {
  409. SpinBlock block(ThreadDestroyListLock);
  410. if (ThreadDestroyList.find(*this)==NotFound)
  411. break;
  412. }
  413. #ifdef _DEBUG
  414. if (st==10)
  415. PROGLOG("Thread::join race");
  416. #endif
  417. Sleep(st); // switch back to exiting thread (not very elegant!)
  418. st++;
  419. if (st>10)
  420. st = 10; // note must be non-zero for high priority threads
  421. }
  422. #ifdef _DEBUG
  423. int c = getLinkCount();
  424. if (c>=DEAD_PSEUDO_COUNT) {
  425. PROGLOG("Dead/Dying thread joined! %d",c);
  426. PrintStackReport();
  427. }
  428. #endif
  429. alive = false; // should be safe here
  430. stopped.signal(); // signal stopped again, to prevent any parallel call from blocking.
  431. return true;
  432. }
  433. Thread::~Thread()
  434. {
  435. ithreadname = &cthreadname; // safer (as derived classes destroyed)
  436. #ifdef _DEBUG
  437. if (alive) {
  438. if (!stopped.wait(0)) { // see if fell out of threadmain and signal stopped
  439. PROGLOG("Live thread killed! %s",getName());
  440. PrintStackReport();
  441. }
  442. // don't need to resignal as we are on way out
  443. }
  444. #endif
  445. Link();
  446. // DBGLOG("Thread %x (%s) destroyed\n", threadid, threadname);
  447. {
  448. CriticalBlock block(ThreadListSem);
  449. ThreadList.zap(*this);
  450. }
  451. free(cthreadname.threadname);
  452. cthreadname.threadname = NULL;
  453. }
  454. unsigned getThreadCount()
  455. {
  456. CriticalBlock block(ThreadListSem);
  457. return ThreadList.ordinality();
  458. }
  459. StringBuffer & getThreadList(StringBuffer &str)
  460. {
  461. CriticalBlock block(ThreadListSem);
  462. ForEachItemIn(i,ThreadList) {
  463. Thread &item=ThreadList.item(i);
  464. item.getInfo(str).append("\n");
  465. }
  466. return str;
  467. }
  468. StringBuffer &getThreadName(int thandle,unsigned tid,StringBuffer &name)
  469. {
  470. CriticalBlock block(ThreadListSem);
  471. bool found=false;
  472. ForEachItemIn(i,ThreadList) {
  473. Thread &item=ThreadList.item(i);
  474. int h;
  475. unsigned t;
  476. const char *s = item.getLogInfo(h,t);
  477. if (s&&*s&&((thandle==0)||(h==thandle))&&((tid==0)||(t==tid))) {
  478. if (found) {
  479. name.clear();
  480. break; // only return if unambiguous
  481. }
  482. name.append(s);
  483. found = true;
  484. }
  485. }
  486. return name;
  487. }
  488. // CThreadedPersistent
  489. CThreadedPersistent::CThreadedPersistent(const char *name, IThreaded *_owner) : athread(*this, name), owner(_owner), state(s_ready)
  490. {
  491. halt = false;
  492. athread.start();
  493. }
  494. CThreadedPersistent::~CThreadedPersistent()
  495. {
  496. join(INFINITE, false);
  497. halt = true;
  498. sem.signal();
  499. athread.join();
  500. }
  501. void CThreadedPersistent::threadmain()
  502. {
  503. for (;;)
  504. {
  505. sem.wait();
  506. if (halt)
  507. break;
  508. try
  509. {
  510. owner->threadmain();
  511. // Note we do NOT call the thread reset hook here - these threads are expected to be able to preserve state, I think
  512. }
  513. catch (IException *e)
  514. {
  515. VStringBuffer errMsg("CThreadedPersistent (%s)", athread.getName());
  516. EXCLOG(e, errMsg.str());
  517. exception.setown(e);
  518. joinSem.signal(); // leave in running state, signal to join to handle
  519. continue;
  520. }
  521. unsigned expected = s_running;
  522. if (!state.compare_exchange_strong(expected, s_ready))
  523. {
  524. expected = s_joining;
  525. if (state.compare_exchange_strong(expected, s_ready))
  526. joinSem.signal();
  527. }
  528. }
  529. }
  530. void CThreadedPersistent::start()
  531. {
  532. unsigned expected = s_ready;
  533. if (!state.compare_exchange_strong(expected, s_running))
  534. {
  535. VStringBuffer msg("CThreadedPersistent::start(%s) - not ready", athread.getName());
  536. IWARNLOG("%s", msg.str());
  537. PrintStackReport();
  538. throw MakeStringExceptionDirect(-1, msg.str());
  539. }
  540. sem.signal();
  541. }
  542. bool CThreadedPersistent::join(unsigned timeout, bool throwException)
  543. {
  544. unsigned expected = s_running;
  545. if (state.compare_exchange_strong(expected, s_joining))
  546. {
  547. if (!joinSem.wait(timeout))
  548. {
  549. unsigned expected = s_joining;
  550. if (state.compare_exchange_strong(expected, s_running)) // if still joining, restore running state
  551. return false;
  552. // if here, threadmain() set s_ready after timeout and has or will signal
  553. if (!joinSem.wait(60000) && throwException) // should be instant
  554. throwUnexpected();
  555. return true;
  556. }
  557. if (throwException && exception.get())
  558. {
  559. // switch back to ready state and throw
  560. Owned<IException> e = exception.getClear();
  561. unsigned expected = s_joining;
  562. if (!state.compare_exchange_strong(expected, s_ready))
  563. throwUnexpected();
  564. throw e.getClear();
  565. }
  566. }
  567. return true;
  568. }
  569. //class CAsyncFor
  570. void CAsyncFor::For(unsigned num,unsigned maxatonce,bool abortFollowingException, bool shuffled)
  571. {
  572. if (num <= 1)
  573. {
  574. if (num == 1)
  575. Do(0);
  576. return;
  577. }
  578. Mutex errmutex;
  579. IException *e=NULL;
  580. Owned<IShuffledIterator> shuffler;
  581. if (shuffled) {
  582. shuffler.setown(createShuffledIterator(num));
  583. shuffler->first(); // prime (needed to make thread safe)
  584. }
  585. unsigned i;
  586. if (maxatonce==1) { // no need for threads
  587. for (i=0;i<num;i++) {
  588. unsigned idx = shuffled?shuffler->lookup(i):i;
  589. try {
  590. Do(idx);
  591. }
  592. catch (IException * _e)
  593. {
  594. if (e)
  595. _e->Release(); // only return first
  596. else
  597. e = _e;
  598. if (abortFollowingException)
  599. break;
  600. }
  601. }
  602. }
  603. else {
  604. if (maxatonce==0 || maxatonce > num)
  605. maxatonce = num;
  606. if (maxatonce < num)
  607. {
  608. class cdothread: public Thread
  609. {
  610. public:
  611. Mutex *errmutex;
  612. Semaphore &ready;
  613. IException *&erre;
  614. unsigned idx;
  615. CAsyncFor *self;
  616. cdothread(CAsyncFor *_self,unsigned _idx,Semaphore &_ready,Mutex *_errmutex,IException *&_e)
  617. : Thread("CAsyncFor"),ready(_ready),erre(_e)
  618. {
  619. errmutex =_errmutex;
  620. idx = _idx;
  621. self = _self;
  622. }
  623. int run()
  624. {
  625. try {
  626. self->Do(idx);
  627. }
  628. catch (IException * _e)
  629. {
  630. synchronized block(*errmutex);
  631. if (erre)
  632. _e->Release(); // only return first
  633. else
  634. erre = _e;
  635. }
  636. #ifndef NO_CATCHALL
  637. catch (...)
  638. {
  639. synchronized block(*errmutex);
  640. if (!erre)
  641. erre = MakeStringException(0, "Unknown exception in Thread %s", getName());
  642. }
  643. #endif
  644. ready.signal();
  645. return 0;
  646. }
  647. };
  648. Semaphore ready;
  649. for (i=0;(i<num)&&(i<maxatonce);i++)
  650. ready.signal();
  651. IArrayOf<Thread> started;
  652. started.ensure(num);
  653. for (i=0;i<num;i++) {
  654. ready.wait();
  655. if (abortFollowingException && e) break;
  656. Owned<Thread> thread = new cdothread(this,shuffled?shuffler->lookup(i):i,ready,&errmutex,e);
  657. thread->start();
  658. started.append(*thread.getClear());
  659. }
  660. ForEachItemIn(idx, started)
  661. {
  662. started.item(idx).join();
  663. }
  664. }
  665. else
  666. {
  667. // Common case of execute all at once can be optimized a little
  668. // Note that shuffle and abortFollowingException are meaningless when executing all at once
  669. class cdothread: public Thread
  670. {
  671. public:
  672. Mutex *errmutex;
  673. IException *&erre;
  674. unsigned idx;
  675. CAsyncFor *self;
  676. cdothread(CAsyncFor *_self,unsigned _idx,Mutex *_errmutex,IException *&_e)
  677. : Thread("CAsyncFor"),erre(_e)
  678. {
  679. errmutex =_errmutex;
  680. idx = _idx;
  681. self = _self;
  682. }
  683. int run()
  684. {
  685. try {
  686. self->Do(idx);
  687. }
  688. catch (IException * _e)
  689. {
  690. synchronized block(*errmutex);
  691. if (erre)
  692. _e->Release(); // only return first
  693. else
  694. erre = _e;
  695. }
  696. #ifndef NO_CATCHALL
  697. catch (...)
  698. {
  699. synchronized block(*errmutex);
  700. if (!erre)
  701. erre = MakeStringException(0, "Unknown exception in Thread %s", getName());
  702. }
  703. #endif
  704. return 0;
  705. }
  706. };
  707. IArrayOf<Thread> started;
  708. started.ensure(num);
  709. for (i=0;i<num-1;i++)
  710. {
  711. Owned<Thread> thread = new cdothread(this,i,&errmutex,e);
  712. thread->start();
  713. started.append(*thread.getClear());
  714. }
  715. Do(num-1);
  716. ForEachItemIn(idx, started)
  717. {
  718. started.item(idx).join();
  719. }
  720. }
  721. }
  722. if (e)
  723. throw e;
  724. }
  725. //---------------------------------------------------------------------------------------------------------------------
  726. class CSimpleFunctionThread : public Thread
  727. {
  728. std::function<void()> func;
  729. public:
  730. inline CSimpleFunctionThread(std::function<void()> _func) : Thread("TaskProcessor"), func(_func) { }
  731. virtual int run()
  732. {
  733. func();
  734. return 1;
  735. }
  736. };
  737. void asyncStart(IThreaded & threaded)
  738. {
  739. CThreaded * thread = new CThreaded("AsyncStart", &threaded);
  740. thread->startRelease();
  741. }
  742. void asyncStart(const char * name, IThreaded & threaded)
  743. {
  744. CThreaded * thread = new CThreaded(name, &threaded);
  745. thread->startRelease();
  746. }
  747. //Experimental - is this a useful function to replace some uses of IThreaded?
  748. void asyncStart(std::function<void()> func)
  749. {
  750. (new CSimpleFunctionThread(func))->startRelease();
  751. }
  752. // ---------------------------------------------------------------------------
  753. // Thread Pools
  754. // ---------------------------------------------------------------------------
  755. class CPooledThreadWrapper;
  756. class CThreadPoolBase
  757. {
  758. public:
  759. virtual ~CThreadPoolBase() {}
  760. protected: friend class CPooledThreadWrapper;
  761. IExceptionHandler *exceptionHandler;
  762. CriticalSection crit;
  763. StringAttr poolname;
  764. int donewaiting;
  765. Semaphore donesem;
  766. PointerArray waitingsems;
  767. UnsignedArray waitingids;
  768. bool stopall;
  769. unsigned defaultmax;
  770. unsigned targetpoolsize;
  771. unsigned delay;
  772. Semaphore availsem;
  773. std::atomic_uint numrunning{0};
  774. virtual void notifyStarted(CPooledThreadWrapper *item)=0;
  775. virtual bool notifyStopped(CPooledThreadWrapper *item)=0;
  776. };
  777. class CPooledThreadWrapper: public Thread
  778. {
  779. PooledThreadHandle handle;
  780. IPooledThread *thread;
  781. Semaphore sem;
  782. CThreadPoolBase &parent;
  783. char *runningname;
  784. public:
  785. CPooledThreadWrapper(CThreadPoolBase &_parent,
  786. PooledThreadHandle _handle,
  787. IPooledThread *_thread) // takes ownership of thread
  788. : Thread(StringBuffer("Member of thread pool: ").append(_parent.poolname).str()), parent(_parent)
  789. {
  790. thread = _thread;
  791. handle = _handle;
  792. runningname = strdup(_parent.poolname);
  793. }
  794. ~CPooledThreadWrapper()
  795. {
  796. thread->Release();
  797. free(runningname);
  798. }
  799. void setName(const char *name) { free(runningname); runningname=strdup(name); }
  800. void setHandle(PooledThreadHandle _handle) { handle = _handle; }
  801. PooledThreadHandle queryHandle() { return handle; }
  802. IPooledThread &queryThread() { return *thread; }
  803. void setThread(IPooledThread *_thread) { thread = _thread; } // takes ownership
  804. bool isStopped() { return (handle==0); }
  805. PooledThreadHandle markStopped()
  806. {
  807. PooledThreadHandle ret=handle;
  808. handle = 0;
  809. if (ret) // JCSMORE - I can't see how handle can not be set if here..
  810. parent.numrunning--;
  811. return ret;
  812. }
  813. void markStarted()
  814. {
  815. parent.numrunning++;
  816. }
  817. int run()
  818. {
  819. do
  820. {
  821. sem.wait();
  822. {
  823. CriticalBlock block(parent.crit); // to synchronize
  824. if (parent.stopall)
  825. break;
  826. }
  827. parent.notifyStarted(this);
  828. try
  829. {
  830. char *&threadname = cthreadname.threadname;
  831. char *temp = threadname; // swap running name and threadname
  832. threadname = runningname;
  833. runningname = temp;
  834. thread->threadmain();
  835. temp = threadname; // and back
  836. threadname = runningname;
  837. runningname = temp;
  838. }
  839. catch (IException *e)
  840. {
  841. char *&threadname = cthreadname.threadname;
  842. char *temp = threadname; // swap back
  843. threadname = runningname;
  844. runningname = temp;
  845. handleException(e);
  846. }
  847. #ifndef NO_CATCHALL
  848. catch (...)
  849. {
  850. char *&threadname = cthreadname.threadname;
  851. char *temp = threadname; // swap back
  852. threadname = runningname;
  853. runningname = temp;
  854. handleException(MakeStringException(0, "Unknown exception in Thread from pool %s", parent.poolname.get()));
  855. }
  856. #endif
  857. callThreadTerminationHooks(true); // Reset any per-thread state.
  858. } while (parent.notifyStopped(this));
  859. return 0;
  860. }
  861. void cycle()
  862. {
  863. sem.signal();
  864. }
  865. void go(void *param)
  866. {
  867. thread->init(param);
  868. cycle();
  869. }
  870. bool stop()
  871. {
  872. if (handle)
  873. return thread->stop();
  874. return true;
  875. }
  876. void handleException(IException *e)
  877. {
  878. CriticalBlock block(parent.crit);
  879. PrintExceptionLog(e,parent.poolname.get());
  880. if (!parent.exceptionHandler||!parent.exceptionHandler->fireException(e)) {
  881. }
  882. e->Release();
  883. }
  884. };
  885. class CPooledThreadIterator: implements IPooledThreadIterator, public CInterface
  886. {
  887. unsigned current;
  888. public:
  889. IArrayOf<IPooledThread> threads;
  890. IMPLEMENT_IINTERFACE;
  891. CPooledThreadIterator()
  892. {
  893. current = 0;
  894. }
  895. bool first()
  896. {
  897. current = 0;
  898. return threads.isItem(current);
  899. }
  900. bool next()
  901. {
  902. current++;
  903. return threads.isItem(current);
  904. }
  905. bool isValid()
  906. {
  907. return threads.isItem(current);
  908. }
  909. IPooledThread & query()
  910. {
  911. return threads.item(current);
  912. }
  913. };
  914. class CThreadPool: public CThreadPoolBase, implements IThreadPool, public CInterface
  915. {
  916. CIArrayOf<CPooledThreadWrapper> threadwrappers;
  917. PooledThreadHandle nextid;
  918. IThreadFactory *factory;
  919. unsigned stacksize;
  920. unsigned timeoutOnRelease;
  921. unsigned traceStartDelayPeriod;
  922. unsigned startsInPeriod;
  923. cycle_t startDelayInPeriod;
  924. CCycleTimer overAllTimer;
  925. PooledThreadHandle _start(void *param,const char *name, bool noBlock, unsigned timeout=0)
  926. {
  927. CCycleTimer startTimer;
  928. bool timedout = defaultmax && !availsem.wait(noBlock ? 0 : (timeout>0?timeout:delay));
  929. PooledThreadHandle ret;
  930. {
  931. CriticalBlock block(crit);
  932. if (timedout)
  933. {
  934. if (!availsem.wait(0)) { // make sure take allocated sem if has become available
  935. if (noBlock || timeout > 0)
  936. throw MakeStringException(0, "No threads available in pool %s", poolname.get());
  937. IWARNLOG("Pool limit exceeded for %s", poolname.get());
  938. }
  939. else
  940. timedout = false;
  941. }
  942. if (traceStartDelayPeriod)
  943. {
  944. ++startsInPeriod;
  945. if (timedout)
  946. {
  947. startDelayInPeriod += startTimer.elapsedCycles();
  948. if (overAllTimer.elapsedCycles() >= queryOneSecCycles()*traceStartDelayPeriod) // check avg. delay per minute
  949. {
  950. double totalDelayMs = (static_cast<double>(cycle_to_nanosec(startDelayInPeriod)))/1000000;
  951. double avgDelayMs = (static_cast<double>(cycle_to_nanosec(startDelayInPeriod/startsInPeriod)))/1000000;
  952. unsigned totalElapsedSecs = overAllTimer.elapsedMs()/1000;
  953. PROGLOG("%s: %u threads started in last %u seconds, total delay = %0.2f milliseconds, average delay = %0.2f milliseconds, currently running = %u", poolname.get(), startsInPeriod, totalElapsedSecs, totalDelayMs, avgDelayMs, runningCount());
  954. startsInPeriod = 0;
  955. startDelayInPeriod = 0;
  956. overAllTimer.reset();
  957. }
  958. }
  959. }
  960. CPooledThreadWrapper &t = allocThread();
  961. if (name)
  962. t.setName(name);
  963. t.go(param);
  964. ret = t.queryHandle();
  965. }
  966. Sleep(0);
  967. return ret;
  968. }
  969. public:
  970. IMPLEMENT_IINTERFACE;
  971. CThreadPool(IThreadFactory *_factory,IExceptionHandler *_exceptionHandler,const char *_poolname,unsigned _defaultmax, unsigned _delay, unsigned _stacksize, unsigned _timeoutOnRelease, unsigned _targetpoolsize)
  972. {
  973. poolname.set(_poolname);
  974. factory = LINK(_factory);
  975. exceptionHandler = _exceptionHandler;
  976. nextid = 1;
  977. stopall = false;
  978. defaultmax = _defaultmax;
  979. delay = _delay;
  980. if (defaultmax)
  981. availsem.signal(defaultmax);
  982. stacksize = _stacksize;
  983. timeoutOnRelease = _timeoutOnRelease;
  984. targetpoolsize = _targetpoolsize?_targetpoolsize:defaultmax;
  985. traceStartDelayPeriod = 0;
  986. startsInPeriod = 0;
  987. startDelayInPeriod = 0;
  988. }
  989. ~CThreadPool()
  990. {
  991. stopAll(true);
  992. if (!joinAll(true, timeoutOnRelease))
  993. IWARNLOG("%s; timedout[%d] waiting for threads in pool", poolname.get(), timeoutOnRelease);
  994. CriticalBlock block(crit);
  995. bool first=true;
  996. ForEachItemIn(i,threadwrappers)
  997. {
  998. CPooledThreadWrapper &t = threadwrappers.item(i);
  999. if (!t.isStopped())
  1000. {
  1001. if (first)
  1002. {
  1003. IWARNLOG("Threads still active: ");
  1004. first = false;
  1005. }
  1006. StringBuffer threadInfo;
  1007. PROGLOG("Active thread: %s, info: %s", t.getName(), t.getInfo(threadInfo).str());
  1008. }
  1009. }
  1010. factory->Release();
  1011. }
  1012. CPooledThreadWrapper &allocThread()
  1013. { // called in critical section
  1014. PooledThreadHandle newid=nextid++;
  1015. if (newid==0)
  1016. newid=nextid++;
  1017. ForEachItemIn(i,threadwrappers) {
  1018. CPooledThreadWrapper &it = threadwrappers.item(i);
  1019. if (it.isStopped()) {
  1020. it.setHandle(newid);
  1021. if (!it.queryThread().canReuse()) {
  1022. it.queryThread().Release();
  1023. it.setThread(factory->createNew());
  1024. }
  1025. return it;
  1026. }
  1027. }
  1028. CPooledThreadWrapper &ret = *new CPooledThreadWrapper(*this,newid,factory->createNew());
  1029. if (stacksize)
  1030. ret.setStackSize(stacksize);
  1031. ret.start();
  1032. threadwrappers.append(ret);
  1033. return ret;
  1034. }
  1035. CPooledThreadWrapper *findThread(PooledThreadHandle handle)
  1036. { // called in critical section
  1037. ForEachItemIn(i,threadwrappers) {
  1038. CPooledThreadWrapper &it = threadwrappers.item(i);
  1039. if (it.queryHandle()==handle)
  1040. return &it;
  1041. }
  1042. return NULL;
  1043. }
  1044. PooledThreadHandle startNoBlock(void *param)
  1045. {
  1046. return _start(param, NULL, true);
  1047. }
  1048. PooledThreadHandle startNoBlock(void *param,const char *name)
  1049. {
  1050. return _start(param, name, true);
  1051. }
  1052. PooledThreadHandle start(void *param)
  1053. {
  1054. return _start(param, NULL, false);
  1055. }
  1056. PooledThreadHandle start(void *param,const char *name)
  1057. {
  1058. return _start(param, name, false);
  1059. }
  1060. PooledThreadHandle start(void *param,const char *name, unsigned timeout)
  1061. {
  1062. return _start(param, name, false, timeout);
  1063. }
  1064. bool stop(PooledThreadHandle handle)
  1065. {
  1066. CriticalBlock block(crit);
  1067. CPooledThreadWrapper *t = findThread(handle);
  1068. if (t)
  1069. return t->stop();
  1070. return true; // already stopped
  1071. }
  1072. bool stopAll(bool tryall=false)
  1073. {
  1074. availsem.signal(1000);
  1075. availsem.wait();
  1076. CriticalBlock block(crit);
  1077. bool ret=true;
  1078. ForEachItemIn(i,threadwrappers) {
  1079. CPooledThreadWrapper &it = threadwrappers.item(i);
  1080. if (!it.stop()) {
  1081. ret = false;
  1082. if (!tryall)
  1083. break;
  1084. }
  1085. }
  1086. return ret;
  1087. }
  1088. bool joinWait(CPooledThreadWrapper &t,unsigned timeout)
  1089. {
  1090. // called in critical section
  1091. if (t.isStopped())
  1092. return true;
  1093. Semaphore sem;
  1094. waitingsems.append(&sem);
  1095. waitingids.append(t.queryHandle());
  1096. crit.leave();
  1097. bool ret = sem.wait(timeout);
  1098. crit.enter();
  1099. unsigned i = waitingsems.find(&sem);
  1100. if (i!=NotFound) {
  1101. waitingids.remove(i);
  1102. waitingsems.remove(i);
  1103. }
  1104. return ret;
  1105. }
  1106. bool join(PooledThreadHandle handle,unsigned timeout=INFINITE)
  1107. {
  1108. CriticalBlock block(crit);
  1109. CPooledThreadWrapper *t = findThread(handle);
  1110. if (!t)
  1111. return true; // already stopped
  1112. return joinWait(*t,timeout);
  1113. }
  1114. virtual bool joinAll(bool del,unsigned timeout=INFINITE)
  1115. { // note timeout is for each join
  1116. CriticalBlock block(crit);
  1117. CIArrayOf<CPooledThreadWrapper> tojoin;
  1118. ForEachItemIn(i1,threadwrappers) {
  1119. CPooledThreadWrapper &it = threadwrappers.item(i1);
  1120. it.Link();
  1121. tojoin.append(it);
  1122. }
  1123. ForEachItemIn(i2,tojoin)
  1124. if (!joinWait(tojoin.item(i2),timeout))
  1125. return false;
  1126. if (del) {
  1127. stopall = true;
  1128. ForEachItemIn(i3,tojoin)
  1129. tojoin.item(i3).cycle();
  1130. {
  1131. CriticalUnblock unblock(crit);
  1132. ForEachItemIn(i4,tojoin)
  1133. tojoin.item(i4).join();
  1134. }
  1135. threadwrappers.kill();
  1136. stopall = false;
  1137. }
  1138. return true;
  1139. }
  1140. IPooledThreadIterator *running()
  1141. {
  1142. CriticalBlock block(crit);
  1143. CPooledThreadIterator *ret = new CPooledThreadIterator;
  1144. ForEachItemIn(i,threadwrappers) {
  1145. CPooledThreadWrapper &it = threadwrappers.item(i);
  1146. if (!it.isStopped()) {
  1147. IPooledThread &t = it.queryThread();
  1148. t.Link();
  1149. ret->threads.append(t);
  1150. }
  1151. }
  1152. return ret;
  1153. }
  1154. unsigned runningCount()
  1155. {
  1156. return numrunning;
  1157. }
  1158. void notifyStarted(CPooledThreadWrapper *item)
  1159. {
  1160. item->markStarted();
  1161. }
  1162. bool notifyStopped(CPooledThreadWrapper *item)
  1163. {
  1164. CriticalBlock block(crit);
  1165. PooledThreadHandle myid = item->markStopped();
  1166. ForEachItemIn(i1,waitingids) { // tell anyone waiting
  1167. if (waitingids.item(i1)==myid)
  1168. ((Semaphore *)waitingsems.item(i1))->signal();
  1169. }
  1170. bool ret = true;
  1171. if (defaultmax) {
  1172. unsigned n=threadwrappers.ordinality();
  1173. for (unsigned i2=targetpoolsize;i2<n;i2++) { // only check excess for efficiency
  1174. if (item==&threadwrappers.item(i2)) {
  1175. threadwrappers.remove(i2);
  1176. ret = false;
  1177. break;
  1178. }
  1179. }
  1180. availsem.signal();
  1181. }
  1182. return ret;
  1183. }
  1184. void setStartDelayTracing(unsigned secs)
  1185. {
  1186. traceStartDelayPeriod = secs;
  1187. }
  1188. };
  1189. IThreadPool *createThreadPool(const char *poolname,IThreadFactory *factory,IExceptionHandler *exceptionHandler,unsigned defaultmax, unsigned delay, unsigned stacksize, unsigned timeoutOnRelease, unsigned targetpoolsize)
  1190. {
  1191. return new CThreadPool(factory,exceptionHandler,poolname,defaultmax,delay,stacksize,timeoutOnRelease,targetpoolsize);
  1192. }
  1193. //=======================================================================================================
  1194. static void CheckAllowedProgram(const char *prog,const char *allowed)
  1195. {
  1196. if (!prog||!allowed||(strcmp(allowed,"*")==0))
  1197. return;
  1198. StringBuffer head;
  1199. bool inq = false;
  1200. // note don't have to be too worried about odd quoting as matching fixed list
  1201. while (*prog&&((*prog!=' ')||inq)) {
  1202. if (*prog=='"')
  1203. inq = !inq;
  1204. head.append(*(prog++));
  1205. }
  1206. StringArray list;
  1207. list.appendList(allowed, ",");
  1208. ForEachItemIn(i,list) {
  1209. if (WildMatch(head.str(),list.item(i)))
  1210. return;
  1211. }
  1212. AERRLOG("Unauthorized pipe program(%s)",head.str());
  1213. throw MakeStringException(-1,"Unauthorized pipe program(%s)",head.str());
  1214. }
  1215. class CPipeProcessException : public CSimpleInterfaceOf<IPipeProcessException>
  1216. {
  1217. int errCode;
  1218. StringAttr msg;
  1219. MessageAudience audience;
  1220. public:
  1221. CPipeProcessException(int _errCode, const char *_msg, MessageAudience _audience = MSGAUD_user) : errCode(_errCode), msg(_msg), audience(_audience)
  1222. {
  1223. }
  1224. virtual int errorCode() const override { return errCode; }
  1225. virtual StringBuffer & errorMessage(StringBuffer &str) const override
  1226. {
  1227. if (msg)
  1228. str.append(msg).append(", ");
  1229. return str.append(strerror(errCode));
  1230. }
  1231. MessageAudience errorAudience() const { return audience; }
  1232. };
  1233. IPipeProcessException *createPipeErrnoException(int code, const char *msg)
  1234. {
  1235. return new CPipeProcessException(code, msg);
  1236. }
  1237. IPipeProcessException *createPipeErrnoExceptionV(int code, const char *msg, ...)
  1238. {
  1239. StringBuffer eStr;
  1240. va_list args;
  1241. va_start(args, msg);
  1242. eStr.limited_valist_appendf(1024, msg, args);
  1243. va_end(args);
  1244. return new CPipeProcessException(code, eStr.str());
  1245. }
  1246. class CSimplePipeStream: implements ISimpleReadStream, public CInterface
  1247. {
  1248. public:
  1249. IMPLEMENT_IINTERFACE;
  1250. CSimplePipeStream(IPipeProcess *_pipe, bool _isStderr) : pipe(_pipe), isStderr(_isStderr) {}
  1251. virtual size32_t read(size32_t sz, void * data)
  1252. {
  1253. if (isStderr)
  1254. return pipe->readError(sz, data);
  1255. else
  1256. return pipe->read(sz, data);
  1257. }
  1258. private:
  1259. Owned<IPipeProcess> pipe;
  1260. bool isStderr;
  1261. };
  1262. #ifdef _WIN32
  1263. class CWindowsPipeProcess: implements IPipeProcess, public CInterface
  1264. {
  1265. HANDLE pipeProcess;
  1266. HANDLE hInput;
  1267. HANDLE hOutput;
  1268. HANDLE hError;
  1269. StringAttr title;
  1270. unsigned retcode;
  1271. CriticalSection sect;
  1272. bool aborted;
  1273. StringAttr allowedprogs;
  1274. StringArray envVars;
  1275. StringArray envValues;
  1276. public:
  1277. IMPLEMENT_IINTERFACE;
  1278. CWindowsPipeProcess(const char *_allowedprogs)
  1279. : allowedprogs(_allowedprogs)
  1280. {
  1281. pipeProcess = (HANDLE)-1;
  1282. hInput=(HANDLE)-1;
  1283. hOutput=(HANDLE)-1;
  1284. hError=(HANDLE)-1;
  1285. retcode = (unsigned)-1;
  1286. aborted = false;
  1287. }
  1288. ~CWindowsPipeProcess()
  1289. {
  1290. kill();
  1291. }
  1292. void kill()
  1293. {
  1294. doCloseInput();
  1295. doCloseOutput();
  1296. doCloseError();
  1297. if (pipeProcess != (HANDLE)-1) {
  1298. CloseHandle(pipeProcess);
  1299. pipeProcess = (HANDLE)-1;
  1300. }
  1301. }
  1302. bool run(const char *_title,const char *prog,const char *dir,bool hasinput,bool hasoutput,bool haserror, size32_t stderrbufsize,bool newProcessGroup)
  1303. {
  1304. // size32_t stderrbufsize ignored as not required (I think)
  1305. CriticalBlock block(sect);
  1306. kill();
  1307. title.clear();
  1308. if (_title) {
  1309. title.set(_title);
  1310. PROGLOG("%s: Creating PIPE process : %s", title.get(), prog);
  1311. }
  1312. CheckAllowedProgram(prog,allowedprogs);
  1313. SECURITY_ATTRIBUTES sa;
  1314. sa.nLength = sizeof(SECURITY_ATTRIBUTES);
  1315. sa.bInheritHandle = TRUE;
  1316. sa.lpSecurityDescriptor = NULL;
  1317. HANDLE hProgOutput=(HANDLE)-1;
  1318. HANDLE hProgInput=(HANDLE)-1;
  1319. HANDLE hProgError=(HANDLE)-1;
  1320. HANDLE h;
  1321. //NB: Create a pipe handles that are not inherited our end
  1322. if (hasinput) {
  1323. CreatePipe(&hProgInput,&h,&sa,0);
  1324. DuplicateHandle(GetCurrentProcess(),h, GetCurrentProcess(), &hInput, 0, FALSE, DUPLICATE_SAME_ACCESS);
  1325. CloseHandle(h);
  1326. }
  1327. if (hasoutput) {
  1328. CreatePipe(&h,&hProgOutput,&sa,0);
  1329. DuplicateHandle(GetCurrentProcess(),h, GetCurrentProcess(), &hOutput, 0, FALSE, DUPLICATE_SAME_ACCESS);
  1330. CloseHandle(h);
  1331. }
  1332. if (haserror) {
  1333. CreatePipe(&h,&hProgError,&sa,0);
  1334. DuplicateHandle(GetCurrentProcess(),h, GetCurrentProcess(), &hError, 0, FALSE, DUPLICATE_SAME_ACCESS);
  1335. CloseHandle(h);
  1336. }
  1337. STARTUPINFO StartupInfo;
  1338. _clear(StartupInfo);
  1339. StartupInfo.cb = sizeof(StartupInfo);
  1340. StartupInfo.wShowWindow = SW_HIDE;
  1341. StartupInfo.dwFlags = STARTF_USESTDHANDLES|STARTF_USESHOWWINDOW ;
  1342. StartupInfo.hStdOutput = hasoutput?hProgOutput:GetStdHandle(STD_OUTPUT_HANDLE);
  1343. StartupInfo.hStdError = haserror?hProgError:GetStdHandle(STD_ERROR_HANDLE);
  1344. StartupInfo.hStdInput = hasinput?hProgInput:GetStdHandle(STD_INPUT_HANDLE);
  1345. PROCESS_INFORMATION ProcessInformation;
  1346. // MORE - should create a new environment block that is copy of parent's, then set all the values in envVars/envValues, and pass it
  1347. if (!CreateProcess(NULL, (char *)prog, NULL,NULL,TRUE,0,NULL, dir&&*dir?dir:NULL, &StartupInfo,&ProcessInformation)) {
  1348. if (_title) {
  1349. StringBuffer errstr;
  1350. formatSystemError(errstr, GetLastError());
  1351. OERRLOG("%s: PIPE process '%s' failed: %s", title.get(), prog, errstr.str());
  1352. }
  1353. return false;
  1354. }
  1355. pipeProcess = ProcessInformation.hProcess;
  1356. CloseHandle(ProcessInformation.hThread);
  1357. if (hasoutput)
  1358. CloseHandle(hProgOutput);
  1359. if (hasinput)
  1360. CloseHandle(hProgInput);
  1361. if (haserror)
  1362. CloseHandle(hProgError);
  1363. return true;
  1364. }
  1365. virtual void setenv(const char *var, const char *value)
  1366. {
  1367. assertex(var);
  1368. if (!value)
  1369. value = "";
  1370. envVars.append(var);
  1371. envValues.append(value);
  1372. }
  1373. size32_t read(size32_t sz, void *buf)
  1374. {
  1375. DWORD sizeRead;
  1376. if (!ReadFile(hOutput, buf, sz, &sizeRead, NULL)) {
  1377. //raise error here
  1378. if(aborted)
  1379. return 0;
  1380. int err=GetLastError();
  1381. switch(err)
  1382. {
  1383. case ERROR_HANDLE_EOF:
  1384. case ERROR_BROKEN_PIPE:
  1385. case ERROR_NO_DATA:
  1386. return 0;
  1387. default:
  1388. aborted = true;
  1389. IException *e = makeOsExceptionV(err, "Pipe: ReadFile failed (size %d)", sz);
  1390. PrintExceptionLog(e, NULL);
  1391. throw e;
  1392. }
  1393. }
  1394. return aborted?((size32_t)-1):((size32_t)sizeRead);
  1395. }
  1396. ISimpleReadStream *getOutputStream()
  1397. {
  1398. return new CSimplePipeStream(LINK(this), false);
  1399. }
  1400. size32_t readError(size32_t sz, void *buf)
  1401. {
  1402. DWORD sizeRead;
  1403. if (!ReadFile(hError, buf, sz, &sizeRead, NULL)) {
  1404. //raise error here
  1405. if(aborted)
  1406. return 0;
  1407. int err=GetLastError();
  1408. switch(err)
  1409. {
  1410. case ERROR_HANDLE_EOF:
  1411. case ERROR_BROKEN_PIPE:
  1412. case ERROR_NO_DATA:
  1413. return 0;
  1414. default:
  1415. aborted = true;
  1416. IException *e = makeOsExceptionV(err, "Pipe: ReadError failed (size %d)", sz);
  1417. PrintExceptionLog(e, NULL);
  1418. throw e;
  1419. }
  1420. }
  1421. return aborted?((size32_t)-1):((size32_t)sizeRead);
  1422. }
  1423. ISimpleReadStream *getErrorStream()
  1424. {
  1425. return new CSimplePipeStream(LINK(this), true);
  1426. }
  1427. size32_t write(size32_t sz, const void *buf)
  1428. {
  1429. DWORD sizeWritten;
  1430. if (!WriteFile(hInput, buf, sz, &sizeWritten, NULL)) {
  1431. int err=GetLastError();
  1432. if ((err==ERROR_HANDLE_EOF)||aborted)
  1433. sizeWritten = 0;
  1434. else {
  1435. IException *e = makeOsExceptionV(err, "Pipe: WriteFile failed (size %d)", sz);
  1436. PrintExceptionLog(e, NULL);
  1437. throw e;
  1438. }
  1439. }
  1440. return aborted?((size32_t)-1):((size32_t)sizeWritten);
  1441. }
  1442. unsigned wait()
  1443. {
  1444. CriticalBlock block(sect);
  1445. if (pipeProcess != (HANDLE)-1) {
  1446. if (title.length())
  1447. PROGLOG("%s: Pipe: Waiting for process to complete %d",title.get(),(unsigned)pipeProcess);
  1448. {
  1449. CriticalUnblock unblock(sect);
  1450. WaitForSingleObject(pipeProcess, INFINITE);
  1451. }
  1452. if (pipeProcess != (HANDLE)-1) {
  1453. GetExitCodeProcess(pipeProcess,(LPDWORD)&retcode); // already got if notified
  1454. CloseHandle(pipeProcess);
  1455. pipeProcess = (HANDLE)-1;
  1456. }
  1457. if (title.length())
  1458. PROGLOG("%s: Pipe: process complete",title.get());
  1459. }
  1460. return retcode;
  1461. }
  1462. unsigned wait(unsigned timeoutms, bool &timedout)
  1463. {
  1464. CriticalBlock block(sect);
  1465. timedout = false;
  1466. if (pipeProcess != (HANDLE)-1) {
  1467. if (title.length())
  1468. PROGLOG("%s: Pipe: Waiting for process to complete %d",title.get(),(unsigned)pipeProcess);
  1469. {
  1470. CriticalUnblock unblock(sect);
  1471. if (WaitForSingleObject(pipeProcess, timeoutms)!=WAIT_OBJECT_0) {
  1472. timedout = true;
  1473. return retcode;
  1474. }
  1475. }
  1476. if (pipeProcess != (HANDLE)-1) {
  1477. GetExitCodeProcess(pipeProcess,(LPDWORD)&retcode); // already got if notified
  1478. CloseHandle(pipeProcess);
  1479. pipeProcess = (HANDLE)-1;
  1480. }
  1481. if (title.length())
  1482. PROGLOG("%s: Pipe: process complete",title.get());
  1483. }
  1484. return retcode;
  1485. }
  1486. void notifyTerminated(HANDLE pid,unsigned _retcode)
  1487. {
  1488. CriticalBlock block(sect);
  1489. if ((pid!=(HANDLE)-1)&&(pid==pipeProcess)) {
  1490. retcode = _retcode;
  1491. pipeProcess = (HANDLE)-1;
  1492. }
  1493. }
  1494. void doCloseInput()
  1495. {
  1496. CriticalBlock block(sect);
  1497. if (hInput != (HANDLE)-1) {
  1498. CloseHandle(hInput);
  1499. hInput = (HANDLE)-1;
  1500. }
  1501. }
  1502. void doCloseOutput()
  1503. {
  1504. CriticalBlock block(sect);
  1505. if (hOutput != (HANDLE)-1) {
  1506. CloseHandle(hOutput);
  1507. hOutput = (HANDLE)-1;
  1508. }
  1509. }
  1510. void doCloseError()
  1511. {
  1512. CriticalBlock block(sect);
  1513. if (hError != (HANDLE)-1) {
  1514. CloseHandle(hError);
  1515. hError = (HANDLE)-1;
  1516. }
  1517. }
  1518. void closeInput()
  1519. {
  1520. doCloseInput();
  1521. }
  1522. void closeOutput()
  1523. {
  1524. doCloseOutput();
  1525. }
  1526. void closeError()
  1527. {
  1528. doCloseError();
  1529. }
  1530. void abort()
  1531. {
  1532. CriticalBlock block(sect);
  1533. if (pipeProcess != (HANDLE)-1) {
  1534. if (title.length())
  1535. PROGLOG("%s: Pipe Aborting",title.get());
  1536. aborted = true;
  1537. //doCloseOutput(); // seems to work better without this
  1538. doCloseInput();
  1539. {
  1540. CriticalUnblock unblock(sect);
  1541. Sleep(100);
  1542. }
  1543. try { // this code is problematic for some reason
  1544. if (pipeProcess != (HANDLE)-1) {
  1545. TerminateProcess(pipeProcess, 255);
  1546. CloseHandle(pipeProcess);
  1547. pipeProcess = (HANDLE)-1;
  1548. }
  1549. }
  1550. catch (...) {
  1551. // ignore errors
  1552. }
  1553. if (title.length())
  1554. PROGLOG("%s: Pipe Aborted",title.get());
  1555. }
  1556. }
  1557. bool hasInput()
  1558. {
  1559. return hInput!=(HANDLE)-1;
  1560. }
  1561. bool hasOutput()
  1562. {
  1563. return hOutput!=(HANDLE)-1;
  1564. }
  1565. bool hasError()
  1566. {
  1567. return hError!=(HANDLE)-1;
  1568. }
  1569. HANDLE getProcessHandle()
  1570. {
  1571. return pipeProcess;
  1572. }
  1573. };
  1574. IPipeProcess *createPipeProcess(const char *allowedprogs)
  1575. {
  1576. return new CWindowsPipeProcess(allowedprogs);
  1577. }
  1578. #else
  1579. class CIgnoreSIGPIPE
  1580. {
  1581. public:
  1582. CIgnoreSIGPIPE()
  1583. {
  1584. oact.sa_handler = SIG_IGN;
  1585. struct sigaction act;
  1586. sigset_t blockset;
  1587. sigemptyset(&blockset);
  1588. act.sa_mask = blockset;
  1589. act.sa_handler = SIG_IGN;
  1590. act.sa_flags = 0;
  1591. sigaction(SIGPIPE, &act, &oact);
  1592. }
  1593. ~CIgnoreSIGPIPE()
  1594. {
  1595. if (oact.sa_handler != SIG_IGN)
  1596. sigaction(SIGPIPE, &oact, NULL);
  1597. }
  1598. private:
  1599. struct sigaction oact;
  1600. };
  1601. #define WHITESPACE " \t\n\r"
  1602. static unsigned dowaitpid(HANDLE pid, int mode)
  1603. {
  1604. while (pid != (HANDLE)-1) {
  1605. int stat=-1;
  1606. int ret = waitpid(pid, &stat, mode);
  1607. if (ret>0)
  1608. {
  1609. if (WIFEXITED(stat))
  1610. return WEXITSTATUS(stat);
  1611. else if (WIFSIGNALED(stat))
  1612. {
  1613. OERRLOG("Program was terminated by signal %u", (unsigned) WTERMSIG(stat));
  1614. if (WTERMSIG(stat)==SIGPIPE)
  1615. return 0;
  1616. return 254;
  1617. }
  1618. else
  1619. {
  1620. return 254;
  1621. }
  1622. }
  1623. if (ret==0)
  1624. break;
  1625. int err = errno;
  1626. if (err == ECHILD)
  1627. break;
  1628. if (err!=EINTR) {
  1629. OERRLOG("dowait failed with errcode %d",err);
  1630. return (unsigned)-1;
  1631. }
  1632. }
  1633. return 0;
  1634. }
  1635. static CriticalSection runsect; // single thread process start to avoid forked handle open/closes interleaving
  1636. class CLinuxPipeProcess: implements IPipeProcess, public CInterface
  1637. {
  1638. class cForkThread: public Thread
  1639. {
  1640. CLinuxPipeProcess *parent;
  1641. public:
  1642. cForkThread(CLinuxPipeProcess *_parent)
  1643. {
  1644. parent = _parent;
  1645. }
  1646. int run()
  1647. {
  1648. parent->run();
  1649. return 0;
  1650. }
  1651. };
  1652. Owned<cForkThread> forkthread;
  1653. class cStdErrorBufferThread: public Thread
  1654. {
  1655. MemoryAttr buf;
  1656. size32_t bufsize;
  1657. Semaphore stopsem;
  1658. CriticalSection &sect;
  1659. int &hError;
  1660. public:
  1661. cStdErrorBufferThread(size32_t maxbufsize,int &_hError,CriticalSection &_sect)
  1662. : sect(_sect), hError(_hError)
  1663. {
  1664. buf.allocate(maxbufsize);
  1665. bufsize = 0;
  1666. }
  1667. int run()
  1668. {
  1669. while (!stopsem.wait(1000)) {
  1670. CriticalBlock block(sect);
  1671. if (hError!=(HANDLE)-1) { // hmm who did that
  1672. fcntl(hError,F_SETFL,O_NONBLOCK); // make sure non-blocking
  1673. if (bufsize<buf.length()) {
  1674. size32_t sizeRead = (size32_t)::read(hError, (byte *)buf.bufferBase()+bufsize, buf.length()-bufsize);
  1675. if ((int)sizeRead>0) {
  1676. bufsize += sizeRead;
  1677. }
  1678. }
  1679. else { // flush (to avoid process blocking)
  1680. byte tmp[1024];
  1681. size32_t totsz = 0;
  1682. for (unsigned i=0;i<1024;i++) {
  1683. size32_t sz = (size32_t)::read(hError, tmp, sizeof(tmp));
  1684. if ((int)sz<=0)
  1685. break;
  1686. totsz+=sz;
  1687. }
  1688. if (totsz)
  1689. IWARNLOG("Lost %d bytes of stderr output",totsz);
  1690. }
  1691. }
  1692. }
  1693. if (hError!=(HANDLE)-1) { // hmm who did that
  1694. fcntl(hError,F_SETFL,0); // read any remaining data in blocking mode
  1695. while (bufsize<buf.length()) {
  1696. size32_t sizeRead = (size32_t)::read(hError, (byte *)buf.bufferBase()+bufsize, buf.length()-bufsize);
  1697. if ((int)sizeRead>0)
  1698. bufsize += sizeRead;
  1699. else
  1700. break;
  1701. }
  1702. }
  1703. return 0;
  1704. }
  1705. void stop()
  1706. {
  1707. stopsem.signal();
  1708. Thread::join();
  1709. }
  1710. size32_t read(size32_t sz,void *out)
  1711. {
  1712. CriticalBlock block(sect);
  1713. if (bufsize<sz)
  1714. sz = bufsize;
  1715. if (sz>0) {
  1716. memcpy(out,buf.bufferBase(),sz);
  1717. if (sz!=bufsize) {
  1718. bufsize -= sz;
  1719. memmove(buf.bufferBase(),(byte *)buf.bufferBase()+sz,bufsize); // not ideal but hopefully not large
  1720. }
  1721. else
  1722. bufsize = 0;
  1723. }
  1724. return sz;
  1725. }
  1726. } *stderrbufferthread;
  1727. protected: friend class PipeWriterThread;
  1728. HANDLE pipeProcess;
  1729. HANDLE hInput;
  1730. HANDLE hOutput;
  1731. HANDLE hError;
  1732. bool hasinput;
  1733. bool hasoutput;
  1734. bool haserror;
  1735. bool newProcessGroup;
  1736. StringAttr title;
  1737. StringAttr cmd;
  1738. StringAttr prog;
  1739. StringAttr dir;
  1740. int retcode;
  1741. CriticalSection sect;
  1742. Semaphore started;
  1743. bool aborted;
  1744. MemoryBuffer stderrbuf;
  1745. size32_t stderrbufsize;
  1746. StringAttr allowedprogs;
  1747. StringArray envVars;
  1748. StringArray envValues;
  1749. void clearUtilityThreads(bool clearStderr)
  1750. {
  1751. Owned<cForkThread> ft;
  1752. cStdErrorBufferThread *et;
  1753. {
  1754. CriticalBlock block(sect); // clear forkthread and optionally stderrbufferthread
  1755. ft.setown(forkthread.getClear());
  1756. et = stderrbufferthread;
  1757. if (clearStderr)
  1758. stderrbufferthread = nullptr;
  1759. }
  1760. if (ft)
  1761. {
  1762. ft->join();
  1763. ft.clear();
  1764. }
  1765. if (et)
  1766. {
  1767. et->stop();
  1768. if (clearStderr)
  1769. delete et;
  1770. }
  1771. }
  1772. public:
  1773. IMPLEMENT_IINTERFACE;
  1774. CLinuxPipeProcess(const char *_allowedprogs)
  1775. : allowedprogs(_allowedprogs)
  1776. {
  1777. pipeProcess = (HANDLE)-1;
  1778. hInput=(HANDLE)-1;
  1779. hOutput=(HANDLE)-1;
  1780. hError=(HANDLE)-1;
  1781. retcode = -1;
  1782. aborted = false;
  1783. stderrbufferthread = NULL;
  1784. newProcessGroup = false;
  1785. }
  1786. ~CLinuxPipeProcess()
  1787. {
  1788. kill();
  1789. }
  1790. void kill()
  1791. {
  1792. closeInput();
  1793. closeOutput();
  1794. closeError();
  1795. clearUtilityThreads(true);
  1796. }
  1797. char **splitargs(const char *line,unsigned &argc)
  1798. {
  1799. char *buf = strdup(line);
  1800. // first count params (this probably could be improved)
  1801. char *s = buf;
  1802. argc = 0;
  1803. while (readarg(s))
  1804. argc++;
  1805. free(buf);
  1806. size32_t l = strlen(line)+1;
  1807. size32_t al = (argc+1)*sizeof(char *);
  1808. char **argv = (char **)malloc(al+l);
  1809. argv[argc] = NULL;
  1810. s = ((char *)argv)+al;
  1811. memcpy(s,line,l);
  1812. for (unsigned i=0;i<argc;i++)
  1813. argv[i] = readarg(s);
  1814. return argv;
  1815. }
  1816. void run()
  1817. {
  1818. int inpipe[2];
  1819. int outpipe[2];
  1820. int errpipe[2];
  1821. if ((hasinput && (::pipe(inpipe)==-1)) ||
  1822. (hasoutput && (::pipe(outpipe)==-1)) ||
  1823. (haserror && (::pipe(errpipe)==-1)))
  1824. {
  1825. retcode = START_FAILURE;
  1826. started.signal();
  1827. throw makeOsException(errno);
  1828. }
  1829. /* NB: Important to call splitargs (which calls malloc) before the fork()
  1830. * and not in the child process. Because performing malloc in the child
  1831. * process, which then calls exec() can cause problems for TBB malloc proxy.
  1832. */
  1833. unsigned argc;
  1834. char **argv=splitargs(prog,argc);
  1835. for (;;)
  1836. {
  1837. pipeProcess = (HANDLE)fork();
  1838. if (pipeProcess!=(HANDLE)-1)
  1839. break;
  1840. if (errno!=EAGAIN) {
  1841. if (hasinput) {
  1842. close(inpipe[0]);
  1843. close(inpipe[1]);
  1844. }
  1845. if (hasoutput) {
  1846. close(outpipe[0]);
  1847. close(outpipe[1]);
  1848. }
  1849. if (haserror) {
  1850. close(errpipe[0]);
  1851. close(errpipe[1]);
  1852. }
  1853. retcode = START_FAILURE;
  1854. started.signal();
  1855. free(argv);
  1856. return;
  1857. }
  1858. }
  1859. if (pipeProcess==0) { // child
  1860. if (newProcessGroup)//Force the child process into its own process group, so we can terminate it and its children.
  1861. setpgid(0,0);
  1862. if (hasinput) {
  1863. dup2(inpipe[0],0);
  1864. close(inpipe[0]);
  1865. close(inpipe[1]);
  1866. }
  1867. if (hasoutput) {
  1868. dup2(outpipe[1],1);
  1869. close(outpipe[0]);
  1870. close(outpipe[1]);
  1871. }
  1872. if (haserror) {
  1873. dup2(errpipe[1],2);
  1874. close(errpipe[0]);
  1875. close(errpipe[1]);
  1876. }
  1877. if (dir.get()) {
  1878. if (chdir(dir) == -1)
  1879. throw MakeStringException(-1, "CLinuxPipeProcess::run: could not change dir to %s", dir.get());
  1880. }
  1881. ForEachItemIn(idx, envVars)
  1882. {
  1883. ::setenv(envVars.item(idx), envValues.item(idx), 1);
  1884. }
  1885. execvp(argv[0],argv);
  1886. if (haserror)
  1887. {
  1888. Owned<IException> e = createPipeErrnoExceptionV(errno, "exec failed: %s", prog.get());
  1889. StringBuffer eStr;
  1890. fprintf(stderr, "ERROR: %d: %s", e->errorCode(), e->errorMessage(eStr).str());
  1891. fflush(stderr);
  1892. }
  1893. _exit(START_FAILURE); // must be _exit!!
  1894. }
  1895. free(argv);
  1896. if (hasinput)
  1897. close(inpipe[0]);
  1898. if (hasoutput)
  1899. close(outpipe[1]);
  1900. if (haserror)
  1901. close(errpipe[1]);
  1902. hInput = hasinput?inpipe[1]:((HANDLE)-1);
  1903. hOutput = hasoutput?outpipe[0]:((HANDLE)-1);
  1904. hError = haserror?errpipe[0]:((HANDLE)-1);
  1905. started.signal();
  1906. retcode = dowaitpid(pipeProcess, 0);
  1907. if (retcode==START_FAILURE)
  1908. closeOutput();
  1909. }
  1910. bool run(const char *_title,const char *_prog,const char *_dir,bool _hasinput,bool _hasoutput, bool _haserror, size32_t stderrbufsize, bool _newProcessGroup)
  1911. {
  1912. CriticalBlock runblock(runsect);
  1913. kill();
  1914. CriticalBlock block(sect);
  1915. hasinput = _hasinput;
  1916. hasoutput = _hasoutput;
  1917. haserror = _haserror;
  1918. newProcessGroup = _newProcessGroup;
  1919. title.clear();
  1920. prog.set(_prog);
  1921. dir.set(_dir);
  1922. if (_title)
  1923. {
  1924. title.set(_title);
  1925. PROGLOG("%s: Creating PIPE program process : '%s' - hasinput=%d, hasoutput=%d stderrbufsize=%d", title.get(), prog.get(),(int)hasinput, (int)hasoutput, stderrbufsize);
  1926. }
  1927. CheckAllowedProgram(prog,allowedprogs);
  1928. retcode = 0;
  1929. if (forkthread)
  1930. {
  1931. {
  1932. CriticalUnblock unblock(sect);
  1933. forkthread->join();
  1934. }
  1935. forkthread.clear();
  1936. }
  1937. forkthread.setown(new cForkThread(this));
  1938. forkthread->start();
  1939. bool joined = false;
  1940. {
  1941. CriticalUnblock unblock(sect);
  1942. started.wait();
  1943. joined = forkthread->join(50); // give a chance to fail
  1944. }
  1945. // only check retcode if we were able to join
  1946. if ( (joined) && (retcode==START_FAILURE) )
  1947. {
  1948. DBGLOG("%s: PIPE process '%s' failed to start", title.get()?title.get():"CLinuxPipeProcess", prog.get());
  1949. forkthread.clear();
  1950. return false;
  1951. }
  1952. if (stderrbufsize)
  1953. {
  1954. if (stderrbufferthread)
  1955. {
  1956. stderrbufferthread->stop();
  1957. delete stderrbufferthread;
  1958. }
  1959. stderrbufferthread = new cStdErrorBufferThread(stderrbufsize,hError,sect);
  1960. stderrbufferthread->start();
  1961. }
  1962. return true;
  1963. }
  1964. virtual void setenv(const char *var, const char *value)
  1965. {
  1966. assertex(var);
  1967. if (!value)
  1968. value = "";
  1969. envVars.append(var);
  1970. envValues.append(value);
  1971. }
  1972. size32_t read(size32_t sz, void *buf)
  1973. {
  1974. CriticalBlock block(sect);
  1975. if (aborted)
  1976. return (size32_t)-1;
  1977. if (hOutput==(HANDLE)-1)
  1978. return 0;
  1979. size32_t sizeRead;
  1980. for (;;) {
  1981. {
  1982. CriticalUnblock unblock(sect);
  1983. sizeRead = (size32_t)::read(hOutput, buf, sz);
  1984. }
  1985. if (sizeRead!=(size32_t)-1)
  1986. break;
  1987. if (aborted)
  1988. break;
  1989. if (errno!=EINTR) {
  1990. aborted = true;
  1991. throw createPipeErrnoExceptionV(errno,"Pipe: read failed (size %d)", sz);
  1992. }
  1993. }
  1994. return aborted?((size32_t)-1):((size32_t)sizeRead);
  1995. }
  1996. ISimpleReadStream *getOutputStream()
  1997. {
  1998. return new CSimplePipeStream(LINK(this), false);
  1999. }
  2000. size32_t write(size32_t sz, const void *buf)
  2001. {
  2002. CriticalBlock block(sect);
  2003. CIgnoreSIGPIPE ignoresigpipe;
  2004. if (aborted)
  2005. return (size32_t)-1;
  2006. if (hInput==(HANDLE)-1)
  2007. return 0;
  2008. size32_t sizeWritten;
  2009. for (;;) {
  2010. {
  2011. CriticalUnblock unblock(sect);
  2012. sizeWritten = (size32_t)::write(hInput, buf, sz);
  2013. }
  2014. if (sizeWritten!=(size32_t)-1)
  2015. break;
  2016. if (aborted)
  2017. break;
  2018. if (errno!=EINTR) {
  2019. throw createPipeErrnoExceptionV(errno, "Pipe: write failed (size %d)", sz);
  2020. }
  2021. }
  2022. return aborted?((size32_t)-1):((size32_t)sizeWritten);
  2023. }
  2024. size32_t readError(size32_t sz, void *buf)
  2025. {
  2026. CriticalBlock block(sect);
  2027. if (stderrbufferthread)
  2028. return stderrbufferthread->read(sz,buf);
  2029. if (aborted)
  2030. return (size32_t)-1;
  2031. if (hError==(HANDLE)-1)
  2032. return 0;
  2033. size32_t sizeRead;
  2034. for (;;) {
  2035. {
  2036. CriticalUnblock unblock(sect);
  2037. sizeRead = (size32_t)::read(hError, buf, sz);
  2038. }
  2039. if (sizeRead!=(size32_t)-1)
  2040. break;
  2041. if (aborted)
  2042. break;
  2043. if (errno!=EINTR) {
  2044. aborted = true;
  2045. throw createPipeErrnoExceptionV(errno, "Pipe: readError failed (size %d)", sz);
  2046. }
  2047. }
  2048. return aborted?((size32_t)-1):((size32_t)sizeRead);
  2049. }
  2050. ISimpleReadStream *getErrorStream()
  2051. {
  2052. return new CSimplePipeStream(LINK(this), true);
  2053. }
  2054. void notifyTerminated(HANDLE pid,unsigned _retcode)
  2055. {
  2056. CriticalBlock block(sect);
  2057. if (((int)pid>0)&&(pid==pipeProcess)) {
  2058. retcode = _retcode;
  2059. pipeProcess = (HANDLE)-1;
  2060. }
  2061. }
  2062. unsigned wait()
  2063. {
  2064. bool timedout;
  2065. return wait(INFINITE, timedout);
  2066. }
  2067. unsigned wait(unsigned timeoutms, bool &timedout)
  2068. {
  2069. timedout = false;
  2070. if (INFINITE != timeoutms)
  2071. {
  2072. CriticalBlock block(sect);
  2073. if (forkthread)
  2074. {
  2075. {
  2076. CriticalUnblock unblock(sect);
  2077. if (!forkthread->join(timeoutms))
  2078. {
  2079. timedout = true;
  2080. return retcode;
  2081. }
  2082. }
  2083. }
  2084. }
  2085. // NOTE - we don't clear stderrbufferthread here, since we want to be able to still read the buffered data
  2086. clearUtilityThreads(false); // NB: will recall forkthread->join(), but doesn't matter
  2087. if (pipeProcess != (HANDLE)-1)
  2088. {
  2089. if (title.length())
  2090. PROGLOG("%s: Pipe: process %d complete %d", title.get(), pipeProcess, retcode);
  2091. pipeProcess = (HANDLE)-1;
  2092. }
  2093. return retcode;
  2094. }
  2095. void closeOutput()
  2096. {
  2097. CriticalBlock block(sect);
  2098. if (hOutput != (HANDLE)-1) {
  2099. ::close(hOutput);
  2100. hOutput = (HANDLE)-1;
  2101. }
  2102. }
  2103. void closeInput()
  2104. {
  2105. CriticalBlock block(sect);
  2106. if (hInput != (HANDLE)-1) {
  2107. ::close(hInput);
  2108. hInput = (HANDLE)-1;
  2109. }
  2110. }
  2111. void closeError()
  2112. {
  2113. CriticalBlock block(sect);
  2114. if (hError != (HANDLE)-1) {
  2115. ::close(hError);
  2116. hError = (HANDLE)-1;
  2117. }
  2118. }
  2119. void abort()
  2120. {
  2121. CriticalBlock block(sect);
  2122. if (pipeProcess != (HANDLE)-1) {
  2123. if (title.length())
  2124. PROGLOG("%s: Pipe Aborting",title.get());
  2125. aborted = true;
  2126. closeInput();
  2127. {
  2128. CriticalUnblock unblock(sect);
  2129. forkthread->join(1000);
  2130. }
  2131. if (pipeProcess != (HANDLE)-1) {
  2132. if (title.length())
  2133. PROGLOG("%s: Forcibly killing pipe process %d",title.get(),pipeProcess);
  2134. if (newProcessGroup)
  2135. ::kill(-pipeProcess,SIGKILL);
  2136. else
  2137. ::kill(pipeProcess,SIGKILL); // if this doesn't kill it we are in trouble
  2138. CriticalUnblock unblock(sect);
  2139. wait();
  2140. }
  2141. if (title.length())
  2142. PROGLOG("%s: Pipe Aborted",title.get());
  2143. retcode = -1;
  2144. forkthread.clear();
  2145. }
  2146. }
  2147. bool hasInput()
  2148. {
  2149. CriticalBlock block(sect);
  2150. return hInput!=(HANDLE)-1;
  2151. }
  2152. bool hasOutput()
  2153. {
  2154. CriticalBlock block(sect);
  2155. return hOutput!=(HANDLE)-1;
  2156. }
  2157. bool hasError()
  2158. {
  2159. CriticalBlock block(sect);
  2160. return hError!=(HANDLE)-1;
  2161. }
  2162. HANDLE getProcessHandle()
  2163. {
  2164. CriticalBlock block(sect);
  2165. return pipeProcess;
  2166. }
  2167. };
  2168. IPipeProcess *createPipeProcess(const char *allowedprogs)
  2169. {
  2170. return new CLinuxPipeProcess(allowedprogs);
  2171. }
  2172. #endif
  2173. // Worker thread
  2174. class CWorkQueueThread: implements IWorkQueueThread, public CInterface
  2175. {
  2176. public:
  2177. IMPLEMENT_IINTERFACE;
  2178. CriticalSection crit;
  2179. unsigned persisttime;
  2180. class cWorkerThread: public Thread
  2181. {
  2182. unsigned persisttime;
  2183. CWorkQueueThread *parent;
  2184. CriticalSection &crit;
  2185. public:
  2186. cWorkerThread(CWorkQueueThread *_parent,CriticalSection &_crit,unsigned _persisttime)
  2187. : crit(_crit)
  2188. {
  2189. parent = _parent;
  2190. persisttime = _persisttime;
  2191. }
  2192. QueueOf<IWorkQueueItem,false> queue;
  2193. Semaphore sem;
  2194. int run()
  2195. {
  2196. for (;;) {
  2197. IWorkQueueItem * work;
  2198. bool wr = sem.wait(persisttime);
  2199. {
  2200. CriticalBlock block(crit);
  2201. if (!wr) {
  2202. wr = sem.wait(0); // catch race
  2203. if (!wr)
  2204. break; // timed out
  2205. }
  2206. work = queue.dequeue();
  2207. }
  2208. if (!work)
  2209. break;
  2210. try {
  2211. work->execute();
  2212. work->Release();
  2213. }
  2214. catch (IException *e)
  2215. {
  2216. EXCLOG(e,"CWorkQueueThread item execute");
  2217. e->Release();
  2218. }
  2219. }
  2220. CriticalBlock block(crit);
  2221. parent->worker=NULL; // this should be safe
  2222. return 0;
  2223. }
  2224. } *worker;
  2225. CWorkQueueThread(unsigned _persisttime)
  2226. {
  2227. persisttime = _persisttime;
  2228. worker = NULL;
  2229. }
  2230. ~CWorkQueueThread()
  2231. {
  2232. wait();
  2233. }
  2234. void post(IWorkQueueItem *packet)
  2235. {
  2236. CriticalBlock block(crit);
  2237. if (!worker) {
  2238. worker = new cWorkerThread(this,crit,persisttime);
  2239. worker->startRelease();
  2240. }
  2241. worker->queue.enqueue(packet);
  2242. worker->sem.signal();
  2243. }
  2244. void wait()
  2245. {
  2246. CriticalBlock block(crit);
  2247. if (worker) {
  2248. worker->queue.enqueue(NULL);
  2249. worker->sem.signal();
  2250. Linked<cWorkerThread> wt;
  2251. wt.set(worker);
  2252. CriticalUnblock unblock(crit);
  2253. wt->join();
  2254. }
  2255. }
  2256. unsigned pending()
  2257. {
  2258. CriticalBlock block(crit);
  2259. unsigned ret = 0;
  2260. if (worker)
  2261. ret = worker->queue.ordinality();
  2262. return ret;
  2263. }
  2264. };
  2265. IWorkQueueThread *createWorkQueueThread(unsigned persisttime)
  2266. {
  2267. return new CWorkQueueThread(persisttime);
  2268. }
  2269. unsigned threadLogID() // for use in logging
  2270. {
  2271. #if defined(__APPLE__)
  2272. return pthread_mach_thread_np(pthread_self());
  2273. #elif !defined(_WIN32)
  2274. #ifdef SYS_gettid
  2275. return (unsigned) (memsize_t) syscall(SYS_gettid);
  2276. #endif
  2277. #endif
  2278. return (unsigned)(memsize_t) GetCurrentThreadId(); // truncated in 64bit
  2279. }