### Minimal Task Pipeline Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/INDEX.md

Demonstrates the basic setup for creating and running a simple task pipeline using GNode and GPipeline. Ensure PyCGraph is imported.

```python
from PyCGraph import GPipeline, GNode, CStatus

class Task(GNode):
    def run(self):
        return CStatus()

pipeline = GPipeline()
pipeline.registerGElement(Task(), set(), "task")
pipeline.process()
```

--------------------------------

### Complete GAdapter Pipeline Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAdapter.md

A comprehensive example showcasing the construction of a complex pipeline using various GAdapter types including GNode, GFunction, and GFence.

```python
from PyCGraph import (
    GPipeline, GNode, GFence, GFunction, 
    CFunctionType, CStatus
)
import time
import json

class DataProcessor(GNode):
    def run(self):
        print("Processing data...")
        time.sleep(1)
        return CStatus()

def save_config():
    """Initialize configuration"""
    print("Loading config...")
    with open("config.json", "w") as f:
        json.dump({"status": "ready"}, f)
    return CStatus()

def validate_results():
    """Validate final results"""
    print("Validating...")
    return CStatus()

# Build complex pipeline with adapters
pipeline = GPipeline()

# Use function for simple config init
config_init = GFunction()
config_init.setFunction(CFunctionType.INIT, save_config)

# Independent data processors
processor1 = DataProcessor()
processor2 = DataProcessor()
processor3 = DataProcessor()

# Fence ensures all processors complete
fence = GFence()

# Use function for final validation
validator = GFunction()
validator.setFunction(CFunctionType.RUN, validate_results)

# Register elements
pipeline.registerGElement(config_init, set(), "config")
pipeline.registerGElement(processor1, {config_init}, "processor1")
pipeline.registerGElement(processor2, {config_init}, "processor2")
pipeline.registerGElement(processor3, {config_init}, "processor3")
pipeline.registerGElement(fence, {processor1, processor2, processor3}, "fence")
pipeline.registerGElement(validator, {fence}, "validator")

# Execute
pipeline.process()
```

--------------------------------

### Complete Example: Simple Logging Event

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GEvent.md

Demonstrates a complete pipeline setup with a custom TimestampEvent that logs the current time when triggered. The event is registered alongside a WorkTask.

```python
from PyCGraph import GPipeline, GNode, GEvent, CStatus
from datetime import datetime

class TimestampEvent(GEvent):
    def trigger(self):
        print(f"[{datetime.now()}] Pipeline event triggered")
        return CStatus()

class WorkTask(GNode):
    def run(self):
        print("Working...")
        return CStatus()

pipeline = GPipeline()

task = WorkTask()
event = TimestampEvent()

pipeline.registerGElement(task, set(), "work")
pipeline.addGEvent(event, "timestamp")

pipeline.process()
```

--------------------------------

### Complete Pipeline Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GGroup.md

Demonstrates how to construct a complex data processing pipeline with sequential and parallel stages using GPipeline, GCluster, GRegion, and custom GNode subclasses. This example shows task registration and pipeline execution.

```python
from PyCGraph import GPipeline, GNode, GCluster, GRegion, CStatus
import time

class FetchTask(GNode):
    def run(self):
        print("Fetching data...")
        time.sleep(1)
        return CStatus()

class ProcessTask(GNode):
    def run(self):
        print("Processing...")
        time.sleep(0.5)
        return CStatus()

class AnalyzeTask(GNode):
    def run(self):
        print("Analyzing...")
        time.sleep(0.5)
        return CStatus()

class SaveTask(GNode):
    def run(self):
        print("Saving...")
        return CStatus()

# Create pipeline with groups
pipeline = GPipeline()

# Sequential fetching and initial processing
cluster = GCluster()

# Parallel analysis
region = GRegion()

fetch = FetchTask()
process = ProcessTask()
analyze1 = AnalyzeTask()
analyze2 = AnalyzeTask()
save = SaveTask()

pipeline.registerGElement(cluster, set(), "sequential", 1)
pipeline.registerGElement(region, {cluster}, "parallel", 1)

pipeline.registerGElement(fetch, set(), "fetch", 1)
pipeline.registerGElement(process, {fetch}, "process", 1)
pipeline.registerGElement(analyze1, set(), "analyze1", 1)
pipeline.registerGElement(analyze2, set(), "analyze2", 1)
pipeline.registerGElement(save, {region}, "save", 1)

# Execute
pipeline.process()
```

--------------------------------

### Install pycgraph from Source Directly

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Install pycgraph directly from its source code using setup.py. Requires Python 3, pybind11, and setuptools.

```shell
$ git clone https://github.com/ChunelFeng/CGraph.git
$ cd CGraph/python                       # 进入对应文件夹
$ python3 setup.py install               # 安装 pycgraph
$ python3 tutorial/T00-HelloCGraph.py    # 运行T00-HelloCGraph.py，并且在终端输出 Hello, pycgraph.
```

--------------------------------

### Install pycgraph using uv

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Install pycgraph using the 'uv' package manager, creating a dedicated environment. This example specifically targets Linux platforms.

```shell
$ uv init cgraph_env              # 通过uv 创建一个名为 cgraph_env 的环境
$ cd cgraph_env 
$ uv add "pycgraph @ git+https://github.com/ChunelFeng/CGraph.git@main#subdirectory=python" --marker "sys_platform == 'linux'"    # 添加 pycgraph 依赖
$ python3 -c "import pycgraph"    # 验证 pycgraph 成功安装
```

--------------------------------

### Example: Full Pipeline Process with Custom Task

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipeline.md

Demonstrates how to define a custom task inheriting from GNode, register it with a GPipeline, and then execute the pipeline using the `process` method. This example shows a complete, self-contained pipeline execution.

```python
from PyCGraph import GPipeline, GNode, CStatus

class MyTask(GNode):
    def run(self):
        print("Task executing")
        return CStatus()

pipeline = GPipeline()
task = MyTask()
pipeline.registerGElement(task, set(), "task1")
status = pipeline.process(runTimes=1)
print(f"Status: {status.getCode()}")
```

--------------------------------

### Build and Install pycgraph from Source (Wheel)

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Install pycgraph from source by building a wheel file. Requires Python 3, pybind11, and setuptools. This method is recommended for source installations.

```shell
$ git clone https://github.com/ChunelFeng/CGraph.git
$ cd CGraph/python                       # 进入对应文件夹
$ python3 setup.py bdist_wheel           # 生成 *.whl 安装包
$ pip3 install dist/pycgraph-xxx.whl     # 安装 *.whl 安装包，xxx 信息与版本和系统有关
$ python3 tutorial/T00-HelloCGraph.py    # 运行T00-HelloCGraph.py，并且在终端输出 Hello, pycgraph.
```

--------------------------------

### System Resource Monitor Daemon Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GDaemon.md

A comprehensive example demonstrating a SystemMonitor daemon that tracks maximum CPU and memory usage while a HeavyTask node runs in the pipeline. The daemon checks every 500ms.

```python
from PyCGraph import GPipeline, GNode, GDaemon, CStatus
import time
import psutil

class SystemMonitor(GDaemon):
    def __init__(self):
        super().__init__()
        self.max_cpu = 0
        self.max_memory = 0
    
    def run(self):
        cpu = psutil.cpu_percent(interval=0.1)
        memory = psutil.virtual_memory().percent
        
        if cpu > self.max_cpu:
            self.max_cpu = cpu
        if memory > self.max_memory:
            self.max_memory = memory
        
        print(f"CPU: {cpu:.1f}% (max: {self.max_cpu:.1f}%), "
              f"Memory: {memory:.1f}% (max: {self.max_memory:.1f}%)")
        return CStatus()

class HeavyTask(GNode):
    def run(self):
        print("Performing heavy computation...")
        time.sleep(2)
        return CStatus()

pipeline = GPipeline()

task = HeavyTask()
monitor = SystemMonitor()

pipeline.registerGElement(task, set(), "task")
pipeline.addGDaemon(monitor, 500)  # Check every 500ms

pipeline.process()
```

--------------------------------

### Compile and Run CGraph on Gitpod

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Compile CGraph and run a tutorial example within an online Gitpod environment. Requires installing CMake first.

```shell
$ sudo apt-get install cmake -y           # 安装cmake
$ ./CGraph-build.sh                       # 编译CGraph工程，生成的内容在同级/build/文件夹中
$ ./build/tutorial/T00-HelloCGraph        # 运行T00-HelloCGraph，并且在终端输出 Hello, CGraph.
```

--------------------------------

### Define Example List in CMake

Source: https://github.com/chunelfeng/cgraph/blob/main/example/CMakeLists.txt

Defines a list of example directories to be built. This variable is used later in the CMake script to add executables for each example.

```cmake
set(CGRAPH_EXAMPLE_LIST
        E01-AutoPilot
        E02-MockGUI
        E03-ThirdFlow
        E04-MapReduce
        E05-HttpServer
        )
```

--------------------------------

### GFence Usage Example with Producers and Consumer

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAdapter.md

Demonstrates how to use GFence to ensure two producer tasks complete before a consumer task begins execution.

```python
from PyCGraph import GPipeline, GNode, GFence, CStatus
import time

class ProducerTask(GNode):
    def run(self):
        print("Producing data...")
        time.sleep(1)
        return CStatus()

class ConsumerTask(GNode):
    def run(self):
        print("Consuming data...")
        return CStatus()

pipeline = GPipeline()

producer1 = ProducerTask()
producer2 = ProducerTask()
consumer = ConsumerTask()
fence = GFence()

pipeline.registerGElement(producer1, set(), "producer1")
pipeline.registerGElement(producer2, set(), "producer2")
pipeline.registerGElement(fence, {producer1, producer2}, "fence")
pipeline.registerGElement(consumer, {fence}, "consumer")

# Both producers complete before fence, consumer waits for fence
pipeline.process()
```

--------------------------------

### init()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

Lifecycle hook called once before the first `run()` execution. Ideal for one-time setup tasks like opening files or establishing database connections.

```APIDOC
## init()

### Description
Called once before the first `run()` execution. Use for one-time setup (opening files, database connections, etc.).

### Returns
`CStatus` — Initialization status.

### Note
Called sequentially in priority order (level-based), not in parallel.

### Example
```python
class DatabaseNode(GNode):
    def init(self):
        print("Initializing database connection")
        self.db_conn = establish_connection()
        return CStatus()
```
```

--------------------------------

### Add Executable for Each Example in CMake

Source: https://github.com/chunelfeng/cgraph/blob/main/example/CMakeLists.txt

Iterates through the defined example list and adds each one as an executable target. It links the main CGraph library objects and the example's source file.

```cmake
foreach(example ${CGRAPH_EXAMPLE_LIST})
    add_executable(${example}
            $<TARGET_OBJECTS:CGraph>
            ${example}.cpp
            )
endforeach()
```

--------------------------------

### Complete GNode Pipeline Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

Demonstrates building and executing a pipeline with custom GNode subclasses for download, processing, and saving tasks. Requires PyCGraph library.

