Tasks and Communication

Task Properties

Rust is described by the developers as a language that concentrates on, amoung other things, safety and concurrency. In order to ensure safe, cocurrent computation, Rust makes use of tasks.

A Rust task is an abstraction that has its own memory space and registers, like a process, but doesn't have a processes' associated operating system cost. Tasks split computation like threads, however do not share memory, thus preventing implicit data races and other perplexing bugs that occur when multiple threads access and modify the same data.

Spawning a Task

Tasks are created using the spawn(fn()) function, which creates a new task and executs the input function, fn() in that task. After the function completes, the task is terminated.

Another way to do this is with the do construct. The

1

do expr { block }

constract is syntactic sugar for

1

expr(proc() { block })

so,

1

do spawn { a; }

is the same as:

1

spawn(proc() { a; })

Note: there are plans to remove do from upcoming versions of Rust, so we don't recommend using it, but since you will see it in lots of example code still describe it here.

Here's an example program that spawns tasks:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

fn count(m: &str, n: int) {
    for i in range(1, n) {
        println!("{:s}{:d}", m, i); 
    }
}

fn main() {
    spawn(proc() { count("A", 10); });
    spawn(proc() { count("B", 10); });
    spawn(proc() { count("C", 10); });
}

If you try running this, you will see something like:

B1
C1
A1
C2
B2
A2
B3
...

Task Communication

Tasks do not share memory, and each task has its own memory space. The tasks all run in the same process, though, the memory isloation is provided by the Rust compiler and run-time, not by the operating system with hardware support as is done for an OS process.

Spawning a task and then trying to access mutable variables from outside the scope of the task itself will result in a compiler error. For example,

1
2
3
4
5
6
7

...
fn main() {
    let n = 10;
    spawn(proc() { count("A", n); });
    spawn(proc() { count("B", n); });
    spawn(proc() { count("C", n); });
}

is okay since n is declared as immutable. But if we change the code to,

1
2
3
4

fn main() {
    let mut n = 10;
    spawn(proc() { count("A", n); });
    ...

the compiler will disallow the code with this error:

sharing.rs:9:31: 9:32 error: mutable variables cannot be implicitly captured
sharing.rs:9     spawn(proc() { count("A", n); });

Ports and Channels

This is accomplished through the use of Port<T> and Chan<T> objects. These can be thought of as a link between a spawned task and its parent (the task which spawned it).

A Port/Chan pair is created as a tuple, as demonstrated in the code below. The type of <T> is unrestricted, however the Port and Chan must have matching types.

The transfer of values between tasks is accomplished with the Chan.send(T) function, which puts a value into the (Port<T>, Chan<T>) pair, and the Port.recv() function, which returns the value that had been sent by the Chan .

The following spawns a very simple task to call our plustwo function, and then send the result back to the parent thread.

1
2
3
4
5
6

   let (port, chan): (Port<int>, Chan<int>) = Chan::new();
   spawn(proc() {
        let x = 10;
        chan.send(plustwo(x));
    });
    let new_x = port.recv(); // new_x == 12

A (Port<T>, Chan<T>) pair can only be sent to from a single task. Once Chan.send(T) has been called from a specific task, that task owns the channel. Trying to call Chan.send(T) from multiple tasks will result in a compile time error. Thus, to send objects back and forth between tasks, two different channels are necessary.

Recieving is synchronous. So, once Port.recv() has been called, the task that called it will not continue until a value is sent.

This can lead to deadlocking programs. For example, if a Port.recv() is waiting on a send from another task, but that other task is waiting to receive something from the original task, neither task can make progress.

Multi-Tasking Map

To put everything together, we modify the mapping code from the last part to update each element in a separate task. If the computation needed to produce each new value is expensive, this will potentially divide the mapping time by the number of cores available (with some overhead for spawning the tasks and the communication channels).

Try running the code below, and see if you can figure out why the order of the statements in the mapr function matters.

 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

type LinkedList = Option<~Node>;

struct Node {
   val: int,
   tail: LinkedList
}

fn construct_list(n: int, x: int) -> LinkedList {
    match n {
        0 => { None }
        _ => { Some(~Node{val: x, tail: construct_list(n - 1, x + 1)}) }
    }
}

fn print_list(p: LinkedList) -> ~str {
    match p {
        None => { ~"" }
        Some(node) => { node.val.to_str() + ", " 
                        + print_list(node.tail) }
    }
}

trait Map {
   fn mapr(&mut self, fn(int) -> int);
}

impl Map for LinkedList {
    fn mapr(&mut self, f: fn(int) -> int) {
         match(*self) {
            None => { }
            Some(ref mut current) => { 
               let (port, chan) : (Port<int>, Chan<int>) = Chan::new();
               let val = current.val; // Can't capture current
               spawn(proc() { chan.send(f(val)); });
               current.tail.mapr(f); // why here?
               current.val = port.recv();
            }
        } 
    } 
}

fn expensive_inc(n: int) -> int { 
   let mut a = 1;
   println!("starting inc: {:d}", n);
   for _ in range(0, 10000) {
        for _ in range(0, 1000000) {
           a = a + 1;
        }
   }
   
   println!("finished inc: {:d} ({:d})", n, a);
   n + 1 
}

fn main() {
    let mut p : LinkedList = construct_list(5, 10);
    p.mapr(expensive_inc);
    println!("List: {:s}", print_list(p.clone()));
}