diff --git a/README.rst b/README.rst
index 7cefb3794..b87cfb192 100644
--- a/README.rst
+++ b/README.rst
@@ -140,6 +140,29 @@ See examples directory for more.
 Changelog
 =========
 
+0.33.0
+======
+
+This release eliminates all threading locks as all non-threadsafe operations
+now happen in the event loop.
+
+* Let connection_lost close the underlying socket (#918) @bdraco
+
+  The socket was closed during shutdown before asyncio's connection_lost
+  handler had a chance to close it which resulted in a traceback on
+  windows.
+
+  Fixed #917
+
+Technically backwards incompatible:
+
+* Removed duplicate unregister_all_services code (#910) @bdraco
+
+  Calling Zeroconf.close from same asyncio event loop zeroconf is running in
+  will now skip unregister_all_services and log a warning as this a blocking
+  operation and is not async safe and never has been.
+
+  Use AsyncZeroconf instead, or for legacy code call async_unregister_all_services before Zeroconf.close
 
 0.32.1
 ======
diff --git a/examples/async_apple_scanner.py b/examples/async_apple_scanner.py
index f10f6ef63..88b54e4a7 100644
--- a/examples/async_apple_scanner.py
+++ b/examples/async_apple_scanner.py
@@ -36,7 +36,7 @@
 def async_on_service_state_change(
     zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
 ) -> None:
-    print("Service %s of type %s state changed: %s" % (name, service_type, state_change))
+    print(f"Service {name} of type {service_type} state changed: {state_change}")
     if state_change is not ServiceStateChange.Added:
         return
     base_name = name[: -len(service_type) - 1]
@@ -55,11 +55,11 @@ async def _async_show_service_info(zeroconf: Zeroconf, service_type: str, name:
         print("  Name: %s" % name)
         print("  Addresses: %s" % ", ".join(addresses))
         print("  Weight: %d, priority: %d" % (info.weight, info.priority))
-        print("  Server: %s" % (info.server,))
+        print(f"  Server: {info.server}")
         if info.properties:
             print("  Properties are:")
             for key, value in info.properties.items():
-                print("    %s: %s" % (key, value))
+                print(f"    {key}: {value}")
         else:
             print("  No properties")
     else:
diff --git a/examples/async_browser.py b/examples/async_browser.py
index f0e0851cf..1cce5c205 100644
--- a/examples/async_browser.py
+++ b/examples/async_browser.py
@@ -17,7 +17,7 @@
 def async_on_service_state_change(
     zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
 ) -> None:
-    print("Service %s of type %s state changed: %s" % (name, service_type, state_change))
+    print(f"Service {name} of type {service_type} state changed: {state_change}")
     if state_change is not ServiceStateChange.Added:
         return
     asyncio.ensure_future(async_display_service_info(zeroconf, service_type, name))
@@ -32,11 +32,11 @@ async def async_display_service_info(zeroconf: Zeroconf, service_type: str, name
         print("  Name: %s" % name)
         print("  Addresses: %s" % ", ".join(addresses))
         print("  Weight: %d, priority: %d" % (info.weight, info.priority))
-        print("  Server: %s" % (info.server,))
+        print(f"  Server: {info.server}")
         if info.properties:
             print("  Properties are:")
             for key, value in info.properties.items():
-                print("    %s: %s" % (key, value))
+                print(f"    {key}: {value}")
         else:
             print("  No properties")
     else:
diff --git a/examples/async_registration.py b/examples/async_registration.py
index 53d14ce1a..c3aab326a 100644
--- a/examples/async_registration.py
+++ b/examples/async_registration.py
@@ -5,27 +5,32 @@
 import asyncio
 import logging
 import socket
-import time
-from typing import List
+from typing import List, Optional
 
 from zeroconf import IPVersion
 from zeroconf.asyncio import AsyncServiceInfo, AsyncZeroconf
 
 
-async def register_services(infos: List[AsyncServiceInfo]) -> None:
-    tasks = [aiozc.async_register_service(info) for info in infos]
-    background_tasks = await asyncio.gather(*tasks)
-    await asyncio.gather(*background_tasks)
-
-
-async def unregister_services(infos: List[AsyncServiceInfo]) -> None:
-    tasks = [aiozc.async_unregister_service(info) for info in infos]
-    background_tasks = await asyncio.gather(*tasks)
-    await asyncio.gather(*background_tasks)
+class AsyncRunner:
+    def __init__(self, ip_version: IPVersion) -> None:
+        self.ip_version = ip_version
+        self.aiozc: Optional[AsyncZeroconf] = None
 
+    async def register_services(self, infos: List[AsyncServiceInfo]) -> None:
+        self.aiozc = AsyncZeroconf(ip_version=self.ip_version)
+        tasks = [self.aiozc.async_register_service(info) for info in infos]
+        background_tasks = await asyncio.gather(*tasks)
+        await asyncio.gather(*background_tasks)
+        print("Finished registration, press Ctrl-C to exit...")
+        while True:
+            await asyncio.sleep(1)
 
-async def close_aiozc(aiozc: AsyncZeroconf) -> None:
-    await aiozc.async_close()
+    async def unregister_services(self, infos: List[AsyncServiceInfo]) -> None:
+        assert self.aiozc is not None
+        tasks = [self.aiozc.async_unregister_service(info) for info in infos]
+        background_tasks = await asyncio.gather(*tasks)
+        await asyncio.gather(*background_tasks)
+        await self.aiozc.async_close()
 
 
 if __name__ == '__main__':
@@ -60,18 +65,10 @@ async def close_aiozc(aiozc: AsyncZeroconf) -> None:
             )
         )
 
-    print("Registration of 250 services, press Ctrl-C to exit...")
-    aiozc = AsyncZeroconf(ip_version=ip_version)
+    print("Registration of 250 services...")
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(register_services(infos))
-    print("Registration complete.")
+    runner = AsyncRunner(ip_version)
     try:
-        while True:
-            time.sleep(0.1)
+        loop.run_until_complete(runner.register_services(infos))
     except KeyboardInterrupt:
-        pass
-    finally:
-        print("Unregistering...")
-        loop.run_until_complete(unregister_services(infos))
-        print("Unregistration complete.")
-        loop.run_until_complete(close_aiozc(aiozc))
+        loop.run_until_complete(runner.unregister_services(infos))
diff --git a/examples/async_service_info_request.py b/examples/async_service_info_request.py
index 885eb99c4..dd8265b7a 100644
--- a/examples/async_service_info_request.py
+++ b/examples/async_service_info_request.py
@@ -36,11 +36,11 @@ async def async_watch_services(aiozc: AsyncZeroconf) -> None:
                 addresses = ["%s:%d" % (addr, cast(int, info.port)) for addr in info.parsed_addresses()]
                 print("  Addresses: %s" % ", ".join(addresses))
                 print("  Weight: %d, priority: %d" % (info.weight, info.priority))
-                print("  Server: %s" % (info.server,))
+                print(f"  Server: {info.server}")
                 if info.properties:
                     print("  Properties are:")
                     for key, value in info.properties.items():
-                        print("    %s: %s" % (key, value))
+                        print(f"    {key}: {value}")
                 else:
                     print("  No properties")
             else:
diff --git a/examples/browser.py b/examples/browser.py
index 8525e9b9b..8c50e4097 100755
--- a/examples/browser.py
+++ b/examples/browser.py
@@ -16,7 +16,7 @@
 def on_service_state_change(
     zeroconf: Zeroconf, service_type: str, name: str, state_change: ServiceStateChange
 ) -> None:
-    print("Service %s of type %s state changed: %s" % (name, service_type, state_change))
+    print(f"Service {name} of type {service_type} state changed: {state_change}")
 
     if state_change is ServiceStateChange.Added:
         info = zeroconf.get_service_info(service_type, name)
@@ -26,11 +26,11 @@ def on_service_state_change(
             addresses = ["%s:%d" % (addr, cast(int, info.port)) for addr in info.parsed_scoped_addresses()]
             print("  Addresses: %s" % ", ".join(addresses))
             print("  Weight: %d, priority: %d" % (info.weight, info.priority))
-            print("  Server: %s" % (info.server,))
+            print(f"  Server: {info.server}")
             if info.properties:
                 print("  Properties are:")
                 for key, value in info.properties.items():
-                    print("    %s: %s" % (key, value))
+                    print(f"    {key}: {value}")
             else:
                 print("  No properties")
         else:
diff --git a/examples/self_test.py b/examples/self_test.py
index 35007db13..2178629b5 100755
--- a/examples/self_test.py
+++ b/examples/self_test.py
@@ -14,7 +14,7 @@
 
     # Test a few module features, including service registration, service
     # query (for Zoe), and service unregistration.
-    print("Multicast DNS Service Discovery for Python, version %s" % (__version__,))
+    print(f"Multicast DNS Service Discovery for Python, version {__version__}")
     r = Zeroconf()
     print("1. Testing registration of a service...")
     desc = {'version': '0.10', 'a': 'test value', 'b': 'another value'}
@@ -40,7 +40,7 @@
     queried_info = r.get_service_info("_http._tcp.local.", "My Service Name._http._tcp.local.")
     assert queried_info
     assert set(queried_info.parsed_addresses()) == expected
-    print("   Getting self: %s" % (queried_info,))
+    print(f"   Getting self: {queried_info}")
     print("   Query done.")
     print("4. Testing unregister of service information...")
     r.unregister_service(info)
diff --git a/requirements-dev.txt b/requirements-dev.txt
index dc2f21de2..3035d59d6 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -1,5 +1,6 @@
 autopep8
 black;implementation_name=="cpython"
+bump2version
 coveralls
 coverage
 # Version restricted because of https://github.com/PyCQA/pycodestyle/issues/741 - is fixed
diff --git a/setup.cfg b/setup.cfg
index e9dc052f1..53810dcd8 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,11 +1,21 @@
+[bumpversion]
+current_version = 0.33.0
+commit = True
+tag = True
+tag_name = {new_version}
+
+[bumpversion:file:zeroconf/__init__.py]
+search = __version__ = '{current_version}'
+replace = __version__ = '{new_version}'
+
 [tool:pytest]
 testpaths = tests
 
 [flake8]
 show-source = 1
-application-import-names=zeroconf
-max-line-length=110
-ignore=E203,W503
+application-import-names = zeroconf
+max-line-length = 110
+ignore = E203,W503,N818
 
 [mypy]
 ignore_missing_imports = true
@@ -18,7 +28,6 @@ warn_redundant_casts = true
 warn_unused_configs = true
 warn_unused_ignores = true
 warn_return_any = true
-# TODO: disallow untyped calls and defs once we have full type hint coverage
 disallow_untyped_calls = false
 disallow_untyped_defs = true
 
diff --git a/tests/conftest.py b/tests/conftest.py
index c05c4b9b7..d4ea1632a 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ conftest for zeroconf tests. """
diff --git a/tests/services/test_browser.py b/tests/services/test_browser.py
index 26684e093..292dee259 100644
--- a/tests/services/test_browser.py
+++ b/tests/services/test_browser.py
@@ -371,7 +371,7 @@ def mock_incoming_msg(
                 zeroconf,
                 mock_incoming_msg(r.ServiceStateChange.Added, service_types[1], service_names[1], 120),
             )
-            zeroconf.wait(100)
+            time.sleep(0.1)
 
             called_with_refresh_time_check = False
 
@@ -693,7 +693,7 @@ def on_service_state_change(zeroconf, service_type, state_change, name):
 
     zc.register_service(info_service)
 
-    zc.wait(1)
+    time.sleep(0.001)
 
     browser.cancel()
 
diff --git a/tests/services/test_info.py b/tests/services/test_info.py
index 8ac8beda7..2060767f7 100644
--- a/tests/services/test_info.py
+++ b/tests/services/test_info.py
@@ -187,8 +187,7 @@ def test_service_info_rejects_expired_records(self):
             ttl,
             b'\x04ff=0\x04ci=3\x04sf=0\x0bsh=6fLM5A==',
         )
-        expired_record.created = 1000
-        expired_record._expiration_time = 1000
+        expired_record.set_created_ttl(1000, 1)
         info.update_record(zc, now, expired_record)
         assert info.properties[b"ci"] == b"2"
         zc.close()
diff --git a/tests/services/test_registry.py b/tests/services/test_registry.py
index 496cc629b..87c048d5c 100644
--- a/tests/services/test_registry.py
+++ b/tests/services/test_registry.py
@@ -23,10 +23,10 @@ def test_only_register_once(self):
         )
 
         registry = r.ServiceRegistry()
-        registry.add(info)
-        self.assertRaises(r.ServiceNameAlreadyRegistered, registry.add, info)
-        registry.remove(info)
-        registry.add(info)
+        registry.async_add(info)
+        self.assertRaises(r.ServiceNameAlreadyRegistered, registry.async_add, info)
+        registry.async_remove(info)
+        registry.async_add(info)
 
     def test_unregister_multiple_times(self):
         """Verify we can unregister a service multiple times.
@@ -46,10 +46,10 @@ def test_unregister_multiple_times(self):
         )
 
         registry = r.ServiceRegistry()
-        registry.add(info)
-        self.assertRaises(r.ServiceNameAlreadyRegistered, registry.add, info)
-        registry.remove(info)
-        registry.remove(info)
+        registry.async_add(info)
+        self.assertRaises(r.ServiceNameAlreadyRegistered, registry.async_add, info)
+        registry.async_remove(info)
+        registry.async_remove(info)
 
     def test_lookups(self):
         type_ = "_test-srvc-type._tcp.local."
@@ -62,13 +62,13 @@ def test_lookups(self):
         )
 
         registry = r.ServiceRegistry()
-        registry.add(info)
+        registry.async_add(info)
 
-        assert registry.get_service_infos() == [info]
-        assert registry.get_info_name(registration_name) == info
-        assert registry.get_infos_type(type_) == [info]
-        assert registry.get_infos_server("ash-2.local.") == [info]
-        assert registry.get_types() == [type_]
+        assert registry.async_get_service_infos() == [info]
+        assert registry.async_get_info_name(registration_name) == info
+        assert registry.async_get_infos_type(type_) == [info]
+        assert registry.async_get_infos_server("ash-2.local.") == [info]
+        assert registry.async_get_types() == [type_]
 
     def test_lookups_upper_case_by_lower_case(self):
         type_ = "_test-SRVC-type._tcp.local."
@@ -81,13 +81,13 @@ def test_lookups_upper_case_by_lower_case(self):
         )
 
         registry = r.ServiceRegistry()
