September Morn Update

Task/Active-object/Groovy/active-object.groovy

@@ -0,0 +1,70 @@
+/**
+ * Integrates input function K over time
+ * S + (t1 - t0) * (K(t1) + K(t0)) / 2
+ */
+class Integrator {
+    interface Function {
+        double apply(double timeSinceStartInSeconds)
+    }
+
+    private final long start
+    private volatile boolean running
+
+    private Function func
+    private double t0
+    private double v0
+    private double sum
+
+    Integrator(Function func) {
+        this.start = System.nanoTime()
+        setFunc(func)
+        new Thread({ this.&integrate() }).start()
+    }
+
+    void setFunc(Function func) {
+        this.func = func
+        def temp = func.apply(0.0.toDouble())
+        v0 = temp
+        t0 = 0.0.doubleValue()
+    }
+
+    double getOutput() {
+        return sum
+    }
+
+    void stop() {
+        running = false
+    }
+
+    private void integrate() {
+        running = true
+        while (running) {
+            try {
+                Thread.sleep(1)
+                update()
+            } catch (InterruptedException ignored) {
+                return
+            }
+        }
+    }
+
+    private void update() {
+        double t1 = (System.nanoTime() - start) / 1.0e9
+        double v1 = func.apply(t1)
+        double rect = (t1 - t0) * (v0 + v1) / 2.0
+        this.sum += rect
+        t0 = t1
+        v0 = v1
+    }
+
+    static void main(String[] args) {
+        Integrator integrator = new Integrator({ t -> Math.sin(Math.PI * t) })
+        Thread.sleep(2000)
+
+        integrator.setFunc({ t -> 0.0.toDouble() })
+        Thread.sleep(500)
+
+        integrator.stop()
+        System.out.println(integrator.getOutput())
+    }
+}

Task/Active-object/Perl-6/active-object.pl6

@@ -0,0 +1,45 @@
+class Integrator {
+    has $.f is rw = sub ($t) { 0 };
+    has $.now is rw;
+    has $.value is rw = 0;
+    has $.integrator is rw;
+
+    method init() {
+        self.value = &(self.f)(0);
+        self.integrator = Thread.new(
+            :code({)
+                loop {
+                    my $t1 = now;
+                    self.value += (&(self.f)(self.now) + &(self.f)($t1)) * ($t1 - self.now) / 2;
+                    self.now = $t1;
+                    sleep .001;
+                }
+            }),
+            :app_lifetime(True)
+        ).run
+    }
+
+    method Input (&f-of-t) {
+        self.f = &f-of-t;
+        self.now = now;
+        self.init;
+    }
+
+    method Output { self.value }
+}
+
+my $a = Integrator.new;
+
+$a.Input( sub ($t) { sin(2 * π * .5 * $t) } );
+
+say "Initial value: ", $a.Output;
+
+sleep 2;
+
+say "After 2 seconds: ", $a.Output;
+
+$a.Input( sub ($t) { 0 } );
+
+sleep .5;
+
+say "f(0): ", $a.Output;

Task/Active-object/Phix/active-object.phix

@@ -0,0 +1,63 @@
+sequence x = {}
+enum TERMINATE, INTERVAL, KFUN, VALUE, T0, K0, ID, ISIZE=$
+integer xlock = init_cs()
+
+function zero(atom /*t*/) return 0 end function
+function sine(atom t) return sin(2*PI*0.5*t) end function
+
+procedure update(integer dx)
+    enter_cs(xlock)
+    atom t1 = time(),
+         k1 = call_func(x[dx][KFUN],{t1})
+    x[dx][VALUE] += (k1 + x[dx][K0]) * (t1 - x[dx][T0]) / 2
+    x[dx][T0] = t1
+    x[dx][K0] = k1
+    leave_cs(xlock)
+end procedure
+
+procedure tick(integer dx)
+    while not x[dx][TERMINATE] do
+        sleep(x[dx][INTERVAL])
+        update(dx)
+    end while
+end procedure
+
+function new_integrator(integer rid, atom interval)
+    x = append(x,repeat(0,ISIZE))
+    integer dx = length(x)
+    x[dx][TERMINATE] = false
+    x[dx][INTERVAL] = interval
+    x[dx][KFUN] = rid
+    x[dx][T0] = time()
+    update(dx)
+    x[dx][ID] = create_thread(routine_id("tick"),{dx})
+    return dx
+end function
+
+procedure set_input(integer dx, rid)
+    enter_cs(xlock)
+    x[dx][KFUN] = rid
+    x[dx][K0] = 0
+    leave_cs(xlock)
+end procedure
+
+function get_output(integer dx)
+    enter_cs(xlock)
+    atom v = x[dx][VALUE]
+    leave_cs(xlock)
+    return v
+end function
+
+procedure stop_integrator(integer dx)
+    x[dx][TERMINATE] = true
+    wait_thread(x[dx][ID])
+end procedure
+
+puts(1,"")
+integer dx = new_integrator(routine_id("sine"),0.01)
+sleep(2)
+printf(1,"%f\n",get_output(dx))
+set_input(dx,routine_id("zero"))
+sleep(0.5)
+printf(1,"%f\n",get_output(dx))
+stop_integrator(dx)

