actori 0.9.0

Actor framework for Rust
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
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106

/*
 * Ring benchmark inspired by Programming Erlang: Software for a
 * Concurrent World, by Joe Armstrong, Chapter 8.11.2
 *
 * "Write a ring benchmark. Create N processes in a ring. Send a
 * message round the ring M times so that a total of N * M messages
 * get sent. Time how long this takes for different values of N and M."
 */
use actori::*;

use std::env;
use std::time::SystemTime;

/// A payload with a counter
struct Payload(usize);

impl Message for Payload {
    type Result = ();
}

struct Node {
    limit: usize,
    next: Recipient<Payload>,
}

impl Actor for Node {
    type Context = Context<Self>;
}

impl Handler<Payload> for Node {
    type Result = ();

    fn handle(&mut self, msg: Payload, _: &mut Context<Self>) {
        if msg.0 >= self.limit {
            println!("Reached limit of {} (payload was {})", self.limit, msg.0);
            System::current().stop();
            return;
        }
        self.next
            .do_send(Payload(msg.0 + 1))
            .expect("Unable to send payload");
    }
}

fn print_usage_and_exit() -> ! {
    eprintln!("Usage; actori-test <num-nodes> <num-times-message-around-ring>");
    ::std::process::exit(1);
}

#[actori_rt::main]
async fn main() {
    let args = env::args().collect::<Vec<_>>();
    if args.len() < 3 {
        print_usage_and_exit();
    }
    let n_nodes = if let Ok(arg_num_nodes) = args[1].parse::<usize>() {
        if arg_num_nodes <= 1 {
            eprintln!("Number of nodes must be > 1");
            ::std::process::exit(1);
        }
        arg_num_nodes
    } else {
        print_usage_and_exit();
    };

    let n_times = if let Ok(arg_ntimes) = args[2].parse::<usize>() {
        arg_ntimes
    } else {
        print_usage_and_exit()
    };

    let now = SystemTime::now();

    println!("Setting up nodes");
    let node = Node::create(move |ctx| {
        let first_addr = ctx.address();
        let mut prev_addr = Node {
            limit: n_nodes * n_times,
            next: first_addr.recipient(),
        }
        .start();

        for _ in 2..n_nodes {
            prev_addr = Node {
                limit: n_nodes * n_times,
                next: prev_addr.recipient(),
            }
            .start();
        }

        Node {
            limit: n_nodes * n_times,
            next: prev_addr.recipient(),
        }
    });
    node.send(Payload(0)).await.unwrap();

    match now.elapsed() {
        Ok(elapsed) => println!(
            "Time taken: {}.{:06} seconds",
            elapsed.as_secs(),
            elapsed.subsec_micros()
        ),
        Err(e) => println!("An error occured: {:?}", e),
    }
}