```python
import time
from PyCGraph import GPipeline, GNode, CStatus

class DownloadTask(GNode):
    def init(self):
        print("Initializing download")
        return CStatus()
    
    def run(self):
        print("Downloading data...")
        time.sleep(1)
        return CStatus()
    
    def destroy(self):
        print("Cleaning up downloads")
        return CStatus()

class ProcessTask(GNode):
    def run(self):
        print("Processing downloaded data...")
        time.sleep(0.5)
        return CStatus()

class SaveTask(GNode):
    def run(self):
        print("Saving results...")
        return CStatus()

# Build pipeline
pipeline = GPipeline()
download = DownloadTask()
process = ProcessTask()
save = SaveTask()

pipeline.registerGElement(download, set(), "download", 1)
pipeline.registerGElement(process, {download}, "process", 1)
pipeline.registerGElement(save, {process}, "save", 1)

# Execute
pipeline.process(runTimes=1)
```

--------------------------------

### Install pycgraph with pip

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Install the pycgraph Python package directly using pip. Includes a command to verify the installation.

```shell
$ pip3 install pycgraph                  # 直接通过 pip3 安装
$ python3 -c "import pycgraph"           # 验证 pycgraph 成功安装
```

--------------------------------

### prepareRun()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

Lifecycle hook called before each `run()` execution, after `init()`. Suitable for per-execution setup that is quicker than full initialization.

```APIDOC
## prepareRun()

### Description
Called before each `run()` execution but after `init()`. Useful for per-execution setup that's faster than full initialization.

### Returns
`CStatus` — Preparation status.

### Example
```python
class StatefulNode(GNode):
    def prepareRun(self):
        print("Preparing for this run iteration")
        return CStatus()
```
```

--------------------------------

### Initialize Node Resources with init()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

Use the `init()` lifecycle hook for one-time setup tasks before the first `run()` execution, such as establishing database connections or opening files. It must return a `CStatus`.

```python
class DatabaseNode(GNode):
    def init(self):
        print("Initializing database connection")
        self.db_conn = establish_connection()
        return CStatus()
```

--------------------------------

### GCoordinator Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAdapter.md

Demonstrates how GCoordinator ensures that a set of tasks execute as an atomic unit within a pipeline.

```python
coordinator = GCoordinator()
task1 = Task()
task2 = Task()
task3 = Task()

# Coordinator ensures all three execute as an atomic unit
pipeline.registerGElement(coordinator, set(), "batch_processor")
```

--------------------------------

### Create GPipelineManager Instance

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Instantiate a GPipelineManager to manage multiple pipelines. This is the starting point for using the manager.

```python
from PyCGraph import GPipelineManager

manager = GPipelineManager()
```

--------------------------------

### Define Tutorial Targets

Source: https://github.com/chunelfeng/cgraph/blob/main/tutorial/CMakeLists.txt

Defines a list of tutorial targets to be built. These correspond to different examples within the CGraph tutorial.

```cmake
set(CGRAPH_TUTORIAL_LIST
         T00-HelloCGraph
         T01-Simple
         T02-Cluster
         T03-Region
         T04-Complex
         T05-Param
         T06-Condition
         T07-MultiPipeline
         T08-Template
         T09-Aspect
         T10-AspectParam
         T11-Singleton
         T12-Function
         T13-Daemon
         T14-Hold
         T15-ElementParam
         T16-MessageSendRecv
         T17-MessagePubSub
         T18-Event
         T19-Cancel
         T20-Suspend
         T21-MultiCondition
         T22-Timeout
         T23-Some
         T24-Fence
         T25-Coordinator
         T26-Mutable
         T27-Trim
         T28-Stage
         T29-Storage
        )
```

--------------------------------

### Compile and Run CGraph with Xmake

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Build CGraph projects and run a tutorial example using Xmake. Supports Linux, MacOS, and Windows.

```shell
$ git clone https://github.com/ChunelFeng/CGraph.git
$ cd CGraph
$ xmake build                             # 编译 tutorial 中的所有项目
$ xmake run T00-HelloCGraph               # 运行 T00-HelloCGraph
```

--------------------------------

### Prepare for Each Run Iteration with prepareRun()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

The `prepareRun()` lifecycle hook is called before each `run()` execution. Use it for per-iteration setup that is quicker than full initialization.

```python
class StatefulNode(GNode):
    def prepareRun(self):
        print("Preparing for this run iteration")
        return CStatus()
```

--------------------------------

### Request Counter Daemon Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GDaemon.md

This example features a RequestCounter daemon that tracks the number of requests processed by RequestHandler nodes. The counter is incremented by handlers and reported every 1000ms.

```python
from PyCGraph import GPipeline, GNode, GDaemon, CStatus
from threading import Lock
import time

class RequestCounter(GDaemon):
    def __init__(self):
        super().__init__()
        self.count = 0
        self.lock = Lock()
    
    def increment(self):
        with self.lock:
            self.count += 1
    
    def run(self):
        with self.lock:
            print(f"Requests processed: {self.count}")
            current = self.count
        return CStatus()

class RequestHandler(GNode):
    def __init__(self, counter):
        super().__init__()
        self.counter = counter
    
    def run(self):
        print("Handling request...")
        time.sleep(0.2)
        self.counter.increment()
        return CStatus()

pipeline = GPipeline()

counter = RequestCounter()
handlers = [RequestHandler(counter) for _ in range(3)]

for i, handler in enumerate(handlers):
    pipeline.registerGElement(handler, set(), f"handler_{i}")

pipeline.addGDaemon(counter, 1000)  # Report every 1000ms

pipeline.process()
```

--------------------------------

### before() Hook

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAspect.md

Called immediately before the element's `run()` method. Use for setup, logging, or preconditions.

