Threaded iterable fix (#227)

* Unlock thread after closing when queue is full
* No test defaults bcause it's env related
* reuseport do not work for unix sockets since 6.12

Author: mosquito

aiomisc/iterator_wrapper.py

@@ -213,6 +213,12 @@ def _in_thread(self) -> None:
 
     def close(self) -> Awaitable[None]:
         self.__channel.close()
+        # if the iterator inside thread is blocked on `.put()`
+        # we need to wake it up to signal that it is closed.
+        try:
+            self.__channel.queue.get()
+        except QueueEmpty:
+            pass
         return asyncio.ensure_future(self.wait_closed())
 
     async def wait_closed(self) -> None:

aiomisc/utils.py

@@ -88,6 +88,10 @@ def bind_socket(
     if not args and ":" in address:
         sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
 
+    unix_address_family = getattr(socket, "AF_UNIX", None)
+    if sock.family == unix_address_family:
+        reuse_port = False
+
     if reuse_addr:
         sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
@@ -99,7 +103,6 @@ def bind_socket(
     for level, option, value in options:
         sock.setsockopt(level, option, value)
 
-    unix_address_family = getattr(socket, "AF_UNIX", None)
     if sock.family == unix_address_family:
         proto_name = proto_name or "unix"
         sock.bind(address)

aiomisc/worker_pool.py

@@ -60,6 +60,7 @@ def _create_socket(self) -> None:
                 socket.AF_UNIX,
                 socket.SOCK_STREAM,
                 address=path,
+                reuse_port=False,
             )
             self.address = path
             chmod(path, 0o600)

tests/test_entrypoint.py

@@ -49,8 +49,6 @@ def unix_socket_udp():
 
     # Behaviour like in the bind_socket
     sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
-    if hasattr(socket, "SO_REUSEPORT"):
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
 
     try:
         sock.bind(socket_path)
@@ -959,10 +957,14 @@ async def test_add_remove_service(entrypoint: aiomisc.Entrypoint):
 @pytest.mark.parametrize(
     "entrypoint_logging_kwargs,basic_config_kwargs", [
         (
-            {},
+            {
+                "log_level": LogLevel.info.name,
+                "log_format": LogFormat.plain,
+                "log_date_format": None,
+            },
             {
                 "level": LogLevel.info.name,
-                "log_format": LogFormat.default(),
+                "log_format": LogFormat.plain,
                 "date_format": None,
             },
         ),

tests/test_sdwatchdog_service.py

@@ -38,6 +38,7 @@ async def handle_datagram(
 
         with bind_socket(
             socket.AF_UNIX, socket.SOCK_DGRAM, address=sock_path,
+            reuse_port=False,
         ) as sock:
             try:
                 os.environ["NOTIFY_SOCKET"] = sock_path