FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Eliminate the reaper thread (#349) · python-zeroconf/python-zeroconf@7816278 · GitHub

/ python-zeroconf Public

Commit 7816278

Browse files
bdraco
authored
Eliminate the reaper thread (#349)
- Cache is now purged between reads when the interval is reached - Reduce locking since we are already making a copy of the readers and not reading under the lock - Simplify shutdown process
1 parent 523aefb commit 7816278

2 files changed

Lines changed: 53 additions & 63 deletions

File tree

‎zeroconf/__init__.py‎

Lines changed: 37 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@
3434
import time
3535
import warnings
3636
from collections import OrderedDict
37-
from typing import Dict, Iterable, List, Optional, Sequence, Union, cast
37+
from typing import Dict, Iterable, List, Optional, Union, cast
3838
from typing import Any, Callable, Set, Tuple # noqa # used in type hints
3939

4040
import ifaddr
@@ -1337,39 +1337,51 @@ def __init__(self, zc: 'Zeroconf') -> None:
13371337
self.zc = zc
13381338
self.readers = {} # type: Dict[socket.socket, Listener]
13391339
self.timeout = 5
1340+
self.cache_cleanup_interval_ms = 10000.0
13401341
self.condition = threading.Condition()
13411342
self.socketpair = socket.socketpair()
1343+
self._last_cache_cleanup = 0.0
13421344
self.start()
13431345
self.name = "zeroconf-Engine-%s" % (getattr(self, 'native_id', self.ident),)
13441346

13451347
def run(self) -> None:
13461348
while not self.zc.done:
1347-
with self.condition:
1348-
rs = list(self.readers.keys())
1349-
if len(rs) == 0:
1350-
# No sockets to manage, but we wait for the timeout
1351-
# or addition of a socket
1349+
rs = list(self.readers.keys())
1350+
if not rs:
1351+
# No sockets to manage, but we wait for the timeout
1352+
# or addition of a socket
1353+
with self.condition:
13521354
self.condition.wait(self.timeout)
1355+
continue
1356+
1357+
try:
1358+
rs.append(self.socketpair[0])
1359+
rr, wr, er = select.select(rs, [], [], self.timeout)
1360+
1361+
if self.zc.done:
1362+
return
1363+
1364+
for socket_ in rr:
1365+
reader = self.readers.get(socket_)
1366+
if reader:
1367+
reader.handle_read(socket_)
1368+
1369+
if self.socketpair[0] in rr:
1370+
# Clear the socket's buffer
1371+
self.socketpair[0].recv(128)
1372+
1373+
except (select.error, socket.error) as e:
1374+
# If the socket was closed by another thread, during
1375+
# shutdown, ignore it and exit
1376+
if e.args[0] not in (errno.EBADF, errno.ENOTCONN) or not self.zc.done:
1377+
raise
1378+
1379+
now = current_time_millis()
1380+
if now - self._last_cache_cleanup >= self.cache_cleanup_interval_ms:
1381+
self._last_cache_cleanup = now
1382+
for record in self.zc.cache.expire(now):
1383+
self.zc.update_record(now, record)
13531384

1354-
if len(rs) != 0:
1355-
try:
1356-
rs = rs + [self.socketpair[0]]
1357-
rr, wr, er = select.select(cast(Sequence[Any], rs), [], [], self.timeout)
1358-
if not self.zc.done:
1359-
for socket_ in rr:
1360-
reader = self.readers.get(socket_)
1361-
if reader:
1362-
reader.handle_read(socket_)
1363-
1364-
if self.socketpair[0] in rr:
1365-
# Clear the socket's buffer
1366-
self.socketpair[0].recv(128)
1367-
1368-
except (select.error, socket.error) as e:
1369-
# If the socket was closed by another thread, during
1370-
# shutdown, ignore it and exit
1371-
if e.args[0] not in (errno.EBADF, errno.ENOTCONN) or not self.zc.done:
1372-
raise
13731385
self.socketpair[0].close()
13741386
self.socketpair[1].close()
13751387

@@ -1464,32 +1476,6 @@ def handle_read(self, socket_: socket.socket) -> None:
14641476
self.zc.handle_response(msg)
14651477

14661478

1467-
class Reaper(threading.Thread):
1468-
1469-
"""A Reaper is used by this module to remove cache entries that
1470-
have expired."""
1471-
1472-
def __init__(self, zc: 'Zeroconf') -> None:
1473-
threading.Thread.__init__(self)
1474-
self.daemon = True
1475-
self.zc = zc
1476-
self.start()
1477-
self.name = "zeroconf-Reaper_%s" % (getattr(self, 'native_id', self.ident),)
1478-
1479-
def run(self) -> None:
1480-
"""Perodic removal of expired entries from the cache."""
1481-
while True:
1482-
with self.zc.reaper_condition:
1483-
self.zc.reaper_condition.wait(10)
1484-
1485-
if self.zc.done:
1486-
return
1487-
1488-
now = current_time_millis()
1489-
for record in self.zc.cache.expire(now):
1490-
self.zc.update_record(now, record)
1491-
1492-
14931479
class Signal:
14941480
def __init__(self) -> None:
14951481
self._handlers = [] # type: List[Callable[..., None]]
@@ -2505,7 +2491,6 @@ def __init__(
25052491
self.cache = DNSCache()
25062492

25072493
self.condition = threading.Condition()
2508-
self.reaper_condition = threading.Condition()
25092494

25102495
# Ensure we create the lock before
25112496
# we add the listener as we could get
@@ -2519,7 +2504,6 @@ def __init__(
25192504
if self.multi_socket:
25202505
for s in self._respond_sockets:
25212506
self.engine.add_reader(self.listener, s)
2522-
self.reaper = Reaper(self)
25232507

25242508
self.debug = None # type: Optional[DNSOutgoing]
25252509

@@ -2538,11 +2522,6 @@ def notify_all(self) -> None:
25382522
with self.condition:
25392523
self.condition.notify_all()
25402524

2541-
def notify_reaper(self) -> None:
2542-
"""Notifies reaper"""
2543-
with self.reaper_condition:
2544-
self.reaper_condition.notify_all()
2545-
25462525
def get_service_info(self, type_: str, name: str, timeout: int = 3000) -> Optional[ServiceInfo]:
25472526
"""Returns network's service information for a particular
25482527
name and type, or None if no service matches by the timeout,
@@ -2987,7 +2966,5 @@ def close(self) -> None:
29872966

29882967
# shutdown the rest
29892968
self.notify_all()
2990-
self.notify_reaper()
2991-
self.reaper.join()
29922969
for s in self._respond_sockets:
29932970
s.close()

‎zeroconf/test.py‎

Lines changed: 16 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -529,6 +529,17 @@ def test_launch_and_close(self):
529529
rv = r.Zeroconf(interfaces=r.InterfaceChoice.Default)
530530
rv.close()
531531

532+
def test_launch_and_close_unicast(self):
533+
rv = r.Zeroconf(interfaces=r.InterfaceChoice.All, unicast=True)
534+
rv.close()
535+
rv = r.Zeroconf(interfaces=r.InterfaceChoice.Default, unicast=True)
536+
rv.close()
537+
538+
def test_close_multiple_times(self):
539+
rv = r.Zeroconf(interfaces=r.InterfaceChoice.Default)
540+
rv.close()
541+
rv.close()
542+
532543
@unittest.skipIf(not socket.has_ipv6, 'Requires IPv6')
533544
@unittest.skipIf(os.environ.get('SKIP_IPV6'), 'IPv6 tests disabled')
534545
def test_launch_and_close_v4_v6(self):
@@ -966,9 +977,11 @@ def test_reaper(self):
966977
zeroconf.cache.add(record_with_10s_ttl)
967978
zeroconf.cache.add(record_with_1s_ttl)
968979
entries_with_cache = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()]))
969-
time.sleep(1.05)
970-
zeroconf.notify_reaper()
971-
time.sleep(0.05)
980+
zeroconf.engine.cache_cleanup_interval_ms = 10
981+
time.sleep(1)
982+
with zeroconf.engine.condition:
983+
zeroconf.engine._notify()
984+
time.sleep(0.1)
972985
entries = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()]))
973986
zeroconf.close()
974987
assert entries != original_entries

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL