/usr/lib/python3/dist-packages/checkbox/lib/selector.py is in python3-checkbox 0.17.6-0ubuntu6.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | #
# This file is part of Checkbox.
#
# Copyright 2011 Canonical Ltd.
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
#
import os
import errno
import select
from checkbox.lib.enum import Enum
__all__ = ["SelectorIO", "SelectorState", "Selector"]
SelectorIO = Enum(
"READ",
"WRITE",
"EXCEPT")
SelectorState = Enum(
"VIRGIN",
"READY",
"TIMED_OUT",
"SIGNALED",
"FAILED")
class Selector:
__slots__ = ("_read_fds", "_write_fds", "_except_fds",
"_save_read_fds", "_save_write_fds", "_save_except_fds",
"_fd_set_size", "_timeout", "_state", "_errno")
_fd_select_size = -1
_cached_read_fds = None
_cached_save_read_fds = None
_cached_write_fds = None
_cached_save_write_fds = None
_cached_except_fds = None
_cached_save_except_fds = None
def __init__(self):
if Selector._cached_read_fds:
self._read_fds = Selector._cached_read_fds
self._write_fds = Selector._cached_write_fds
self._except_fds = Selector._cached_except_fds
self._save_read_fds = Selector._cached_save_read_fds
self._save_write_fds = Selector._cached_save_write_fds
self._save_except_fds = Selector._cached_save_except_fds
Selector._cached_read_fds = None
Selector._cached_write_fds = None
Selector._cached_except_fds = None
Selector._cached_save_read_fds = None
Selector._cached_save_write_fds = None
Selector._cached_save_except_fds = None
else:
self._read_fds = []
self._write_fds = []
self._except_fds = []
self._save_read_fds = []
self._save_write_fds = []
self._save_except_fds = []
self.reset()
def __del__(self):
if Selector._cached_read_fds is None:
Selector._cached_read_fds = self._read_fds
Selector._cached_write_fds = self._write_fds
Selector._cached_except_fds = self._except_fds
Selector._cached_save_read_fds = self._save_read_fds
Selector._cached_save_write_fds = self._save_write_fds
Selector._cached_save_except_fds = self._save_except_fds
def reset(self):
self._errno = 0
self._state = SelectorState.VIRGIN
self._timeout = None
self._save_read_fds = []
self._save_write_fds = []
self._save_except_fds = []
@staticmethod
def get_fd_select_size():
if Selector._fd_select_size == -1:
try:
Selector._fd_select_size = os.sysconf("SC_OPEN_MAX")
except (AttributeError, ValueError):
Selector._fd_select_size = 1024
return Selector._fd_select_size
def get_errno(self):
return self._errno
def get_errstr(self):
return os.strerror(self._errno)
def set_timeout(self, timeout):
self._timeout = timeout
def unset_timeout(self):
self._timeout = None
def has_timed_out(self):
return self._state == SelectorState.TIMED_OUT
def has_signaled(self):
return self._state == SelectorState.SIGNALED
def has_failed(self):
return self._state == SelectorState.FAILED
def has_ready(self):
return self._state == SelectorState.READY
def add_fd(self, fd, interest):
if fd < 0 or fd >= Selector.get_fd_select_size():
raise Exception("File descriptor %d outside of range 0-%d"
% (fd, Selector._fd_select_size))
if interest == SelectorIO.READ:
self._save_read_fds.append(fd)
elif interest == SelectorIO.WRITE:
self._save_write_fds.append(fd)
elif interest == SelectorIO.EXCEPT:
self._save_except_fds.append(fd)
def remove_fd(self, fd, interest):
if fd < 0 or fd >= Selector.get_fd_select_size():
raise Exception("File descriptor %d outside of range 0-%d"
% (fd, Selector._fd_select_size))
if interest == SelectorIO.READ:
self._save_read_fds.remove(fd)
elif interest == SelectorIO.WRITE:
self._save_write_fds.remove(fd)
elif interest == SelectorIO.EXCEPT:
self._save_except_fds.remove(fd)
def execute(self):
try:
self._read_fds, self._write_fds, self._except_fds = select.select(
self._save_read_fds, self._save_write_fds,
self._save_except_fds, self._timeout)
except select.error as e:
self._errno = e.errno
if e.errno == errno.EINTR:
self._state = SelectorState.SIGNALED
else:
self._state = SelectorState.FAILED
return
# Just in case
self._errno = 0
if not self._read_fds \
and not self._write_fds \
and not self._except_fds:
self._state = SelectorState.TIMED_OUT
else:
self._state = SelectorState.READY
def is_fd_ready(self, fd, interest):
if self._state != SelectorState.READY \
and self._state != SelectorState.TIMED_OUT:
raise Exception(
"Selector requested descriptor not in ready state")
if fd < 0 or fd >= Selector.get_fd_select_size():
return False
if interest == SelectorIO.READ:
return fd in self._read_fds
elif interest == SelectorIO.WRITE:
return fd in self._write_fds
elif interest == SelectorIO.EXCEPT:
return fd in self._except_fds
return False
|