```APIDOC
## before()

### Description
Called immediately before the element's `run()` method. Use for setup, logging, or preconditions.

### Returns
`CStatus` — Status; non-OK status prevents element execution.

### Example
```python
class MeasurementAspect(GAspect):
    def before(self):
        import time
        self.start_time = time.time()
        print("Element execution starting")
        return CStatus()
```
```

--------------------------------

### GFunction Usage Example with Inline Callables

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAdapter.md

Demonstrates using GFunction with inline lambda functions for INIT, RUN, and DESTROY stages to create lightweight tasks.

```python
from PyCGraph import GPipeline, GFunction, CFunctionType, CStatus
import time

pipeline = GPipeline()

# Create inline function-based tasks
init_func = GFunction()
init_func.setFunction(CFunctionType.INIT, lambda: (print("Initializing"), CStatus()))

run_func = GFunction()
run_func.setFunction(CFunctionType.RUN, lambda: (print("Running task"), time.sleep(1), CStatus()))

cleanup_func = GFunction()
cleanup_func.setFunction(CFunctionType.DESTROY, lambda: (print("Cleaning up"), CStatus()))

pipeline.registerGElement(init_func, set(), "init_task")
pipeline.registerGElement(run_func, {init_func}, "run_task")
pipeline.registerGElement(cleanup_func, {run_func}, "cleanup_task")

pipeline.process()
```

--------------------------------

### Compile and Run CGraph with CMake

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Build CGraph using CMake and run a tutorial example. Supports Linux, MacOS, and Windows.

```shell
$ git clone https://github.com/ChunelFeng/CGraph.git
$ cd CGraph
$ cmake . -Bbuild
$ cd build && make -j8
$ ./tutorial/T00-HelloCGraph              # 运行 T00-HelloCGraph
```

--------------------------------

### Complete Example: Metrics Collection Event

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GEvent.md

Illustrates a pipeline with a MetricsEvent that records the time of each trigger. The event is associated with a MonitoredTask that simulates work.

```python
from PyCGraph import GPipeline, GNode, GEvent, CStatus
import time

class MetricsEvent(GEvent):
    def __init__(self):
        super().__init__()
        self.event_times = []
    
    def trigger(self):
        self.event_times.append(time.time())
        print(f"Metrics collected: {len(self.event_times)} events")
        return CStatus()

class MonitoredTask(GNode):
    def run(self):
        print("Monitored task executing")
        time.sleep(0.5)
        return CStatus()

pipeline = GPipeline()

metrics = MetricsEvent()
task = MonitoredTask()

pipeline.registerGElement(task, set(), "task")
pipeline.addGEvent(metrics, "metrics_collector")

# Task runs, event fires at completion
pipeline.process(runTimes=3)

print(f"Total events: {len(metrics.event_times)}")
```

--------------------------------

### Multi-Stage ETL Pipeline Creation

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Demonstrates building a complex, multi-stage ETL pipeline with custom task nodes. This example shows how to define dependencies between tasks within a pipeline.

```python
from PyCGraph import (
    GPipeline, GPipelineManager, GNode, GCluster, CStatus
)
import time

class FetchTask(GNode):
    def run(self):
        print(f"[{self.getName()}] Fetching...")
        time.sleep(0.1)
        return CStatus()

class ProcessTask(GNode):
    def run(self):
        print(f"[{self.getName()}] Processing...")
        time.sleep(0.2)
        return CStatus()

class SaveTask(GNode):
    def run(self):
        print(f"[{self.getName()}] Saving...")
        time.sleep(0.1)
        return CStatus()

def create_etl_pipeline(pipeline_id):
    """Create an ETL pipeline"""
    pipeline = GPipeline()
    
    fetch = FetchTask()
    process = ProcessTask()
    save = SaveTask()
    
    fetch.setName(f"fetch_{pipeline_id}")
    process.setName(f"process_{pipeline_id}")
    save.setName(f"save_{pipeline_id}")
    
    pipeline.registerGElement(fetch, set(), f"fetch_{pipeline_id}")
    pipeline.registerGElement(process, {fetch}, f"process_{pipeline_id}")
    pipeline.registerGElement(save, {process}, f"save_{pipeline_id}")
    
    return pipeline

# Create manager and pipelines
manager = GPipelineManager()

for i in range(5):
    etl = create_etl_pipeline(i)
    manager.add(etl)
```

--------------------------------

### Periodic Cleanup Daemon Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GDaemon.md

Implement a cleanup daemon to remove old temporary files. This daemon runs periodically to maintain a clean temporary directory.

```Python
from PyCGraph import GPipeline, GNode, GDaemon, CStatus
import os
import time
from datetime import datetime, timedelta

class CleanupDaemon(GDaemon):
    def __init__(self, temp_dir="/tmp/cgraph_temp"):
        super().__init__()
        self.temp_dir = temp_dir
    
    def run(self):
        try:
            # Clean up temporary files older than 1 hour
            cutoff = datetime.now() - timedelta(hours=1)
            cutoff_time = cutoff.timestamp()
            
            for filename in os.listdir(self.temp_dir):
                filepath = os.path.join(self.temp_dir, filename)
                if os.path.getmtime(filepath) < cutoff_time:
                    os.remove(filepath)
                    print(f"Cleaned up: {filename}")
            
            return CStatus()
        except Exception as e:
            return CStatus(-1, str(e))

class DataProcessor(GNode):
    def run(self):
        print("Processing data...")
        # Generate temporary files
        return CStatus()

pipeline = GPipeline()

processor = DataProcessor()
cleanup = CleanupDaemon()

pipeline.registerGElement(processor, set(), "processor")
pipeline.addGDaemon(cleanup, 5000)  # Clean every 5 seconds

pipeline.process()
```

--------------------------------

### Implement before() Hook

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAspect.md

