How to use executor in linux with "spawn" method in version 4.0.0

[ardevik]

With version 3 i can use:

from multiprocessing import get_context
from apscheduler.executor.pool import ProcessPoolExecutor

executor = ProcessPoolExecutor(max_instance=4, pool_kwargs={"mp_context": get_context("spawn")}

but how do this on version 4.0.0 on linux, i want to make sure multiprocessing always use spawn and not fork

[agronholm]

APScheduler 4 doesn't use multiprocessing at all. It uses AnyIO's to_process.run_sync() method which never uses fork() anyway.

[ardevik]

oke, because i will use apscheduler to run polars library that not work well when using fork

[ardevik]

Hi @agronholm,

can you show me example how to use apscheduler.executors.subprocess.ProcessPoolJobExecutors

im trying use this code but still get error

from apscheduler.executors.subprocess import ProcessPoolJobExecutors

job_executors = {"custom_pool": ProcessPoolJobExecutors(max_workers=1)}

def tick():
print("tick")

with Scheduler(data_store,event_broker, job_executors=job_executors ) as scheduler:
scheduler.add_schedule(tick, job_executor="custom_pool")
scheduler.run_until_stopped()

[agronholm]

I can only assume that your scheduled task function was in the same module that you used as the application entry point. This is not recommended. You can fix this by moving it to another module.

[ardevik]

Is scheduled task need in the same project as worker ?
I plan separate the project, worker and scheduler with fastapi, but get this error on worker side when i add task from separate project

INFO:apscheduler._schedulers.sync:Connection established
INFO:apscheduler._schedulers.sync:Scheduler started
ERROR:apscheduler._schedulers.sync:Scheduler crashed
  + Exception Group Traceback (most recent call last):
  |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 784, in run_until_stopped
  |     async with create_task_group() as task_group:
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 1012, in _process_jobs
    |     func = self._get_task_callable(task)
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 965, in _get_task_callable
    |     func = self._task_callables[task.id] = callable_from_ref(task.func)
    |                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 101, in callable_from_ref
    |     raise LookupError(f"Error resolving reference {ref!r}: could not import module")
    | LookupError: Error resolving reference 'tick:tick': could not import module
    |
    | During handling of the above exception, another exception occurred:
    |
    | Exception Group Traceback (most recent call last):
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 988, in _process_jobs
    |     async with AsyncExitStack() as exit_stack:
    |   File "/usr/lib64/python3.12/contextlib.py", line 754, in __aexit__
    |     raise exc_details[1]
    |   File "/usr/lib64/python3.12/contextlib.py", line 737, in __aexit__
    |     cb_suppress = await cb(*exc_details)
    |                   ^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
    |     raise BaseExceptionGroup(
    | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
    +-+---------------- 1 ----------------
      | Traceback (most recent call last):
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 961, in _get_task_callable
      |     return self._task_callables[task.id]
      |            ~~~~~~~~~~~~~~~~~~~~^^^^^^^^^
      | KeyError: 'tick:tick'
      |
      | During handling of the above exception, another exception occurred:
      |
      | Traceback (most recent call last):
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 99, in callable_from_ref
      |     obj = __import__(modulename, fromlist=[rest])
      |           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      | ModuleNotFoundError: No module named 'tick'
      |
      | During handling of the above exception, another exception occurred:
      |
      | Traceback (most recent call last):
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 1012, in _process_jobs
      |     func = self._get_task_callable(task)
      |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 965, in _get_task_callable
      |     func = self._task_callables[task.id] = callable_from_ref(task.func)
      |                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 101, in callable_from_ref
      |     raise LookupError(f"Error resolving reference {ref!r}: could not import module")
      | LookupError: Error resolving reference 'tick:tick': could not import module
      +------------------------------------
INFO:apscheduler._schedulers.sync:Stream finished
  + Exception Group Traceback (most recent call last):
  |   File "/root/main.py", line 18, in <module>
  |     scheduler.run_until_stopped()
  |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/sync.py", line 383, in run_until_stopped
  |     self._portal.call(self._async_scheduler.run_until_stopped)
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 287, in call
  |     return cast(T_Retval, self.start_task_soon(func, *args).result())
  |                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib64/python3.12/concurrent/futures/_base.py", line 456, in result
  |     return self.__get_result()
  |            ^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib64/python3.12/concurrent/futures/_base.py", line 401, in __get_result
  |     raise self._exception
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 218, in _call_func
  |     retval = await retval_or_awaitable
  |              ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 784, in run_until_stopped
  |     async with create_task_group() as task_group:
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 961, in _get_task_callable
    |     return self._task_callables[task.id]
    |            ~~~~~~~~~~~~~~~~~~~~^^^^^^^^^
    | KeyError: 'tick:tick'
    |
    | During handling of the above exception, another exception occurred:
    |
    | Traceback (most recent call last):
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 99, in callable_from_ref
    |     obj = __import__(modulename, fromlist=[rest])
    |           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    | ModuleNotFoundError: No module named 'tick'
    |
    | During handling of the above exception, another exception occurred:
    |
    | Traceback (most recent call last):
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 1012, in _process_jobs
    |     func = self._get_task_callable(task)
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 965, in _get_task_callable
    |     func = self._task_callables[task.id] = callable_from_ref(task.func)
    |                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 101, in callable_from_ref
    |     raise LookupError(f"Error resolving reference {ref!r}: could not import module")
    | LookupError: Error resolving reference 'tick:tick': could not import module
    |
    | During handling of the above exception, another exception occurred:
    |
    | Exception Group Traceback (most recent call last):
    |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 988, in _process_jobs
    |     async with AsyncExitStack() as exit_stack:
    |   File "/usr/lib64/python3.12/contextlib.py", line 754, in __aexit__
    |     raise exc_details[1]
    |   File "/usr/lib64/python3.12/contextlib.py", line 737, in __aexit__
    |     cb_suppress = await cb(*exc_details)
    |                   ^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
    |     raise BaseExceptionGroup(
    | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
    +-+---------------- 1 ----------------
      | Traceback (most recent call last):
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 1012, in _process_jobs
      |     func = self._get_task_callable(task)
      |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 965, in _get_task_callable
      |     func = self._task_callables[task.id] = callable_from_ref(task.func)
      |                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 101, in callable_from_ref
      |     raise LookupError(f"Error resolving reference {ref!r}: could not import module")
      | LookupError: Error resolving reference 'tick:tick': could not import module
      +------------------------------------

During handling of the above exception, another exception occurred:

  + Exception Group Traceback (most recent call last):
  |   File "/root/main.py", line 17, in <module>
  |     with Scheduler(data_store, event_broker, job_executors=job_executors) as scheduler:
  |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/sync.py", line 133, in __exit__
  |     self._exit_stack.__exit__(exc_type, exc_val, exc_tb)
  |   File "/usr/lib64/python3.12/contextlib.py", line 610, in __exit__
  |     raise exc_details[1]
  |   File "/usr/lib64/python3.12/contextlib.py", line 595, in __exit__
  |     if cb(*exc_details):
  |        ^^^^^^^^^^^^^^^^
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 131, in __exit__
  |     return self._exit_future.result()
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib64/python3.12/concurrent/futures/_base.py", line 456, in result
  |     return self.__get_result()
  |            ^^^^^^^^^^^^^^^^^^^
  |   File "/usr/lib64/python3.12/concurrent/futures/_base.py", line 401, in __get_result
  |     raise self._exception
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 218, in _call_func
  |     retval = await retval_or_awaitable
  |              ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 115, in run_async_cm
  |     result = await self._async_cm.__aexit__(*self._exit_exc_info)
  |              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 178, in __aexit__
  |     await self._exit_stack.__aexit__(exc_type, exc_val, exc_tb)
  |   File "/usr/lib64/python3.12/contextlib.py", line 754, in __aexit__
  |     raise exc_details[1]
  |   File "/usr/lib64/python3.12/contextlib.py", line 737, in __aexit__
  |     cb_suppress = await cb(*exc_details)
  |                   ^^^^^^^^^^^^^^^^^^^^^^
  |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Exception Group Traceback (most recent call last):
    |   File "/usr/lib64/python3.12/contextlib.py", line 737, in __aexit__
    |     cb_suppress = await cb(*exc_details)
    |                   ^^^^^^^^^^^^^^^^^^^^^^
    |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
    |     raise BaseExceptionGroup(
    | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
    +-+---------------- 1 ----------------
      | Exception Group Traceback (most recent call last):
      |   File "/root/main.py", line 18, in <module>
      |     scheduler.run_until_stopped()
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/sync.py", line 383, in run_until_stopped
      |     self._portal.call(self._async_scheduler.run_until_stopped)
      |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 287, in call
      |     return cast(T_Retval, self.start_task_soon(func, *args).result())
      |                           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/usr/lib64/python3.12/concurrent/futures/_base.py", line 456, in result
      |     return self.__get_result()
      |            ^^^^^^^^^^^^^^^^^^^
      |   File "/usr/lib64/python3.12/concurrent/futures/_base.py", line 401, in __get_result
      |     raise self._exception
      |   File "/root/.venv/lib64/python3.12/site-packages/anyio/from_thread.py", line 218, in _call_func
      |     retval = await retval_or_awaitable
      |              ^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 784, in run_until_stopped
      |     async with create_task_group() as task_group:
      |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
      |     raise BaseExceptionGroup(
      | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
      +-+---------------- 1 ----------------
        | Exception Group Traceback (most recent call last):
        |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 988, in _process_jobs
        |     async with AsyncExitStack() as exit_stack:
        |   File "/usr/lib64/python3.12/contextlib.py", line 754, in __aexit__
        |     raise exc_details[1]
        |   File "/usr/lib64/python3.12/contextlib.py", line 737, in __aexit__
        |     cb_suppress = await cb(*exc_details)
        |                   ^^^^^^^^^^^^^^^^^^^^^^
        |   File "/root/.venv/lib64/python3.12/site-packages/anyio/_backends/_asyncio.py", line 680, in __aexit__
        |     raise BaseExceptionGroup(
        | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
        +-+---------------- 1 ----------------
          | Traceback (most recent call last):
          |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 1012, in _process_jobs
          |     func = self._get_task_callable(task)
          |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
          |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_schedulers/async_.py", line 965, in _get_task_callable
          |     func = self._task_callables[task.id] = callable_from_ref(task.func)
          |                                            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
          |   File "/root/.venv/lib64/python3.12/site-packages/apscheduler/_marshalling.py", line 101, in callable_from_ref
          |     raise LookupError(f"Error resolving reference {ref!r}: could not import module")
          | LookupError: Error resolving reference 'tick:tick': could not import module
          +------------------------------------

[agronholm]

How could the worker import the task function if it's not available to it?

[ardevik]

so if task function need to be in the same project as worker, how i can register the task function to the worker without run it, then how i can trigger the task function to run from other project(scheduler or web project), please help still not find any code example or can you give me the link to code example ?

[agronholm]

The worker just needs to be able to import the module (tick) and retrieve the tick() function from it. You don't need to do anything extra in the worker.