test for hang

.github/workflows/nodes_ci.yml

@@ -22,13 +22,13 @@ jobs:
       matrix:
         os:
           - macos-latest
-          - ubuntu-latest
-          - windows-latest
+          # - ubuntu-latest
+          # - windows-latest
         python-version:
           - 3.9
           - "3.10"
           - "3.11"
-          - "3.12"
+          # - "3.12"
     steps:
       - name: Get PyTorch Channel
         shell: bash

test/nodes/test_map.py

@@ -108,54 +108,54 @@ def test_in_order_process(self):
     def test_out_of_order_process(self):
         self._test_map(False, "process")
 
-    @parameterized.expand(
-        itertools.product(
-            [0, 7, 14],
-            [True],  # TODO: define and fix in_order = False
-            [0, 1, 9],  # TODO: define and fix in_order = False
-        )
-    )
-    def test_save_load_state_thread(self, midpoint: int, in_order: bool, snapshot_frequency: int):
-        method = "thread"
-        batch_size = 6
-        n = 80
-        multiprocessing_context = None if IS_WINDOWS else "forkserver"
-        src = MockSource(num_samples=n)
-        node = Batcher(src, batch_size=batch_size, drop_last=False)
-        node = ParallelMapper(
-            node,
-            RandomSleepUdf(),
-            num_workers=4,
-            in_order=in_order,
-            method=method,
-            multiprocessing_context=multiprocessing_context,
-            snapshot_frequency=snapshot_frequency,
-        )
-        node = Prefetcher(node, prefetch_factor=2)
-        run_test_save_load_state(self, node, midpoint)
-
-    @parameterized.expand(
-        itertools.product(
-            [0, 7, 14],
-            [True],  # TODO: define and fix in_order = False
-            [0, 1, 9],  # TODO: define and fix in_order = False
-        )
-    )
-    def test_save_load_state_process(self, midpoint: int, in_order: bool, snapshot_frequency: int):
-        method = "process"
-        batch_size = 6
-        n = 80
-        multiprocessing_context = None if IS_WINDOWS else "forkserver"
-        src = MockSource(num_samples=n)
-        node = Batcher(src, batch_size=batch_size, drop_last=False)
-        node = ParallelMapper(
-            node,
-            RandomSleepUdf(),
-            num_workers=4,
-            in_order=in_order,
-            method=method,
-            multiprocessing_context=multiprocessing_context,
-            snapshot_frequency=snapshot_frequency,
-        )
-        node = Prefetcher(node, prefetch_factor=2)
-        run_test_save_load_state(self, node, midpoint)
+    # @parameterized.expand(
+    #     itertools.product(
+    #         [0, 7, 14],
+    #         [True],  # TODO: define and fix in_order = False
+    #         [0, 1, 9],  # TODO: define and fix in_order = False
+    #     )
+    # )
+    # def test_save_load_state_thread(self, midpoint: int, in_order: bool, snapshot_frequency: int):
+    #     method = "thread"
+    #     batch_size = 6
+    #     n = 80
+    #     multiprocessing_context = None if IS_WINDOWS else "forkserver"
+    #     src = MockSource(num_samples=n)
+    #     node = Batcher(src, batch_size=batch_size, drop_last=False)
+    #     node = ParallelMapper(
+    #         node,
+    #         RandomSleepUdf(),
+    #         num_workers=4,
+    #         in_order=in_order,
+    #         method=method,
+    #         multiprocessing_context=multiprocessing_context,
+    #         snapshot_frequency=snapshot_frequency,
+    #     )
+    #     node = Prefetcher(node, prefetch_factor=2)
+    #     run_test_save_load_state(self, node, midpoint)
+
+    # @parameterized.expand(
+    #     itertools.product(
+    #         [0, 7, 14],
+    #         [True],  # TODO: define and fix in_order = False
+    #         [0, 1, 9],  # TODO: define and fix in_order = False
+    #     )
+    # )
+    # def test_save_load_state_process(self, midpoint: int, in_order: bool, snapshot_frequency: int):
+    #     method = "process"
+    #     batch_size = 6
+    #     n = 80
+    #     multiprocessing_context = None if IS_WINDOWS else "forkserver"
+    #     src = MockSource(num_samples=n)
+    #     node = Batcher(src, batch_size=batch_size, drop_last=False)
+    #     node = ParallelMapper(
+    #         node,
+    #         RandomSleepUdf(),
+    #         num_workers=4,
+    #         in_order=in_order,
+    #         method=method,
+    #         multiprocessing_context=multiprocessing_context,
+    #         snapshot_frequency=snapshot_frequency,
+    #     )
+    #     node = Prefetcher(node, prefetch_factor=2)
+    #     run_test_save_load_state(self, node, midpoint)