• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1[/
2  (C) Copyright 2013 Vicente J. Botet Escriba.
3  Distributed under the Boost Software License, Version 1.0.
4  (See accompanying file LICENSE_1_0.txt or copy at
5  http://www.boost.org/LICENSE_1_0.txt).
6]
7
8[section:latches Latches -- EXPERIMENTAL]
9
10[////////////////////]
11[section Introdcution]
12
13Latches are a thread co-ordination mechanism that allow one or more threads to block until one or more threads have reached a point.
14
15[/
16An individual latch is a reusable object; once the operation has been completed, the threads can re-use the same barrier. It is thus useful for managing repeated tasks handled by multiple threads.
17
18A completion latch is like a latch that allows to associate a completion function which will be called once the internal counter reaches the value 0 and all the consumer threads have taken care of the notification.
19]
20
21[endsect]
22[////////////////]
23[section Examples]
24
25Sample use cases for the latch include:
26
27* Setting multiple threads to perform a task, and then waiting until all threads have reached a common point.
28* Creating multiple threads, which wait for a signal before advancing beyond a common point.
29
30An example of the first use case would be as follows:
31
32  void DoWork(thread_pool* pool) {
33    latch completion_latch(NTASKS);
34    for (int i = 0; i < NTASKS; ++i) {
35      pool->submit([&] {
36        // perform work
37        ...
38        completion_latch.count_down();
39      }));
40    }
41    // Block until work is done
42    completion_latch.wait();
43  }
44
45An example of the second use case is shown below. We need to load data and then process it using a number of threads. Loading the data is I/O bound, whereas starting threads and creating data structures is CPU bound. By running these in parallel, throughput can be increased.
46
47  void DoWork() {
48    latch start_latch(1);
49    vector<thread*> workers;
50    for (int i = 0; i < NTHREADS; ++i) {
51      workers.push_back(new thread([&] {
52        // Initialize data structures. This is CPU bound.
53        ...
54        start_latch.wait();
55        // perform work
56        ...
57      }));
58    }
59    // Load input data. This is I/O bound.
60    ...
61    // Threads can now start processing
62    start_latch.count_down();
63    }
64
65[/
66The completion latches can be used to co-ordinate also a set of threads carrying out a repeated task. The number of threads can be adjusted dynamically to respond to changing requirements.
67
68In the example below, a number of threads are performing a multi-stage task. Some tasks may require fewer steps than others, meaning that some threads may finish before others. We reduce the number of threads waiting on the latch when this happens.
69
70  void DoWork() {
71    Tasks& tasks;
72    size_t initial_threads;
73    atomic<size_t> current_threads(initial_threads)
74    vector<thread*> workers;
75
76    // Create a barrier, and set a lambda that will be invoked every time the
77    // barrier counts down. If one or more active threads have completed,
78    // reduce the number of threads.
79    completion_latch task_barrier(n_threads);
80    task_barrier.then([&] {
81      task_barrier.reset(current_threads);
82    });
83
84    for (int i = 0; i < n_threads; ++i) {
85        workers.push_back(new thread([&] {
86        bool active = true;
87        while(active) {
88          Task task = tasks.get();
89          // perform task
90          ...
91          if (finished(task)) {
92            current_threads--;
93            active = false;
94          }
95          task_barrier.count_down_and_wait();
96         }
97       });
98    }
99
100    // Read each stage of the task until all stages are complete.
101    while (!finished()) {
102      GetNextStage(tasks);
103    }
104  }
105]
106
107[endsect]
108[///////////////////////////]
109[section:latch Class `latch`]
110
111    #include <boost/thread/latch.hpp>
112
113    class latch
114    {
115    public:
116        latch(latch const&) = delete;
117        latch& operator=(latch const&) = delete;
118
119        latch(std::size_t count);
120        ~latch();
121
122        void wait();
123        bool try_wait();
124        template <class Rep, class Period>
125        cv_status wait_for(const chrono::duration<Rep, Period>& rel_time);
126        template <class lock_type, class Clock, class Duration>
127        cv_status wait_until(const chrono::time_point<Clock, Duration>& abs_time);
128        void count_down();
129        void count_down_and_wait();
130
131    };
132
133
134[/
135        void reset(std::size_t count);
136]
137
138A latch maintains an internal counter that is initialized when the latch is created. One or more threads may block waiting until the counter is decremented to 0.
139
140Instances of __latch__ are not copyable or movable.
141
142[///////////////////]
143[section Constructor `latch(std::size_t)`]
144
145    latch(std::size_t count);
146
147[variablelist
148
149[[Effects:] [Construct a latch with is initial value for the internal counter.]]
150
151[[Note:] [The counter could be zero.]]
152
153[[Throws:] [Nothing.]]
154
155]
156
157[endsect]
158[//////////////////]
159[section Destructor `~latch()`]
160
161    ~latch();
162
163[variablelist
164
165[[Precondition:] [No threads are waiting or invoking  count_down on `*this`.]]
166
167[[Effects:] [Destroys `*this` latch.]]
168
169[[Throws:] [Nothing.]]
170
171]
172
173[endsect]
174[/////////////////////////////////////]
175[section:wait Member Function `wait()`]
176
177    void wait();
178
179[variablelist
180
181[[Effects:] [Block the calling thread until the internal count reaches the value zero. Then all waiting threads
182are unblocked. ]]
183
184[[Throws:] [
185 - __thread_resource_error__ if an error occurs.
186
187 - __thread_interrupted__ if the wait was interrupted by a call to __interrupt__ on the __thread__ object associated with the current thread of execution.
188]]
189
190[[Notes:] [`wait()` is an ['interruption point].]]
191
192]
193
194[endsect]
195[/////////////////////////////////////////////]
196[section:try_wait Member Function `try_wait()`]
197
198        bool try_wait();
199
200[variablelist
201
202[[Returns:] [Returns true if the internal count is 0, and false otherwise. Does not block the calling thread. ]]
203
204[[Throws:] [
205 - __thread_resource_error__ if an error occurs.
206]]
207
208]
209
210[endsect]
211[//////////////////////////////////////////////]
212[section:wait_for Member Function `wait_for() `]
213
214        template <class Rep, class Period>
215        cv_status wait_for(const chrono::duration<Rep, Period>& rel_time);
216
217[variablelist
218
219[[Effects:] [Block the calling thread until the internal count reaches the value zero or the duration has been elapsed. If no timeout, all waiting threads are unblocked. ]]
220[[Returns:] [cv_status::no_timeout  if the internal count is 0, and cv_status::timeout if duration has been elapsed. ]]
221
222[[Throws:] [
223 - __thread_resource_error__ if an error occurs.
224
225 - __thread_interrupted__ if the wait was interrupted by a call to __interrupt__ on the __thread__ object associated with the current thread of execution.
226]]
227
228[[Notes:] [`wait_for()` is an ['interruption point].]]
229
230]
231
232[endsect]
233[/////////////////////////////////////////////////]
234[section:wait_until Member Function `wait_until()`]
235
236        template <class lock_type, class Clock, class Duration>
237        cv_status wait_until(const chrono::time_point<Clock, Duration>& abs_time);
238
239[variablelist
240
241[[Effects:] [Block the calling thread until the internal count reaches the value zero or the time_point has been reached. If no timeout, all waiting threads are unblocked. ]]
242[[Returns:] [cv_status::no_timeout  if the internal count is 0, and cv_status::timeout if time_point has been reached.]]
243
244[[Throws:] [
245 - __thread_resource_error__ if an error occurs.
246
247 - __thread_interrupted__ if the wait was interrupted by a call to __interrupt__ on the __thread__ object associated with the current thread of execution.
248]]
249
250[[Notes:] [`wait_until()` is an ['interruption point].]]
251
252]
253
254[endsect]
255[/////////////////////////////////////////////////]
256[section:count_down Member Function `count_down()`]
257
258        void count_down();
259
260[variablelist
261
262[[Requires:] [The internal counter is non zero.]]
263[[Effects:] [Decrements the internal count by 1, and returns. If the count reaches 0, any threads blocked in wait() will be released. ]]
264
265[[Throws:] [
266 - __thread_resource_error__ if an error occurs.
267
268 - __thread_interrupted__ if the wait was interrupted by a call to __interrupt__ on the __thread__ object associated with the current thread of execution.
269]]
270[[Notes:] [`count_down()` is an ['interruption point].]]
271
272]
273
274[endsect]
275[///////////////////////////////////////////////////////////////////]
276[section:count_down_and_wait Member Function `count_down_and_wait()`]
277
278        void count_down_and_wait();
279
280[variablelist
281
282[[Requires:] [The internal counter is non zero.]]
283[[Effects:] [Decrements the internal count by 1. If the resulting count is not 0, blocks the calling thread until the internal count is decremented to 0 by one or more other threads calling count_down() or count_down_and_wait(). ]]
284
285[[Throws:] [
286 - __thread_resource_error__ if an error occurs.
287
288 - __thread_interrupted__ if the wait was interrupted by a call to __interrupt__ on the __thread__ object associated with the current thread of execution.
289]]
290[[Notes:] [`count_down_and_wait()` is an ['interruption point].]]
291
292]
293
294[endsect]
295[///////////////////////////////////////]
296[
297[section:reset Member Function `reset()`]
298
299    reset( size_t );
300
301[variablelist
302
303[[Requires:] [This function may only be invoked when there are no other threads currently inside the waiting functions.]]
304
305[[Returns:] [Resets the latch with a new value for the initial thread count.  ]]
306
307[[Throws:] [
308 - __thread_resource_error__ if an error occurs.
309]]
310
311]
312
313[endsect]
314]
315[endsect] [/ latch]
316
317[/
318[//////////////////////////////////////////////////]
319[section:completion_latch Class `completion_latch `]
320
321    #include <boost/thread/completion_latch.hpp>
322
323    class completion_latch
324    {
325    public:
326        typedef 'implementation defined' completion_function;
327
328        completion_latch(completion_latch const&) = delete;
329        completion_latch& operator=(completion_latch const&) = delete;
330
331        completion_latch(std::size_t count);
332        template <typename F>
333        completion_latch(std::size_t count, F&& funct);
334        ~completion_latch();
335
336        void wait();
337        bool try_wait();
338        template <class Rep, class Period>
339        cv_status wait_for(const chrono::duration<Rep, Period>& rel_time);
340        template <class lock_type, class Clock, class Duration>
341        cv_status wait_until(const chrono::time_point<Clock, Duration>& abs_time);
342        void count_down();
343        void count_down_and_wait();
344        void reset(std::size_t count);
345        template <typename F>
346        completion_function then(F&& funct);
347    };
348
349
350A completion latch is like a latch that allows to associate a completion function which will be called once the internal counter reaches the value 0 and all the consumer threads have taken care of the notification.
351
352Instances of completion_latch are not copyable or movable.
353
354Only the additional functions are documented.
355
356[/////////////////////]
357[section:c Constructor]
358
359    completion_latch(std::size_t count);
360
361[variablelist
362
363[[Effects:] [Construct a completion_latch with is initial value for the internal counter and a noop completion function.]]
364
365[[Note:] [The counter could be zero and rest later.]]
366
367[[Throws:] [Nothing.]]
368
369]
370
371[endsect]
372[///////////////////////////////////////////////]
373[section:cf Constructor with completion function]
374
375        template <typename F>
376        completion_latch(std::size_t count, F&& funct);
377
378[variablelist
379
380[[Effects:] [Construct a completion_latch with is initial value for the internal counter and the completion function `funct`.]]
381
382[[Note:] [The counter could be zero and reset later.]]
383
384[[Throws:] [
385
386 - Any exception thrown by the copy/move construction of funct.
387
388]]
389
390]
391
392[endsect]
393[///////////////////////////////////]
394[section:then Member Function `then`]
395
396        template <typename F>
397        completion_function then(F&& funct);
398
399[variablelist
400
401[[Requires:] [This function may only be invoked when there are no other threads currently inside the waiting functions. It may also be invoked from within the registered completion function. ]]
402
403[[Effects:] [Associates the parameter `funct` as completion function of the latch. The next time the internal count reaches 0, this function will be invoked.]]
404[[Returns:] [The old completion function.]]
405
406[[Throws:] [
407 - __thread_resource_error__ if an error occurs.
408
409 - Any exception thrown by the copy/move construction of completion functions.
410]]
411
412]
413
414[endsect]
415
416[endsect] [/ completion_latch]
417]
418
419[endsect] [/ Latches]
420