Error using multiprocessing in Ops (is it allowed?)

[whillas-yabble]

hi,
I'm doing a grid search over some clustering params which is intensive and an 'embarrassingly parallelizable problem' so i just copy pasted from the concurrent.futures examples with my function and it sometimes works. Lately its been half way through the search (running in VSCode debugger, running in a test using job.execute_in_process()) and I'm getting this error starting to crop up:

Traceback (most recent call last):
  File "/usr/lib/python3.12/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/usr/lib/python3.12/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/usr/lib/python3.12/multiprocessing/pool.py", line 114, in worker
    task = get()
           ^^^^^
  File "/usr/lib/python3.12/multiprocessing/queues.py", line 386, in get
    with self._rlock:
  File "/usr/lib/python3.12/multiprocessing/synchronize.py", line 95, in __enter__
    return self._semlock.__enter__()
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/alex/dev/yabble/ai-dags-theme-counting/__venv__/lib/python3.12/site-packages/dagster/_utils/interrupts.py", line 83, in _new_signal_handler
    raise error_cls()
dagster._core.errors.DagsterExecutionInterruptedError" 

Is using multiprocessing inside a dagster Op OK? Is it just because I'm using using job.execute_in_process()? Something else?

Using dagster==1.10.18 & python 3.12.3

[whillas-yabble]

Looks like it does have something to do with the job.execute_in_process as running the dag via dagster dev seems to work... not sure where this leaves me in terms of debugging :(

[whillas-yabble]

Actually, it looks like it fails with

A process in the process pool was terminated abruptly while the future was running or pending.

:(

[cmpadden]

Hi @whillas-yabble, when using job.execute_in_process() the executor ignores the job's configured executor and uses the in_process_executor:

• https://docs.dagster.io/guides/build/jobs/job-execution#default-job-executor

You might want to consider using the multiprocess_executor which will execute each step in its own spawned process.

import dagster as dg

@dg.job(
    config={
        "execution": {
            "config": {
                "multiprocess": {
                    "max_concurrent": 4,
                }
            }
        }
    }
)
def my_job():
    # your ops here
    pass

There's also this open GitHub issues about running multiprocessing within an asset that might be relevant to you. It talks about a workaround of setting the PYTHONPATH.

• Support Python multiprocessing within assets #19781 (comment)