Reintroduce weight function (#4723)

* Reintroduce weight function

* Remove newline

* Add deprecation cycle

distributed/tests/test_utils.py

@@ -24,6 +24,7 @@
     LoopRunner,
     TimeoutError,
     _maybe_complex,
+    deprecated,
     ensure_bytes,
     ensure_ip,
     format_dashboard_link,
@@ -617,3 +618,20 @@ def test_lru():
 async def test_offload():
     assert (await offload(inc, 1)) == 2
     assert (await offload(lambda x, y: x + y, 1, y=2)) == 3
+
+
+def test_deprecated():
+    @deprecated()
+    def foo():
+        return "bar"
+
+    with pytest.warns(DeprecationWarning, match="foo is deprecated"):
+        assert foo() == "bar"
+
+    # Explicit version specified
+    @deprecated(version_removed="1.2.3")
+    def foo():
+        return "bar"
+
+    with pytest.warns(DeprecationWarning, match="removed in version 1.2.3"):
+        assert foo() == "bar"

distributed/tests/test_worker.py

@@ -54,7 +54,7 @@
     s,
     slowinc,
 )
-from distributed.worker import Worker, error_message, logger, parse_memory_limit
+from distributed.worker import Worker, error_message, logger, parse_memory_limit, weight
 
 
 @pytest.mark.asyncio
@@ -1787,3 +1787,8 @@ async def test_story(c, s, w):
     ts = w.tasks[future.key]
     assert ts.state in str(w.story(ts))
     assert w.story(ts) == w.story(ts.key)
+
+
+def test_weight_deprecated():
+    with pytest.warns(DeprecationWarning):
+        weight("foo", "bar")

distributed/utils.py

@@ -1073,6 +1073,34 @@ def time_warn(duration, text):
         print("TIME WARNING", text, end - start)
 
 
+def deprecated(*, version_removed: str = None):
+    """Decorator to mark a function as deprecated
+
+    Parameters
+    ----------
+    version_removed : str, optional
+        If specified, include the version in which the deprecated function
+        will be removed. Defaults to "a future release".
+    """
+
+    def decorator(func):
+        nonlocal version_removed
+        msg = f"{funcname(func)} is deprecated and will be removed in"
+        if version_removed is not None:
+            msg += f" version {version_removed}"
+        else:
+            msg += " a future release"
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            warnings.warn(msg, DeprecationWarning, stacklevel=2)
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator
+
+
 def json_load_robust(fn, load=json.load):
     """ Reads a JSON file from disk that may be being written as we read """
     while not os.path.exists(fn):

distributed/worker.py

@@ -56,6 +56,7 @@
     LRU,
     TimeoutError,
     _maybe_complex,
+    deprecated,
     get_ip,
     has_arg,
     import_file,
@@ -3720,6 +3721,11 @@ def convert_kwargs_to_str(kwargs, max_len=None):
         return "{{{}}}".format(", ".join(strs))
 
 
+@deprecated(version_removed="2021.06.0")
+def weight(k, v):
+    return sizeof(v)
+
+
 async def run(server, comm, function, args=(), kwargs=None, is_coro=None, wait=True):
     kwargs = kwargs or {}
     function = pickle.loads(function)