mio 0.1.0 

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
132
133
134
135
136
137
138
139
140

use std::mem;
use nix::fcntl::Fd;
use nix::sys::event::*;
use error::{MioResult, MioError};
use os::IoDesc;
use poll::{IoEvent, IOREADABLE, IOWRITABLE, IOERROR, IOHINTED, IOHUPHINT};

pub struct Selector {
    kq: Fd,
    changes: Events
}

impl Selector {
    pub fn new() -> MioResult<Selector> {
        Ok(Selector {
            kq: try!(kqueue().map_err(MioError::from_sys_error)),
            changes: Events::new()
        })
    }

    pub fn select(&mut self, evts: &mut Events, timeout_ms: uint) -> MioResult<()> {
        let cnt = try!(kevent(self.kq, self.changes.as_slice(),
                              evts.as_mut_slice(), timeout_ms)
                                  .map_err(MioError::from_sys_error));

        self.changes.len = 0;

        evts.len = cnt;
        Ok(())
    }

    pub fn register(&mut self, io: &IoDesc, token: uint) -> MioResult<()> {
        let flag = EV_ADD | EV_CLEAR;

        try!(self.ev_push(io, EVFILT_READ, flag, FilterFlag::empty(), token));
        try!(self.ev_push(io, EVFILT_WRITE, flag, FilterFlag::empty(), token));

        Ok(())
    }

    // Queues an event change. Events will get submitted to the OS on the next
    // call to select or when the change buffer fills up.
    fn ev_push(&mut self,
               io: &IoDesc,
               filter: EventFilter,
               flags: EventFlag,
               fflags: FilterFlag,
               token: uint) -> MioResult<()> {

        // If the change buffer is full, flush it
        try!(self.maybe_flush_changes());

        let idx = self.changes.len;
        let ev = &mut self.changes.events[idx];

        ev_set(ev, io.fd as uint, filter, flags, fflags, token);

        self.changes.len += 1;

        Ok(())
    }

    fn maybe_flush_changes(&mut self) -> MioResult<()> {
        if self.changes.is_full() {
            try!(kevent(self.kq, self.changes.as_slice(), &mut [], 0)
                    .map_err(MioError::from_sys_error));
            self.changes.len = 0;
        }

        Ok(())
    }
}

pub struct Events {
    len: uint,
    events: [KEvent, ..1024]
}

impl Events {
    pub fn new() -> Events {
        Events {
            len: 0,
            events: unsafe { mem::uninitialized() }
        }
    }

    #[inline]
    pub fn len(&self) -> uint {
        self.len
    }

    // TODO: We will get rid of this eventually in favor of an iterator
    #[inline]
    pub fn get(&self, idx: uint) -> IoEvent {
        if idx >= self.len {
            panic!("invalid index");
        }

        let ev = &self.events[idx];
        let token = ev.udata;

        // When the read end of the socket is closed, EV_EOF is set on the
        // flags, and fflags contains the error, if any.

        let mut kind = IOHINTED;

        if ev.filter == EVFILT_READ {
            kind = kind | IOREADABLE;
        } else if ev.filter == EVFILT_WRITE {
            kind = kind | IOWRITABLE;
        } else {
            unimplemented!();
        }

        if ev.flags.contains(EV_EOF) {
            kind = kind | IOHUPHINT;

            // When the read end of the socket is closed, EV_EOF is set on
            // flags, and fflags contains the error if there is one.
            if !ev.fflags.is_empty() {
                kind = kind | IOERROR;
            }
        }

        IoEvent::new(kind, token)
    }

    #[inline]
    fn is_full(&self) -> bool {
        self.len == self.events.len()
    }

    fn as_slice(&self) -> &[KEvent] {
        self.events.slice_to(self.len)
    }

    fn as_mut_slice(&mut self) -> &mut [KEvent] {
        self.events.as_mut_slice()
    }
}