-        registry.add(info)
+        registry.async_add(info)
 
-        assert registry.get_service_infos() == [info]
-        assert registry.get_info_name(registration_name.lower()) == info
-        assert registry.get_infos_type(type_.lower()) == [info]
-        assert registry.get_infos_server("ash-2.local.") == [info]
-        assert registry.get_types() == [type_.lower()]
+        assert registry.async_get_service_infos() == [info]
+        assert registry.async_get_info_name(registration_name.lower()) == info
+        assert registry.async_get_infos_type(type_.lower()) == [info]
+        assert registry.async_get_infos_server("ash-2.local.") == [info]
+        assert registry.async_get_types() == [type_.lower()]
 
     def test_lookups_lower_case_by_upper_case(self):
         type_ = "_test-srvc-type._tcp.local."
@@ -100,10 +100,10 @@ def test_lookups_lower_case_by_upper_case(self):
         )
 
         registry = r.ServiceRegistry()
-        registry.add(info)
+        registry.async_add(info)
 
-        assert registry.get_service_infos() == [info]
-        assert registry.get_info_name(registration_name.upper()) == info
-        assert registry.get_infos_type(type_.upper()) == [info]
-        assert registry.get_infos_server("ASH-2.local.") == [info]
-        assert registry.get_types() == [type_]
+        assert registry.async_get_service_infos() == [info]
+        assert registry.async_get_info_name(registration_name.upper()) == info
+        assert registry.async_get_infos_type(type_.upper()) == [info]
+        assert registry.async_get_infos_server("ASH-2.local.") == [info]
+        assert registry.async_get_types() == [type_]
diff --git a/tests/services/test_types.py b/tests/services/test_types.py
index d14a8b250..f4206cf43 100644
--- a/tests/services/test_types.py
+++ b/tests/services/test_types.py
@@ -50,7 +50,7 @@ def test_integration_with_listener(self):
             "ash-2.local.",
             addresses=[socket.inet_aton("10.0.1.2")],
         )
-        zeroconf_registrar.registry.add(info)
+        zeroconf_registrar.registry.async_add(info)
         try:
             with patch.object(
                 zeroconf_registrar.engine.protocols[0], "suppress_duplicate_packet", return_value=False
@@ -87,7 +87,7 @@ def test_integration_with_listener_v6_records(self):
             "ash-2.local.",
             addresses=[socket.inet_pton(socket.AF_INET6, addr)],
         )
-        zeroconf_registrar.registry.add(info)
+        zeroconf_registrar.registry.async_add(info)
         try:
             with patch.object(
                 zeroconf_registrar.engine.protocols[0], "suppress_duplicate_packet", return_value=False
@@ -124,7 +124,7 @@ def test_integration_with_listener_ipv6(self):
             "ash-2.local.",
             addresses=[socket.inet_pton(socket.AF_INET6, addr)],
         )
-        zeroconf_registrar.registry.add(info)
+        zeroconf_registrar.registry.async_add(info)
         try:
             with patch.object(
                 zeroconf_registrar.engine.protocols[0], "suppress_duplicate_packet", return_value=False
@@ -160,7 +160,7 @@ def test_integration_with_subtype_and_listener(self):
             "ash-2.local.",
             addresses=[socket.inet_aton("10.0.1.2")],
         )
-        zeroconf_registrar.registry.add(info)
+        zeroconf_registrar.registry.async_add(info)
         try:
             with patch.object(
                 zeroconf_registrar.engine.protocols[0], "suppress_duplicate_packet", return_value=False
diff --git a/tests/test_asyncio.py b/tests/test_asyncio.py
index f4722389d..9ec5e496c 100644
--- a/tests/test_asyncio.py
+++ b/tests/test_asyncio.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """Unit tests for aio.py."""
@@ -57,11 +56,9 @@ def verify_threads_ended():
     yield
     threads_after = frozenset(threading.enumerate())
     non_executor_threads = frozenset(
-        [
-            thread
-            for thread in threads_after
-            if "asyncio" not in thread.name and "ThreadPoolExecutor" not in thread.name
-        ]
+        thread
+        for thread in threads_after
+        if "asyncio" not in thread.name and "ThreadPoolExecutor" not in thread.name
     )
     threads = non_executor_threads - threads_before
     assert not threads
@@ -104,6 +101,7 @@ async def test_async_with_sync_passed_in_closed_in_async() -> None:
 @pytest.mark.asyncio
 async def test_sync_within_event_loop_executor() -> None:
     """Test sync version still works from an executor within an event loop."""
+
     def sync_code():
         zc = Zeroconf(interfaces=['127.0.0.1'])
         assert zc.get_service_info("_neverused._tcp.local.", "xneverused._neverused._tcp.local.", 10) is None
@@ -118,7 +116,7 @@ async def test_async_service_registration() -> None:
     aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
     type_ = "_test1-srvc-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     calls = []
 
@@ -178,7 +176,7 @@ async def test_async_service_registration_name_conflict() -> None:
     aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
     type_ = "_test-srvc2-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
@@ -226,7 +224,7 @@ async def test_async_service_registration_name_does_not_match_type() -> None:
     aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
     type_ = "_test-srvc3-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
@@ -253,7 +251,7 @@ async def test_async_tasks() -> None:
     aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
     type_ = "_test-srvc4-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     calls = []
 
@@ -319,7 +317,7 @@ async def test_async_wait_unblocks_on_update() -> None:
     aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
     type_ = "_test-srvc4-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
@@ -355,8 +353,8 @@ async def test_service_info_async_request() -> None:
     type_ = "_test1-srvc-type._tcp.local."
     name = "xxxyyy"
     name2 = "abc"
-    registration_name = "%s.%s" % (name, type_)
-    registration_name2 = "%s.%s" % (name2, type_)
+    registration_name = f"{name}.{type_}"
+    registration_name2 = f"{name2}.{type_}"
 
     # Start a tasks BEFORE the registration that will keep trying
     # and see the registration a bit later
@@ -453,7 +451,7 @@ async def test_async_service_browser() -> None:
     aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
     type_ = "_test9-srvc-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     calls = []
 
@@ -512,7 +510,7 @@ async def test_async_context_manager() -> None:
     """Test using an async context manager."""
     type_ = "_test10-sr-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     async with AsyncZeroconf(interfaces=['127.0.0.1']) as aiozc:
         info = ServiceInfo(
@@ -538,8 +536,8 @@ async def test_async_unregister_all_services() -> None:
     type_ = "_test1-srvc-type._tcp.local."
     name = "xxxyyy"
     name2 = "abc"
-    registration_name = "%s.%s" % (name, type_)
-    registration_name2 = "%s.%s" % (name2, type_)
+    registration_name = f"{name}.{type_}"
+    registration_name2 = f"{name2}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
@@ -593,7 +591,7 @@ async def test_async_unregister_all_services() -> None:
 async def test_async_zeroconf_service_types():
     type_ = "_test-srvc-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     zeroconf_registrar = AsyncZeroconf(interfaces=['127.0.0.1'])
     desc = {'path': '/~paulsm/'}
@@ -807,14 +805,14 @@ async def test_info_asking_default_is_asking_qm_questions_after_the_first_qu():
     zeroconf_info = aiozc.zeroconf
 
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")]
     )
 
-    zeroconf_info.registry.add(info)
+    zeroconf_info.registry.async_add(info)
 
     # we are going to patch the zeroconf send to check query transmission
     old_send = zeroconf_info.async_send
diff --git a/tests/test_cache.py b/tests/test_cache.py
index 4b3a8a18e..559b43573 100644
--- a/tests/test_cache.py
+++ b/tests/test_cache.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._cache. """
@@ -98,7 +97,7 @@ def test_async_all_by_details(self):
         record2 = r.DNSAddress('a', const._TYPE_A, const._CLASS_IN, 1, b'b')
         cache = r.DNSCache()
         cache.async_add_records([record1, record2])
-        assert set(cache.async_all_by_details('a', const._TYPE_A, const._CLASS_IN)) == set([record1, record2])
+        assert set(cache.async_all_by_details('a', const._TYPE_A, const._CLASS_IN)) == {record1, record2}
 
     def test_async_entries_with_server(self):
         record1 = r.DNSService(
@@ -109,8 +108,8 @@ def test_async_entries_with_server(self):
         )
         cache = r.DNSCache()
         cache.async_add_records([record1, record2])
-        assert set(cache.async_entries_with_server('ab')) == set([record1, record2])
-        assert set(cache.async_entries_with_server('AB')) == set([record1, record2])
+        assert set(cache.async_entries_with_server('ab')) == {record1, record2}
+        assert set(cache.async_entries_with_server('AB')) == {record1, record2}
 
     def test_async_entries_with_name(self):
         record1 = r.DNSService(
@@ -121,8 +120,8 @@ def test_async_entries_with_name(self):
         )
         cache = r.DNSCache()
         cache.async_add_records([record1, record2])
-        assert set(cache.async_entries_with_name('irrelevant')) == set([record1, record2])
-        assert set(cache.async_entries_with_name('Irrelevant')) == set([record1, record2])
+        assert set(cache.async_entries_with_name('irrelevant')) == {record1, record2}
+        assert set(cache.async_entries_with_name('Irrelevant')) == {record1, record2}
 
 
 # These functions have been seen in other projects so
@@ -152,7 +151,7 @@ def test_get_all_by_details(self):
         record2 = r.DNSAddress('a', const._TYPE_A, const._CLASS_IN, 1, b'b')
         cache = r.DNSCache()
         cache.async_add_records([record1, record2])
-        assert set(cache.get_all_by_details('a', const._TYPE_A, const._CLASS_IN)) == set([record1, record2])
+        assert set(cache.get_all_by_details('a', const._TYPE_A, const._CLASS_IN)) == {record1, record2}
 
     def test_entries_with_server(self):
         record1 = r.DNSService(
@@ -163,8 +162,8 @@ def test_entries_with_server(self):
         )
         cache = r.DNSCache()
         cache.async_add_records([record1, record2])
-        assert set(cache.entries_with_server('ab')) == set([record1, record2])
-        assert set(cache.entries_with_server('AB')) == set([record1, record2])
+        assert set(cache.entries_with_server('ab')) == {record1, record2}
+        assert set(cache.entries_with_server('AB')) == {record1, record2}
 
     def test_entries_with_name(self):
         record1 = r.DNSService(
@@ -175,8 +174,8 @@ def test_entries_with_name(self):
         )
         cache = r.DNSCache()
         cache.async_add_records([record1, record2])
-        assert set(cache.entries_with_name('irrelevant')) == set([record1, record2])
-        assert set(cache.entries_with_name('Irrelevant')) == set([record1, record2])
+        assert set(cache.entries_with_name('irrelevant')) == {record1, record2}
+        assert set(cache.entries_with_name('Irrelevant')) == {record1, record2}
 
     def test_current_entry_with_name_and_alias(self):
         record1 = r.DNSPointer(
diff --git a/tests/test_core.py b/tests/test_core.py
index d80514f7c..fd45b1ee5 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._core """
@@ -55,28 +54,27 @@ async def test_reaper():
         aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
         zeroconf = aiozc.zeroconf
         cache = zeroconf.cache
-        original_entries = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()]))
+        original_entries = list(itertools.chain(*(cache.entries_with_name(name) for name in cache.names())))
         record_with_10s_ttl = r.DNSAddress('a', const._TYPE_SOA, const._CLASS_IN, 10, b'a')
         record_with_1s_ttl = r.DNSAddress('a', const._TYPE_SOA, const._CLASS_IN, 1, b'b')
         zeroconf.cache.async_add_records([record_with_10s_ttl, record_with_1s_ttl])
         question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN)
         now = r.current_time_millis()
