Add more logs on execution info of operands (#2940)

Author: hekaisheng

mars/core/entity/executable.py

@@ -94,7 +94,11 @@ def cb(_, sess=ref(session)):
                 # if decref in isolation, means that this tileable
                 # is not required for main thread, thus we do not need
                 # to wait for decref, otherwise, wait a bit
-                fut.result(0.5)
+                try:
+                    fut.result(0.5)
+                except concurrent.futures.TimeoutError:
+                    # ignore timeout
+                    pass
 
         self.tileable = ref(tileable, cb)
 

mars/dataframe/base/bloom_filter.py

@@ -72,7 +72,7 @@ def tile(cls, op: "DataFrameBloomFilter"):
 
         # union all chunk filters
         combine_size = options.combine_size
-        while len(chunks) > 4:
+        while len(chunks) > combine_size:
             new_chunks = []
             for i in range(0, len(chunks), combine_size):
                 chks = chunks[i : i + combine_size]

mars/dataframe/groupby/aggregation.py

@@ -14,6 +14,7 @@
 
 import functools
 import itertools
+import logging
 import typing
 import uuid
 from typing import List
@@ -59,6 +60,8 @@
 cp = lazy_import("cupy", globals=globals(), rename="cp")
 cudf = lazy_import("cudf", globals=globals())
 
+logger = logging.getLogger(__name__)
+
 _support_get_group_without_as_index = pd_release_version[:2] > (1, 0)
 
 
@@ -617,9 +620,11 @@ def _tile_auto(
         if cls._choose_tree_method(
             raw_sizes, agg_sizes, len(chunks), len(in_df.chunks), op.chunk_store_limit
         ):
+            logger.debug("Choose tree method for groupby operand %s", op)
             return cls._combine_tree(op, chunks + left_chunks, out_df, func_infos)
         else:
             # otherwise, use shuffle
+            logger.debug("Choose shuffle method for groupby operand %s", op)
             return cls._perform_shuffle(
                 op, chunks + left_chunks, in_df, out_df, func_infos
             )

mars/dataframe/merge/merge.py

@@ -627,6 +627,12 @@ def tile(cls, op: "DataFrameMerge"):
             if op.how == "inner" and op.bloom_filter:
                 if has_unknown_shape(left, right):
                     yield left.chunks + right.chunks
+                small_one = right if len(left.chunks) > len(right.chunks) else left
+                logger.debug(
+                    "Apply bloom filter for operand %s, use DataFrame %s to build bloom filter.",
+                    op,
+                    small_one,
+                )
                 left, right = yield from recursive_tile(
                     *cls._apply_bloom_filter(left, right, left_on, right_on, op)
                 )
@@ -638,6 +644,7 @@ def tile(cls, op: "DataFrameMerge"):
             if op.method == "auto":
                 # if method is auto, select new method after auto merge
                 method = cls._choose_merge_method(op, left, right)
+        logger.debug("Choose %s method for merge operand %s", method, op)
         if method == MergeMethod.one_chunk:
             ret = cls._tile_one_chunk(op, left, right)
         elif method == MergeMethod.broadcast:

mars/serialization/core.pyx

@@ -15,6 +15,7 @@
 
 import asyncio
 import datetime
+import hashlib
 import inspect
 import sys
 from cpython cimport PyObject
@@ -27,7 +28,6 @@ import numpy as np
 import pandas as pd
 
 from .._utils cimport TypeDispatcher
-from ..utils import tokenize_int
 
 import cloudpickle
 
@@ -142,7 +142,9 @@ cdef class Serializer:
 
     @classmethod
     def calc_default_serializer_id(cls):
-        return tokenize_int(f"{cls.__module__}.{cls.__qualname__}") % _SERIALIZER_ID_PRIME
+        s = f"{cls.__module__}.{cls.__qualname__}"
+        h = hashlib.md5(s.encode())
+        return int(h.hexdigest(), 16) % _SERIALIZER_ID_PRIME
 
     @classmethod
     def register(cls, obj_type):

mars/services/scheduling/worker/execution.py

@@ -111,16 +111,18 @@ def _fill_subtask_result_with_exception(
     _, exc, tb = sys.exc_info()
     if isinstance(exc, asyncio.CancelledError):
         status = SubtaskStatus.cancelled
-        log_str = "Cancel"
+        logger.exception(
+            "Cancel run subtask %s on band %s",
+            subtask.subtask_id,
+            subtask_info.band_name,
+        )
     else:
         status = SubtaskStatus.errored
-        log_str = "Failed to"
-    logger.exception(
-        "%s run subtask %s on band %s",
-        log_str,
-        subtask.subtask_id,
-        subtask_info.band_name,
-    )
+        logger.exception(
+            "Failed to run subtask %s on band %s",
+            subtask.subtask_id,
+            subtask_info.band_name,
+        )
     subtask_info.result.status = status
     subtask_info.result.progress = 1.0
     subtask_info.result.error = exc

mars/services/session/supervisor/core.py

@@ -14,6 +14,8 @@
 
 import asyncio
 import functools
+import logging
+import os
 import time
 from typing import Dict, List, Optional
 
@@ -23,6 +25,8 @@
 from ...core import NodeRole, create_service_session, destroy_service_session
 from ..core import SessionInfo
 
+logger = logging.getLogger(__name__)
+
 
 class SessionManagerActor(mo.Actor):
     def __init__(self, service_config: Optional[Dict] = None):
@@ -164,6 +168,11 @@ async def __post_create__(self):
             address=self.address,
             uid=CustomLogMetaActor.gen_uid(self._session_id),
         )
+        logger.debug(
+            "Session %s actor created on pid: %s",
+            self._session_id,
+            os.getpid(),
+        )
 
     async def remove(self):
         await destroy_service_session(

setup.py

@@ -72,10 +72,13 @@
     cythonize_kw["compiler_directives"] = {"linetrace": True}
 
 if "MSC" in sys.version:
-    extra_compile_args = ["/Ot", "/I" + os.path.join(repo_root, "misc")]
+    extra_compile_args = ["/std:c11", "/Ot", "/I" + os.path.join(repo_root, "misc")]
     cy_extension_kw["extra_compile_args"] = extra_compile_args
 else:
     extra_compile_args = ["-O3"]
+    if sys.platform != "darwin":
+        # for macOS, we assume that C++ 11 is enabled by default
+        extra_compile_args.append("-std=c++0x")
     cy_extension_kw["extra_compile_args"] = extra_compile_args