Try to fix unclosed resources.

I think these might be one of the few places that create those socket,
so maybe closing them is the right call. I'm not sure.

There are random failures on some tests.

as _handle_pipe_msg is use in a single place and `self._async_pipe_in1`
is used only to be passed between _handle_pipe_msgs and _handle_pipe_msg
I'm tempted to refactor and pass _async_pipe_in1 as a parameter maybe ?
and also rename to _handle_single_pipe_msg to make it visually a bit
more distinctive.

Author: Carreau

ipykernel/iostream.py

@@ -170,20 +170,20 @@ async def _handle_event(self):
         *all* waiting events are processed in order.
         """
         # create async wrapper within coroutine
-        pipe_in = zmq.asyncio.Socket(self._pipe_in0)
-        try:
-            while True:
-                await pipe_in.recv()
-                # freeze event count so new writes don't extend the queue
-                # while we are processing
-                n_events = len(self._events)
-                for _ in range(n_events):
-                    event_f = self._events.popleft()
-                    event_f()
-        except Exception:
-            if self.thread.__stop.is_set():
-                return
-            raise
+        with zmq.asyncio.Socket(self._pipe_in0) as pipe_in:
+            try:
+                while True:
+                    await pipe_in.recv()
+                    # freeze event count so new writes don't extend the queue
+                    # while we are processing
+                    n_events = len(self._events)
+                    for _ in range(n_events):
+                        event_f = self._events.popleft()
+                        event_f()
+            except Exception:
+                if self.thread.__stop.is_set():
+                    return
+                raise
 
     def _setup_pipe_in(self):
         """setup listening pipe for IOPub from forked subprocesses"""
@@ -218,6 +218,8 @@ async def _handle_pipe_msgs(self):
             if self.thread.__stop.is_set():
                 return
             raise
+        finally:
+            self._async_pipe_in1.close()
 
     async def _handle_pipe_msg(self, msg=None):
         """handle a pipe message from a subprocess"""