-        other_known_answers = set(
-            [
-                r.DNSPointer(
-                    "_hap._tcp.local.",
-                    const._TYPE_PTR,
-                    const._CLASS_IN,
-                    10000,
-                    'known-to-other._hap._tcp.local.',
-                )
-            ]
-        )
+        other_known_answers = {
+            r.DNSPointer(
+                "_hap._tcp.local.",
+                const._TYPE_PTR,
+                const._CLASS_IN,
+                10000,
+                'known-to-other._hap._tcp.local.',
+            )
+        }
         zeroconf.question_history.add_question_at_time(question, now, other_known_answers)
         assert zeroconf.question_history.suppresses(question, now, other_known_answers)
-        entries_with_cache = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()]))
+        entries_with_cache = list(itertools.chain(*(cache.entries_with_name(name) for name in cache.names())))
         await asyncio.sleep(1.2)
-        entries = list(itertools.chain(*[cache.entries_with_name(name) for name in cache.names()]))
+        entries = list(itertools.chain(*(cache.entries_with_name(name) for name in cache.names())))
+        assert zeroconf.cache.get(record_with_1s_ttl) is None
         await aiozc.async_close()
         assert not zeroconf.question_history.suppresses(question, now, other_known_answers)
         assert entries != original_entries
@@ -85,6 +83,24 @@ async def test_reaper():
         assert record_with_1s_ttl not in entries
 
 
+@pytest.mark.asyncio
+async def test_reaper_aborts_when_done():
+    """Ensure cache cleanup stops when zeroconf is done."""
+    with patch.object(_core, "_CACHE_CLEANUP_INTERVAL", 10):
+        assert _core._CACHE_CLEANUP_INTERVAL == 10
+        aiozc = AsyncZeroconf(interfaces=['127.0.0.1'])
+        zeroconf = aiozc.zeroconf
+        record_with_10s_ttl = r.DNSAddress('a', const._TYPE_SOA, const._CLASS_IN, 10, b'a')
+        record_with_1s_ttl = r.DNSAddress('a', const._TYPE_SOA, const._CLASS_IN, 1, b'b')
+        zeroconf.cache.async_add_records([record_with_10s_ttl, record_with_1s_ttl])
+        assert zeroconf.cache.get(record_with_10s_ttl) is not None
+        assert zeroconf.cache.get(record_with_1s_ttl) is not None
+        await aiozc.async_close()
+        await asyncio.sleep(1.2)
+        assert zeroconf.cache.get(record_with_10s_ttl) is not None
+        assert zeroconf.cache.get(record_with_1s_ttl) is not None
+
+
 class Framework(unittest.TestCase):
     def test_launch_and_close(self):
         rv = r.Zeroconf(interfaces=r.InterfaceChoice.All)
@@ -338,11 +354,11 @@ def test_goodbye_all_services():
     info = r.ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
+    zc.registry.async_add(info)
     out = zc.generate_unregister_all_services()
     assert out is not None
     first_packet = out.packets()
-    zc.registry.add(info)
+    zc.registry.async_add(info)
     out2 = zc.generate_unregister_all_services()
     assert out2 is not None
     second_packet = out.packets()
@@ -351,7 +367,7 @@ def test_goodbye_all_services():
     # Verify the registery is empty
     out3 = zc.generate_unregister_all_services()
     assert out3 is None
-    assert zc.registry.get_service_infos() == []
+    assert zc.registry.async_get_service_infos() == []
 
     zc.close()
 
