jthread.cpp 67 KB

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