Merge pull request #596 from esythan/master

barrier_worker in ps_trainer

doc/online_trainer.md

@@ -41,7 +41,7 @@
 ## 模型
 流式训练采用静态图参数服务器方式训练，在组网时需要注意几点：
 1. embedding层需使用paddle.static.nn.sparse_embedding，其中size参数的第一维可指定任意值，第二维为embedding向量的维度。  
-2. 在组网中增加inference_feed_vars、inference_target_var两个变量的赋值，指明inference_model的输入和输出，供线上预测使用。
+2. 在组网中增加inference_feed_vars、inference_target_var两个变量的赋值，指明inference_model的输入和输出，供在线推理使用。
 3. 在组网中增加all_vars变量的赋值，可用于在线离线一致性检查。
 4. 如果希望在训练过程中dump出组网中的变量和网络参数（主要用于训练中的调试和异常检查），请赋值train_dump_fields和train_dump_params；如果希望在预测过程中dump出组网中的变量（主要用于线上预测所需特征的离线灌库），请赋值infer_dump_fields。
 
@@ -65,5 +65,5 @@ fleetrun --server_num=1 --worker_num=1 ../../../tools/static_ps_online_trainer.p
 目录下的embedding.shard目录为sparse特征对应的embedding，其中.txt文件为具体的embedding值和优化方法需要的统计量，.meta文件指明.txt文件的具体schema。  
 目录下的其他文件为dense参数，文件名即为这些参数在组网中对应的var_name。  
 ### inference_model
-用于线上预测的模型，保存于model_save_path/day/inference_model_{$pass_id}中，分为model、sparse、dense三个部分。  
+用于在线推理的模型，保存于model_save_path/day/inference_model_{$pass_id}中，分为model、sparse、dense三个部分。  
 其中sparse和dense参数与checkpoint模型类似，多出一个名为“__model__”的文件，保存的是在线服务使用的组网（可能经过裁剪），线上服务可以直接加载。  

doc/onoff_diff.md

@@ -2,10 +2,10 @@
 在线上服务搭建完成之后，为保证正确性，需要对同一批数据进行在线和离线预测，并对比结果，结果一致，才能启动线上服务。  
 需要注意的是，在线/离线两个环境，预测的数据、加载的模型需保持一致。  
 
-## 在线预测
-参见在线预测模块(tools/inference)。  
-需要开启debug模式，数据预测过程中打印每一层的输出tensor。  
-如果线上预测是通过kv存储获取embedding向量，则还需要将预测数据中的所有feasign对应的embedding输出到文件中。  
+## 在线推理
+参见在线推理模块(tools/inference)。  
+需要开启debug模式，数据推理过程中打印每一层的输出tensor。  
+如果在线推理是通过kv存储获取embedding向量，则还需要将数据中的所有feasign对应的embedding输出到文件中。  
 
 ## 离线预测
 如果sparse embedding过大，离线预测无法单机进行，则需要从线上拿到预测数据所有feasign对应的embedding，并转换成离线能够加载的模型格式，这部分操作在get_xbox_model.py脚本中进行。  
@@ -19,6 +19,6 @@
 1. 启动前准备：  
     准备预测数据，放入在线环境相应位置及离线slot_dnn目录的infer_data/online下，同时在离线slot_dnn目录中建立infer_data/offline空文件夹  
     将离线训练好的模型cp到在线环境中（包括slot_dnn目录下的all_vars.txt文件及模型保存目录下的model和参数文件）  
-2. 启动在线预测：生成cube.result及std.log文件  
+2. 启动在线推理：生成cube.result及std.log文件  
 3. 将上述两个文件cp至tools/onoff_diff/data文件夹下，std.log重命名为log.online  
 4. 在tools/onoff_diff目录下运行sh run.sh，完成离线预测及在线离线一致性检查工作  

doc/whole_process.md

@@ -19,10 +19,10 @@ PaddleRec支持多种数据源，包括文件、kafka、odps、tf_record等，
 ## 特征重要性
 在训练过程中，如果你想要衡量模型中特征的重要性，可以参考[特征重要性](./feature_importance.md)
 
-## 预测
-分布式训练过程中，会保存inference_model用于线上预测。
-1. 如果你想要搭建一个独立的预测服务，请参考[Serving部署](./serving.md)
-2. 如果你已经拥有一个线上服务，想要在其中增加Paddle的推荐预测内容，请参考[Inference部署](../tools/inference/README.md)
+## 在线推理
+分布式训练过程中，会保存inference_model用于在线推理。
+1. 如果你想要搭建一个独立的推理服务，请参考[Paddle Serving部署](./serving.md)
+2. 如果你已经拥有一个在线服务，想要在其中增加Paddle的推理内容，请参考[Paddle Inference部署](../tools/inference/README.md)
 
 ## 在线离线一致性检查
 在线上服务搭建完成，正式投入使用之前，需要进行在线离线一致性检查，确保在线服务的正确性，具体内容参见[在线离线一致性检查](./onoff_diff.md)

tools/static_gpubox_trainer.py

@@ -179,6 +179,7 @@ def run_worker(self):
                     self.exe, model_dir,
                     [feed.name for feed in self.input_data],
                     self.inference_target_var)
+            fleet.barrier_worker()
             self.reader.release_memory()
             self.PSGPU.end_pass()
             logger.info("finish {} epoch training....".format(epoch))

tools/static_ps_online_trainer.py

@@ -4,7 +4,7 @@
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
-#     http://www.apache.org/licenses/LICENSE-2.0
+#     http://www.apache.org/licenses/LICENSE-2.0  
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
@@ -407,6 +407,7 @@ def run_online_worker(self):
                         xbox_base_key=model_base_key,
                         train_local=self.train_local,
                         client=self.hadoop_client)
+            fleet.barrier_worker()
             day = get_next_day(day)
 
     def dataset_train_loop(self, cur_dataset, day, pass_index,

tools/static_ps_trainer.py

@@ -166,6 +166,7 @@ def run_worker(self):
                         model_dir,
                         [feed.name for feed in self.inference_feed_var],
                         [self.inference_target_var], self.exe)
+            fleet.barrier_worker()
 
         if reader_type == "InmemoryDataset":
             self.reader.release_memory()