From a785f3fc95752dc2cfced5e7ccef710e189acb9c Mon Sep 17 00:00:00 2001
From: Eric Holk <eric.holk@gmail.com>
Date: Wed, 16 May 2012 09:47:00 -0700
Subject: Adding a module with parallel vector operations.

This should go in libstd, but currently resolve bugs make this not
work.
---
 src/libstd/par.rs | 103 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
 src/libstd/std.rc |   2 ++
 2 files changed, 105 insertions(+)
 create mode 100644 src/libstd/par.rs

(limited to 'src/libstd')

diff --git a/src/libstd/par.rs b/src/libstd/par.rs
new file mode 100644
index 00000000000..5160e50fd9a
--- /dev/null
+++ b/src/libstd/par.rs
@@ -0,0 +1,103 @@
+import comm::port;
+import comm::chan;
+import comm::send;
+import comm::recv;
+import task::spawn;
+
+export future;
+export map;
+export alli;
+
+iface future<T: send> {
+    fn get() -> T;
+}
+
+type future_<T: send> = {
+    mut slot : option<T>,
+    port : port<T>,
+};
+
+impl<T: send> of future<T> for future_<T> {
+    fn get() -> T {
+        alt(self.slot) {
+          some(x) { x }
+          none {
+            let x = recv(self.port);
+            self.slot = some(x);
+            x
+          }
+        }
+    }
+}
+
+
+#[doc="Executes a bit of code asynchronously.
+
+Returns a handle that can be used to retrieve the result at your
+leisure."]
+fn future<T: send>(thunk : fn~() -> T) -> future<T> {
+    let p = port();
+    let c = chan(p);
+
+    spawn() {||
+        send(c, thunk());
+    }
+
+    {mut slot: none::<T>, port : p} as future::<T>
+}
+
+#[doc="The maximum number of tasks this module will spawn for a single
+ operationg."]
+const max_tasks : uint = 32u;
+
+#[doc="The minimum number of elements each task will process."]
+const min_granularity : uint = 1024u;
+
+#[doc="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."]
+fn map_slices<A: send, B: send>(xs: [A], f: fn~(uint, [A]) -> B) -> [B] {
+    let len = xs.len();
+    if len < min_granularity {
+        // This is a small vector, fall back on the normal map.
+        [f(0u, xs)]
+    }
+    else {
+        let num_tasks = uint::min(max_tasks, len / min_granularity);
+
+        let items_per_task = len / num_tasks;
+
+        let mut futures = [];
+        let mut base = 0u;
+        while base < len {
+            let slice = vec::slice(xs, base,
+                                   uint::min(len, base + items_per_task));
+            futures += [future() {|copy base|
+                f(base, slice)
+            }];
+            base += items_per_task;
+        }
+
+        futures.map() {|ys|
+            ys.get()
+        }
+    }
+}
+
+#[doc="A parallel version of map."]
+fn map<A: send, B: send>(xs: [A], f: fn~(A) -> B) -> [B] {
+    vec::concat(map_slices(xs) {|_base, slice|
+        map(slice, f)
+    })
+}
+
+#[doc="Returns true if the function holds for all elements in the vector."]
+fn alli<A: send>(xs: [A], f: fn~(uint, A) -> bool) -> bool {
+    vec::all(map_slices(xs) {|base, slice|
+        slice.alli() {|i, x|
+            f(i + base, x)
+        }
+    }) {|x| x }
+}
diff --git a/src/libstd/std.rc b/src/libstd/std.rc
index 40a03ae25c2..f25ceb87c3c 100644
--- a/src/libstd/std.rc
+++ b/src/libstd/std.rc
@@ -19,6 +19,7 @@ export bitv, deque, fun_treemap, list, map, smallintmap, sort, treemap;
 export rope, arena;
 export ebml, dbg, getopts, json, rand, sha1, term, time, prettyprint;
 export test, tempfile, serialization;
+export par;
 
 // General io and system-services modules
 
@@ -58,6 +59,7 @@ mod getopts;
 mod json;
 mod sha1;
 mod md4;
+mod par;
 mod tempfile;
 mod term;
 mod time;
-- 
cgit 1.4.1-3-g733a5