kernel 1.1.0

Abstract Reactive Streams
Documentation 
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
107
108
109
110
111
112
113
114
115
116
117
118
119

use reactors::task::{Task, Context, TaskId, T3, Termination};
use reactors::job::Job;
use reactors::system::IO;
use reactors::cps::CpsTask;
use intercore::message::*;
use intercore::bus::{Memory, Channel, send};
use intercore::server::handle_intercore;
use queues::publisher::Publisher;
use std::{thread, time};
use std::ffi::CString;
use handle::{from_raw, into_raw, UnsafeShared, use_};
use reactors::console::Console;
use reactors::selector::{Selector, Async, Pool};
use std::str;

const TASKS_MAX_CNT: usize = 256;

pub struct Scheduler<'a> {
    pub tasks: Vec<T3<Job<'a>>>,
    pub bus: Channel,
    pub queues: Memory,
    pub io: IO,
}

impl<'a> Scheduler<'a> {
    pub fn with_channel(id: usize) -> Self {
        let chan = Channel {
            id: id,
            publisher: Publisher::with_mirror(CString::new(format!("/pub_{}", id)).unwrap(), 88),
            subscribers: Vec::new(),
        };
        Scheduler {
            tasks: Vec::with_capacity(TASKS_MAX_CNT),
            bus: chan,
            io: IO::new(),
            queues: Memory::new(),
        }
    }

    pub fn spawn(&'a mut self, t: Job<'a>, l: Termination, input: Option<&'a str>) -> TaskId {
        let last = self.tasks.len();
        self.tasks.push(T3(t, l));
        self.tasks.last_mut().expect("Scheduler: can't retrieve a task.").0.init(input, last);
        TaskId(last, self.bus.id)
    }

    pub fn exec(&'a mut self, t: TaskId, input: Option<&'a str>) {
        self.tasks.get_mut(t.0).expect("Scheduler: can't retrieve a task.").0.exec(input);
    }

    #[inline]
    fn terminate(&'a mut self, t: Termination, i: usize) {
        if t == Termination::Recursive {
            self.tasks.remove(i);
        }
    }

    #[inline]
    fn poll_bus(&mut self) {
        let x = into_raw(self);
        for s in &from_raw(x).bus.subscribers {
            handle_intercore(from_raw(x), s.recv(), &mut from_raw(x).bus, s);
            s.commit();
        }
    }

    pub fn handle_shell(&mut self, buf: &'a [u8], shell: TaskId) {
        if let Ok(x) = self.io.cmd(buf) {
            send(&self.bus, Message::Exec(shell.0, x.to_string()));
        }
    }

    pub fn hibernate(&mut self) {
        thread::sleep(time::Duration::from_millis(10)); // Green Peace
    }

    #[inline]
    fn poll_tasks(&'a mut self) {
        let a = into_raw(self);
        let l = from_raw(a).tasks.len();
        for i in 1..l {
            from_raw(a).tasks[i].0.poll(Context::Nil, from_raw(a));
        }
    }

    pub fn mem(&mut self) -> UnsafeShared<Memory> {
        unsafe { UnsafeShared::new(&mut self.queues as *mut Memory) }
    }

    pub fn run0(&mut self, input: Option<&'a str>) {
        println!("BSP core {:?}", self.bus.id);
        self.io.spawn(Selector::Rx(Console::new()));
        let x = into_raw(self);
        let shell = from_raw(x).spawn(Job::Cps(CpsTask::new(self.mem())),
                                      Termination::Corecursive,
                                      input);
        send(&self.bus,
             Message::Exec(shell.0, input.unwrap().to_string()));

        loop {
            self.poll_bus();
            match from_raw(x).io.poll() {
                Async::Ready((_, Pool::Raw(buf))) => self.handle_shell(buf, shell),
                _ => (),
            }
            self.hibernate();
            use_(self).poll_tasks();
        }
    }

    pub fn run(&mut self) {
        println!("AP core {:?}", self.bus.id);
        loop {
            self.poll_bus();
            self.hibernate();
            use_(self).poll_tasks();
        }
    }
}