Deprecate utilities which have moved to dask (#4966)

distributed/batched.py

@@ -5,9 +5,9 @@
 from tornado.ioloop import IOLoop
 
 import dask
+from dask.utils import parse_timedelta
 
 from .core import CommClosedError
-from .utils import parse_timedelta
 
 logger = logging.getLogger(__name__)
 

distributed/cfexecutor.py

@@ -4,8 +4,10 @@
 from tlz import merge
 from tornado import gen
 
+from dask.utils import parse_timedelta
+
 from .metrics import time
-from .utils import TimeoutError, parse_timedelta, sync
+from .utils import TimeoutError, sync
 
 
 @gen.coroutine

distributed/client.py

@@ -36,6 +36,7 @@
     ensure_dict,
     format_bytes,
     funcname,
+    parse_timedelta,
     stringify,
 )
 
@@ -78,7 +79,6 @@
     key_split,
     log_errors,
     no_default,
-    parse_timedelta,
     sync,
     thread_state,
 )

distributed/comm/core.py

@@ -8,11 +8,12 @@
 from contextlib import suppress
 
 import dask
+from dask.utils import parse_timedelta
 
 from ..metrics import time
 from ..protocol import pickle
 from ..protocol.compression import get_default_compression
-from ..utils import TimeoutError, parse_timedelta
+from ..utils import TimeoutError
 from . import registry
 from .addressing import parse_address
 

distributed/comm/tcp.py

@@ -20,11 +20,12 @@
 from tornado.tcpserver import TCPServer
 
 import dask
+from dask.utils import parse_timedelta
 
 from ..protocol.utils import pack_frames_prelude, unpack_frames
 from ..system import MEMORY_LIMIT
 from ..threadpoolexecutor import ThreadPoolExecutor
-from ..utils import ensure_ip, get_ip, get_ipv6, nbytes, parse_timedelta
+from ..utils import ensure_ip, get_ip, get_ipv6, nbytes
 from .addressing import parse_host_port, unparse_host_port
 from .core import Comm, CommClosedError, Connector, FatalCommClosedError, Listener
 from .registry import Backend, backends

distributed/comm/ucx.py

@@ -10,8 +10,9 @@
 import weakref
 
 import dask
+from dask.utils import parse_bytes
 
-from ..utils import ensure_ip, get_ip, get_ipv6, log_errors, nbytes, parse_bytes
+from ..utils import ensure_ip, get_ip, get_ipv6, log_errors, nbytes
 from .addressing import parse_host_port, unparse_host_port
 from .core import Comm, CommClosedError, Connector, Listener
 from .registry import Backend, backends

distributed/core.py

@@ -18,6 +18,7 @@
 from tornado.ioloop import IOLoop, PeriodicCallback
 
 import dask
+from dask.utils import parse_timedelta
 
 from . import profile, protocol
 from .comm import (
@@ -36,7 +37,6 @@
     get_traceback,
     has_keyword,
     is_coroutine_function,
-    parse_timedelta,
     truncate_exception,
 )
 

distributed/deploy/adaptive.py

@@ -2,9 +2,10 @@
 from inspect import isawaitable
 
 import dask.config
+from dask.utils import parse_timedelta
 
 from ..protocol import pickle
-from ..utils import log_errors, parse_timedelta
+from ..utils import log_errors
 from .adaptive_core import AdaptiveCore
 
 logger = logging.getLogger(__name__)

distributed/deploy/adaptive_core.py

@@ -6,8 +6,9 @@
 import tlz as toolz
 from tornado.ioloop import IOLoop, PeriodicCallback
 
+from dask.utils import parse_timedelta
+
 from ..metrics import time
-from ..utils import parse_timedelta
 
 logger = logging.getLogger(__name__)
 

distributed/deploy/cluster.py

@@ -9,19 +9,11 @@
 from tornado.ioloop import PeriodicCallback
 
 import dask.config
-from dask.utils import _deprecated, format_bytes
+from dask.utils import _deprecated, format_bytes, parse_timedelta
 
 from ..core import Status
 from ..objects import SchedulerInfo
-from ..utils import (
-    Log,
-    Logs,
-    format_dashboard_link,
-    log_errors,
-    parse_timedelta,
-    sync,
-    thread_state,
-)
+from ..utils import Log, Logs, format_dashboard_link, log_errors, sync, thread_state
 from .adaptive import Adaptive
 
 logger = logging.getLogger(__name__)

distributed/deploy/spec.py

@@ -11,18 +11,12 @@
 from tornado import gen
 
 import dask
+from dask.utils import parse_bytes, parse_timedelta
 
 from ..core import CommClosedError, Status, rpc
 from ..scheduler import Scheduler
 from ..security import Security
-from ..utils import (
-    LoopRunner,
-    TimeoutError,
-    import_term,
-    parse_bytes,
-    parse_timedelta,
-    silence_logging,
-)
+from ..utils import LoopRunner, TimeoutError, import_term, silence_logging
 from .adaptive import Adaptive
 from .cluster import Cluster
 

distributed/diagnostics/progressbar.py