The before() method is called before the element's run() method. Use it for setup or logging preconditions.

```python
class MeasurementAspect(GAspect):
    def before(self):
        import time
        self.start_time = time.time()
        print("Element execution starting")
        return CStatus()
```

--------------------------------

### Complete Example: Conditional Event

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GEvent.md

Shows a pipeline with a ConditionalEvent that can be configured with a threshold. The event's trigger logic can access this threshold for conditional behavior.

```python
from PyCGraph import GPipeline, GNode, GEvent, CStatus, GParam

class ConditionalEvent(GEvent):
    def __init__(self, threshold):
        super().__init__()
        self.threshold = threshold
    
    def trigger(self):
        # Event logic can access shared state
        print(f"Event fired: threshold={self.threshold}")
        return CStatus()

class DataTask(GNode):
    def run(self):
        print("Processing data")
        return CStatus()

pipeline = GPipeline()

data_task = DataTask()
event = ConditionalEvent(threshold=100)

pipeline.registerGElement(data_task, set(), "data")
pipeline.addGEvent(event, "conditional_event")

pipeline.process()
```

--------------------------------

### Element Timeout Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/errors.md

Demonstrates a GNode that will intentionally timeout due to a long sleep. The pipeline will detect this timeout.

```python
from PyCGraph import GNode, GElementTimeoutStrategy, CStatus
import time

class SlowTask(GNode):
    def run(self):
        self.setTimeout(1000, GElementTimeoutStrategy.AS_ERROR)
        
        # This will timeout
        time.sleep(2)
        return CStatus()

# Pipeline detects timeout and halts this element
pipeline.process()
```

--------------------------------

### Invalid Element Registration Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/errors.md

Shows how invalid element registration (e.g., passing None or creating circular dependencies) can lead to CException errors.

```python
from PyCGraph import GPipeline, GNode, CException

class BadRegistration:
    pipeline = GPipeline()
    
    try:
        # Invalid: element not properly created
        pipeline.registerGElement(None, set(), "invalid")
    except CException as e:
        print(f"Registration failed: {e}")

# Invalid: circular dependency (if supported)
try:
    node1.addDependGElements({node2})
    node2.addDependGElements({node1})
except CException:
    pass
```

--------------------------------

### Implement Core Daemon Logic

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GDaemon.md

Override the run() method to implement the core logic of your daemon. This example demonstrates a daemon that monitors CPU and memory usage using the psutil library.

```python
class ResourceMonitor(GDaemon):
    def run(self):
        import psutil
        print(f"CPU: {psutil.cpu_percent()}%")
        print(f"Memory: {psutil.virtual_memory().percent}%")
        return CStatus()
```

--------------------------------

### Parameter Data Flow Pattern Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GParam.md

Demonstrates a basic data flow pipeline with nodes for producing, processing, and consuming data. Parameter updates are noted as occurring via the pipeline interface.

```python
from PyCGraph import GPipeline, GNode, GParam, CStatus
import time

class DataProducer(GNode):
    def run(self):
        # Produce data
        data = {"timestamp": time.time(), "value": 42}
        # Parameter updates would happen here via pipeline interface
        print(f"Produced: {data}")
        return CStatus()

class DataProcessor(GNode):
    def run(self):
        # Process data from previous step
        # Access via parameter would be:
        # with pipeline.getParam("shared_data") as data:
        #     processed = transform(data)
        print("Processing data...")
        return CStatus()

class DataConsumer(GNode):
    def run(self):
        # Consume processed data
        # Access via parameter would be:
        # with pipeline.getParam("shared_data") as data:
        #     final_result = data.processed_value
        print("Consuming data...")
        return CStatus()

pipeline = GPipeline()

producer = DataProducer()
processor = DataProcessor()
consumer = DataConsumer()

pipeline.registerGElement(producer, set(), "producer")
pipeline.registerGElement(processor, {producer}, "processor")
pipeline.registerGElement(consumer, {processor}, "consumer")

pipeline.process()
```

--------------------------------

### Parameter Synchronization Example with Thread-Safe Counter

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GParam.md

Illustrates parameter synchronization using a thread-safe counter shared between multiple increment tasks and a read task. Shows how to register elements and process the pipeline.

```python
from PyCGraph import GPipeline, GNode, GParam, CStatus
from threading import Lock
import time

class CounterParam(GParam):
    """Thread-safe counter shared between elements"""
    def __init__(self):
        super().__init__()
        self.count = 0
        self.lock = Lock()
    
    def increment(self):
        with self.lock:
            self.count += 1
            return self.count

class IncrementTask(GNode):
    def run(self):
        # Simulate shared counter increment
        # In real usage, would be:
        # with pipeline.getParam("counter") as counter:
        #     new_val = counter.increment()
        print("Incrementing counter...")
        time.sleep(0.1)
        return CStatus()

class ReadCounterTask(GNode):
    def run(self):
        # Read final counter value
        # In real usage, would be:
        # with pipeline.getParam("counter") as counter:
        #     final = counter.count
        print("Reading counter...")
        return CStatus()

pipeline = GPipeline()

# Create multiple increment tasks
increment_tasks = [IncrementTask() for _ in range(5)]
read_task = ReadCounterTask()

# Register increment tasks in parallel
for i, task in enumerate(increment_tasks):
    pipeline.registerGElement(task, set(), f"increment_{i}")

# Register read task after all increments
pipeline.registerGElement(read_task, set(increment_tasks), "read_counter")

pipeline.process()
```

--------------------------------

### Batch Processing Daemon Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GDaemon.md

Implement a batch queue daemon to process items in batches. This daemon manages a queue and processes items when a batch size is reached.

```Python
from PyCGraph import GPipeline, GNode, GDaemon, CStatus
from collections import deque
from threading import Lock
import time

class BatchQueue(GDaemon):
    def __init__(self, batch_size=5):
        super().__init__()
        self.batch_size = batch_size
        self.queue = deque()
        self.lock = Lock()
        self.processed = 0
    
    def enqueue(self, item):
        with self.lock:
            self.queue.append(item)
    
    def run(self):
        with self.lock:
            pending = len(self.queue)
            if pending > 0:
                print(f"Pending items: {pending}, "
                      f"Processed: {self.processed}")
        
        # Process batch
        with self.lock:
            if len(self.queue) >= self.batch_size:
                batch = [self.queue.popleft() for _ in range(self.batch_size)]
                self.processed += len(batch)
                print(f"Processing batch of {len(batch)} items")
        
        return CStatus()

class ItemProducer(GNode):
    def __init__(self, queue):
        super().__init__()
        self.queue = queue
    
    def run(self):
        print("Producing items...")
        for i in range(3):
            self.queue.enqueue(f"item_{i}")
        time.sleep(0.5)
        return CStatus()

pipeline = GPipeline()

batch_queue = BatchQueue(batch_size=5)
producer = ItemProducer(batch_queue)

pipeline.registerGElement(producer, set(), "producer")
pipeline.addGDaemon(batch_queue, 1000)  # Check every 1 second

pipeline.process()
```

--------------------------------

### GSingleton Usage Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAdapter.md

Illustrates how GSingleton ensures an element runs only once, even with multiple pipeline runs. Useful for initialization patterns.

```python
from PyCGraph import GSingleton

singleton = GSingleton()

class SingletonInit(GSingleton):
    def run(self):
        print("This runs only once, even if loop > 1")
        return CStatus()

init_task = SingletonInit()

pipeline = GPipeline()
pipeline.registerGElement(init_task, set(), "singleton")
pipeline.process(runTimes=5)  # Prints message only once
```

--------------------------------

### Example: Performance Aspect

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAspect.md

This aspect measures the execution time of an element and stores the timings. It demonstrates how to use before() and after() for performance monitoring.

```python
from PyCGraph import GPipeline, GNode, GAspect, CStatus
import time

class PerfAspect(GAspect):
    def __init__(self):
        super().__init__()
        self.timings = {}
    
    def before(self):
        self.start = time.time()
        return CStatus()
    
    def after(self):
        elapsed = time.time() - self.start
        element_name = "unknown"  # Can enhance to get element name
        if element_name not in self.timings:
            self.timings[element_name] = []
        self.timings[element_name].append(elapsed)
        print(f"Execution time: {elapsed:.3f}s")
        return CStatus()

class LongTask(GNode):
    def run(self):
        print("Executing long task...")
        time.sleep(1)
        return CStatus()

pipeline = GPipeline()

task = LongTask()
perf = PerfAspect()
task.addGAspect(perf)

pipeline.registerGElement(task, set(), "long_task")
pipeline.process()

print(f"Timings: {perf.timings}")
```

--------------------------------

### Resource Management Aspect Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAspect.md

Implement a ResourceAspect to manage external resources like database connections. The 'before' method connects, and the 'after' method disconnects.

```python
from PyCGraph import GPipeline, GNode, GAspect, CStatus

class DatabaseConnection:
    def connect(self):
        print("Connecting to database...")
    
    def disconnect(self):
        print("Disconnecting from database...")

class ResourceAspect(GAspect):
    def __init__(self, resource):
        super().__init__()
        self.resource = resource
    
    def before(self):
        try:
            self.resource.connect()
            return CStatus()
        except Exception as e:
            return CStatus(-1, str(e))
    
    def after(self):
        try:
            self.resource.disconnect()
            return CStatus()
        except Exception as e:
            return CStatus(-1, str(e))

class DatabaseQuery(GNode):
    def run(self):
        print("Executing query...")
        return CStatus()

pipeline = GPipeline()

db_conn = DatabaseConnection()
db_aspect = ResourceAspect(db_conn)

task = DatabaseQuery()
task.addGAspect(db_aspect)

pipeline.registerGElement(task, set(), "db_query")
pipeline.process()
```

--------------------------------

### Performance Tip: Use Level/Priority

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/REFERENCE.md

Utilize level/priority settings for init/destroy ordering to control execution flow.

```text
2. Use level/priority for init/destroy ordering
```

--------------------------------

### Python DAG Pipeline Example

Source: https://github.com/chunelfeng/cgraph/blob/main/README_en.md

Illustrates how to build a DAG pipeline using the Python API of CGraph. It defines custom nodes and sets up their execution order and dependencies within the pipeline.

```python
import time
from datetime import datetime

from PyCGraph import GNode, GPipeline, CStatus


class MyNode1(GNode):
    def run(self):
        print("[{0}] {1}, enter MyNode1 run function. Sleep for 1 second ... ".format(datetime.now(), self.getName()))
        time.sleep(1)
        return CStatus()

class MyNode2(GNode):
    def run(self):
        print("[{0}] {1}, enter MyNode2 run function. Sleep for 2 second ... ".format(datetime.now(), self.getName()))
        time.sleep(2)
        return CStatus()


if __name__ == '__main__':
    pipeline = GPipeline()
    a, b, c, d = MyNode1(), MyNode2(), MyNode1(), MyNode2()

    pipeline.registerGElement(a, set(), "nodeA")
    pipeline.registerGElement(b, {a}, "nodeB")
    pipeline.registerGElement(c, {a}, "nodeC")
    pipeline.registerGElement(d, {b, c}, "nodeD")

    pipeline.process()

```

--------------------------------

### Dependency Violation Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/errors.md

