Integrating CGraph Pipelines with Python Using the pycgraph Binding

The pycgraph binding exposes CGraph's high-performance C++ DAG engine to Python via pybind11, allowing developers to construct and execute parallel pipelines using native Python classes while retaining lock-free scheduling and thread-pool optimizations.

The pycgraph package provides a thin, zero-dependency Python wrapper around the CGraph C++ library, enabling seamless integration of high-throughput directed acyclic graph (DAG) pipelines into Python applications. By leveraging pybind11, the binding in chunelfeng/cgraph exposes core abstractions such as GPipeline, GNode, and CStatus as first-class Python objects, making it straightforward to script complex workflows without sacrificing the performance of the underlying lock-free execution engine.

Architecture and Data Flow

The binding architecture centers on python/PyCGraph.cpp, which declares the pycgraph module using PYBIND11_MODULE(pycgraph, cg). This file registers all CGraph types with Python via py::class_ declarations, mapping C++ methods directly to Python instance methods.

The wrapper headers in python/wrapper/—including PyWrapperInclude.h, PywGNode.h, and PywGPipelineDeleter.h—provide thin C++ subclasses that adapt native CGraph classes for Python consumption. These wrappers handle Python-specific concerns such as constructor adaptation and reference counting.

Module Initialization and Class Registration

The module initialization in python/PyCGraph.cpp uses the standard pybind11 macro to expose the CGraph API:

PYBIND11_MODULE(pycgraph, cg) {
    cg.doc() = "CGraph with python api, github: https://github.com/ChunelFeng/CGraph";
    // ... class and enum registrations ...
}

This registration exposes critical classes including UThreadPoolConfig for thread-pool tuning, GElementRelation for dependency inspection, and the core GPipeline, GNode, and CStatus types. Enums such as GEngineType and GElementTimeoutStrategy are mapped to Python enum types for readable configuration.

Lifetime Management and Object Safety

To prevent premature garbage collection of Python nodes registered with a C++ pipeline, the binding employs keep_alive<1, 2>() policies on methods like registerGElement. This ensures that as long as the GPipeline exists, its registered GNode objects remain valid in Python memory.

For base classes that should not be deleted by Python, the binding uses py::nodelete, allowing the C++ side to own object lifetimes. These policies prevent dangling pointers when pipelines outlive temporary Python objects.

Thread Safety and GIL Handling

Methods that invoke the C++ thread pool or block on asynchronous results release the Python Global Interpreter Lock (GIL) using py::call_guard<py::gil_scoped_release>(). This allows true parallel execution of CGraph nodes across multiple CPU cores without Python thread contention, enabling the C++ scheduler to utilize the UThreadPool while other Python threads remain unblocked.

Building and Installing the pycgraph Module