@@ -367,7 +383,7 @@ def test_register_service_with_custom_ttl():
     name = "MyTestHome"
     info_service = r.ServiceInfo(
         type_,
-        '%s.%s' % (name, type_),
+        f'{name}.{type_}',
         80,
         0,
         0,
@@ -423,9 +439,9 @@ def test_tc_bit_defers():
     name2 = "knownname2"
     name3 = "knownname3"
 
-    registration_name = "%s.%s" % (name, type_)
-    registration2_name = "%s.%s" % (name2, type_)
-    registration3_name = "%s.%s" % (name3, type_)
+    registration_name = f"{name}.{type_}"
+    registration2_name = f"{name2}.{type_}"
+    registration3_name = f"{name3}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
@@ -441,9 +457,9 @@ def test_tc_bit_defers():
     info3 = r.ServiceInfo(
         type_, registration3_name, 80, 0, 0, desc, server_name3, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
-    zc.registry.add(info2)
-    zc.registry.add(info3)
+    zc.registry.async_add(info)
+    zc.registry.async_add(info2)
+    zc.registry.async_add(info3)
 
     protocol = zc.engine.protocols[0]
     now = r.current_time_millis()
@@ -502,9 +518,9 @@ def test_tc_bit_defers_last_response_missing():
     name2 = "knownname2"
     name3 = "knownname3"
 
-    registration_name = "%s.%s" % (name, type_)
-    registration2_name = "%s.%s" % (name2, type_)
-    registration3_name = "%s.%s" % (name3, type_)
+    registration_name = f"{name}.{type_}"
+    registration2_name = f"{name2}.{type_}"
+    registration3_name = f"{name3}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
@@ -520,9 +536,9 @@ def test_tc_bit_defers_last_response_missing():
     info3 = r.ServiceInfo(
         type_, registration3_name, 80, 0, 0, desc, server_name3, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
-    zc.registry.add(info2)
-    zc.registry.add(info3)
+    zc.registry.async_add(info)
+    zc.registry.async_add(info2)
+    zc.registry.async_add(info3)
 
     protocol = zc.engine.protocols[0]
     now = r.current_time_millis()
@@ -584,7 +600,7 @@ def test_tc_bit_defers_last_response_missing():
     assert source_ip not in protocol._timers
 
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     zc.close()
 
 
@@ -729,7 +745,7 @@ def test_shutdown_while_register_in_process():
     name = "MyTestHome"
     info_service = r.ServiceInfo(
         type_,
-        '%s.%s' % (name, type_),
+        f'{name}.{type_}',
         80,
         0,
         0,
diff --git a/tests/test_dns.py b/tests/test_dns.py
index 197357067..071e1f650 100644
--- a/tests/test_dns.py
+++ b/tests/test_dns.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._dns. """
@@ -101,7 +100,7 @@ def test_dns_record_reset_ttl(self):
     def test_service_info_dunder(self):
         type_ = "_test-srvc-type._tcp.local."
         name = "xxxyyy"
-        registration_name = "%s.%s" % (name, type_)
+        registration_name = f"{name}.{type_}"
         info = ServiceInfo(
             type_,
             registration_name,
@@ -119,7 +118,7 @@ def test_service_info_dunder(self):
     def test_service_info_text_properties_not_given(self):
         type_ = "_test-srvc-type._tcp.local."
         name = "xxxyyy"
-        registration_name = "%s.%s" % (name, type_)
+        registration_name = f"{name}.{type_}"
         info = ServiceInfo(
             type_=type_,
             name=registration_name,
@@ -166,7 +165,7 @@ def test_dns_record_hashablity_does_not_consider_ttl():
     record1 = r.DNSAddress('irrelevant', const._TYPE_A, const._CLASS_IN, const._DNS_OTHER_TTL, b'same')
     record2 = r.DNSAddress('irrelevant', const._TYPE_A, const._CLASS_IN, const._DNS_HOST_TTL, b'same')
 
-    record_set = set([record1, record2])
+    record_set = {record1, record2}
     assert len(record_set) == 1
 
     record_set.add(record1)
@@ -187,7 +186,7 @@ def test_dns_address_record_hashablity():
     address3 = r.DNSAddress('irrelevant', const._TYPE_A, const._CLASS_IN, 1, b'c')
     address4 = r.DNSAddress('irrelevant', const._TYPE_AAAA, const._CLASS_IN, 1, b'c')
 
-    record_set = set([address1, address2, address3, address4])
+    record_set = {address1, address2, address3, address4}
     assert len(record_set) == 4
 
     record_set.add(address1)
@@ -199,9 +198,9 @@ def test_dns_address_record_hashablity():
     assert len(record_set) == 4
 
     # Verify we can remove records
-    additional_set = set([address1, address2])
+    additional_set = {address1, address2}
     record_set -= additional_set
-    assert record_set == set([address3, address4])
+    assert record_set == {address3, address4}
 
 
 def test_dns_hinfo_record_hashablity():
@@ -209,7 +208,7 @@ def test_dns_hinfo_record_hashablity():
     hinfo1 = r.DNSHinfo('irrelevant', const._TYPE_HINFO, 0, 0, 'cpu1', 'os')
     hinfo2 = r.DNSHinfo('irrelevant', const._TYPE_HINFO, 0, 0, 'cpu2', 'os')
 
-    record_set = set([hinfo1, hinfo2])
+    record_set = {hinfo1, hinfo2}
     assert len(record_set) == 2
 
     record_set.add(hinfo1)
@@ -228,7 +227,7 @@ def test_dns_pointer_record_hashablity():
     ptr1 = r.DNSPointer('irrelevant', const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, '123')
     ptr2 = r.DNSPointer('irrelevant', const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, '456')
 
-    record_set = set([ptr1, ptr2])
+    record_set = {ptr1, ptr2}
     assert len(record_set) == 2
 
     record_set.add(ptr1)
@@ -249,7 +248,7 @@ def test_dns_text_record_hashablity():
     text3 = r.DNSText('irrelevant', 0, 1, const._DNS_OTHER_TTL, b'12345678901')
     text4 = r.DNSText('irrelevant', 0, 0, const._DNS_OTHER_TTL, b'ABCDEFGHIJK')
 
-    record_set = set([text1, text2, text3, text4])
+    record_set = {text1, text2, text3, text4}
 
     assert len(record_set) == 4
 
@@ -271,7 +270,7 @@ def test_dns_service_record_hashablity():
     srv3 = r.DNSService('irrelevant', const._TYPE_SRV, const._CLASS_IN, const._DNS_HOST_TTL, 0, 0, 81, 'a')
     srv4 = r.DNSService('irrelevant', const._TYPE_SRV, const._CLASS_IN, const._DNS_HOST_TTL, 0, 0, 80, 'ab')
 
-    record_set = set([srv1, srv2, srv3, srv4])
+    record_set = {srv1, srv2, srv3, srv4}
 
     assert len(record_set) == 4
 
@@ -288,6 +287,31 @@ def test_dns_service_record_hashablity():
     assert len(record_set) == 4
 
 
+def test_dns_nsec_record_hashablity():
+    """Test DNSNsec are hashable."""
+    nsec1 = r.DNSNsec(
+        'irrelevant', const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, 'irrelevant', [1, 2, 3]
+    )
+    nsec2 = r.DNSNsec(
+        'irrelevant', const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, 'irrelevant', [1, 2]
+    )
+
+    record_set = {nsec1, nsec2}
+    assert len(record_set) == 2
+
+    record_set.add(nsec1)
+    assert len(record_set) == 2
+
+    nsec2_dupe = r.DNSNsec(
+        'irrelevant', const._TYPE_PTR, const._CLASS_IN, const._DNS_OTHER_TTL, 'irrelevant', [1, 2]
+    )
+    assert nsec2 == nsec2_dupe
+    assert nsec2.__hash__() == nsec2_dupe.__hash__()
+
+    record_set.add(nsec2_dupe)
+    assert len(record_set) == 2
+
+
 def test_rrset_does_not_consider_ttl():
     """Test DNSRRSet does not consider the ttl in the hash."""
 
diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py
index aa2f74f60..47e68b75f 100644
--- a/tests/test_exceptions.py
+++ b/tests/test_exceptions.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._exceptions """
@@ -107,7 +106,7 @@ def test_bad_types(self):
         bad_names_to_try = (
             '._x._tcp.local.',
             'a' * 64 + '._sub._http._tcp.local.',
-            'a' * 62 + u'â._sub._http._tcp.local.',
+            'a' * 62 + 'â._sub._http._tcp.local.',
         )
         for name in bad_names_to_try:
             self.assertRaises(r.BadTypeInNameException, r.service_type_name, name)
@@ -129,7 +128,7 @@ def test_good_service_names(self):
             ('_12345-67890-abc._udp.local.', '_12345-67890-abc._udp.local.'),
             ('x._sub._http._tcp.local.', '_http._tcp.local.'),
             ('a' * 63 + '._sub._http._tcp.local.', '_http._tcp.local.'),
-            ('a' * 61 + u'â._sub._http._tcp.local.', '_http._tcp.local.'),
+            ('a' * 61 + 'â._sub._http._tcp.local.', '_http._tcp.local.'),
         )
 
         for name, result in good_names_to_try:
@@ -140,7 +139,7 @@ def test_good_service_names(self):
     def test_invalid_addresses(self):
         type_ = "_test-srvc-type._tcp.local."
         name = "xxxyyy"
-        registration_name = "%s.%s" % (name, type_)
+        registration_name = f"{name}.{type_}"
 
         bad = ('127.0.0.1', '::1', 42)
         for addr in bad:
diff --git a/tests/test_handlers.py b/tests/test_handlers.py
index bab50a558..d4144721b 100644
--- a/tests/test_handlers.py
+++ b/tests/test_handlers.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._handlers """
@@ -46,7 +45,7 @@ def test_ttl(self):
         # service definition
         type_ = "_test-srvc-type._tcp.local."
         name = "xxxyyy"
-        registration_name = "%s.%s" % (name, type_)
+        registration_name = f"{name}.{type_}"
 
         desc = {'path': '/~paulsm/'}
         info = ServiceInfo(
@@ -88,7 +87,7 @@ def _process_outgoing_packet(out):
         expected_ttl = None
         for _ in range(3):
             _process_outgoing_packet(zc.generate_service_query(info))
-        zc.registry.add(info)
+        zc.registry.async_add(info)
         for _ in range(3):
             _process_outgoing_packet(zc.generate_service_broadcast(info, None))
         assert nbr_answers == 12 and nbr_additionals == 0 and nbr_authorities == 3
@@ -113,7 +112,7 @@ def _process_outgoing_packet(out):
 
         # unregister
         expected_ttl = 0
-        zc.registry.remove(info)
+        zc.registry.async_remove(info)
         for _ in range(3):
             _process_outgoing_packet(zc.generate_service_broadcast(info, 0))
         assert nbr_answers == 12 and nbr_additionals == 0 and nbr_authorities == 0
@@ -122,7 +121,7 @@ def _process_outgoing_packet(out):
         expected_ttl = None
         for _ in range(3):
             _process_outgoing_packet(zc.generate_service_query(info))
-        zc.registry.add(info)
+        zc.registry.async_add(info)
         # register service with custom TTL
         expected_ttl = const._DNS_HOST_TTL * 2
         assert expected_ttl != const._DNS_HOST_TTL
@@ -148,7 +147,7 @@ def _process_outgoing_packet(out):
 
         # unregister
         expected_ttl = 0
-        zc.registry.remove(info)
+        zc.registry.async_remove(info)
         for _ in range(3):
             _process_outgoing_packet(zc.generate_service_broadcast(info, 0))
         assert nbr_answers == 12 and nbr_additionals == 0 and nbr_authorities == 0
@@ -160,7 +159,7 @@ def test_name_conflicts(self):
         zc = Zeroconf(interfaces=['127.0.0.1'])
         type_ = "_homeassistant._tcp.local."
         name = "Home"
-        registration_name = "%s.%s" % (name, type_)
+        registration_name = f"{name}.{type_}"
 
         info = ServiceInfo(
             type_,
@@ -189,7 +188,7 @@ def test_register_and_lookup_type_by_uppercase_name(self):
         zc = Zeroconf(interfaces=['127.0.0.1'])
         type_ = "_mylowertype._tcp.local."
         name = "Home"
-        registration_name = "%s.%s" % (name, type_)
+        registration_name = f"{name}.{type_}"
 
         info = ServiceInfo(
             type_,
@@ -224,7 +223,7 @@ def test_ptr_optimization():
     # service definition
     type_ = "_test-srvc-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
@@ -280,12 +279,12 @@ def test_any_query_for_ptr():
     zc = Zeroconf(interfaces=['127.0.0.1'])
     type_ = "_anyptr._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
     ipv6_address = socket.inet_pton(socket.AF_INET6, "2001:db8::1")
     info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, server_name, addresses=[ipv6_address])
-    zc.registry.add(info)
+    zc.registry.async_add(info)
 
     _clear_cache(zc)
     generated = r.DNSOutgoing(const._FLAGS_QR_QUERY)
@@ -298,7 +297,7 @@ def test_any_query_for_ptr():
     assert multicast_out.answers[0][0].name == type_
     assert multicast_out.answers[0][0].alias == registration_name
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     zc.close()
 
 
@@ -307,12 +306,12 @@ def test_aaaa_query():
     zc = Zeroconf(interfaces=['127.0.0.1'])
     type_ = "_knownaaaservice._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
     ipv6_address = socket.inet_pton(socket.AF_INET6, "2001:db8::1")
     info = ServiceInfo(type_, registration_name, 80, 0, 0, desc, server_name, addresses=[ipv6_address])
-    zc.registry.add(info)
+    zc.registry.async_add(info)
 
     generated = r.DNSOutgoing(const._FLAGS_QR_QUERY)
     question = r.DNSQuestion(server_name, const._TYPE_AAAA, const._CLASS_IN)
@@ -323,7 +322,7 @@ def test_aaaa_query():
     )
     assert multicast_out.answers[0][0].address == ipv6_address
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     zc.close()
 
 
@@ -332,7 +331,7 @@ def test_a_and_aaaa_record_fate_sharing():
     zc = Zeroconf(interfaces=['127.0.0.1'])
     type_ = "_a-and-aaaa-service._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
     ipv6_address = socket.inet_pton(socket.AF_INET6, "2001:db8::1")
@@ -343,7 +342,7 @@ def test_a_and_aaaa_record_fate_sharing():
     aaaa_record = info.dns_addresses(version=r.IPVersion.V6Only)[0]
     a_record = info.dns_addresses(version=r.IPVersion.V4Only)[0]
 
-    zc.registry.add(info)
+    zc.registry.async_add(info)
 
     # Test AAAA query
     generated = r.DNSOutgoing(const._FLAGS_QR_QUERY)
@@ -376,7 +375,7 @@ def test_a_and_aaaa_record_fate_sharing():
     assert len(multicast_out.answers) == 1
     assert len(multicast_out.additionals) == 1
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     zc.close()
 
 
@@ -388,13 +387,13 @@ def test_unicast_response():
     # service definition
     type_ = "_test-srvc-type._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")]
     )
     # register
-    zc.registry.add(info)
+    zc.registry.async_add(info)
     _clear_cache(zc)
 
     # query
@@ -421,7 +420,7 @@ def test_unicast_response():
         assert has_srv and has_txt and has_a
 
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     zc.close()
 
 
@@ -434,8 +433,8 @@ def test_qu_response():
     type_ = "_test-srvc-type._tcp.local."
     other_type_ = "_notthesame._tcp.local."
     name = "xxxyyy"
-    registration_name = "%s.%s" % (name, type_)
-    registration_name2 = "%s.%s" % (name, other_type_)
+    registration_name = f"{name}.{type_}"
+    registration_name2 = f"{name}.{other_type_}"
     desc = {'path': '/~paulsm/'}
     info = ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, "ash-2.local.", addresses=[socket.inet_aton("10.0.1.2")]
@@ -530,13 +529,13 @@ def test_known_answer_supression():
     zc = Zeroconf(interfaces=['127.0.0.1'])
     type_ = "_knownanswersv8._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
     info = ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, server_name, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
+    zc.registry.async_add(info)
 
     now = current_time_millis()
     _clear_cache(zc)
@@ -632,7 +631,7 @@ def test_known_answer_supression():
     assert not multicast_out or not multicast_out.answers
 
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     zc.close()
 
 
@@ -643,9 +642,9 @@ def test_multi_packet_known_answer_supression():
     name2 = "knownname2"
     name3 = "knownname3"
 
-    registration_name = "%s.%s" % (name, type_)
-    registration2_name = "%s.%s" % (name2, type_)
-    registration3_name = "%s.%s" % (name3, type_)
+    registration_name = f"{name}.{type_}"
+    registration2_name = f"{name2}.{type_}"
+    registration3_name = f"{name3}.{type_}"
 
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
@@ -661,9 +660,9 @@ def test_multi_packet_known_answer_supression():
     info3 = ServiceInfo(
         type_, registration3_name, 80, 0, 0, desc, server_name3, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
-    zc.registry.add(info2)
-    zc.registry.add(info3)
+    zc.registry.async_add(info)
+    zc.registry.async_add(info2)
+    zc.registry.async_add(info3)
 
     now = current_time_millis()
     _clear_cache(zc)
@@ -684,9 +683,9 @@ def test_multi_packet_known_answer_supression():
     assert unicast_out is None
     assert multicast_out is None
     # unregister
-    zc.registry.remove(info)
-    zc.registry.remove(info2)
-    zc.registry.remove(info3)
+    zc.registry.async_remove(info)
+    zc.registry.async_remove(info2)
+    zc.registry.async_remove(info3)
     zc.close()
 
 
@@ -694,23 +693,23 @@ def test_known_answer_supression_service_type_enumeration_query():
     zc = Zeroconf(interfaces=['127.0.0.1'])
     type_ = "_otherknown._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
     info = ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, server_name, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
+    zc.registry.async_add(info)
 
     type_2 = "_otherknown2._tcp.local."
     name = "knownname"
-    registration_name2 = "%s.%s" % (name, type_2)
+    registration_name2 = f"{name}.{type_2}"
     desc = {'path': '/~paulsm/'}
     server_name2 = "ash-3.local."
     info2 = ServiceInfo(
         type_2, registration_name2, 80, 0, 0, desc, server_name2, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info2)
+    zc.registry.async_add(info2)
     now = current_time_millis()
     _clear_cache(zc)
 
@@ -756,8 +755,8 @@ def test_known_answer_supression_service_type_enumeration_query():
     assert not multicast_out or not multicast_out.answers
 
     # unregister
-    zc.registry.remove(info)
-    zc.registry.remove(info2)
+    zc.registry.async_remove(info)
+    zc.registry.async_remove(info2)
     zc.close()
 
 
@@ -772,23 +771,23 @@ async def test_qu_response_only_sends_additionals_if_sends_answer():
 
     type_ = "_addtest1._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "ash-2.local."
     info = ServiceInfo(
         type_, registration_name, 80, 0, 0, desc, server_name, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info)
+    zc.registry.async_add(info)
 
     type_2 = "_addtest2._tcp.local."
     name = "knownname"
-    registration_name2 = "%s.%s" % (name, type_2)
+    registration_name2 = f"{name}.{type_2}"
     desc = {'path': '/~paulsm/'}
     server_name2 = "ash-3.local."
     info2 = ServiceInfo(
         type_2, registration_name2, 80, 0, 0, desc, server_name2, addresses=[socket.inet_aton("10.0.1.2")]
     )
-    zc.registry.add(info2)
+    zc.registry.async_add(info2)
 
     ptr_record = info.dns_pointer()
 
@@ -889,7 +888,7 @@ async def test_qu_response_only_sends_additionals_if_sends_answer():
     assert info2.dns_service() in unicast_out.additionals
 
     # unregister
-    zc.registry.remove(info)
+    zc.registry.async_remove(info)
     await aiozc.async_close()
 
 
@@ -904,7 +903,7 @@ async def test_cache_flush_bit():
 
     type_ = "_cacheflush._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "server-uu1.local."
     info = ServiceInfo(
@@ -992,7 +991,7 @@ def async_update_records(self, zc: 'Zeroconf', now: float, records: List[r.Recor
 
     type_ = "_cacheflush._tcp.local."
     name = "knownname"
-    registration_name = "%s.%s" % (name, type_)
+    registration_name = f"{name}.{type_}"
     desc = {'path': '/~paulsm/'}
     server_name = "server-uu1.local."
     info = ServiceInfo(
@@ -1013,11 +1012,11 @@ def async_update_records(self, zc: 'Zeroconf', now: float, records: List[r.Recor
     )
     await asyncio.sleep(0)  # flush out the call_soon_threadsafe
 
-    assert set([record.new for record in updated]) == set([ptr_record, a_record])
+    assert {record.new for record in updated} == {ptr_record, a_record}
 
     # The old records should be None so we trigger Add events
     # in service browsers instead of Update events
-    assert set([record.old for record in updated]) == set([None])
+    assert {record.old for record in updated} == {None}
 
     await aiozc.async_close()
 
@@ -1044,7 +1043,7 @@ async def test_questions_query_handler_populates_the_question_history_from_qm_qu
     )
     assert unicast_out is None
     assert multicast_out is None
-    assert zc.question_history.suppresses(question, now, set([known_answer]))
+    assert zc.question_history.suppresses(question, now, {known_answer})
 
     await aiozc.async_close()
 
@@ -1071,7 +1070,7 @@ async def test_questions_query_handler_does_not_put_qu_questions_in_history():
     )
     assert unicast_out is None
     assert multicast_out is None
-    assert not zc.question_history.suppresses(question, now, set([known_answer]))
+    assert not zc.question_history.suppresses(question, now, {known_answer})
 
     await aiozc.async_close()
 
diff --git a/tests/test_history.py b/tests/test_history.py
index 89159dff6..9da6b5679 100644
--- a/tests/test_history.py
+++ b/tests/test_history.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """Unit tests for _history.py."""
@@ -14,20 +13,16 @@ def test_question_suppression():
 
     question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN)
     now = r.current_time_millis()
-    other_known_answers = set(
-        [
-            r.DNSPointer(
-                "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.'
-            )
-        ]
-    )
-    our_known_answers = set(
-        [
-            r.DNSPointer(
-                "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-us._hap._tcp.local.'
-            )
-        ]
-    )
+    other_known_answers = {
+        r.DNSPointer(
+            "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.'
+        )
+    }
+    our_known_answers = {
+        r.DNSPointer(
+            "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-us._hap._tcp.local.'
+        )
+    }
 
     history.add_question_at_time(question, now, other_known_answers)
 
@@ -52,13 +47,11 @@ def test_question_expire():
 
     question = r.DNSQuestion("_hap._tcp._local.", const._TYPE_PTR, const._CLASS_IN)
     now = r.current_time_millis()
-    other_known_answers = set(
-        [
-            r.DNSPointer(
-                "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.'
-            )
-        ]
-    )
+    other_known_answers = {
+        r.DNSPointer(
+            "_hap._tcp.local.", const._TYPE_PTR, const._CLASS_IN, 10000, 'known-to-other._hap._tcp.local.'
+        )
+    }
     history.add_question_at_time(question, now, other_known_answers)
 
     # Verify the question is suppressed if the known answers are the same
diff --git a/tests/test_init.py b/tests/test_init.py
index 5005a75d9..1d1f7086b 100644
--- a/tests/test_init.py
+++ b/tests/test_init.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf.py """
@@ -127,7 +126,7 @@ def verify_name_change(self, zc, type_, name, number_hosts):
         desc = {'path': '/~paulsm/'}
         info_service = ServiceInfo(
             type_,
-            '%s.%s' % (name, type_),
+            f'{name}.{type_}',
             80,
             0,
             0,
@@ -147,7 +146,7 @@ def verify_name_change(self, zc, type_, name, number_hosts):
         # in the registry
         info_service2 = ServiceInfo(
             type_,
-            '%s.%s' % (name, type_),
+            f'{name}.{type_}',
             80,
             0,
             0,
@@ -160,7 +159,7 @@ def verify_name_change(self, zc, type_, name, number_hosts):
 
     def generate_many_hosts(self, zc, type_, name, number_hosts):
         block_size = 25
-        number_hosts = int(((number_hosts - 1) / block_size + 1)) * block_size
+        number_hosts = int((number_hosts - 1) / block_size + 1) * block_size
         out = r.DNSOutgoing(const._FLAGS_QR_RESPONSE | const._FLAGS_AA)
         for i in range(1, number_hosts + 1):
             next_name = name if i == 1 else '%s-%d' % (name, i)
diff --git a/tests/test_logger.py b/tests/test_logger.py
index 205ce0ff9..cedda7e95 100644
--- a/tests/test_logger.py
+++ b/tests/test_logger.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """Unit tests for logger.py."""
diff --git a/tests/test_protocol.py b/tests/test_protocol.py
index ebdb71105..75a69d5e1 100644
--- a/tests/test_protocol.py
+++ b/tests/test_protocol.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._protocol """
@@ -195,8 +194,8 @@ def test_dns_hinfo(self):
         generated.add_additional_answer(DNSHinfo('irrelevant', const._TYPE_HINFO, 0, 0, 'cpu', 'os'))
         parsed = r.DNSIncoming(generated.packets()[0])
         answer = cast(r.DNSHinfo, parsed.answers[0])
-        assert answer.cpu == u'cpu'
-        assert answer.os == u'os'
+        assert answer.cpu == 'cpu'
+        assert answer.os == 'os'
 
         generated = r.DNSOutgoing(0)
         generated.add_additional_answer(DNSHinfo('irrelevant', const._TYPE_HINFO, 0, 0, 'cpu', 'x' * 257))
@@ -722,6 +721,22 @@ def test_qu_packet_parser():
     assert ",QU," in str(parsed.questions[0])
 
 
+def test_parse_packet_with_nsec_record():
+    """Test we can parse a packet with an NSEC record."""
+    nsec_packet = (
+        b"\x00\x00\x84\x00\x00\x00\x00\x01\x00\x00\x00\x03\x08_meshcop\x04_udp\x05local\x00\x00\x0c\x00"
+        b"\x01\x00\x00\x11\x94\x00\x0f\x0cMyHome54 (2)\xc0\x0c\xc0+\x00\x10\x80\x01\x00\x00\x11\x94\x00"
+        b")\x0bnn=MyHome54\x13xp=695034D148CC4784\x08tv=0.0.0\xc0+\x00!\x80\x01\x00\x00\x00x\x00\x15\x00"
+        b"\x00\x00\x00\xc0'\x0cMaster-Bed-2\xc0\x1a\xc0+\x00/\x80\x01\x00\x00\x11\x94\x00\t\xc0+\x00\x05"
+        b"\x00\x00\x80\x00@"
+    )
+    parsed = DNSIncoming(nsec_packet)
+    nsec_record = parsed.answers[3]
+    assert "nsec," in str(nsec_record)
+    assert nsec_record.rdtypes == [16, 33]
+    assert nsec_record.next_name == "MyHome54 (2)._meshcop._udp.local."
+
+
 def test_records_same_packet_share_fate():
     """Test records in the same packet all have the same created time."""
     out = r.DNSOutgoing(const._FLAGS_QR_QUERY | const._FLAGS_AA)
diff --git a/tests/test_services.py b/tests/test_services.py
index b1e2d8904..7994cbdc5 100644
--- a/tests/test_services.py
+++ b/tests/test_services.py
@@ -1,5 +1,4 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
 """ Unit tests for zeroconf._services. """
@@ -48,7 +47,7 @@ def test_integration_with_listener_class(self):
         type_ = "_http._tcp.local."
         subtype = subtype_name + "._sub." + type_
         name = "UPPERxxxyyyæøå"
-        registration_name = "%s.%s" % (name, subtype)
+        registration_name = f"{name}.{subtype}"
 
         class MyListener(r.ServiceListener):
             def add_service(self, zeroconf, type, name):
diff --git a/tests/test_updates.py b/tests/test_updates.py
index 1f6f8ad4b..ecdf89d4b 100644
--- a/tests/test_updates.py
+++ b/tests/test_updates.py
@@ -1,11 +1,11 @@
 #!/usr/bin/env python
-# -*- coding: utf-8 -*-
 
 
-""" Unit tests for zeroconf._services. """
+""" Unit tests for zeroconf._updates. """
 
 import logging
 import socket
+import time
 from threading import Event
 
 import pytest
@@ -67,7 +67,7 @@ def on_service_state_change(zeroconf, service_type, state_change, name):
 
     info_service = ServiceInfo(
         type_,
-        '%s.%s' % (name, type_),
+        f'{name}.{type_}',
         80,
         0,
         0,
@@ -78,7 +78,7 @@ def on_service_state_change(zeroconf, service_type, state_change, name):
 
     zc.register_service(info_service)
 
-    zc.wait(1)
+    time.sleep(0.001)
 
     browser.cancel()
 
diff --git a/zeroconf/__init__.py b/zeroconf/__init__.py
index 56c61ff3f..a5ba52740 100644
--- a/zeroconf/__init__.py
+++ b/zeroconf/__init__.py
@@ -28,6 +28,7 @@
     DNSAddress,
     DNSEntry,
     DNSHinfo,
+    DNSNsec,
     DNSPointer,
     DNSQuestion,
     DNSRecord,
@@ -79,7 +80,7 @@
 
 __author__ = 'Paul Scott-Murphy, William McBrine'
 __maintainer__ = 'Jakub Stasiak <jakub@stasiak.at>'
-__version__ = '0.32.1'
+__version__ = '0.33.0'
 __license__ = 'LGPL'
 
 
diff --git a/zeroconf/_core.py b/zeroconf/_core.py
index 2f5ef5074..37b72d593 100644
--- a/zeroconf/_core.py
+++ b/zeroconf/_core.py
@@ -21,15 +21,13 @@
 """
 
 import asyncio
-import concurrent.futures
-import contextlib
 import itertools
 import random
 import socket
 import sys
 import threading
 from types import TracebackType  # noqa # used in type hints
-from typing import Dict, List, Optional, Tuple, Type, Union, cast
+from typing import Awaitable, Dict, List, Optional, Tuple, Type, Union, cast
 
 from ._cache import DNSCache
 from ._dns import DNSQuestion, DNSQuestionType
@@ -43,7 +41,13 @@
 from ._services.info import ServiceInfo, instance_name_from_service_info
 from ._services.registry import ServiceRegistry
 from ._updates import RecordUpdate, RecordUpdateListener
-from ._utils.asyncio import get_running_loop, shutdown_loop, wait_event_or_timeout
+from ._utils.asyncio import (
+    await_awaitable,
+    get_running_loop,
+    run_coro_with_timeout,
+    shutdown_loop,
+    wait_event_or_timeout,
+)
 from ._utils.name import service_type_name
 from ._utils.net import (
     IPVersion,
@@ -62,7 +66,6 @@
     _FLAGS_AA,
     _FLAGS_QR_QUERY,
     _FLAGS_QR_RESPONSE,
-    _LOADED_SYSTEM_TIMEOUT,
     _MAX_MSG_ABSOLUTE,
     _MDNS_ADDR,
     _MDNS_ADDR6,
@@ -73,7 +76,8 @@
 )
 
 _TC_DELAY_RANDOM_INTERVAL = (400, 500)
-_CLOSE_TIMEOUT = 3
+_CLOSE_TIMEOUT = 3000  # ms
+_REGISTER_BROADCASTS = 3
 
 
 class AsyncEngine:
@@ -92,7 +96,7 @@ def __init__(
         self.senders: List[asyncio.DatagramTransport] = []
         self._listen_socket = listen_socket
         self._respond_sockets = respond_sockets
-        self._cache_cleanup_task: Optional[asyncio.Task] = None
+        self._cleanup_timer: Optional[asyncio.TimerHandle] = None
         self._running_event: Optional[asyncio.Event] = None
 
     def setup(self, loop: asyncio.AbstractEventLoop, loop_thread_ready: Optional[threading.Event]) -> None:
@@ -104,8 +108,10 @@ def setup(self, loop: asyncio.AbstractEventLoop, loop_thread_ready: Optional[thr
     async def _async_setup(self, loop_thread_ready: Optional[threading.Event]) -> None:
         """Set up the instance."""
         assert self.loop is not None
+        self._cleanup_timer = self.loop.call_later(
+            millis_to_seconds(_CACHE_CLEANUP_INTERVAL), self._async_cache_cleanup
+        )
         await self._async_create_endpoints()
-        self._cache_cleanup_task = self.loop.create_task(self._async_cache_cleanup())
         assert self._running_event is not None
         self._running_event.set()
         if loop_thread_ready:
@@ -136,33 +142,30 @@ async def _async_create_endpoints(self) -> None:
             if s in sender_sockets:
                 self.senders.append(cast(asyncio.DatagramTransport, transport))
 
-    async def _async_cache_cleanup(self) -> None:
+    def _async_cache_cleanup(self) -> None:
         """Periodic cache cleanup."""
-        while not self.zc.done:
-            now = current_time_millis()
-            self.zc.question_history.async_expire(now)
-            self.zc.record_manager.async_updates(
-                now, [RecordUpdate(record, None) for record in self.zc.cache.async_expire(now)]
-            )
-            self.zc.record_manager.async_updates_complete()
-            await asyncio.sleep(millis_to_seconds(_CACHE_CLEANUP_INTERVAL))
+        now = current_time_millis()
+        self.zc.question_history.async_expire(now)
+        self.zc.record_manager.async_updates(
+            now, [RecordUpdate(record, None) for record in self.zc.cache.async_expire(now)]
+        )
+        self.zc.record_manager.async_updates_complete()
+        assert self.loop is not None
+        self._cleanup_timer = self.loop.call_later(
+            millis_to_seconds(_CACHE_CLEANUP_INTERVAL), self._async_cache_cleanup
+        )
 
     async def _async_close(self) -> None:
         """Cancel and wait for the cleanup task to finish."""
         self._async_shutdown()
-        if self._cache_cleanup_task:
-            self._cache_cleanup_task.cancel()
-            with contextlib.suppress(asyncio.CancelledError):
-                await self._cache_cleanup_task
-            self._cache_cleanup_task = None
         await asyncio.sleep(0)  # flush out any call soons
+        assert self._cleanup_timer is not None
+        self._cleanup_timer.cancel()
 
     def _async_shutdown(self) -> None:
         """Shutdown transports and sockets."""
         for transport in itertools.chain(self.senders, self.readers):
             transport.close()
-        for s in self._respond_sockets:
-            s.close()
 
     def close(self) -> None:
         """Close from sync context."""
@@ -173,9 +176,7 @@ def close(self) -> None:
             return
         if not self.loop.is_running():
             return
-        asyncio.run_coroutine_threadsafe(self._async_close(), self.loop).result(
-            _CLOSE_TIMEOUT + _LOADED_SYSTEM_TIMEOUT
-        )
+        run_coro_with_timeout(self._async_close(), self.loop, _CLOSE_TIMEOUT)
 
 
 class AsyncListener(asyncio.Protocol, QuietLogger):
@@ -325,6 +326,9 @@ def error_received(self, exc: Exception) -> None:
     def connection_made(self, transport: asyncio.BaseTransport) -> None:
         self.transport = cast(asyncio.DatagramTransport, transport)
 
+    def connection_lost(self, exc: Optional[Exception]) -> None:
+        """Handle connection lost."""
+
 
 class Zeroconf(QuietLogger):
 
@@ -418,14 +422,6 @@ def done(self) -> bool:
     def listeners(self) -> List[RecordUpdateListener]:
         return self.record_manager.listeners
 
-    def wait(self, timeout: float) -> None:
-        """Calling task waits for a given number of milliseconds or until notified."""
-        assert self.loop is not None
-        with contextlib.suppress(concurrent.futures.TimeoutError):
-            asyncio.run_coroutine_threadsafe(self.async_wait(timeout), self.loop).result(
-                millis_to_seconds(timeout)
-            )
-
     async def async_wait(self, timeout: float) -> None:
         """Calling task waits for a given number of milliseconds or until notified."""
         assert self.notify_event is not None
@@ -478,6 +474,28 @@ def register_service(
         allow_name_change: bool = False,
         cooperating_responders: bool = False,
     ) -> None:
+        """Registers service information to the network with a default TTL.
+        Zeroconf will then respond to requests for information for that
+        service.  The name of the service may be changed if needed to make
+        it unique on the network. Additionally multiple cooperating responders
+        can register the same service on the network for resilience
+        (if you want this behavior set `cooperating_responders` to `True`)."""
+        assert self.loop is not None
+        run_coro_with_timeout(
+            await_awaitable(
+                self.async_register_service(info, ttl, allow_name_change, cooperating_responders)
+            ),
+            self.loop,
+            _REGISTER_TIME * _REGISTER_BROADCASTS,
+        )
+
+    async def async_register_service(
+        self,
+        info: ServiceInfo,
+        ttl: Optional[int] = None,
+        allow_name_change: bool = False,
+        cooperating_responders: bool = False,
+    ) -> Awaitable:
         """Registers service information to the network with a default TTL.
         Zeroconf will then respond to requests for information for that
         service.  The name of the service may be changed if needed to make
@@ -489,36 +507,34 @@ def register_service(
             # Setting TTLs via ServiceInfo is preferred
             info.host_ttl = ttl
             info.other_ttl = ttl
-        self.check_service(info, allow_name_change, cooperating_responders)
-        self.registry.add(info)
-        self._broadcast_service(info, _REGISTER_TIME, None)
+
+        await self.async_wait_for_start()
+        await self.async_check_service(info, allow_name_change, cooperating_responders)
+        self.registry.async_add(info)
+        return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
 
     def update_service(self, info: ServiceInfo) -> None:
         """Registers service information to the network with a default TTL.
         Zeroconf will then respond to requests for information for that
         service."""
+        assert self.loop is not None
+        run_coro_with_timeout(
+            await_awaitable(self.async_update_service(info)), self.loop, _REGISTER_TIME * _REGISTER_BROADCASTS
+        )
 
-        self.registry.update(info)
-        self._broadcast_service(info, _REGISTER_TIME, None)
+    async def async_update_service(self, info: ServiceInfo) -> Awaitable:
+        """Registers service information to the network with a default TTL.
+        Zeroconf will then respond to requests for information for that
+        service."""
+        self.registry.async_update(info)
+        return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
 
-    def _broadcast_service(self, info: ServiceInfo, interval: int, ttl: Optional[int]) -> None:
+    async def _async_broadcast_service(self, info: ServiceInfo, interval: int, ttl: Optional[int]) -> None:
         """Send a broadcasts to announce a service at intervals."""
-        now = current_time_millis()
-        next_time = now
-        i = 0
-        while i < 3:
-            if now < next_time:
-                self.wait(next_time - now)
-                now = current_time_millis()
-                continue
-
-            self.send_service_broadcast(info, ttl)
-            i += 1
-            next_time += interval
-
-    def send_service_broadcast(self, info: ServiceInfo, ttl: Optional[int]) -> None:
-        """Send a broadcast to announce a service."""
-        self.send(self.generate_service_broadcast(info, ttl))
+        for i in range(_REGISTER_BROADCASTS):
+            if i != 0:
+                await asyncio.sleep(millis_to_seconds(interval))
+            self.async_send(self.generate_service_broadcast(info, ttl))
 
     def generate_service_broadcast(self, info: ServiceInfo, ttl: Optional[int]) -> DNSOutgoing:
         """Generate a broadcast to announce a service."""
@@ -526,10 +542,6 @@ def generate_service_broadcast(self, info: ServiceInfo, ttl: Optional[int]) -> D
         self._add_broadcast_answer(out, info, ttl)
         return out
 
-    def send_service_query(self, info: ServiceInfo) -> None:
-        """Send a query to lookup a service."""
-        self.send(self.generate_service_query(info))
-
     def generate_service_query(self, info: ServiceInfo) -> DNSOutgoing:  # pylint: disable=no-self-use
         """Generate a query to lookup a service."""
         out = DNSOutgoing(_FLAGS_QR_QUERY | _FLAGS_AA)
@@ -560,39 +572,51 @@ def _add_broadcast_answer(  # pylint: disable=no-self-use
 
     def unregister_service(self, info: ServiceInfo) -> None:
         """Unregister a service."""
-        self.registry.remove(info)
-        self._broadcast_service(info, _UNREGISTER_TIME, 0)
+        assert self.loop is not None
+        run_coro_with_timeout(
+            self.async_unregister_service(info), self.loop, _UNREGISTER_TIME * _REGISTER_BROADCASTS
+        )
+
+    async def async_unregister_service(self, info: ServiceInfo) -> Awaitable:
+        """Unregister a service."""
+        self.registry.async_remove(info)
+        return asyncio.ensure_future(self._async_broadcast_service(info, _UNREGISTER_TIME, 0))
 
     def generate_unregister_all_services(self) -> Optional[DNSOutgoing]:
         """Generate a DNSOutgoing goodbye for all services and remove them from the registry."""
-        service_infos = self.registry.get_service_infos()
+        service_infos = self.registry.async_get_service_infos()
         if not service_infos:
             return None
         out = DNSOutgoing(_FLAGS_QR_RESPONSE | _FLAGS_AA)
         for info in service_infos:
             self._add_broadcast_answer(out, info, 0)
-        self.registry.remove(service_infos)
+        self.registry.async_remove(service_infos)
         return out
 
-    def unregister_all_services(self) -> None:
-        """Unregister all registered services."""
+    async def async_unregister_all_services(self) -> None:
+        """Unregister all registered services.
+
+        Unlike async_register_service and async_unregister_service, this
+        method does not return a future and is always expected to be
+        awaited since its only called at shutdown.
+        """
         # Send Goodbye packets https://datatracker.ietf.org/doc/html/rfc6762#section-10.1
         out = self.generate_unregister_all_services()
         if not out:
             return
-        now = current_time_millis()
-        next_time = now
-        i = 0
-        while i < 3:
-            if now < next_time:
-                self.wait(next_time - now)
-                now = current_time_millis()
-                continue
-            self.send(out)
-            i += 1
-            next_time += _UNREGISTER_TIME
+        for i in range(_REGISTER_BROADCASTS):
+            if i != 0:
+                await asyncio.sleep(millis_to_seconds(_UNREGISTER_TIME))
+            self.async_send(out)
 
-    def check_service(
+    def unregister_all_services(self) -> None:
+        """Unregister all registered services."""
+        assert self.loop is not None
+        run_coro_with_timeout(
+            self.async_unregister_all_services(), self.loop, _UNREGISTER_TIME * _REGISTER_BROADCASTS
+        )
+
+    async def async_check_service(
         self, info: ServiceInfo, allow_name_change: bool, cooperating_responders: bool = False
     ) -> None:
         """Checks the network for a unique service name, modifying the
@@ -603,25 +627,25 @@ def check_service(
         next_instance_number = 2
         next_time = now = current_time_millis()
         i = 0
-        while i < 3:
+        while i < _REGISTER_BROADCASTS:
             # check for a name conflict
             while self.cache.current_entry_with_name_and_alias(info.type, info.name):
                 if not allow_name_change:
                     raise NonUniqueNameException
 
                 # change the name and look for a conflict
-                info.name = '%s-%s.%s' % (instance_name, next_instance_number, info.type)
+                info.name = f'{instance_name}-{next_instance_number}.{info.type}'
                 next_instance_number += 1
                 service_type_name(info.name)
                 next_time = now
                 i = 0
 
             if now < next_time:
-                self.wait(next_time - now)
+                await self.async_wait(next_time - now)
                 now = current_time_millis()
                 continue
 
-            self.send_service_query(info)
+            self.async_send(self.generate_service_query(info))
             i += 1
             next_time += _CHECK_TIME
 
@@ -756,7 +780,14 @@ def close(self) -> None:
 
         This method is idempotent and irreversible.
         """
-        self.unregister_all_services()
+        assert self.loop is not None
+        if self.loop.is_running():
+            if self.loop == get_running_loop():
+                log.warning(
+                    "unregister_all_services skipped as it does blocking i/o; use AsyncZeroconf with asyncio"
+                )
+            else:
+                self.unregister_all_services()
         self._close()
         self.engine.close()
         self._shutdown_threads()
diff --git a/zeroconf/_dns.py b/zeroconf/_dns.py
index 93db9859c..0f7a5e114 100644
--- a/zeroconf/_dns.py
+++ b/zeroconf/_dns.py
@@ -22,7 +22,7 @@
 
 import enum
 import socket
-from typing import Any, Dict, Iterable, Optional, TYPE_CHECKING, Tuple, Union, cast
+from typing import Any, Dict, Iterable, List, Optional, TYPE_CHECKING, Tuple, Union, cast
 
 from ._exceptions import AbstractMethodException
 from ._utils.net import _is_v6_address
@@ -31,9 +31,6 @@
     _CLASSES,
     _CLASS_MASK,
     _CLASS_UNIQUE,
-    _EXPIRE_FULL_TIME_PERCENT,
-    _EXPIRE_STALE_TIME_PERCENT,
-    _RECENT_TIME_PERCENT,
     _TYPES,
     _TYPE_ANY,
 )
@@ -45,6 +42,11 @@
 _BASE_MAX_SIZE = _LEN_SHORT + _LEN_SHORT + _LEN_INT + _LEN_SHORT  # type  # class  # ttl  # length
 _NAME_COMPRESSION_MIN_SIZE = _LEN_BYTE * 2
 
+_EXPIRE_FULL_TIME_MS = 1000
+_EXPIRE_STALE_TIME_MS = 500
+_RECENT_TIME_MS = 250
+
+
 if TYPE_CHECKING:
     # https://github.com/PyCQA/pylint/issues/3525
     from ._protocol import DNSIncoming, DNSOutgoing  # pylint: disable=cyclic-import
@@ -100,7 +102,7 @@ def get_type(t: int) -> str:
 
     def entry_to_string(self, hdr: str, other: Optional[Union[bytes, str]]) -> str:
         """String representation with additional information"""
-        return "%s[%s,%s%s,%s]%s" % (
+        return "{}[{},{}{},{}]{}".format(
             hdr,
             self.get_type(self.type),
             self.get_class_(self.class_),
@@ -116,11 +118,7 @@ class DNSQuestion(DNSEntry):
 
     def answered_by(self, rec: 'DNSRecord') -> bool:
         """Returns true if the question is answered by the record"""
-        return (
-            self.class_ == rec.class_
-            and (self.type == rec.type or self.type == _TYPE_ANY)
-            and self.name == rec.name
-        )
+        return self.class_ == rec.class_ and self.type in (rec.type, _TYPE_ANY) and self.name == rec.name
 
     def __hash__(self) -> int:
         return hash((self.name, self.class_, self.type))
@@ -146,7 +144,7 @@ def unicast(self, value: bool) -> None:
 
     def __repr__(self) -> str:
         """String representation"""
-        return "%s[question,%s,%s,%s]" % (
+        return "{}[question,{},{},{}]".format(
             self.get_type(self.type),
             "QU" if self.unicast else "QM",
             self.get_class_(self.class_),
@@ -158,7 +156,7 @@ class DNSRecord(DNSEntry):
 
     """A DNS record - like a DNS entry, but has a TTL"""
 
-    __slots__ = ('ttl', 'created', '_expiration_time', '_stale_time', '_recent_time')
+    __slots__ = ('ttl', 'created')
 
     # TODO: Switch to just int ttl
     def __init__(
@@ -167,9 +165,6 @@ def __init__(
         super().__init__(name, type_, class_)
         self.ttl = ttl
         self.created = created or current_time_millis()
-        self._expiration_time: Optional[float] = None
-        self._stale_time: Optional[float] = None
-        self._recent_time: Optional[float] = None
 
     def __eq__(self, other: Any) -> bool:  # pylint: disable=no-self-use
         """Abstract method"""
@@ -193,27 +188,19 @@ def get_expiration_time(self, percent: int) -> float:
     # TODO: Switch to just int here
     def get_remaining_ttl(self, now: float) -> Union[int, float]:
         """Returns the remaining TTL in seconds."""
-        if self._expiration_time is None:
-            self._expiration_time = self.get_expiration_time(_EXPIRE_FULL_TIME_PERCENT)
-        return max(0, millis_to_seconds(self._expiration_time - now))
+        return max(0, millis_to_seconds((self.created + (_EXPIRE_FULL_TIME_MS * self.ttl)) - now))
 
     def is_expired(self, now: float) -> bool:
         """Returns true if this record has expired."""
-        if self._expiration_time is None:
-            self._expiration_time = self.get_expiration_time(_EXPIRE_FULL_TIME_PERCENT)
-        return self._expiration_time <= now
+        return self.created + (_EXPIRE_FULL_TIME_MS * self.ttl) <= now
 
     def is_stale(self, now: float) -> bool:
         """Returns true if this record is at least half way expired."""
-        if self._stale_time is None:
-            self._stale_time = self.get_expiration_time(_EXPIRE_STALE_TIME_PERCENT)
-        return self._stale_time <= now
+        return self.created + (_EXPIRE_STALE_TIME_MS * self.ttl) <= now
 
     def is_recent(self, now: float) -> bool:
         """Returns true if the record more than one quarter of its TTL remaining."""
-        if self._recent_time is None:
-            self._recent_time = self.get_expiration_time(_RECENT_TIME_PERCENT)
-        return self._recent_time > now
+        return self.created + (_RECENT_TIME_MS * self.ttl) > now
 
     def reset_ttl(self, other: 'DNSRecord') -> None:
         """Sets this record's TTL and created time to that of
@@ -224,9 +211,6 @@ def set_created_ttl(self, created: float, ttl: Union[float, int]) -> None:
         """Set the created and ttl of a record."""
         self.created = created
         self.ttl = ttl
-        self._expiration_time = None
-        self._stale_time = None
-        self._recent_time = None
 
     def write(self, out: 'DNSOutgoing') -> None:  # pylint: disable=no-self-use
         """Abstract method"""
@@ -234,7 +218,7 @@ def write(self, out: 'DNSOutgoing') -> None:  # pylint: disable=no-self-use
 
     def to_string(self, other: Union[bytes, str]) -> str:
         """String representation with additional information"""
-        arg = "%s/%s,%s" % (self.ttl, int(self.get_remaining_ttl(current_time_millis())), cast(Any, other))
+        arg = f"{self.ttl}/{int(self.get_remaining_ttl(current_time_millis()))},{cast(Any, other)}"
         return DNSEntry.entry_to_string(self, "record", arg)
 
 
@@ -443,7 +427,47 @@ def __hash__(self) -> int:
 
     def __repr__(self) -> str:
         """String representation"""
-        return self.to_string("%s:%s" % (self.server, self.port))
+        return self.to_string(f"{self.server}:{self.port}")
+
+
+class DNSNsec(DNSRecord):
+
+    """A DNS NSEC record"""
+
+    __slots__ = ('next_name', 'rdtypes')
+
+    def __init__(
+        self,
+        name: str,
+        type_: int,
+        class_: int,
+        ttl: int,
+        next_name: str,
+        rdtypes: List[int],
+        created: Optional[float] = None,
+    ) -> None:
+        super().__init__(name, type_, class_, ttl, created)
+        self.next_name = next_name
+        self.rdtypes = rdtypes
+
+    def __eq__(self, other: Any) -> bool:
+        """Tests equality on cpu and os"""
+        return (
+            isinstance(other, DNSNsec)
+            and self.next_name == other.next_name
+            and self.rdtypes == other.rdtypes
+            and DNSEntry.__eq__(self, other)
+        )
+
+    def __hash__(self) -> int:
+        """Hash to compare like DNSNSec."""
+        return hash((*self._entry_tuple(), self.next_name, *self.rdtypes))
+
+    def __repr__(self) -> str:
+        """String representation"""
+        return self.to_string(
+            self.next_name + "," + "|".join([self.get_type(type_) for type_ in self.rdtypes])
+        )
 
 
 class DNSRRSet:
diff --git a/zeroconf/_handlers.py b/zeroconf/_handlers.py
index 617be4083..07b9ac833 100644
--- a/zeroconf/_handlers.py
+++ b/zeroconf/_handlers.py
@@ -140,11 +140,11 @@ def _construct_outgoing_from_record_set(
 
     def _additionals_from_answers_rrset(self, rrset: Set[DNSRecord]) -> Set[DNSRecord]:
         additionals: Set[DNSRecord] = set()
-        return additionals.union(*[self._additionals[record] for record in rrset])
+        return additionals.union(*(self._additionals[record] for record in rrset))
 
     def _suppress_mcasts_from_last_second(self, rrset: Set[DNSRecord]) -> None:
         """Remove any records that were already sent in the last second."""
-        rrset -= set(record for record in rrset if self._has_mcast_record_in_last_second(record))
+        rrset -= {record for record in rrset if self._has_mcast_record_in_last_second(record)}
 
     def _has_mcast_within_one_quarter_ttl(self, record: DNSRecord) -> bool:
         """Check to see if a record has been mcasted recently.
@@ -185,7 +185,7 @@ def _add_service_type_enumeration_query_answers(
 
         https://datatracker.ietf.org/doc/html/rfc6763#section-9
         """
-        for stype in self.registry.get_types():
+        for stype in self.registry.async_get_types():
             dns_pointer = DNSPointer(
                 _SERVICE_TYPE_ENUMERATION_NAME, _TYPE_PTR, _CLASS_IN, _DNS_OTHER_TTL, stype, now
             )
@@ -196,18 +196,16 @@ def _add_pointer_answers(
         self, name: str, answer_set: _AnswerWithAdditionalsType, known_answers: DNSRRSet, now: float
     ) -> None:
         """Answer PTR/ANY question."""
-        for service in self.registry.get_infos_type(name):
+        for service in self.registry.async_get_infos_type(name):
             # Add recommended additional answers according to
             # https://tools.ietf.org/html/rfc6763#section-12.1.
             dns_pointer = service.dns_pointer(created=now)
             if not known_answers.suppresses(dns_pointer):
-                answer_set[dns_pointer] = set(
-                    [
-                        service.dns_service(created=now),
-                        service.dns_text(created=now),
-                        *service.dns_addresses(created=now),
-                    ]
-                )
+                answer_set[dns_pointer] = {
+                    service.dns_service(created=now),
+                    service.dns_text(created=now),
+                    *service.dns_addresses(created=now),
+                }
 
     def _add_address_answers(
         self,
@@ -218,7 +216,7 @@ def _add_address_answers(
         type_: int,
     ) -> None:
         """Answer A/AAAA/ANY question."""
-        for service in self.registry.get_infos_server(name):
+        for service in self.registry.async_get_infos_server(name):
             answers: List[DNSAddress] = []
             additionals: Set[DNSRecord] = set()
             for dns_address in service.dns_addresses(created=now):
@@ -249,7 +247,7 @@ def _answer_question(
             self._add_address_answers(question.name, answer_set, known_answers, now, type_)
 
         if type_ in (_TYPE_SRV, _TYPE_TXT, _TYPE_ANY):
-            service = self.registry.get_info_name(question.name)  # type: ignore
+            service = self.registry.async_get_info_name(question.name)  # type: ignore
             if service is not None:
                 if type_ in (_TYPE_SRV, _TYPE_ANY):
                     # Add recommended additional answers according to
@@ -271,7 +269,7 @@ def async_response(  # pylint: disable=unused-argument
         threadsafe.
         """
         ucast_source = port != _MDNS_PORT
-        known_answers = DNSRRSet(itertools.chain(*[msg.answers for msg in msgs]))
+        known_answers = DNSRRSet(itertools.chain(*(msg.answers for msg in msgs)))
         query_res = _QueryResponse(self.cache, msgs[0], ucast_source)
 
         for msg in msgs:
diff --git a/zeroconf/_protocol.py b/zeroconf/_protocol.py
index 79f483def..f987ce2e2 100644
--- a/zeroconf/_protocol.py
+++ b/zeroconf/_protocol.py
@@ -25,7 +25,7 @@
 from typing import Any, Dict, List, Optional, TYPE_CHECKING, Tuple, Union, cast
 
 
-from ._dns import DNSAddress, DNSHinfo, DNSPointer, DNSQuestion, DNSRecord, DNSService, DNSText
+from ._dns import DNSAddress, DNSHinfo, DNSNsec, DNSPointer, DNSQuestion, DNSRecord, DNSService, DNSText
 from ._exceptions import IncomingDecodeError, NamePartTooLongException
 from ._logger import QuietLogger, log
 from ._utils.struct import int2byte
@@ -43,6 +43,7 @@
     _TYPE_AAAA,
     _TYPE_CNAME,
     _TYPE_HINFO,
+    _TYPE_NSEC,
     _TYPE_PTR,
     _TYPE_SRV,
     _TYPE_TXT,
@@ -201,6 +202,18 @@ def read_others(self) -> None:
                 rec = DNSAddress(
                     domain, type_, class_, ttl, self.read_string(16), created=self.now, scope_id=self.scope_id
                 )
+            elif type_ == _TYPE_NSEC:
+                name_start = self.offset
+                name = self.read_name()
+                rec = DNSNsec(
+                    domain,
+                    type_,
+                    class_,
+                    ttl,
+                    name,
+                    self.read_bitmap(name_start + length),
+                    self.now,
+                )
             else:
                 # Try to ignore types we don't know about
                 # Skip the payload for the resource record so the next
@@ -210,6 +223,19 @@ def read_others(self) -> None:
             if rec is not None:
                 self.answers.append(rec)
 
+    def read_bitmap(self, end: int) -> List[int]:
+        """Reads an NSEC bitmap from the packet."""
+        rdtypes = []
+        while self.offset < end:
+            window = self.data[self.offset]
+            bitmap_length = self.data[self.offset + 1]
+            for i, byte in enumerate(self.data[self.offset + 2 : self.offset + 2 + bitmap_length]):
+                for bit in range(0, 8):
+                    if byte & (0x80 >> bit):
+                        rdtypes.append(bit + window * 256 + i * 8)
+            self.offset += 2 + bitmap_length
+        return rdtypes
+
     def read_name(self) -> str:
         """Reads a domain name from the packet"""
         result = ''
@@ -233,10 +259,10 @@ def read_name(self) -> str:
                     next_ = off + 1
                 off = ((length & 0x3F) << 8) | self.data[off]
                 if off >= first:
-                    raise IncomingDecodeError("Bad domain name (circular) at %s" % (off,))
+                    raise IncomingDecodeError(f"Bad domain name (circular) at {off}")
                 first = off
             else:
-                raise IncomingDecodeError("Bad domain name at %s" % (off,))
+                raise IncomingDecodeError(f"Bad domain name at {off}")
 
         if next_ >= 0:
             self.offset = next_
@@ -497,7 +523,7 @@ def _write_record(self, record: DNSRecord, now: float) -> bool:
         self.write_short(0)  # Will get replaced with the actual size
         record.write(self)
         # Adjust size for the short we will write before this record
-        length = sum((len(d) for d in self.data[index + 1 :]))
+        length = sum(len(d) for d in self.data[index + 1 :])
         # Here we replace the 0 length short we wrote
         # before with the actual length
         self._replace_short(index, length)
diff --git a/zeroconf/_services/info.py b/zeroconf/_services/info.py
index 7bc81c8df..d1bf17e9b 100644
--- a/zeroconf/_services/info.py
+++ b/zeroconf/_services/info.py
@@ -20,7 +20,6 @@
     USA
 """
 
-import asyncio
 import ipaddress
 import socket
 from typing import Any, Dict, List, Optional, TYPE_CHECKING, Union, cast
@@ -29,7 +28,7 @@
 from .._exceptions import BadTypeInNameException
 from .._protocol import DNSOutgoing
 from .._updates import RecordUpdate, RecordUpdateListener
-from .._utils.asyncio import get_running_loop
+from .._utils.asyncio import get_running_loop, run_coro_with_timeout
 from .._utils.name import service_type_name
 from .._utils.net import (
     IPVersion,
@@ -37,7 +36,7 @@
     _is_v6_address,
 )
 from .._utils.struct import int2byte
-from .._utils.time import current_time_millis, millis_to_seconds
+from .._utils.time import current_time_millis
 from ..const import (
     _CLASS_IN,
     _CLASS_UNIQUE,
@@ -45,7 +44,6 @@
     _DNS_OTHER_TTL,
     _FLAGS_QR_QUERY,
     _LISTENER_TIME,
-    _LOADED_SYSTEM_TIMEOUT,
     _TYPE_A,
     _TYPE_AAAA,
     _TYPE_PTR,
@@ -426,9 +424,7 @@ def request(
         assert zc.loop is not None and zc.loop.is_running()
         if zc.loop == get_running_loop():
             raise RuntimeError("Use AsyncServiceInfo.async_request from the event loop")
-        return asyncio.run_coroutine_threadsafe(
-            self.async_request(zc, timeout, question_type), zc.loop
-        ).result(millis_to_seconds(timeout) + _LOADED_SYSTEM_TIMEOUT)
+        return bool(run_coro_with_timeout(self.async_request(zc, timeout, question_type), zc.loop, timeout))
 
     async def async_request(
         self, zc: 'Zeroconf', timeout: float, question_type: Optional[DNSQuestionType] = None
diff --git a/zeroconf/_services/registry.py b/zeroconf/_services/registry.py
index ebf5abbb6..4e64c8d7b 100644
--- a/zeroconf/_services/registry.py
+++ b/zeroconf/_services/registry.py
@@ -20,7 +20,6 @@
     USA
 """
 
-import threading
 from typing import Dict, List, Optional, Union
 
 
@@ -31,9 +30,8 @@
 class ServiceRegistry:
     """A registry to keep track of services.
 
-    This class exists to ensure services can
-    be safely added and removed with thread
-    safety.
+    The registry must only be accessed from
+    the event loop as it is not thread safe.
     """
 
     def __init__(
@@ -43,56 +41,43 @@ def __init__(
         self._services: Dict[str, ServiceInfo] = {}
         self.types: Dict[str, List] = {}
         self.servers: Dict[str, List] = {}
-        self._lock = threading.Lock()  # add and remove services thread safe
 
-    def add(self, info: ServiceInfo) -> None:
+    def async_add(self, info: ServiceInfo) -> None:
         """Add a new service to the registry."""
-        with self._lock:
-            self._add(info)
+        self._add(info)
 
-    def remove(self, info: Union[List[ServiceInfo], ServiceInfo]) -> None:
+    def async_remove(self, info: Union[List[ServiceInfo], ServiceInfo]) -> None:
         """Remove a new service from the registry."""
-        infos = info if isinstance(info, list) else [info]
+        self._remove(info if isinstance(info, list) else [info])
 
-        with self._lock:
-            self._remove(infos)
-
-    def update(self, info: ServiceInfo) -> None:
+    def async_update(self, info: ServiceInfo) -> None:
         """Update new service in the registry."""
+        self._remove([info])
+        self._add(info)
 
-        with self._lock:
-            self._remove([info])
-            self._add(info)
-
-    def get_service_infos(self) -> List[ServiceInfo]:
+    def async_get_service_infos(self) -> List[ServiceInfo]:
         """Return all ServiceInfo."""
         return list(self._services.values())
 
-    def get_info_name(self, name: str) -> Optional[ServiceInfo]:
+    def async_get_info_name(self, name: str) -> Optional[ServiceInfo]:
         """Return all ServiceInfo for the name."""
         return self._services.get(name.lower())
 
-    def get_types(self) -> List[str]:
+    def async_get_types(self) -> List[str]:
         """Return all types."""
         return list(self.types.keys())
 
-    def get_infos_type(self, type_: str) -> List[ServiceInfo]:
+    def async_get_infos_type(self, type_: str) -> List[ServiceInfo]:
         """Return all ServiceInfo matching type."""
-        return self._get_by_index("types", type_)
+        return self._async_get_by_index("types", type_)
 
-    def get_infos_server(self, server: str) -> List[ServiceInfo]:
+    def async_get_infos_server(self, server: str) -> List[ServiceInfo]:
         """Return all ServiceInfo matching server."""
-        return self._get_by_index("servers", server)
+        return self._async_get_by_index("servers", server)
 
-    def _get_by_index(self, attr: str, key: str) -> List[ServiceInfo]:
+    def _async_get_by_index(self, attr: str, key: str) -> List[ServiceInfo]:
         """Return all ServiceInfo matching the index."""
-        # Since we do not get under a lock since it would be
-        # a performance issue, its possible
-        # the service can be unregistered during the get
-        # so we must check if info is None
-        return list(
-            filter(None, [self._services.get(name) for name in getattr(self, attr).get(key.lower(), [])[:]])
-        )
+        return [self._services[name] for name in getattr(self, attr).get(key.lower(), [])]
 
     def _add(self, info: ServiceInfo) -> None:
         """Add a new service under the lock."""
diff --git a/zeroconf/_utils/asyncio.py b/zeroconf/_utils/asyncio.py
index c68c0f00e..10b8b3d97 100644
--- a/zeroconf/_utils/asyncio.py
+++ b/zeroconf/_utils/asyncio.py
@@ -23,8 +23,12 @@
 import asyncio
 import contextlib
 import queue
-from typing import Any, List, Optional, Set, cast
+from typing import Any, Awaitable, Coroutine, List, Optional, Set, cast
 
+from .time import millis_to_seconds
+from ..const import _LOADED_SYSTEM_TIMEOUT
+
+# The combined timeouts should be lower than _CLOSE_TIMEOUT + _WAIT_FOR_LOOP_TASKS_TIMEOUT
 _TASK_AWAIT_TIMEOUT = 1
 _GET_ALL_TASKS_TIMEOUT = 3
 _WAIT_FOR_LOOP_TASKS_TIMEOUT = 3  # Must be larger than _TASK_AWAIT_TIMEOUT
@@ -80,6 +84,19 @@ async def _wait_for_loop_tasks(wait_tasks: Set[asyncio.Task]) -> None:
     await asyncio.wait(wait_tasks, timeout=_TASK_AWAIT_TIMEOUT)
 
 
+async def await_awaitable(aw: Awaitable) -> None:
+    """Wait on an awaitable and the task it returns."""
+    task = await aw
+    await task
+
+
+def run_coro_with_timeout(aw: Coroutine, loop: asyncio.AbstractEventLoop, timeout: float) -> Any:
+    """Run a coroutine with a timeout."""
+    return asyncio.run_coroutine_threadsafe(aw, loop).result(
+        millis_to_seconds(timeout) + _LOADED_SYSTEM_TIMEOUT
+    )
+
+
 def shutdown_loop(loop: asyncio.AbstractEventLoop) -> None:
     """Wait for pending tasks and stop an event loop."""
     pending_tasks = set(
diff --git a/zeroconf/asyncio.py b/zeroconf/asyncio.py
index 67ff1c120..ef7e7f649 100644
--- a/zeroconf/asyncio.py
+++ b/zeroconf/asyncio.py
@@ -26,20 +26,15 @@
 
 from ._core import Zeroconf
 from ._dns import DNSQuestionType
-from ._exceptions import NonUniqueNameException
 from ._services import ServiceListener
 from ._services.browser import _ServiceBrowserBase
-from ._services.info import ServiceInfo, instance_name_from_service_info
+from ._services.info import ServiceInfo
 from ._services.types import ZeroconfServiceTypes
 from ._utils.net import IPVersion, InterfaceChoice, InterfacesType
-from ._utils.time import millis_to_seconds
 from .const import (
     _BROWSER_TIME,
-    _CHECK_TIME,
     _MDNS_PORT,
-    _REGISTER_TIME,
     _SERVICE_TYPE_ENUMERATION_NAME,
-    _UNREGISTER_TIME,
 )
 
 
@@ -172,16 +167,11 @@ def __init__(
         )
         self.async_browsers: Dict[ServiceListener, AsyncServiceBrowser] = {}
 
-    async def _async_broadcast_service(self, info: ServiceInfo, interval: int, ttl: Optional[int]) -> None:
-        """Send a broadcasts to announce a service at intervals."""
-        for i in range(3):
-            if i != 0:
-                await asyncio.sleep(millis_to_seconds(interval))
-            self.zeroconf.async_send(self.zeroconf.generate_service_broadcast(info, ttl))
-
     async def async_register_service(
         self,
         info: ServiceInfo,
+        ttl: Optional[int] = None,
+        allow_name_change: bool = False,
         cooperating_responders: bool = False,
     ) -> Awaitable:
         """Registers service information to the network with a default TTL.
@@ -194,10 +184,9 @@ async def async_register_service(
         The service will be broadcast in a task. This task is returned
         and therefore can be awaited if necessary.
         """
-        await self.zeroconf.async_wait_for_start()
-        await self.async_check_service(info, cooperating_responders)
-        self.zeroconf.registry.add(info)
-        return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
+        return await self.zeroconf.async_register_service(
+            info, ttl, allow_name_change, cooperating_responders
+        )
 
     async def async_unregister_all_services(self) -> None:
         """Unregister all registered services.
@@ -206,30 +195,7 @@ async def async_unregister_all_services(self) -> None:
         method does not return a future and is always expected to be
         awaited since its only called at shutdown.
         """
-        out = self.zeroconf.generate_unregister_all_services()
-        if not out:
-            return
-        for i in range(3):
-            if i != 0:
-                await asyncio.sleep(millis_to_seconds(_UNREGISTER_TIME))
-            self.zeroconf.async_send(out)
-
-    async def async_check_service(self, info: ServiceInfo, cooperating_responders: bool = False) -> None:
-        """Checks the network for a unique service name."""
-        instance_name_from_service_info(info)
-        if cooperating_responders:
-            return
-        self._raise_on_name_conflict(info)
-        for i in range(3):
-            if i != 0:
-                await asyncio.sleep(millis_to_seconds(_CHECK_TIME))
-            self.zeroconf.async_send(self.zeroconf.generate_service_query(info))
-            self._raise_on_name_conflict(info)
-
-    def _raise_on_name_conflict(self, info: ServiceInfo) -> None:
-        """Raise NonUniqueNameException if the ServiceInfo has a conflict."""
-        if self.zeroconf.cache.current_entry_with_name_and_alias(info.type, info.name):
-            raise NonUniqueNameException
+        await self.zeroconf.async_unregister_all_services()
 
     async def async_unregister_service(self, info: ServiceInfo) -> Awaitable:
         """Unregister a service.
@@ -237,8 +203,7 @@ async def async_unregister_service(self, info: ServiceInfo) -> Awaitable:
         The service will be broadcast in a task. This task is returned
         and therefore can be awaited if necessary.
         """
-        self.zeroconf.registry.remove(info)
-        return asyncio.ensure_future(self._async_broadcast_service(info, _UNREGISTER_TIME, 0))
+        return await self.zeroconf.async_unregister_service(info)
 
     async def async_update_service(self, info: ServiceInfo) -> Awaitable:
         """Registers service information to the network with a default TTL.
@@ -248,8 +213,7 @@ async def async_update_service(self, info: ServiceInfo) -> Awaitable:
         The service will be broadcast in a task. This task is returned
         and therefore can be awaited if necessary.
         """
-        self.zeroconf.registry.update(info)
-        return asyncio.ensure_future(self._async_broadcast_service(info, _REGISTER_TIME, None))
+        return await self.zeroconf.async_update_service(info)
 
     async def async_close(self) -> None:
         """Ends the background threads, and prevent this instance from
@@ -287,7 +251,7 @@ async def async_remove_service_listener(self, listener: ServiceListener) -> None
     async def async_remove_all_service_listeners(self) -> None:
         """Removes a listener from the set that is currently listening."""
         await asyncio.gather(
-            *[self.async_remove_service_listener(listener) for listener in list(self.async_browsers)]
+            *(self.async_remove_service_listener(listener) for listener in list(self.async_browsers))
         )
 
     async def __aenter__(self) -> 'AsyncZeroconf':
diff --git a/zeroconf/const.py b/zeroconf/const.py
index afdcb2d4b..27dc817f8 100644
--- a/zeroconf/const.py
+++ b/zeroconf/const.py
@@ -103,6 +103,7 @@
 _TYPE_TXT = 16
 _TYPE_AAAA = 28
 _TYPE_SRV = 33
+_TYPE_NSEC = 47
 _TYPE_ANY = 255
 
 # Mapping constants to names
@@ -136,6 +137,7 @@
     _TYPE_AAAA: "quada",
     _TYPE_SRV: "srv",
     _TYPE_ANY: "any",
+    _TYPE_NSEC: "nsec",
 }
 
 _HAS_A_TO_Z = re.compile(r'[A-Za-z]')
@@ -143,10 +145,7 @@
 _HAS_ONLY_A_TO_Z_NUM_HYPHEN_UNDERSCORE = re.compile(r'^[A-Za-z0-9\-\_]+$')
 _HAS_ASCII_CONTROL_CHARS = re.compile(r'[\x00-\x1f\x7f]')
 
-_EXPIRE_FULL_TIME_PERCENT = 100
-_EXPIRE_STALE_TIME_PERCENT = 50
 _EXPIRE_REFRESH_TIME_PERCENT = 75
-_RECENT_TIME_PERCENT = 25
 
 _LOCAL_TRAILER = '.local.'
 _TCP_PROTOCOL_LOCAL_TRAILER = '._tcp.local.'