small fixes

Author: domenSedlar

GraphCreation/MyCode/pipeline/read_and_emit.py

@@ -22,18 +22,29 @@ def read_and_emit(self):
         state = {}
         current_t = None
 
-        # Read in very small chunks using pyarrow iter_batches
         for batch in pq_file.iter_batches(batch_size=100):
-            batch_df = batch.to_pandas()
-            for _, row in batch_df.iterrows():
+            # Convert to Python objects column-wise without Pandas
+            timestamps = batch.column("timestamp")
+            nodes = batch.column("node")
+            
+            # Convert once for the entire batch to Python scalars
+            # Avoid per-row overhead
+            ts_values = timestamps.to_pylist()
+            node_values = nodes.to_pylist()
+
+            # Precompute the row dicts once
+            # This avoids deepcopies of the same Arrow Row multiple times
+            all_rows = batch.to_pylist()
+
+            for i, ts in enumerate(ts_values):
                 if current_t is None:
-                    current_t = row["timestamp"]
-                elif current_t != row["timestamp"]:
+                    current_t = ts
+                elif ts != current_t:
                     self.buffer.put(deepcopy(state))
-                    state = {}
-                    current_t = row["timestamp"]
-                state[row["node"]] = deepcopy(row.to_dict())
+                    state.clear()
+                    current_t = ts
 
+                state[node_values[i]] = all_rows[i]
 
         self.buffer.put(state)
         self.buffer.put(None)

GraphCreation/MyCode/run_pipeline.py

@@ -3,6 +3,7 @@
 import threading
 import time
 from queue import Queue
+import cProfile
 
 sys.path.append(os.path.join(os.path.dirname(__file__), 'pipeline'))
 
@@ -48,4 +49,4 @@ def main():
 
 
 if __name__ == '__main__':
-    main() 
+    main()

SignalProcessing/common/logger.py

@@ -2,7 +2,7 @@
 import os
 
 class Logger:
-    def __init__(self, name=__name__, log_file='pipeline.log', log_dir='logs'):
+    def __init__(self, name=__name__, log_file='pipeline.log', log_dir='logs', rack='None'):
         self.logger_fake = logging.getLogger("idk")
 
         self.logger = logging.getLogger(name)
@@ -22,7 +22,7 @@ def __init__(self, name=__name__, log_file='pipeline.log', log_dir='logs'):
             self.log_dir = log_dir
 
             current_time = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M")
-            log_filename = f"{os.path.splitext(os.path.basename(name if name != '__main__' else 'main'))[0]}.log"
+            log_filename = f"{os.path.splitext(os.path.basename(name if name != '__main__' else 'main'))[0]}_{rack}.log"
             log_path = os.path.join(os.getcwd(), self.log_dir, current_time)
             os.makedirs(log_path, exist_ok=True)
             log_file = os.path.join(log_path, log_filename)

SignalProcessing/pipeline/file_reading/node_manager.py

