OpenVX 1.3.2: implement pipelining, streaming, and event-queue extension

[simonCatBot]

Summary

Implements the Khronos OpenVX 1.3.2 vx_khr_pipelining extension (graph pipelining, streaming, and event queue support) in AMD OpenVX.

What changed

• Added OPENVX_USE_PIPELINING CMake option (default ON) so the extension can be compiled out when not needed.
• New internal engine: amd_openvx/openvx/ago/ago_pipelining.cpp
  • Graph-parameter queue management (ready / consumed / done refs)
  • QUEUE_AUTO background executor and QUEUE_MANUAL synchronous drain
  • Graph-streaming executor thread
  • Event registration/delivery (vxRegisterEvent, vxSendUserEvent, graph/node events)
  • Source/sink node pipe-up pre-fill and per-node state tracking
  • Replicated graph-parameter handling using object-array/pyramid sibling substitution
• New extension API layer: amd_openvx/openvx/api/vx_pipelining_api.cpp
  • vxSetGraphScheduleConfig, enqueue/dequeue/check done refs, event APIs, streaming start/stop
• Wired pipelining hooks into the existing graph scheduling, wait, execute, and verify paths.
• Relaxed vxAddUserKernel to accept a NULL validate callback (required by CTS source/sink kernels) and added output-metadata inference so source/sink graphs verify.

Conformance results

Run with the minimal OpenVX-CTS runner built against this libopenvx.so:

Suite	Result
GraphPipeline.*	109/109 pass
GraphStreaming.*	24/24 pass
Combined	133/133 pass

Build

• Default: -DOPENVX_USE_PIPELINING=ON — extension enabled.
• Optional: -DOPENVX_USE_PIPELINING=OFF — extension APIs return VX_ERROR_NOT_SUPPORTED and the legacy vxProcessGraph path remains functional.

Related

• Closes conformance gap for OpenVX 1.3.2 Pipelining/Streaming extension tests.

[Copilot]

Pull request overview

Adds Khronos OpenVX 1.3.2 vx_khr_pipelining support (graph parameter queues, streaming, and event queue APIs) to the AMD OpenVX engine, including a build-time toggle and CTS CI coverage to close the conformance gap.

Changes:

• Introduces OPENVX_USE_PIPELINING CMake option and wires new pipelining/streaming/event APIs into the core engine.
• Adds new internal pipelining engine (ago_pipelining.cpp) and public API layer (vx_pipelining_api.cpp) implementing queueing, streaming threads, and event delivery.
• Expands CI conformance workflows to run CTS GraphPipeline.* and GraphStreaming.* suites on CPU and HIP.

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 13 comments.

Show a summary per file

File	Description
amd_openvx/openvx/CMakeLists.txt	Adds OPENVX_USE_PIPELINING option and builds new pipelining sources.
amd_openvx/openvx/api/vx_pipelining_api.cpp	New public extension API implementation (+ stubs when disabled).
amd_openvx/openvx/api/vx_api.cpp	Adds/updates context/graph/kernel/node/reference attributes needed for pipelining + relaxes vxAddUserKernel validate requirement.
amd_openvx/openvx/ago/ago_util.cpp	Adds new pipelining/event system state to core structs and cleans up in destructors.
amd_openvx/openvx/ago/ago_platform.h	Includes pipelining header and adds STL synchronization primitives.
amd_openvx/openvx/ago/ago_pipelining.cpp	New internal pipelining execution helpers, queue handling, and event emission.
amd_openvx/openvx/ago/ago_internal.h	Adds pipelining/event system structs and new fields on graph/context/kernel/node.
amd_openvx/openvx/ago/ago_interface.cpp	Wires event notifications and pipelining execution into verify/execute/process/wait paths.
.github/workflows/conformance.yml	Adds CTS configuration vars and new CPU pipelining/streaming jobs.
.github/workflows/conformance-hip.yml	Adds CTS configuration vars and new HIP pipelining/streaming jobs.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[simonCatBot]

@Copilot — thanks for the detailed review. The latest push addresses all of the items below. PTAL.

1. VX_CONTEXT_EVENT_TIMEOUT guard (vx_api.cpp) — vxSetContextAttribute now returns VX_ERROR_NOT_SUPPORTED when agoGetContextEventSystem() returns nullptr (i.e. pipelining compiled out).

