fix(multiprocessing): remove chuncking for now

causy-dev · Oct 1, 2023 · ed37fda · ed37fda
1 parent 23aef65
commit ed37fda
Show file tree

Hide file tree

Showing 3 changed files with 5,044 additions and 11 deletions.
diff --git a/graph.py b/graph.py
@@ -314,6 +314,7 @@ def execute_pipeline_step(self, test_fn: IndependenceTestInterface):
 
         # run all combinations in parallel
         for result in iterator:
+            print(".")
             # for result in args:
             # result = unpack_run(result)
             if not isinstance(result, list):

diff --git a/independence_tests.py b/independence_tests.py
@@ -29,7 +29,7 @@
 
 class CalculateCorrelations(IndependenceTestInterface):
     NUM_OF_COMPARISON_ELEMENTS = 2
-    CHUNK_SIZE_PARALLEL_PROCESSING = 10000
+    CHUNK_SIZE_PARALLEL_PROCESSING = 1
     PARALLEL = False
 
     def test(
@@ -56,7 +56,7 @@ def test(
 
 class CorrelationCoefficientTest(IndependenceTestInterface):
     NUM_OF_COMPARISON_ELEMENTS = 2
-    CHUNK_SIZE_PARALLEL_PROCESSING = 10000
+    CHUNK_SIZE_PARALLEL_PROCESSING = 1
     PARALLEL = True
 
     def test(
@@ -93,7 +93,7 @@ def test(
 
 class PartialCorrelationTest(IndependenceTestInterface):
     NUM_OF_COMPARISON_ELEMENTS = 3
-    CHUNK_SIZE_PARALLEL_PROCESSING = 10000
+    CHUNK_SIZE_PARALLEL_PROCESSING = 1
     PARALLEL = True
 
     def test(
@@ -159,7 +159,7 @@ def test(
 
 class ExtendedPartialCorrelationTest(IndependenceTestInterface):
     NUM_OF_COMPARISON_ELEMENTS = ComparisonSettings(min=5, max=AS_MANY_AS_FIELDS)
-    CHUNK_SIZE_PARALLEL_PROCESSING = 1000
+    CHUNK_SIZE_PARALLEL_PROCESSING = 1
     PARALLEL = True
 
     def test(
@@ -208,7 +208,7 @@ def test(
 
 class UnshieldedTriplesTest(IndependenceTestInterface):
     NUM_OF_COMPARISON_ELEMENTS = 2
-    CHUNK_SIZE_PARALLEL_PROCESSING = 1000
+    CHUNK_SIZE_PARALLEL_PROCESSING = 1
 
     def test(
         self, nodes: Tuple[str], graph: BaseGraphInterface
@@ -255,7 +255,7 @@ def test(
 
 class ExtendedPartialCorrelationTest2(IndependenceTestInterface):
     NUM_OF_COMPARISON_ELEMENTS = ComparisonSettings(min=4, max=AS_MANY_AS_FIELDS)
-    CHUNK_SIZE_PARALLEL_PROCESSING = 100
+    CHUNK_SIZE_PARALLEL_PROCESSING = 1
     PARALLEL = True
 
     def test(
@@ -292,20 +292,20 @@ def test(
                 if i == k:
                     continue
 
-
                 # print(partial_correlation_coefficients[i][k])
                 try:
                     t, critical_t = get_t_and_critial_t(
                         sample_size,
                         nb_of_control_vars,
-                        (- precision_matrix[i][k] / math.sqrt(precision_matrix[i][i]*precision_matrix[k][k])),
+                        (
+                            -precision_matrix[i][k]
+                            / math.sqrt(precision_matrix[i][i] * precision_matrix[k][k])
+                        ),
                         self.threshold,
                     )
                 except ValueError:
                     # TODO: @sof fiugre out why this happens
-                    logger.debug(
-                        f"ValueError {i} {k} ({precision_matrix[i][k]})"
-                    )
+                    logger.debug(f"ValueError {i} {k} ({precision_matrix[i][k]})")
                     continue
 
                 if abs(t) < critical_t: