Remove @avoid_ci from test_steal (#6872)

Author: crusaderky

distributed/tests/test_active_memory_manager.py

@@ -1085,6 +1085,21 @@ async def tensordot_stress(c):
     assert await c.compute(b) == 2134.398
 
 
+@pytest.mark.slow
+@pytest.mark.avoid_ci(reason="distributed#5371")
+@gen_cluster(
+    client=True,
+    nthreads=[("", 1)] * 4,
+    Worker=Nanny,
+    config=NO_AMM_START,
+)
+async def test_noamm_stress(c, s, *nannies):
+    """Test the tensordot_stress helper without AMM. This is to figure out if a
+    stability issue is AMM-specific or not.
+    """
+    await tensordot_stress(c)
+
+
 @pytest.mark.slow
 @pytest.mark.avoid_ci(reason="distributed#5371")
 @gen_cluster(
@@ -1148,6 +1163,8 @@ async def test_ReduceReplicas_stress(c, s, *nannies):
             {"class": "distributed.active_memory_manager.ReduceReplicas"},
         ],
     },
+    scheduler_kwargs={"transition_counter_max": 500_000},
+    worker_kwargs={"transition_counter_max": 500_000},
 )
 async def test_RetireWorker_stress(c, s, *nannies, use_ReduceReplicas):
     """It is safe to retire the best part of a cluster in the middle of a computation"""

distributed/tests/test_steal.py

@@ -85,18 +85,18 @@ async def test_steal_cheap_data_slow_computation(c, s, a, b):
     assert abs(len(a.data) - len(b.data)) <= 5
 
 
-@pytest.mark.avoid_ci
-@gen_cluster(client=True, nthreads=[("127.0.0.1", 1)] * 2)
+@pytest.mark.slow
+@gen_cluster(client=True, nthreads=[("", 1)] * 2)
 async def test_steal_expensive_data_slow_computation(c, s, a, b):
     np = pytest.importorskip("numpy")
 
-    x = c.submit(slowinc, 100, delay=0.2, workers=a.address)
+    x = c.submit(slowinc, 1, delay=0.2, workers=a.address)
     await wait(x)  # learn that slowinc is slow
 
-    x = c.submit(np.arange, 1000000, workers=a.address)  # put expensive data
+    x = c.submit(np.arange, 1_000_000, workers=a.address)  # put expensive data
     await wait(x)
 
-    slow = [c.submit(slowinc, x, delay=0.1, pure=False) for i in range(20)]
+    slow = [c.submit(slowinc, x, delay=0.1, pure=False) for _ in range(20)]
     await wait(slow)
     assert len(s.tasks[x.key].who_has) > 1