Make test_session_set_progress more stable under Ray tests (#3103)

Author: zhongchun

mars/deploy/oscar/tests/test_local.py

@@ -608,14 +608,15 @@ async def test_session_get_progress(create_cluster):
     assert session.session_id is not None
 
     raw = np.random.rand(100, 4)
-    t = mt.tensor(raw, chunk_size=5)
+    t = mt.tensor(raw, chunk_size=50)
 
     def f1(c):
         time.sleep(0.5)
         return c
 
     t1 = t.sum()
-    r = t1.map_chunk(f1)
+    t2 = t1.map_chunk(f1)
+    r = t2.map_chunk(f1)
     info = await session.execute(r)
 
     for _ in range(100):

mars/deploy/oscar/tests/test_ray_dag.py

@@ -60,13 +60,15 @@
 @pytest.mark.parametrize(indirect=True)
 @pytest.fixture
 async def create_cluster(request):
+    param = getattr(request, "param", {})
     start_method = os.environ.get("POOL_START_METHOD", None)
     client = await new_cluster(
         subprocess_start_method=start_method,
         backend="ray",
         n_worker=2,
         n_cpu=2,
         use_uvloop=False,
+        config=param.get("config", None),
     )
     async with client:
         assert client.session.client is not None
@@ -123,7 +125,11 @@ def test_sync_execute(ray_start_regular_shared2, config):
 
 
 @require_ray
-@pytest.mark.skip("Enable when ray progress got fixed")
+@pytest.mark.parametrize(
+    "create_cluster",
+    [{"config": {"task.execution_config.ray.subtask_monitor_interval": 0}}],
+    indirect=True,
+)
 @pytest.mark.asyncio
 async def test_session_get_progress(ray_start_regular_shared2, create_cluster):
     await test_local.test_session_get_progress(create_cluster)