polling 3.3.2

Portable interface to epoll, kqueue, event ports, and IOCP
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
123
124
125
126
127
128
129
130
131

use std::io::{self, Write};
use std::net::{TcpListener, TcpStream};
use std::thread;
use std::time::Duration;

use easy_parallel::Parallel;
use polling::{Event, Events, Poller};

#[test]
fn concurrent_add() -> io::Result<()> {
    let (reader, mut writer) = tcp_pair()?;
    let poller = Poller::new()?;

    let mut events = Events::new();

    let result = Parallel::new()
        .add(|| {
            poller.wait(&mut events, None)?;
            Ok(())
        })
        .add(|| {
            thread::sleep(Duration::from_millis(100));
            unsafe {
                poller.add(&reader, Event::readable(0))?;
            }
            writer.write_all(&[1])?;
            Ok(())
        })
        .run()
        .into_iter()
        .collect::<io::Result<()>>();

    poller.delete(&reader)?;
    result?;

    assert_eq!(events.len(), 1);
    assert_eq!(
        events.iter().next().unwrap().with_no_extra(),
        Event::readable(0)
    );

    Ok(())
}

#[test]
fn concurrent_modify() -> io::Result<()> {
    let (reader, mut writer) = tcp_pair()?;
    let poller = Poller::new()?;
    unsafe {
        poller.add(&reader, Event::none(0))?;
    }

    let mut events = Events::new();

    Parallel::new()
        .add(|| {
            poller.wait(&mut events, Some(Duration::from_secs(10)))?;
            Ok(())
        })
        .add(|| {
            thread::sleep(Duration::from_millis(100));
            poller.modify(&reader, Event::readable(0))?;
            writer.write_all(&[1])?;
            Ok(())
        })
        .run()
        .into_iter()
        .collect::<io::Result<()>>()?;

    assert_eq!(events.len(), 1);
    assert_eq!(
        events.iter().next().unwrap().with_no_extra(),
        Event::readable(0)
    );

    Ok(())
}

#[cfg(unix)]
#[test]
fn concurrent_interruption() -> io::Result<()> {
    struct MakeItSend<T>(T);
    unsafe impl<T> Send for MakeItSend<T> {}

    let (reader, _writer) = tcp_pair()?;
    let poller = Poller::new()?;
    unsafe {
        poller.add(&reader, Event::none(0))?;
    }

    let mut events = Events::new();
    let events_borrow = &mut events;
    let (sender, receiver) = std::sync::mpsc::channel();

    Parallel::new()
        .add(move || {
            // Register a signal handler so that the syscall is actually interrupted. A signal that
            // is ignored by default does not cause an interrupted syscall.
            signal_hook::flag::register(signal_hook::consts::signal::SIGURG, Default::default())?;

            // Signal to the other thread how to send a signal to us
            sender
                .send(MakeItSend(unsafe { libc::pthread_self() }))
                .unwrap();

            poller.wait(events_borrow, Some(Duration::from_secs(1)))?;
            Ok(())
        })
        .add(move || {
            let MakeItSend(target_thread) = receiver.recv().unwrap();
            thread::sleep(Duration::from_millis(100));
            assert_eq!(0, unsafe {
                libc::pthread_kill(target_thread, libc::SIGURG)
            });
            Ok(())
        })
        .run()
        .into_iter()
        .collect::<io::Result<()>>()?;

    assert_eq!(events.len(), 0);

    Ok(())
}

fn tcp_pair() -> io::Result<(TcpStream, TcpStream)> {
    let listener = TcpListener::bind("127.0.0.1:0")?;
    let a = TcpStream::connect(listener.local_addr()?)?;
    let (b, _) = listener.accept()?;
    Ok((a, b))
}