condition.d 27 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939
  1. /**
  2. * The condition module provides a primitive for synchronized condition
  3. * checking.
  4. *
  5. * Copyright: Copyright Sean Kelly 2005 - 2009.
  6. * License: $(LINK2 http://www.boost.org/LICENSE_1_0.txt, Boost License 1.0)
  7. * Authors: Sean Kelly
  8. * Source: $(DRUNTIMESRC core/sync/_condition.d)
  9. */
  10. /* Copyright Sean Kelly 2005 - 2009.
  11. * Distributed under the Boost Software License, Version 1.0.
  12. * (See accompanying file LICENSE or copy at
  13. * http://www.boost.org/LICENSE_1_0.txt)
  14. */
  15. module core.sync.condition;
  16. public import core.sync.exception;
  17. public import core.sync.mutex;
  18. public import core.time;
  19. version (Windows)
  20. {
  21. import core.sync.semaphore;
  22. import core.sys.windows.basetsd /+: HANDLE+/;
  23. import core.sys.windows.winbase /+: CloseHandle, CreateSemaphoreA, CRITICAL_SECTION,
  24. DeleteCriticalSection, EnterCriticalSection, INFINITE, InitializeCriticalSection,
  25. LeaveCriticalSection, ReleaseSemaphore, WAIT_OBJECT_0, WaitForSingleObject+/;
  26. import core.sys.windows.windef /+: BOOL, DWORD+/;
  27. import core.sys.windows.winerror /+: WAIT_TIMEOUT+/;
  28. }
  29. else version (Posix)
  30. {
  31. import core.sync.config;
  32. import core.stdc.errno;
  33. import core.sys.posix.pthread;
  34. import core.sys.posix.time;
  35. }
  36. else
  37. {
  38. static assert(false, "Platform not supported");
  39. }
  40. ////////////////////////////////////////////////////////////////////////////////
  41. // Condition
  42. //
  43. // void wait();
  44. // void notify();
  45. // void notifyAll();
  46. ////////////////////////////////////////////////////////////////////////////////
  47. /**
  48. * This class represents a condition variable as conceived by C.A.R. Hoare. As
  49. * per Mesa type monitors however, "signal" has been replaced with "notify" to
  50. * indicate that control is not transferred to the waiter when a notification
  51. * is sent.
  52. */
  53. class Condition
  54. {
  55. ////////////////////////////////////////////////////////////////////////////
  56. // Initialization
  57. ////////////////////////////////////////////////////////////////////////////
  58. /**
  59. * Initializes a condition object which is associated with the supplied
  60. * mutex object.
  61. *
  62. * Params:
  63. * m = The mutex with which this condition will be associated.
  64. *
  65. * Throws:
  66. * SyncError on error.
  67. */
  68. this( Mutex m ) nothrow @safe
  69. {
  70. this(m, true);
  71. }
  72. /// ditto
  73. this( shared Mutex m ) shared nothrow @safe
  74. {
  75. this(m, true);
  76. }
  77. //
  78. private this(this Q, M)( M m, bool _unused_ ) nothrow @trusted
  79. if ((is(Q == Condition) && is(M == Mutex)) ||
  80. (is(Q == shared Condition) && is(M == shared Mutex)))
  81. {
  82. version (Windows)
  83. {
  84. static if (is(Q == Condition))
  85. {
  86. alias HANDLE_TYPE = void*;
  87. }
  88. else
  89. {
  90. alias HANDLE_TYPE = shared(void*);
  91. }
  92. m_blockLock = cast(HANDLE_TYPE) CreateSemaphoreA( null, 1, 1, null );
  93. if ( m_blockLock == m_blockLock.init )
  94. throw new SyncError( "Unable to initialize condition" );
  95. scope(failure) CloseHandle( cast(void*) m_blockLock );
  96. m_blockQueue = cast(HANDLE_TYPE) CreateSemaphoreA( null, 0, int.max, null );
  97. if ( m_blockQueue == m_blockQueue.init )
  98. throw new SyncError( "Unable to initialize condition" );
  99. scope(failure) CloseHandle( cast(void*) m_blockQueue );
  100. InitializeCriticalSection( cast(RTL_CRITICAL_SECTION*) &m_unblockLock );
  101. m_assocMutex = m;
  102. }
  103. else version (Posix)
  104. {
  105. m_assocMutex = m;
  106. static if ( is( typeof( pthread_condattr_setclock ) ) )
  107. {
  108. () @trusted
  109. {
  110. pthread_condattr_t attr = void;
  111. int rc = pthread_condattr_init( &attr );
  112. if ( rc )
  113. throw new SyncError( "Unable to initialize condition" );
  114. rc = pthread_condattr_setclock( &attr, CLOCK_MONOTONIC );
  115. if ( rc )
  116. throw new SyncError( "Unable to initialize condition" );
  117. rc = pthread_cond_init( cast(pthread_cond_t*) &m_hndl, &attr );
  118. if ( rc )
  119. throw new SyncError( "Unable to initialize condition" );
  120. rc = pthread_condattr_destroy( &attr );
  121. if ( rc )
  122. throw new SyncError( "Unable to initialize condition" );
  123. } ();
  124. }
  125. else
  126. {
  127. int rc = pthread_cond_init( cast(pthread_cond_t*) &m_hndl, null );
  128. if ( rc )
  129. throw new SyncError( "Unable to initialize condition" );
  130. }
  131. }
  132. }
  133. ~this()
  134. {
  135. version (Windows)
  136. {
  137. BOOL rc = CloseHandle( m_blockLock );
  138. assert( rc, "Unable to destroy condition" );
  139. rc = CloseHandle( m_blockQueue );
  140. assert( rc, "Unable to destroy condition" );
  141. DeleteCriticalSection( &m_unblockLock );
  142. }
  143. else version (Posix)
  144. {
  145. int rc = pthread_cond_destroy( &m_hndl );
  146. assert( !rc, "Unable to destroy condition" );
  147. }
  148. }
  149. ////////////////////////////////////////////////////////////////////////////
  150. // General Properties
  151. ////////////////////////////////////////////////////////////////////////////
  152. /**
  153. * Gets the mutex associated with this condition.
  154. *
  155. * Returns:
  156. * The mutex associated with this condition.
  157. */
  158. @property Mutex mutex()
  159. {
  160. return m_assocMutex;
  161. }
  162. /// ditto
  163. @property shared(Mutex) mutex() shared
  164. {
  165. return m_assocMutex;
  166. }
  167. // undocumented function for internal use
  168. final @property Mutex mutex_nothrow() pure nothrow @safe @nogc
  169. {
  170. return m_assocMutex;
  171. }
  172. // ditto
  173. final @property shared(Mutex) mutex_nothrow() shared pure nothrow @safe @nogc
  174. {
  175. return m_assocMutex;
  176. }
  177. ////////////////////////////////////////////////////////////////////////////
  178. // General Actions
  179. ////////////////////////////////////////////////////////////////////////////
  180. /**
  181. * Wait until notified.
  182. *
  183. * Throws:
  184. * SyncError on error.
  185. */
  186. void wait()
  187. {
  188. wait!(typeof(this))(true);
  189. }
  190. /// ditto
  191. void wait() shared
  192. {
  193. wait!(typeof(this))(true);
  194. }
  195. /// ditto
  196. void wait(this Q)( bool _unused_ )
  197. if (is(Q == Condition) || is(Q == shared Condition))
  198. {
  199. version (Windows)
  200. {
  201. timedWait( INFINITE );
  202. }
  203. else version (Posix)
  204. {
  205. int rc = pthread_cond_wait( cast(pthread_cond_t*) &m_hndl, (cast(Mutex) m_assocMutex).handleAddr() );
  206. if ( rc )
  207. throw new SyncError( "Unable to wait for condition" );
  208. }
  209. }
  210. /**
  211. * Suspends the calling thread until a notification occurs or until the
  212. * supplied time period has elapsed.
  213. *
  214. * Params:
  215. * val = The time to wait.
  216. *
  217. * In:
  218. * val must be non-negative.
  219. *
  220. * Throws:
  221. * SyncError on error.
  222. *
  223. * Returns:
  224. * true if notified before the timeout and false if not.
  225. */
  226. bool wait( Duration val )
  227. {
  228. return wait!(typeof(this))(val, true);
  229. }
  230. /// ditto
  231. bool wait( Duration val ) shared
  232. {
  233. return wait!(typeof(this))(val, true);
  234. }
  235. /// ditto
  236. bool wait(this Q)( Duration val, bool _unused_ )
  237. if (is(Q == Condition) || is(Q == shared Condition))
  238. in
  239. {
  240. assert( !val.isNegative );
  241. }
  242. do
  243. {
  244. version (Windows)
  245. {
  246. auto maxWaitMillis = dur!("msecs")( uint.max - 1 );
  247. while ( val > maxWaitMillis )
  248. {
  249. if ( timedWait( cast(uint)
  250. maxWaitMillis.total!"msecs" ) )
  251. return true;
  252. val -= maxWaitMillis;
  253. }
  254. return timedWait( cast(uint) val.total!"msecs" );
  255. }
  256. else version (Posix)
  257. {
  258. timespec t = void;
  259. mktspec( t, val );
  260. int rc = pthread_cond_timedwait( cast(pthread_cond_t*) &m_hndl,
  261. (cast(Mutex) m_assocMutex).handleAddr(),
  262. &t );
  263. if ( !rc )
  264. return true;
  265. if ( rc == ETIMEDOUT )
  266. return false;
  267. throw new SyncError( "Unable to wait for condition" );
  268. }
  269. }
  270. /**
  271. * Notifies one waiter.
  272. *
  273. * Throws:
  274. * SyncError on error.
  275. */
  276. void notify()
  277. {
  278. notify!(typeof(this))(true);
  279. }
  280. /// ditto
  281. void notify() shared
  282. {
  283. notify!(typeof(this))(true);
  284. }
  285. /// ditto
  286. void notify(this Q)( bool _unused_ )
  287. if (is(Q == Condition) || is(Q == shared Condition))
  288. {
  289. version (Windows)
  290. {
  291. notify_( false );
  292. }
  293. else version (Posix)
  294. {
  295. // Since OS X 10.7 (Lion), pthread_cond_signal returns EAGAIN after retrying 8192 times,
  296. // so need to retrying while it returns EAGAIN.
  297. //
  298. // 10.7.0 (Lion): http://www.opensource.apple.com/source/Libc/Libc-763.11/pthreads/pthread_cond.c
  299. // 10.8.0 (Mountain Lion): http://www.opensource.apple.com/source/Libc/Libc-825.24/pthreads/pthread_cond.c
  300. // 10.10.0 (Yosemite): http://www.opensource.apple.com/source/libpthread/libpthread-105.1.4/src/pthread_cond.c
  301. // 10.11.0 (El Capitan): http://www.opensource.apple.com/source/libpthread/libpthread-137.1.1/src/pthread_cond.c
  302. // 10.12.0 (Sierra): http://www.opensource.apple.com/source/libpthread/libpthread-218.1.3/src/pthread_cond.c
  303. // 10.13.0 (High Sierra): http://www.opensource.apple.com/source/libpthread/libpthread-301.1.6/src/pthread_cond.c
  304. // 10.14.0 (Mojave): http://www.opensource.apple.com/source/libpthread/libpthread-330.201.1/src/pthread_cond.c
  305. // 10.14.1 (Mojave): http://www.opensource.apple.com/source/libpthread/libpthread-330.220.2/src/pthread_cond.c
  306. int rc;
  307. do {
  308. rc = pthread_cond_signal( cast(pthread_cond_t*) &m_hndl );
  309. } while ( rc == EAGAIN );
  310. if ( rc )
  311. throw new SyncError( "Unable to notify condition" );
  312. }
  313. }
  314. /**
  315. * Notifies all waiters.
  316. *
  317. * Throws:
  318. * SyncError on error.
  319. */
  320. void notifyAll()
  321. {
  322. notifyAll!(typeof(this))(true);
  323. }
  324. /// ditto
  325. void notifyAll() shared
  326. {
  327. notifyAll!(typeof(this))(true);
  328. }
  329. /// ditto
  330. void notifyAll(this Q)( bool _unused_ )
  331. if (is(Q == Condition) || is(Q == shared Condition))
  332. {
  333. version (Windows)
  334. {
  335. notify_( true );
  336. }
  337. else version (Posix)
  338. {
  339. // Since OS X 10.7 (Lion), pthread_cond_broadcast returns EAGAIN after retrying 8192 times,
  340. // so need to retrying while it returns EAGAIN.
  341. //
  342. // 10.7.0 (Lion): http://www.opensource.apple.com/source/Libc/Libc-763.11/pthreads/pthread_cond.c
  343. // 10.8.0 (Mountain Lion): http://www.opensource.apple.com/source/Libc/Libc-825.24/pthreads/pthread_cond.c
  344. // 10.10.0 (Yosemite): http://www.opensource.apple.com/source/libpthread/libpthread-105.1.4/src/pthread_cond.c
  345. // 10.11.0 (El Capitan): http://www.opensource.apple.com/source/libpthread/libpthread-137.1.1/src/pthread_cond.c
  346. // 10.12.0 (Sierra): http://www.opensource.apple.com/source/libpthread/libpthread-218.1.3/src/pthread_cond.c
  347. // 10.13.0 (High Sierra): http://www.opensource.apple.com/source/libpthread/libpthread-301.1.6/src/pthread_cond.c
  348. // 10.14.0 (Mojave): http://www.opensource.apple.com/source/libpthread/libpthread-330.201.1/src/pthread_cond.c
  349. // 10.14.1 (Mojave): http://www.opensource.apple.com/source/libpthread/libpthread-330.220.2/src/pthread_cond.c
  350. int rc;
  351. do {
  352. rc = pthread_cond_broadcast( cast(pthread_cond_t*) &m_hndl );
  353. } while ( rc == EAGAIN );
  354. if ( rc )
  355. throw new SyncError( "Unable to notify condition" );
  356. }
  357. }
  358. private:
  359. version (Windows)
  360. {
  361. bool timedWait(this Q)( DWORD timeout )
  362. if (is(Q == Condition) || is(Q == shared Condition))
  363. {
  364. static if (is(Q == Condition))
  365. {
  366. auto op(string o, T, V1)(ref T val, V1 mod)
  367. {
  368. return mixin("val " ~ o ~ "mod");
  369. }
  370. }
  371. else
  372. {
  373. auto op(string o, T, V1)(ref shared T val, V1 mod)
  374. {
  375. import core.atomic: atomicOp;
  376. return atomicOp!o(val, mod);
  377. }
  378. }
  379. int numSignalsLeft;
  380. int numWaitersGone;
  381. DWORD rc;
  382. rc = WaitForSingleObject( cast(HANDLE) m_blockLock, INFINITE );
  383. assert( rc == WAIT_OBJECT_0 );
  384. op!"+="(m_numWaitersBlocked, 1);
  385. rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
  386. assert( rc );
  387. m_assocMutex.unlock();
  388. scope(failure) m_assocMutex.lock();
  389. rc = WaitForSingleObject( cast(HANDLE) m_blockQueue, timeout );
  390. assert( rc == WAIT_OBJECT_0 || rc == WAIT_TIMEOUT );
  391. bool timedOut = (rc == WAIT_TIMEOUT);
  392. EnterCriticalSection( &m_unblockLock );
  393. scope(failure) LeaveCriticalSection( &m_unblockLock );
  394. if ( (numSignalsLeft = m_numWaitersToUnblock) != 0 )
  395. {
  396. if ( timedOut )
  397. {
  398. // timeout (or canceled)
  399. if ( m_numWaitersBlocked != 0 )
  400. {
  401. op!"-="(m_numWaitersBlocked, 1);
  402. // do not unblock next waiter below (already unblocked)
  403. numSignalsLeft = 0;
  404. }
  405. else
  406. {
  407. // spurious wakeup pending!!
  408. m_numWaitersGone = 1;
  409. }
  410. }
  411. if ( op!"-="(m_numWaitersToUnblock, 1) == 0 )
  412. {
  413. if ( m_numWaitersBlocked != 0 )
  414. {
  415. // open the gate
  416. rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
  417. assert( rc );
  418. // do not open the gate below again
  419. numSignalsLeft = 0;
  420. }
  421. else if ( (numWaitersGone = m_numWaitersGone) != 0 )
  422. {
  423. m_numWaitersGone = 0;
  424. }
  425. }
  426. }
  427. else if ( op!"+="(m_numWaitersGone, 1) == int.max / 2 )
  428. {
  429. // timeout/canceled or spurious event :-)
  430. rc = WaitForSingleObject( cast(HANDLE) m_blockLock, INFINITE );
  431. assert( rc == WAIT_OBJECT_0 );
  432. // something is going on here - test of timeouts?
  433. op!"-="(m_numWaitersBlocked, m_numWaitersGone);
  434. rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
  435. assert( rc == WAIT_OBJECT_0 );
  436. m_numWaitersGone = 0;
  437. }
  438. LeaveCriticalSection( &m_unblockLock );
  439. if ( numSignalsLeft == 1 )
  440. {
  441. // better now than spurious later (same as ResetEvent)
  442. for ( ; numWaitersGone > 0; --numWaitersGone )
  443. {
  444. rc = WaitForSingleObject( cast(HANDLE) m_blockQueue, INFINITE );
  445. assert( rc == WAIT_OBJECT_0 );
  446. }
  447. // open the gate
  448. rc = ReleaseSemaphore( cast(HANDLE) m_blockLock, 1, null );
  449. assert( rc );
  450. }
  451. else if ( numSignalsLeft != 0 )
  452. {
  453. // unblock next waiter
  454. rc = ReleaseSemaphore( cast(HANDLE) m_blockQueue, 1, null );
  455. assert( rc );
  456. }
  457. m_assocMutex.lock();
  458. return !timedOut;
  459. }
  460. void notify_(this Q)( bool all )
  461. if (is(Q == Condition) || is(Q == shared Condition))
  462. {
  463. static if (is(Q == Condition))
  464. {
  465. auto op(string o, T, V1)(ref T val, V1 mod)
  466. {
  467. return mixin("val " ~ o ~ "mod");
  468. }
  469. }
  470. else
  471. {
  472. auto op(string o, T, V1)(ref shared T val, V1 mod)
  473. {
  474. import core.atomic: atomicOp;
  475. return atomicOp!o(val, mod);
  476. }
  477. }
  478. DWORD rc;
  479. EnterCriticalSection( &m_unblockLock );
  480. scope(failure) LeaveCriticalSection( &m_unblockLock );
  481. if ( m_numWaitersToUnblock != 0 )
  482. {
  483. if ( m_numWaitersBlocked == 0 )
  484. {
  485. LeaveCriticalSection( &m_unblockLock );
  486. return;
  487. }
  488. if ( all )
  489. {
  490. op!"+="(m_numWaitersToUnblock, m_numWaitersBlocked);
  491. m_numWaitersBlocked = 0;
  492. }
  493. else
  494. {
  495. op!"+="(m_numWaitersToUnblock, 1);
  496. op!"-="(m_numWaitersBlocked, 1);
  497. }
  498. LeaveCriticalSection( &m_unblockLock );
  499. }
  500. else if ( m_numWaitersBlocked > m_numWaitersGone )
  501. {
  502. rc = WaitForSingleObject( cast(HANDLE) m_blockLock, INFINITE );
  503. assert( rc == WAIT_OBJECT_0 );
  504. if ( 0 != m_numWaitersGone )
  505. {
  506. op!"-="(m_numWaitersBlocked, m_numWaitersGone);
  507. m_numWaitersGone = 0;
  508. }
  509. if ( all )
  510. {
  511. m_numWaitersToUnblock = m_numWaitersBlocked;
  512. m_numWaitersBlocked = 0;
  513. }
  514. else
  515. {
  516. m_numWaitersToUnblock = 1;
  517. op!"-="(m_numWaitersBlocked, 1);
  518. }
  519. LeaveCriticalSection( &m_unblockLock );
  520. rc = ReleaseSemaphore( cast(HANDLE) m_blockQueue, 1, null );
  521. assert( rc );
  522. }
  523. else
  524. {
  525. LeaveCriticalSection( &m_unblockLock );
  526. }
  527. }
  528. // NOTE: This implementation uses Algorithm 8c as described here:
  529. // http://groups.google.com/group/comp.programming.threads/
  530. // browse_frm/thread/1692bdec8040ba40/e7a5f9d40e86503a
  531. HANDLE m_blockLock; // auto-reset event (now semaphore)
  532. HANDLE m_blockQueue; // auto-reset event (now semaphore)
  533. Mutex m_assocMutex; // external mutex/CS
  534. CRITICAL_SECTION m_unblockLock; // internal mutex/CS
  535. int m_numWaitersGone = 0;
  536. int m_numWaitersBlocked = 0;
  537. int m_numWaitersToUnblock = 0;
  538. }
  539. else version (Posix)
  540. {
  541. Mutex m_assocMutex;
  542. pthread_cond_t m_hndl;
  543. }
  544. }
  545. ////////////////////////////////////////////////////////////////////////////////
  546. // Unit Tests
  547. ////////////////////////////////////////////////////////////////////////////////
  548. unittest
  549. {
  550. import core.thread;
  551. import core.sync.mutex;
  552. import core.sync.semaphore;
  553. void testNotify()
  554. {
  555. auto mutex = new Mutex;
  556. auto condReady = new Condition( mutex );
  557. auto semDone = new Semaphore;
  558. auto synLoop = new Object;
  559. int numWaiters = 10;
  560. int numTries = 10;
  561. int numReady = 0;
  562. int numTotal = 0;
  563. int numDone = 0;
  564. int numPost = 0;
  565. void waiter()
  566. {
  567. for ( int i = 0; i < numTries; ++i )
  568. {
  569. synchronized( mutex )
  570. {
  571. while ( numReady < 1 )
  572. {
  573. condReady.wait();
  574. }
  575. --numReady;
  576. ++numTotal;
  577. }
  578. synchronized( synLoop )
  579. {
  580. ++numDone;
  581. }
  582. semDone.wait();
  583. }
  584. }
  585. auto group = new ThreadGroup;
  586. for ( int i = 0; i < numWaiters; ++i )
  587. group.create( &waiter );
  588. for ( int i = 0; i < numTries; ++i )
  589. {
  590. for ( int j = 0; j < numWaiters; ++j )
  591. {
  592. synchronized( mutex )
  593. {
  594. ++numReady;
  595. condReady.notify();
  596. }
  597. }
  598. while ( true )
  599. {
  600. synchronized( synLoop )
  601. {
  602. if ( numDone >= numWaiters )
  603. break;
  604. }
  605. Thread.yield();
  606. }
  607. for ( int j = 0; j < numWaiters; ++j )
  608. {
  609. semDone.notify();
  610. }
  611. }
  612. group.joinAll();
  613. assert( numTotal == numWaiters * numTries );
  614. }
  615. void testNotifyAll()
  616. {
  617. auto mutex = new Mutex;
  618. auto condReady = new Condition( mutex );
  619. int numWaiters = 10;
  620. int numReady = 0;
  621. int numDone = 0;
  622. bool alert = false;
  623. void waiter()
  624. {
  625. synchronized( mutex )
  626. {
  627. ++numReady;
  628. while ( !alert )
  629. condReady.wait();
  630. ++numDone;
  631. }
  632. }
  633. auto group = new ThreadGroup;
  634. for ( int i = 0; i < numWaiters; ++i )
  635. group.create( &waiter );
  636. while ( true )
  637. {
  638. synchronized( mutex )
  639. {
  640. if ( numReady >= numWaiters )
  641. {
  642. alert = true;
  643. condReady.notifyAll();
  644. break;
  645. }
  646. }
  647. Thread.yield();
  648. }
  649. group.joinAll();
  650. assert( numReady == numWaiters && numDone == numWaiters );
  651. }
  652. void testWaitTimeout()
  653. {
  654. auto mutex = new Mutex;
  655. auto condReady = new Condition( mutex );
  656. bool waiting = false;
  657. bool alertedOne = true;
  658. bool alertedTwo = true;
  659. void waiter()
  660. {
  661. synchronized( mutex )
  662. {
  663. waiting = true;
  664. // we never want to miss the notification (30s)
  665. alertedOne = condReady.wait( dur!"seconds"(30) );
  666. // but we don't want to wait long for the timeout (10ms)
  667. alertedTwo = condReady.wait( dur!"msecs"(10) );
  668. }
  669. }
  670. auto thread = new Thread( &waiter );
  671. thread.start();
  672. while ( true )
  673. {
  674. synchronized( mutex )
  675. {
  676. if ( waiting )
  677. {
  678. condReady.notify();
  679. break;
  680. }
  681. }
  682. Thread.yield();
  683. }
  684. thread.join();
  685. assert( waiting );
  686. assert( alertedOne );
  687. assert( !alertedTwo );
  688. }
  689. testNotify();
  690. testNotifyAll();
  691. testWaitTimeout();
  692. }
  693. unittest
  694. {
  695. import core.thread;
  696. import core.sync.mutex;
  697. import core.sync.semaphore;
  698. void testNotify()
  699. {
  700. auto mutex = new shared Mutex;
  701. auto condReady = new shared Condition( mutex );
  702. auto semDone = new Semaphore;
  703. auto synLoop = new Object;
  704. int numWaiters = 10;
  705. int numTries = 10;
  706. int numReady = 0;
  707. int numTotal = 0;
  708. int numDone = 0;
  709. int numPost = 0;
  710. void waiter()
  711. {
  712. for ( int i = 0; i < numTries; ++i )
  713. {
  714. synchronized( mutex )
  715. {
  716. while ( numReady < 1 )
  717. {
  718. condReady.wait();
  719. }
  720. --numReady;
  721. ++numTotal;
  722. }
  723. synchronized( synLoop )
  724. {
  725. ++numDone;
  726. }
  727. semDone.wait();
  728. }
  729. }
  730. auto group = new ThreadGroup;
  731. for ( int i = 0; i < numWaiters; ++i )
  732. group.create( &waiter );
  733. for ( int i = 0; i < numTries; ++i )
  734. {
  735. for ( int j = 0; j < numWaiters; ++j )
  736. {
  737. synchronized( mutex )
  738. {
  739. ++numReady;
  740. condReady.notify();
  741. }
  742. }
  743. while ( true )
  744. {
  745. synchronized( synLoop )
  746. {
  747. if ( numDone >= numWaiters )
  748. break;
  749. }
  750. Thread.yield();
  751. }
  752. for ( int j = 0; j < numWaiters; ++j )
  753. {
  754. semDone.notify();
  755. }
  756. }
  757. group.joinAll();
  758. assert( numTotal == numWaiters * numTries );
  759. }
  760. void testNotifyAll()
  761. {
  762. auto mutex = new shared Mutex;
  763. auto condReady = new shared Condition( mutex );
  764. int numWaiters = 10;
  765. int numReady = 0;
  766. int numDone = 0;
  767. bool alert = false;
  768. void waiter()
  769. {
  770. synchronized( mutex )
  771. {
  772. ++numReady;
  773. while ( !alert )
  774. condReady.wait();
  775. ++numDone;
  776. }
  777. }
  778. auto group = new ThreadGroup;
  779. for ( int i = 0; i < numWaiters; ++i )
  780. group.create( &waiter );
  781. while ( true )
  782. {
  783. synchronized( mutex )
  784. {
  785. if ( numReady >= numWaiters )
  786. {
  787. alert = true;
  788. condReady.notifyAll();
  789. break;
  790. }
  791. }
  792. Thread.yield();
  793. }
  794. group.joinAll();
  795. assert( numReady == numWaiters && numDone == numWaiters );
  796. }
  797. void testWaitTimeout()
  798. {
  799. auto mutex = new shared Mutex;
  800. auto condReady = new shared Condition( mutex );
  801. bool waiting = false;
  802. bool alertedOne = true;
  803. bool alertedTwo = true;
  804. void waiter()
  805. {
  806. synchronized( mutex )
  807. {
  808. waiting = true;
  809. // we never want to miss the notification (30s)
  810. alertedOne = condReady.wait( dur!"seconds"(30) );
  811. // but we don't want to wait long for the timeout (10ms)
  812. alertedTwo = condReady.wait( dur!"msecs"(10) );
  813. }
  814. }
  815. auto thread = new Thread( &waiter );
  816. thread.start();
  817. while ( true )
  818. {
  819. synchronized( mutex )
  820. {
  821. if ( waiting )
  822. {
  823. condReady.notify();
  824. break;
  825. }
  826. }
  827. Thread.yield();
  828. }
  829. thread.join();
  830. assert( waiting );
  831. assert( alertedOne );
  832. assert( !alertedTwo );
  833. }
  834. testNotify();
  835. testNotifyAll();
  836. testWaitTimeout();
  837. }