diff --git a/dask_jobqueue/core.py b/dask_jobqueue/core.py index baf54146..b973eb1d 100644 --- a/dask_jobqueue/core.py +++ b/dask_jobqueue/core.py @@ -1,4 +1,5 @@ from contextlib import contextmanager, suppress +import asyncio import logging import math import os @@ -11,6 +12,7 @@ import tempfile import copy import warnings +import functools import dask @@ -94,7 +96,22 @@ """.strip() -class Job(ProcessInterface, abc.ABC): +class FixedProcessInterface(ProcessInterface): + def __init__(self, scheduler=None, name=None): + self.address = getattr(self, "address", None) + self.external_address = None + self.status = Status.created + + @functools.cached_property + def lock(self): + return asyncio.Lock() + + @functools.cached_property + def _event_finished(self): + return asyncio.Event() + + +class Job(FixedProcessInterface, abc.ABC): """ Base class to launch Dask workers on Job queues This class should not be used directly, use a class appropriate for