[Bug fix] Fix bugs for dualpipe when using PipelineDatasetPreprocessor (#76212)

Author: zhangyuqin1998

python/paddle/distributed/fleet/meta_parallel/dualpipev.py

@@ -34,6 +34,7 @@
 from .pipeline_parallel import (
     FakeMicroDataset,
     HybridParallelOptimizer,
+    PipelineDatasetPreprocessor,
     PipelineParallel,
 )
 from .pp_utils.batch_comm_helper import BatchCommHelper
@@ -634,6 +635,9 @@ def _wrap_data(self, data, phase):
         """
         for backward compatibility, wrap data to Fake FakeMicroDataset if it is of type list or tuple
         """
+        if isinstance(data, PipelineDatasetPreprocessor):
+            data = data()
+
         if (not isinstance(data, tuple)) and (not isinstance(data, list)):
             return data