A C++ engine wearing a Python API
pyNetX exposes a single class — NetconfClient — but underneath it runs a
compact C++ runtime: libssh2 sessions in non-blocking mode, a global least-loaded
thread pool, a shared asyncio future dispatcher, and an epoll reactor pool
for notification streams. Python never blocks; the engine does the waiting.
Async-focused surface
The async APIs are now the primary path: get_config_async(),
subscribe_async() and friends return awaitable asyncio.Future
objects completed off-thread. Legacy sync-flow methods remain for compatibility in
v2.0.7, but are deprecated for a future major release.
Shared worker pool
All clients share one C++ ThreadPool (default 4 workers,
set_threadpool_size(n)). Each enqueue picks the worker with the smallest
in-flight count — load balancing by queue depth, not round-robin.
Epoll notifications
Notification sockets are watched by dedicated reactor threads using Linux
epoll. Hundreds of idle devices cost almost nothing; one thread scales
to hundreds of FDs.
Crash-hardened
Reactor callbacks, pool workers and the dispatcher are exception-guarded.
Dead FDs are unregistered, subscriptions marked inactive, and the Python process
keeps running. Clients are held by weak_ptr — no stale pointers.
The layer cake, exploded
A call travels down the left edge — Python coroutine → pybind11 binding → C++ engine → libssh2 framing → kernel socket — and the device's XML answer climbs back up the right. The GIL is only held at the very top; everything below runs free of it.
Why two SSH sessions per client?
RPCs are strictly request → reply → next request on one channel. A long-lived
notification stream would starve that channel, so each subscribed client opens a
second SSH session whose only job is carrying
<notification> messages — and whose socket FD is handed to the
epoll reactor.
Non-blocking all the way down
Async paths use libssh2 in non-blocking mode. When a read returns
EAGAIN, the worker parks on poll() until the socket is
readable (v2.0.4) — no busy-spinning, no burned CPU while a device thinks.
Messages are framed and read until the NETCONF ]]>]]>
end-of-message marker.
RAII everywhere
Sockets, epoll descriptors, libssh2 sessions and channels are wrapped in RAII
types (SocketRAII, EpollRAII, custom
unique_ptr deleters). Cleanup happens deterministically even on
exception paths — the fix that ended the v1.0.9 crash era.
From await to <rpc-reply>
Watch a single await client.get_async() make the round trip. The glowing
payload below is your request: green ring = your event loop, columns = the shared
thread pool, the far node = the device, the magenta octahedron = the shared
AsyncFutureDispatcher that completes your Future. The step cards light up
in sync with the animation.
- Submit. The pybind11 binding wraps your call,
enqueues a task on the least-loaded ThreadPool worker, creates an
asyncio.Futureon your running loop, and registers a poller with the shared dispatcher. Your coroutine suspends. - Serialize. A worker thread picks the task up and
takes
session_mutex_— one RPC at a time per client channel, so request/reply ordering can never interleave. - Transmit. The XML RPC (with
message-id) is written through libssh2 over the non-blocking TCP socket to the device's NETCONF subsystem. - Receive. The worker reads until the
]]>]]>end-of-message marker, parking onpoll()whenever libssh2 saysEAGAIN. An<rpc-error>becomes a typed C++ exception. - Complete. The C++
std::futureturns ready. The shared dispatcher thread — batch-polling every 50 ms — notices, acquires the GIL, and converts the result (or maps the C++ exception to its Python twin). - Resume.
loop.call_soon_threadsafe(set_result)hands the value to your event loop. If you already cancelled,fut_pending()skips the write — noInvalidStateError. Yourawaitreturns.
One epoll, hundreds of devices
After subscribe_async() gets its <rpc-reply> — and only
then — the notification socket is registered with a reactor. From that point a single
epoll_wait() loop hears every device. Pushes flow right-to-left below:
device → reactor → per-client deque → next_notification_async() or the
GIL-free next_notification() helper. Red flashes are dying FDs being cleaned
up; an over-full queue drops the newest message and emits health events.
Registration is reply-gated
Registering the FD before the subscription reply caused a classic race:
the reactor could swallow the <rpc-reply> meant for
subscribe_*(). Since v2.0.3 the FD only joins epoll after
<create-subscription> succeeds.
Reactor pool, least-loaded
set_notification_reactor_count(n) spins up n reactor
threads, each with its own epoll FD. New subscriptions land on the reactor watching
the fewest devices, and existing FDs are rebalanced when the count changes. Without
the call, one reactor is created lazily on first subscribe.
Bounded or unbounded queues
notif_queue_size=-1 buffers without limit. A non-negative size bounds
the deque: when full, new notifications are dropped rather than stalling the
reactor. v2.0.6+ exposes notification_queue_size(),
peek_notifications(), and next_notification_async(timeout_ms).
The synchronous helper next_notification() now releases the GIL while waiting.
v2.0.7 stream parser
The notification receiver now keeps a per-subscription receive buffer. It can split
multiple complete ]]>]]> frames from one read, preserve trailing partial
bytes for the next callback, recover missing-EOM notifications for inspection, and emit
malformed_notification or incomplete_notification health events
when a device sends broken stream data.
Failure is survivable
On EPOLLERR / EPOLLHUP / EPOLLRDHUP or a read failure, the reactor
unregisters the FD, marks the subscription inactive
(is_subscription_active() → False) and moves on. Reactors
hold weak_ptrs, so a garbage-collected client can never be touched
through a stale pointer.
Notification streams are parsed as streams, not socket reads
v2.0.7 adds a persistent per-subscription notification receive buffer in front of
the existing queue and health-event pipeline. Coalesced notifications, trailing
fragments, abandoned partials, empty EOM frames, orphan bytes and malformed XML are
now classified before data reaches Python, while the v2.0.6 label and
UTC timestamp event fields remain intact for dashboards and event pipelines.
Persistent receive state
Each subscription keeps notification bytes that were not complete yet, so a reactor callback no longer has to pretend one SSH read equals one notification.
FIFO split for many frames
Multiple ]]>]]>-delimited notifications arriving in one read are split and queued as separate entries, preserving delivery order.
Trailing partials survive
A complete notification followed by the start of the next one is handled correctly: the complete frame is queued and the partial tail stays buffered.
Malformed stream events
Invalid frames, empty EOM frames, orphan bytes and recovered missing-EOM notifications emit malformed_notification events for inspection.
Process-wide health stream
Use pyNetX.next_notification_event() or
await pyNetX.next_notification_event_async() to observe queue pressure,
dropped notification deltas, recovery events, incomplete reads, partial byte counts,
and health-event bus drops. Each event can be converted with event.as_dict()
and now includes timestamp and label.
Device-friendly labels
NetconfClient(..., label="blr-leaf-01") attaches a human-readable
identifier to health events. The default is the string "None", and global
timeout events also use label="None" because they are not generated by a
specific device client.
Timestamped at generation
Events are timestamped when generated using UTC, so operators can correlate queue pressure and notification drops with device logs, Kafka pipelines, application logs and monitoring systems without depending on local timezone settings.
Same event pipeline, richer parser signals
The existing process-wide health bus remains the operator-facing plane. v2.0.7 simply
feeds it better parser diagnostics: malformed_notification for invalid stream
data and incomplete_notification when partial bytes are abandoned or time out.
Every thread, FD and lock in the process
The whole runtime is a fixed, predictable set of threads — nothing is spawned per call, per device, or per notification.
Python main thread
× 1Runs your asyncio event loop. Awaits Futures, fires callbacks scheduled by the
dispatcher, and can call the GIL-free next_notification() helper. The
event loop can also await next_notification_async().
ThreadPool workers
× N (default 4)Execute every NETCONF operation. Per-worker task queues with least-in-flight
dispatch; exception-guarded so a bad task can't kill the pool. Resize with
set_threadpool_size(n).
AsyncFutureDispatcher
× 1 · sharedA leaked-singleton, detached thread that batch-polls registered
std::futures every 50 ms and completes asyncio Futures via
call_soon_threadsafe. Replaced the old one-watcher-thread-per-call design
in v2.0.4.
Notification reactors
× M (default 1, lazy)Each owns one epoll FD and a 1-second-tick epoll_wait() loop over up to
64 events per wake. Reads notification byte streams through the v2.0.7 persistent parser, pushes
complete or inspectable payloads into the owning client's deque, emits health events,
and skips stuck partial messages using v2.0.5+ guards.
Scale with set_notification_reactor_count(m).
File descriptors
Per NetconfClient:
Per reactor:
Locks & synchronization
Want true parallelism against one device? Use separate
NetconfClient instances — the per-channel lock is a feature, not a
limitation: it protects NETCONF's request/reply framing.
One class, two tempos
pyNetX is now async-focused. Use *_async for connection handling, RPCs,
configuration operations and subscriptions. Legacy *_sync flow methods remain
in v2.0.7 for compatibility, but emit DeprecationWarning and will be removed
in a future major release. Common helpers such as next_notification(),
queue inspection and event-stream APIs are not deprecated.
connect_sync, disconnect_sync, send_rpc_sync,
get_config_sync, subscribe_sync, and related configuration RPC methods.
Use the corresponding *_async methods for new work. Common/helper APIs remain supported.
Deprecated sync flow · compatibility only
Asynchronous · returns asyncio Futures
Module-level · tune the engine
Constructor knobs
Exceptions cross the bridge typed
C++ exceptions are mapped to Python classes with sensible built-in bases — and since v2.0.4 the async path preserves the same public pyNetX exception classes as the compatibility sync path.
| C++ exception | Python exception | Python base | Raised when |
|---|---|---|---|
NetconfConnectionRefused | NetconfConnectionRefusedError | ConnectionError | TCP/SSH connect is refused or times out |
NetconfAuthError | NetconfAuthError | PermissionError | SSH authentication fails |
NetconfChannelError | NetconfChannelError | OSError | channel open / subsystem / IO failures |
NetconfException | NetconfException | RuntimeError | everything else, incl. <rpc-error> replies |
How the engine got here
Each release attacked one bottleneck: scalability, then stability, then the bridge, then production-grade observability, test coverage and finally byte-stream-safe notification parsing.
Hardened notification stream parser
Persistent per-subscription receive buffers now split coalesced EOM-delimited
notifications, preserve trailing partial bytes, recover missing-EOM payloads for
inspection, report abandoned partials, and emit malformed_notification
diagnostics for broken stream data.
Labeled, timestamped events and expanded release testing
NotificationHealthEvent now includes timestamp and
label, NetconfClient accepts label="...", timeout
events explicitly report label="None", and the release flow documents
repaired-wheel pytest checks plus optional Netopeer2/Sysrepo validation.
Notification observability and async-first direction
Process-wide NotificationHealthEvent stream,
next_notification_async(), early GIL release in
next_notification(), incomplete-notification guards
(notif_incomplete_max_kb, notif_incomplete_timeout),
queue inspection helpers, and sync-flow deprecation warnings.
The bridge grows up
Shared AsyncFutureDispatcher replaces one watcher thread per async call.
poll()-on-EAGAIN kills busy-spinning reads. New
socket_connect_timeout. Async finally raises the same typed exceptions as sync.
Notifications hardened
Reactor exceptions can no longer kill the interpreter. FDs register only after the
subscription reply. weak_ptr client refs, mutex-guarded notification state,
is_subscription_active().
Stability under load
Destructor/memory-cleanup crash fixes, defensive exception handling across the API,
bounded notification queues via notif_queue_size, wheels for Python 3.11–3.14.
Cancellation-safe bridge
fut_pending() guard skips set_result/set_exception on
finished or cancelled Futures — goodbye sporadic InvalidStateError.
select → epoll
The notification subsystem re-built on Linux epoll: ~85% fewer CPU wake-ups at 500 FDs, linear scaling, no thread per notification. Task pool switched to dynamic least-loaded dispatch (+40% mixed-traffic throughput).
Tested as installed wheels, then against a real NETCONF server
v2.0.7 keeps the installed-wheel and Netopeer2 release gates, and adds deep notification stream parsing coverage beyond unit checks. The release gate installs each repaired manylinux wheel into a clean virtual environment and runs pytest from outside the source tree, so imports come from the packaged wheel. Optional Netopeer2/Sysrepo tests validate real NETCONF-over-SSH behavior.
Default wheel gate
Inside the manylinux build container, each CPython wheel is built, repaired with
auditwheel, installed into a fresh venv, imported from
site-packages, and tested with pytest -m "not netopeer".
Fake NETCONF integration
The normal suite includes a local Paramiko-based NETCONF-over-SSH server for async connect/RPC/subscribe flows, queue-full/recovery events, fragmentation, coalesced raw chunks, malformed frames, orphan prefixes, large payloads, Unicode payloads, two-client behavior and cleanup paths.
Optional Netopeer2 validation
The netopeer marker runs against a containerized or externally supplied
Netopeer2/Sysrepo server. These tests are skipped when the target is not reachable and
pass once 127.0.0.1:830 is listening with the configured credentials.
# manylinux wheel gate: run from outside the source tree in a fresh venv
python -m pytest -c /io/test/pytest.ini /io/test -m "not netopeer" -ra --tb=short
# optional real NETCONF target on host
sudo docker run -d --name pynetx-netopeer2 -p 830:830 sysrepo/sysrepo-netopeer2:latest
cd /tmp/pynetx-netopeer-test311
PYNETX_RUN_NETOPEER=1 \
PYNETX_NETOPEER_HOST=127.0.0.1 \
PYNETX_NETOPEER_PORT=830 \
PYNETX_NETOPEER_USERNAME=netconf \
PYNETX_NETOPEER_PASSWORD=netconf \
python -m pytest \
-c /path/to/netconf_pybind/test/pytest.ini \
/path/to/netconf_pybind/test \
-m netopeer -ra --tb=shortSixty seconds to running code
Tune the engine once at startup, then drive devices concurrently from plain asyncio.
import asyncio
import pyNetX
from pyNetX import NetconfClient, NetconfAuthError
pyNetX.set_threadpool_size(10) # 10 NETCONF ops in flight, app-wide
async def fetch(host):
client = NetconfClient(
hostname=host, port=830,
username="admin", password="admin",
connect_timeout=30, read_timeout=30,
socket_connect_timeout=5,
label=f"device-{host}",
)
await client.connect_async()
try:
# same channel = serialized; different clients = parallel
config = await client.get_config_async(source="running")
return host, config
finally:
await client.disconnect_async()
async def main():
hosts = ["10.0.0.1", "10.0.0.2", "10.0.0.3"]
results = await asyncio.gather(
*(fetch(h) for h in hosts), return_exceptions=True
)
for r in results:
if isinstance(r, NetconfAuthError):
print("auth failed:", r)
asyncio.run(main())import asyncio
import pyNetX
from pyNetX import NetconfClient
pyNetX.set_threadpool_size(10)
pyNetX.set_notification_reactor_count(4)
async def health_monitor():
while True:
event = await pyNetX.next_notification_event_async(timeout_ms=-1)
if event.valid:
# includes timestamp, label, hostname, queue counters, malformed/incomplete diagnostics and drop deltas
print("health:", event.as_dict())
async def watch(client: NetconfClient):
await client.connect_async()
await client.subscribe_async(stream="NETCONF")
while client.is_subscription_active():
xml = await client.next_notification_async(timeout_ms=1000)
if xml:
handle(xml)
async def main():
client = NetconfClient(
hostname="10.0.0.1", port=830,
username="admin", password="admin",
notif_queue_size=1000,
notif_incomplete_max_kb=1024,
notif_incomplete_timeout=5,
notif_drop_event_threshold=1,
label="blr-leaf-01",
)
asyncio.create_task(health_monitor())
await watch(client)
def handle(xml: str):
print("notification:", xml[:120], "...")