[auto config] show error info on csv (#60101)

xysheng-baidu · web-flow · commit beac9cb49ce1 · 2023-12-20T16:55:34.000+08:00
* [auto config] show error info on csv

* [auto config] fix error on show error on csv
diff --git a/python/paddle/distributed/auto_tuner/utils.py b/python/paddle/distributed/auto_tuner/utils.py
@@ -1229,6 +1229,48 @@ def read_log(
     return res_metric, res_memory, err_code
 
 
+def get_error_info(filename):
+    """
+    get error info from log file
+    return:
+        error_info: Specific error message
+    """
+    error_infos = []
+    error_pattern = r"Error"
+    with open(filename, 'r') as file:
+        lines = file.readlines()
+        last_lines = lines[-100:]
+        for line in last_lines:
+            error_info = re.findall(error_pattern, line, re.IGNORECASE)
+            if error_info:
+                if "Out of memory" in line:
+                    error_infos.append("Out of memory")
+                else:
+                    error_infos.append(line)
+    return list(set(error_infos))
+
+
+def find_error_from_log(path):
+    """
+    find error infos from log directory
+    return:
+        error_info: all error message on log directory
+    """
+    unique_error_info = ""
+    all_error_infos = []
+    for root, dirs, files in os.walk(path):
+        for file in files:
+            if not file.startswith("workerlog"):
+                continue
+            error_infos = get_error_info(path + "/" + file)
+            all_error_infos += error_infos
+    all_error_infos = list(set(all_error_infos))
+    for info in all_error_infos:
+        unique_error_info = unique_error_info + info + ","
+    unique_error_info = unique_error_info[:-1]
+    return unique_error_info
+
+
 def three_mul_combinations(target):
     """Return the combinations of three numbers which product is target."""
     results = []
diff --git a/python/paddle/distributed/launch/main.py b/python/paddle/distributed/launch/main.py
@@ -307,6 +307,7 @@ def launch():
         from ..auto_tuner.tuner import AutoTuner
         from ..auto_tuner.utils import (
             add_overlap_performance,
+            find_error_from_log,
             gen_new_args,
             gen_new_ctx,
             read_log,
@@ -916,10 +917,61 @@ def launch():
                         f"bw_{bw}_{tuner_cfg['metric_cfg']['name']}"
                     ] = multi_dp_performace
 
+            error_info = None
             cur_cfg["has_error"] = has_error
             if has_error:
+                error_info = []
                 error_task_nums += 1
-            error_info = None
+                if OOM_flag:
+                    error_info.append("Out of memory")
+                else:
+                    if actual_nnodes > 1:
+                        path = f"auto_tuner/error/{job_id}/{ip}"
+                        single_error_info = find_error_from_log(
+                            ctx.args.log_dir
+                        )
+                        if len(single_error_info) > 0:
+                            while not client.put(
+                                path, single_error_info.encode('latin-1')
+                            ):
+                                time.sleep(1)
+                            ctx.logger.info(
+                                f"Put Error info: {single_error_info} to {path}"
+                            )
+                            logger.info(
+                                f"Put Error info: {single_error_info} to {path}"
+                            )
+                        else:
+                            while not client.put(path, "OK".encode('latin-1')):
+                                time.sleep(1)
+                            ctx.logger.info(f"Put OK to {path}")
+                            logger.info(f"Put OK to {path}")
+
+                        result = list(
+                            client.get_prefix(f"auto_tuner/error/{job_id}/")
+                        )
+                        size = len(result)
+                        while size != actual_nnodes:
+                            time.sleep(1)
+                            result = list(
+                                client.get_prefix(f"auto_tuner/error/{job_id}/")
+                            )
+                            size = len(result)
+
+                        status = [
+                            i[0].decode()
+                            for i in result
+                            if "OK" not in i[0].decode()
+                        ]
+                        error_info = list(set(status))
+                        ctx.logger.info(
+                            f"Status of auto_tuner/error/{job_id}/: {error_info}"
+                        )
+                        logger.info(
+                            f"Status of auto_tuner/error/{job_id}/: {error_info}"
+                        )
+                    else:
+                        error_info.append(find_error_from_log(ctx.args.log_dir))
             cur_cfg["error_info"] = error_info
             task_nums = len(auto_tuner.algo.all_tasks)
             cur_task_id = auto_tuner.algo.idx
