Multiplexers can take function pointers directly

This makes multiplexers slightly more ergonomic.

Due to rust-lang/rust#28229 (functions implement Copy
but not Clone) it's not currently possible to do with closures yet but it's
better than nothing

Author: dking

src/bin/bench.rs

@@ -1,35 +1,43 @@
 extern crate time;
 
 extern crate rs_ducts;
+
 use rs_ducts::multiplex;
 use rs_ducts::map;
 
 use rs_ducts::Pipeline;
 
+// tuneables for different workloads
+const THREAD_COUNT: usize = 1000;
 const WORK_COUNT: u64 = 1000;
 const WORK_FACTOR: u64 = 34;
 const BUFFSIZE: usize = 5;
 
 fn bench_single() {
     let source: Vec<u64> = (1..WORK_COUNT).collect();
 
-    Pipeline::new(source, 5)
-        .map(|x| fib(WORK_FACTOR) + x, BUFFSIZE)
-        .drain();
+    Pipeline::new(source, 5).map(fib_work, BUFFSIZE).drain();
 }
 
+
 fn bench_multi() {
     let source: Vec<u64> = (1..WORK_COUNT).collect();
 
-    let mappers = (0..4)
-        .map(|_| map::Mapper::new(|x| fib(WORK_FACTOR) + x))
-        .collect();
     Pipeline::new(source, BUFFSIZE)
-        .then(multiplex::Multiplex::new(mappers, BUFFSIZE), BUFFSIZE)
+        .then(multiplex::Multiplex::from(map::Mapper::new(fib_work),
+                                         THREAD_COUNT,
+                                         BUFFSIZE),
+              BUFFSIZE)
         .drain();
 }
 
+
 // just something expensive
+fn fib_work(n: u64) -> u64 {
+    fib(WORK_FACTOR) + n
+}
+
+
 fn fib(n: u64) -> u64 {
     if n == 0 || n == 1 {
         1
@@ -38,6 +46,7 @@ fn fib(n: u64) -> u64 {
     }
 }
 
+
 pub fn timeit<F>(name: &str, func: F)
     where F: FnOnce() -> () + Copy
 {

src/lib.rs

@@ -1,13 +1,15 @@
 use std::sync::mpsc;
 use std::thread;
 
+
 #[derive(Debug)]
 pub struct Pipeline<Output>
     where Output: Send + 'static
 {
     rx: mpsc::Receiver<Output>,
 }
 
+
 impl<Output> Pipeline<Output>
     where Output: Send
 {
@@ -73,6 +75,7 @@ impl<Output> Pipeline<Output>
     }
 }
 
+
 impl<Output> IntoIterator for Pipeline<Output>
     where Output: Send
 {
@@ -85,6 +88,7 @@ impl<Output> IntoIterator for Pipeline<Output>
     }
 }
 
+
 pub trait PipelineEntry<In, Out> {
     fn process<I: IntoIterator<Item = In>>(self,
                                            rx: I,
@@ -134,8 +138,22 @@ pub mod map {
             }
         }
     }
+
+    impl<In, Out, Func> Clone for Mapper<In, Out, Func>
+        where Func: Fn(In) -> Out + Copy
+    {
+        fn clone(&self) -> Self {
+            Mapper::new(self.func)
+        }
+    }
+
+    impl<In, Out, Func> Copy for Mapper<In, Out, Func>
+        where Func: Fn(In) -> Out + Copy
+    {
+    }
 }
 
+
 pub mod filter {
     use std::marker::PhantomData;
     use std::sync::mpsc;
@@ -180,6 +198,10 @@ pub mod filter {
 
 
 pub mod multiplex {
+    // work around https://github.com/rust-lang/rust/issues/28229
+    // (functions implement Copy but not Clone)
+    #![cfg_attr(feature="cargo-clippy", allow(expl_impl_clone_on_copy))]
+
     use std::marker::PhantomData;
     use std::sync::mpsc;
     use std::sync::{Arc, Mutex};
@@ -189,7 +211,7 @@ pub mod multiplex {
 
     #[derive(Debug)]
     pub struct Multiplex<In, Out, Entry>
-        where Entry: PipelineEntry<In, Out>
+        where Entry: PipelineEntry<In, Out> + Send
     {
         entries: Vec<Entry>,
         buffsize: usize,
@@ -199,6 +221,14 @@ pub mod multiplex {
         out_: PhantomData<Out>,
     }
 
+    impl<In, Out, Entry> Multiplex<In, Out, Entry>
+        where Entry: PipelineEntry<In, Out> + Send + Copy
+    {
+        pub fn from(entry: Entry, workers: usize, buffsize: usize) -> Self {
+            Self::new((0..workers).map(|_| entry).collect(), buffsize)
+        }
+    }
+
     impl<In, Out, Entry> Multiplex<In, Out, Entry>
         where Entry: PipelineEntry<In, Out> + Send
     {
@@ -324,21 +354,59 @@ mod tests {
         assert_eq!(produced, expect);
     }
 
+    // just something expensive
+    fn fib_work(n: u64) -> u64 {
+        const WORK_FACTOR: u64 = 10;
+        fib(WORK_FACTOR) + n
+    }
+
+    fn fib(n: u64) -> u64 {
+        if n == 0 || n == 1 {
+            1
+        } else {
+            fib(n - 1) + fib(n - 2)
+        }
+    }
+
+    #[test]
+    fn multiplex_map_function() {
+        // we have two signatures for Multiplex, one that takes a function
+        // pointer and one that can take a closure. THis is the function pointer
+        // side
+
+        let buffsize: usize = 10;
+        let workers: usize = 10;
+
+        let source: Vec<u64> = (1..1000).collect();
+        let expect: Vec<u64> =
+            source.clone().into_iter().map(fib_work).collect();
+
+        let pbb: Pipeline<u64> = Pipeline::new(source, buffsize)
+            .then(multiplex::Multiplex::from(map::Mapper::new(fib_work),
+                                             workers,
+                                             buffsize),
+                  buffsize);
+        let mut produced: Vec<u64> = pbb.into_iter().collect();
+
+        produced.sort(); // these may arrive out of order
+        assert_eq!(produced, expect);
+    }
+
     #[test]
-    fn multiplex_map() {
+    fn multiplex_map_closure() {
         let buffsize: usize = 10;
+        let workers: usize = 10;
 
         let source: Vec<i32> = (1..1000).collect();
         let expect: Vec<i32> = source.iter().map(|x| x * 2).collect();
 
         let pbb: Pipeline<i32> = Pipeline::new(source, buffsize)
-            // TOOD multiplex takes a list of PipelineEntry but it would be
-            // nicer if it just took one and was able to clone it
-            .then(
-                multiplex::Multiplex::new(
-                    (0..10).map(|_| map::Mapper::new(|i| i*2)).collect(),
-                    buffsize),
-                buffsize);
+            .then(multiplex::Multiplex::new((0..workers)
+                .map(|_| {
+                        map::Mapper::new(|i| i * 2)
+                    }).collect(),
+                buffsize),
+            buffsize);
         let mut produced: Vec<i32> = pbb.into_iter().collect();
 
         produced.sort(); // these may arrive out of order