jthread.cpp 62 KB

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