Fix supervisor n_process (#2950)

Co-authored-by: hekaisheng <[email protected]>

Author: wjsi

mars/core/graph/tests/test_graph.py

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import logging
 import pytest
 
 from .... import tensor as mt
 from ....tests.core import flaky
+from ....utils import to_str
 from .. import DAG, GraphContainsCycleError
 
 
@@ -103,22 +103,5 @@ def test_to_dot():
     arr2 = arr + arr_add
     graph = arr2.build_graph(fuse_enabled=False, tile=True)
 
-    dot = str(graph.to_dot(trunc_key=5))
-    try:
-        assert all(str(n.key)[5] in dot for n in graph) is True
-    except AssertionError:
-        graph_reprs = []
-        for n in graph:
-            graph_reprs.append(
-                f"{n.op.key} -> {[succ.key for succ in graph.successors(n)]}"
-            )
-        logging.error(
-            "Unexpected error in test_to_dot.\ndot = %r\ngraph_repr = %r",
-            dot,
-            "\n".join(graph_reprs),
-        )
-        missing_prefix = next(n.key for n in graph if str(n.key)[5] not in dot)
-        logging.error(
-            "Missing prefix %r (type: %s)", missing_prefix, type(missing_prefix)
-        )
-        raise
+    dot = to_str(graph.to_dot(trunc_key=5))
+    assert all(to_str(n.key)[:5] in dot for n in graph) is True

mars/deploy/kubernetes/config.py

@@ -15,6 +15,7 @@
 
 import abc
 import functools
+import math
 import re
 
 from ... import __version__ as mars_version
@@ -605,6 +606,8 @@ def build_container_command(self):
             cmd += ["-p", str(self._service_port)]
         if self._web_port:
             cmd += ["-w", str(self._web_port)]
+        if self._cpu:
+            cmd += ["--n-process", str(int(math.ceil(self._cpu)))]
         return cmd
 
 

mars/deploy/oscar/supervisor.py

@@ -38,7 +38,7 @@ def config_args(self, parser):
         super().config_args(parser)
         parser.add_argument("-w", "--web-port", help="web port of the service")
         parser.add_argument(
-            "--n-process", help="number of supervisor processes", default="0"
+            "--n-process", help="number of supervisor processes", default="1"
         )
 
     def parse_args(self, parser, argv, environ=None):

mars/oscar/backends/context.py

@@ -50,12 +50,12 @@ class ProfilingContext:
 
 
 class MarsActorContext(BaseActorContext):
-    __slots__ = "_address", "_caller"
+    __slots__ = ("_caller",)
 
     support_allocate_strategy = True
 
     def __init__(self, address: str = None):
-        self._address = address
+        BaseActorContext.__init__(self, address)
         self._caller = ActorCaller()
 
     def __del__(self):

mars/oscar/backends/mars/pool.py

@@ -17,8 +17,10 @@
 import logging.config
 import multiprocessing
 import os
+import random
 import signal
 import sys
+import uuid
 from dataclasses import dataclass
 from types import TracebackType
 from typing import List
@@ -164,6 +166,9 @@ def _start_sub_pool(
     ):
         ensure_coverage()
 
+        # make sure enough randomness for every sub pool
+        random.seed(uuid.uuid1().bytes)
+
         conf = actor_config.get_pool_config(process_index)
         suspend_sigint = conf["suspend_sigint"]
         if suspend_sigint:

mars/oscar/backends/mars/tests/test_mars_actor_context.py

@@ -242,7 +242,7 @@ def get_call_log(self):
 
 @pytest.mark.parametrize(indirect=True)
 @pytest.fixture(params=[False, True])
-async def actor_pool_context(request):
+async def actor_pool(request):
     start_method = (
         os.environ.get("POOL_START_METHOD", "forkserver")
         if sys.platform != "win32"
@@ -266,9 +266,10 @@ async def actor_pool_context(request):
 
 
 @pytest.mark.asyncio
-async def test_simple_local_actor_pool(actor_pool_context):
-    pool = actor_pool_context
-    actor_ref = await mo.create_actor(DummyActor, 100, address=pool.external_address)
+async def test_simple_local_actor_pool(actor_pool):
+    actor_ref = await mo.create_actor(
+        DummyActor, 100, address=actor_pool.external_address
+    )
     assert await actor_ref.add(1) == 101
     await actor_ref.add(1)
 
@@ -279,17 +280,18 @@ async def test_simple_local_actor_pool(actor_pool_context):
     assert actor_ref.address == ref2.address
     assert actor_ref.uid == ref2.uid
 
-    ref = await mo.actor_ref(uid=actor_ref.uid, address=pool.external_address)
+    ref = await mo.actor_ref(uid=actor_ref.uid, address=actor_pool.external_address)
     assert await ref.add(2) == 104
 
 
 @pytest.mark.asyncio
-async def test_mars_post_create_pre_destroy(actor_pool_context):
-    pool = actor_pool_context
+async def test_mars_post_create_pre_destroy(actor_pool):
     rec_ref = await mo.create_actor(
-        RecordActor, uid=RecordActor.default_uid(), address=pool.external_address
+        RecordActor, uid=RecordActor.default_uid(), address=actor_pool.external_address
+    )
+    actor_ref = await mo.create_actor(
+        CreateDestroyActor, address=actor_pool.external_address
     )
-    actor_ref = await mo.create_actor(CreateDestroyActor, address=pool.external_address)
     await actor_ref.destroy()
 
     records = await rec_ref.get_records()
@@ -299,72 +301,78 @@ async def test_mars_post_create_pre_destroy(actor_pool_context):
 
 
 @pytest.mark.asyncio
-async def test_mars_create_actor(actor_pool_context):
-    pool = actor_pool_context
-    actor_ref = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+async def test_mars_create_actor(actor_pool):
+    actor_ref = await mo.create_actor(
+        DummyActor, 1, address=actor_pool.external_address
+    )
     # create actor inside on_receive
-    r = await actor_ref.create(DummyActor, 5, address=pool.external_address)
-    ref = await mo.actor_ref(r, address=pool.external_address)
+    r = await actor_ref.create(DummyActor, 5, address=actor_pool.external_address)
+    ref = await mo.actor_ref(r, address=actor_pool.external_address)
     assert await ref.add(10) == 15
     # create actor inside on_receive and send message
     r = await actor_ref.create_send(
-        DummyActor, 5, method="add", method_args=(1,), address=pool.external_address
+        DummyActor,
+        5,
+        method="add",
+        method_args=(1,),
+        address=actor_pool.external_address,
     )
     assert r == 6
 
 
 @pytest.mark.asyncio
-async def test_mars_create_actor_error(actor_pool_context):
-    pool = actor_pool_context
+async def test_mars_create_actor_error(actor_pool):
     ref1 = await mo.create_actor(
-        DummyActor, 1, uid="dummy1", address=pool.external_address
+        DummyActor, 1, uid="dummy1", address=actor_pool.external_address
     )
     with pytest.raises(mo.ActorAlreadyExist):
         await mo.create_actor(
-            DummyActor, 1, uid="dummy1", address=pool.external_address
+            DummyActor, 1, uid="dummy1", address=actor_pool.external_address
         )
     await mo.destroy_actor(ref1)
 
     with pytest.raises(ValueError):
-        await mo.create_actor(DummyActor, -1, address=pool.external_address)
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+        await mo.create_actor(DummyActor, -1, address=actor_pool.external_address)
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     with pytest.raises(ValueError):
-        await ref1.create(DummyActor, -2, address=pool.external_address)
+        await ref1.create(DummyActor, -2, address=actor_pool.external_address)
 
 
 @pytest.mark.asyncio
-async def test_mars_send(actor_pool_context):
-    pool = actor_pool_context
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+async def test_mars_send(actor_pool):
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     ref2 = await mo.actor_ref(
-        await ref1.create(DummyActor, 2, address=pool.external_address)
+        await ref1.create(DummyActor, 2, address=actor_pool.external_address)
     )
     assert await ref1.send(ref2, "add", 3) == 5
 
-    ref3 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+    ref3 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     ref4 = await mo.create_actor(
-        DummyActor, 2, address=pool.external_address, allocate_strategy=RandomSubPool()
+        DummyActor,
+        2,
+        address=actor_pool.external_address,
+        allocate_strategy=RandomSubPool(),
     )
     assert await ref4.send(ref3, "add", 3) == 4
 
 
 @pytest.mark.asyncio
-async def test_mars_send_error(actor_pool_context):
-    pool = actor_pool_context
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+async def test_mars_send_error(actor_pool):
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     with pytest.raises(TypeError):
         await ref1.add(1.0)
-    ref2 = await mo.create_actor(DummyActor, 2, address=pool.external_address)
+    ref2 = await mo.create_actor(DummyActor, 2, address=actor_pool.external_address)
     with pytest.raises(TypeError):
         await ref1.send(ref2, "add", 1.0)
     with pytest.raises(mo.ActorNotExist):
-        await (await mo.actor_ref("fake_uid", address=pool.external_address)).add(1)
+        await (await mo.actor_ref("fake_uid", address=actor_pool.external_address)).add(
+            1
+        )
 
 
 @pytest.mark.asyncio
-async def test_mars_tell(actor_pool_context):
-    pool = actor_pool_context
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+async def test_mars_tell(actor_pool):
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     ref2 = await mo.actor_ref(await ref1.create(DummyActor, 2))
     await ref1.tell(ref2, "add", 3)
     assert await ref2.get_value() == 5
@@ -382,9 +390,8 @@ async def test_mars_tell(actor_pool_context):
 
 
 @pytest.mark.asyncio
-async def test_mars_batch_method(actor_pool_context):
-    pool = actor_pool_context
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+async def test_mars_batch_method(actor_pool):
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     batch_result = await ref1.add_ret.batch(
         ref1.add_ret.delay(1), ref1.add_ret.delay(2), ref1.add_ret.delay(3)
     )
@@ -401,11 +408,10 @@ async def test_mars_batch_method(actor_pool_context):
 
 
 @pytest.mark.asyncio
-async def test_gather_exception(actor_pool_context):
+async def test_gather_exception(actor_pool):
     try:
         Router.get_instance_or_empty()._cache.clear()
-        pool = actor_pool_context
-        ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+        ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
         router = Router.get_instance_or_empty()
         client = next(iter(router._cache.values()))
 
@@ -439,9 +445,8 @@ class MyException(Exception):
 
 
 @pytest.mark.asyncio
-async def test_mars_destroy_has_actor(actor_pool_context):
-    pool = actor_pool_context
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+async def test_mars_destroy_has_actor(actor_pool):
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     ref2 = await mo.actor_ref(ref1)
     ref2_add_method = ref2.add
     assert isinstance(ref1, ActorRef)
@@ -465,7 +470,7 @@ async def test_mars_destroy_has_actor(actor_pool_context):
         await ref2_add_method(1)
 
     ref1 = await mo.create_actor(
-        DummyActor, 1, uid=ref1.uid, address=pool.external_address
+        DummyActor, 1, uid=ref1.uid, address=actor_pool.external_address
     )
 
     # the ref2 should be works after actor is recreated.
@@ -479,8 +484,8 @@ async def test_mars_destroy_has_actor(actor_pool_context):
     assert not await mo.has_actor(ref1)
     assert not await mo.has_actor(ref2)
 
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
-    ref2 = await ref1.create(DummyActor, 2, address=pool.external_address)
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
+    ref2 = await ref1.create(DummyActor, 2, address=actor_pool.external_address)
 
     assert await mo.has_actor(ref2)
 
@@ -489,23 +494,24 @@ async def test_mars_destroy_has_actor(actor_pool_context):
 
     with pytest.raises(mo.ActorNotExist):
         await mo.destroy_actor(
-            await mo.actor_ref("fake_uid", address=pool.external_address)
+            await mo.actor_ref("fake_uid", address=actor_pool.external_address)
         )
 
-    ref1 = await mo.create_actor(DummyActor, 1, address=pool.external_address)
+    ref1 = await mo.create_actor(DummyActor, 1, address=actor_pool.external_address)
     with pytest.raises(mo.ActorNotExist):
-        await ref1.delete(await mo.actor_ref("fake_uid", address=pool.external_address))
+        await ref1.delete(
+            await mo.actor_ref("fake_uid", address=actor_pool.external_address)
+        )
 
     # test self destroy
-    ref1 = await mo.create_actor(DummyActor, 2, address=pool.external_address)
+    ref1 = await mo.create_actor(DummyActor, 2, address=actor_pool.external_address)
     await ref1.destroy()
     assert not await mo.has_actor(ref1)
 
 
 @pytest.mark.asyncio
-async def test_mars_resource_lock(actor_pool_context):
-    pool = actor_pool_context
-    ref = await mo.create_actor(ResourceLockActor, address=pool.external_address)
+async def test_mars_resource_lock(actor_pool):
+    ref = await mo.create_actor(ResourceLockActor, address=actor_pool.external_address)
     event_list = []
 
     async def test_task(idx):
@@ -525,13 +531,12 @@ async def test_task(idx):
 
 
 @pytest.mark.asyncio
-async def test_promise_chain(actor_pool_context):
-    pool = actor_pool_context
+async def test_promise_chain(actor_pool):
     lock_ref = await mo.create_actor(
-        ResourceLockActor, 2, address=pool.external_address
+        ResourceLockActor, 2, address=actor_pool.external_address
     )
     promise_test_ref = await mo.create_actor(
-        PromiseTestActor, lock_ref, address=pool.external_address
+        PromiseTestActor, lock_ref, address=actor_pool.external_address
     )
 
     delay_val = 1.0
@@ -609,8 +614,8 @@ async def __pre_destroy__(self):
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("in_sub_pool", [True, False])
-async def test_error_in_pre_destroy(actor_pool_context, in_sub_pool):
-    pool = actor_pool_context
+async def test_error_in_pre_destroy(actor_pool, in_sub_pool):
+    pool = actor_pool
 
     strategy = None if not in_sub_pool else RandomSubPool()
     a = await mo.create_actor(

mars/oscar/backends/pool.py

@@ -724,8 +724,14 @@ async def create_actor(self, message: CreateActorMessage) -> ResultMessageType:
     async def actor_ref(self, message: ActorRefMessage) -> ResultMessageType:
         result = await super().actor_ref(message)
         if isinstance(result, ErrorMessage):
-            message.actor_ref.address = self._main_address
-            result = await self.call(self._main_address, message)
+            # need a new message id to call main actor
+            main_message = ActorRefMessage(
+                new_message_id(),
+                create_actor_ref(self._main_address, message.actor_ref.uid),
+            )
+            result = await self.call(self._main_address, main_message)
+            # rewrite to message_id of the original request
+            result.message_id = message.message_id
         return result
 
     @implements(AbstractActorPool.destroy_actor)

mars/oscar/context.pxd

@@ -14,7 +14,7 @@
 
 
 cdef class BaseActorContext:
-    pass
+    cdef public str _address
 
 
 cpdef get_context()