Repository: dask/distributed

Issue:
Spill to constrained disk space
# Use case
This has been raised offline by a power user.
Their workers have limited disk space - frequently less than the amount of RAM. At the moment, the user has completely disabled spilling as the spill file will occupy all available space and when that happens OSErrors will start being raised and data will be lost.

# Proposed behaviour

- If spill-to-disk limit is hit, stop spilling to disk and keep data in memory.
- Buildup of memory will then pause worker


# Proposed design
1. Add a line to the dask config to put an upper limit to the size of the spill file.
2. Keep track of the current size of the spill files on disk. This may not be exactly the same as the size of the keys due to discrepancies between sizeof() and pickle output. Note that this measure can be done without I/O by intercepting the calls to ``zict.File.__setitem__``.
3. Ahead of spilling, add the ``sizeof()`` of the key to be spilled to the current size of the spill files on disk. If the maximum size would be exceeded, log a warning and don't spill. If memory pressure keeps building up, this will in turn cause the worker to eventually reach the pause threshold.

Discussion:
github-actions[bot]: ## Unit Test Results
       18 files  +       1         18 suites  +1   10h 39m 33s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "duration of all tests") + 37m 15s
  2 602 tests +       6    2 515 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests")  -        1       78 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests")  -     2    8 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") +  8  1 [:fire:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "test errors") +1 
23 284 runs  +1 420  21 722 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") +1 270  1 517 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") +105  44 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") +44  1 [:fire:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "test errors") +1 

For more details on these failures and errors, see [this check](https://github.com/dask/distributed/runs/5186444273).

Results for commit 18898106. ± Comparison against base commit 91f899e7.

ncclementi: @crusaderky With the changes on this PR, Does this documentation need extra information?
https://github.com/dask/distributed/blob/main/docs/source/worker.rst#using-the-dashboard-to-monitor-memory-usage
crusaderky: > @crusaderky With the changes on this PR, Does this documentation need extra information? https://github.com/dask/distributed/blob/main/docs/source/worker.rst#using-the-dashboard-to-monitor-memory-usage

updated
crusaderky: > @crusaderky it looks like we are having a related failure in a `test_scheduler.py::test_memory` see https://github.com/dask/distributed/runs/5186888061?check_suite_focus=true#step:12:1260

This is... a major pain. The problem is that Python 3.7/3.8 CI does installs neither lz4 nor snappy, so ``distributed.comm.compression`` falls back to no compression. However, I can't force it to zlib either (which is always available, because of a design flaw in ``distributed.protocol.compression``. Which is not trivial to fix. I'll look into it more but this looks like yet another blocker issue.
crusaderky: All green. Ready for review (but not merge!)
ncclementi: I think we are missing renaming some things in the dashboard to be consistent with the changes done in this PR, unless we think it is clear enough that "managed" is "managed_in_memory".

- The [WorkerTable](https://github.com/dask/distributed/blob/922d987261637152636087d93bf5ca333e8faefb/distributed/dashboard/components/scheduler.py#L3021)  it uses the word "memory_managed" and there is an assignment of `"memory_managed": "managed",` if I'm not wrong this should be "memory_managed": "managed in memory",

- The [ClusterMemory](https://github.com/dask/distributed/blob/922d987261637152636087d93bf5ca333e8faefb/distributed/dashboard/components/scheduler.py#L237) and [WorkersMemory](https://github.com/dask/distributed/blob/922d987261637152636087d93bf5ca333e8faefb/distributed/dashboard/components/scheduler.py#L352) uses as name also "managed", 
crusaderky: > * "memory_managed": "managed"



> I think we are missing renaming some things in the dashboard to be consistent with the changes done in this PR, unless we think it is clear enough that "managed" is "managed_in_memory".
> 
> * The [WorkerTable](https://github.com/dask/distributed/blob/922d987261637152636087d93bf5ca333e8faefb/distributed/dashboard/components/scheduler.py#L3021)  it uses the word "memory_managed" and there is an assignment of `"memory_managed": "managed"` if I'm not wrong this should be `"memory_managed": "managed in memory"`,
> * The [ClusterMemory](https://github.com/dask/distributed/blob/922d987261637152636087d93bf5ca333e8faefb/distributed/dashboard/components/scheduler.py#L237) and [WorkersMemory](https://github.com/dask/distributed/blob/922d987261637152636087d93bf5ca333e8faefb/distributed/dashboard/components/scheduler.py#L352) use also as name  "managed",

I've renamed the variables for clarity. Note that the measure being shown was already managed_in_memory. The choice to just write "managed" in the UI is a conscious one to avoid confusing users.
quasiben: @shwina given you recent work with visualizing RMM data, this PR may be of interest as it illuminates issues with how spilled data is visualized

Patch:
diff --git a/distributed/dashboard/components/scheduler.py b/distributed/dashboard/components/scheduler.py
index d8eab514..1dc529b1 100644
--- a/distributed/dashboard/components/scheduler.py
+++ b/distributed/dashboard/components/scheduler.py
@@ -3044,7 +3044,7 @@ class WorkerTable(DashboardComponent):
             "memory",
             "memory_limit",
             "memory_percent",
-            "memory_managed",
+            "memory_managed_in_memory",
             "memory_unmanaged_old",
             "memory_unmanaged_recent",
             "memory_spilled",
@@ -3072,7 +3072,7 @@ class WorkerTable(DashboardComponent):
             "memory",
             "memory_limit",
             "memory_percent",
-            "memory_managed",
+            "memory_managed_in_memory",
             "memory_unmanaged_old",
             "memory_unmanaged_recent",
             "memory_spilled",
@@ -3083,7 +3083,7 @@ class WorkerTable(DashboardComponent):
         column_title_renames = {
             "memory_limit": "limit",
             "memory_percent": "memory %",
-            "memory_managed": "managed",
+            "memory_managed_in_memory": "managed",
             "memory_unmanaged_old": "unmanaged old",
             "memory_unmanaged_recent": "unmanaged recent",
             "memory_spilled": "spilled",
@@ -3104,7 +3104,7 @@ class WorkerTable(DashboardComponent):
             "memory_percent": NumberFormatter(format="0.0 %"),
             "memory": NumberFormatter(format="0.0 b"),
             "memory_limit": NumberFormatter(format="0.0 b"),
-            "memory_managed": NumberFormatter(format="0.0 b"),
+            "memory_managed_in_memory": NumberFormatter(format="0.0 b"),
             "memory_unmanaged_old": NumberFormatter(format="0.0 b"),
             "memory_unmanaged_recent": NumberFormatter(format="0.0 b"),
             "memory_spilled": NumberFormatter(format="0.0 b"),
@@ -3231,7 +3231,7 @@ class WorkerTable(DashboardComponent):
             else:
                 data["memory_percent"][-1] = ""
             data["memory_limit"][-1] = ws.memory_limit
-            data["memory_managed"][-1] = minfo.managed_in_memory
+            data["memory_managed_in_memory"][-1] = minfo.managed_in_memory
             data["memory_unmanaged_old"][-1] = minfo.unmanaged_old
             data["memory_unmanaged_recent"][-1] = minfo.unmanaged_recent
             data["memory_unmanaged_recent"][-1] = minfo.unmanaged_recent
diff --git a/distributed/scheduler.py b/distributed/scheduler.py
index 6f259e59..231c4ef7 100644
--- a/distributed/scheduler.py
+++ b/distributed/scheduler.py
@@ -295,13 +295,16 @@ class MemoryState:
     """Memory readings on a worker or on the whole cluster.
 
     managed
-        Sum of the output of sizeof() for all dask keys held by the worker, both in
-        memory and spilled to disk
+        Sum of the output of sizeof() for all dask keys held by the worker in memory,
+        plus number of bytes spilled to disk
     managed_in_memory
-        Sum of the output of sizeof() for the dask keys held in RAM
+        Sum of the output of sizeof() for the dask keys held in RAM. Note that this may
+        be inaccurate, which may cause inaccurate unmanaged memory (see below).
     managed_spilled
-        Sum of the output of sizeof() for the dask keys spilled to the hard drive.
-        Note that this is the size in memory; serialized size may be different.
+        Number of bytes  for the dask keys spilled to the hard drive.
+        Note that this is the size on disk; size in memory may be different due to
+        compression and inaccuracies in sizeof(). In other words, given the same keys,
+        'managed' will change depending if the keys are in memory or spilled.
     process
         Total RSS memory measured by the OS on the worker process.
         This is always exactly equal to managed_in_memory + unmanaged.
@@ -341,7 +344,7 @@ class MemoryState:
         *,
         process: Py_ssize_t,
         unmanaged_old: Py_ssize_t,
-        managed: Py_ssize_t,
+        managed_in_memory: Py_ssize_t,
         managed_spilled: Py_ssize_t,
     ):
         # Some data arrives with the heartbeat, some other arrives in realtime as the
@@ -349,9 +352,9 @@ class MemoryState:
         # This can cause glitches where a partial measure is larger than the whole, so
         # we need to force all numbers to add up exactly by definition.
         self._process = process
-        self._managed_spilled = min(managed_spilled, managed)
+        self._managed_in_memory = min(self._process, managed_in_memory)
+        self._managed_spilled = managed_spilled
         # Subtractions between unsigned ints guaranteed by construction to be >= 0
-        self._managed_in_memory = min(managed - self._managed_spilled, process)
         self._unmanaged_old = min(unmanaged_old, process - self._managed_in_memory)
 
     @property
@@ -372,14 +375,22 @@ class MemoryState:
 
     @classmethod
     def sum(cls, *infos: "MemoryState") -> "MemoryState":
-        out = MemoryState(process=0, unmanaged_old=0, managed=0, managed_spilled=0)
+        process = 0
+        unmanaged_old = 0
+        managed_in_memory = 0
+        managed_spilled = 0
         ms: MemoryState
         for ms in infos:
-            out._process += ms._process
-            out._managed_spilled += ms._managed_spilled
-            out._managed_in_memory += ms._managed_in_memory
-            out._unmanaged_old += ms._unmanaged_old
-        return out
+            process += ms._process
+            unmanaged_old += ms._unmanaged_old
+            managed_spilled += ms._managed_spilled
+            managed_in_memory += ms._managed_in_memory
+        return MemoryState(
+            process=process,
+            unmanaged_old=unmanaged_old,
+            managed_in_memory=managed_in_memory,
+            managed_spilled=managed_spilled,
+        )
 
     @property
     def managed(self) -> Py_ssize_t:
@@ -684,8 +695,11 @@ class WorkerState:
             # metrics["memory"] is None if the worker sent a heartbeat before its
             # SystemMonitor ever had a chance to run
             process=self._metrics["memory"] or 0,
-            managed=self._nbytes,
-            managed_spilled=self._metrics["spilled_nbytes"],
+            # self._nbytes is instantaneous; metrics may lag behind by a heartbeat
+            managed_in_memory=max(
+                0, self._nbytes - self._metrics["spilled_nbytes"]["memory"]
+            ),
+            managed_spilled=self._metrics["spilled_nbytes"]["disk"],
             unmanaged_old=self._memory_unmanaged_old,
         )
 
@@ -4330,7 +4344,10 @@ class Scheduler(SchedulerState, ServerNode):
         # SystemMonitor ever had a chance to run.
         # ws._nbytes is updated at a different time and sizeof() may not be accurate,
         # so size may be (temporarily) negative; floor it to zero.
-        size = max(0, (metrics["memory"] or 0) - ws._nbytes + metrics["spilled_nbytes"])
+        size = max(
+            0,
+            (metrics["memory"] or 0) - ws._nbytes + metrics["spilled_nbytes"]["memory"],
+        )
 
         ws._memory_other_history.append((local_now, size))
         if not memory_unmanaged_old:
diff --git a/distributed/spill.py b/distributed/spill.py
index c44dfdf1..12df540d 100644
--- a/distributed/spill.py
+++ b/distributed/spill.py
@@ -5,7 +5,7 @@ import time
 from collections.abc import Mapping
 from contextlib import contextmanager
 from functools import partial
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, NamedTuple
 
 import zict
 from packaging.version import parse as parse_version
@@ -20,6 +20,21 @@ logger = logging.getLogger(__name__)
 has_zict_210 = parse_version(zict.__version__) > parse_version("2.0.0")
 
 
+class SpilledSize(NamedTuple):
+    """Size of a key/value pair when spilled to disk, in bytes"""
+
+    # output of sizeof()
+    memory: int
+    # pickled size
+    disk: int
+
+    def __add__(self, other: SpilledSize) -> SpilledSize:  # type: ignore
+        return SpilledSize(self.memory + other.memory, self.disk + other.disk)
+
+    def __sub__(self, other: SpilledSize) -> SpilledSize:  # type: ignore
+        return SpilledSize(self.memory - other.memory, self.disk - other.disk)
+
+
 class SpillBuffer(zict.Buffer):
     """MutableMapping that automatically spills out dask key/value pairs to disk when
     the total size of the stored data exceeds the target. If max_spill is provided the
@@ -182,9 +197,14 @@ class SpillBuffer(zict.Buffer):
         return self.slow
 
     @property
-    def spilled_total(self) -> int:
-        """Number of bytes spilled to disk.
-        Note that this is the pickled size, which may differ from the output of sizeof().
+    def spilled_total(self) -> SpilledSize:
+        """Number of bytes spilled to disk. Tuple of
+
+        - output of sizeof()
+        - pickled size
+
+        The two may differ substantially, e.g. if sizeof() is inaccurate or in case of
+        compression.
         """
         return self.slow.total_weight
 
@@ -208,8 +228,8 @@ class HandledError(Exception):
 
 class Slow(zict.Func):
     max_weight: int | Literal[False]
-    weight_by_key: dict[str, int]
-    total_weight: int
+    weight_by_key: dict[str, SpilledSize]
+    total_weight: SpilledSize
 
     def __init__(self, spill_directory: str, max_weight: int | Literal[False] = False):
         super().__init__(
@@ -219,7 +239,7 @@ class Slow(zict.Func):
         )
         self.max_weight = max_weight
         self.weight_by_key = {}
-        self.total_weight = 0
+        self.total_weight = SpilledSize(0, 0)
 
     def __setitem__(self, key: str, value: Any) -> None:
         try:
@@ -239,11 +259,11 @@ class Slow(zict.Func):
             assert key not in self.weight_by_key
         else:
             self.d.pop(key, None)
-            self.total_weight -= self.weight_by_key.pop(key, 0)
+            self.total_weight -= self.weight_by_key.pop(key, SpilledSize(0, 0))
 
         if (
             self.max_weight is not False
-            and self.total_weight + pickled_size > self.max_weight
+            and self.total_weight.disk + pickled_size > self.max_weight
         ):
             # Stop callbacks and ensure that the key ends up in SpillBuffer.fast
             # To be caught by SpillBuffer.__setitem__
@@ -253,8 +273,9 @@ class Slow(zict.Func):
         # This may raise OSError, which is caught by SpillBuffer above.
         self.d[key] = pickled
 
-        self.weight_by_key[key] = pickled_size
-        self.total_weight += pickled_size
+        weight = SpilledSize(safe_sizeof(value), pickled_size)
+        self.weight_by_key[key] = weight
+        self.total_weight += weight
 
     def __delitem__(self, key: str) -> None:
         super().__delitem__(key)
diff --git a/distributed/widgets/templates/scheduler_info.html.j2 b/distributed/widgets/templates/scheduler_info.html.j2
index 68036bca..0362bd47 100644
--- a/distributed/widgets/templates/scheduler_info.html.j2
+++ b/distributed/widgets/templates/scheduler_info.html.j2
@@ -116,7 +116,7 @@
                             <strong>Memory usage: </strong> {{ worker["metrics"]["memory"] | format_bytes}}
                         </td>
                         <td style="text-align: left;">
-                            <strong>Spilled bytes: </strong> {{ worker["metrics"]["spilled_nbytes"] | format_bytes }}
+                            <strong>Spilled bytes: </strong> {{ worker["metrics"]["spilled_nbytes"]["disk"] | format_bytes }}
                         </td>
                     </tr>
                     <tr>
diff --git a/distributed/worker.py b/distributed/worker.py
index fa15b20f..92f34fff 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -1159,7 +1159,13 @@ class Worker(ServerNode):
         elif self._status != Status.closed:
             self.loop.call_later(0.05, self._send_worker_status_change)
 
-    async def get_metrics(self):
+    async def get_metrics(self) -> dict:
+        try:
+            spilled_memory, spilled_disk = self.data.spilled_total  # type: ignore
+        except AttributeError:
+            # spilling is disabled
+            spilled_memory, spilled_disk = 0, 0
+
         out = dict(
             executing=self.executing_count,
             in_memory=len(self.data),
@@ -1170,7 +1176,10 @@ class Worker(ServerNode):
                 "workers": dict(self.bandwidth_workers),
                 "types": keymap(typename, self.bandwidth_types),
             },
-            spilled_nbytes=getattr(self.data, "spilled_total", 0),
+            spilled_nbytes={
+                "memory": spilled_memory,
+                "disk": spilled_disk,
+            },
         )
         out.update(self.monitor.recent())
 
diff --git a/docs/source/worker.rst b/docs/source/worker.rst
index bd6fe2d5..49586535 100644
--- a/docs/source/worker.rst
+++ b/docs/source/worker.rst
@@ -285,7 +285,8 @@ unmanaged recent
 
 spilled
     managed memory that has been spilled to disk. This is not included in the 'managed'
-    measure above.
+    measure above. This measure reports the number of bytes actually spilled to disk,
+    which may differ from the output of ``sizeof`` particularly in case of compression.
 
 The sum of managed + unmanaged + unmanaged recent is equal by definition to the process
 memory.


Test patch:
diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py
index 8a948821..93cacaa6 100644
--- a/distributed/tests/test_scheduler.py
+++ b/distributed/tests/test_scheduler.py
@@ -2406,7 +2406,12 @@ async def test_quiet_cluster_round_robin(c, s, a, b):
 
 
 def test_memorystate():
-    m = MemoryState(process=100, unmanaged_old=15, managed=80, managed_spilled=12)
+    m = MemoryState(
+        process=100,
+        unmanaged_old=15,
+        managed_in_memory=68,
+        managed_spilled=12,
+    )
     assert m.process == 100
     assert m.managed == 80
     assert m.managed_in_memory == 68
@@ -2431,8 +2436,18 @@ def test_memorystate():
 
 
 def test_memorystate_sum():
-    m1 = MemoryState(process=100, unmanaged_old=15, managed=80, managed_spilled=12)
-    m2 = MemoryState(process=80, unmanaged_old=10, managed=60, managed_spilled=2)
+    m1 = MemoryState(
+        process=100,
+        unmanaged_old=15,
+        managed_in_memory=68,
+        managed_spilled=12,
+    )
+    m2 = MemoryState(
+        process=80,
+        unmanaged_old=10,
+        managed_in_memory=58,
+        managed_spilled=2,
+    )
     m3 = MemoryState.sum(m1, m2)
     assert m3.process == 180
     assert m3.unmanaged_old == 25
@@ -2441,14 +2456,17 @@ def test_memorystate_sum():
 
 
 @pytest.mark.parametrize(
-    "process,unmanaged_old,managed,managed_spilled", list(product(*[[0, 1, 2, 3]] * 4))
+    "process,unmanaged_old,managed_in_memory,managed_spilled",
+    list(product(*[[0, 1, 2, 3]] * 4)),
 )
-def test_memorystate_adds_up(process, unmanaged_old, managed, managed_spilled):
+def test_memorystate_adds_up(
+    process, unmanaged_old, managed_in_memory, managed_spilled
+):
     """Input data is massaged by __init__ so that everything adds up by construction"""
     m = MemoryState(
         process=process,
         unmanaged_old=unmanaged_old,
-        managed=managed,
+        managed_in_memory=managed_in_memory,
         managed_spilled=managed_spilled,
     )
     assert m.managed_in_memory + m.unmanaged == m.process
@@ -2572,12 +2590,28 @@ async def test_memory(c, s, *nannies):
         ]
     )
 
-    # Timeout needs to be enough to spill 100 MiB to disk
-    await asyncio.gather(
-        assert_memory(a, "managed_spilled", 50, 51, timeout=10),
-        assert_memory(b, "managed_spilled", 50, 51, timeout=10),
-        assert_memory(s, "managed_spilled", 100, 101, timeout=10.1),
-    )
+    # dask serialization compresses ("x" * 50 * 2**20) from 50 MiB to ~200 kiB.
+    # Test that managed_spilled reports the actual size on disk and not the output of
+    # sizeof().
+    # FIXME https://github.com/dask/distributed/issues/5807
+    #       This would be more robust if we could just enable zlib compression in
+    #       @gen_cluster
+    from distributed.protocol.compression import default_compression
+
+    if default_compression:
+        await asyncio.gather(
+            assert_memory(a, "managed_spilled", 0.1, 0.5, timeout=3),
+            assert_memory(b, "managed_spilled", 0.1, 0.5, timeout=3),
+            assert_memory(s, "managed_spilled", 0.2, 1.0, timeout=3.1),
+        )
+    else:
+        # Long timeout to allow spilling 100 MiB to disk
+        await asyncio.gather(
+            assert_memory(a, "managed_spilled", 50, 51, timeout=10),
+            assert_memory(b, "managed_spilled", 50, 51, timeout=10),
+            assert_memory(s, "managed_spilled", 100, 102, timeout=10.1),
+        )
+
     # FIXME on Windows and MacOS we occasionally observe managed_in_memory = 49 bytes
     await asyncio.gather(
         assert_memory(a, "managed_in_memory", 0, 0.1, timeout=0),
diff --git a/distributed/tests/test_spill.py b/distributed/tests/test_spill.py
index 713393cb..c30aa6ce 100644
--- a/distributed/tests/test_spill.py
+++ b/distributed/tests/test_spill.py
@@ -16,8 +16,11 @@ from distributed.spill import SpillBuffer
 from distributed.utils_test import captured_logger
 
 
-def psize(*objs) -> int:
-    return sum(len(frame) for obj in objs for frame in serialize_bytelist(obj))
+def psize(*objs) -> tuple[int, int]:
+    return (
+        sum(sizeof(o) for o in objs),
+        sum(len(frame) for obj in objs for frame in serialize_bytelist(obj)),
+    )
 
 
 def test_spillbuffer(tmpdir):
@@ -27,27 +30,27 @@ def test_spillbuffer(tmpdir):
     assert buf.disk is buf.slow
 
     assert not buf.slow.weight_by_key
-    assert buf.slow.total_weight == 0
-    assert buf.spilled_total == 0
+    assert buf.slow.total_weight == (0, 0)
+    assert buf.spilled_total == (0, 0)
 
     a, b, c, d = "a" * 100, "b" * 99, "c" * 98, "d" * 97
 
     # Test assumption made by this test, mostly for non CPython implementations
     assert 100 < sizeof(a) < 200
-    assert sizeof(a) != psize(a)
+    assert psize(a)[0] != psize(a)[1]
 
     buf["a"] = a
     assert not buf.slow
     assert buf.fast.weights == {"a": sizeof(a)}
     assert buf.fast.total_weight == sizeof(a)
     assert buf.slow.weight_by_key == {}
-    assert buf.slow.total_weight == 0
+    assert buf.slow.total_weight == (0, 0)
     assert buf["a"] == a
 
     buf["b"] = b
     assert not buf.slow
     assert not buf.slow.weight_by_key
-    assert buf.slow.total_weight == 0
+    assert buf.slow.total_weight == (0, 0)
 
     buf["c"] = c
     assert set(buf.slow) == {"a"}
@@ -182,7 +185,7 @@ def test_spillbuffer_maxlim(tmpdir):
     # max_spill
 
     a_large = "a" * 500
-    assert psize(a_large) > 600  # size of max_spill
+    assert psize(a_large)[1] > 600  # size of max_spill
 
     with captured_logger(logging.getLogger("distributed.spill")) as logs_alarge:
         buf["a"] = a_large