yaks 0.1.0

Minimalistic framework for automatic multithreading of hecs via rayon
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161

use hecs::{Entity, Fetch, Query, QueryBorrow};

#[cfg_attr(not(feature = "parallel"), allow(unused_variables))]
/// Distributes over a `rayon` thread pool the work of applying a function to items in a query.
/// See [`hecs::QueryBorrow::iter_batched()`](../hecs/struct.QueryBorrow.html#method.iter_batched).
///
/// If the default `parallel` feature is disabled the functionality is identical
/// to `query_borrow.iter().for_each(for_each)`.
///
/// Calling `batch()` standalone will use the global `rayon` thread pool:
/// ```rust
/// # struct Pos;
/// # struct Vel;
/// # impl std::ops::AddAssign<&Vel> for Pos {
/// #     fn add_assign(&mut self, _: &Vel) {}
/// # }
/// # let world = hecs::World::new();
/// # let num_entities = 64;
/// yaks::batch(
///     &mut world.query::<(&mut Pos, &Vel)>(),
///     num_entities / 16,
///     |_entity, (pos, vel)| {
///         *pos += vel;
///     },
/// );
/// ```
/// Alternatively, a specific thread pool can be used via
/// [`rayon::ThreadPool::install()`](../rayon/struct.ThreadPool.html#method.install):
/// ```rust
/// # struct Pos;
/// # struct Vel;
/// # impl std::ops::AddAssign<&Vel> for Pos {
/// #     fn add_assign(&mut self, _: &Vel) {}
/// # }
/// # let world = hecs::World::new();
/// # let num_entities = 64;
/// # #[cfg(feature = "parallel")]
/// # let thread_pool =
/// # {
/// #     rayon::ThreadPoolBuilder::new().build().unwrap()
/// # };
/// # #[cfg(not(feature = "parallel"))]
/// # let thread_pool =
/// # {
/// #     struct DummyPool;
/// #     impl DummyPool {
/// #         fn install(&self, mut closure: impl FnMut()) {
/// #             closure();
/// #         }
/// #     }
/// #     DummyPool
/// # };
/// thread_pool.install(|| {
///     yaks::batch(
///         &mut world.query::<(&mut Pos, &Vel)>(),
///         num_entities / 16,
///         |_entity, (pos, vel)| {
///             *pos += vel;
///         },
///     )
/// });
/// ```
/// `batch()` can be called in systems, where it will use whichever thread pool is used by
/// the system or the executor it's in:
/// ```rust
/// # use yaks::{QueryMarker, Executor};
/// # struct Pos;
/// # struct Vel;
/// # impl std::ops::AddAssign<&Vel> for Pos {
/// #     fn add_assign(&mut self, _: &Vel) {}
/// # }
/// # let world = hecs::World::new();
/// # let mut num_entities = 64;
/// # #[cfg(feature = "parallel")]
/// # let thread_pool =
/// # {
/// #     rayon::ThreadPoolBuilder::new().build().unwrap()
/// # };
/// # #[cfg(not(feature = "parallel"))]
/// # let thread_pool =
/// # {
/// #     struct DummyPool;
/// #     impl DummyPool {
/// #         fn install(&self, mut closure: impl FnMut()) {
/// #             closure();
/// #         }
/// #     }
/// #     DummyPool
/// # };
/// let mut executor = Executor::<(u32, )>::builder()
///     .system(|context, num_entities: &u32, query: QueryMarker<(&mut Pos, &Vel)>| {
///         yaks::batch(
///             &mut context.query(query),
///             num_entities / 16,
///             |_entity, (pos, vel)| {
///                 *pos += vel;
///             },
///         )
///     })
///     .build();
///
/// executor.run(&world, &mut num_entities);
///
/// thread_pool.install(|| {
///     executor.run(&world, &mut num_entities);
/// });
/// ```
pub fn batch<'query, 'world, Q, F>(
    query_borrow: &'query mut QueryBorrow<'world, Q>,
    batch_size: u32,
    for_each: F,
) where
    Q: Query + Send + Sync + 'query,
    F: Fn(Entity, <<Q as Query>::Fetch as Fetch<'query>>::Item) + Send + Sync,
{
    #[cfg(feature = "parallel")]
    {
        use rayon::prelude::{ParallelBridge, ParallelIterator};
        query_borrow
            .iter_batched(batch_size)
            .par_bridge()
            .for_each(|batch| batch.for_each(|(entity, components)| for_each(entity, components)));
    }
    #[cfg(not(feature = "parallel"))]
    {
        query_borrow
            .iter()
            .for_each(|(entity, components)| for_each(entity, components));
    }
}

#[cfg(feature = "parallel")]
#[test]
fn thread_pool_installation() {
    use hecs::World;
    use std::{
        thread,
        time::{Duration, Instant},
    };
    struct A(usize);
    struct B(usize);

    let mut world = World::new();
    world.spawn_batch((0..20).map(|_| (B(0),)));
    let a = A(1);
    let thread_pool = rayon::ThreadPoolBuilder::new().build().unwrap();
    let time = Instant::now();
    thread_pool.install(|| {
        batch(&mut world.query::<&mut B>(), 4, |_, b| {
            b.0 += a.0;
            thread::sleep(Duration::from_millis(10));
        });
    });
    #[cfg(not(feature = "parallel"))]
    assert!(time.elapsed() > Duration::from_millis(200));
    #[cfg(feature = "parallel")]
    assert!(time.elapsed() < Duration::from_millis(200));
    for (_, b) in world.query::<&B>().iter() {
        assert_eq!(b.0, 1);
    }
}