统一 ps 开发 - python

python/paddle/distributed/fleet/meta_optimizers/ps_optimizer.py

@@ -31,7 +31,6 @@ def __init__(self, optimizer):
         self.inner_opt = optimizer
         # we do not allow meta optimizer to be inner optimizer currently
         self.meta_optimizers_white_list = []
-        self.attrs = {}
         self.pass_ctx = PassContext()
 
     def _set_basic_info(self, loss, role_maker, user_defined_optimizer,
@@ -40,50 +39,48 @@ def _set_basic_info(self, loss, role_maker, user_defined_optimizer,
             loss, role_maker, user_defined_optimizer, user_defined_strategy)
 
     def _init_ps_pass_context(self, loss, startup_program):
+        attrs = {}
         # trainer
-        self.attrs["env"] = get_dist_env()
+        attrs["env"] = get_dist_env()
 
-        self.attrs['loss'] = loss
-        self.attrs['min_block_size'] = 81920
-        self.attrs['origin_main_program'] = loss.block.program
-        self.attrs['origin_startup_program'] = startup_program
+        attrs['loss'] = loss
+        attrs['min_block_size'] = 81920
+        attrs['origin_main_program'] = loss.block.program
+        attrs['origin_startup_program'] = startup_program
 
-        self.attrs['cloned_main'] = loss.block.program.clone()
-        self.attrs['cloned_startup'] = startup_program.clone()
+        attrs['cloned_main'] = attrs['origin_main_program'].clone()
+        attrs['cloned_startup'] = attrs['origin_startup_program'].clone()
 
-        self.attrs['user_defined_strategy'] = self.user_defined_strategy
-        self.attrs['trainer'] = TrainerRuntimeConfig(self.user_defined_strategy)
-        self.attrs['ps_mode'] = self.attrs['trainer'].mode
+        attrs['user_defined_strategy'] = self.user_defined_strategy
+        attrs['trainer'] = TrainerRuntimeConfig(self.user_defined_strategy)
+        attrs['ps_mode'] = attrs['trainer'].mode
 
-        self.attrs['role_maker'] = self.role_maker
-        self.attrs[
+        attrs['role_maker'] = self.role_maker
+        attrs[
             'is_heter_ps_mode'] = self.role_maker._is_heter_parameter_server_mode
-        self.attrs['is_worker'] = self.role_maker._is_worker()
-        self.attrs['is_server'] = self.role_maker._is_server()
-        self.attrs['is_heter_worker'] = self.role_maker._is_heter_worker()
+        attrs['is_worker'] = self.role_maker._is_worker()
+        attrs['is_server'] = self.role_maker._is_server()
+        attrs['is_heter_worker'] = self.role_maker._is_heter_worker()
 
-        self.attrs['use_ps_gpu'] = self.user_defined_strategy.a_sync_configs[
+        attrs['use_ps_gpu'] = self.user_defined_strategy.a_sync_configs[
             "use_ps_gpu"]
-        self.attrs[
-            'lr_decay_steps'] = self.user_defined_strategy.a_sync_configs[
-                "lr_decay_steps"]
-        self.attrs['k_steps'] = self.user_defined_strategy.a_sync_configs[
-            "k_steps"]
-        self.attrs[
-            'launch_barrier'] = self.user_defined_strategy.a_sync_configs[
-                "launch_barrier"]
-
-        self.attrs['launch_barrier_flag'] = int(
+        attrs['lr_decay_steps'] = self.user_defined_strategy.a_sync_configs[
+            "lr_decay_steps"]
+        attrs['k_steps'] = self.user_defined_strategy.a_sync_configs["k_steps"]
+        attrs['launch_barrier'] = self.user_defined_strategy.a_sync_configs[
+            "launch_barrier"]
+
+        attrs['launch_barrier_flag'] = int(
             os.getenv("FLAGS_LAUNCH_BARRIER", "1"))
 
-        build_var_distributed(self.attrs)
+        build_var_distributed(attrs)
 
         # server 
-        self.attrs['_main_server'] = fluid.Program()
-        self.attrs['_startup_server'] = fluid.Program()
-        self.attrs['tensor_table'] = {}
+        attrs['_main_server'] = fluid.Program()
+        attrs['_startup_server'] = fluid.Program()
+        attrs['tensor_table'] = {}
 
-        self.pass_ctx._attrs = self.attrs
+        self.pass_ctx._attrs = attrs
 
     def _is_graph_out(self):
         return False

python/paddle/distributed/passes/ps_server_pass.py

@@ -115,11 +115,11 @@ def _apply_single_impl(self, main_program, startup_program, pass_ctx):
                           LRScheduler), "must be LRScheduler"
 
         ops = get_optimize_ops(attrs['origin_main_program'])
-        lr_decay_main_program, lr_decay_startup_program, lr_name = _get_lr_sheduler_program(
+        lr_decay_main_program, lr_decay_startup_program, lr_name = self._get_lr_sheduler_program(
             attrs['origin_main_program'].lr_sheduler, attrs['lr_decay_steps'])
-        _add_tensor_table(attrs, "@LR_DECAY_COUNTER@", lr_name,
-                          lr_decay_startup_program, lr_decay_main_program,
-                          "GlobalStepTable")
+        self._add_tensor_table(attrs, "@LR_DECAY_COUNTER@", lr_name,
+                               lr_decay_startup_program, lr_decay_main_program,
+                               "GlobalStepTable")
         return
 
 

python/paddle/distributed/ps/utils/ps_program_builder.py

@@ -41,7 +41,7 @@ def _optimize_programs(self):
         pass
 
     def _build_trainer_programs(self):
-        pass
+        raise NotImplementedError
 
     def _build_pserver_programs(self):
         is_sgd_adam = False
@@ -60,11 +60,13 @@ def _build_pserver_programs(self):
 
     def _build_programs(self):
         if self.attrs['is_worker']:
+            logger.info("start building trainer program")
             self._build_trainer_programs()
             fluid.framework.switch_startup_program(self.cloned_startup)
             self.loss.block.program = self.cloned_main
 
         elif self.attrs['is_server']:
+            logger.info("start building pserver program")
             self._build_pserver_programs()
             self.loss.block.program = self.attrs['_main_server']
             fluid.framework.switch_startup_program(self.attrs[
@@ -73,6 +75,7 @@ def _build_programs(self):
 
 class GeoPsProgramBuilder(PsProgramBuilder):  # 仅 CPU 模式
     def __init__(self, pass_ctx):
+        logger.info("start building geo-ps program")
         super(GeoPsProgramBuilder, self).__init__(pass_ctx)
         if self.ps_mode != DistributedMode.GEO:
             raise ValueError("ps mode: {} not matched {}",
@@ -92,6 +95,7 @@ def _build_trainer_programs(self):
 
 class CpuSyncPsProgramBuilder(PsProgramBuilder):
     def __init__(self, pass_ctx):
+        logger.info("start building cpu-sync-ps program")
         super(CpuSyncPsProgramBuilder, self).__init__(pass_ctx)
         if self.ps_mode == DistributedMode.GEO:
             raise ValueError("ps mode: {} not matched {}",
@@ -130,14 +134,17 @@ def _build_trainer_programs(self):
 
 class CpuAsyncPsProgramBuilder(CpuSyncPsProgramBuilder):
     def __init__(self, pass_ctx):
+        logger.info("start building cpu-async-ps program")
         super(CpuAsyncPsProgramBuilder, self).__init__(pass_ctx)
 
 
-class GpuPsProgramBuilder(PsProgramBuilder):  # 和 geo、sync、async 等模式无关 
+class GpuPsProgramBuilder(PsProgramBuilder):
     def __init__(self, pass_ctx):
+        logger.info("start building gpu-ps program")
         super(GpuPsProgramBuilder, self).__init__(pass_ctx)
 
     def _build_trainer_programs(self):
+
         add_lr_decay_table_pass = new_pass("add_lr_decay_table_pass",
                                            self.attrs)
         add_lr_decay_table_pass.apply([], [], self.pass_ctx)
@@ -152,7 +159,8 @@ def _build_trainer_programs(self):
         ps_gpu_pass.apply([self.cloned_main], [None], self.pass_ctx)
 
         ps_transpile_pass = new_pass("ps_transpile_pass", self.attrs)
-        ps_transpile_pass.apply([_main], [_startup], self.pass_ctx)
+        ps_transpile_pass.apply([self.cloned_main], [self.cloned_startup],
+                                self.pass_ctx)
 
         self.attrs['origin_main_program'] = self.cloned_main
         self.attrs['origin_startup_program'] = self.cloned_startup
@@ -165,6 +173,7 @@ def _build_trainer_programs(self):
 
 class HeterAsyncPsProgramBuilder(PsProgramBuilder):
     def __init__(self, pass_ctx):
+        logger.info("start building heter-async-ps program")
         super(HeterAsyncPsProgramBuilder, self).__init__(pass_ctx)
         if self.use_ps_gpu or self.ps_mode == DistributedMode.GEO or self.attrs[
                 'is_heter_ps_mode'] == False:

python/paddle/distributed/ps/utils/public.py

@@ -27,6 +27,10 @@
 import paddle.fluid.framework as framework
 import paddle.distributed.fleet as fleet
 
+#logging.basicConfig(
+#    format='%(levelname)s - %(asctime)s - %(pathname)s: %(lineno)s - %(message)s', level=logging.INFO)
+#logger = logging.getLogger(__name__)
+
 OP_NAME_SCOPE = "op_namescope"
 CLIP_OP_NAME_SCOPE = "gradient_clip"
 STEP_COUNTER = "@PS_STEP_COUNTER@"
@@ -43,6 +47,24 @@
 SPARSE_OP_TYPE_DICT = {"lookup_table": "W", "lookup_table_v2": "W"}
 
 
+def logger_config(log_path, logging_name):
+    logger = logging.getLogger(logging_name)
+    logger.setLevel(level=logging.DEBUG)
+    handler = logging.FileHandler(log_path, mode='a', encoding='UTF-8')
+    handler.setLevel(logging.INFO)
+    formatter = logging.Formatter(
+        '%(levelname)s - %(asctime)s - %(pathname)s: %(lineno)s - %(message)s')
+    handler.setFormatter(formatter)
+    console = logging.StreamHandler()
+    console.setLevel(logging.DEBUG)
+    logger.addHandler(handler)
+    logger.addHandler(console)
+    return logger
+
+
+logger = logger_config(log_path='/ps_log', logging_name='ps_log')
+
+
 class DistributedMode:
     SYNC = 0
     ASYNC = 1

python/paddle/fluid/tests/unittests/distributed_passes/ps_pass_test_base.py

@@ -36,20 +36,48 @@ def setUp(self):
     def tearDown(self):
         print('Ps tearDown...')
 
-    def ps_launch(self, config):
-        cmd = [
-            sys.executable,
-            "-u",
-        ] + [
-            "-m", "launch", "--log_dir", config['log_dir'], "--worker_num",
-            config['worker_num'], "--server_num", config['server_num'],
-            "../ps/ps_dnn_trainer.py", "-m", config['ps_mode_config'],
-            "--run_minimize", config['run_minimize'], "--run_single_pass",
-            config['run_single_pass'], "--debug_new_pass",
-            config['debug_new_pass'], "--debug_new_minimize",
-            config['debug_new_minimize'], "--applied_pass_name",
-            config['applied_pass_name']
-        ]
+    def ps_launch(self, config, ps_mode="cpu-ps"):
+        if ps_mode == "cpu-ps":
+            os.environ['WITH_DISTRIBUTE'] = 'ON'
+
+            cmd = [
+                sys.executable,
+                "-u",
+            ] + [
+                "-m", "launch", "--log_dir", config['log_dir'], "--worker_num",
+                config['worker_num'], "--server_num", config['server_num'],
+                "../ps/ps_dnn_trainer.py", "-m", config['ps_mode_config'],
+                "--run_minimize", config['run_minimize'], "--run_single_pass",
+                config['run_single_pass'], "--debug_new_pass",
+                config['debug_new_pass'], "--debug_new_minimize",
+                config['debug_new_minimize'], "--applied_pass_name",
+                config['applied_pass_name']
+            ]
+        elif ps_mode == "gpu-ps":
+            os.environ['FLAGS_LAUNCH_BARRIER'] = '0'
+            os.environ['PADDLE_PSERVER_NUMS'] = '1'
+            os.environ['PADDLE_TRAINERS_NUM'] = '1'
+            os.environ['POD_IP'] = '127.0.0.1'
+            os.environ['PADDLE_PSERVERS_IP_PORT_LIST'] = '127.0.0.1:29011'
+            os.environ['PADDLE_PORT'] = '29011'
+            os.environ['FLAGS_selected_gpus'] = '0,1,2,3,4,5,6,7'
+            # pserver
+            # os.environ['TRAINING_ROLE'] = 'PSERVER'
+
+            # trainer
+            os.environ['TRAINING_ROLE'] = 'TRAINER'
+            os.environ['PADDLE_TRAINER_ID'] = '0'
+
+            cmd = [
+                sys.executable, "-u", "../ps/ps_dnn_trainer.py", "-m",
+                config['ps_mode_config'], "--run_minimize",
+                config['run_minimize'], "--run_single_pass",
+                config['run_single_pass'], "--debug_new_pass",
+                config['debug_new_pass'], "--debug_new_minimize",
+                config['debug_new_minimize'], "--applied_pass_name",
+                config['applied_pass_name']
+            ]
+
         cmd = [shlex.quote(c) for c in cmd]
         prepare_python_path_and_return_module(__file__)
         exitcode = os.system(' '.join(cmd))

python/paddle/fluid/tests/unittests/distributed_passes/test_ps_trainer_pass.py

@@ -21,6 +21,7 @@
 
 import paddle
 from ps_pass_test_base import *
+from paddle.distributed.ps.utils.public import logger
 from paddle.fluid.tests.unittests.ps.ps_dnn_trainer import DnnTrainer
 
 
@@ -38,30 +39,43 @@ def init(self):
         self.config['applied_pass_name'] = ""
 
     def setUp(self):
-        print('TestPsTrainerPass setUp...')
+        pass
 
     def tearDown(self):
-        print('TestPsTrainerPass tearDown...')
+        pass
 
     def check(self):
         pass
 
-    def test_ps_optimizer_minimize(self):
+    def test_ps_optimizer_minimize_cpu(self):
         self.init()
         self.config['run_minimize'] = '1'
 
         self.config['debug_new_minimize'] = '0'
-        self.config['log_dir'] = "/log_old_minimize"
+        self.config['log_dir'] = "/cpu_log_old_minimize"
         remove_path_if_exists(self.config['log_dir'])
         self.ps_launch(self.config)
 
         self.config['debug_new_minimize'] = '1'
-        self.config['log_dir'] = "/log_new_minimize"
+        self.config['log_dir'] = "/cpu_log_new_minimize"
         remove_path_if_exists(self.config['log_dir'])
         self.ps_launch(self.config)
 
         self.check()
 
+    def test_ps_optimizer_minimize_gpu(self):
+        self.init()
+        self.config['run_minimize'] = '1'
+        self.config['ps_mode_config'] = "../ps/gpu_ps_config.yaml"
+
+        self.config['debug_new_minimize'] = '0'
+        self.ps_launch(self.config, "gpu-ps")
+
+        self.config['debug_new_minimize'] = '1'
+        self.ps_launch(self.config, "gpu-ps")
+
+        self.check()
+
     def test_append_send_ops_pass(self):
         self.init()
         self.config['run_single_pass'] = '1'
@@ -70,12 +84,12 @@ def test_append_send_ops_pass(self):
         self.config['debug_new_pass'] = '0'
         self.config['log_dir'] = "/log_old_" + self.config['applied_pass_name']
         remove_path_if_exists(self.config['log_dir'])
-        self.ps_launch(self.config)
+        self.ps_launch(self.config, "cpu-ps")
 
         self.config['debug_new_pass'] = '1'
         self.config['log_dir'] = "/log_new_" + self.config['applied_pass_name']
         remove_path_if_exists(self.config['log_dir'])
-        self.ps_launch(self.config)
+        self.ps_launch(self.config, "cpu-ps")
 
         self.check()
 
@@ -84,4 +98,5 @@ def test_distributed_ops_pass(self):
 
 
 if __name__ == '__main__':
+    remove_path_if_exists('/ps_log')
     unittest.main()

python/paddle/fluid/tests/unittests/ps/cpu_async_ps_config.yaml

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+# refer to PaddleRec/models/rank/dnn/benchmark.yaml
+
 hyper_parameters:
   optimizer:
     class: Adam