rm FLAGS_enable_p2p_comm_opt

Author: liym27

python/paddle/distributed/passes/pass_utils.py

@@ -46,9 +46,6 @@
 ]
 
 logger = get_logger(logging.INFO)
-from paddle.distributed.utils.stream_utils import (
-    ExecutionStreamType,
-)
 
 
 # NOTE: Here stream is just a presentation with different name,
@@ -242,27 +239,6 @@ def _get_required_vars_of_program(program):
     """
     Get all vars in the program that are non-persistable and not in op's no_need_buffer.
     """
-    if paddle.base.framework.get_flags("FLAGS_enable_pir_api")[
-        "FLAGS_enable_pir_api"
-    ]:
-        return _get_required_vars_of_program_in_pir(program)
-    else:
-        return _get_required_vars_of_program_in_old_ir(program)
-
-
-def _get_required_vars_of_program_in_pir(program):
-    required_vars = set(program.list_vars())
-    no_need_buffer_vars = core.get_no_need_buffer_values(program)
-    required_vars -= no_need_buffer_vars
-    persistable_vars = set()
-    for var in required_vars:
-        if var.persistable:
-            persistable_vars.add(var)
-    required_vars -= persistable_vars
-    return required_vars
-
-
-def _get_required_vars_of_program_in_old_ir(program):
     required_vars = set()
     for block in program.blocks:
         for op in block.ops:
@@ -603,21 +579,7 @@ def _pir_overlap_send_recv(program):
             elif op.name() == "pd_op.recv_v2":
                 op.set_bool_attr("dynamic_shape", False)
                 op.set_bool_attr("use_calc_stream", True)
-                if os.getenv("FLAGS_enable_p2p_comm_opt", 0) in [
-                    'True',
-                    'true',
-                    '1',
-                ]:
-                    # and os.getenv("FLAGS_1f1b", 0) in [
-                    #     'True',
-                    #     'true',
-                    #     '1',
-                    # ]:
-                    op.set_execution_stream(
-                        ExecutionStreamType.DefaultStream.value
-                    )
-                else:
-                    op.set_execution_stream("recv_stream")
+                op.set_execution_stream("recv_stream")
                 op.set_scheduling_priority(0)
 
 

python/paddle/distributed/passes/pipeline_scheduler_pass/pipeline_1f1b.py

@@ -13,7 +13,6 @@
 # limitations under the License.
 
 import logging
-import os
 
 import paddle
 from paddle.base import core
@@ -28,7 +27,6 @@
     AutoParallelStreamType,
     _add_event_dependency,
     _program_for_fthenb_and_1f1b,
-    _split_program_into_forward_backward_optimize,
     forward_complete_op_role,
     split_program,
 )
@@ -179,14 +177,6 @@ def _backward_forward_overlap(self, backward_program, forward_program):
         )
 
     def _create_job_list(self):
-
-        if os.getenv("FLAGS_enable_p2p_comm_opt", 0) in [
-            'True',
-            'true',
-            '1',
-        ]:
-            return self._create_job_list_send_recv()
-
         num_micro_batches = self.get_attr("num_micro_batches")
         pp_stage = self.get_attr("pp_stage")
         pp_degree = self.get_attr("pp_degree")
@@ -201,18 +191,24 @@ def _create_job_list(self):
 
         forward_micro_batch_id = 0
         for i in range(micro_batch_in_warmup):
+            recv_fwd_job = core.Job(RECV_FORWARD)
+            recv_fwd_job.set_micro_batch_id(forward_micro_batch_id)
+            job_list.append(recv_fwd_job)
+
             forward_job = core.Job(FORWARD)
             forward_job.set_micro_batch_id(forward_micro_batch_id)
             job_list.append(forward_job)
             forward_micro_batch_id += 1
 
         backward_micro_batch_id = 0
+        jobs_in_stable_phase = [BACKWARD, RECV_FORWARD, SEND_BACKWARD, FORWARD]
         for i in range(micro_batch_in_1f1b):
-            for job_type in self.jobs_in_stable_phase:
+            for job_type in jobs_in_stable_phase:
                 job = core.Job(job_type)
                 micro_batch_id = (
                     forward_micro_batch_id
                     if job_type.startswith(FORWARD)
+                    or job_type.startswith(RECV_FORWARD)
                     else backward_micro_batch_id
                 )
                 job.set_micro_batch_id(micro_batch_id)
