diff --git a/newsfragments/1587.misc.rst b/newsfragments/1587.misc.rst
new file mode 100644
index 0000000000..8152340ba2
--- /dev/null
+++ b/newsfragments/1587.misc.rst
@@ -0,0 +1,7 @@
+We refactored `trio.testing.MockClock` so that it no longer needs to
+run an internal task to manage autojumping. This should be mostly
+invisible to users, but there is one semantic change: the interaction
+between `trio.testing.wait_all_tasks_blocked` and the autojump clock
+was fixed. Now, the autojump will always wait until after all
+`~trio.testing.wait_all_tasks_blocked` calls have finished before
+firing, instead of it depending on which threshold values you passed.
diff --git a/trio/_core/__init__.py b/trio/_core/__init__.py
index e22485149d..b2fc7d4300 100644
--- a/trio/_core/__init__.py
+++ b/trio/_core/__init__.py
@@ -71,6 +71,8 @@
 
 from ._thread_cache import start_thread_soon
 
+from ._mock_clock import MockClock
+
 # Kqueue imports
 try:
     from ._run import current_kqueue, monitor_kevent, wait_kevent
diff --git a/trio/testing/_mock_clock.py b/trio/_core/_mock_clock.py
similarity index 56%
rename from trio/testing/_mock_clock.py
rename to trio/_core/_mock_clock.py
index 843f51197f..6d9332fd23 100644
--- a/trio/testing/_mock_clock.py
+++ b/trio/_core/_mock_clock.py
@@ -2,6 +2,7 @@
 from math import inf
 
 from .. import _core
+from ._run import GLOBAL_RUN_CONTEXT
 from .._abc import Clock
 from .._util import SubclassingDeprecatedIn_v0_15_0
 
@@ -51,32 +52,13 @@ class MockClock(Clock, metaclass=SubclassingDeprecatedIn_v0_15_0):
        above) then just set it to zero, and the clock will jump whenever all
        tasks are blocked.
 
-       .. warning::
-
-          If you're using :func:`wait_all_tasks_blocked` and
-          :attr:`autojump_threshold` together, then you have to be
-          careful. Setting :attr:`autojump_threshold` acts like a background
-          task calling::
-
-             while True:
-                 await wait_all_tasks_blocked(
-                   cushion=clock.autojump_threshold, tiebreaker=float("inf"))
-
-          This means that if you call :func:`wait_all_tasks_blocked` with a
-          cushion *larger* than your autojump threshold, then your call to
-          :func:`wait_all_tasks_blocked` will never return, because the
-          autojump task will keep waking up before your task does, and each
-          time it does it'll reset your task's timer. However, if your cushion
-          and the autojump threshold are the *same*, then the autojump's
-          tiebreaker will prevent them from interfering (unless you also set
-          your tiebreaker to infinity for some reason. Don't do that). As an
-          important special case: this means that if you set an autojump
-          threshold of zero and use :func:`wait_all_tasks_blocked` with the
-          default zero cushion, then everything will work fine.
-
-          **Summary**: you should set :attr:`autojump_threshold` to be at
-          least as large as the largest cushion you plan to pass to
-          :func:`wait_all_tasks_blocked`.
+       .. note:: If you use ``autojump_threshold`` and
+          `wait_all_tasks_blocked` at the same time, then you might wonder how
+          they interact, since they both cause things to happen after the run
+          loop goes idle for some time. The answer is:
+          `wait_all_tasks_blocked` takes priority. If there's a task blocked
+          in `wait_all_tasks_blocked`, then the autojump feature treats that
+          as active task and does *not* jump the clock.
 
     """
 
@@ -88,8 +70,6 @@ def __init__(self, rate=0.0, autojump_threshold=inf):
         self._virtual_base = 0.0
         self._rate = 0.0
         self._autojump_threshold = 0.0
-        self._autojump_task = None
-        self._autojump_cancel_scope = None
         # kept as an attribute so that our tests can monkeypatch it
         self._real_clock = time.perf_counter
 
@@ -124,47 +104,31 @@ def autojump_threshold(self):
     @autojump_threshold.setter
     def autojump_threshold(self, new_autojump_threshold):
         self._autojump_threshold = float(new_autojump_threshold)
-        self._maybe_spawn_autojump_task()
-        if self._autojump_cancel_scope is not None:
-            # Task is running and currently blocked on the old setting, wake
-            # it up so it picks up the new setting
-            self._autojump_cancel_scope.cancel()
-
-    async def _autojumper(self):
-        while True:
-            with _core.CancelScope() as cancel_scope:
-                self._autojump_cancel_scope = cancel_scope
-                try:
-                    # If the autojump_threshold changes, then the setter does
-                    # cancel_scope.cancel(), which causes the next line here
-                    # to raise Cancelled, which is absorbed by the cancel
-                    # scope above, and effectively just causes us to skip back
-                    # to the start the loop, like a 'continue'.
-                    await _core.wait_all_tasks_blocked(self._autojump_threshold, inf)
-                    statistics = _core.current_statistics()
-                    jump = statistics.seconds_to_next_deadline
-                    if 0 < jump < inf:
-                        self.jump(jump)
-                    else:
-                        # There are no deadlines, nothing is going to happen
-                        # until some actual I/O arrives (or maybe another
-                        # wait_all_tasks_blocked task wakes up). That's fine,
-                        # but if our threshold is zero then this will become a
-                        # busy-wait -- so insert a small-but-non-zero _sleep to
-                        # avoid that.
-                        if self._autojump_threshold == 0:
-                            await _core.wait_all_tasks_blocked(0.01)
-                finally:
-                    self._autojump_cancel_scope = None
-
-    def _maybe_spawn_autojump_task(self):
-        if self._autojump_threshold < inf and self._autojump_task is None:
-            try:
-                clock = _core.current_clock()
-            except RuntimeError:
-                return
-            if clock is self:
-                self._autojump_task = _core.spawn_system_task(self._autojumper)
+        self._try_resync_autojump_threshold()
+
+    # runner.clock_autojump_threshold is an internal API that isn't easily
+    # usable by custom third-party Clock objects. If you need access to this
+    # functionality, let us know, and we'll figure out how to make a public
+    # API. Discussion:
+    #
+    #     https://github.com/python-trio/trio/issues/1587
+    def _try_resync_autojump_threshold(self):
+        try:
+            runner = GLOBAL_RUN_CONTEXT.runner
+            if runner.is_guest:
+                runner.force_guest_tick_asap()
+        except AttributeError:
+            pass
+        else:
+            runner.clock_autojump_threshold = self._autojump_threshold
+
+    # Invoked by the run loop when runner.clock_autojump_threshold is
+    # exceeded.
+    def _autojump(self):
+        statistics = _core.current_statistics()
+        jump = statistics.seconds_to_next_deadline
+        if 0 < jump < inf:
+            self.jump(jump)
 
     def _real_to_virtual(self, real):
         real_offset = real - self._real_base
@@ -172,8 +136,7 @@ def _real_to_virtual(self, real):
         return self._virtual_base + virtual_offset
 
     def start_clock(self):
-        token = _core.current_trio_token()
-        token.run_sync_soon(self._maybe_spawn_autojump_task)
+        self._try_resync_autojump_threshold()
 
     def current_time(self):
         return self._real_to_virtual(self._real_clock())
diff --git a/trio/_core/_run.py b/trio/_core/_run.py
index 465aebe6f5..8b112051e5 100644
--- a/trio/_core/_run.py
+++ b/trio/_core/_run.py
@@ -12,6 +12,7 @@
 import collections.abc
 from contextlib import contextmanager, closing
 import warnings
+import enum
 
 from contextvars import copy_context
 from math import inf
@@ -112,6 +113,11 @@ def deadline_to_sleep_time(self, deadline):
         return deadline - self.current_time()
 
 
+class IdlePrimedTypes(enum.Enum):
+    WAITING_FOR_IDLE = 1
+    AUTOJUMP_CLOCK = 2
+
+
 ################################################################
 # CancelScope and friends
 ################################################################
@@ -1209,6 +1215,9 @@ class Runner:
     entry_queue = attr.ib(factory=EntryQueue)
     trio_token = attr.ib(default=None)
 
+    # If everything goes idle for this long, we call clock._autojump()
+    clock_autojump_threshold = attr.ib(default=inf)
+
     # Guest mode stuff
     is_guest = attr.ib(default=False)
     guest_tick_scheduled = attr.ib(default=False)
@@ -1999,12 +2008,18 @@ def unrolled_run(runner, async_fn, args, host_uses_signal_set_wakeup_fd=False):
                 timeout = _MAX_TIMEOUT
             timeout = min(max(0, timeout), _MAX_TIMEOUT)
 
-            idle_primed = False
+            idle_primed = None
             if runner.waiting_for_idle:
                 cushion, tiebreaker, _ = runner.waiting_for_idle.keys()[0]
                 if cushion < timeout:
                     timeout = cushion
-                    idle_primed = True
+                    idle_primed = IdlePrimedTypes.WAITING_FOR_IDLE
+            # We use 'elif' here because if there are tasks in
+            # wait_all_tasks_blocked, then those tasks will wake up without
+            # jumping the clock, so we don't need to autojump.
+            elif runner.clock_autojump_threshold < timeout:
+                timeout = runner.clock_autojump_threshold
+                idle_primed = IdlePrimedTypes.AUTOJUMP_CLOCK
 
             if runner.instruments:
                 runner.instrument("before_io_wait", timeout)
@@ -2024,18 +2039,18 @@ def unrolled_run(runner, async_fn, args, host_uses_signal_set_wakeup_fd=False):
                 if deadline <= now:
                     # This removes the given scope from runner.deadlines:
                     cancel_scope.cancel()
-                    idle_primed = False
+                    idle_primed = None
                 else:
                     break
 
-            # idle_primed=True means: if the IO wait hit the timeout, and still
-            # nothing is happening, then we should start waking up
-            # wait_all_tasks_blocked tasks. But there are some subtleties in
-            # defining "nothing is happening".
+            # idle_primed != None means: if the IO wait hit the timeout, and
+            # still nothing is happening, then we should start waking up
+            # wait_all_tasks_blocked tasks or autojump the clock. But there
+            # are some subtleties in defining "nothing is happening".
             #
             # 'not runner.runq' means that no tasks are currently runnable.
             # 'not events' means that the last IO wait call hit its full
-            # timeout. These are very similar, and if idle_primed=True and
+            # timeout. These are very similar, and if idle_primed != None and
             # we're running in regular mode then they always go together. But,
             # in *guest* mode, they can happen independently, even when
             # idle_primed=True:
@@ -2049,14 +2064,18 @@ def unrolled_run(runner, async_fn, args, host_uses_signal_set_wakeup_fd=False):
             #   before we got here.
             #
             # So we need to check both.
-            if idle_primed and not runner.runq and not events:
-                while runner.waiting_for_idle:
-                    key, task = runner.waiting_for_idle.peekitem(0)
-                    if key[:2] == (cushion, tiebreaker):
-                        del runner.waiting_for_idle[key]
-                        runner.reschedule(task)
-                    else:
-                        break
+            if idle_primed is not None and not runner.runq and not events:
+                if idle_primed is IdlePrimedTypes.WAITING_FOR_IDLE:
+                    while runner.waiting_for_idle:
+                        key, task = runner.waiting_for_idle.peekitem(0)
+                        if key[:2] == (cushion, tiebreaker):
+                            del runner.waiting_for_idle[key]
+                            runner.reschedule(task)
+                        else:
+                            break
+                else:
+                    assert idle_primed is IdlePrimedTypes.AUTOJUMP_CLOCK
+                    runner.clock._autojump()
 
             # Process all runnable tasks, but only the ones that are already
             # runnable now. Anything that becomes runnable during this cycle
diff --git a/trio/_core/tests/test_guest_mode.py b/trio/_core/tests/test_guest_mode.py
index 46e741e392..e58ec3ebc2 100644
--- a/trio/_core/tests/test_guest_mode.py
+++ b/trio/_core/tests/test_guest_mode.py
@@ -7,6 +7,7 @@
 import signal
 import socket
 import threading
+import time
 
 import trio
 import trio.testing
@@ -473,3 +474,26 @@ async def trio_main_raising(in_host):
     assert excinfo.value.__context__ is final_exc
 
     assert signal.getsignal(signal.SIGINT) is signal.default_int_handler
+
+
+def test_guest_mode_autojump_clock_threshold_changing():
+    # This is super obscure and probably no-one will ever notice, but
+    # technically mutating the MockClock.autojump_threshold from the host
+    # should wake up the guest, so let's test it.
+
+    clock = trio.testing.MockClock()
+
+    DURATION = 120
+
+    async def trio_main(in_host):
+        assert trio.current_time() == 0
+        in_host(lambda: setattr(clock, "autojump_threshold", 0))
+        await trio.sleep(DURATION)
+        assert trio.current_time() == DURATION
+
+    start = time.monotonic()
+    trivial_guest_run(trio_main, clock=clock)
+    end = time.monotonic()
+    # Should be basically instantaneous, but we'll leave a generous buffer to
+    # account for any CI weirdness
+    assert end - start < DURATION / 2
diff --git a/trio/_core/tests/test_mock_clock.py b/trio/_core/tests/test_mock_clock.py
new file mode 100644
index 0000000000..aef4cef498
--- /dev/null
+++ b/trio/_core/tests/test_mock_clock.py
@@ -0,0 +1,181 @@
+from math import inf
+import time
+
+import pytest
+
+from trio import sleep
+from ... import _core
+from .. import wait_all_tasks_blocked
+from .._mock_clock import MockClock
+from .tutil import slow
+
+
+def test_mock_clock():
+    REAL_NOW = 123.0
+    c = MockClock()
+    c._real_clock = lambda: REAL_NOW
+    repr(c)  # smoke test
+    assert c.rate == 0
+    assert c.current_time() == 0
+    c.jump(1.2)
+    assert c.current_time() == 1.2
+    with pytest.raises(ValueError):
+        c.jump(-1)
+    assert c.current_time() == 1.2
+    assert c.deadline_to_sleep_time(1.1) == 0
+    assert c.deadline_to_sleep_time(1.2) == 0
+    assert c.deadline_to_sleep_time(1.3) > 999999
+
+    with pytest.raises(ValueError):
+        c.rate = -1
+    assert c.rate == 0
+
+    c.rate = 2
+    assert c.current_time() == 1.2
+    REAL_NOW += 1
+    assert c.current_time() == 3.2
+    assert c.deadline_to_sleep_time(3.1) == 0
+    assert c.deadline_to_sleep_time(3.2) == 0
+    assert c.deadline_to_sleep_time(4.2) == 0.5
+
+    c.rate = 0.5
+    assert c.current_time() == 3.2
+    assert c.deadline_to_sleep_time(3.1) == 0
+    assert c.deadline_to_sleep_time(3.2) == 0
+    assert c.deadline_to_sleep_time(4.2) == 2.0
+
+    c.jump(0.8)
+    assert c.current_time() == 4.0
+    REAL_NOW += 1
+    assert c.current_time() == 4.5
+
+    c2 = MockClock(rate=3)
+    assert c2.rate == 3
+    assert c2.current_time() < 10
+
+
+async def test_mock_clock_autojump(mock_clock):
+    assert mock_clock.autojump_threshold == inf
+
+    mock_clock.autojump_threshold = 0
+    assert mock_clock.autojump_threshold == 0
+
+    real_start = time.perf_counter()
+
+    virtual_start = _core.current_time()
+    for i in range(10):
+        print("sleeping {} seconds".format(10 * i))
+        await sleep(10 * i)
+        print("woke up!")
+        assert virtual_start + 10 * i == _core.current_time()
+        virtual_start = _core.current_time()
+
+    real_duration = time.perf_counter() - real_start
+    print("Slept {} seconds in {} seconds".format(10 * sum(range(10)), real_duration))
+    assert real_duration < 1
+
+    mock_clock.autojump_threshold = 0.02
+    t = _core.current_time()
+    # this should wake up before the autojump threshold triggers, so time
+    # shouldn't change
+    await wait_all_tasks_blocked()
+    assert t == _core.current_time()
+    # this should too
+    await wait_all_tasks_blocked(0.01)
+    assert t == _core.current_time()
+
+    # This should wake up at the same time as the autojump_threshold, and
+    # confuse things. There is no deadline, so it shouldn't actually jump
+    # the clock. But does it handle the situation gracefully?
+    await wait_all_tasks_blocked(cushion=0.02, tiebreaker=float("inf"))
+    # And again with threshold=0, because that has some special
+    # busy-wait-avoidance logic:
+    mock_clock.autojump_threshold = 0
+    await wait_all_tasks_blocked(tiebreaker=float("inf"))
+
+    # set up a situation where the autojump task is blocked for a long long
+    # time, to make sure that cancel-and-adjust-threshold logic is working
+    mock_clock.autojump_threshold = 10000
+    await wait_all_tasks_blocked()
+    mock_clock.autojump_threshold = 0
+    # if the above line didn't take affect immediately, then this would be
+    # bad:
+    await sleep(100000)
+
+
+async def test_mock_clock_autojump_interference(mock_clock):
+    mock_clock.autojump_threshold = 0.02
+
+    mock_clock2 = MockClock()
+    # messing with the autojump threshold of a clock that isn't actually
+    # installed in the run loop shouldn't do anything.
+    mock_clock2.autojump_threshold = 0.01
+
+    # if the autojump_threshold of 0.01 were in effect, then the next line
+    # would block forever, as the autojump task kept waking up to try to
+    # jump the clock.
+    await wait_all_tasks_blocked(0.015)
+
+    # but the 0.02 limit does apply
+    await sleep(100000)
+
+
+def test_mock_clock_autojump_preset():
+    # Check that we can set the autojump_threshold before the clock is
+    # actually in use, and it gets picked up
+    mock_clock = MockClock(autojump_threshold=0.1)
+    mock_clock.autojump_threshold = 0.01
+    real_start = time.perf_counter()
+    _core.run(sleep, 10000, clock=mock_clock)
+    assert time.perf_counter() - real_start < 1
+
+
+async def test_mock_clock_autojump_0_and_wait_all_tasks_blocked_0(mock_clock):
+    # Checks that autojump_threshold=0 doesn't interfere with
+    # calling wait_all_tasks_blocked with the default cushion=0 and arbitrary
+    # tiebreakers.
+
+    mock_clock.autojump_threshold = 0
+
+    record = []
+
+    async def sleeper():
+        await sleep(100)
+        record.append("yawn")
+
+    async def waiter():
+        for i in range(10):
+            await wait_all_tasks_blocked(tiebreaker=i)
+            record.append(i)
+        await sleep(1000)
+        record.append("waiter done")
+
+    async with _core.open_nursery() as nursery:
+        nursery.start_soon(sleeper)
+        nursery.start_soon(waiter)
+
+    assert record == list(range(10)) + ["yawn", "waiter done"]
+
+
+@slow
+async def test_mock_clock_autojump_0_and_wait_all_tasks_blocked_nonzero(mock_clock):
+    # Checks that autojump_threshold=0 doesn't interfere with
+    # calling wait_all_tasks_blocked with a non-zero cushion.
+
+    mock_clock.autojump_threshold = 0
+
+    record = []
+
+    async def sleeper():
+        await sleep(100)
+        record.append("yawn")
+
+    async def waiter():
+        await wait_all_tasks_blocked(1)
+        record.append("waiter done")
+
+    async with _core.open_nursery() as nursery:
+        nursery.start_soon(sleeper)
+        nursery.start_soon(waiter)
+
+    assert record == ["waiter done", "yawn"]
diff --git a/trio/testing/__init__.py b/trio/testing/__init__.py
index 8c730ffeb5..aa15c4743e 100644
--- a/trio/testing/__init__.py
+++ b/trio/testing/__init__.py
@@ -1,9 +1,7 @@
-from .._core import wait_all_tasks_blocked
+from .._core import wait_all_tasks_blocked, MockClock
 
 from ._trio_test import trio_test
 
-from ._mock_clock import MockClock
-
 from ._checkpoints import assert_checkpoints, assert_no_checkpoints
 
 from ._sequencer import Sequencer
diff --git a/trio/tests/test_testing.py b/trio/tests/test_testing.py
index 460200b28c..304f18cb88 100644
--- a/trio/tests/test_testing.py
+++ b/trio/tests/test_testing.py
@@ -1,7 +1,5 @@
 # XX this should get broken up, like testing.py did
 
-import time
-from math import inf
 import tempfile
 
 import pytest
@@ -260,156 +258,6 @@ async def child(i):
 ################################################################
 
 
-def test_mock_clock():
-    REAL_NOW = 123.0
-    c = MockClock()
-    c._real_clock = lambda: REAL_NOW
-    repr(c)  # smoke test
-    assert c.rate == 0
-    assert c.current_time() == 0
-    c.jump(1.2)
-    assert c.current_time() == 1.2
-    with pytest.raises(ValueError):
-        c.jump(-1)
-    assert c.current_time() == 1.2
-    assert c.deadline_to_sleep_time(1.1) == 0
-    assert c.deadline_to_sleep_time(1.2) == 0
-    assert c.deadline_to_sleep_time(1.3) > 999999
-
-    with pytest.raises(ValueError):
-        c.rate = -1
-    assert c.rate == 0
-
-    c.rate = 2
-    assert c.current_time() == 1.2
-    REAL_NOW += 1
-    assert c.current_time() == 3.2
-    assert c.deadline_to_sleep_time(3.1) == 0
-    assert c.deadline_to_sleep_time(3.2) == 0
-    assert c.deadline_to_sleep_time(4.2) == 0.5
-
-    c.rate = 0.5
-    assert c.current_time() == 3.2
-    assert c.deadline_to_sleep_time(3.1) == 0
-    assert c.deadline_to_sleep_time(3.2) == 0
-    assert c.deadline_to_sleep_time(4.2) == 2.0
-
-    c.jump(0.8)
-    assert c.current_time() == 4.0
-    REAL_NOW += 1
-    assert c.current_time() == 4.5
-
-    c2 = MockClock(rate=3)
-    assert c2.rate == 3
-    assert c2.current_time() < 10
-
-
-async def test_mock_clock_autojump(mock_clock):
-    assert mock_clock.autojump_threshold == inf
-
-    mock_clock.autojump_threshold = 0
-    assert mock_clock.autojump_threshold == 0
-
-    real_start = time.perf_counter()
-
-    virtual_start = _core.current_time()
-    for i in range(10):
-        print("sleeping {} seconds".format(10 * i))
-        await sleep(10 * i)
-        print("woke up!")
-        assert virtual_start + 10 * i == _core.current_time()
-        virtual_start = _core.current_time()
-
-    real_duration = time.perf_counter() - real_start
-    print("Slept {} seconds in {} seconds".format(10 * sum(range(10)), real_duration))
-    assert real_duration < 1
-
-    mock_clock.autojump_threshold = 0.02
-    t = _core.current_time()
-    # this should wake up before the autojump threshold triggers, so time
-    # shouldn't change
-    await wait_all_tasks_blocked()
-    assert t == _core.current_time()
-    # this should too
-    await wait_all_tasks_blocked(0.01)
-    assert t == _core.current_time()
-
-    # This should wake up at the same time as the autojump_threshold, and
-    # confuse things. There is no deadline, so it shouldn't actually jump
-    # the clock. But does it handle the situation gracefully?
-    await wait_all_tasks_blocked(cushion=0.02, tiebreaker=float("inf"))
-    # And again with threshold=0, because that has some special
-    # busy-wait-avoidance logic:
-    mock_clock.autojump_threshold = 0
-    await wait_all_tasks_blocked(tiebreaker=float("inf"))
-
-    # set up a situation where the autojump task is blocked for a long long
-    # time, to make sure that cancel-and-adjust-threshold logic is working
-    mock_clock.autojump_threshold = 10000
-    await wait_all_tasks_blocked()
-    mock_clock.autojump_threshold = 0
-    # if the above line didn't take affect immediately, then this would be
-    # bad:
-    await sleep(100000)
-
-
-async def test_mock_clock_autojump_interference(mock_clock):
-    mock_clock.autojump_threshold = 0.02
-
-    mock_clock2 = MockClock()
-    # messing with the autojump threshold of a clock that isn't actually
-    # installed in the run loop shouldn't do anything.
-    mock_clock2.autojump_threshold = 0.01
-
-    # if the autojump_threshold of 0.01 were in effect, then the next line
-    # would block forever, as the autojump task kept waking up to try to
-    # jump the clock.
-    await wait_all_tasks_blocked(0.015)
-
-    # but the 0.02 limit does apply
-    await sleep(100000)
-
-
-def test_mock_clock_autojump_preset():
-    # Check that we can set the autojump_threshold before the clock is
-    # actually in use, and it gets picked up
-    mock_clock = MockClock(autojump_threshold=0.1)
-    mock_clock.autojump_threshold = 0.01
-    real_start = time.perf_counter()
-    _core.run(sleep, 10000, clock=mock_clock)
-    assert time.perf_counter() - real_start < 1
-
-
-async def test_mock_clock_autojump_0_and_wait_all_tasks_blocked(mock_clock):
-    # Checks that autojump_threshold=0 doesn't interfere with
-    # calling wait_all_tasks_blocked with the default cushion=0 and arbitrary
-    # tiebreakers.
-
-    mock_clock.autojump_threshold = 0
-
-    record = []
-
-    async def sleeper():
-        await sleep(100)
-        record.append("yawn")
-
-    async def waiter():
-        for i in range(10):
-            await wait_all_tasks_blocked(tiebreaker=i)
-            record.append(i)
-        await sleep(1000)
-        record.append("waiter done")
-
-    async with _core.open_nursery() as nursery:
-        nursery.start_soon(sleeper)
-        nursery.start_soon(waiter)
-
-    assert record == list(range(10)) + ["yawn", "waiter done"]
-
-
-################################################################
-
-
 async def test__assert_raises():
     with pytest.raises(AssertionError):
         with _assert_raises(RuntimeError):