par.rs 4.2 KB
Newer Older
1
import future_spawn = future::spawn;
E
Eric Holk 已提交
2

3
export map, mapi, alli, any, mapi_factory;
E
Eric Holk 已提交
4

5 6 7 8
/**
 * The maximum number of tasks this module will spawn for a single
 * operation.
 */
E
Eric Holk 已提交
9 10
const max_tasks : uint = 32u;

11
/// The minimum number of elements each task will process.
E
Eric Holk 已提交
12 13
const min_granularity : uint = 1024u;

14 15 16 17 18 19 20
/**
 * An internal helper to map a function over a large vector and
 * return the intermediate results.
 *
 * This is used to build most of the other parallel vector functions,
 * like map or alli.
 */
21
fn map_slices<A: copy send, B: copy send>(
22 23 24
    xs: ~[A],
    f: fn() -> fn~(uint, v: &[A]) -> B)
    -> ~[B] {
E
Eric Holk 已提交
25 26 27

    let len = xs.len();
    if len < min_granularity {
28
        log(info, ~"small slice");
E
Eric Holk 已提交
29
        // This is a small vector, fall back on the normal map.
30
        ~[f()(0u, xs)]
E
Eric Holk 已提交
31 32
    }
    else {
33
        let num_tasks = uint::min(max_tasks, len / min_granularity);
E
Eric Holk 已提交
34 35 36

        let items_per_task = len / num_tasks;

37
        let mut futures = ~[];
E
Eric Holk 已提交
38
        let mut base = 0u;
39
        log(info, ~"spawning tasks");
E
Eric Holk 已提交
40
        while base < len {
41
            let end = uint::min(len, base + items_per_task);
T
Tim Chevalier 已提交
42
            // FIXME: why is the ::<A, ()> annotation required here? (#2617)
43
            do vec::as_buf::<A, ()>(xs) |p, _len| {
44
                let f = f();
B
Brian Anderson 已提交
45
                let f = do future_spawn() |copy base| {
E
Eric Holk 已提交
46 47 48 49
                    unsafe {
                        let len = end - base;
                        let slice = (ptr::offset(p, base),
                                     len * sys::size_of::<A>());
P
Paul Stansifer 已提交
50
                        log(info, fmt!("pre-slice: %?", (base, slice)));
51
                        let slice : &[A] =
E
Eric Holk 已提交
52
                            unsafe::reinterpret_cast(slice);
P
Paul Stansifer 已提交
53 54
                        log(info, fmt!("slice: %?",
                                       (base, vec::len(slice), end - base)));
E
Eric Holk 已提交
55
                        assert(vec::len(slice) == end - base);
56
                        f(base, slice)
E
Eric Holk 已提交
57
                    }
58 59
                };
                vec::push(futures, f);
E
Eric Holk 已提交
60 61 62
            };
            base += items_per_task;
        }
63
        log(info, ~"tasks spawned");
E
Eric Holk 已提交
64

P
Paul Stansifer 已提交
65
        log(info, fmt!("num_tasks: %?", (num_tasks, futures.len())));
E
Eric Holk 已提交
66 67
        assert(num_tasks == futures.len());

B
Brian Anderson 已提交
68
        let r = do futures.map() |ys| {
E
Eric Holk 已提交
69 70 71 72 73 74 75
            ys.get()
        };
        assert(r.len() == futures.len());
        r
    }
}

76
/// A parallel version of map.
77
fn map<A: copy send, B: copy send>(xs: ~[A], f: fn~(A) -> B) -> ~[B] {
B
Brian Anderson 已提交
78
    vec::concat(map_slices(xs, || {
79
        fn~(_base: uint, slice : &[A], copy f) -> ~[B] {
80
            vec::map(slice, |x| f(x))
81
        }
82
    }))
E
Eric Holk 已提交
83 84
}

85
/// A parallel version of mapi.
86 87
fn mapi<A: copy send, B: copy send>(xs: ~[A],
                                    f: fn~(uint, A) -> B) -> ~[B] {
B
Brian Anderson 已提交
88
    let slices = map_slices(xs, || {
89
        fn~(base: uint, slice : &[A], copy f) -> ~[B] {
B
Brian Anderson 已提交
90
            vec::mapi(slice, |i, x| {
91
                f(i + base, x)
92
            })
93
        }
94
    });
95 96 97 98 99 100
    let r = vec::concat(slices);
    log(info, (r.len(), xs.len()));
    assert(r.len() == xs.len());
    r
}

101 102 103 104 105 106
/**
 * A parallel version of mapi.
 *
 * In this case, f is a function that creates functions to run over the
 * inner elements. This is to skirt the need for copy constructors.
 */
107
fn mapi_factory<A: copy send, B: copy send>(
108
    xs: ~[A], f: fn() -> fn~(uint, A) -> B) -> ~[B] {
B
Brian Anderson 已提交
109
    let slices = map_slices(xs, || {
110
        let f = f();
111
        fn~(base: uint, slice : &[A], move f) -> ~[B] {
B
Brian Anderson 已提交
112
            vec::mapi(slice, |i, x| {
113
                f(i + base, x)
114
            })
E
Eric Holk 已提交
115
        }
116
    });
E
Eric Holk 已提交
117 118 119 120 121 122
    let r = vec::concat(slices);
    log(info, (r.len(), xs.len()));
    assert(r.len() == xs.len());
    r
}

123
/// Returns true if the function holds for all elements in the vector.
124
fn alli<A: copy send>(xs: ~[A], f: fn~(uint, A) -> bool) -> bool {
B
Brian Anderson 已提交
125
    do vec::all(map_slices(xs, || {
126
        fn~(base: uint, slice : &[A], copy f) -> bool {
B
Brian Anderson 已提交
127
            vec::alli(slice, |i, x| {
E
Eric Holk 已提交
128
                f(i + base, x)
129
            })
E
Eric Holk 已提交
130
        }
B
Brian Anderson 已提交
131
    })) |x| { x }
E
Eric Holk 已提交
132 133
}

134
/// Returns true if the function holds for any elements in the vector.
135
fn any<A: copy send>(xs: ~[A], f: fn~(A) -> bool) -> bool {
B
Brian Anderson 已提交
136
    do vec::any(map_slices(xs, || {
137
        fn~(_base : uint, slice: &[A], copy f) -> bool {
138
            vec::any(slice, |x| f(x))
E
Eric Holk 已提交
139
        }
B
Brian Anderson 已提交
140
    })) |x| { x }
E
Eric Holk 已提交
141
}