Illustrates correct and incorrect dependency registration for GNodes in a GPipeline. Ensures tasks run in the correct order.

```python
from PyCGraph import GPipeline, GNode, CStatus

class Task1(GNode):
    def run(self):
        return CStatus()

class Task2(GNode):
    def run(self):
        # Assumes Task1 completed (has dependency)
        return CStatus()

pipeline = GPipeline()
task1 = Task1()
task2 = Task2()

# Correct: task2 depends on task1
pipeline.registerGElement(task1, set(), "task1")
pipeline.registerGElement(task2, {task1}, "task2")

# Without dependency, task2 might run before task1
```

--------------------------------

### C++ DAG Pipeline Example

Source: https://github.com/chunelfeng/cgraph/blob/main/README_en.md

Demonstrates creating a Directed Acyclic Graph (DAG) pipeline in C++ using CGraph. It shows how to define custom nodes inheriting from GNode, register them with dependencies, and process the pipeline.

```cpp
#include "CGraph.h"

using namespace CGraph;

class MyNode1 : public GNode {
public:
    CStatus run() override {
        printf("[%s], sleep for 1 second ...\n", this->getName().c_str());
        CGRAPH_SLEEP_SECOND(1)
        return CStatus();
    }
};

class MyNode2 : public GNode {
public:
    CStatus run() override {
        printf("[%s], sleep for 2 second ...\n", this->getName().c_str());
        CGRAPH_SLEEP_SECOND(2)
        return CStatus();
    }
};


int main() {
    GPipelinePtr pipeline = GPipelineFactory::create();
    GElementPtr a, b, c, d = nullptr;

    pipeline->registerGElement<MyNode1>(&a, {}, "nodeA");
    pipeline->registerGElement<MyNode2>(&b, {a}, "nodeB");
    pipeline->registerGElement<MyNode1>(&c, {a}, "nodeC");
    pipeline->registerGElement<MyNode2>(&d, {b, c}, "nodeD");

    pipeline->process();

    GPipelineFactory::remove(pipeline);

    return 0;
}

```

--------------------------------

### init()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Initializes all managed pipelines. Sets up each pipeline's internal resources.

```APIDOC
## init()

### Description
Initializes all managed pipelines. Sets up each pipeline's internal resources.

### Returns
`CStatus` — Initialization status.
```

--------------------------------

### Example: Logging Aspect

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GAspect.md

This aspect logs the start and end of an element's execution using Python's logging module. It shows how to integrate custom logging into the aspect lifecycle.

```python
from PyCGraph import GPipeline, GNode, GAspect, CStatus
from datetime import datetime
import logging

class LoggingAspect(GAspect):
    def __init__(self, logger=None):
        super().__init__()
        self.logger = logger or logging.getLogger(__name__)
    
    def before(self):
        self.logger.info("Element started")
        return CStatus()
    
    def after(self):
        self.logger.info("Element finished")
        return CStatus()

class DataTask(GNode):
    def run(self):
        print("Processing data...")
        return CStatus()

logging.basicConfig(level=logging.INFO)

pipeline = GPipeline()
task = DataTask()
logging_aspect = LoggingAspect()
task.addGAspect(logging_aspect)

pipeline.registerGElement(task, set(), "data_task")
pipeline.process()
```

--------------------------------

### Best Practice: Keep event processing brief

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GEvent.md

Example of a GEvent designed for quick operations. Ensure that event processing is fast to avoid blocking pipeline execution.

```python
class QuickEvent(GEvent):
    def trigger(self):
        # Fast operation only
        log_metric()
        return CStatus()
```

--------------------------------

### Define and Execute a Simple Task

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/REFERENCE.md

Demonstrates the basic steps to define a task, create a pipeline, register the task, and execute it.

```python
from PyCGraph import GPipeline, GNode, CStatus

# 1. Define a task
class MyTask(GNode):
    def run(self):
        print("Executing task")
        return CStatus()

# 2. Create pipeline
pipeline = GPipeline()

# 3. Register elements
task = MyTask()
pipeline.registerGElement(task, set(), "my_task")

# 4. Execute
pipeline.process()
```

--------------------------------

### Health Check Daemon Example

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GDaemon.md

Implement a health check daemon that monitors active tasks against a maximum limit. This daemon runs periodically to report its status.

```Python
from PyCGraph import GPipeline, GNode, GDaemon, CStatus
from datetime import datetime
import time

class HealthCheckDaemon(GDaemon):
    def __init__(self, max_tasks=10):
        super().__init__()
        self.max_tasks = max_tasks
        self.active_tasks = 0
        self.task_count = 0
    
    def record_task(self):
        self.task_count += 1
    
    def run(self):
        timestamp = datetime.now().isoformat()
        status = "HEALTHY" if self.active_tasks < self.max_tasks else "OVERLOADED"
        print(f"[{timestamp}] Health: {status}, "
              f"Active: {self.active_tasks}, Total: {self.task_count}")
        return CStatus()

class WorkTask(GNode):
    def __init__(self, health_daemon):
        super().__init__()
        self.health = health_daemon
    
    def run(self):
        self.health.active_tasks += 1
        self.health.record_task()
        print("Task working...")
        time.sleep(0.5)
        self.health.active_tasks -= 1
        return CStatus()

pipeline = GPipeline()

health = HealthCheckDaemon(max_tasks=5)
tasks = [WorkTask(health) for _ in range(3)]

for i, task in enumerate(tasks):
    pipeline.registerGElement(task, set(), f"task_{i}")

pipeline.addGDaemon(health, 2000)  # Check every 2 seconds

pipeline.process()
```

--------------------------------

### Suspend GPipeline Execution

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipeline.md

