debug for simple graph

Author: tatoao pro

processor_pipeline/new_core/graph.py

@@ -70,14 +70,14 @@ def __init__(self, nodes: List[Node], edges: List[Edge], **kwargs):
         self.node_init_variables: Dict[str, Any] = {}
 
 
-    async def dynamic_fan_in_pipes_task(self, node: Node):
+    def dynamic_fan_in_pipes_task(self, node: Node):
         """
         Fan in pipes for a node.
         """
         precious_nodes = get_previous_nodes(self.nx_graph, node.processor_unique_name)
 
         if len(precious_nodes) == 0:
-            return
+            return asyncio.sleep(0)
 
         previous_input_pipes = [
             self.processor_pipes[previous_node.processor_unique_name].output_pipe 
@@ -87,12 +87,12 @@ async def dynamic_fan_in_pipes_task(self, node: Node):
             self._fan_in_pipes(
                 previous_input_pipes, 
                 self.processor_pipes[node.processor_unique_name].input_pipe), 
-            name=f"fan_in_pipes_task_{node.processor_unique_name}_to_{'='.join([previous_node.processor_unique_name for previous_node in precious_nodes])}"
+            name=f"fan_in_pipes_task__from_{'='.join([previous_node.processor_unique_name for previous_node in precious_nodes])} to {node.processor_unique_name}"
         )
 
-        await task
+        return task
 
-    async def dynamic_fan_out_pipes_task(self, node: Node):
+    def dynamic_fan_out_pipes_task(self, node: Node):
         """
         Fan out pipes for a node.
         """
@@ -112,20 +112,11 @@ async def dynamic_fan_out_pipes_task(self, node: Node):
                 ),
                 name=f"fan_out_pipes_task_{node.processor_unique_name}_to_{'='.join([next_node.processor_unique_name for next_node in next_nodes])}"
             )
-            await task
+            return task
         else:
-            await asyncio.sleep(0)
+            return asyncio.sleep(0)
 
 
-
-    async def add_processor_into_graph(self, node: Node):
-        self.initialize_node(node)
-        # merge input pipes
-        self.background_tasks.append(self.dynamic_fan_in_pipes_task(node))
-        # merge output pipes
-        self.background_tasks.append(self.dynamic_fan_out_pipes_task(node))
-
-
     def initialize(self):
 
         for node in self.nodes:
@@ -149,24 +140,33 @@ def initialize(self):
 
 
 
-        for node in self.nodes:
-            # merge input pipes
-            self.background_tasks.append(self.dynamic_fan_in_pipes_task(node))
-
-            # merge output pipes
-            self.background_tasks.append(self.dynamic_fan_out_pipes_task(node))
 
 
         root_node = get_root_nodes(self.nx_graph)[0]
 
         # Create graph's own input pipe (separate from root node's input pipe)
+    
+        async def root_task():
+            await self._fan_in_pipes(
+                [self.input_pipe], 
+                self.processor_pipes[root_node.processor_unique_name].input_pipe
+            )
+            self.logger.warning("Graph pipe into root input pipe task completed")
+
         graph_pipe_into_root_input_pipe_task = asyncio.create_task(
-            self._fan_out_pipes(
-                self.input_pipe, 
-                [self.processor_pipes[root_node.processor_unique_name].input_pipe]
-            ),
+            root_task(),
             name="graph_pipe_into_root_input_pipe_task"
         )
+        self.background_tasks.append(graph_pipe_into_root_input_pipe_task)
+
+
+        for node in self.nodes:
+            # merge input pipes
+            self.background_tasks.append(self.dynamic_fan_in_pipes_task(node))
+
+            # # merge output pipes
+            # self.background_tasks.append(self.dynamic_fan_out_pipes_task(node))
+
 
 
         # create the task for all leaf nodes
@@ -180,8 +180,7 @@ def initialize(self):
             name="leaf_nodes_output_pipe_task"
         )
 
-        self.background_tasks.insert(0, graph_pipe_into_root_input_pipe_task)
-        self.background_tasks.insert(0, leaf_nodes_output_pipe_task)
+        self.background_tasks.append(leaf_nodes_output_pipe_task)
         self.logger.info(f"Graph initialized: {self.processor_id}, {self.background_tasks}")
 
 
