forked from mofeng-git/One-KVM
-
Notifications
You must be signed in to change notification settings - Fork 0
/
inotify.py
300 lines (253 loc) · 10.9 KB
/
inotify.py
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
# ========================================================================== #
# #
# KVMD - The main PiKVM daemon. #
# #
# Copyright (C) 2018-2024 Maxim Devaev <[email protected]> #
# #
# This source file is partially based on python-watchdog module. #
# #
# This program is free software: you can redistribute it and/or modify #
# it under the terms of the GNU General Public License as published by #
# the Free Software Foundation, either version 3 of the License, or #
# (at your option) any later version. #
# #
# This program 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 this program. If not, see <https://www.gnu.org/licenses/>. #
# #
# ========================================================================== #
import sys
import os
import asyncio
import struct
import dataclasses
import types
import errno
from typing import Generator
from .logging import get_logger
from .languages import Languages
from . import aiotools
from . import libc
# =====
_EVENT_HEAD_FMT = "iIII"
_EVENT_HEAD_SIZE = struct.calcsize(_EVENT_HEAD_FMT)
_EVENTS_BUFFER_LENGTH = 4096 * (_EVENT_HEAD_SIZE + 256) # count * (head + max_file_name_size + null_character)
_FS_FALLBACK_ENCODING = "utf-8"
_FS_ENCODING = (sys.getfilesystemencoding() or _FS_FALLBACK_ENCODING)
# =====
def _inotify_parsed_buffer(data: bytes) -> Generator[tuple[int, int, int, bytes], None, None]:
offset = 0
while offset + _EVENT_HEAD_SIZE <= len(data):
(wd, mask, cookie, length) = struct.unpack_from("iIII", data, offset)
name = data[
offset + _EVENT_HEAD_SIZE # noqa: E203
:
offset + _EVENT_HEAD_SIZE + length
].rstrip(b"\0")
offset += _EVENT_HEAD_SIZE + length
if wd >= 0:
yield (wd, mask, cookie, name)
def _inotify_check(retval: int) -> int:
if retval < 0:
c_errno = libc.get_errno()
if c_errno == errno.ENOSPC: # pylint: disable=no-else-raise
raise OSError(c_errno, "Inotify watch limit reached")
elif c_errno == errno.EMFILE:
raise OSError(c_errno, "Inotify instance limit reached")
else:
raise OSError(c_errno, os.strerror(c_errno))
return retval
def _fs_encode(path: str) -> bytes:
try:
return path.encode(_FS_ENCODING, "strict")
except UnicodeEncodeError:
return path.encode(_FS_FALLBACK_ENCODING, "strict")
def _fs_decode(path: bytes) -> str:
try:
return path.decode(_FS_ENCODING, "strict")
except UnicodeDecodeError:
return path.decode(_FS_FALLBACK_ENCODING, "strict")
# =====
class InotifyMask:
# Userspace events
ACCESS = 0x00000001 # File was accessed
ATTRIB = 0x00000004 # Meta-data changed
CLOSE_WRITE = 0x00000008 # Writable file was closed
CLOSE_NOWRITE = 0x00000010 # Unwritable file closed
CREATE = 0x00000100 # Subfile was created
DELETE = 0x00000200 # Subfile was deleted
DELETE_SELF = 0x00000400 # Self was deleted
MODIFY = 0x00000002 # File was modified
MOVE_SELF = 0x00000800 # Self was moved
MOVED_FROM = 0x00000040 # File was moved from X
MOVED_TO = 0x00000080 # File was moved to Y
OPEN = 0x00000020 # File was opened
# Events sent by the kernel to a watch
IGNORED = 0x00008000 # File was ignored
ISDIR = 0x40000000 # Event occurred against directory
Q_OVERFLOW = 0x00004000 # Event queued overflowed
UNMOUNT = 0x00002000 # Backing file system was unmounted
# Helper userspace events
# CLOSE = CLOSE_WRITE | CLOSE_NOWRITE # Close
# MOVE = MOVED_FROM | MOVED_TO # Moves
# Helper for userspace events
# ALL_EVENTS = (
# ACCESS
# | ATTRIB
# | CLOSE_WRITE
# | CLOSE_NOWRITE
# | CREATE
# | DELETE
# | DELETE_SELF
# | MODIFY
# | MOVE_SELF
# | MOVED_FROM
# | MOVED_TO
# | OPEN
# )
# Helper for all modify events
ALL_MODIFY_EVENTS = (
CLOSE_WRITE
| CREATE
| DELETE
| DELETE_SELF
| MODIFY
| MOVE_SELF
| MOVED_FROM
| MOVED_TO
)
# Special flags for watch()
# DONT_FOLLOW = 0x02000000 # Don't follow a symbolic link
# EXCL_UNLINK = 0x04000000 # Exclude events on unlinked objects
# MASK_CREATE = 0x10000000 # Don't overwrite existent watchers (since 4.18)
# MASK_ADD = 0x20000000 # Add to the mask of an existing watch
# ONESHOT = 0x80000000 # Only send event once
# ONLYDIR = 0x01000000 # Only watch the path if it's a directory
@classmethod
def to_string(cls, mask: int) -> str:
flags: list[str] = []
for name in dir(cls):
if (
name[0].isupper()
and not name.startswith("ALL_")
and name not in ["CLOSE", "MOVE"]
and mask & getattr(cls, name)
):
flags.append(name)
return "|".join(flags)
@dataclasses.dataclass(frozen=True, repr=False)
class InotifyEvent:
wd: int
mask: int
cookie: int
name: str
path: str
def __repr__(self) -> str:
return (
f"<InotifyEvent: wd={self.wd}, mask={InotifyMask.to_string(self.mask)},"
f" cookie={self.cookie}, name={self.name}, path={self.path}>"
)
class Inotify:
def __init__(self) -> None:
self.__fd = -1
self.__wd_by_path: dict[str, int] = {}
self.__path_by_wd: dict[int, str] = {}
self.__moved: dict[int, str] = {}
self.__events_queue: "asyncio.Queue[InotifyEvent]" = asyncio.Queue()
async def watch(self, mask: int, *paths: str) -> None:
for path in paths:
path = os.path.normpath(path)
assert path not in self.__wd_by_path, path
get_logger().info(Languages().gettext("Watching for %s"), path)
# Асинхронно, чтобы не висло на NFS
wd = _inotify_check(await aiotools.run_async(libc.inotify_add_watch, self.__fd, _fs_encode(path), mask))
self.__wd_by_path[path] = wd
self.__path_by_wd[wd] = path
# def unwatch(self, path: str) -> None:
# path = os.path.normpath(path)
# assert path in self.__wd_by_path, path
# get_logger().info("Unwatching %s", path)
# wd = self.__wd_by_path[path]
# _inotify_check(_inotify_rm_watch(self.__fd, wd))
# del self.__wd_by_path[path]
# del self.__path_by_wd[wd]
# def has_events(self) -> bool:
# return (not self.__events_queue.empty())
async def get_event(self, timeout: float) -> (InotifyEvent | None):
assert timeout > 0
try:
return (await asyncio.wait_for(
asyncio.ensure_future(self.__events_queue.get()),
timeout=timeout,
))
except asyncio.TimeoutError:
return None
async def get_series(self, timeout: float) -> list[InotifyEvent]:
series: list[InotifyEvent] = []
event = await self.get_event(timeout)
if event:
series.append(event)
while event:
event = await self.get_event(timeout)
if event:
series.append(event)
return series
def __read_and_queue_events(self) -> None:
logger = get_logger()
for event in self.__read_parsed_events():
# XXX: Ни в коем случае не приводить self.__read_parsed_events() к списку.
# Он использует self.__wd_by_path и self.__path_by_wd, содержимое которых
# корректируется кодом ниже. В противном случае все сломается.
if event.mask & InotifyMask.MOVED_FROM:
self.__moved[event.cookie] = event.path # Save moved_from_path
elif event.mask & InotifyMask.MOVED_TO:
moved_from_path = self.__moved.pop(event.cookie, None)
if moved_from_path is not None:
wd = self.__wd_by_path.pop(moved_from_path, None)
if wd is not None:
self.__wd_by_path[event.path] = wd
self.__path_by_wd[wd] = event.path
if event.mask & InotifyMask.IGNORED:
ignored_path = self.__path_by_wd[event.wd]
if self.__wd_by_path[ignored_path] == event.wd:
logger.info(Languages().gettext("Unwatching %s because IGNORED was received"), ignored_path)
del self.__wd_by_path[ignored_path]
continue
self.__events_queue.put_nowait(event)
def __read_parsed_events(self) -> Generator[InotifyEvent, None, None]:
for (wd, mask, cookie, name_bytes) in _inotify_parsed_buffer(self.__read_buffer()):
wd_path = self.__path_by_wd.get(wd, None)
if wd_path is not None:
name = _fs_decode(name_bytes)
path = (os.path.join(wd_path, name) if name else wd_path) # Avoid trailing slash
yield InotifyEvent(wd, mask, cookie, name, path)
def __read_buffer(self) -> bytes:
while True:
try:
return os.read(self.__fd, _EVENTS_BUFFER_LENGTH)
except OSError as err:
if err.errno == errno.EINTR:
pass
def __enter__(self) -> "Inotify":
assert self.__fd < 0
self.__fd = _inotify_check(libc.inotify_init())
asyncio.get_event_loop().add_reader(self.__fd, self.__read_and_queue_events)
return self
def __exit__(
self,
_exc_type: type[BaseException],
_exc: BaseException,
_tb: types.TracebackType,
) -> None:
if self.__fd >= 0:
asyncio.get_event_loop().remove_reader(self.__fd)
for wd in list(self.__wd_by_path.values()):
libc.inotify_rm_watch(self.__fd, wd)
try:
os.close(self.__fd)
except Exception:
pass