Lines Matching +full:test +full:- +full:pool
2 // file at the top-level directory of this distribution and at
3 // http://rust-lang.org/COPYRIGHT.
5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
11 //! A thread pool used to execute functions in parallel.
13 //! Spawns a specified number of worker threads and replenishes the pool if any worker threads
28 //! let pool = ThreadPool::new(n_workers);
33 //! pool.execute(move|| {
34 //! tx.send(1).expect("channel will be there waiting for the pool");
43 //! Keep in mind, if a barrier synchronizes more jobs than you have workers in the pool,
46 //! https://doc.rust-lang.org/reference/behavior-not-considered-unsafe.html).
56 //! let pool = ThreadPool::new(n_workers);
67 //! pool.execute(move|| {
107 fn new(shared_data: &'a Arc<ThreadPoolSharedData>) -> Sentinel<'a> { in new()
152 /// let pool = threadpool::Builder::new()
174 pub fn new() -> Builder { in new()
182 /// Set the maximum number of worker-threads that will be alive at any given moment by the built
193 /// No more than eight threads will be alive simultaneously for this pool:
198 /// let pool = threadpool::Builder::new()
203 /// pool.execute(|| {
208 pub fn num_threads(mut self, num_threads: usize) -> Builder { in num_threads()
215 /// specified, threads spawned by the thread pool will be unnamed.
221 /// Each thread spawned by this pool will have the name "foo":
226 /// let pool = threadpool::Builder::new()
231 /// pool.execute(|| {
236 pub fn thread_name(mut self, name: String) -> Builder { in thread_name()
245 /// [thread]: https://doc.rust-lang.org/nightly/std/thread/index.html#stack-size
250 /// Each thread spawned by this pool will have a 4 MB stack:
253 /// let pool = threadpool::Builder::new()
258 /// pool.execute(|| {
263 pub fn thread_stack_size(mut self, size: usize) -> Builder { in thread_stack_size()
276 /// let pool = threadpool::Builder::new()
281 pub fn build(self) -> ThreadPool { in build()
325 fn has_work(&self) -> bool { in has_work()
329 /// Notify all observers joining this pool if there is no more work to do.
341 /// Abstraction of a thread pool for basic parallelism.
352 /// Creates a new thread pool capable of executing `num_threads` number of jobs concurrently.
360 /// Create a new thread pool capable of executing four jobs concurrently:
365 /// let pool = ThreadPool::new(4);
367 pub fn new(num_threads: usize) -> ThreadPool { in new()
371 /// Creates a new thread pool capable of executing `num_threads` number of jobs concurrently.
384 /// let pool = ThreadPool::with_name("worker".into(), 2);
386 /// pool.execute(|| {
393 /// pool.join();
396 /// [thread name]: https://doc.rust-lang.org/std/thread/struct.Thread.html#method.name
397 pub fn with_name(name: String, num_threads: usize) -> ThreadPool { in with_name()
407 pub fn new_with_name(name: String, num_threads: usize) -> ThreadPool { in new_with_name()
411 /// Executes the function `job` on a thread in the pool.
415 /// Execute four jobs on a thread pool that can run two jobs concurrently:
420 /// let pool = ThreadPool::new(2);
421 /// pool.execute(|| println!("hello"));
422 /// pool.execute(|| println!("world"));
423 /// pool.execute(|| println!("foo"));
424 /// pool.execute(|| println!("bar"));
425 /// pool.join();
437 /// Returns the number of jobs waiting to executed in the pool.
446 /// let pool = ThreadPool::new(2);
448 /// pool.execute(|| {
454 /// assert_eq!(8, pool.queued_count());
456 pub fn queued_count(&self) -> usize { in queued_count()
469 /// let pool = ThreadPool::new(4);
471 /// pool.execute(move || {
477 /// assert_eq!(4, pool.active_count());
479 pub fn active_count(&self) -> usize { in active_count()
483 /// Returns the maximum number of threads the pool will execute concurrently.
490 /// let mut pool = ThreadPool::new(4);
491 /// assert_eq!(4, pool.max_count());
493 /// pool.set_num_threads(8);
494 /// assert_eq!(8, pool.max_count());
496 pub fn max_count(&self) -> usize { in max_count()
500 /// Returns the number of panicked threads over the lifetime of the pool.
507 /// let pool = ThreadPool::new(4);
509 /// pool.execute(move || {
516 /// pool.join();
518 /// assert_eq!(5, pool.panic_count());
520 pub fn panic_count(&self) -> usize { in panic_count()
530 /// Sets the number of worker-threads to use as `num_threads`.
545 /// let mut pool = ThreadPool::new(4);
547 /// pool.execute(move || {
553 /// assert_eq!(4, pool.active_count());
554 /// assert_eq!(6, pool.queued_count());
556 /// // Increase thread capacity of the pool
557 /// pool.set_num_threads(8);
560 /// assert_eq!(8, pool.active_count());
561 /// assert_eq!(2, pool.queued_count());
563 /// // Decrease thread capacity of the pool
565 /// pool.set_num_threads(4);
567 /// assert_eq!(8, pool.active_count());
568 /// assert_eq!(2, pool.queued_count());
584 /// Block the current thread until all jobs in the pool have been executed.
586 /// Calling `join` on an empty pool will cause an immediate return.
589 /// event will exit together even if the pool is processing new jobs by the
592 /// Calling `join` from a thread within the pool will cause a deadlock. This
602 /// let pool = ThreadPool::new(8);
607 /// pool.execute(move || {
612 /// pool.join();
640 /// Cloning a pool will create a new handle to the pool.
641 /// The behavior is similar to [Arc](https://doc.rust-lang.org/stable/std/sync/struct.Arc.html).
650 /// let pool = ThreadPool::with_name("clone example".into(), 2);
654 /// let pool = pool.clone();
659 /// pool.execute(move || {
676 fn clone(&self) -> ThreadPool { in clone()
684 /// Create a thread pool with one thread per CPU.
688 fn default() -> Self { in default()
694 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { in fmt()
705 /// Check if you are working with the same pool
720 fn eq(&self, other: &ThreadPool) -> bool { in eq()
744 // Shutdown this thread if the pool has become smaller in spawn_in_pool()
780 #[cfg(test)]
781 mod test { module
791 #[test]
794 let mut pool = ThreadPool::new(TEST_TASKS); in test_set_num_threads_increasing() localVariable
796 pool.execute(move || sleep(Duration::from_secs(23))); in test_set_num_threads_increasing()
799 assert_eq!(pool.active_count(), TEST_TASKS); in test_set_num_threads_increasing()
801 pool.set_num_threads(new_thread_amount); in test_set_num_threads_increasing()
803 for _ in 0..(new_thread_amount - TEST_TASKS) { in test_set_num_threads_increasing()
804 pool.execute(move || sleep(Duration::from_secs(23))); in test_set_num_threads_increasing()
807 assert_eq!(pool.active_count(), new_thread_amount); in test_set_num_threads_increasing()
809 pool.join(); in test_set_num_threads_increasing()
812 #[test]
815 let mut pool = ThreadPool::new(TEST_TASKS); in test_set_num_threads_decreasing() localVariable
817 pool.execute(move || { in test_set_num_threads_decreasing()
821 pool.set_num_threads(new_thread_amount); in test_set_num_threads_decreasing()
823 pool.execute(move || sleep(Duration::from_secs(23))); in test_set_num_threads_decreasing()
826 assert_eq!(pool.active_count(), new_thread_amount); in test_set_num_threads_decreasing()
828 pool.join(); in test_set_num_threads_decreasing()
831 #[test]
833 let pool = ThreadPool::new(TEST_TASKS); in test_active_count() localVariable
835 pool.execute(move || loop { in test_active_count()
840 let active_count = pool.active_count(); in test_active_count()
842 let initialized_count = pool.max_count(); in test_active_count()
846 #[test]
848 let pool = ThreadPool::new(TEST_TASKS); in test_works() localVariable
853 pool.execute(move || { in test_works()
861 #[test]
867 #[test]
869 let pool = ThreadPool::new(TEST_TASKS); in test_recovery_from_subtask_panic() localVariable
873 pool.execute(move || panic!("Ignore this panic, it must!")); in test_recovery_from_subtask_panic()
875 pool.join(); in test_recovery_from_subtask_panic()
877 assert_eq!(pool.panic_count(), TEST_TASKS); in test_recovery_from_subtask_panic()
883 pool.execute(move || { in test_recovery_from_subtask_panic()
891 #[test]
893 let pool = ThreadPool::new(TEST_TASKS); in test_should_not_panic_on_drop_if_subtasks_panic_after_drop() localVariable
899 pool.execute(move || { in test_should_not_panic_on_drop_if_subtasks_panic_after_drop()
905 drop(pool); in test_should_not_panic_on_drop_if_subtasks_panic_after_drop()
911 #[test]
915 let pool = ThreadPool::new(TEST_TASKS); in test_massive_task_creation() localVariable
925 pool.execute(move || { in test_massive_task_creation()
926 // Wait until the pool has been filled once. in test_massive_task_creation()
929 // wait so the pool can be measured in test_massive_task_creation()
938 assert_eq!(pool.active_count(), TEST_TASKS); in test_massive_task_creation()
942 pool.join(); in test_massive_task_creation()
944 let atomic_active_count = pool.active_count(); in test_massive_task_creation()
952 #[test]
956 let mut pool = ThreadPool::new(test_tasks_begin); in test_shrink() localVariable
962 pool.execute(move || { in test_shrink()
973 pool.execute(move || { in test_shrink()
980 pool.set_num_threads(TEST_TASKS); in test_shrink()
982 assert_eq!(pool.active_count(), test_tasks_begin); in test_shrink()
986 assert_eq!(pool.active_count(), TEST_TASKS); in test_shrink()
990 #[test]
992 let name = "test"; in test_name()
993 let mut pool = ThreadPool::with_name(name.to_owned(), 2); in test_name() localVariable
996 // initial thread should share the name "test" in test_name()
999 pool.execute(move || { in test_name()
1005 // new spawn thread should share the name "test" too. in test_name()
1006 pool.set_num_threads(3); in test_name()
1008 pool.execute(move || { in test_name()
1014 // recover thread should share the name "test" too. in test_name()
1015 pool.execute(move || { in test_name()
1025 #[test]
1027 let pool = ThreadPool::new(4); in test_debug() localVariable
1028 let debug = format!("{:?}", pool); in test_debug()
1034 let pool = ThreadPool::with_name("hello".into(), 4); in test_debug() localVariable
1035 let debug = format!("{:?}", pool); in test_debug()
1041 let pool = ThreadPool::new(4); in test_debug() localVariable
1042 pool.execute(move || sleep(Duration::from_secs(5))); in test_debug()
1044 let debug = format!("{:?}", pool); in test_debug()
1051 #[test]
1053 let pool = ThreadPool::with_name("repeate join test".into(), 8); in test_repeate_join() localVariable
1058 pool.execute(move || { in test_repeate_join()
1064 println!("{:?}", pool); in test_repeate_join()
1065 pool.join(); in test_repeate_join()
1070 pool.execute(move || { in test_repeate_join()
1075 pool.join(); in test_repeate_join()
1079 #[test]
1101 error(format!("p1: {} -=- {:?}\n", i, pool0_)); in test_multi_join()
1104 tx.send(i).expect("send i from pool1 -> main"); in test_multi_join()
1114 error(format!("pool0.join() complete =-= {:?}", pool1)); in test_multi_join()
1123 #[test]
1125 // Joining an empty pool must return imminently in test_empty_pool()
1126 let pool = ThreadPool::new(4); in test_empty_pool() localVariable
1128 pool.join(); in test_empty_pool()
1133 #[test]
1141 let pool = ThreadPool::with_name("no fun or joy".into(), 8); in test_no_fun_or_joy() localVariable
1143 pool.execute(sleepy_function); in test_no_fun_or_joy()
1145 let p_t = pool.clone(); in test_no_fun_or_joy()
1150 pool.join(); in test_no_fun_or_joy()
1153 #[test]
1155 let pool = ThreadPool::with_name("clone example".into(), 2); in test_clone() localVariable
1157 // This batch of jobs will occupy the pool for some time in test_clone()
1159 pool.execute(move || { in test_clone()
1164 // The following jobs will be inserted into the pool in a random fashion in test_clone()
1166 let pool = pool.clone(); in test_clone() localVariable
1169 pool.join(); in test_clone()
1174 pool.execute(move || { in test_clone()
1184 let pool = pool.clone(); in test_clone() localVariable
1187 pool.join(); in test_clone()
1192 pool.execute(move || { in test_clone()
1214 #[test]
1220 #[test]
1226 #[test]
1232 #[test]
1239 #[test]
1241 /// of joins has completed. So once one thread joining on a pool has
1242 /// succeded other threads joining on the same pool must get out even if
1247 /// groups of four because the waiter pool has four workers.
1305 let mut dur = after - now; in test_join_wavesurfer()
1306 if dur >= n_cycles - 1 { in test_join_wavesurfer()
1307 dur = n_cycles - 1; in test_join_wavesurfer()