The python/setup.py script compiles the CGraph core sources (src/*.cpp) alongside the pybind11 wrappers into a shared library named pycgraph. The Extension API pulls in all necessary headers from src/ and python/wrapper/, linking the resulting module into an importable Python package.

To install from source:

git clone https://github.com/chunelfeng/cgraph.git
cd cgraph/python
pip install .

This builds the native extension and makes import pycgraph available in your Python environment.

Practical Python Implementation Example

The following snippet demonstrates a complete end-to-end pipeline built in Python, mirroring the C++ "HelloCGraph" demo. It defines custom nodes by subclassing GNode, registers them with dependencies, and executes the DAG.

import time
from datetime import datetime
from pycgraph import GNode, GPipeline, CStatus

# ----------------------------------------------------------------------

# Define custom node classes by subclassing GNode

# ----------------------------------------------------------------------

class MyNode1(GNode):
    """Sleep 1 s and report execution."""
    def run(self):
        print(f"[{datetime.now()}] {self.getName()} → MyNode1.run (sleep 1 s)")
        time.sleep(1)
        return CStatus()

class MyNode2(GNode):
    """Sleep 2 s and report execution."""
    def run(self):
        print(f"[{datetime.now()}] {self.getName()} → MyNode2.run (sleep 2 s)")
        time.sleep(2)
        return CStatus()

# ----------------------------------------------------------------------

# Build the pipeline

# ----------------------------------------------------------------------

pipeline = GPipeline()

# Instantiate node objects

a, b, c, d = MyNode1(), MyNode2(), MyNode1(), MyNode2()

# Register nodes and their dependencies (DAG edges)

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

# Execute the graph (process runs all nodes respecting dependencies)

pipeline.process()

Under the hood, GPipeline() creates a C++ GPipeline object wrapped by py::class_<GPipeline>. The registerGElement method stores Python-side GNode objects inside the C++ pipeline, with keep_alive policies guaranteeing they are not garbage-collected prematurely. When process() is called, the C++ scheduler executes nodeA, then runs nodeB and nodeC in parallel using the default UThreadPool, and finally executes nodeD once its dependencies complete.

Key Source Files in the Binding Layer

The following files constitute the complete binding layer between the CGraph C++ engine and Python:

File	Purpose
python/PyCGraph.cpp	Central pybind11 module definition that registers all CGraph types, enums, and methods with the pycgraph Python module.
python/wrapper/PyWrapperInclude.h	Aggregate header that includes all wrapper-specific adaptations required for Python compatibility.
python/wrapper/PywGNode.h	Defines PywGNode, a thin subclass adapting GNode for Python inheritance and method overriding.
python/wrapper/PywGPipelineDeleter.h	Manages custom deletion policies for pipeline objects to ensure proper C++ cleanup from Python.
python/setup.py	setuptools configuration that compiles CGraph core sources and pybind11 wrappers into the pycgraph shared library.
src/CGraph.h	Public header for the core C++ engine; defines the underlying GPipeline, GNode, and CStatus implementations.

Summary

• pycgraph is a thin Python wrapper around the CGraph C++ engine, built with pybind11 to expose high-performance DAG scheduling to Python developers.
• The binding is defined in python/PyCGraph.cpp and uses wrapper headers in python/wrapper/ to adapt C++ classes for Python inheritance and lifetime management.
• Lifetime safety is enforced via keep_alive<1, 2>() policies on registerGElement, ensuring Python nodes persist as long as the C++ pipeline references them.
• Thread safety is achieved by releasing the Python GIL during C++ execution with py::call_guard<py::gil_scoped_release>(), enabling true parallel node execution across multiple CPU cores.
• Developers can install the module via pip install from python/setup.py, which compiles the native extension against the CGraph core sources.

Frequently Asked Questions

How do I install pycgraph from source?

Clone the chunelfeng/cgraph repository and install the Python package from the python directory. The setup.py script automatically compiles the C++ sources and pybind11 wrappers into a shared library using the Extension API. Run pip install ./python from the repository root to build and install the pycgraph module.

Can I mix Python-defined nodes with C++ nodes in the same pipeline?

Yes. The pycgraph binding allows you to register Python subclasses of GNode alongside native C++ nodes within the same GPipeline. The C++ scheduler treats both uniformly, executing Python-defined run() methods while managing dependencies through the underlying DAG engine. The keep_alive policies ensure that Python objects remain valid throughout the C++ execution lifecycle.

Does pycgraph support multi-threading in Python?

Yes. The binding releases the Python Global Interpreter Lock (GIL) during C++ execution using py::call_guard<py::gil_scoped_release>(). This allows CGraph's UThreadPool to utilize multiple CPU cores concurrently, even from a single Python process. Consequently, Python-defined nodes can run in parallel without the contention typically associated with Python threading.

How are memory leaks prevented when passing Python objects to C++?

The binding uses pybind11's keep_alive<1, 2>() policy on methods like registerGElement in python/PyCGraph.cpp. This policy instructs Python to keep the second argument (the GNode) alive as long as the first argument (the GPipeline) exists. Additionally, py::nodelete is used for base classes where C++ owns the lifetime, preventing Python from deleting objects that the C++ engine still references.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}