@@ -217,14 +216,13 @@ async def _fan_in_pipes(self, input_pipes: List[PipeInterface], output_pipe: Pip
         """Merge multiple input pipes into a single output pipe using asyncio.as_completed"""
         async def read_pipe_task(pipe):
             """Read all data from a single pipe"""
-            # import ipdb; ipdb.set_trace()
             async for message_id, data in pipe:
                 self.logger.info(f"Fan in pipe: {data}: from {pipe._pipe_id} to {output_pipe._pipe_id}")
                 if data is None:
-                    await output_pipe.put(data)
                     break
                 await output_pipe.put(data)
 
+
         try:
             # Create tasks for reading from each input pipe
             tasks = [asyncio.create_task(read_pipe_task(pipe), name=f"read_pipe_task_{pipe._pipe_id}") for pipe in input_pipes]
@@ -258,7 +256,9 @@ async def _fan_out_pipes(self, source_pipe: PipeInterface, output_pipes: List[Pi
             # import ipdb; ipdb.set_trace()
             for pipe in output_pipes:
                 try:
+                    self.logger.info(f"Fan out pipe: None: from {source_pipe._pipe_id} to {pipe._pipe_id}")
                     await pipe.put(None)
+
                 except Exception as e:
                     self.logger.error(f"Error signaling end-of-stream to pipe {getattr(pipe, '_pipe_id', 'unknown')}: {e}")
 
@@ -318,16 +318,21 @@ async def execute(self, data: Any, session_id: Optional[str] = None, *args, **kw
                 tasks.append(task)
 
             # Wait for all processors to complete
-            await asyncio.gather(*self.background_tasks)
             await asyncio.gather(*tasks)
+            
+            # Close input pipe to signal end-of-stream to background tasks
+            await self.input_pipe.close()
+            
+            self.logger.warning(f"Graph background tasks: {self.background_tasks}")
+            await asyncio.gather(*self.background_tasks)
 
 
         except Exception as e:
             self.logger.error(f"Error executing graph: {e}")
             self.logger.error(traceback.format_exc())
             caught_exception = e
         finally:
-            await self.input_pipe.close()
+            # Input pipe is already closed above, only close output pipe here
             await self.output_pipe.close()
             if caught_exception:
                 raise caught_exception
@@ -337,4 +342,4 @@ async def process(self, data: Any, *args, **kwargs) -> Any:
         """
         Process data.
         """
-        pass
+        pass

processor_pipeline/new_core/logger.py

@@ -0,0 +1,32 @@
+from loguru import logger
+
+# Configure loguru for better formatting and control
+def _ensure_extra_fields(record):
+    # Always provide a default session_id for formatting
+    record["extra"].setdefault("session_id", "-")
+    # Optional suffix to distinguish different inputs within the same session
+    record["extra"].setdefault("sid_suffix", "")
+    # Default object_name for formatting
+    record["extra"].setdefault("object_name", "UNKNOWN")
+    return True
+
+
+logger.remove()  # Remove default handler
+logger.add(
+    "logs/processor_pipeline.log",
+    rotation="10 MB",
+    retention="7 days",
+    level="INFO",
+    # Include session_id in all log lines
+    format="<green>{time:YYYY-MM-DD HH:mm:ss:SSS}</green> | sid={extra[session_id]}{extra[sid_suffix]} | <level>{level: <8}</level> | <cyan>{extra[object_name]}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
+    backtrace=True,
+    diagnose=True,
+    filter=_ensure_extra_fields,
+)
+logger.add(
+    lambda msg: print(msg, end=""),
+    level="DEBUG",
+    format="<green>{time:HH:mm:ss:SSS}</green> | sid={extra[session_id]}{extra[sid_suffix]} | <level>{level: <8}</level> | <cyan>{extra[object_name]}</cyan> - <level>{message}</level>",
+    colorize=True,
+    filter=_ensure_extra_fields,
+)

processor_pipeline/new_core/pipe.py

@@ -1,6 +1,6 @@
 import asyncio
 import uuid
-from loguru import logger
+from .logger import logger
 from .core_interfaces import PipeInterface, PipeMeta
 from typing import Any, Optional, Dict, AsyncGenerator, List, Tuple, Set
 from pydantic import BaseModel
@@ -37,10 +37,10 @@ def __init__(self, maxsize: int = -1, pipe_id: Optional[str] = None, logger=logg
             self._pipe_id = pipe_id
 
         self.queue = asyncio.Queue(maxsize)
-        self.logger = logger or logger.bind(name=self._pipe_id)
+        self.logger = logger or logger.bind(object_name=self._pipe_id)
         # ensure logger has default session_id to avoid formatting issues when not bound
         try:
-            self.logger = self.logger.bind(session_id="-")
+            self.logger = self.logger.bind(session_id="-", object_name=self._pipe_id)
         except Exception:
             pass
 
@@ -73,7 +73,7 @@ async def put(self, data: Any) -> None:
             return
 
         message_id = self.generate_random_message_id(data)
-        self.logger.debug(f"[{self._pipe_id}] [PUT] data: {data} with message_id: {message_id}")
+        self.logger.debug(f"[PUT] data: {data} with message_id: {message_id}")
 
         # Put data in main queue
         await self.queue.put((message_id, data))
@@ -88,9 +88,10 @@ async def get(self, timeout: Optional[float] = None) -> Any:
         self.statistics.historic_get_count += 1
         data = await self.queue.get()
         if data is None:
+            self.logger.debug("[GET] data is None, Closing pipe")
             return None
         message_id, data = data
-        self.logger.debug(f"[{self._pipe_id}] [GET] data: {data} with message_id: {message_id}")
+        self.logger.debug(f"[GET] data: {data} with message_id: {message_id}")
         return (message_id, data)
 
     async def __aiter__(self) -> AsyncGenerator[Any, None]:
@@ -123,7 +124,7 @@ async def register_observer(self, observer_id: Optional[str] = None) -> str:
                 raise ValueError(f"Observer {observer_id} already exists")
             self._observers[observer_id] = asyncio.Queue()
 
-        self.logger.debug(f"[{self._pipe_id}] Registered observer: {observer_id}")
+        self.logger.debug(f"[REGISTERED OBSERVER] {observer_id}")
         return observer_id
 
     async def unregister_observer(self, observer_id: str) -> None:
@@ -133,7 +134,7 @@ async def unregister_observer(self, observer_id: str) -> None:
                 # Close the observer queue
                 self._observers[observer_id].put_nowait(None)
                 del self._observers[observer_id]
-                self.logger.debug(f"[{self._pipe_id}] Unregistered observer: {observer_id}")
+                self.logger.debug(f"[UNREGISTERED OBSERVER] {observer_id}")
 
     async def peek_aiter(self, observer_id: Optional[str] = None) -> AsyncGenerator[Any, None]:
         """Create an async iterator for peeking without consuming from main queue"""
@@ -266,7 +267,7 @@ async def get(self, timeout: Optional[float] = None) -> Any:
         if data is None:
             return None
         message_id, data = data
-        self.logger.debug(f"[{self._pipe_id}] [GET] data: {data} with message_id: {message_id}")
+        self.logger.debug(f"[GET] data: {data} with message_id: {message_id}")
         return (message_id, data)
 
 

processor_pipeline/new_core/processor.py

@@ -9,35 +9,7 @@
 from .pipe import BufferPipe
 from pydantic import BaseModel, computed_field
 from threading import Thread
-from loguru import logger
-
-# Configure loguru for better formatting and control
-def _ensure_session_id(record):
-    # Always provide a default session_id for formatting
-    record["extra"].setdefault("session_id", "-")
-    # Optional suffix to distinguish different inputs within the same session
-    record["extra"].setdefault("sid_suffix", "")
-    return True
-
-logger.remove()  # Remove default handler
-logger.add(
-    "logs/processor_pipeline.log",
-    rotation="10 MB",
-    retention="7 days",
-    level="INFO",
-    # Include session_id in all log lines
-    format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | sid={extra[session_id]}{extra[sid_suffix]} | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
-    backtrace=True,
-    diagnose=True,
-    filter=_ensure_session_id,
-)
-logger.add(
-    lambda msg: print(msg, end=""),
-    level="DEBUG",
-    format="<green>{time:HH:mm:ss}</green> | sid={extra[session_id]}{extra[sid_suffix]} | <level>{level: <8}</level> | <cyan>{name}</cyan> - <level>{message}</level>",
-    colorize=True,
-    filter=_ensure_session_id,
-)
+from .logger import logger
 
 
 class ProcessorStatistics(BaseModel):
@@ -112,6 +84,7 @@ def __init__(self,
         # Get pipe types from meta (defaults are now set in ProcessorMeta)
         self.input_pipe_type = meta["input_pipe_type"]
         self.output_pipe_type = meta["output_pipe_type"]
+        self.background_tasks: List[asyncio.Task] = []
 
 
         self.semaphore = asyncio.Semaphore(max_concurrent)
@@ -122,10 +95,9 @@ def __init__(self,
         # statistics
         self._statistics = ProcessorStatistics()
 
-        self.background_tasks: List[asyncio.Task] = []
 
         # Debug logging for initialization
-        self.logger = logger
+        self.logger = logger.bind(object_name=self.processor_id)
         self.logger.debug(f"Initialized processor {self.processor_id} with output_strategy={self.output_strategy.value}, max_concurrent={max_concurrent}")
         self.logger.debug(f"Processor meta: {meta}")
 
@@ -138,17 +110,17 @@ def bind_session_id(self, session_id: str) -> None:
             self.session = {"session_id": session_id}
 
         # Bind logger for this processor
-        self.logger = self.logger.bind(session_id=session_id)
+        self.logger = self.logger.bind(session_id=session_id, object_name=self.processor_id)
 
         # Bind logger for pipes if available
         if getattr(self, "input_pipe", None) is not None and getattr(self.input_pipe, "logger", None) is not None:
             try:
-                self.input_pipe.logger = self.input_pipe.logger.bind(session_id=session_id)
+                self.input_pipe.logger = self.input_pipe.logger.bind(session_id=session_id, object_name=self.input_pipe._pipe_id)
             except Exception:
                 pass
         if getattr(self, "output_pipe", None) is not None and getattr(self.output_pipe, "logger", None) is not None:
             try:
-                self.output_pipe.logger = self.output_pipe.logger.bind(session_id=session_id)
+                self.output_pipe.logger = self.output_pipe.logger.bind(session_id=session_id, object_name=self.output_pipe._pipe_id)
             except Exception:
                 pass
 
@@ -406,10 +378,11 @@ async def execute(self, data: Any = None, session_id: Optional[str] = None, *arg
         self.bind_session_id(session_id)
 
         self.logger.info(f"Starting execute for {self.processor_id} with output strategy: {self.output_strategy.value}")
-        
+
         # Determine intake method based on data type
         if data is AsyncGenerator or inspect.isasyncgenfunction(data) or hasattr(data, '__aiter__'):
             generator = data
+
             async def intake_method():
                 item_count = 0
                 async for item in generator:
@@ -431,9 +404,7 @@ async def intake_method():
             intake_task = asyncio.create_task(asyncio.sleep(0))
         else:
             # a one time intake input
-            intake_task1 = asyncio.create_task(self.input_pipe.put(data))
-            intake_task2 = asyncio.create_task(self.input_pipe.put(None))
-            intake_task = asyncio.gather(intake_task1, intake_task2)
+            intake_task = asyncio.create_task(self.input_pipe.put(data))
 
         # Create main processing task based on output strategy
         if self.output_strategy == OutputStrategy.ASAP:
@@ -446,7 +417,7 @@ async def intake_method():
         else:
             raise ValueError(f"Invalid output strategy: {self.output_strategy}")
 
-        _, result, *_ = await asyncio.gather(intake_task, main_task, *self.background_tasks)
+        _, result = await asyncio.gather(intake_task, main_task)
         self.logger.info(f"Execute completed for {self.processor_id}")
         return result