Merge branch 'master' into fix_setup_print

Author: MrChengmo

core/trainers/framework/dataset.py

@@ -15,13 +15,13 @@
 from __future__ import print_function
 
 import os
-import warnings
 
 import paddle.fluid as fluid
 from paddlerec.core.utils import envs
 from paddlerec.core.utils import dataloader_instance
 from paddlerec.core.reader import SlotReader
 from paddlerec.core.trainer import EngineMode
+from paddlerec.core.utils.util import split_files
 
 __all__ = ["DatasetBase", "DataLoader", "QueueDataset"]
 
@@ -123,7 +123,8 @@ def _get_dataset(self, dataset_name, context):
             for x in os.listdir(train_data_path)
         ]
         if context["engine"] == EngineMode.LOCAL_CLUSTER:
-            file_list = context["fleet"].split_files(file_list)
+            file_list = split_files(file_list, context["fleet"].worker_index(),
+                                    context["fleet"].worker_num())
 
         dataset.set_filelist(file_list)
         for model_dict in context["phases"]:

core/utils/dataloader_instance.py

@@ -19,6 +19,7 @@
 from paddlerec.core.utils.envs import get_runtime_environ
 from paddlerec.core.reader import SlotReader
 from paddlerec.core.trainer import EngineMode
+from paddlerec.core.utils.util import split_files
 
 
 def dataloader_by_name(readerclass,
@@ -39,7 +40,8 @@ def dataloader_by_name(readerclass,
 
     files = [str(data_path) + "/%s" % x for x in os.listdir(data_path)]
     if context["engine"] == EngineMode.LOCAL_CLUSTER:
-        files = context["fleet"].split_files(files)
+        files = split_files(files, context["fleet"].worker_index(),
+                            context["fleet"].worker_num())
         print("file_list : {}".format(files))
 
     reader = reader_class(yaml_file)
@@ -80,7 +82,8 @@ def slotdataloader_by_name(readerclass, dataset_name, yaml_file, context):
 
     files = [str(data_path) + "/%s" % x for x in os.listdir(data_path)]
     if context["engine"] == EngineMode.LOCAL_CLUSTER:
-        files = context["fleet"].split_files(files)
+        files = split_files(files, context["fleet"].worker_index(),
+                            context["fleet"].worker_num())
         print("file_list: {}".format(files))
 
     sparse = get_global_env(name + "sparse_slots", "#")
@@ -133,7 +136,8 @@ def slotdataloader(readerclass, train, yaml_file, context):
 
     files = [str(data_path) + "/%s" % x for x in os.listdir(data_path)]
     if context["engine"] == EngineMode.LOCAL_CLUSTER:
-        files = context["fleet"].split_files(files)
+        files = split_files(files, context["fleet"].worker_index(),
+                            context["fleet"].worker_num())
         print("file_list: {}".format(files))
 
     sparse = get_global_env("sparse_slots", "#", namespace)

core/utils/envs.py

@@ -18,7 +18,9 @@
 import os
 import socket
 import sys
+import six
 import traceback
+import six
 
 global_envs = {}
 global_envs_flatten = {}
@@ -101,6 +103,12 @@ def fatten_env_namespace(namespace_nests, local_envs):
             name = ".".join(["dataset", dataset["name"], "type"])
             global_envs[name] = "DataLoader"
 
+    if get_platform() == "LINUX" and six.PY3:
+        print("QueueDataset can not support PY3, change to DataLoader")
+        for dataset in envs["dataset"]:
+            name = ".".join(["dataset", dataset["name"], "type"])
+            global_envs[name] = "DataLoader"
+
 
 def get_global_env(env_name, default_value=None, namespace=None):
     """
@@ -253,11 +261,19 @@ def load_yaml(config):
             use_full_loader = False
 
     if os.path.isfile(config):
-        with open(config, 'r') as rb:
-            if use_full_loader:
-                _config = yaml.load(rb.read(), Loader=yaml.FullLoader)
-            else:
-                _config = yaml.load(rb.read())
-            return _config
+        if six.PY2:
+            with open(config, 'r') as rb:
+                if use_full_loader:
+                    _config = yaml.load(rb.read(), Loader=yaml.FullLoader)
+                else:
+                    _config = yaml.load(rb.read())
+                return _config
+        else:
+            with open(config, 'r', encoding="utf-8") as rb:
+                if use_full_loader:
+                    _config = yaml.load(rb.read(), Loader=yaml.FullLoader)
+                else:
+                    _config = yaml.load(rb.read())
+                return _config
     else:
         raise ValueError("config {} can not be supported".format(config))

core/utils/util.py

@@ -19,11 +19,8 @@
 import numpy as np
 from paddle import fluid
 
-from paddlerec.core.utils import fs as fs
-
 
 def save_program_proto(path, program=None):
-
     if program is None:
         _program = fluid.default_main_program()
     else:
@@ -171,6 +168,39 @@ def print_cost(cost, params):
     return log_str
 
 
+def split_files(files, trainer_id, trainers):
+    """
+    split files before distributed training,
+    example 1: files is [a, b, c ,d, e]  and trainer_num = 2, then trainer
+               0 gets [a, b, c] and trainer 1 gets [d, e].
+    example 2: files is [a, b], and trainer_num = 3, then trainer 0 gets
+               [a], trainer 1 gets [b],  trainer 2 gets []
+
+    Args:
+        files(list): file list need to be read.
+
+    Returns:
+        list: files belongs to this worker.
+    """
+    if not isinstance(files, list):
+        raise TypeError("files should be a list of file need to be read.")
+
+    remainder = len(files) % trainers
+    blocksize = int(len(files) / trainers)
+
+    blocks = [blocksize] * trainers
+    for i in range(remainder):
+        blocks[i] += 1
+
+    trainer_files = [[]] * trainers
+    begin = 0
+    for i in range(trainers):
+        trainer_files[i] = files[begin:begin + blocks[i]]
+        begin += blocks[i]
+
+    return trainer_files[trainer_id]
+
+
 class CostPrinter(object):
     """
     For count cost time && print cost log

models/contentunderstanding/readme.md

@@ -86,7 +86,7 @@ python -m paddlerec.run -m ./config.yaml # 自定义修改超参后，指定配
 # 修改对应模型的config.yaml, workspace配置为当前目录的绝对路径
 # 修改对应模型的config.yaml，mode配置infer_runner
 # 示例: mode: train_runner -> mode: infer_runner
-# infer_runner中 class配置为 class: single_infer
+# infer_runner中 class配置为 class: infer
 # 修改phase阶段为infer的配置，参照config注释
 
 # 修改完config.yaml后 执行:
@@ -106,7 +106,7 @@ python -m paddlerec.run -m ./config.yaml # 自定义修改超参后，指定配
 # 修改对应模型的config.yaml, workspace配置为当前目录的绝对路径
 # 修改对应模型的config.yaml，mode配置infer_runner
 # 示例: mode: train_runner -> mode: infer_runner
-# infer_runner中 class配置为 class: single_infer
+# infer_runner中 class配置为 class: infer
 # 修改phase阶段为infer的配置，参照config注释
 
 # 修改完config.yaml后 执行:

models/demo/movie_recommand/rank/config.yaml

@@ -64,8 +64,7 @@ runner:
   device: cpu
 
 - name: runner_infer
-  epochs: 1
-  class: single_infer
+  class: infer
   print_interval: 10000
   init_model_path: "increment/9" # load model path
 

models/demo/movie_recommand/recall/config.yaml

@@ -64,8 +64,7 @@ runner:
   device: cpu
 
 - name: runner_infer
-  epochs: 1
-  class: single_infer
+  class: infer
   print_interval: 10000
   init_model_path: "increment/9" # load model path
 

models/match/dssm/config.yaml

@@ -56,9 +56,7 @@ runner:
   init_model_path: "" # load model path
   print_interval: 2
 - name: infer_runner
-  class: single_infer
-  # num of epochs
-  epochs: 1
+  class: infer
   # device to run training or infer
   device: cpu
   print_interval: 1

models/match/multiview-simnet/config.yaml

@@ -63,9 +63,7 @@ runner:
   init_model_path: "" # load model path
   print_interval: 1
 - name: infer_runner
-  class: single_infer
-  # num of epochs
-  epochs: 1
+  class: infer
   # device to run training or infer
   device: cpu
   print_interval: 1

models/match/readme.md

@@ -43,7 +43,7 @@ python -m paddlerec.run -m paddlerec.models.match.multiview-simnet # multiview-s
 # 修改对应模型的config.yaml, workspace配置为当前目录的绝对路径
 # 修改对应模型的config.yaml，mode配置infer_runner
 # 示例: mode: train_runner -> mode: infer_runner
-# infer_runner中 class配置为 class: single_infer
+# infer_runner中 class配置为 class: infer
 # 修改phase阶段为infer的配置，参照config注释
 
 # 修改完config.yaml后 执行: