core.sync.barrier source code

1 /**
2  * The barrier module provides a primitive for synchronizing the progress of
3  * a group of threads.
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/_barrier.d)
9  */
10 
11 /*          Copyright Sean Kelly 2005 - 2009.
12  * Distributed under the Boost Software License, Version 1.0.
13  *    (See accompanying file LICENSE or copy at
14  *          http://www.boost.org/LICENSE_1_0.txt)
15  */
16 module core.sync.barrier;
17 
18 
19 public import core.sync.exception;
20 import core.sync.condition;
21 import core.sync.mutex;
22 
23 
24 ////////////////////////////////////////////////////////////////////////////////
25 // Barrier
26 //
27 // void wait();
28 ////////////////////////////////////////////////////////////////////////////////
29 
30 
31 /**
32  * This class represents a barrier across which threads may only travel in
33  * groups of a specific size.
34  */
35 class Barrier
36 {
37     ////////////////////////////////////////////////////////////////////////////
38     // Initialization
39     ////////////////////////////////////////////////////////////////////////////
40 
41 
42     /**
43      * Initializes a barrier object which releases threads in groups of limit
44      * in size.
45      *
46      * Params:
47      *  limit = The number of waiting threads to release in unison.
48      *
49      * Throws:
50      *  SyncError on error.
51      */
52     this( uint limit )
53     in
54     {
55         assert( limit > 0 );
56     }
57     do
58     {
59         m_lock  = new Mutex;
60         m_cond  = new Condition( m_lock );
61         m_group = 0;
62         m_limit = limit;
63         m_count = limit;
64     }
65 
66 
67     ////////////////////////////////////////////////////////////////////////////
68     // General Actions
69     ////////////////////////////////////////////////////////////////////////////
70 
71 
72     /**
73      * Wait for the pre-determined number of threads and then proceed.
74      *
75      * Throws:
76      *  SyncError on error.
77      */
78     void wait()
79     {
80         synchronized( m_lock )
81         {
82             uint group = m_group;
83 
84             if ( --m_count == 0 )
85             {
86                 m_group++;
87                 m_count = m_limit;
88                 m_cond.notifyAll();
89             }
90             while ( group == m_group )
91                 m_cond.wait();
92         }
93     }
94 
95 
96 private:
97     Mutex       m_lock;
98     Condition   m_cond;
99     uint        m_group;
100     uint        m_limit;
101     uint        m_count;
102 }
103 
104 
105 ////////////////////////////////////////////////////////////////////////////////
106 // Unit Tests
107 ////////////////////////////////////////////////////////////////////////////////
108 
109 unittest
110 {
111     import core.thread;
112 
113     int  numThreads = 10;
114     auto barrier    = new Barrier( numThreads );
115     auto synInfo    = new Object;
116     int  numReady   = 0;
117     int  numPassed  = 0;
118 
119     void threadFn()
120     {
121         synchronized( synInfo )
122         {
123             ++numReady;
124         }
125         barrier.wait();
126         synchronized( synInfo )
127         {
128             ++numPassed;
129         }
130     }
131 
132     auto group = new ThreadGroup;
133 
134     for ( int i = 0; i < numThreads; ++i )
135     {
136         group.create( &threadFn );
137     }
138     group.joinAll();
139     assert( numReady == numThreads && numPassed == numThreads );
140 }