ipykernel 7 support

.github/workflows/test.yml

@@ -48,6 +48,8 @@ jobs:
           - python: "3.9"
             runs_on: macos-14
           - python: "3.11"
+          - python: "3.12"
+            pre: pre
 
     steps:
       - uses: actions/checkout@v4
@@ -122,7 +124,12 @@ jobs:
 
       - name: Install Python dependencies
         run: |
-          pip install --pre --upgrade ipyparallel[test]
+          pip install --upgrade ipyparallel[test]
+
+      - name: Install pre-release dependencies
+        if: ${{ matrix.pre }}
+        run: |
+          pip install --pre --upgrade ipyparallel[test] 'https://github.com/ipython/ipykernel/archive/main.tar.gz#egg=ipykernel'
 
       - name: Install extra Python packages
         if: ${{ ! startsWith(matrix.python, '3.11') }}
@@ -161,8 +168,8 @@ jobs:
         run: |
           set -x
           docker ps -a
-          docker logs slurmctld
           docker exec -i slurmctld squeue --states=all
           docker exec -i slurmctld sinfo
+          docker logs slurmctld
           docker logs c1
           docker logs c2

ipyparallel/apps/baseapp.py

@@ -7,7 +7,6 @@
 import re
 import sys
 
-import traitlets
 from IPython.core.application import BaseIPythonApplication
 from IPython.core.application import base_aliases as base_ip_aliases
 from IPython.core.application import base_flags as base_ip_flags
@@ -21,15 +20,6 @@
 
 from .._version import __version__
 
-# FIXME: CUnicode is needed for cli parsing
-# with traitlets 4
-# bump when we require traitlets 5, which requires Python 3.7
-if int(traitlets.__version__.split(".", 1)[0]) < 5:
-    from traitlets import CUnicode
-else:
-    # don't need CUnicode with traitlets 4
-    CUnicode = Unicode
-
 # -----------------------------------------------------------------------------
 # Module errors
 # -----------------------------------------------------------------------------
@@ -107,7 +97,7 @@ def _work_dir_changed(self, change):
         '', config=True, help="The ZMQ URL of the iplogger to aggregate logging."
     )
 
-    cluster_id = CUnicode(
+    cluster_id = Unicode(
         '',
         config=True,
         help="""String id to add to runtime files, to prevent name collisions when

ipyparallel/cluster/launcher.py

@@ -563,6 +563,10 @@ async def join(self, timeout=None):
         """Wait for the process to exit"""
         if self._wait_thread is not None:
             self._wait_thread.join(timeout=timeout)
+            if self._wait_thread.is_alive():
+                raise TimeoutError(
+                    f"Process {self.process.pid} did not exit in {timeout} seconds."
+                )
 
     def _stream_file(self, path):
         """Stream one file"""
@@ -1886,6 +1890,17 @@ def __init__(self, work_dir='.', config=None, **kwargs):
         # trigger program_changed to populate default context arguments
         self._program_changed()
 
+    def _run_command(self, command, **kwargs):
+        joined_command = shlex_join(command)
+        self.log.debug("Running command: %s", joined_command)
+        output = check_output(
+            command,
+            stdin=None,
+            **kwargs,
+        ).decode("utf8", "replace")
+        self.log.debug("Command %s output: %s", command[0], output)
+        return output
+
     def parse_job_id(self, output):
         """Take the output of the submit command and return the job id."""
         m = self.job_id_regexp.search(output)
@@ -1960,42 +1975,24 @@ def start(self, n=1):
 
         env = os.environ.copy()
         env.update(self.get_env())
-        output = check_output(self.args, env=env)
-        output = output.decode("utf8", 'replace')
-        self.log.debug(f"Submitted {shlex_join(self.args)}. Output: {output}")
+        output = self._run_command(self.args, env=env)
 
         job_id = self.parse_job_id(output)
         self.notify_start(job_id)
         return job_id
 
     def stop(self):
-        try:
-            output = check_output(
-                self.delete_command + [self.job_id],
-                stdin=None,
-            ).decode("utf8", 'replace')
-        except Exception:
-            self.log.exception(
-                "Problem stopping cluster with command: %s"
-                % (self.delete_command + [self.job_id])
-            )
-            output = ""
+        command = self.delete_command + [self.job_id]
+        output = self._run_command(command)
 
         self.notify_stop(
             dict(job_id=self.job_id, output=output)
         )  # Pass the output of the kill cmd
         return output
 
     def signal(self, sig):
-        cmd = self.signal_command + [str(sig), self.job_id]
-        try:
-            output = check_output(
-                cmd,
-                stdin=None,
-            ).decode("utf8", 'replace')
-        except Exception:
-            self.log.exception("Problem sending signal with: {shlex_join(cmd)}")
-            output = ""
+        command = self.signal_command + [str(sig), self.job_id]
+        self._run_command(command)
 
     # same local-file implementation as LocalProcess
     # should this be on the base class?