Merge remote-tracking branch 'origin/main' into ssh_windows_support

# Conflicts:
#	.github/workflows/test.yml

Author: minrk

.github/workflows/test.yml

@@ -47,6 +47,8 @@ jobs:
           - python: "3.9"
             runs_on: macos-14
           - python: "3.11"
+          - python: "3.12"
+            pre: pre
 
     steps:
       - uses: actions/checkout@v4
@@ -87,7 +89,7 @@ jobs:
           export DOCKER_BUILDKIT=1
           export COMPOSE_DOCKER_CLI_BUILD=1
           cd ci/slurm
-          docker-compose up -d --build
+          docker compose up -d --build
 
       - name: Install Python (conda) ${{ matrix.python }}
         if: ${{ matrix.cluster_type == 'mpi' }}
@@ -113,16 +115,17 @@ 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') }}
         run: |
           pip install distributed joblib
-          # pip install --only-binary :all: matplotlib || echo "no matplotlib"#
-          # the || syntax doesn't work under windows powershell, but since the command (currently) works for all
-          # python version (>3.8), the "||" branch can be simply removed. Otherwise we would need a splitted
-          # step for Windows and Linux (as for the docker-compose command)
           pip install --only-binary :all: matplotlib
 
       - name: Show environment
@@ -156,8 +159,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

docs/source/tutorial/process.md

@@ -789,6 +789,8 @@ in some scratch directory. This can be set with:
 c.IPEngine.work_dir = u'/path/to/scratch/'
 ```
 
+[^cite_mpi]: Message Passing Interface (MPI) <https://www.mpi-forum.org>
+
 [^cite_mongodb]: MongoDB database <https://www.mongodb.com/>
 
 [^cite_pbs]: Portable Batch System <https://www.mcs.anl.gov/research/projects/openpbs/>

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

@@ -564,6 +564,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"""
@@ -1873,6 +1877,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)
@@ -1947,42 +1962,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?