2. vxQueryNode default: label (vx_api.cpp) — restored the explicit default: label in the vxQueryNode switch.

3. AgoGraphParameterQueue uninitialized fields (ago_internal.h) — added a constructor that initializes index, max_depth, and enabled.

4. vxWaitEvent timeout (vx_pipelining_api.cpp) — now honors evsys->timeout_ms: waits forever when VX_TIMEOUT_WAIT_FOREVER, otherwise uses wait_for(..., timeout_ms) and returns VX_FAILURE on timeout.

5. vxGraphParameterDequeueDoneRef timeout (vx_pipelining_api.cpp) — now uses pipe->timeout_ms with a timed wait on done_cv.

6. Bogus -DOPENVX_USE_STREAMING=ON in CI (.github/workflows/conformance.yml and .github/workflows/conformance-hip.yml) — removed the unused flag.

7. VX_KERNEL_PIPEUP_ query guard* (vx_api.cpp) — vxQueryKernel only reports pipeup depths when OPENVX_USE_PIPELINING is enabled; otherwise returns VX_ERROR_NOT_SUPPORTED.

8. VX_KERNEL_PIPEUP_ set guard* (vx_api.cpp) — vxSetKernelAttribute only accepts pipeup depths when OPENVX_USE_PIPELINING is enabled.

9. VX_REFERENCE_ENQUEUE_COUNT (vx_api.cpp + ago_pipelining.cpp) — implemented agoGetReferenceEnqueueCount() and wired it into vxQueryReference; returns the number of graph-parameter queues that currently hold the reference in ready/consumed/done.

10. Event-registration cleanup (ago_pipelining.cpp, ago_util.cpp) — agoRemoveEventRegistrations() is now invoked from AgoGraph::~AgoGraph, AgoNode::~AgoNode, and AgoData::~AgoData to prevent stale registrations/use-after-free.

11. QUEUE_AUTO polling (ago_pipelining.cpp) — replaced the busy sleep_for loop with a condition_variable wait on pipe->enqueue_cv, waking whenever refs are enqueued or the executor is stopped.

12. Duplicate/early GPU node-completed events (ago_interface.cpp) — reviewed the event emission points. VX_EVENT_NODE_COMPLETED is emitted once per node execution in the CPU path and once after GPU synchronization in the GPU path, not twice.

Also fixed a pre-existing flaky race in the graph-scheduling thread that caused GraphPipeline.ManualSchedule/8 to hang in combined CTS runs:

• Made the graph thread counters atomic (threadScheduleCount, threadExecuteCount, etc.).
• Fixed WaitForSingleObject for semaphores so it does not lose wake-ups when a release happens before the wait.
• Removed the spurious ReleaseSemaphore calls in agoWaitGraph that were generating phantom graph executions and desynchronizing the schedule/wait counts.
• Made the graph-thread loop resilient to spurious semaphore returns.

Verification:

• GraphPipeline.*:GraphStreaming.* combined: 133/133 pass (no AddressSanitizer leaks).
• GraphPipeline.ManualSchedule/8 after Graph.VirtualImage: 20/20 stress runs pass.
• Legacy Graph.VirtualImage repro and And_alt.gdf still pass.

[Copilot]

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated no new comments.

Comments suppressed due to low confidence (7)

amd_openvx/openvx/api/vx_pipelining_api.cpp:230

• vxGraphParameterDequeueDoneRef() returns VX_FAILURE on timeout, but the extension spec (vx_khr_pipelining.h) requires returning VX_ERROR_TIMEOUT when VX_GRAPH_TIMEOUT is finite and the wait expires.

            if (!q->done_cv.wait_for(lock, std::chrono::milliseconds(pipe->timeout_ms),
                                       [q]() { return !q->done_refs.empty(); }))
                return VX_FAILURE;
        }

amd_openvx/openvx/api/vx_pipelining_api.cpp:313

• vxWaitEvent() returns VX_FAILURE on timeout, but the extension spec requires VX_ERROR_TIMEOUT when VX_CONTEXT_EVENT_TIMEOUT is finite and the wait expires.

        } else {
            if (!evsys->events_cv.wait_for(lock, std::chrono::milliseconds(evsys->timeout_ms),
                                             [&evsys]() { return !evsys->events.empty(); }))
                return VX_FAILURE;
        }

amd_openvx/openvx/api/vx_pipelining_api.cpp:412

