Fix dataloader error in static graph (#59878)

Wong4j · web-flow · commit d252541c8c1b · 2023-12-14T10:59:28.000+08:00
* remove .value() for static graph

* use logging instead of print
diff --git a/python/paddle/incubate/multiprocessing/reductions.py b/python/paddle/incubate/multiprocessing/reductions.py
@@ -84,15 +84,15 @@ def _rebuild_tensor(cls, lodtensor, metadata):
         size, stop_gradient = metadata
         tensor = paddle.base.core.eager.Tensor()
         if lodtensor._is_initialized():
-            tensor.value().get_tensor()._share_data_with(lodtensor)
+            tensor.get_tensor()._share_data_with(lodtensor)
         else:
             tensor = paddle.to_tensor([], dtype=lodtensor._dtype())
         tensor.stop_gradient = stop_gradient
     return tensor
 
 
 def _reduce_tensor(tensor):
-    lodtensor = tensor.value().get_tensor()
+    lodtensor = tensor.get_tensor()
 
     if not tensor.stop_gradient and not tensor.is_leaf:
         raise RuntimeError(
diff --git a/python/paddle/io/dataloader/dataloader_iter.py b/python/paddle/io/dataloader/dataloader_iter.py
@@ -622,7 +622,7 @@ def _thread_loop(self, legacy_expected_place):
                                 if isinstance(
                                     slot, (paddle.Tensor, core.eager.Tensor)
                                 ):
-                                    slot = slot.value().get_tensor()
+                                    slot = slot.get_tensor()
                                 elif not isinstance(slot, core.LoDTensor):
                                     tmp = core.LoDTensor()
                                     tmp.set(slot, core.CPUPlace())
diff --git a/test/legacy_test/test_multiprocess_dataloader_dynamic.py b/test/legacy_test/test_multiprocess_dataloader_dynamic.py
@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
 import sys
 import time
 import unittest
@@ -33,6 +34,8 @@
 from paddle.io import DataLoader
 from paddle.nn import Linear
 
+logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s")
+
 
 class SimpleFCNet(paddle.nn.Layer):
     def __init__(self):
@@ -74,8 +77,24 @@ def forward(self, image):
         return out
 
 
+def collate_batch(batch_list):
+    batch_size = len(batch_list)
+    image = np.stack([item[0] for item in batch_list], axis=0).astype('float32')
+    image = paddle.to_tensor(image).reshape([batch_size, -1])
+    label = np.stack([item[1] for item in batch_list], axis=0).astype('int64')
+    label = paddle.to_tensor(label).reshape([batch_size, -1])
+    return image, label
+
+
 class TestDygraphDataLoader(unittest.TestCase):
-    def run_main(self, num_workers, places, persistent_workers):
+    def run_main(
+        self,
+        num_workers,
+        places,
+        persistent_workers,
+        collate_fn,
+        use_shared_memory,
+    ):
         base.default_startup_program().random_seed = 1
         base.default_main_program().random_seed = 1
         with base.dygraph.guard(places[0]):
@@ -89,6 +108,8 @@ def run_main(self, num_workers, places, persistent_workers):
                 batch_size=BATCH_SIZE,
                 drop_last=True,
                 persistent_workers=persistent_workers,
+                collate_fn=collate_fn,
+                use_shared_memory=use_shared_memory,
             )
             assert len(dataloader) == int(SAMPLE_NUM / BATCH_SIZE)
 
@@ -117,36 +138,44 @@ def run_main(self, num_workers, places, persistent_workers):
             "step": step_list,
             "loss": np.array(loss_list),
         }
-        print("time cost", ret['time'], 'step_list', ret['step'])
+        logging.info(f"time cost {ret['time']} step_list {ret['step']}")
         return ret
 
     def test_main(self):
         for p in prepare_places():
             for persistent_workers in [False, True]:
-                results = []
-                for num_workers in [0, 2]:
-                    print(
-                        self.__class__.__name__,
-                        p,
-                        num_workers,
-                        persistent_workers,
-                    )
-                    sys.stdout.flush()
-                    ret = self.run_main(
-                        num_workers=num_workers,
-                        places=p,
-                        persistent_workers=persistent_workers,
-                    )
-                    results.append(ret)
-                diff = np.max(
-                    np.abs(results[0]['loss'] - results[1]['loss'])
-                    / np.abs(results[0]['loss'])
-                )
-                self.assertLess(diff, 1e-2)
+                for collate_fn in [None, collate_batch]:
+                    for use_shared_memory in [False, True]:
+                        results = []
+                        for num_workers in [0, 2]:
+                            logging.info(
+                                f"{self.__class__.__name__} {p} {num_workers} {persistent_workers} {collate_fn} {use_shared_memory}"
+                            )
+                            sys.stdout.flush()
+                            ret = self.run_main(
+                                num_workers=num_workers,
+                                places=p,
+                                persistent_workers=persistent_workers,
+                                collate_fn=collate_fn,
+                                use_shared_memory=use_shared_memory,
+                            )
+                            results.append(ret)
+                        diff = np.max(
+                            np.abs(results[0]['loss'] - results[1]['loss'])
+                            / np.abs(results[0]['loss'])
+                        )
+                        self.assertLess(diff, 1e-2)
 
 
 class TestDygraphDataLoaderWithBatchedDataset(TestDygraphDataLoader):
-    def run_main(self, num_workers, places, persistent_workers):
+    def run_main(
+        self,
+        num_workers,
+        places,
+        persistent_workers,
+        collate_fn,
+        use_shared_memory,
+    ):
         base.default_startup_program().random_seed = 1
         base.default_main_program().random_seed = 1
         with base.dygraph.guard(places[0]):
@@ -160,6 +189,8 @@ def run_main(self, num_workers, places, persistent_workers):
                 batch_size=None,
                 drop_last=True,
                 persistent_workers=persistent_workers,
+                collate_fn=None,
+                use_shared_memory=use_shared_memory,
             )
             assert len(dataloader) == int(SAMPLE_NUM / BATCH_SIZE)
 
@@ -188,7 +219,7 @@ def run_main(self, num_workers, places, persistent_workers):
             "step": step_list,
             "loss": np.array(loss_list),
         }
-        print("time cost", ret['time'], 'step_list', ret['step'])
+        logging.info(f"time cost {ret['time']} step_list {ret['step']}")
         return ret
 
 
