rtipc 0.9.0

Real-Time IPC, based on a zero-copy, wait-free circular message queue implementation
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
120
121
122

use nix::sys::socket::Backlog;

use std::time::Duration;

use rtipc::ChannelGroup;
use rtipc::Consumer;
use rtipc::PopResult;
use rtipc::Producer;

use rtipc::TryPushResult;

use rtipc::Server;

use crate::common::CommandId;
use crate::common::MsgCommand;
use crate::common::MsgEvent;
use crate::common::MsgResponse;

use crate::common::wait_pollin;

mod common;

struct App {
    command: Consumer<MsgCommand>,
    response: Producer<MsgResponse>,
    event: Producer<MsgEvent>,
}

fn print_group(grp: &ChannelGroup) {
    let attr = grp.get_attr();
    let grp_info = str::from_utf8(attr.info.iter().as_slice()).unwrap();
    let cmd_info = str::from_utf8(&attr.consumers.get(0).unwrap().info).unwrap();
    let rsp_info = str::from_utf8(&attr.producers.get(0).unwrap().info).unwrap();
    let evt_info = str::from_utf8(&attr.producers.get(1).unwrap().info).unwrap();
    println!(
        "server received request grp={} cmd={} rsp={} evt={}",
        grp_info, cmd_info, rsp_info, evt_info
    );
}

impl App {
    pub fn new(mut grp: ChannelGroup) -> Self {
        print_group(&grp);
        let command = grp.acquire_consumer(0).unwrap();
        let response = grp.acquire_producer(0).unwrap();
        let event = grp.acquire_producer(1).unwrap();

        Self {
            command,
            response,
            event,
        }
    }
    fn run(&mut self) {
        let mut run = true;
        let mut cnt = 0;

        while run {
            let eventfd = self.command.eventfd().unwrap();
            let _ = wait_pollin(eventfd, Duration::from_millis(10));
            match self.command.pop().unwrap() {
                PopResult::NoMessage => continue,
                PopResult::NoNewMessage => continue,
                PopResult::Success => {}
                PopResult::SuccessMessagesDiscarded => {}
            };
            let cmd = self.command.current_message().unwrap();
            self.response.current_message().id = cmd.id;
            println!("server received command: id = {}", cmd.id);

            let cmdid: CommandId = unsafe { ::std::mem::transmute(cmd.id) };
            self.response.current_message().result = match cmdid {
                CommandId::Hello => 0,
                CommandId::Stop => {
                    run = false;
                    0
                }
                CommandId::SendEvent => self.send_events(
                    unsafe { cmd.args.send.id },
                    unsafe { cmd.args.send.num },
                    unsafe { cmd.args.send.force },
                ),
                CommandId::Div => {
                    let (err, res) = self.div(unsafe { cmd.args.div.divident }, unsafe {
                        cmd.args.div.divisor
                    });
                    self.response.current_message().data.quotient = res;
                    err
                }
            };
            self.response.force_push().unwrap();

            cnt += 1;
        }
    }
    fn send_events(&mut self, id: u32, num: u32, force: bool) -> i32 {
        for i in 0..num {
            let event = self.event.current_message();
            event.id = id;
            event.nr = i;
            if force {
                self.event.force_push().unwrap();
            } else {
                if self.event.try_push().unwrap() == TryPushResult::QueueFull {
                    return i as i32;
                }
            }
        }
        num as i32
    }
    fn div(&mut self, a: f64, b: f64) -> (i32, f64) {
        if b == 0.0 { (-1, 0.0) } else { (0, a / b) }
    }
}

fn main() {
    let backlog = Backlog::new(1).unwrap();
    let server = Server::new("rtipc.sock", backlog).unwrap();
    let grp = server.conditional_accept(|_| true).unwrap();
    let mut app = App::new(grp);
    app.run();
}