Task/Active-object/Rust/active-object.rust

@@ -0,0 +1,172 @@
+#![feature(mpsc_select)]
+
+extern crate num;
+extern crate schedule_recv;
+
+use num::traits::Zero;
+use num::Float;
+use schedule_recv::periodic_ms;
+use std::f64::consts::PI;
+use std::ops::Mul;
+use std::sync::mpsc::{self, SendError, Sender};
+use std::sync::{Arc, Mutex};
+use std::thread;
+use std::time::Duration;
+
+pub type Actor<S> = Sender<Box<Fn(u32) -> S + Send>>;
+pub type ActorResult<S> = Result<(), SendError<Box<Fn(u32) -> S + Send>>>;
+
+/// Rust supports both shared-memory and actor models of concurrency, and the `Integrator` utilizes
+/// both.  We use an `Actor` to send the `Integrator` new functions, while we use a `Mutex`
+/// (shared-memory concurrency) to hold the result of the integration.
+///
+/// Note that these are not the only options here--there are many, many ways you can deal with
+/// concurrent access.  But when in doubt, a plain old `Mutex` is often a good bet.  For example,
+/// this might look like a good situation for a `RwLock`--after all, there's no reason for a read
+/// in the main task to block writes.  Unfortunately, unless you have significantly more reads than
+/// writes (which is certainly not the case here), a `Mutex` will usually outperform a `RwLock`.
+pub struct Integrator<S: 'static, T: Send> {
+    input: Actor<S>,
+    output: Arc<Mutex<T>>,
+}
+
+/// In Rust, time durations are strongly typed.  This is usually exactly what you want, but for a
+/// problem like this--where the integrated value has unusual (unspecified?) units--it can actually
+/// be a bit tricky.  Right now, `Duration`s can only be multiplied or divided by `i32`s, so in
+/// order to be able to actually do math with them we say that the type parameter `S` (the result
+/// of the function being integrated) must yield `T` (the type of the integrated value) when
+/// multiplied by `f64`.  We could possibly replace `f64` with a generic as well, but it would make
+/// things a bit more complex.
+impl<S, T> Integrator<S, T>
+where
+    S: Mul<f64, Output = T> + Float + Zero,
+    T: 'static + Clone + Send + Float,
+{
+    pub fn new(frequency: u32) -> Integrator<S, T> {
+        // We create a pipe allowing functions to be sent from tx (the sending end) to input (the
+        // receiving end).  In order to change the function we are integrating from the task in
+        // which the Integrator lives, we simply send the function through tx.
+        let (tx, input) = mpsc::channel();
+        // The easiest way to do shared-memory concurrency in Rust is to use atomic reference
+        // counting, or Arc, around a synchronized type (like Mutex<T>).  Arc gives you a guarantee
+        // that memory will not be freed as long as there is at least one reference to it.
+        // It is similar to C++'s shared_ptr, but it is guaranteed to be safe and is never
+        // incremented unless explicitly cloned (by default, it is moved).
+        let s: Arc<Mutex<T>> = Arc::new(Mutex::new(Zero::zero()));
+        let integrator = Integrator {
+            input: tx,
+            // Here is the aforementioned clone.  We have to do it before s enters the closure,
+            // because once that happens it is moved into the closure (and later, the new task) and
+            // becomes inaccessible to the outside world.
+            output: Arc::clone(&s),
+        };
+        thread::spawn(move || -> () {
+            // The frequency is how often we want to "tick" as we update our integrated total.  In
+            // Rust, timers can yield Receivers that are periodically notified with an empty
+            // message (where the period is the frequency).  This is useful because it lets us wait
+            // on either a tick or another type of message (in this case, a request to change the
+            // function we are integrating).
+            let periodic = periodic_ms(frequency);
+            let mut t = 0;
+            let mut k: Box<Fn(u32) -> S + Send> = Box::new(|_| Zero::zero());
+            let mut k_0: S = Zero::zero();
+            loop {
+                // Here's the selection we talked about above.  Note that we are careful to call
+                // the *non*-failing function, recv(), here.  The reason we do this is because
+                // recv() will return Err when the sending end of a channel is dropped.  While
+                // this is unlikely to happen for the timer (so again, you could argue for failure
+                // there), it's normal behavior for the sending end of input to be dropped, since
+                // it just happens when the Integrator falls out of scope.  So we handle it cleanly
+                // and break out of the loop, rather than failing.
+                select! {
+                    res = periodic.recv() => match res {
+                        Ok(_) => {
+                            t += frequency;
+                            let k_1: S = k(t);
+                            // Rust Mutexes are a bit different from Mutexes in many other
+                            // languages, in that the protected data is actually encapsulated by
+                            // the Mutex.  The reason for this is that Rust is actually capable of
+                            // enforcing (via its borrow checker) the invariant that the contents
+                            // of a Mutex may only be read when you have acquired its lock.  This
+                            // is enforced by way of a MutexGuard, the return value of lock(),
+                            // which implements some special traits (Deref and DerefMut) that allow
+                            // access to the inner element "through" the guard.  The element so
+                            // acquired has a lifetime bounded by that of the MutexGuard, the
+                            // MutexGuard can only be acquired by taking a lock, and the only way
+                            // to release the lock is by letting the MutexGuard fall out of scope,
+                            // so it's impossible to access the data incorrectly.  There are some
+                            // additional subtleties around the actual implementation, but that's
+                            // the basic idea.
+                            let mut s = s.lock().unwrap();
+                            *s = *s + (k_1 + k_0) * (f64::from(frequency) / 2.);
+                            k_0 = k_1;
+                        }
+                        Err(_) => break,
+                    },
+                    res = input.recv() => match res {
+                        Ok(k_new) => k = k_new,
+                        Err(_) => break,
+                    }
+                }
+            }
+        });
+        integrator
+    }
+
+    pub fn input(&self, k: Box<Fn(u32) -> S + Send>) -> ActorResult<S> {
+        // The meat of the work is done in the other thread, so to set the
+        // input we just send along the Sender we set earlier...
+        self.input.send(k)
+    }
+
+    pub fn output(&self) -> T {
+        // ...and to read the input, we simply acquire a lock on the output Mutex and return a
+        // copy. Why do we have to copy it?  Because, as mentioned above, Rust won't let us
+        // retain access to the interior of the Mutex unless we have possession of its lock.  There
+        // are ways and circumstances in which one can avoid this (e.g. by using atomic types) but
+        // a copy is a perfectly reasonable solution as well, and a lot easier to reason about :)
+        *self.output.lock().unwrap()
+    }
+}
+
+/// This function is fairly straightforward.  We create the integrator, set its input function k(t)
+/// to 2pi * f * t, and then wait as described in the Rosetta stone problem.
+fn integrate() -> f64 {
+    let object = Integrator::new(10);
+    object
+        .input(Box::new(|t: u32| {
+            let two_seconds_ms = 2 * 1000;
+            let f = 1. / f64::from(two_seconds_ms);
+            (2. * PI * f * f64::from(t)).sin()
+        }))
+        .expect("Failed to set input");
+    thread::sleep(Duration::from_secs(2));
+    object.input(Box::new(|_| 0.)).expect("Failed to set input");
+    thread::sleep(Duration::from_millis(500));
+    object.output()
+}
+
+fn main() {
+    println!("{}", integrate());
+}
+
+/// Will fail on a heavily loaded machine
+#[test]
+#[ignore]
+fn solution() {
+    // We should just be able to call integrate, but can't represent the closure properly due to
+    // rust-lang/rust issue #17060 if we make frequency or period a variable.
+    // FIXME(pythonesque): When unboxed closures are fixed, fix integrate() to take two arguments.
+    let object = Integrator::new(10);
+    object
+        .input(Box::new(|t: u32| {
+            let two_seconds_ms = 2 * 1000;
+            let f = 1. / (two_seconds_ms / 10) as f64;
+            (2. * PI * f * t as f64).sin()
+        }))
+        .expect("Failed to set input");
+    thread::sleep(Duration::from_millis(200));
+    object.input(Box::new(|_| 0.)).expect("Failed to set input");
+    thread::sleep(Duration::from_millis(100));
+    assert_eq!(object.output() as u32, 0)
+}