Pauses the pipeline's current execution. Any elements that have already started their current run will complete, but new elements will not start until `resume()` is called.

```python
status = pipeline.suspend()
```

--------------------------------

### Get Node Loop Count

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

Gets the current loop count for the node, indicating how many times it executes per run. Useful for understanding iterative behavior.

```python
loop_count = node.getLoop()
print(f"Executes {loop_count} times per run")
```

--------------------------------

### Performance Tip: Use makeSerial()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/REFERENCE.md

Use makeSerial() for linear pipelines to ensure sequential execution.

```text
6. Use `makeSerial()` for linear pipelines
```

--------------------------------

### Performance Tip: Profile with perf()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/REFERENCE.md

Profile your pipeline with perf() before attempting optimizations.

```text
8. Profile with `perf()` before optimizing
```

--------------------------------

### Build Tutorial Executables

Source: https://github.com/chunelfeng/cgraph/blob/main/tutorial/CMakeLists.txt

Iterates through the defined tutorial list and creates an executable for each. Each executable is compiled from its C++ source file and linked against the CGraph library.

```cmake
foreach(tut ${CGRAPH_TUTORIAL_LIST})
    add_executable(${tut}
            # 在自己的工程中引入CGraph功能，仅需引入 CGraph-env-include.cmake 后，加入这一句话即可
            $<TARGET_OBJECTS:CGraph>
            ${tut}.cpp
            )
endforeach()
```

--------------------------------

### Performance Tip: Enable CPU Binding

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/REFERENCE.md

Enable CPU binding for NUMA systems to optimize thread-to-core affinity.

```text
3. Enable CPU binding for NUMA systems
```

--------------------------------

### getSize()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Gets the number of managed pipelines.

```APIDOC
## getSize()

### Description
Gets the number of managed pipelines.

### Returns
`int` — Number of pipelines in collection.
```

--------------------------------

### Execute All Pipelines

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Demonstrates the basic lifecycle of initializing, running, and destroying all pipelines managed by GPipelineManager.

```python
print("Starting batch execution...")
manager.init()
manager.run()
manager.destroy()
print("Batch execution completed")
```

--------------------------------

### Get Group Name

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GGroup.md

Retrieves the human-readable name assigned to the group.

```python
name = group.getName()
```

--------------------------------

### Dynamic Pipeline Creation and Execution

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Demonstrates how to dynamically create pipelines, add them to a manager, and execute them. Use this pattern when pipelines are not known at compile time.

```python
from PyCGraph import GPipeline, GPipelineManager, GNode, CStatus

class DynamicPipelineManager:
    def __init__(self):
        self.manager = GPipelineManager()
    
    def create_and_add_pipeline(self, pipeline_id):
        pipeline = GPipeline()
        
        # Configure pipeline
        task = GNode()
        pipeline.registerGElement(task, set(), f"task_{pipeline_id}")
        
        # Add to manager
        self.manager.add(pipeline)
        return pipeline
    
    def execute_all(self):
        self.manager.init()
        status = self.manager.run()
        self.manager.destroy()
        return status
    
    def remove_pipeline(self, pipeline):
        return self.manager.remove(pipeline)

# Usage
mgr = DynamicPipelineManager()
p1 = mgr.create_and_add_pipeline(1)
p2 = mgr.create_and_add_pipeline(2)
p3 = mgr.create_and_add_pipeline(3)

status = mgr.execute_all()
```

--------------------------------

### getLoop()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GNode.md

Gets the current loop count configured for the node's execution.

```APIDOC
## getLoop()

### Description
Gets the current loop count.

### Method
`getLoop() -> int`

### Parameters
None

### Request Example
```python
loop_count = node.getLoop()
print(f"Executes {loop_count} times per run")
```

### Response
#### Success Response (200)
`int` — Loop count.

#### Response Example
```json
{
  "example": 5
}
```
```

--------------------------------

### Get Pipeline Count

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipelineManager.md

Retrieves the number of pipelines currently managed by the GPipelineManager.

```python
num_pipelines = manager.getSize()
print(f"Executed {num_pipelines} ETL pipelines")
```

--------------------------------

### Parallel Execution of Tasks

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/REFERENCE.md

Shows how to set up tasks that can run in parallel, with dependencies only on an initial task.

```python
task1 = NodeA()
task2 = NodeB()
task3 = NodeC()

# No dependencies between 2 and 3
pipeline.registerGElement(task1, set(), "task1")
pipeline.registerGElement(task2, {task1}, "task2")
pipeline.registerGElement(task3, {task1}, "task3")
```

--------------------------------

### Get Group Relationships

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GGroup.md

Retrieves information about the relationships the group has with other elements in the graph, such as dependencies.

```python
relation = group.getRelation()
```

--------------------------------

### run()

Source: https://github.com/chunelfeng/cgraph/blob/main/_autodocs/api-reference/GPipeline.md

Executes all registered elements once according to their dependencies. This method must be called after `init()` and before `destroy()`.

```APIDOC
## run()

### Description
Executes all registered elements once according to their dependencies. Must be preceded by `init()` and followed by `destroy()`.

### Parameters
No parameters

### Returns
`CStatus` — Execution status.
```

--------------------------------

### Compile and Run CGraph with Bazel

Source: https://github.com/chunelfeng/cgraph/blob/main/COMPILE.md

Build CGraph tutorial targets using Bazel and run a specific tutorial. Supports Linux, MacOS, and Windows.

```shell
$ git clone https://github.com/ChunelFeng/CGraph.git
$ cd CGraph
$ bazel build //tutorial/...              # 编译 tutorial路径下的所有targets
$ bazel run //tutorial:T00-HelloCGraph    # 运行 T00-HelloCGraph
```