make copy_input_file async

Author: andrii-i

jupyter_scheduler/scheduler.py

@@ -41,8 +41,10 @@
 from jupyter_scheduler.orm import Job, JobDefinition, create_session
 from jupyter_scheduler.utils import (
     copy_directory,
+    copy_file,
     create_output_directory,
     create_output_filename,
+    spawn_process,
 )
 
 
@@ -422,11 +424,9 @@ def db_session(self):
         return self._db_session
 
     def copy_input_file(self, input_uri: str, copy_to_path: str):
-        """Copies the input file to the staging directory"""
+        """Copies the input file to the staging directory in a new process."""
         input_filepath = os.path.join(self.root_dir, input_uri)
-        with fsspec.open(input_filepath) as input_file:
-            with fsspec.open(copy_to_path, "wb") as output_file:
-                output_file.write(input_file.read())
+        spawn_process(target=copy_file, input_filepath=input_filepath, copy_to_path=copy_to_path)
 
     def copy_input_folder(self, input_uri: str, nb_copy_to_path: str) -> List[str]:
         """Copies the input file along with the input directory to the staging directory, returns the list of copied files relative to the staging directory"""
@@ -478,13 +478,6 @@ def create_job(self, model: CreateJob) -> str:
             else:
                 self.copy_input_file(model.input_uri, staging_paths["input"])
 
-            # The MP context forces new processes to not be forked on Linux.
-            # This is necessary because `asyncio.get_event_loop()` is bugged in
-            # forked processes in Python versions below 3.12. This method is
-            # called by `jupyter_core` by `nbconvert` in the default executor.
-            #
-            # See: https://github.com/python/cpython/issues/66285
-            # See also: https://github.com/jupyter/jupyter_core/pull/362
             mp_ctx = mp.get_context("spawn")
             p = mp_ctx.Process(
                 target=self.execution_manager_class(

jupyter_scheduler/utils.py

@@ -116,6 +116,13 @@ def copy_directory(
     return copied_files
 
 
+def copy_file(input_filepath: str, copy_to_path: str):
+    """Copies the file at input_filepath to copy_to_path"""
+    with fsspec.open(input_filepath) as input_file:
+        with fsspec.open(copy_to_path, "wb") as output_file:
+            output_file.write(input_file.read())
+
+
 def spawn_process(target, *args, **kwargs) -> SpawnProcess:
     """
     Spawns a new process using the 'spawn' context with the given target and