@@ -8,10 +8,12 @@
 from tlz import valmap
 from tornado.ioloop import IOLoop
 
+import dask
+
 from ..client import default_client, futures_of
 from ..core import CommClosedError, coerce_to_address, connect
 from ..protocol.pickle import dumps
-from ..utils import LoopRunner, is_kernel, key_split, parse_timedelta
+from ..utils import LoopRunner, is_kernel, key_split
 from .progress import MultiProgress, Progress, format_time
 
 logger = logging.getLogger(__name__)
@@ -34,7 +36,7 @@ def __init__(self, keys, scheduler=None, interval="100ms", complete=True):
                 break
 
         self.keys = {k.key if hasattr(k, "key") else k for k in keys}
-        self.interval = parse_timedelta(interval, default="s")
+        self.interval = dask.utils.parse_timedelta(interval, default="s")
         self.complete = complete
         self._start_time = default_timer()
 

distributed/event.py

@@ -4,8 +4,10 @@
 from collections import defaultdict
 from contextlib import suppress
 
+from dask.utils import parse_timedelta
+
 from .client import Client
-from .utils import TimeoutError, log_errors, parse_timedelta
+from .utils import TimeoutError, log_errors
 from .worker import get_worker
 
 logger = logging.getLogger(__name__)

distributed/lock.py

@@ -3,8 +3,10 @@
 import uuid
 from collections import defaultdict, deque
 
+from dask.utils import parse_timedelta
+
 from .client import Client
-from .utils import TimeoutError, log_errors, parse_timedelta
+from .utils import TimeoutError, log_errors
 from .worker import get_worker
 
 logger = logging.getLogger(__name__)

distributed/multi_lock.py

@@ -4,8 +4,10 @@
 from collections import defaultdict
 from typing import Hashable, List
 
+from dask.utils import parse_timedelta
+
 from .client import Client
-from .utils import TimeoutError, log_errors, parse_timedelta
+from .utils import TimeoutError, log_errors
 from .worker import get_worker
 
 logger = logging.getLogger(__name__)

distributed/nanny.py

@@ -17,6 +17,7 @@
 
 import dask
 from dask.system import CPU_COUNT
+from dask.utils import parse_timedelta
 
 from . import preloading
 from .comm import get_address_host, unparse_host_port
@@ -32,7 +33,6 @@
     json_load_robust,
     mp_context,
     parse_ports,
-    parse_timedelta,
     silence_logging,
 )
 from .worker import Worker, parse_memory_limit, run

distributed/profile.py

@@ -33,8 +33,10 @@
 
 import tlz as toolz
 
+from dask.utils import format_time, parse_timedelta
+
 from .metrics import time
-from .utils import color_of, format_time, parse_timedelta
+from .utils import color_of
 
 
 def identifier(frame):

distributed/pubsub.py

@@ -4,10 +4,12 @@
 import weakref
 from collections import defaultdict, deque
 
+from dask.utils import parse_timedelta
+
 from .core import CommClosedError
 from .metrics import time
 from .protocol.serialize import to_serialize
-from .utils import TimeoutError, parse_timedelta, sync
+from .utils import TimeoutError, sync
 
 logger = logging.getLogger(__name__)
 

distributed/queues.py

@@ -3,10 +3,10 @@
 import uuid
 from collections import defaultdict
 
-from dask.utils import stringify
+from dask.utils import parse_timedelta, stringify
 
 from .client import Client, Future
-from .utils import parse_timedelta, sync, thread_state
+from .utils import sync, thread_state
 from .worker import get_client, get_worker
 
 logger = logging.getLogger(__name__)

distributed/scheduler.py

@@ -38,6 +38,7 @@
 
 import dask
 from dask.highlevelgraph import HighLevelGraph
+from dask.utils import format_bytes, format_time, parse_bytes, parse_timedelta
 
 from . import preloading, profile
 from . import versions as version_module
@@ -70,15 +71,11 @@
     All,
     TimeoutError,
     empty_context,
-    format_bytes,
-    format_time,
     get_fileno_limit,
     key_split,
     key_split_group,
     log_errors,
     no_default,
-    parse_bytes,
-    parse_timedelta,
     tmpfile,
     validate_key,
 )

distributed/semaphore.py

@@ -8,11 +8,12 @@
 from tornado.ioloop import IOLoop, PeriodicCallback
 
 import dask
+from dask.utils import parse_timedelta
 
 from distributed.utils_comm import retry_operation
 
 from .metrics import time
-from .utils import log_errors, parse_timedelta, sync, thread_state
+from .utils import log_errors, sync, thread_state
 from .worker import get_client, get_worker
 
 logger = logging.getLogger(__name__)

distributed/stealing.py

@@ -7,11 +7,12 @@
 from tornado.ioloop import PeriodicCallback
 
 import dask
+from dask.utils import parse_timedelta
 
 from .comm.addressing import get_address_host
 from .core import CommClosedError
 from .diagnostics.plugin import SchedulerPlugin
-from .utils import log_errors, parse_timedelta
+from .utils import log_errors
 
 LATENCY = 10e-3