@@ -224,6 +220,11 @@ def _create_job_list(self):
             backward_job = core.Job(BACKWARD)
             backward_job.set_micro_batch_id(backward_micro_batch_id)
             job_list.append(backward_job)
+
+            send_bwd_job = core.Job(SEND_BACKWARD)
+            send_bwd_job.set_micro_batch_id(backward_micro_batch_id)
+            job_list.append(send_bwd_job)
+
             backward_micro_batch_id += 1
 
         opt_job = core.Job(OPT)
@@ -362,20 +363,9 @@ def _partial_pir_programs(self, program):
             not enable_send_recv_overlap
         ), "PIR does not support 1F1B with enable_send_recv_overlap yet."
 
-        if os.getenv("FLAGS_enable_p2p_comm_opt", 0) in [
-            'True',
-            'true',
-            '1',
-        ]:
-            types = [RECV_FORWARD, FORWARD, BACKWARD, SEND_BACKWARD, OPT]
-            prog_splitter = ProgramSplitter(program, types)
-            sub_program_list = prog_splitter._split_programs()
-
-        else:
-            types = [FORWARD, BACKWARD, OPT]
-            sub_program_list = _split_program_into_forward_backward_optimize(
-                program, enable_send_recv_overlap
-            )
+        types = [RECV_FORWARD, FORWARD, BACKWARD, SEND_BACKWARD, OPT]
+        prog_splitter = ProgramSplitter(program, types)
+        sub_program_list = prog_splitter._split_programs()
 
         for i in range(len(types)):
             logger.debug(
@@ -406,62 +396,6 @@ def is_comm_op_valid_to_overlap(self, op):
             == AutoParallelStreamType.CALC_STREAM.value
         )
 
-    def _create_job_list_send_recv(self):
-        num_micro_batches = self.get_attr("num_micro_batches")
-        pp_stage = self.get_attr("pp_stage")
-        pp_degree = self.get_attr("pp_degree")
-
-        job_list = []
-        assert (
-            pp_degree <= num_micro_batches
-        ), "Num of micro batches should larger than or equal to pp degree."
-
-        micro_batch_in_warmup = pp_degree - pp_stage
-        micro_batch_in_1f1b = num_micro_batches - micro_batch_in_warmup
-
-        forward_micro_batch_id = 0
-        for i in range(micro_batch_in_warmup):
-            recv_fwd_job = core.Job(RECV_FORWARD)
-            recv_fwd_job.set_micro_batch_id(forward_micro_batch_id)
-            job_list.append(recv_fwd_job)
-
-            forward_job = core.Job(FORWARD)
-            forward_job.set_micro_batch_id(forward_micro_batch_id)
-            job_list.append(forward_job)
-            forward_micro_batch_id += 1
-
-        backward_micro_batch_id = 0
-        jobs_in_stable_phase = [BACKWARD, RECV_FORWARD, SEND_BACKWARD, FORWARD]
-        for i in range(micro_batch_in_1f1b):
-            for job_type in jobs_in_stable_phase:
-                job = core.Job(job_type)
-                micro_batch_id = (
-                    forward_micro_batch_id
-                    if job_type.startswith(FORWARD)
-                    or job_type.startswith(RECV_FORWARD)
-                    else backward_micro_batch_id
-                )
-                job.set_micro_batch_id(micro_batch_id)
-                job_list.append(job)
-            forward_micro_batch_id += 1
-            backward_micro_batch_id += 1
-
-        for i in range(micro_batch_in_warmup):
-            backward_job = core.Job(BACKWARD)
-            backward_job.set_micro_batch_id(backward_micro_batch_id)
-            job_list.append(backward_job)
-
-            send_bwd_job = core.Job(SEND_BACKWARD)
-            send_bwd_job.set_micro_batch_id(backward_micro_batch_id)
-            job_list.append(send_bwd_job)
-
-            backward_micro_batch_id += 1
-
-        opt_job = core.Job(OPT)
-        opt_job.set_micro_batch_id(0)
-        job_list.append(opt_job)
-        return job_list
-
 
 class ProgramSplitter: