Repository: dask/dask
Pull request #10901 (2024-02-06 16:07:41)

Update deployment options in docs homepage
Applies some of the same changes in https://github.com/dask/dask/pull/10882 to the docs homepage 

cc @mrocklin

github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/dask/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

     15 files  ±0       15 suites  ±0   3h 17m 36s :stopwatch: + 1m 44s
 12 986 tests ±0   12 057 :white_check_mark: +1     929 :zzz: ±0  0 :x:  - 1 
160 477 runs  ±0  143 949 :white_check_mark: +2  16 528 :zzz:  - 1  0 :x:  - 1 

Results for commit 07b9c6e0. ± Comparison against base commit 9d81762b.

[test-results]:data:application/gzip;base64,H4sIABthwmUC/03MyQ7CIBSF4VdpWLtgugy+jKEMCbGDobBqfHcRK+3y/05ydhTi5Dd0HwjcBrSVmM9yJZkc1+XbRIGoVNfcdqpV78dWrG2IQZ74jK+KmupOwcSpEu7gU1rTIaks7VlgLuXR/ZkzzXXH3zMRQNXfLtetr892neeYayAsR22Fx94AV5iCqTUKZoILhFvHPNPgVJDo/QEosp0qGQEAAA==

mrocklin: Fine by me

Patch:
diff --git a/dask/array/ufunc.py b/dask/array/ufunc.py
index 9a0f527a7..d9ebf06b5 100644
--- a/dask/array/ufunc.py
+++ b/dask/array/ufunc.py
@@ -7,7 +7,7 @@ import numpy as np
 
 from dask import core
 from dask.array.core import Array, apply_infer_dtype, asarray, blockwise, elemwise
-from dask.base import is_dask_collection, normalize_function
+from dask.base import is_dask_collection, normalize_token
 from dask.highlevelgraph import HighLevelGraph
 from dask.utils import derived_from, funcname
 
@@ -42,7 +42,7 @@ class da_frompyfunc:
         return "da.frompyfunc<%s, %d, %d>" % (self._name, self.nin, self.nout)
 
     def __dask_tokenize__(self):
-        return (normalize_function(self._func), self.nin, self.nout)
+        return (normalize_token(self._func), self.nin, self.nout)
 
     def __reduce__(self):
         return (da_frompyfunc, (self._func, self.nin, self.nout))
diff --git a/dask/base.py b/dask/base.py
index 1cdbdb28f..ae4745e29 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -5,7 +5,6 @@ import datetime
 import decimal
 import hashlib
 import inspect
-import os
 import pathlib
 import pickle
 import random
@@ -21,7 +20,8 @@ from enum import Enum
 from functools import partial
 from numbers import Integral, Number
 from operator import getitem
-from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeVar
+from types import MappingProxyType
+from typing import Any, Literal, TypeVar
 
 from tlz import curry, groupby, identity, merge
 from tlz.functoolz import Compose
@@ -80,9 +80,6 @@ __all__ = (
     "clone_key",
 )
 
-if TYPE_CHECKING:
-    from _typeshed import ReadableBuffer
-
 _annotations: ContextVar[dict[str, Any]] = ContextVar("annotations", default={})
 
 
@@ -1012,31 +1009,25 @@ def persist(*args, traverse=True, optimize_graph=True, scheduler=None, **kwargs)
 ############
 
 
-class _HashFactory(Protocol):
-    def __call__(
-        self, string: ReadableBuffer = b"", *, usedforsecurity: bool = True
-    ) -> hashlib._Hash:
-        ...
-
-
-# Pass `usedforsecurity=False` to support FIPS builds of Python
-def _md5(x: ReadableBuffer, _hashlib_md5: _HashFactory = hashlib.md5) -> hashlib._Hash:
-    return _hashlib_md5(x, usedforsecurity=False)
-
-
 def tokenize(*args, **kwargs):
     """Deterministic token
 
     >>> tokenize([1, 2, '3'])
-    '7d6a880cd9ec03506eee6973ff551339'
+    '06961e8de572e73c2e74b51348177918'
 
     >>> tokenize('Hello') == tokenize('Hello')
     True
     """
-    hasher = _md5(str(tuple(map(normalize_token, args))).encode())
-    if kwargs:
-        hasher.update(str(normalize_token(kwargs)).encode())
-    return hasher.hexdigest()
+    tok = _seen.set({})
+    try:
+        token = _normalize_seq_func(args)
+        if kwargs:
+            token = token, _normalize_seq_func(sorted(kwargs.items()))
+    finally:
+        _seen.reset(tok)
+
+    # Pass `usedforsecurity=False` to support FIPS builds of Python
+    return hashlib.md5(str(token).encode(), usedforsecurity=False).hexdigest()
 
 
 normalize_token = Dispatch()
@@ -1047,7 +1038,6 @@ normalize_token.register(
         str,
         bytes,
         type(None),
-        type,
         slice,
         complex,
         type(Ellipsis),
@@ -1062,35 +1052,60 @@ normalize_token.register(
 )
 
 
+@normalize_token.register(type)
+def normalize_type(typ):
+    # This allows for some ambiguity in case a type is redefined in a local
+    # scope of a module however, defining this further breaks tokenization for
+    # local objects entirely
+    return typ.__name__, typ.__module__
+
+
+@normalize_token.register(MappingProxyType)
 @normalize_token.register(dict)
 def normalize_dict(d):
-    return normalize_token(sorted(d.items(), key=str))
+    return "dict", _normalize_seq_func(sorted(d.items(), key=str))
 
 
 @normalize_token.register(OrderedDict)
 def normalize_ordered_dict(d):
-    return type(d).__name__, normalize_token(list(d.items()))
+    return _normalize_seq_func((type(d), list(d.items())))
 
 
 @normalize_token.register(set)
 def normalize_set(s):
-    return normalize_token(sorted(s, key=str))
+    return "set", _normalize_seq_func(sorted(s, key=str))
 
 
-def _normalize_seq_func(seq):
-    # Defined outside normalize_seq to avoid unnecessary redefinitions and
-    # therefore improving computation times.
-    try:
-        return list(map(normalize_token, seq))
-    except RecursionError:
-        if not config.get("tokenize.ensure-deterministic"):
-            return uuid.uuid4().hex
+# Circular reference breaker used by _normalize_seq_func.
+# This variable is recreated anew every time you call tokenize(). Note that this means
+# that you could call tokenize() from inside tokenize() and they would be fully
+# independent.
+#
+# It is a map of {id(obj): (<first seen incremental int>, obj)} which causes an object
+# to be tokenized as ("__seen", <incremental>) the second time it's encountered while
+# traversing collections. A strong reference to the object is stored in the context to
+# prevent ids from being reused by different objects.
+_seen: ContextVar[dict[int, tuple[int, object]]] = ContextVar("_seen")
 
-        raise RuntimeError(
-            f"Sequence {str(seq)} cannot be deterministically hashed. Please, see "
-            "https://docs.dask.org/en/latest/custom-collections.html#implementing-deterministic-hashing "
-            "for more information"
-        )
+
+def _normalize_seq_func(seq):
+    seen = _seen.get()
+    out = []
+
+    for item in seq:
+        if isinstance(item, (str, bytes, int, float, bool, type(None))):
+            # Basic data type. This is just for performance and compactness of the
+            # output. It doesn't need to be a comprehensive list.
+            norm_item = item
+        elif id(item) in seen:
+            # May or may not be a circular recursion. Maybe just a double reference.
+            seen_when, _ = seen[id(item)]
+            norm_item = "__seen", seen_when
+        else:
+            seen[id(item)] = len(seen), item
+            norm_item = normalize_token(item)
+        out.append(norm_item)
+    return out
 
 
 @normalize_token.register((tuple, list))
@@ -1105,12 +1120,12 @@ def normalize_literal(lit):
 
 @normalize_token.register(range)
 def normalize_range(r):
-    return list(map(normalize_token, [r.start, r.stop, r.step]))
+    return "range", _normalize_seq_func((r.start, r.stop, r.step))
 
 
 @normalize_token.register(Enum)
 def normalize_enum(e):
-    return type(e).__name__, e.name, e.value
+    return _normalize_seq_func((type(e), e.name, e.value))
 
 
 @normalize_token.register(random.Random)
@@ -1133,7 +1148,7 @@ def normalize_object(o):
         if self is not None and not inspect.ismodule(self) and isinstance(name, str):
             return normalize_token(self), name
 
-        return normalize_function(o)
+        return normalize_callable(o)
 
     if dataclasses.is_dataclass(o):
         return normalize_dataclass(o)
@@ -1142,7 +1157,7 @@ def normalize_object(o):
         return uuid.uuid4().hex
 
     raise RuntimeError(
-        f"Object {str(o)} cannot be deterministically hashed. Please, see "
+        f"Object {o!r} cannot be deterministically hashed. Please, see "
         "https://docs.dask.org/en/latest/custom-collections.html#implementing-deterministic-hashing "
         "for more information"
     )
@@ -1152,11 +1167,11 @@ function_cache: dict[Callable, Callable | tuple | str | bytes] = {}
 function_cache_lock = threading.Lock()
 
 
-def normalize_function(func: Callable) -> Callable | tuple | str | bytes:
+def normalize_callable(func: Callable) -> Callable | tuple | str | bytes:
     try:
         return function_cache[func]
     except KeyError:
-        result = _normalize_function(func)
+        result = _normalize_callable(func)
         if len(function_cache) >= 500:  # clear half of cache if full
             with function_cache_lock:
                 if len(function_cache) >= 500:
@@ -1165,14 +1180,17 @@ def normalize_function(func: Callable) -> Callable | tuple | str | bytes:
         function_cache[func] = result
         return result
     except TypeError:  # not hashable
-        return _normalize_function(func)
+        return _normalize_callable(func)
 
 
-def _normalize_function(func: Callable) -> tuple | str | bytes:
+normalize_function = normalize_callable
+
+
+def _normalize_callable(func: Callable) -> tuple | str | bytes:
     if isinstance(func, Compose):
         first = getattr(func, "first", None)
         funcs = reversed((first,) + func.funcs) if first else func.funcs
-        return tuple(normalize_function(f) for f in funcs)
+        return tuple(normalize_callable(f) for f in funcs)
     elif isinstance(func, (partial, curry)):
         args = tuple(normalize_token(i) for i in func.args)
         if func.keywords:
@@ -1181,7 +1199,7 @@ def _normalize_function(func: Callable) -> tuple | str | bytes:
             )
         else:
             kws = None
-        return (normalize_function(func.func), args, kws)
+        return normalize_callable(func.func), args, kws
     else:
         try:
             result = pickle.dumps(func, protocol=4)
@@ -1198,7 +1216,7 @@ def _normalize_function(func: Callable) -> tuple | str | bytes:
                 return str(func)
         else:
             raise RuntimeError(
-                f"Function {str(func)} may not be deterministically hashed by "
+                f"Function {func!r} may not be deterministically hashed by "
                 "cloudpickle. See: https://github.com/cloudpipe/cloudpickle/issues/385 "
                 "for more information."
             )
@@ -1206,11 +1224,16 @@ def _normalize_function(func: Callable) -> tuple | str | bytes:
 
 def normalize_dataclass(obj):
     fields = [
-        (field.name, getattr(obj, field.name)) for field in dataclasses.fields(obj)
+        (field.name, normalize_token(getattr(obj, field.name, None)))
+        for field in dataclasses.fields(obj)
     ]
+    params = obj.__dataclass_params__
+    params = [(attr, getattr(params, attr)) for attr in params.__slots__]
+
     return (
-        normalize_function(type(obj)),
-        _normalize_seq_func(fields),
+        normalize_callable(type(obj)),
+        params,
+        fields,
     )
 
 
@@ -1302,29 +1325,16 @@ def register_pyarrow():
 def register_numpy():
     import numpy as np
 
+    @normalize_token.register(np.generic)
     @normalize_token.register(np.ndarray)
     def normalize_array(x):
         if not x.shape:
+            # Note: this path is used for both scalar and empty arrays
             return (x.item(), x.dtype)
-        if hasattr(x, "mode") and getattr(x, "filename", None):
-            if hasattr(x.base, "ctypes"):
-                offset = (
-                    x.ctypes._as_parameter_.value - x.base.ctypes._as_parameter_.value
-                )
-            else:
-                offset = 0  # root memmap's have mmap object as base
-            if hasattr(
-                x, "offset"
-            ):  # offset numpy used while opening, and not the offset to the beginning of file
-                offset += x.offset
-            return (
-                x.filename,
-                os.path.getmtime(x.filename),
-                x.dtype,
-                x.shape,
-                x.strides,
-                offset,
-            )
+
+        if x.size == 0:
+            return (None, x.dtype, x.shape)
+
         if x.dtype.hasobject:
             try:
                 try:
@@ -1346,7 +1356,7 @@ def register_numpy():
                         data = uuid.uuid4().hex
                     else:
                         raise RuntimeError(
-                            f"``np.ndarray`` with object ``dtype`` {str(x)} cannot "
+                            f"``np.ndarray`` with object ``dtype`` {x!r} cannot "
                             "be deterministically hashed. Please, see "
                             "https://docs.dask.org/en/latest/custom-collections.html#implementing-deterministic-hashing "  # noqa: E501
                             "for more information"
@@ -1356,14 +1366,13 @@ def register_numpy():
                 data = hash_buffer_hex(x.ravel(order="K").view("i1"))
             except (BufferError, AttributeError, ValueError):
                 data = hash_buffer_hex(x.copy().ravel(order="K").view("i1"))
-        return (data, x.dtype, x.shape, x.strides)
+        return (data, x.dtype, x.shape)
 
     @normalize_token.register(np.matrix)
     def normalize_matrix(x):
-        return type(x).__name__, normalize_array(x.view(type=np.ndarray))
+        return _normalize_seq_func((type(x), x.view(type=np.ndarray)))
 
     normalize_token.register(np.dtype, repr)
-    normalize_token.register(np.generic, repr)
 
     @normalize_token.register(np.ufunc)
     def normalize_ufunc(x):
@@ -1372,7 +1381,7 @@ def register_numpy():
             if getattr(np, name) is x:
                 return "np." + name
         except AttributeError:
-            return normalize_function(x)
+            return normalize_callable(x)
 
     @normalize_token.register(np.random.RandomState)
     def normalize_np_random_state(state):
@@ -1388,10 +1397,7 @@ def register_scipy():
     import scipy.sparse as sp
 
     def normalize_sparse_matrix(x, attrs):
-        return (
-            type(x).__name__,
-            normalize_seq(normalize_token(getattr(x, key)) for key in attrs),
-        )
+        return _normalize_seq_func((type(x), [getattr(x, key) for key in attrs]))
 
     for cls, attrs in [
         (sp.dia_matrix, ("data", "offsets", "shape")),
@@ -1405,7 +1411,7 @@ def register_scipy():
 
     @normalize_token.register(sp.dok_matrix)
     def normalize_dok_matrix(x):
-        return type(x).__name__, normalize_token(sorted(x.items()))
+        return _normalize_seq_func((type(x), x.tocoo()))
 
 
 def _colorize(t):
@@ -1664,11 +1670,11 @@ def clone_key(key: KeyOrStrT, seed: Hashable) -> KeyOrStrT:
     Examples
     --------
     >>> clone_key("x", 123)
-    'x-dc2b8d1c184c72c19faa81c797f8c6b0'
+    'x-c4fb64ccca807af85082413d7ef01721'
     >>> clone_key("inc-cbb1eca3bafafbb3e8b2419c4eebb387", 123)
-    'inc-f81b5a88038a2132882aa29a9fcfec06'
+    'inc-bc629c23014a4472e18b575fdaf29ee7'
     >>> clone_key(("sum-cbb1eca3bafafbb3e8b2419c4eebb387", 4, 3), 123)
-    ('sum-fd6be9e9fe07fc232ad576fa997255e8', 4, 3)
+    ('sum-c053f3774e09bd0f7de6044dbc40e71d', 4, 3)
     """
     if isinstance(key, tuple) and key and isinstance(key[0], str):
         return (clone_key(key[0], seed),) + key[1:]
diff --git a/dask/utils.py b/dask/utils.py
index 0da2953c8..0e974fe47 100644
--- a/dask/utils.py
+++ b/dask/utils.py
@@ -744,15 +744,6 @@ class Dispatch:
         """Return the function implementation for the given ``cls``"""
         lk = self._lookup
         for cls2 in cls.__mro__:
-            try:
-                impl = lk[cls2]
-            except KeyError:
-                pass
-            else:
-                if cls is not cls2:
-                    # Cache lookup
-                    lk[cls] = impl
-                return impl
             # Is a lazy registration function present?
             toplevel, _, _ = cls2.__module__.partition(".")
             try:
@@ -763,6 +754,15 @@ class Dispatch:
                 register()
                 self._lazy.pop(toplevel, None)
                 return self.dispatch(cls)  # recurse
+            try:
+                impl = lk[cls2]
+            except KeyError:
+                pass
+            else:
+                if cls is not cls2:
+                    # Cache lookup
+                    lk[cls] = impl
+                return impl
         raise TypeError(f"No dispatch for {cls}")
 
     def __call__(self, arg, *args, **kwargs):
diff --git a/docs/source/deploying.rst b/docs/source/deploying.rst
index 05f7a02c2..4d65e48cd 100644
--- a/docs/source/deploying.rst
+++ b/docs/source/deploying.rst
@@ -170,7 +170,7 @@ deployments this is often handled by other Kubernetes services.
        from dask_kubernetes.operator import KubeCluster
        cluster = KubeCluster(
            name="my-dask-cluster",
-           image='ghcr.io/dask/dask:latest',
+           image="ghcr.io/dask/dask:latest",
            resources={"requests": {"memory": "2Gi"}, "limits": {"memory": "64Gi"}},
        )
        cluster.scale(10)
diff --git a/docs/source/index.rst b/docs/source/index.rst
index 6056b1ce2..7b611cf2e 100644
--- a/docs/source/index.rst
+++ b/docs/source/index.rst
@@ -206,7 +206,7 @@ Dask provides several APIs.  Choose one that works best for you:
 How to Install Dask
 -------------------
 
-Installing Dask is easy with ``pip`` or ``conda``
+Installing Dask is easy with ``pip`` or ``conda``.
 
 Learn more at :bdg-link-primary:`Install Documentation <install.html>`
 
@@ -227,7 +227,7 @@ Learn more at :bdg-link-primary:`Install Documentation <install.html>`
 How to Deploy Dask
 ------------------
 
-You can then use Dask on a single machine, or deploy it on distributed hardware
+You can use Dask on a single machine, or deploy it on distributed hardware.
 
 Learn more at :bdg-link-primary:`Deploy Documentation <deploying.html>`
 
@@ -249,97 +249,64 @@ Learn more at :bdg-link-primary:`Deploy Documentation <deploying.html>`
        Alternatively, you can skip this part, and Dask will operate within a
        thread pool contained entirely with your local process.
 
-    .. tab-item:: Kubernetes
-
-       The `dask-kubernetes project <https://kubernetes.dask.org>`_ provides
-       a Dask Kubernetes Operator.
-
-       .. code-block:: python
-
-          from dask_kubernetes.operator import KubeCluster
-          cluster = KubeCluster(
-             name="my-dask-cluster",
-             image='ghcr.io/dask/dask:latest'
-          )
-          cluster.scale(10)
-
-       Learn more at :bdg-link-primary:`Dask Kubernetes Documentation <https://kubernetes.dask.org>`
-
-    .. tab-item:: HPC
-
-        The `dask-jobqueue project <https://jobqueue.dask.org>`_ interfaces
-        with popular job submission projects, like SLURM, PBS, SGE, LSF,
-        Torque, Condor, and others.
-
-
-        .. code-block:: python
-
-           from dask_jobqueue import SLURMCluster
-
-           cluster = SLURMCluster()
-           cluster.scale(jobs=10)
-
+    .. tab-item:: Cloud
 
-        You can also deploy Dask with MPI
+        `Coiled <https://docs.coiled.io/user_guide/index.html?utm_source=dask-docs&utm_medium=homepage>`_
+        is a commercial SaaS product that deploys Dask clusters on cloud platforms like AWS, GCP, and Azure.
 
         .. code-block:: python
 
-           # myscript.py
-           from dask_mpi import initialize
-           initialize()
-
-           from dask.distributed import Client
-           client = Client()  # Connect this local process to remote workers
-
-        .. code-block::
+            import coiled
+            cluster = coiled.Cluster(
+               n_workers=100,
+               region="us-east-2",
+               worker_memory="16 GiB",
+               spot_policy="spot_with_fallback",
+            )
+            client = cluster.get_client()
 
-           $ mpirun -np 4 python myscript.py
+        Learn more at :bdg-link-primary:`Coiled Documentation <https://docs.coiled.io/user_guide/index.html?utm_source=dask-docs&utm_medium=homepage>`
 
-        Learn more at :bdg-link-primary:`Dask Jobqueue Documentation <https://jobqueue.dask.org>` and the :bdg-link-primary:`Dask MPI Documentation <https://mpi.dask.org>`.
 
-    .. tab-item:: Cloud
+    .. tab-item:: HPC
 
-        The `dask-cloudprovider project <https://cloudprovider.dask.org>`_ interfaces
-        with popular cloud platforms like AWS, GCP, Azure, and Digital Ocean.
+        The `Dask-Jobqueue project <https://jobqueue.dask.org>`_ deploys
+        Dask clusters on popular HPC job submission systems like SLURM, PBS, SGE, LSF,
+        Torque, Condor, and others.
 
         .. code-block:: python
 
-           from dask_cloudprovider.aws import FargateCluster
-           cluster = FargateCluster(
-               # Cluster manager specific config kwargs
-           )
-
-        Learn more at :bdg-link-primary:`Dask CloudProvider Documentation <https://cloudprovider.dask.org>`
-
-    .. tab-item:: Managed/Commercial
+            from dask_jobqueue import PBSCluster
+            cluster = PBSCluster(
+               cores=24,
+               memory="100GB",
+               queue="regular",
+               account="my-account",
+            )
+            cluster.scale(jobs=100)
+            client = cluster.get_client()
 
-        Several companies offer commercial Dask products.  These are not open
-        source, but tend to be easier, safer, cheaper, more fully featured,
-        etc..  All options here include solid free offerings for individuals.
+        Learn more at :bdg-link-primary:`Dask-Jobqueue Documentation <https://jobqueue.dask.org>`
 
-        .. _Coiled: https://coiled.io/?utm_source=dask-docs&utm_medium=homepage
-        .. |coiled| replace:: **Coiled**
-        .. _saturn: https://saturncloud.io
-        .. |saturn| replace:: **Saturn Cloud**
-        .. _nebari: https://nebari.dev
-        .. |nebari| replace:: **Nebari**
+    .. tab-item:: Kubernetes
 
-        -  |Coiled|_ provides a standalone Dask deployment product that works
-           in AWS and GCP.
+       The `Dask Kubernetes project <https://kubernetes.dask.org>`_ provides
+       a Dask Kubernetes Operator for deploying Dask on Kubernetes clusters.
 
-           Coiled notably employs many of the active Dask maintainers today.
+       .. code-block:: python
 
-           Learn more at :bdg-link-primary:`Coiled <https://coiled.io/?utm_source=dask-docs&utm_medium=homepage>`
+            from dask_kubernetes.operator import KubeCluster
+            cluster = KubeCluster(
+               name="my-dask-cluster",
+               image="ghcr.io/dask/dask:latest",
+               resources={"requests": {"memory": "2Gi"}, "limits": {"memory": "64Gi"}},
+            )
+            cluster.scale(10)
+            client = cluster.get_client()
 
-        -  |saturn|_ provides Dask as part of their hosted platform
-           including Jupyter and other products.
 
-           Learn more at :bdg-link-primary:`Saturn Cloud <https://saturncloud.io>`
-        -  |nebari|_ from Quansight provides Dask as part of a Kubernetes-based
-           git-ops managed platform along with Jupyter and other products
-           suitable for on-prem deployments.
+       Learn more at :bdg-link-primary:`Dask Kubernetes Documentation <https://kubernetes.dask.org>`
 
-           Learn more at :bdg-link-primary:`Nebari <https://nebari.dev>`
 
 Learn with Examples
 -------------------


Test patch:
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index f4d4c5323..db2d769ee 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -105,7 +105,7 @@ def test_is_dask_collection_dask_expr_does_not_materialize():
 def test_unpack_collections():
     @dataclasses.dataclass
     class ADataClass:
-        a: int
+        a: int  # type: ignore[annotation-unchecked]
 
     class ANamedTuple(NamedTuple):
         a: int  # type: ignore[annotation-unchecked]
@@ -1007,12 +1007,12 @@ def test_optimizations_ctd():
 
 
 def test_clone_key():
-    assert clone_key("inc-1-2-3", 123) == "inc-4dfeea2f9300e67a75f30bf7d6182ea4"
-    assert clone_key("x", 123) == "x-dc2b8d1c184c72c19faa81c797f8c6b0"
-    assert clone_key("x", 456) == "x-b76f061b547b00d18b9c7a18ccc47e2d"
-    assert clone_key(("x", 1), 456) == ("x-b76f061b547b00d18b9c7a18ccc47e2d", 1)
+    assert clone_key("inc-1-2-3", 123) == "inc-73db79fdf4518507ddc84796726d4844"
+    assert clone_key("x", 123) == "x-c4fb64ccca807af85082413d7ef01721"
+    assert clone_key("x", 456) == "x-d4b538b4d4cf68fca214077609feebae"
+    assert clone_key(("x", 1), 456) == ("x-d4b538b4d4cf68fca214077609feebae", 1)
     assert clone_key(("sum-1-2-3", h1, 1), 123) == (
-        "sum-1efd41f02035dc802f4ebb9995d07e9d",
+        "sum-822e7622aa1262cef988b3033c32aa37",
         h1,
         1,
     )
diff --git a/dask/tests/test_delayed.py b/dask/tests/test_delayed.py
index dc9b796ad..a5a1e4335 100644
--- a/dask/tests/test_delayed.py
+++ b/dask/tests/test_delayed.py
@@ -649,15 +649,11 @@ def identity(x):
     return x
 
 
-def test_name_consistent_across_instances():
+def test_deterministic_name():
     func = delayed(identity, pure=True)
-
-    data = {"x": 1, "y": 25, "z": [1, 2, 3]}
-    assert func(data)._key == "identity-4f318f3c27b869239e97c3ac07f7201a"
-
-    data = {"x": 1, 1: "x"}
-    assert func(data)._key == func(data)._key
-    assert func(1)._key == "identity-7258833899272585e16d0ec36b21a3de"
+    data1 = {"x": 1, "y": 25, "z": [1, 2, 3]}
+    data2 = {"x": 1, "y": 25, "z": [1, 2, 3]}
+    assert func(data1)._key == func(data2)._key
 
 
 def test_sensitive_to_partials():
diff --git a/dask/tests/test_highgraph.py b/dask/tests/test_highgraph.py
index 03a21c763..2f65387e5 100644
--- a/dask/tests/test_highgraph.py
+++ b/dask/tests/test_highgraph.py
@@ -9,7 +9,7 @@ from concurrent.futures import ThreadPoolExecutor
 import pytest
 
 import dask
-from dask.base import normalize_token
+from dask.base import tokenize
 from dask.blockwise import Blockwise, blockwise_token
 from dask.highlevelgraph import HighLevelGraph, Layer, MaterializedLayer, to_graphviz
 from dask.utils_test import inc
@@ -309,5 +309,5 @@ def test_tokenize_hlg():
     a = db.from_sequence(list(range(10)), npartitions=2).max()
     b = db.from_sequence(list(range(10)), npartitions=2).max()
     c = db.from_sequence(list(range(10)), npartitions=3).max()
-    assert normalize_token(a.dask) == normalize_token(b.dask)
-    assert normalize_token(a.dask) != normalize_token(c.dask)
+    assert tokenize(a.dask) == tokenize(b.dask)
+    assert tokenize(a.dask) != tokenize(c.dask)
diff --git a/dask/tests/test_tokenize.py b/dask/tests/test_tokenize.py
index cb29333a1..59d9a2c6a 100644
--- a/dask/tests/test_tokenize.py
+++ b/dask/tests/test_tokenize.py
@@ -13,6 +13,7 @@ import textwrap
 from enum import Enum, Flag, IntEnum, IntFlag
 from typing import Union
 
+import cloudpickle
 import pytest
 from tlz import compose, curry, partial
 
@@ -22,6 +23,7 @@ from dask.core import literal
 from dask.utils import tmpfile
 from dask.utils_test import import_or_none
 
+da = import_or_none("dask.array")
 dd = import_or_none("dask.dataframe")
 np = import_or_none("numpy")
 sp = import_or_none("scipy.sparse")
@@ -62,69 +64,175 @@ def test_normalize_function():
     assert normalize_function(curry(f2, b=1)) != normalize_function(curry(f2, b=2))
 
 
+def _clear_function_cache():
+    from dask.base import function_cache, function_cache_lock
+
+    with function_cache_lock:
+        function_cache.clear()
+
+
+def check_tokenize(*args, idempotent=True, deterministic=None, copy=None, **kwargs):
+    """Wrapper around tokenize
+
+    Parameters
+    ----------
+    args, kwargs: passed to tokenize
+    idempotent: True or False
+        If True, expect tokenize() called on the same object twice to produce the same
+        result. If False, expect different results. Default: True
+    deterministic: True, False, or "maybe"
+        If True, expect tokenize() called on two identical copies of an object to
+        produce the same result. If False, expect different results. If "maybe", expect
+        nothing. Default: same as idempotent
+    copy: callable or False
+        f(T)->T that deep-copies the object. Default: distributed serialize
+    """
+    if deterministic is None:
+        deterministic = idempotent
+
+    if copy is None:
+        copy = lambda x: cloudpickle.loads(cloudpickle.dumps(x))
+
+    ensure_deterministic = deterministic is True  # not maybe
+    with dask.config.set({"tokenize.ensure-deterministic": ensure_deterministic}):
+        before = tokenize(*args, **kwargs)
+
+        # Test idempotency (the same object tokenizes to the same value)
+        _clear_function_cache()
+        after = tokenize(*args, **kwargs)
+
+        assert (before == after) is idempotent
+
+        # Test same-interpreter determinism (two identical objects tokenize to
+        # the same value as long as you do it on the same interpreter)
+        if copy:
+            args2, kwargs2 = copy((args, kwargs))
+            _clear_function_cache()
+            after = tokenize(*args2, **kwargs2)
+
+            if deterministic != "maybe":
+                assert (before == after) is deterministic
+
+        # Skip: different interpreter determinism
+
+    return before
+
+
+def test_check_tokenize():
+    # idempotent and deterministic
+    check_tokenize(123)
+    # returns tokenized value
+    assert tokenize(123) == check_tokenize(123)
+
+    # idempotent, but not deterministic
+    class A:
+        def __init__(self):
+            self.tok = random.random()
+
+        def __reduce_ex__(self, protocol):
+            return A, ()
+
+        def __dask_tokenize__(self):
+            return self.tok
+
+    a = A()
+    with pytest.raises(AssertionError):
+        check_tokenize(a)
+    check_tokenize(a, deterministic=False)
+    check_tokenize(a, deterministic="maybe")
+
+    # Not idempotent
+    class B:
+        def __dask_tokenize__(self):
+            return random.random()
+
+    b = B()
+    check_tokenize(b, idempotent=False)
+    with pytest.raises(AssertionError):
+        check_tokenize(b)
+    with pytest.raises(AssertionError):
+        check_tokenize(b, deterministic=False)
+
+
 def test_tokenize():
     a = (1, 2, 3)
-    assert isinstance(tokenize(a), (str, bytes))
+    assert isinstance(tokenize(a), str)
+
+
+@pytest.mark.skipif("not np")
+def test_tokenize_scalar():
+    assert check_tokenize(np.int64(1)) != check_tokenize(1)
+    assert check_tokenize(np.int64(1)) != check_tokenize(np.int32(1))
+    assert check_tokenize(np.int64(1)) != check_tokenize(np.uint32(1))
+    assert check_tokenize(np.int64(1)) != check_tokenize("1")
+    assert check_tokenize(np.int64(1)) != check_tokenize(np.float64(1))
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_array_consistent_on_values():
-    assert tokenize(np.random.RandomState(1234).random_sample(1000)) == tokenize(
+    assert check_tokenize(
         np.random.RandomState(1234).random_sample(1000)
-    )
+    ) == check_tokenize(np.random.RandomState(1234).random_sample(1000))
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_array_supports_uneven_sizes():
-    tokenize(np.random.random(7).astype(dtype="i2"))
+    check_tokenize(np.random.random(7).astype(dtype="i2"))
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_discontiguous_numpy_array():
-    tokenize(np.random.random(8)[::2])
+    check_tokenize(np.random.random(8)[::2])
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_datetime():
-    tokenize(np.array(["2000-01-01T12:00:00"], dtype="M8[ns]"))
+    check_tokenize(np.array(["2000-01-01T12:00:00"], dtype="M8[ns]"))
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_scalar():
-    assert tokenize(np.array(1.0, dtype="f8")) == tokenize(np.array(1.0, dtype="f8"))
-    assert tokenize(
+    assert check_tokenize(np.array(1.0, dtype="f8")) == check_tokenize(
+        np.array(1.0, dtype="f8")
+    )
+    assert check_tokenize(
         np.array([(1, 2)], dtype=[("a", "i4"), ("b", "i8")])[0]
-    ) == tokenize(np.array([(1, 2)], dtype=[("a", "i4"), ("b", "i8")])[0])
+    ) == check_tokenize(np.array([(1, 2)], dtype=[("a", "i4"), ("b", "i8")])[0])
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_scalar_string_rep():
     # Test tokenizing numpy scalars doesn't depend on their string representation
     with np.printoptions(formatter={"all": lambda x: "foo"}):
-        assert tokenize(np.array(1)) != tokenize(np.array(2))
+        assert check_tokenize(np.array(1)) != check_tokenize(np.array(2))
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_array_on_object_dtype():
     a = np.array(["a", "aa", "aaa"], dtype=object)
-    assert tokenize(a) == tokenize(a)
-    assert tokenize(np.array(["a", None, "aaa"], dtype=object)) == tokenize(
+    assert check_tokenize(a) == check_tokenize(a)
+    assert check_tokenize(np.array(["a", None, "aaa"], dtype=object)) == check_tokenize(
         np.array(["a", None, "aaa"], dtype=object)
     )
-    assert tokenize(
+    assert check_tokenize(
         np.array([(1, "a"), (1, None), (1, "aaa")], dtype=object)
-    ) == tokenize(np.array([(1, "a"), (1, None), (1, "aaa")], dtype=object))
+    ) == check_tokenize(np.array([(1, "a"), (1, None), (1, "aaa")], dtype=object))
 
     # Trigger non-deterministic hashing for object dtype
     class NoPickle:
         pass
 
     x = np.array(["a", None, NoPickle], dtype=object)
-    assert tokenize(x) != tokenize(x)
+    check_tokenize(x, idempotent=False)
+
 
-    with dask.config.set({"tokenize.ensure-deterministic": True}):
-        with pytest.raises(RuntimeError, match="cannot be deterministically hashed"):
-            tokenize(x)
+@pytest.mark.skipif("not np")
+def test_empty_numpy_array():
+    arr = np.array([])
+    assert arr.strides
+    assert check_tokenize(arr) == check_tokenize(arr)
+    arr2 = np.array([], dtype=np.int64())
+    assert check_tokenize(arr) != check_tokenize(arr2)
 
 
 @pytest.mark.skipif("not np")
@@ -138,40 +246,49 @@ def test_tokenize_numpy_memmap_offset(tmpdir):
     with open(fn, "rb") as f:
         mmap1 = np.memmap(f, dtype=np.uint8, mode="r", offset=0, shape=5)
         mmap2 = np.memmap(f, dtype=np.uint8, mode="r", offset=5, shape=5)
-
-        assert tokenize(mmap1) != tokenize(mmap2)
+        mmap3 = np.memmap(f, dtype=np.uint8, mode="r", offset=0, shape=5)
+        assert check_tokenize(mmap1) == check_tokenize(mmap1)
+        assert check_tokenize(mmap1) == check_tokenize(mmap3)
+        assert check_tokenize(mmap1) != check_tokenize(mmap2)
         # also make sure that they tokenize correctly when taking sub-arrays
+        assert check_tokenize(mmap1[1:-1]) == check_tokenize(mmap1[1:-1])
+        assert check_tokenize(mmap1[1:-1]) == check_tokenize(mmap3[1:-1])
+        assert check_tokenize(mmap1[1:2]) == check_tokenize(mmap3[1:2])
+        assert check_tokenize(mmap1[1:2]) != check_tokenize(mmap1[1:3])
+        assert check_tokenize(mmap1[1:2]) != check_tokenize(mmap3[1:3])
         sub1 = mmap1[1:-1]
         sub2 = mmap2[1:-1]
-        assert tokenize(sub1) != tokenize(sub2)
+        assert check_tokenize(sub1) != check_tokenize(sub2)
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_memmap():
     with tmpfile(".npy") as fn:
-        x = np.arange(5)
-        np.save(fn, x)
-        y = tokenize(np.load(fn, mmap_mode="r"))
+        x1 = np.arange(5)
+        np.save(fn, x1)
+        y = check_tokenize(np.load(fn, mmap_mode="r"))
 
     with tmpfile(".npy") as fn:
-        x = np.arange(5)
-        np.save(fn, x)
-        z = tokenize(np.load(fn, mmap_mode="r"))
+        x2 = np.arange(5)
+        np.save(fn, x2)
+        z = check_tokenize(np.load(fn, mmap_mode="r"))
 
-    assert y != z
+    assert check_tokenize(x1) == check_tokenize(x2)
+    # Memory maps should behave similar to ordinary arrays
+    assert y == z
 
     with tmpfile(".npy") as fn:
         x = np.random.normal(size=(10, 10))
         np.save(fn, x)
         mm = np.load(fn, mmap_mode="r")
         mm2 = np.load(fn, mmap_mode="r")
-        a = tokenize(mm[0, :])
-        b = tokenize(mm[1, :])
-        c = tokenize(mm[0:3, :])
-        d = tokenize(mm[:, 0])
+        a = check_tokenize(mm[0, :])
+        b = check_tokenize(mm[1, :])
+        c = check_tokenize(mm[0:3, :])
+        d = check_tokenize(mm[:, 0])
         assert len({a, b, c, d}) == 4
-        assert tokenize(mm) == tokenize(mm2)
-        assert tokenize(mm[1, :]) == tokenize(mm2[1, :])
+        assert check_tokenize(mm) == check_tokenize(mm2)
+        assert check_tokenize(mm[1, :]) == check_tokenize(mm2[1, :])
 
 
 @pytest.mark.skipif("not np")
@@ -184,39 +301,32 @@ def test_tokenize_numpy_memmap_no_filename():
 
         a = np.load(fn1, mmap_mode="r")
         b = a + a
-        assert tokenize(b) == tokenize(b)
+        assert check_tokenize(b) == check_tokenize(b)
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_numpy_ufunc_consistent():
-    assert tokenize(np.sin) == "02106e2c67daf452fb480d264e0dac21"
-    assert tokenize(np.cos) == "c99e52e912e4379882a9a4b387957a0b"
+    assert check_tokenize(np.sin) == check_tokenize("np.sin")
+    assert check_tokenize(np.cos) == check_tokenize("np.cos")
 
     # Make a ufunc that isn't in the numpy namespace. Similar to
     # any found in other packages.
     inc = np.frompyfunc(lambda x: x + 1, 1, 1)
-    assert tokenize(inc) == tokenize(inc)
+    check_tokenize(inc, copy=False, deterministic=False)
+
+    np_ufunc = np.sin
+    np_ufunc2 = np.cos
+    assert isinstance(np_ufunc, np.ufunc)
+    assert isinstance(np_ufunc2, np.ufunc)
+    assert check_tokenize(np_ufunc) != check_tokenize(np_ufunc2)
 
 
 def test_tokenize_partial_func_args_kwargs_consistent():
     f = partial(f3, f2, c=f1)
-    res = normalize_token(f)
-    sol = (
-        b"\x80\x04\x95#\x00\x00\x00\x00\x00\x00\x00\x8c\x18"
-        b"dask.tests.test_tokenize\x94\x8c\x02f3\x94\x93\x94.",
-        (
-            b"\x80\x04\x95#\x00\x00\x00\x00\x00\x00\x00\x8c\x18"
-            b"dask.tests.test_tokenize\x94\x8c\x02f2\x94\x93\x94.",
-        ),
-        (
-            (
-                "c",
-                b"\x80\x04\x95#\x00\x00\x00\x00\x00\x00\x00\x8c\x18"
-                b"dask.tests.test_tokenize\x94\x8c\x02f1\x94\x93\x94.",
-            ),
-        ),
-    )
-    assert res == sol
+    g = partial(f3, f2, c=f1)
+    h = partial(f3, f2, c=5)
+    assert check_tokenize(f) == check_tokenize(g)
+    assert check_tokenize(f) != check_tokenize(h)
 
 
 def test_normalize_base():
@@ -233,33 +343,50 @@ def test_normalize_base():
 
 
 def test_tokenize_object():
-    o = object()
-    # Defaults to non-deterministic tokenization
-    assert normalize_token(o) != normalize_token(o)
+    check_tokenize(object(), idempotent=False)
 
-    with dask.config.set({"tokenize.ensure-deterministic": True}):
-        with pytest.raises(RuntimeError, match="cannot be deterministically hashed"):
-            normalize_token(o)
 
+_GLOBAL = 1
 
-def test_tokenize_function_cloudpickle():
-    a, b = (lambda x: x, lambda x: x)
-    # No error by default
-    tokenize(a)
 
-    if dd and dd._dask_expr_enabled():
-        return  # dask-expr does a check and serializes if possible
+def test_tokenize_local_functions():
+    a, b, c, d, e = (
+        # Note: Same line, same code lambdas cannot be distinguished
+        lambda x: x,
+        lambda x: x + 1,
+        lambda x: x,
+        lambda y: y,
+        lambda y: y + 1,
+    )
 
-    with dask.config.set({"tokenize.ensure-deterministic": True}):
-        with pytest.raises(RuntimeError, match="may not be deterministically hashed"):
-            tokenize(b)
+    def f(x):
+        return x
 
+    local_scope = 1
+
+    def g():
+        nonlocal local_scope
+        local_scope += 1
+        return e(local_scope)
+
+    def h():
+        global _GLOBAL
+        _GLOBAL += 1
+        return _GLOBAL
+
+    all_funcs = [a, b, c, d, e, f, g, h]
+
+    # Lambdas serialize differently after a cloudpickle roundtrip
+    tokens = [check_tokenize(func, deterministic="maybe") for func in all_funcs]
+    assert len(set(tokens)) == len(all_funcs)
 
-def test_tokenize_callable():
-    def my_func(a, b, c=1):
-        return a + b + c
 
-    assert tokenize(my_func) == tokenize(my_func)  # Consistent token
+def my_func(a, b, c=1):
+    return a + b + c
+
+
+def test_tokenize_callable():
+    check_tokenize(my_func)
 
 
 @pytest.mark.skipif("not pd")
@@ -267,16 +394,16 @@ def test_tokenize_pandas():
     a = pd.DataFrame({"x": [1, 2, 3], "y": ["4", "asd", None]}, index=[1, 2, 3])
     b = pd.DataFrame({"x": [1, 2, 3], "y": ["4", "asd", None]}, index=[1, 2, 3])
 
-    assert tokenize(a) == tokenize(b)
+    assert check_tokenize(a) == check_tokenize(b)
     b.index.name = "foo"
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
     a = pd.DataFrame({"x": [1, 2, 3], "y": ["a", "b", "a"]})
     b = pd.DataFrame({"x": [1, 2, 3], "y": ["a", "b", "a"]})
     a["z"] = a.y.astype("category")
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
     b["z"] = a.y.astype("category")
-    assert tokenize(a) == tokenize(b)
+    assert check_tokenize(a) == check_tokenize(b)
 
 
 @pytest.mark.skipif("not pd")
@@ -285,7 +412,7 @@ def test_tokenize_pandas_invalid_unicode():
     df = pd.DataFrame(
         {"x\ud83d": [1, 2, 3], "y\ud83d": ["4", "asd\ud83d", None]}, index=[1, 2, 3]
     )
-    tokenize(df)
+    check_tokenize(df)
 
 
 @pytest.mark.skipif("not pd")
@@ -294,7 +421,7 @@ def test_tokenize_pandas_mixed_unicode_bytes():
         {"ö".encode(): [1, 2, 3], "ö": ["ö", "ö".encode(), None]},
         index=[1, 2, 3],
     )
-    tokenize(df)
+    check_tokenize(df)
 
 
 @pytest.mark.skipif("not pd")
@@ -304,7 +431,7 @@ def test_tokenize_pandas_no_pickle():
         pass
 
     df = pd.DataFrame({"x": ["foo", None, NoPickle()]})
-    tokenize(df)
+    check_tokenize(df, idempotent=False)
 
 
 @pytest.mark.skipif("not dd")
@@ -329,12 +456,12 @@ def test_tokenize_pandas_extension_array():
     )
 
     for arr in arrays:
-        assert tokenize(arr) == tokenize(arr)
+        check_tokenize(arr)
 
 
 @pytest.mark.skipif("not pd")
 def test_tokenize_na():
-    assert tokenize(pd.NA) == tokenize(pd.NA)
+    check_tokenize(pd.NA)
 
 
 @pytest.mark.skipif("not pd")
@@ -348,35 +475,24 @@ def test_tokenize_offset():
         pd.DateOffset(months=7),
         pd.DateOffset(days=10),
     ]:
-        assert tokenize(offset) == tokenize(offset)
+        check_tokenize(offset)
 
 
 @pytest.mark.skipif("not pd")
 def test_tokenize_pandas_index():
     idx = pd.Index(["a", "b"])
-    assert tokenize(idx) == tokenize(idx)
+    check_tokenize(idx)
 
     idx = pd.MultiIndex.from_product([["a", "b"], [0, 1]])
-    assert tokenize(idx) == tokenize(idx)
+    check_tokenize(idx)
 
 
 def test_tokenize_kwargs():
-    assert tokenize(5, x=1) == tokenize(5, x=1)
-    assert tokenize(5) != tokenize(5, x=1)
-    assert tokenize(5, x=1) != tokenize(5, x=2)
-    assert tokenize(5, x=1) != tokenize(5, y=1)
-    assert tokenize(5, foo="bar") != tokenize(5, {"foo": "bar"})
-
-
-def test_tokenize_same_repr():
-    class Foo:
-        def __init__(self, x):
-            self.x = x
-
-        def __repr__(self):
-            return "a foo"
-
-    assert tokenize(Foo(1)) != tokenize(Foo(2))
+    check_tokenize(5, x=1)
+    assert check_tokenize(5) != check_tokenize(5, x=1)
+    assert check_tokenize(5, x=1) != check_tokenize(5, x=2)
+    assert check_tokenize(5, x=1) != check_tokenize(5, y=1)
+    assert check_tokenize(5, foo="bar") != check_tokenize(5, {"foo": "bar"})
 
 
 def test_tokenize_method():
@@ -391,70 +507,91 @@ def test_tokenize_method():
             return "Hello world"
 
     a, b = Foo(1), Foo(2)
-    assert tokenize(a) == tokenize(a)
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
-    assert tokenize(a.hello) == tokenize(a.hello)
-    assert tokenize(a.hello) != tokenize(b.hello)
-
-    for ensure in [True, False]:
-        with dask.config.set({"tokenize.ensure-deterministic": ensure}):
-            assert tokenize(a) == tokenize(a)
-            assert tokenize(a.hello) == tokenize(a.hello)
+    assert check_tokenize(a.hello) != check_tokenize(b.hello)
 
     # dispatch takes precedence
-    before = tokenize(a)
+    before = check_tokenize(a)
     normalize_token.register(Foo, lambda self: self.x + 1)
-    after = tokenize(a)
+    after = check_tokenize(a)
     assert before != after
+    del normalize_token._lookup[Foo]
 
 
-def test_staticmethods():
-    class Foo:
-        def __init__(self, x):
+def test_tokenize_callable_class():
+    """___dask_tokenize__ takes precedence over callable()"""
+
+    class C:
+        def __init__(self, x, y):
             self.x = x
+            self.y = y
 
-        def normal_method(self):
-            pass
+        def __dask_tokenize__(self):
+            return self.x
 
-        @staticmethod
-        def static_method():
+        def __call__(self):
             pass
 
-        @classmethod
-        def class_method(cls):
-            pass
+    assert check_tokenize(C(1, 2)) == check_tokenize(C(1, 3))
+    assert check_tokenize(C(1, 2)) != check_tokenize(C(2, 2))
+
 
-    class Bar(Foo):
+class HasStaticMethods:
+    def __init__(self, x):
+        self.x = x
+
+    def __dask_tokenize__(self):
+        return normalize_token(type(self)), self.x
+
+    def normal_method(self):
         pass
 
-    a, b, c = Foo(1), Foo(2), Bar(1)
-    assert tokenize(a) != tokenize(b)
-    assert tokenize(a.normal_method) != tokenize(b.normal_method)
-    assert tokenize(a.static_method) == tokenize(b.static_method)
-    assert tokenize(a.static_method) == tokenize(c.static_method)
-    assert tokenize(a.class_method) == tokenize(b.class_method)
-    assert tokenize(a.class_method) != tokenize(c.class_method)
+    @staticmethod
+    def static_method():
+        pass
+
+    @classmethod
+    def class_method(cls):
+        pass
+
+
+class HasStaticMethods2(HasStaticMethods):
+    pass
+
+
+def test_staticmethods():
+    a, b, c = HasStaticMethods(1), HasStaticMethods(2), HasStaticMethods2(1)
+    # These invoke HasStaticMethods.__dask_tokenize__()
+    assert check_tokenize(a.normal_method) != check_tokenize(b.normal_method)
+    assert check_tokenize(a.normal_method) != check_tokenize(c.normal_method)
+    # These don't
+    assert check_tokenize(a.static_method) == check_tokenize(b.static_method)
+    assert check_tokenize(a.static_method) == check_tokenize(c.static_method)
+    assert check_tokenize(a.class_method) == check_tokenize(b.class_method)
+    assert check_tokenize(a.class_method) != check_tokenize(c.class_method)
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_sequences():
-    assert tokenize([1]) != tokenize([2])
-    assert tokenize([1]) != tokenize((1,))
-    assert tokenize([1]) == tokenize([1])
+    assert check_tokenize([1]) != check_tokenize([2])
+    assert check_tokenize([1]) != check_tokenize((1,))
+    assert check_tokenize([1]) == check_tokenize([1])
 
     x = np.arange(2000)  # long enough to drop information in repr
     y = np.arange(2000)
     y[1000] = 0  # middle isn't printed in repr
-    assert tokenize([x]) != tokenize([y])
+    assert check_tokenize([x]) != check_tokenize([y])
 
 
 def test_tokenize_dict():
-    assert tokenize({"x": 1, 1: "x"}) == tokenize({"x": 1, 1: "x"})
+    assert check_tokenize({"x": 1, 1: "x"}) == check_tokenize({"x": 1, 1: "x"})
 
 
 def test_tokenize_set():
-    assert tokenize({1, 2, "x", (1, "x")}) == tokenize({1, 2, "x", (1, "x")})
+    assert check_tokenize({1, 2, "x", (1, "x")}) == check_tokenize(
+        {1, 2, "x", (1, "x")}
+    )
 
 
 def test_tokenize_ordered_dict():
@@ -464,13 +601,17 @@ def test_tokenize_ordered_dict():
     b = OrderedDict([("a", 1), ("b", 2)])
     c = OrderedDict([("b", 2), ("a", 1)])
 
-    assert tokenize(a) == tokenize(b)
-    assert tokenize(a) != tokenize(c)
+    assert check_tokenize(a) == check_tokenize(b)
+    assert check_tokenize(a) != check_tokenize(c)
 
 
 def test_tokenize_timedelta():
-    assert tokenize(datetime.timedelta(days=1)) == tokenize(datetime.timedelta(days=1))
-    assert tokenize(datetime.timedelta(days=1)) != tokenize(datetime.timedelta(days=2))
+    assert check_tokenize(datetime.timedelta(days=1)) == check_tokenize(
+        datetime.timedelta(days=1)
+    )
+    assert check_tokenize(datetime.timedelta(days=1)) != check_tokenize(
+        datetime.timedelta(days=2)
+    )
 
 
 @pytest.mark.parametrize("enum_type", [Enum, IntEnum, IntFlag, Flag])
@@ -479,8 +620,8 @@ def test_tokenize_enum(enum_type):
         RED = 1
         BLUE = 2
 
-    assert tokenize(Color.RED) == tokenize(Color.RED)
-    assert tokenize(Color.RED) != tokenize(Color.BLUE)
+    assert check_tokenize(Color.RED) == check_tokenize(Color.RED)
+    assert check_tokenize(Color.RED) != check_tokenize(Color.BLUE)
 
 
 @dataclasses.dataclass
@@ -493,54 +634,72 @@ class BDataClass:
     a: float
 
 
+@dataclasses.dataclass
+class NoValueDataClass:
+    a: int = dataclasses.field(init=False)
+
+
+class GlobalClass:
+    def __init__(self, val) -> None:
+        self.val = val
+
+
 def test_tokenize_dataclass():
     a1 = ADataClass(1)
     a2 = ADataClass(2)
-    assert tokenize(a1) == tokenize(a1)
-    assert tokenize(a1) != tokenize(a2)
+    check_tokenize(a1)
+    assert check_tokenize(a1) != check_tokenize(a2)
 
     # Same field names and values, but dataclass types are different
     b1 = BDataClass(1)
-    assert tokenize(a1) != tokenize(b1)
+    assert check_tokenize(a1) != check_tokenize(b1)
 
     class SubA(ADataClass):
         pass
 
-    assert dataclasses.is_dataclass(
-        SubA
-    ), "Python regression: SubA should be considered a dataclass"
-    assert tokenize(SubA(1)) == tokenize(SubA(1))
-    assert tokenize(SubA(1)) != tokenize(a1)
+    assert dataclasses.is_dataclass(SubA)
+    assert check_tokenize(SubA(1), deterministic=False) != check_tokenize(a1)
 
     # Same name, same values, new definition: tokenize differently
     ADataClassRedefinedDifferently = dataclasses.make_dataclass(
         "ADataClass", [("a", Union[int, str])]
     )
-    assert tokenize(a1) != tokenize(ADataClassRedefinedDifferently(1))
+    assert check_tokenize(a1) != check_tokenize(
+        ADataClassRedefinedDifferently(1), deterministic=False
+    )
+
+    # Dataclass with unpopulated value
+    nv = NoValueDataClass()
+    check_tokenize(nv)
 
 
-def test_tokenize_range():
-    assert tokenize(range(5, 10, 2)) == tokenize(range(5, 10, 2))  # Identical ranges
-    assert tokenize(range(5, 10, 2)) != tokenize(range(1, 10, 2))  # Different start
-    assert tokenize(range(5, 10, 2)) != tokenize(range(5, 15, 2))  # Different stop
-    assert tokenize(range(5, 10, 2)) != tokenize(range(5, 10, 1))  # Different step
+@pytest.mark.parametrize(
+    "other",
+    [
+        (1, 10, 2),  # Different start
+        (5, 15, 2),  # Different stop
+        (5, 10, 1),  # Different step
+    ],
+)
+def test_tokenize_range(other):
+    assert check_tokenize(range(5, 10, 2)) != check_tokenize(range(*other))
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_object_array_with_nans():
     a = np.array(["foo", "Jos\xe9", np.nan], dtype="O")
-    assert tokenize(a) == tokenize(a)
+    check_tokenize(a)
 
 
 @pytest.mark.parametrize(
     "x", [1, True, "a", b"a", 1.0, 1j, 1.0j, [], (), {}, None, str, int]
 )
 def test_tokenize_base_types(x):
-    assert tokenize(x) == tokenize(x), x
+    check_tokenize(x)
 
 
 def test_tokenize_literal():
-    assert tokenize(literal(["x", 1])) == tokenize(literal(["x", 1]))
+    assert check_tokenize(literal(["x", 1])) != check_tokenize(literal(["x", 2]))
 
 
 @pytest.mark.skipif("not np")
@@ -549,21 +708,21 @@ def test_tokenize_numpy_matrix():
     rng = np.random.RandomState(1234)
     a = np.asmatrix(rng.rand(100))
     b = a.copy()
-    assert tokenize(a) == tokenize(b)
+    assert check_tokenize(a) == check_tokenize(b)
 
     b[:10] = 1
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
 
 @pytest.mark.skipif("not sp")
-@pytest.mark.parametrize("cls_name", ("dia", "bsr", "coo", "csc", "csr", "dok", "lil"))
+@pytest.mark.parametrize("cls_name", ("dok",))
 def test_tokenize_dense_sparse_array(cls_name):
     rng = np.random.RandomState(1234)
 
-    a = sp.rand(10, 10000, random_state=rng).asformat(cls_name)
+    a = sp.rand(10, 100, random_state=rng).asformat(cls_name)
     b = a.copy()
 
-    assert tokenize(a) == tokenize(b)
+    assert check_tokenize(a) == check_tokenize(b)
 
     # modifying the data values
     if hasattr(b, "data"):
@@ -572,70 +731,85 @@ def test_tokenize_dense_sparse_array(cls_name):
         b[3, 3] = 1
     else:
         raise ValueError
-
-    assert tokenize(a) != tokenize(b)
+    check_tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
     # modifying the data indices
     b = a.copy().asformat("coo")
     b.row[:10] = np.arange(10)
     b = b.asformat(cls_name)
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
 
-@pytest.mark.skipif(
-    sys.platform == "win32" and sys.version_info[:2] == (3, 9),
-    reason="https://github.com/ipython/ipython/issues/12197",
-)
-def test_tokenize_object_with_recursion_error():
-    cycle = dict(a=None)
-    cycle["a"] = cycle
+def test_tokenize_circular_recursion():
+    a = [1, 2]
+    a[0] = a
 
-    assert len(tokenize(cycle)) == 32
+    # Test that tokenization doesn't stop as soon as you hit the circular recursion
+    b = [1, 3]
+    b[0] = b
 
-    with dask.config.set({"tokenize.ensure-deterministic": True}):
-        with pytest.raises(RuntimeError, match="cannot be deterministically hashed"):
-            tokenize(cycle)
+    def copy(x):
+        return pickle.loads(pickle.dumps(x))
 
+    assert check_tokenize(a, copy=copy) != check_tokenize(b, copy=copy)
+
+    # Different circular recursions tokenize differently
+    c = [[], []]
+    c[0].append(c[0])
+    c[1].append(c[1])
+
+    d = [[], []]
+    d[0].append(d[1])
+    d[1].append(d[0])
+    assert check_tokenize(c, copy=copy) != check_tokenize(d, copy=copy)
+
+    # For dicts, the dict itself is not passed to _normalize_seq_func
+    e = {}
+    e[0] = e
+    check_tokenize(e, copy=copy)
 
-def test_tokenize_datetime_date():
-    # Same date
-    assert tokenize(datetime.date(2021, 6, 25)) == tokenize(datetime.date(2021, 6, 25))
-    # Different year
-    assert tokenize(datetime.date(2021, 6, 25)) != tokenize(datetime.date(2022, 6, 25))
-    # Different month
-    assert tokenize(datetime.date(2021, 6, 25)) != tokenize(datetime.date(2021, 7, 25))
-    # Different day
-    assert tokenize(datetime.date(2021, 6, 25)) != tokenize(datetime.date(2021, 6, 26))
+
+@pytest.mark.parametrize(
+    "other",
+    [
+        (2002, 6, 25),  # Different year
+        (2021, 7, 25),  # Different month
+        (2021, 6, 26),  # Different day
+    ],
+)
+def test_tokenize_datetime_date(other):
+    a = datetime.date(2021, 6, 25)
+    b = datetime.date(*other)
+    assert check_tokenize(a) != check_tokenize(b)
 
 
 def test_tokenize_datetime_time():
     # Same time
-    assert tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc)) == tokenize(
-        datetime.time(1, 2, 3, 4, datetime.timezone.utc)
-    )
-    assert tokenize(datetime.time(1, 2, 3, 4)) == tokenize(datetime.time(1, 2, 3, 4))
-    assert tokenize(datetime.time(1, 2, 3)) == tokenize(datetime.time(1, 2, 3))
-    assert tokenize(datetime.time(1, 2)) == tokenize(datetime.time(1, 2))
+    check_tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc))
+    check_tokenize(datetime.time(1, 2, 3, 4))
+    check_tokenize(datetime.time(1, 2, 3))
+    check_tokenize(datetime.time(1, 2))
     # Different hour
-    assert tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc)) != tokenize(
-        datetime.time(2, 2, 3, 4, datetime.timezone.utc)
-    )
+    assert check_tokenize(
+        datetime.time(1, 2, 3, 4, datetime.timezone.utc)
+    ) != check_tokenize(datetime.time(2, 2, 3, 4, datetime.timezone.utc))
     # Different minute
-    assert tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc)) != tokenize(
-        datetime.time(1, 3, 3, 4, datetime.timezone.utc)
-    )
+    assert check_tokenize(
+        datetime.time(1, 2, 3, 4, datetime.timezone.utc)
+    ) != check_tokenize(datetime.time(1, 3, 3, 4, datetime.timezone.utc))
     # Different second
-    assert tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc)) != tokenize(
-        datetime.time(1, 2, 4, 4, datetime.timezone.utc)
-    )
+    assert check_tokenize(
+        datetime.time(1, 2, 3, 4, datetime.timezone.utc)
+    ) != check_tokenize(datetime.time(1, 2, 4, 4, datetime.timezone.utc))
     # Different micros
-    assert tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc)) != tokenize(
-        datetime.time(1, 2, 3, 5, datetime.timezone.utc)
-    )
+    assert check_tokenize(
+        datetime.time(1, 2, 3, 4, datetime.timezone.utc)
+    ) != check_tokenize(datetime.time(1, 2, 3, 5, datetime.timezone.utc))
     # Different tz
-    assert tokenize(datetime.time(1, 2, 3, 4, datetime.timezone.utc)) != tokenize(
-        datetime.time(1, 2, 3, 4)
-    )
+    assert check_tokenize(
+        datetime.time(1, 2, 3, 4, datetime.timezone.utc)
+    ) != check_tokenize(datetime.time(1, 2, 3, 4))
 
 
 def test_tokenize_datetime_datetime():
@@ -644,40 +818,40 @@ def test_tokenize_datetime_datetime():
     optional = [4, 5, 6, 7, datetime.timezone.utc]
     for i in range(len(optional) + 1):
         args = required + optional[:i]
-        assert tokenize(datetime.datetime(*args)) == tokenize(datetime.datetime(*args))
+        check_tokenize(datetime.datetime(*args))
 
     # Different year
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(2, 2, 3, 4, 5, 6, 7, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(2, 2, 3, 4, 5, 6, 7, datetime.timezone.utc))
     # Different month
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 1, 3, 4, 5, 6, 7, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(1, 1, 3, 4, 5, 6, 7, datetime.timezone.utc))
     # Different day
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 2, 2, 4, 5, 6, 7, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(1, 2, 2, 4, 5, 6, 7, datetime.timezone.utc))
     # Different hour
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 2, 3, 3, 5, 6, 7, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(1, 2, 3, 3, 5, 6, 7, datetime.timezone.utc))
     # Different minute
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 2, 3, 4, 4, 6, 7, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(1, 2, 3, 4, 4, 6, 7, datetime.timezone.utc))
     # Different second
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 2, 3, 4, 5, 5, 7, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(1, 2, 3, 4, 5, 5, 7, datetime.timezone.utc))
     # Different micros
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 2, 3, 4, 5, 6, 6, datetime.timezone.utc))
+    ) != check_tokenize(datetime.datetime(1, 2, 3, 4, 5, 6, 6, datetime.timezone.utc))
     # Different tz
-    assert tokenize(
+    assert check_tokenize(
         datetime.datetime(1, 2, 3, 4, 5, 6, 7, datetime.timezone.utc)
-    ) != tokenize(datetime.datetime(1, 2, 3, 4, 5, 6, 7, None))
+    ) != check_tokenize(datetime.datetime(1, 2, 3, 4, 5, 6, 7, None))
 
 
 def test_tokenize_functions_main():
@@ -722,13 +896,13 @@ def test_tokenize_functions_main():
 
 
 def test_normalize_function_limited_size():
+    _clear_function_cache()
     for _ in range(1000):
         normalize_function(lambda x: x)
-
     assert 50 < len(function_cache) < 600
 
 
-def test_normalize_function_dataclass_field_no_repr():
+def test_tokenize_dataclass_field_no_repr():
     A = dataclasses.make_dataclass(
         "A",
         [("param", float, dataclasses.field(repr=False))],
@@ -737,26 +911,24 @@ def test_normalize_function_dataclass_field_no_repr():
 
     a1, a2 = A(1), A(2)
 
-    assert normalize_function(a1) == normalize_function(a1)
-    assert normalize_function(a1) != normalize_function(a2)
+    assert check_tokenize(a1) != check_tokenize(a2)
 
 
 def test_tokenize_operator():
     """Top-level functions in the operator module have a __self__ attribute, which is
     the module itself
     """
-    assert tokenize(operator.add) == tokenize(operator.add)
-    assert tokenize(operator.add) != tokenize(operator.mul)
+    assert check_tokenize(operator.add) != check_tokenize(operator.mul)
 
 
 def test_tokenize_random_state():
     a = random.Random(123)
     b = random.Random(123)
     c = random.Random(456)
-    assert tokenize(a) == tokenize(b)
-    assert tokenize(a) != tokenize(c)
+    assert check_tokenize(a) == check_tokenize(b)
+    assert check_tokenize(a) != check_tokenize(c)
     a.random()
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
 
 @pytest.mark.skipif("not np")
@@ -764,10 +936,10 @@ def test_tokenize_random_state_numpy():
     a = np.random.RandomState(123)
     b = np.random.RandomState(123)
     c = np.random.RandomState(456)
-    assert tokenize(a) == tokenize(b)
-    assert tokenize(a) != tokenize(c)
+    assert check_tokenize(a) == check_tokenize(b)
+    assert check_tokenize(a) != check_tokenize(c)
     a.random()
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) != check_tokenize(b)
 
 
 @pytest.mark.parametrize(
@@ -781,35 +953,31 @@ def test_tokenize_random_functions(module):
 
     a = module.random
     b = pickle.loads(pickle.dumps(a))
-    assert tokenize(a) == tokenize(b)
+    assert check_tokenize(a) == check_tokenize(b)
 
     # Drawing elements or reseeding changes the global state
     a()
     c = pickle.loads(pickle.dumps(a))
-    assert tokenize(a) == tokenize(c)
-    assert tokenize(a) != tokenize(b)
+    assert check_tokenize(a) == check_tokenize(c)
+    assert check_tokenize(a) != check_tokenize(b)
 
     module.seed(123)
     d = pickle.loads(pickle.dumps(a))
-    assert tokenize(a) == tokenize(d)
-    assert tokenize(a) != tokenize(c)
+    assert check_tokenize(a) == check_tokenize(d)
+    assert check_tokenize(a) != check_tokenize(c)
 
 
 def test_tokenize_random_functions_with_state():
     a = random.Random(123).random
-    b = random.Random(123).random
-    c = random.Random(456).random
-    assert tokenize(a) == tokenize(b)
-    assert tokenize(a) != tokenize(c)
+    b = random.Random(456).random
+    assert check_tokenize(a) != check_tokenize(b)
 
 
 @pytest.mark.skipif("not np")
 def test_tokenize_random_functions_with_state_numpy():
     a = np.random.RandomState(123).random
-    b = np.random.RandomState(123).random
-    c = np.random.RandomState(456).random
-    assert tokenize(a) == tokenize(b)
-    assert tokenize(a) != tokenize(c)
+    b = np.random.RandomState(456).random
+    assert check_tokenize(a) != check_tokenize(b)
 
 
 @pytest.mark.skipif("not pa")