• vxWaitGraphEvent() is currently implemented as a thin wrapper over vxWaitEvent() on the context. That means it can return events from other graphs (no graph filtering) and it ignores the VX_GRAPH_EVENT_TIMEOUT attribute (spec requires using the graph-level timeout for this API).

VX_API_ENTRY vx_status VX_API_CALL vxWaitGraphEvent(vx_graph graph, vx_event_t *event, vx_bool do_not_block)
{
    if (!agoIsValidGraph((AgoGraph *)graph) || !event)
        return VX_ERROR_INVALID_REFERENCE;
    AgoContext * context = ((AgoGraph *)graph)->ref.context;
    return vxWaitEvent(context, event, do_not_block);
}

amd_openvx/openvx/api/vx_pipelining_api.cpp:192

• vxGraphParameterEnqueueReadyRef() pushes into q->ready_refs but never enforces q->max_depth (which is set via vxSetGraphScheduleConfig / first enqueue). This allows unbounded growth and breaks the queue-depth contract; once the queue is full the function should fail (VX_FAILURE) instead of silently accepting more refs.

            std::lock_guard<std::mutex> lock(q->mtx);
            q->ready_refs.push_back((AgoData *)refs[i]);
        }

amd_openvx/openvx/api/vx_pipelining_api.cpp:395

• vxRegisterEvent() currently (1) allows registering events after vxVerifyGraph (spec requires registration before verification), (2) does not validate that the event type matches the reference type (graph vs node), and (3) appends duplicate registrations instead of updating the stored app_value when the same event is registered twice (spec says duplicates should update). This can lead to incorrect app_value resolution and unexpected duplicate registrations.

This issue also appears on line 406 of the same file.

VX_API_ENTRY vx_status VX_API_CALL vxRegisterEvent(vx_reference ref, enum vx_event_type_e type, vx_uint32 param, vx_uint32 app_value)
{
    if (!ref || !agoIsValidReference((AgoReference *)ref))
        return VX_ERROR_INVALID_REFERENCE;
    AgoReference * r = (AgoReference *)ref;
    AgoContextEventSystem * evsys = agoGetContextEventSystem(r->context);
    if (!evsys)
        return VX_FAILURE;
    if (type != VX_EVENT_GRAPH_PARAMETER_CONSUMED &&
        type != VX_EVENT_GRAPH_COMPLETED &&
        type != VX_EVENT_NODE_COMPLETED &&
        type != VX_EVENT_NODE_ERROR) {
        return VX_ERROR_NOT_SUPPORTED;
    }

    AgoEventRegistration reg;
    reg.ref = ref;
    reg.event_type = type;
    reg.app_value = app_value;
    reg.graph_parameter_index = (type == VX_EVENT_GRAPH_PARAMETER_CONSUMED) ? param : 0;
    std::lock_guard<std::mutex> lock(evsys->registrations_mtx);
    evsys->registrations.push_back(reg);
    return VX_SUCCESS;
}

amd_openvx/openvx/ago/ago_pipelining.cpp:448

• agoMoveConsumedRefsToDone() emits VX_EVENT_GRAPH_PARAMETER_CONSUMED whenever done_refs is non-empty. Since done_refs can remain non-empty across executions until the app dequeues, this will re-emit "consumed" events on subsequent executions even when nothing was consumed in that execution.

        if (!q->done_refs.empty()) {
            // Notify that a reference at this parameter was consumed during this execution.
            agoNotifyGraphParameterConsumed(graph, q->index);
        }

amd_openvx/openvx/api/vx_api.cpp:2871

• VX_KERNEL_PIPEUP_INPUT_DEPTH can be set/queried on the kernel, but the value is never propagated to nodes or used by the scheduler/executor. As-is, the attribute is effectively a no-op, which can break kernels that rely on the pipeup input-depth semantics described in vx_khr_pipelining.h (e.g., sink nodes retaining inputs during PIPEUP).

            case VX_KERNEL_PIPEUP_INPUT_DEPTH:
                if (size == sizeof(vx_uint32)) {
                    vx_uint32 v = *(const vx_uint32 *)ptr;
                    if (v < 1 || kernel->finalized) status = VX_ERROR_INVALID_PARAMETERS;
                    else { kernel->pipeup_input_depth = v; status = VX_SUCCESS; }
                }
                break;