@@ -249,7 +249,7 @@ def iterate_batches(self, limit_rows=None, stop_event=None, final_log_frequency=
                 break
             to_remove = []
 
-            logger.debug("reading data from nodes")
+            # logger.debug("reading data from nodes")
 
             for node_id in active_nodes:
                 if stop_event and stop_event.is_set():
@@ -296,13 +296,6 @@ def iterate_batches(self, limit_rows=None, stop_event=None, final_log_frequency=
                     break
 
         self.buffer.put(None)
-        # After processing, log expected vs actual rows per node
-        logger.info("NodeManager: Checking processed row counts per node...")
-        for node_id, man in self.node_managers.items():
-            if self.expected_rows[node_id] != self.processed_rows[node_id]:
-                logger.warning(f"Node {node_id}: Expected {self.expected_rows[node_id]} rows, processed {self.processed_rows[node_id]} rows!")
-            else:
-                logger.info(f"Node {node_id}: Processed all {self.processed_rows[node_id]} rows as expected.")
 
 def main():
     import argparse

SignalProcessing/pipeline/persist.py

@@ -24,7 +24,7 @@ class StatePersister:
     def __init__(self, input_queue, output_file='latest_state.json', batch_write_size=25):
         self.input_queue = input_queue
         self.output_file = output_file
-        self.batch_write_size = 25  # Reduced from 100 for more aggressive memory management
+        self.batch_write_size = batch_write_size
         self.state_buffer = []  # Buffer for batch writing
 
         self.writer = None
@@ -130,9 +130,9 @@ def run(self, timeout=0):
         batch_count = 0
         while True:
             if self.input_queue.empty():
-                logger.info("waiting, queue empty")
+                # logger.info("waiting, queue empty")
                 state_data = self.input_queue.get()
-                logger.info("continuing")
+                # logger.info("continuing")
             else:
                 state_data = self.input_queue.get()
 

SignalProcessing/pipeline/state_builder.py

@@ -9,7 +9,7 @@
 
 from common.memory_utils import log_memory_usage, force_memory_cleanup
 from common.logger import Logger
-logger = Logger(name=__name__.split('.')[-1], log_dir='logs').get_logger_real()
+logger = Logger(name=__name__.split('.')[-1], log_dir='logs').get_logger()
 
 class StateBuilder:
     def __init__(self, input_queue, output_queue, batch_size=5, max_queue_size=50):

SignalProcessing/run_pipeline.py

@@ -1,5 +1,6 @@
 import threading
 import queue
+import multiprocessing
 from pipeline.file_reading.node_manager import NodeManager
 from pipeline.changes.change_detector import ChangeLevelDetector
 from pipeline.state_builder import StateBuilder
@@ -10,15 +11,41 @@
 import sys
 
 from common.logger import Logger
-logger = Logger(name=__name__.split('.')[-1], log_dir='logs').get_logger_real()
+logger = Logger(name=__name__.split('.')[-1], log_dir='logs', rack=sys.argv[1]).get_logger_real()
+
+def node_manager_process(buffer_queue, stop_event, limit_nodes, limit_racks, temp_dir, rows_in_mem):
+    """NodeManager process function that can be pickled"""
+    node_manager = NodeManager(
+        buffer=buffer_queue, 
+        limit_nodes=limit_nodes, 
+        temp_dir=temp_dir, 
+        rows_in_mem=rows_in_mem, 
+        limit_racks=limit_racks
+    )
+    node_manager.iterate_batches(stop_event=stop_event, final_log_frequency=500)
+
+def change_detector_process(buffer_queue, change_queue, delta, clock):
+    """ChangeDetector process function that can be pickled"""
+    change_detector = ChangeLevelDetector(buffer_queue, change_queue, delta=delta, clock=clock)
+    change_detector.run()
+
+def state_builder_process(change_queue, state_queue):
+    """StateBuilder process function that can be pickled"""
+    state_builder = StateBuilder(change_queue, state_queue)
+    state_builder.run()
+
+def state_persister_process(state_queue, output_file):
+    """StatePersister process function that can be pickled"""
+    state_persister = StatePersister(state_queue, output_file=output_file)
+    state_persister.run()
 
 def run():
     limit_nodes = None
     limit_racks = int(sys.argv[1])
     delta=0.5
     clock=3
-    bq_max_size=300
-    rows_in_mem=300
+    rows_in_mem=1000
+    bq_max_size=2*rows_in_mem
     temp_dir_loc="E:/temp_parquet_files"
 
     vars_to_log = ['limit_nodes', 'limit_racks', 'delta', 'clock', 'bq_max_size', 'rows_in_mem']
@@ -30,48 +57,58 @@ def run():
     # Initialize memory monitor
     memory_monitor = MemoryMonitor(log_interval=50)
 
-        # Set up queues for each stage with size limits for backpressure
-    # Create queues with smaller sizes for more aggressive memory management
-    buffer_queue = queue.Queue(maxsize=bq_max_size)     # NodeManager → ChangeLevelDetector (reduced from 200)
-    change_queue = queue.Queue(maxsize=500)     # ChangeLevelDetector → StateBuilder (reduced from 100)
-    state_queue = queue.Queue(maxsize=500)     # StateBuilder → StatePersister (reduced from 500)
+    # Set up queues for each stage with size limits for backpressure
+    # Use multiprocessing.Queue for inter-process communication
+    buffer_queue = multiprocessing.Queue(maxsize=bq_max_size)     # NodeManager → ChangeLevelDetector
+    change_queue = multiprocessing.Queue(maxsize=bq_max_size)     # ChangeLevelDetector → StateBuilder
+    state_queue = multiprocessing.Queue(maxsize=bq_max_size)     # StateBuilder → StatePersister
 
-    output_file = f'./outputs/threaded_pipeline_state_{datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")}.parquet'
+    output_file = f'./outputs/threaded_pipeline_state_{datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")}_rack{limit_racks}.parquet'
 
     # Remove output file if it exists
     if os.path.exists(output_file):
         os.remove(output_file)
 
-    # Create the stop event
-    stop_event = threading.Event()
-
-    # Set up pipeline stages
-    node_manager = NodeManager(buffer=buffer_queue, limit_nodes=limit_nodes, temp_dir=temp_dir_loc, rows_in_mem=rows_in_mem, limit_racks=limit_racks)
-    change_detector = ChangeLevelDetector(buffer_queue, change_queue, delta=delta, clock=clock)
-    state_builder = StateBuilder(change_queue, state_queue)
-    state_persister = StatePersister(state_queue, output_file=output_file)
+    # Create the stop event (multiprocessing.Event)
+    stop_event = multiprocessing.Event()
 
-    # Create threads
-    threads = [
-        threading.Thread(target=lambda: node_manager.iterate_batches(stop_event=stop_event, final_log_frequency=5000), name="NodeManagerThread"),
-        threading.Thread(target=change_detector.run, name="ChangeLevelDetectorThread"),
-        threading.Thread(target=state_builder.run, name="StateBuilderThread"),
-        threading.Thread(target=state_persister.run, name="StatePersisterThread"),
+    # Create processes with function-based targets that can be pickled
+    processes = [
+        multiprocessing.Process(
+            target=node_manager_process, 
+            args=(buffer_queue, stop_event, limit_nodes, limit_racks, temp_dir_loc, rows_in_mem),
+            name="NodeManagerProcess"
+        ),
+        multiprocessing.Process(
+            target=change_detector_process, 
+            args=(buffer_queue, change_queue, delta, clock),
+            name="ChangeLevelDetectorProcess"
+        ),
+        multiprocessing.Process(
+            target=state_builder_process, 
+            args=(change_queue, state_queue),
+            name="StateBuilderProcess"
+        ),
+        multiprocessing.Process(
+            target=state_persister_process, 
+            args=(state_queue, output_file),
+            name="StatePersisterProcess"
+        ),
     ]
 
-    # Start threads
-    for t in threads:
-        t.start()
+    # Start processes
+    for p in processes:
+        p.start()
 
-    logger.info(f"Started all threads")
+    logger.info(f"Started all processes")
 
     try:
-        while any(t.is_alive() for t in threads):
+        while any(p.is_alive() for p in processes):
             # Monitor memory usage
             memory_monitor.check_memory("Pipeline-Main")
 
-            for t in threads:
-                t.join(timeout=0.5)
+            for p in processes:
+                p.join(timeout=0.5)
     except KeyboardInterrupt:
         logger.info("KeyboardInterrupt received! Setting stop event and sending sentinels.")
         stop_event.set()
@@ -80,8 +117,8 @@ def run():
             change_queue.put(None)
             state_queue.put(None)
         logger.info("Sentinels sent to all queues.")
-        for t in threads:
-            t.join(timeout=5)
+        for p in processes:
+            p.join(timeout=5)
         logger.info("Pipeline killed by user.")
 
 if __name__ == "__main__":