diff --git a/.github/workflows/conformance-hip.yml b/.github/workflows/conformance-hip.yml index f04757b51..3f7378840 100644 --- a/.github/workflows/conformance-hip.yml +++ b/.github/workflows/conformance-hip.yml @@ -284,6 +284,8 @@ jobs: -DOPENVX_INCLUDES="$GITHUB_WORKSPACE/amd_openvx/openvx/include" \ -DOPENVX_LIBRARIES="$GITHUB_WORKSPACE/install/lib/libopenvx.so;$GITHUB_WORKSPACE/install/lib/libvxu.so;pthread;dl;m;rt" \ -DOPENVX_CONFORMANCE_VISION=ON \ + -DOPENVX_USE_PIPELINING=ON \ + -DOPENVX_USE_STREAMING=ON \ -DCMAKE_C_STANDARD_LIBRARIES="-L${ROCM_PATH}/lib -lamdhip64" \ -DCMAKE_CXX_STANDARD_LIBRARIES="-L${ROCM_PATH}/lib -lamdhip64" \ -DCMAKE_EXE_LINKER_FLAGS="-Wl,-rpath-link,${ROCM_PATH}/lib" @@ -1031,6 +1033,88 @@ jobs: path: profraw/ if-no-files-found: ignore + pipelining-hip: + name: KHR EXT - Pipelining (HIP) + runs-on: + group: linux-shark39-runner-group + container: + image: mivisionx/ubuntu-24.04:rocm-20260710 + options: --device /dev/kfd --device /dev/dri + needs: build-hip-debug + steps: + - name: Download debug build artifacts + uses: actions/download-artifact@v4 + with: + name: build-artifacts-hip-debug + - name: Validate ROCm + run: | + rocminfo + amd-smi || true + - name: Run pipelining tests + run: | + mkdir -p profraw + chmod +x OpenVX-cts/build/bin/vx_test_conformance + cd OpenVX-cts/build + export LD_LIBRARY_PATH=$GITHUB_WORKSPACE/install/lib:${ROCM_PATH}/lib:$GITHUB_WORKSPACE/OpenVX-cts/build/lib + export VX_TEST_DATA_PATH=$GITHUB_WORKSPACE/OpenVX-cts/test_data/ + export HIP_VISIBLE_DEVICES=0 # ROCR_VISIBLE_DEVICES already pins the assigned GPU; the runner also sets HIP_VISIBLE_DEVICES to the same host index, which stacks and yields hipErrorNoDevice on index-1 GPUs. After ROCR filtering the device is index 0 for HIP. + export AGO_LOG_STDERR=1 + export LLVM_PROFILE_FILE=$GITHUB_WORKSPACE/profraw/pipelining_%p.profraw + for target in CPU GPU; do + echo "::group::pipelining (AGO_DEFAULT_TARGET=$target)" + AGO_DEFAULT_TARGET=$target timeout 1800 ./bin/vx_test_conformance --filter="GraphPipeline.*" + echo "::endgroup::" + done + - name: Upload profraw + if: always() + continue-on-error: true + uses: actions/upload-artifact@v4 + with: + name: profraw-hip-pipelining + path: profraw/ + if-no-files-found: ignore + + streaming-hip: + name: KHR EXT - Streaming (HIP) + runs-on: + group: linux-shark39-runner-group + container: + image: mivisionx/ubuntu-24.04:rocm-20260710 + options: --device /dev/kfd --device /dev/dri + needs: build-hip-debug + steps: + - name: Download debug build artifacts + uses: actions/download-artifact@v4 + with: + name: build-artifacts-hip-debug + - name: Validate ROCm + run: | + rocminfo + amd-smi || true + - name: Run streaming tests + run: | + mkdir -p profraw + chmod +x OpenVX-cts/build/bin/vx_test_conformance + cd OpenVX-cts/build + export LD_LIBRARY_PATH=$GITHUB_WORKSPACE/install/lib:${ROCM_PATH}/lib:$GITHUB_WORKSPACE/OpenVX-cts/build/lib + export VX_TEST_DATA_PATH=$GITHUB_WORKSPACE/OpenVX-cts/test_data/ + export HIP_VISIBLE_DEVICES=0 # ROCR_VISIBLE_DEVICES already pins the assigned GPU; the runner also sets HIP_VISIBLE_DEVICES to the same host index, which stacks and yields hipErrorNoDevice on index-1 GPUs. After ROCR filtering the device is index 0 for HIP. + export AGO_LOG_STDERR=1 + export LLVM_PROFILE_FILE=$GITHUB_WORKSPACE/profraw/streaming_%p.profraw + for target in CPU GPU; do + echo "::group::streaming (AGO_DEFAULT_TARGET=$target)" + AGO_DEFAULT_TARGET=$target timeout 900 ./bin/vx_test_conformance --filter="GraphStreaming.*" + echo "::endgroup::" + done + - name: Upload profraw + if: always() + continue-on-error: true + uses: actions/upload-artifact@v4 + with: + name: profraw-hip-streaming + path: profraw/ + if-no-files-found: ignore + # --- Local Tests (API + GDF) --- api-tests-hip: diff --git a/.github/workflows/conformance.yml b/.github/workflows/conformance.yml index 0f108ad6a..57770f309 100644 --- a/.github/workflows/conformance.yml +++ b/.github/workflows/conformance.yml @@ -158,7 +158,9 @@ jobs: -DCMAKE_POLICY_VERSION_MINIMUM=3.5 \ -DOPENVX_INCLUDES="${{ github.workspace }}/amd_openvx/openvx/include" \ -DOPENVX_LIBRARIES="${INSTALL_PREFIX}/lib/libopenvx.so;${INSTALL_PREFIX}/lib/libvxu.so;pthread;dl;m;rt" \ - -DOPENVX_CONFORMANCE_VISION=ON + -DOPENVX_CONFORMANCE_VISION=ON \ + -DOPENVX_USE_PIPELINING=ON \ + -DOPENVX_USE_STREAMING=ON make -j$(nproc) - name: Upload debug build artifacts @@ -637,6 +639,60 @@ jobs: path: profraw/ if-no-files-found: ignore + pipelining-cpu: + name: KHR EXT - Pipelining (CPU) + runs-on: ubuntu-24.04 + needs: build-debug + steps: + - uses: actions/checkout@v4 + - name: Download debug build artifacts + uses: actions/download-artifact@v4 + with: + name: build-artifacts-debug + - name: Run pipelining tests + run: | + mkdir -p profraw + chmod +x OpenVX-cts/build/bin/vx_test_conformance + cd OpenVX-cts/build + export LD_LIBRARY_PATH=${{ github.workspace }}/install/lib:${{ github.workspace }}/OpenVX-cts/build/lib + export VX_TEST_DATA_PATH=${{ github.workspace }}/OpenVX-cts/test_data/ + export LLVM_PROFILE_FILE=${{ github.workspace }}/profraw/pipelining_%p.profraw + timeout 1800 ./bin/vx_test_conformance --filter="GraphPipeline.*" + - name: Upload profraw + if: always() + uses: actions/upload-artifact@v4 + with: + name: profraw-pipelining + path: profraw/ + if-no-files-found: ignore + + streaming-cpu: + name: KHR EXT - Streaming (CPU) + runs-on: ubuntu-24.04 + needs: build-debug + steps: + - uses: actions/checkout@v4 + - name: Download debug build artifacts + uses: actions/download-artifact@v4 + with: + name: build-artifacts-debug + - name: Run streaming tests + run: | + mkdir -p profraw + chmod +x OpenVX-cts/build/bin/vx_test_conformance + cd OpenVX-cts/build + export LD_LIBRARY_PATH=${{ github.workspace }}/install/lib:${{ github.workspace }}/OpenVX-cts/build/lib + export VX_TEST_DATA_PATH=${{ github.workspace }}/OpenVX-cts/test_data/ + export LLVM_PROFILE_FILE=${{ github.workspace }}/profraw/streaming_%p.profraw + timeout 900 ./bin/vx_test_conformance --filter="GraphStreaming.*" + - name: Upload profraw + if: always() + uses: actions/upload-artifact@v4 + with: + name: profraw-streaming + path: profraw/ + if-no-files-found: ignore + # --- Local Tests (API + GDF) --- api-tests: diff --git a/amd_openvx/openvx/CMakeLists.txt b/amd_openvx/openvx/CMakeLists.txt index 5c76524b9..6e47f9eb4 100644 --- a/amd_openvx/openvx/CMakeLists.txt +++ b/amd_openvx/openvx/CMakeLists.txt @@ -32,6 +32,13 @@ set(VERSION "1.3.2") project(openvx VERSION ${VERSION} LANGUAGES CXX) +option(OPENVX_USE_PIPELINING "Enable OpenVX Graph Pipelining/Streaming/Event extension (vx_khr_pipelining)" ON) +if(OPENVX_USE_PIPELINING) + add_definitions(-DOPENVX_USE_PIPELINING=1) +else() + add_definitions(-DOPENVX_USE_PIPELINING=0) +endif() + include_directories(include ago api) list(APPEND SOURCES @@ -63,6 +70,7 @@ list(APPEND SOURCES ago/ago_interface.cpp ago/ago_kernel_api.cpp ago/ago_kernel_list.cpp + ago/ago_pipelining.cpp ago/ago_platform.cpp ago/ago_util.cpp ago/ago_util_opencl.cpp @@ -70,6 +78,7 @@ list(APPEND SOURCES api/vxu.cpp api/vx_api.cpp api/vx_nodes.cpp + api/vx_pipelining_api.cpp ) add_library(openvx SHARED ${SOURCES}) diff --git a/amd_openvx/openvx/ago/ago_interface.cpp b/amd_openvx/openvx/ago/ago_interface.cpp index 2e5e88a21..1bae77cec 100644 --- a/amd_openvx/openvx/ago/ago_interface.cpp +++ b/amd_openvx/openvx/ago/ago_interface.cpp @@ -31,20 +31,23 @@ static void agoGraphThreadFunction(LPVOID graph_) #endif { AgoGraph * graph = (AgoGraph *)graph_; - while (WaitForSingleObject(graph->hSemToThread, INFINITE) == WAIT_OBJECT_0) { - graph->threadThreadWaitState = 2; - if (graph->threadThreadTerminationState) + for (;;) { + DWORD w = WaitForSingleObject(graph->hSemToThread, INFINITE); + if (graph->threadThreadTerminationState.load()) break; + // Ignore spurious/error returns from the semaphore wait and retry. + if (w != WAIT_OBJECT_0) + continue; // execute graph graph->status = agoProcessGraph(graph); // inform caller - graph->threadExecuteCount++; + graph->threadExecuteCount.fetch_add(1); ReleaseSemaphore(graph->hSemFromThread, 1, nullptr); } // inform caller about termination - graph->threadThreadTerminationState = 2; + graph->threadThreadTerminationState.store(2); ReleaseSemaphore(graph->hSemFromThread, 1, nullptr); #if _WIN32 return 0; @@ -159,8 +162,32 @@ AgoGraph * agoCreateGraph(AgoContext * acontext) return (AgoGraph *)agraph; } +// Signals the graph scheduling thread to exit and waits for it to do so. Must be called +// without holding the graph or context critical section: the thread executes the graph +// inside both of them, so joining it from within either one deadlocks. +static void agoStopGraphThread(AgoGraph * agraph) +{ + if (!agraph->hThread || agraph->threadThreadTerminationState.load() == 2) + return; + agraph->threadThreadTerminationState.store(1); + ReleaseSemaphore(agraph->hSemToThread, 1, nullptr); + while (agraph->threadThreadTerminationState.load() == 1) { + // give a chance for the thread to run in case it is waititng + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + ReleaseSemaphore(agraph->hSemToThread, 1, nullptr); + } +} + int agoReleaseGraph(AgoGraph * agraph) { + // Stop every thread that executes this graph before taking any lock, so that the + // graph is not being torn down underneath them and so that joining them cannot + // deadlock against a section they are trying to enter. + if (agraph->ref.external_count <= 1) { + agoStopGraphPipelining(agraph); + agoStopGraphThread(agraph); + } + CAgoLock lock(agraph->ref.context->cs); int status = 0; @@ -169,15 +196,7 @@ int agoReleaseGraph(AgoGraph * agraph) agraph->ref.context->num_active_references--; if (agraph->ref.external_count == 0) { EnterCriticalSection(&agraph->cs); - // stop graph thread if (agraph->hThread) { - agraph->threadThreadTerminationState = 1; - ReleaseSemaphore(agraph->hSemToThread, 1, nullptr); - while (agraph->threadThreadTerminationState == 1) { - // give a chance for the thread to run in case it is waititng - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - ReleaseSemaphore(agraph->hSemToThread, 1, nullptr); - } if (agraph->hSemToThread) { CloseHandle(agraph->hSemToThread); } @@ -1368,8 +1387,15 @@ vx_status agoVerifyNode(AgoNode * node) if (data) { if ((kernel->argConfig[arg] & (AGO_KERNEL_ARG_INPUT_FLAG | AGO_KERNEL_ARG_OUTPUT_FLAG)) == AGO_KERNEL_ARG_OUTPUT_FLAG) { vx_meta_format meta = &node->metaList[arg]; + // For user kernels without a validate callback, infer output meta from the + // bound object so source/sink kernels can verify. + if (kernel->user_kernel && !kernel->validate_f && data && kernel->argType[arg] && kernel->argType[arg] != VX_TYPE_REFERENCE) { + meta->data.ref.type = data->ref.type; + meta->data.u = data->u; + } if (kernel->argType[arg] && kernel->argType[arg] != VX_TYPE_REFERENCE && (meta->data.ref.type != kernel->argType[arg])) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: output argument type mismatch for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } else if (meta->data.ref.type == VX_TYPE_IMAGE) { @@ -1520,6 +1546,7 @@ vx_status agoVerifyNode(AgoNode * node) // make sure that the data come from output validator matches with object if (data->u.arr.itemtype != meta->data.u.arr.itemtype) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid array type for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } else if (!data->u.arr.capacity || (meta->data.u.arr.capacity && meta->data.u.arr.capacity > data->u.arr.capacity)) { @@ -1552,6 +1579,7 @@ vx_status agoVerifyNode(AgoNode * node) // make sure that the data come from output validator matches with object if (data->u.objarr.itemtype != meta->data.u.objarr.itemtype) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid object-array type for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } else if (!data->u.objarr.numitems || (meta->data.u.objarr.numitems && meta->data.u.objarr.numitems > data->u.objarr.numitems)) { @@ -1572,6 +1600,7 @@ vx_status agoVerifyNode(AgoNode * node) // make sure that the data come from output validator matches with object if (data->u.scalar.type != meta->data.u.scalar.type) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid type for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } } @@ -1579,6 +1608,7 @@ vx_status agoVerifyNode(AgoNode * node) // make sure that the data come from output validator matches with object if ((data->u.mat.type != meta->data.u.mat.type) || (data->u.mat.columns != meta->data.u.mat.columns) || (data->u.mat.rows != meta->data.u.mat.rows)) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid matrix meta for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } } @@ -1588,6 +1618,7 @@ vx_status agoVerifyNode(AgoNode * node) else if (meta->data.ref.type == VX_TYPE_THRESHOLD) { if ((data->u.thr.thresh_type != meta->data.u.thr.thresh_type)) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid threshold meta for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } } @@ -1609,6 +1640,7 @@ vx_status agoVerifyNode(AgoNode * node) } if (mismatched) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid tensor meta for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } } @@ -1623,6 +1655,7 @@ vx_status agoVerifyNode(AgoNode * node) } else if (kernel->argType[arg]) { agoAddLogEntry(&kernel->ref, VX_ERROR_INVALID_TYPE, "ERROR: agoVerifyGraph: kernel %s: invalid type for argument#%d\n", kernel->name, arg); + return VX_ERROR_INVALID_TYPE; } } @@ -1696,7 +1729,6 @@ int agoVerifyGraph(AgoGraph * graph) graph->enable_node_level_gpu_flush = false; } #endif - return status; } @@ -1837,11 +1869,11 @@ vx_status agoComputeImageValidRectangleOutputs(AgoGraph * graph) AgoData * data = node->paramList[i]; if (data && param->direction == VX_OUTPUT) { if (data->ref.type == VX_TYPE_IMAGE) { - printf("valid_rect [ %5d %5d %5d %5d ] image %s\n", data->u.img.rect_valid.start_x, data->u.img.rect_valid.start_y, data->u.img.rect_valid.end_x, data->u.img.rect_valid.end_y, data->name.c_str()); + fprintf(stderr,"valid_rect [ %5d %5d %5d %5d ] image %s\n", data->u.img.rect_valid.start_x, data->u.img.rect_valid.start_y, data->u.img.rect_valid.end_x, data->u.img.rect_valid.end_y, data->name.c_str()); } else if (data->ref.type == VX_TYPE_PYRAMID) { for (vx_size level = 0; level < data->u.pyr.levels; level++) { - printf("valid_rect [ %5d %5d %5d %5d ] pyrL%d %s\n", data->children[level]->u.img.rect_valid.start_x, data->children[level]->u.img.rect_valid.start_y, data->children[level]->u.img.rect_valid.end_x, data->children[level]->u.img.rect_valid.end_y, (int)level, data->name.c_str()); + fprintf(stderr,"valid_rect [ %5d %5d %5d %5d ] pyrL%d %s\n", data->children[level]->u.img.rect_valid.start_x, data->children[level]->u.img.rect_valid.start_y, data->children[level]->u.img.rect_valid.end_x, data->children[level]->u.img.rect_valid.end_y, (int)level, data->name.c_str()); } } } @@ -2483,8 +2515,10 @@ int agoExecuteGraph(AgoGraph * graph) if (status) { if (status == VX_ERROR_GRAPH_ABANDONED) agoAddLogEntry((vx_reference)graph, VX_FAILURE, "INFO: kernel %s exec returned graph_stopped status: VX_ERROR_GRAPH_ABANDONED (%d)\n", kernel->name, status); - else + else { + agoNotifyNodeError(graph, node, status); agoAddLogEntry((vx_reference)graph, VX_FAILURE, "ERROR: kernel %s exec failed (%d:%s)\n", kernel->name, status, agoEnum2Name(status)); + } return status; } agoPerfCaptureStop(&node->perf); @@ -2525,6 +2559,7 @@ int agoExecuteGraph(AgoGraph * graph) return VX_ERROR_GRAPH_ABANDONED; } } + agoNotifyNodeCompleted(graph, node); } } } @@ -2552,6 +2587,11 @@ int agoExecuteGraph(AgoGraph * graph) } #endif } + // Notify completion for GPU nodes launched in this graph execution. + for (AgoNode * node = graph->nodeList.head; node; node = node->next) { + if (node->attr_affinity.device_type == AGO_KERNEL_FLAG_DEVICE_GPU) + agoNotifyNodeCompleted(graph, node); + } agoPerfProfileEntry(graph, ago_profile_type_wait_end, &graph->ref); graph->gpu_perf_total.kernel_enqueue += graph->gpu_perf.kernel_enqueue; graph->gpu_perf_total.kernel_wait += graph->gpu_perf.kernel_wait; @@ -2572,7 +2612,14 @@ int agoExecuteGraph(AgoGraph * graph) if (status == VX_SUCCESS) graph->state = VX_GRAPH_STATE_COMPLETED; - + // Advance streaming node state for each executed node. + for (AgoNode * node = graph->nodeList.head; node; node = node->next) { + node->node_exec_count++; + vx_uint32 threshold = node->pipeup_output_depth > 0 ? (node->pipeup_output_depth - 1) : 0; + if (node->node_exec_count >= threshold && node->node_state == VX_NODE_STATE_PIPEUP) { + node->node_state = VX_NODE_STATE_STEADY; + } + } return status; } @@ -2843,6 +2890,40 @@ vx_status agoGraphDumpPerformanceProfile(AgoGraph * graph, const char * fileName return VX_SUCCESS; } +static void agoPreFillSourceNodePipeup(AgoGraph * graph) +{ + for (AgoNode * node = graph->nodeList.head; node; node = node->next) { + vx_uint32 depth = node->pipeup_output_depth; + if (depth <= 1) + continue; + // source nodes have no input parameters + bool has_input = false; + for (vx_uint32 i = 0; i < node->paramCount; i++) { + if (node->parameters[i].direction == VX_INPUT) { + has_input = true; + break; + } + } + if (has_input) + continue; + // execute the node in pipeup state until one frame before steady + vx_uint32 target = depth > 0 ? (depth - 1) : 0; + + AgoKernel * kernel = node->akernel; + while (node->node_exec_count < target) { + vx_status s = VX_SUCCESS; + if (kernel && kernel->kernel_f) { + s = kernel->kernel_f(node, (vx_reference *)node->paramList, node->paramCount); + } + if (s != VX_SUCCESS) + break; + node->node_exec_count++; + } + if (node->node_exec_count >= target && node->node_state == VX_NODE_STATE_PIPEUP) + node->node_state = VX_NODE_STATE_STEADY; + } +} + int agoProcessGraph(AgoGraph * graph) { vx_status status = VX_ERROR_INVALID_REFERENCE; @@ -2855,7 +2936,24 @@ int agoProcessGraph(AgoGraph * graph) } // execute graph if possible if (status == VX_SUCCESS) { - if (graph->verified && graph->isReadyToExecute) { + if (graph->verified && graph->pipelining) { + AgoGraphPipeliningState * pipe = graph->pipelining; + if (pipe->schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_MANUAL || + pipe->schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO || + pipe->streaming_enabled) { + status = agoExecuteGraphPipelined(graph); + } + else if (graph->isReadyToExecute) { + status = agoExecuteGraph(graph); + } + else { + agoAddLogEntry(&graph->ref, VX_FAILURE, "ERROR: agoProcessGraph: not verified (%d) or not ready to execute (%d)\n", graph->verified, graph->isReadyToExecute); + status = VX_FAILURE; + } + } + else if (graph->verified && graph->isReadyToExecute) { + // For non-streaming execution, pre-fill source-node pipeup queues. + agoPreFillSourceNodePipeup(graph); status = agoExecuteGraph(graph); } else { @@ -2872,7 +2970,6 @@ int agoScheduleGraph(AgoGraph * graph) vx_status status = VX_ERROR_INVALID_REFERENCE; if (agoIsValidGraph(graph)) { status = VX_SUCCESS; - graph->threadScheduleCount++; if (graph->hThread) { if (!graph->verified) { // make sure to verify the graph in master thread @@ -2880,13 +2977,18 @@ int agoScheduleGraph(AgoGraph * graph) status = vxVerifyGraph(graph); } if (status == VX_SUCCESS) { + // count the request before waking the graph thread, so a waiter can + // never observe the completion token without the matching request + graph->threadScheduleCount.fetch_add(1); // inform graph thread to execute if (!ReleaseSemaphore(graph->hSemToThread, 1, nullptr)) { + graph->threadScheduleCount.fetch_sub(1); status = VX_ERROR_NO_RESOURCES; } } } else { + graph->threadScheduleCount.fetch_add(1); status = agoProcessGraph(graph); } } @@ -2898,27 +3000,94 @@ int agoWaitGraph(AgoGraph * graph) vx_status status = VX_ERROR_INVALID_REFERENCE; if (agoIsValidGraph(graph)) { status = VX_SUCCESS; - graph->threadWaitCount++; - if (graph->threadScheduleCount <= 0) // the graph was never scheduled so return VX_FAILURE + if (graph->pipelining) { + AgoGraphPipeliningState * pipe = graph->pipelining; + // Streaming graphs are driven by the streaming thread; vxWaitGraph just + // needs to observe that no execution is active. + if (pipe->streaming_enabled) { + while (!pipe->streaming_stop.load()) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + // wait for streaming thread to finish + if (pipe->streaming_thread.joinable()) + pipe->streaming_thread.join(); + return status; + } + // QUEUE_AUTO: stop the background executor, drain any refs that were + // enqueued but not yet processed, then restart the executor so future + // enqueues continue to be handled automatically. + if (pipe->schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO) { + agoStopGraphPipelining(graph); + { + CAgoLock lock(graph->cs); + for (;;) { + bool any_ready = false; + for (auto& q : pipe->param_queues) { + if (q->enabled) { + std::lock_guard qlock(q->mtx); + if (!q->ready_refs.empty()) { + any_ready = true; + break; + } + } + } + if (!any_ready) + break; + int exec_status = agoExecutePipelinedGraphOnce(graph); + if (exec_status != VX_SUCCESS) { + status = exec_status; + break; + } + } + } + agoStartGraphPipeliningAutoExecutor(graph); + return status; + } + // QUEUE_MANUAL without a graph thread: drain synchronously. + if (!graph->hThread && pipe->schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_MANUAL) { + CAgoLock lock(graph->cs); + for (;;) { + bool any_ready = false; + for (auto& q : pipe->param_queues) { + if (q->enabled) { + std::lock_guard qlock(q->mtx); + if (!q->ready_refs.empty()) { + any_ready = true; + break; + } + } + } + if (!any_ready) + break; + int exec_status = agoExecuteGraphPipelined(graph); + if (exec_status != VX_SUCCESS) { + status = exec_status; + break; + } + } + return status; + } + } + graph->threadWaitCount.fetch_add(1); + if (graph->threadScheduleCount.load() <= 0) // the graph was never scheduled so return VX_FAILURE return VX_FAILURE; if (graph->hThread) { - graph->threadThreadWaitState = 1; - while (graph->threadThreadWaitState == 1) { - // wait for the agoGraphThreadFunction to be done - std::this_thread::sleep_for(std::chrono::milliseconds(1)); - // release the semaphore in case the agoScheduleGraph was called before the agoGraphThreadFunction - ReleaseSemaphore(graph->hSemToThread, 1, nullptr); - } - while (graph->threadExecuteCount < graph->threadScheduleCount) { + // The graph thread posts exactly one completion token per scheduled execution, + // so claim a token for every execution scheduled so far that no earlier wait has + // accounted for. Counting tokens rather than polling thread state keeps this + // correct no matter whether the graph thread runs before or after this call. + vx_int32 target = graph->threadScheduleCount.load(); + while (graph->threadCompletionCount.load() < target) { if (WaitForSingleObject(graph->hSemFromThread, INFINITE) != WAIT_OBJECT_0) { agoAddLogEntry(&graph->ref, VX_FAILURE, "ERROR: agoWaitGraph: WaitForSingleObject failed\n"); status = VX_FAILURE; break; } + graph->threadCompletionCount.fetch_add(1); } } if(status == VX_SUCCESS) status = graph->status; } return status; -} \ No newline at end of file +} diff --git a/amd_openvx/openvx/ago/ago_internal.h b/amd_openvx/openvx/ago/ago_internal.h index dce5b1332..24f64abaa 100644 --- a/amd_openvx/openvx/ago/ago_internal.h +++ b/amd_openvx/openvx/ago/ago_internal.h @@ -546,6 +546,8 @@ struct AgoKernel { vx_uint32 gpu_buffer_update_param_index; vx_bool opencl_buffer_access_enable; vx_uint32 importing_module_index_plus1; + vx_uint32 pipeup_output_depth; + vx_uint32 pipeup_input_depth; public: AgoKernel(); ~AgoKernel(); @@ -616,6 +618,9 @@ struct AgoNode { vx_uint32 hierarchical_level; vx_status status; vx_perf_t perf; + vx_uint32 node_state; + vx_uint32 node_exec_count; + vx_uint32 pipeup_output_depth; vx_bool local_data_change_is_enabled; vx_bool local_data_set_by_implementation; struct { bool enable; int paramIndexScalar; int paramIndexArray; } gpu_scalar_array_output_sync; @@ -663,13 +668,83 @@ struct AgoNodeList { AgoNode * tail; AgoNode * trash; }; +struct AgoGraphParameterQueue { + std::mutex mtx; + std::condition_variable done_cv; + std::deque ready_refs; + std::deque consumed_refs; + std::deque done_refs; + std::vector valid_refs; + vx_uint32 index; + vx_uint32 max_depth; + bool enabled; + AgoGraphParameterQueue() : index(0), max_depth(0), enabled(false) {} +}; + +struct AgoGraphPipeliningState { + vx_enum schedule_mode; + vx_uint32 timeout_ms; + vx_uint32 event_timeout_ms; + vx_uint32 pipeline_depth; + bool streaming_enabled; + AgoNode * trigger_node; + std::atomic streaming_stop; + std::thread streaming_thread; + std::atomic active_executions; + std::mutex active_mtx; + std::condition_variable active_cv; + std::mutex execution_mtx; + std::thread executor_thread; + std::atomic executor_stop; + std::mutex enqueue_mtx; + std::condition_variable enqueue_cv; + std::vector> param_queues; +public: + AgoGraphPipeliningState(); + ~AgoGraphPipeliningState(); +}; + +struct AgoEvent { + vx_enum event_type; + vx_uint64 timestamp; + vx_uint64 app_value; + AgoGraph * graph; + AgoNode * node; + vx_uint32 graph_parameter_index; + vx_status status; + void * user_parameter; +}; + +struct AgoEventRegistration { + vx_reference ref; + vx_enum event_type; + vx_uint32 app_value; + vx_uint32 graph_parameter_index; +}; + +struct AgoContextEventSystem { + std::mutex events_mtx; + std::condition_variable events_cv; + std::deque events; + std::mutex registrations_mtx; + std::vector registrations; + bool enabled; + vx_uint32 timeout_ms; +public: + AgoContextEventSystem(); + ~AgoContextEventSystem(); +}; + struct AgoGraph { AgoReference ref; std::string name; AgoGraph * next; CRITICAL_SECTION cs; HANDLE hThread, hSemToThread, hSemFromThread; - vx_int32 threadScheduleCount, threadExecuteCount, threadWaitCount, threadThreadTerminationState, threadThreadWaitState; + // threadScheduleCount counts executions handed to the graph thread, threadExecuteCount + // counts executions it finished, and threadCompletionCount counts the completions already + // claimed by a waiter -- one completion token on hSemFromThread per scheduled execution. + std::atomic threadScheduleCount, threadExecuteCount, threadCompletionCount, threadWaitCount, threadThreadTerminationState; AgoDataList dataList; AgoNodeList nodeList; vx_bool isReadyToExecute; @@ -707,6 +782,7 @@ struct AgoGraph { bool enable_performance_profiling; std::vector performance_profile; std::map moduleHandle; + AgoGraphPipeliningState * pipelining; public: AgoGraph(); ~AgoGraph(); @@ -791,6 +867,7 @@ struct AgoContext { vx_size hip_mem_release_count; #endif AgoTargetAffinityInfo_ attr_affinity; + AgoContextEventSystem * events; public: AgoContext(); ~AgoContext(); @@ -907,6 +984,24 @@ void agoPerfCopyNormalize(AgoContext * context, vx_perf_t * perfDst, vx_perf_t * // log void agoRegisterLogCallback(vx_context context, vx_log_callback_f callback, vx_bool reentrant); void agoAddLogEntry(AgoReference * ref, vx_status status, const char *message, ...); +// pipelining +AgoGraphPipeliningState * agoGetGraphPipeliningState(AgoGraph * graph); +AgoContextEventSystem * agoGetContextEventSystem(AgoContext * context); +void agoStopGraphPipelining(AgoGraph * graph); +void agoStartGraphPipeliningAutoExecutor(AgoGraph * graph); +void agoStartGraphStreamingThread(AgoGraph * graph); +void agoPushEvent(AgoContext * context, const AgoEvent& evt); +int agoExecuteGraphPipelined(AgoGraph * graph); +int agoExecutePipelinedGraphOnce(AgoGraph * graph); +int agoExecuteGraphQueueManual(AgoGraph * graph); +// event notifications +void agoNotifyGraphCompleted(AgoGraph * graph); +void agoNotifyNodeCompleted(AgoGraph * graph, AgoNode * node); +void agoNotifyNodeError(AgoGraph * graph, AgoNode * node, vx_status status); +void agoNotifyGraphParameterConsumed(AgoGraph * graph, vx_uint32 graph_parameter_index); +bool agoGraphHasNodeEventRegistrations(AgoGraph * graph); +vx_uint32 agoGetReferenceEnqueueCount(AgoContext * context, AgoReference * ref); +void agoRemoveEventRegistrations(AgoContext * context, vx_reference ref); #if (ENABLE_OPENCL || ENABLE_HIP) int agoGpuOclAllocBuffers(AgoGraph * graph); diff --git a/amd_openvx/openvx/ago/ago_pipelining.cpp b/amd_openvx/openvx/ago/ago_pipelining.cpp new file mode 100644 index 000000000..e1219172b --- /dev/null +++ b/amd_openvx/openvx/ago/ago_pipelining.cpp @@ -0,0 +1,681 @@ +/* +Copyright (c) 2015 - 2026 Advanced Micro Devices, Inc. All rights reserved. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. +*/ + +#include "ago_internal.h" +#include + +// +// OpenVX Pipelining Extension - AGO internal helpers +// +// This file implements the core state management and execution helpers for +// the Khronos OpenVX pipelining/streaming/event-queue extension +// (vx_khr_pipelining.h). +// + +#if OPENVX_USE_PIPELINING + +AgoGraphPipeliningState * agoGetGraphPipeliningState(AgoGraph * graph) +{ + if (!graph) + return nullptr; + if (!graph->pipelining) { + graph->pipelining = new AgoGraphPipeliningState(); + } + return graph->pipelining; +} + +AgoContextEventSystem * agoGetContextEventSystem(AgoContext * context) +{ + if (!context) + return nullptr; + if (!context->events) { + context->events = new AgoContextEventSystem(); + } + return context->events; +} + +static void agoStopGraphPipeliningExecutor(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph ? graph->pipelining : nullptr; + if (!pipe) + return; + + pipe->executor_stop.store(true); + { + std::lock_guard lock(pipe->enqueue_mtx); + pipe->enqueue_cv.notify_all(); + } + if (pipe->executor_thread.joinable()) { + pipe->executor_thread.join(); + } + + pipe->streaming_stop.store(true); + if (pipe->streaming_thread.joinable()) { + pipe->streaming_thread.join(); + } +} + +void agoStopGraphPipelining(AgoGraph * graph) +{ + if (!graph) + return; + agoStopGraphPipeliningExecutor(graph); +} + +static vx_uint64 agoCurrentTimestampNs() +{ + auto now = std::chrono::steady_clock::now(); + auto ns = std::chrono::duration_cast(now.time_since_epoch()).count(); + return (vx_uint64)ns; +} + +bool agoGraphHasNodeEventRegistrations(AgoGraph * graph) +{ + if (!graph || !graph->ref.context) + return false; + AgoContextEventSystem * evsys = agoGetContextEventSystem(graph->ref.context); + if (!evsys) + return false; + std::lock_guard lock(evsys->registrations_mtx); + for (const auto& reg : evsys->registrations) { + if (reg.event_type == VX_EVENT_NODE_COMPLETED || reg.event_type == VX_EVENT_NODE_ERROR) { + AgoReference * r = (AgoReference *)reg.ref; + if (r && r->type == VX_TYPE_NODE && r->scope == (vx_reference)graph) + return true; + } + } + return false; +} + +static bool agoIsPipeliningGraph(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph ? graph->pipelining : nullptr; + if (!pipe) + return false; + return pipe->schedule_mode != VX_GRAPH_SCHEDULE_MODE_NORMAL || pipe->streaming_enabled; +} + +static AgoGraphParameterQueue * agoGetGraphParameterQueue(AgoGraphPipeliningState * pipe, vx_uint32 index) +{ + if (!pipe || index >= pipe->param_queues.size()) + return nullptr; + return pipe->param_queues[index].get(); +} + +// + +static vx_uint32 agoFindEventAppValue(AgoContext * context, vx_reference ref, vx_enum event_type, vx_uint32 graph_parameter_index) +{ + AgoContextEventSystem * evsys = agoGetContextEventSystem(context); + if (!evsys) + return 0; + std::lock_guard lock(evsys->registrations_mtx); + for (const auto& reg : evsys->registrations) { + if (reg.ref == ref && reg.event_type == event_type && + (event_type != VX_EVENT_GRAPH_PARAMETER_CONSUMED || reg.graph_parameter_index == graph_parameter_index)) { + return reg.app_value; + } + } + return 0; +} + +void agoRemoveEventRegistrations(AgoContext * context, vx_reference ref) +{ + AgoContextEventSystem * evsys = agoGetContextEventSystem(context); + if (!evsys) + return; + std::lock_guard lock(evsys->registrations_mtx); + auto& regs = evsys->registrations; + regs.erase(std::remove_if(regs.begin(), regs.end(), + [ref](const AgoEventRegistration& reg) { return reg.ref == ref; }), regs.end()); +} + +vx_uint32 agoGetReferenceEnqueueCount(AgoContext * context, AgoReference * ref) +{ + if (!context || !ref) + return 0; + CAgoLock lock(context->cs); + vx_uint32 count = 0; + for (AgoGraph * graph = context->graphList.head; graph; graph = graph->next) { + AgoGraphPipeliningState * pipe = graph->pipelining; + if (!pipe) + continue; + for (auto& qptr : pipe->param_queues) { + AgoGraphParameterQueue * q = qptr.get(); + if (!q || !q->enabled) + continue; + std::lock_guard qlock(q->mtx); + for (AgoData * d : q->ready_refs) + if ((AgoReference *)d == ref) ++count; + for (AgoData * d : q->consumed_refs) + if ((AgoReference *)d == ref) ++count; + for (AgoData * d : q->done_refs) + if ((AgoReference *)d == ref) ++count; + } + } + return count; +} + +// Event helpers +// + +static void agoInternalPushEvent(AgoContext * context, const AgoEvent& evt) +{ + AgoContextEventSystem * evsys = agoGetContextEventSystem(context); + if (!evsys || !evsys->enabled) + return; + { + std::lock_guard lock(evsys->events_mtx); + evsys->events.push_back(evt); + } + evsys->events_cv.notify_one(); +} + +void agoPushEvent(AgoContext * context, const AgoEvent& evt) +{ + agoInternalPushEvent(context, evt); +} + +void agoNotifyGraphCompleted(AgoGraph * graph) +{ + if (!graph || !graph->ref.context) + return; + AgoContextEventSystem * evsys = agoGetContextEventSystem(graph->ref.context); + if (!evsys || !evsys->enabled) + return; + AgoEvent evt; + evt.event_type = VX_EVENT_GRAPH_COMPLETED; + evt.timestamp = agoCurrentTimestampNs(); + evt.app_value = agoFindEventAppValue(graph->ref.context, (vx_reference)graph, VX_EVENT_GRAPH_COMPLETED, 0); + evt.graph = graph; + evt.node = nullptr; + evt.graph_parameter_index = 0; + evt.status = VX_SUCCESS; + evt.user_parameter = nullptr; + agoInternalPushEvent(graph->ref.context, evt); +} + +void agoNotifyNodeCompleted(AgoGraph * graph, AgoNode * node) +{ + if (!graph || !node || !graph->ref.context) + return; + AgoContextEventSystem * evsys = agoGetContextEventSystem(graph->ref.context); + if (!evsys || !evsys->enabled) + return; + AgoEvent evt; + evt.event_type = VX_EVENT_NODE_COMPLETED; + evt.timestamp = agoCurrentTimestampNs(); + evt.app_value = agoFindEventAppValue(graph->ref.context, (vx_reference)node, VX_EVENT_NODE_COMPLETED, 0); + evt.graph = graph; + evt.node = node; + evt.graph_parameter_index = 0; + evt.status = VX_SUCCESS; + evt.user_parameter = nullptr; + agoInternalPushEvent(graph->ref.context, evt); +} + +void agoNotifyNodeError(AgoGraph * graph, AgoNode * node, vx_status status) +{ + if (!graph || !node || !graph->ref.context) + return; + AgoContextEventSystem * evsys = agoGetContextEventSystem(graph->ref.context); + if (!evsys || !evsys->enabled) + return; + AgoEvent evt; + evt.event_type = VX_EVENT_NODE_ERROR; + evt.timestamp = agoCurrentTimestampNs(); + evt.app_value = agoFindEventAppValue(graph->ref.context, (vx_reference)node, VX_EVENT_NODE_ERROR, 0); + evt.graph = graph; + evt.node = node; + evt.graph_parameter_index = 0; + evt.status = status; + evt.user_parameter = nullptr; + agoInternalPushEvent(graph->ref.context, evt); +} + +static void agoEmitRegisteredNodeEvents(AgoGraph * graph, vx_enum event_type, vx_status err_status) +{ + if (!graph || !graph->ref.context) + return; + AgoContextEventSystem * evsys = agoGetContextEventSystem(graph->ref.context); + if (!evsys || !evsys->enabled) + return; + std::lock_guard lock(evsys->registrations_mtx); + for (const auto& reg : evsys->registrations) { + if (reg.event_type != event_type) + continue; + AgoReference * r = (AgoReference *)reg.ref; + if (!r || r->type != VX_TYPE_NODE || r->scope != (vx_reference)graph) + continue; + AgoEvent evt; + evt.event_type = event_type; + evt.timestamp = agoCurrentTimestampNs(); + evt.app_value = reg.app_value; + evt.graph = graph; + evt.node = (AgoNode *)r; + evt.graph_parameter_index = 0; + evt.status = err_status; + evt.user_parameter = nullptr; + agoInternalPushEvent(graph->ref.context, evt); + } +} + +void agoNotifyGraphParameterConsumed(AgoGraph * graph, vx_uint32 graph_parameter_index) +{ + if (!graph || !graph->ref.context) + return; + AgoContextEventSystem * evsys = agoGetContextEventSystem(graph->ref.context); + if (!evsys || !evsys->enabled) + return; + AgoEvent evt; + evt.event_type = VX_EVENT_GRAPH_PARAMETER_CONSUMED; + evt.timestamp = agoCurrentTimestampNs(); + evt.app_value = agoFindEventAppValue(graph->ref.context, (vx_reference)graph, VX_EVENT_GRAPH_PARAMETER_CONSUMED, graph_parameter_index); + evt.graph = graph; + evt.node = nullptr; + evt.graph_parameter_index = graph_parameter_index; + evt.status = VX_SUCCESS; + evt.user_parameter = nullptr; + agoInternalPushEvent(graph->ref.context, evt); +} + +// +// Reference substitution for pipelined execution. +// After graph optimization the graph parameter may be attached to a wrapper +// node, while the actual work happens in internally created/rewired nodes. +// Rather than swapping only the graph-parameter node's paramList entries, we +// replace every occurrence of the default bound data object in the entire +// graph with the queued reference, execute, then swap back. +// + +struct AgoParamBinding { + AgoData * original; + AgoData * queued; +}; + +static void agoSwapDataRefInGraph(AgoGraph * graph, AgoData * dataFind, AgoData * dataReplace) +{ + if (dataFind == dataReplace) + return; + // Replace in all node parameter lists. + for (AgoNode * node = graph->nodeList.head; node; node = node->next) { + for (vx_uint32 i = 0; i < node->paramCount; i++) { + if (node->paramList[i] == dataFind) { + node->paramList[i] = dataReplace; + } + } + } + // Replace in supernode data lists (GPU path). +#if (ENABLE_OPENCL||ENABLE_HIP) + for (AgoSuperNode * super = graph->supernodeList; super; super = super->next) { + for (size_t i = 0; i < super->dataList.size(); i++) { + if (super->dataList[i] == dataFind) { + super->dataList[i] = dataReplace; + } + } + for (size_t i = 0; i < super->dataListForAgeDelay.size(); i++) { + if (super->dataListForAgeDelay[i] == dataFind) { + super->dataListForAgeDelay[i] = dataReplace; + } + } + } +#endif + // Replace ROI master links. + for (AgoData * adata = graph->dataList.head; adata; adata = adata->next) { + if (adata->ref.type == VX_TYPE_IMAGE && adata->u.img.isROI && adata->u.img.roiMasterImage == dataFind) { + adata->u.img.roiMasterImage = dataReplace; + } + } +} + +// Swap a graph parameter binding, expanding object-array/pyramid siblings when +// the queued reference belongs to a replicated object array/pyramid. +static void agoApplyDataRefSwapWithSiblings(AgoGraph * graph, AgoData * original, AgoData * queued) +{ + if (original == queued) + return; + AgoData * origParent = original ? original->parent : nullptr; + AgoData * queuedParent = queued ? queued->parent : nullptr; + if (origParent && queuedParent && origParent != queuedParent && + origParent->numChildren > 1 && + (origParent->ref.type == VX_TYPE_OBJECT_ARRAY || origParent->ref.type == VX_TYPE_PYRAMID) && + origParent->ref.type == queuedParent->ref.type && + origParent->numChildren == queuedParent->numChildren) { + for (vx_uint32 i = 0; i < (vx_uint32)origParent->numChildren; i++) { + agoSwapDataRefInGraph(graph, origParent->children[i], queuedParent->children[i]); + } + } else { + agoSwapDataRefInGraph(graph, original, queued); + } +} + +static std::vector agoCollectGraphParameterBindings(AgoGraph * graph) +{ + std::vector bindings; + bindings.resize(graph->parameters.size()); + for (vx_uint32 i = 0; i < (vx_uint32)graph->parameters.size(); i++) { + vx_parameter param = graph->parameters[i]; + if (!param || param->scope->type != VX_TYPE_NODE) { + bindings[i] = { nullptr, nullptr }; + continue; + } + AgoNode * node = (AgoNode *)param->scope; + if (!node) { + bindings[i] = { nullptr, nullptr }; + continue; + } + AgoData * original = (param->index < node->paramCount) ? node->paramList[param->index] : nullptr; + bindings[i] = { original, nullptr }; + } + return bindings; +} + +static void agoApplyQueuedRefsToBindings(AgoGraph * graph, + AgoGraphPipeliningState * pipe, + std::vector& bindings, + std::vector& consumed_refs) +{ + consumed_refs.assign(bindings.size(), nullptr); + for (size_t i = 0; i < bindings.size(); i++) { + AgoGraphParameterQueue * q = agoGetGraphParameterQueue(pipe, (vx_uint32)i); + if (!q) + continue; + + AgoData * ref = nullptr; + { + std::lock_guard lock(q->mtx); + if (!q->ready_refs.empty()) { + ref = q->ready_refs.front(); + q->ready_refs.pop_front(); + q->consumed_refs.push_back(ref); + } + } + if (!ref) + continue; + + consumed_refs[i] = ref; + agoRetainData(graph, ref, false); + bindings[i].queued = ref; + if (bindings[i].original) { + agoApplyDataRefSwapWithSiblings(graph, bindings[i].original, ref); + } + } +} + +static void agoRestoreBindings(AgoGraph * graph, std::vector& bindings) +{ + for (auto& b : bindings) { + if (b.original && b.queued) { + agoApplyDataRefSwapWithSiblings(graph, b.queued, b.original); + } + } +} + +static void agoMoveConsumedRefsToDone(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph->pipelining; + if (!pipe) + return; + for (auto& q : pipe->param_queues) { + std::lock_guard lock(q->mtx); + + while (!q->consumed_refs.empty()) { + q->done_refs.push_back(q->consumed_refs.front()); + q->consumed_refs.pop_front(); + } + if (!q->done_refs.empty()) { + // Notify that a reference at this parameter was consumed during this execution. + agoNotifyGraphParameterConsumed(graph, q->index); + } + } + for (auto& q : pipe->param_queues) { + q->done_cv.notify_all(); + } +} + +// +// Single pipelined execution instance (pipeline depth = 1 serialized path). +// +int agoExecutePipelinedGraphOnce(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph->pipelining; + if (!pipe) + return VX_FAILURE; + + // Collect default data references bound to each graph parameter. + std::vector bindings = agoCollectGraphParameterBindings(graph); + // Pop one ref from each configured queue and substitute into the graph. + std::vector consumed_refs; + agoApplyQueuedRefsToBindings(graph, pipe, bindings, consumed_refs); + + // Execute the graph synchronously using the normal path. + int status = agoExecuteGraph(graph); + + // Restore original bindings so the next execution sees the static defaults. + agoRestoreBindings(graph, bindings); + + // Release references retained for this execution. + for (AgoData * ref : consumed_refs) { + if (ref) { + agoReleaseData(ref, false); + } + } + + // Move consumed refs to done queues and wake waiters. + agoMoveConsumedRefsToDone(graph); + + // Emit node completion events for all user-registered nodes. This covers + // the case where graph optimization rewrote the user-visible nodes. + if (status == VX_SUCCESS) { + agoEmitRegisteredNodeEvents(graph, VX_EVENT_NODE_COMPLETED, VX_SUCCESS); + } + + // Emit graph completion event. + if (status == VX_SUCCESS) { + agoNotifyGraphCompleted(graph); + } + + return status; +} + +// +// QUEUE_MANUAL: drain all ready queues, executing one graph instance per +// complete set of ready refs. +// +int agoExecuteGraphQueueManual(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph->pipelining; + if (!pipe) + return VX_FAILURE; + + int overall_status = VX_SUCCESS; + for (;;) { + // Check if every enabled queue has at least one ready ref. + bool all_ready = true; + for (auto& q : pipe->param_queues) { + if (!q->enabled) + continue; + std::lock_guard lock(q->mtx); + if (q->ready_refs.empty()) { + all_ready = false; + break; + } + } + + if (!all_ready) + break; + + int status = agoExecutePipelinedGraphOnce(graph); + if (status != VX_SUCCESS) { + overall_status = status; + break; + } + } + return overall_status; +} + +// +// Background executor loop for QUEUE_AUTO. +// +static void agoGraphQueueAutoExecutorLoop(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph->pipelining; + if (!pipe) + return; + + auto anyReady = [&pipe]() -> bool { + for (auto& q : pipe->param_queues) { + if (!q->enabled) + continue; + std::lock_guard qlock(q->mtx); + if (!q->ready_refs.empty()) + return true; + } + return false; + }; + + while (!pipe->executor_stop.load()) { + { + CAgoLock lock(graph->cs); + if (pipe->schedule_mode != VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO) + break; + + // Wait until all enabled queues have at least one ready ref. + bool all_ready = true; + for (auto& q : pipe->param_queues) { + if (!q->enabled) + continue; + std::lock_guard qlock(q->mtx); + if (q->ready_refs.empty()) { + all_ready = false; + break; + } + } + if (all_ready) { + agoExecutePipelinedGraphOnce(graph); + continue; + } + } + // Nothing to do: block until a ref is enqueued or we're asked to stop. + std::unique_lock elock(pipe->enqueue_mtx); + pipe->enqueue_cv.wait_for(elock, std::chrono::milliseconds(1), [&pipe, &anyReady]() { + return pipe->executor_stop.load() || anyReady(); + }); + } +} + +// +// Streaming executor loop. +// +static void agoGraphStreamingExecutorLoop(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = graph->pipelining; + if (!pipe) + return; + + while (!pipe->streaming_stop.load()) { + { + CAgoLock lock(graph->cs); + if (!pipe->streaming_enabled) + break; + agoExecutePipelinedGraphOnce(graph); + } + std::this_thread::sleep_for(std::chrono::microseconds(100)); + } +} + +// +// Public internal entry point used by agoProcessGraph/agoScheduleGraph when +// the graph is in a pipelining schedule mode. +// +int agoExecuteGraphPipelined(AgoGraph * graph) +{ + if (!agoIsValidGraph(graph)) + return VX_ERROR_INVALID_REFERENCE; + + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + + if (pipe->schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_MANUAL) { + return agoExecuteGraphQueueManual(graph); + } + + if (pipe->schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO) { + // QUEUE_AUTO runs via the background executor; synchronous entry has + // nothing to do because the executor wakes whenever refs are enqueued. + return VX_SUCCESS; + } + + // Streaming mode handled by streaming thread, not via this entry. + return VX_SUCCESS; +} + +// +// Start the QUEUE_AUTO background executor if not already running. +// +void agoStartGraphPipeliningAutoExecutor(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return; + if (!pipe->executor_thread.joinable()) { + pipe->executor_stop.store(false); + pipe->executor_thread = std::thread([graph]() { + agoGraphQueueAutoExecutorLoop(graph); + }); + } +} + +// +// Start the streaming thread. +// +void agoStartGraphStreamingThread(AgoGraph * graph) +{ + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return; + if (!pipe->streaming_thread.joinable()) { + pipe->streaming_stop.store(false); + pipe->streaming_thread = std::thread([graph]() { + agoGraphStreamingExecutorLoop(graph); + }); + } +} +#else +// Stubs when the pipelining/streaming/event extension is disabled. +AgoGraphPipeliningState * agoGetGraphPipeliningState(AgoGraph *) { return nullptr; } +AgoContextEventSystem * agoGetContextEventSystem(AgoContext *) { return nullptr; } +void agoStopGraphPipelining(AgoGraph *) {} +bool agoGraphHasNodeEventRegistrations(AgoGraph *) { return false; } +void agoPushEvent(AgoContext *, const AgoEvent&) {} +void agoNotifyGraphCompleted(AgoGraph *) {} +void agoNotifyNodeCompleted(AgoGraph *, AgoNode *) {} +void agoNotifyNodeError(AgoGraph *, AgoNode *, vx_status) {} +void agoNotifyGraphParameterConsumed(AgoGraph *, vx_uint32) {} +int agoExecuteGraphPipelined(AgoGraph *) { return VX_ERROR_NOT_SUPPORTED; } +int agoExecutePipelinedGraphOnce(AgoGraph *) { return VX_ERROR_NOT_SUPPORTED; } +void agoStartGraphPipeliningAutoExecutor(AgoGraph *) {} +void agoStartGraphStreamingThread(AgoGraph *) {} +vx_uint32 agoGetReferenceEnqueueCount(AgoContext *, AgoReference *) { return 0; } +void agoRemoveEventRegistrations(AgoContext *, vx_reference) {} +#endif diff --git a/amd_openvx/openvx/ago/ago_platform.cpp b/amd_openvx/openvx/ago/ago_platform.cpp index 3418d7071..461506c3c 100644 --- a/amd_openvx/openvx/ago/ago_platform.cpp +++ b/amd_openvx/openvx/ago/ago_platform.cpp @@ -281,7 +281,9 @@ typedef struct { typedef struct { int type; // should be VX_CRITICAL_SECTION - mutex mtx; + // recursive to match Win32 CRITICAL_SECTION semantics: several call paths + // re-enter the same section, e.g. agoProcessGraph -> vxVerifyGraph + recursive_mutex mtx; } vx_critical_section; @@ -289,7 +291,7 @@ typedef struct { void EnterCriticalSection(CRITICAL_SECTION* cs) { vx_critical_section * crit_sec = (vx_critical_section *)*cs; - std::lock_guard lock(crit_sec->mtx); + crit_sec->mtx.lock(); } // Emulates LeaveCriticalSection for non_windows platform @@ -354,10 +356,9 @@ DWORD WaitForSingleObject(HANDLE h, DWORD dwMilliseconds) vx_semaphore * sem = (vx_semaphore *)h; { unique_lock lk(sem->mtx); - sem->cv.wait(lk); // TBD: implement with timeout - } - { - lock_guard lk(sem->mtx); + // Wait only if the semaphore count is currently zero; otherwise + // a notification that arrived before this wait would be lost. + sem->cv.wait(lk, [&sem]() { return sem->count > 0; }); sem->count--; } } diff --git a/amd_openvx/openvx/ago/ago_platform.h b/amd_openvx/openvx/ago/ago_platform.h index 8adfab027..eeee9c0db 100644 --- a/amd_openvx/openvx/ago/ago_platform.h +++ b/amd_openvx/openvx/ago/ago_platform.h @@ -33,6 +33,7 @@ THE SOFTWARE. #define _USE_MATH_DEFINES #include #include +#include #include #include #include @@ -49,6 +50,10 @@ THE SOFTWARE. #include #include #include +#include +#include +#include +#include #if _WIN32 #include diff --git a/amd_openvx/openvx/ago/ago_util.cpp b/amd_openvx/openvx/ago/ago_util.cpp index b48374412..27763e10d 100644 --- a/amd_openvx/openvx/ago/ago_util.cpp +++ b/amd_openvx/openvx/ago/ago_util.cpp @@ -3004,6 +3004,7 @@ AgoNode * agoCreateNode(AgoGraph * graph, AgoKernel * kernel) node->attr_affinity = graph->attr_affinity; node->ref.internal_count = 1; node->akernel = kernel; + node->pipeup_output_depth = kernel->pipeup_output_depth; node->attr_border_mode.mode = VX_BORDER_MODE_UNDEFINED; node->localDataSize = kernel->localDataSize; node->localDataPtr = NULL; @@ -3300,6 +3301,8 @@ AgoData::AgoData() } AgoData::~AgoData() { + if (ref.context) + agoRemoveEventRegistrations(ref.context, (vx_reference)this); #if ENABLE_OPENCL agoGpuOclReleaseData(this); #elif ENABLE_HIP @@ -3313,6 +3316,10 @@ AgoData::~AgoData() agoReleaseMemory(reserved_allocated); reserved_allocated = nullptr; } + if (children) { + delete[] children; + children = nullptr; + } } AgoMetaFormat::AgoMetaFormat() : set_valid_rectangle_callback{ nullptr } @@ -3331,7 +3338,8 @@ AgoKernel::AgoKernel() kernel_f{ nullptr }, validate_f{ nullptr }, input_validate_f{ nullptr }, output_validate_f{ nullptr }, initialize_f{ nullptr }, deinitialize_f{ nullptr }, query_target_support_f{ nullptr }, opencl_codegen_callback_f{ nullptr }, regen_callback_f{ nullptr }, opencl_global_work_update_callback_f{ nullptr }, gpu_buffer_update_callback_f{ nullptr }, gpu_buffer_update_param_index{ 0 }, - opencl_buffer_access_enable{ vx_false_e }, importing_module_index_plus1{ 0 } + opencl_buffer_access_enable{ vx_false_e }, importing_module_index_plus1{ 0 }, + pipeup_output_depth{ 1 }, pipeup_input_depth{ 1 } { memset(&name, 0, sizeof(name)); memset(&argConfig, 0, sizeof(argConfig)); @@ -3362,7 +3370,8 @@ AgoSuperNode::~AgoSuperNode() AgoNode::AgoNode() : next{ nullptr }, akernel{ nullptr }, flags{ 0 }, localDataSize{ 0 }, localDataPtr{ nullptr }, localDataPtr_allocated{ nullptr }, valid_rect_reset{ vx_true_e }, valid_rect_num_inputs{ 0 }, valid_rect_num_outputs{ 0 }, valid_rect_inputs{ nullptr }, valid_rect_outputs{ nullptr }, - paramCount{ 0 }, callback{ nullptr }, supernode{ nullptr }, initialized{ false }, target_support_flags{ 0 }, hierarchical_level{ 0 }, status{ VX_SUCCESS } + paramCount{ 0 }, callback{ nullptr }, supernode{ nullptr }, initialized{ false }, target_support_flags{ 0 }, hierarchical_level{ 0 }, status{ VX_SUCCESS }, + node_state{ VX_NODE_STATE_PIPEUP }, node_exec_count{ 0 }, pipeup_output_depth{ 0 } , drama_divide_invoked{ false } #if ENABLE_OPENCL , opencl_type{ 0 }, opencl_param_mem2reg_mask{ 0 }, opencl_param_discard_mask{ 0 }, opencl_param_as_value_mask{ 0 }, @@ -3386,6 +3395,8 @@ AgoNode::AgoNode() } AgoNode::~AgoNode() { + if (ref.context) + agoRemoveEventRegistrations(ref.context, (vx_reference)this); agoShutdownNode(this); if (valid_rect_inputs) { delete[] valid_rect_inputs; @@ -3412,9 +3423,9 @@ AgoNode::~AgoNode() } AgoGraph::AgoGraph() : next{ nullptr }, hThread{ nullptr }, hSemToThread{ nullptr }, hSemFromThread{ nullptr }, - threadScheduleCount{ 0 }, threadExecuteCount{ 0 }, threadWaitCount{ 0 }, threadThreadTerminationState{ 0 }, + threadScheduleCount{ 0 }, threadExecuteCount{ 0 }, threadCompletionCount{ 0 }, threadWaitCount{ 0 }, threadThreadTerminationState{ 0 }, isReadyToExecute{ vx_false_e }, detectedInvalidNode{ false }, status{ VX_SUCCESS }, - virtualDataGenerationCount{ 0 }, optimizer_flags{ AGO_GRAPH_OPTIMIZER_FLAGS_DEFAULT }, verified{ false }, enable_performance_profiling{ false }, execFrameCount{ 0 } + virtualDataGenerationCount{ 0 }, optimizer_flags{ AGO_GRAPH_OPTIMIZER_FLAGS_DEFAULT }, verified{ false }, enable_performance_profiling{ false }, execFrameCount{ 0 }, pipelining{ nullptr } #if ENABLE_OPENCL , supernodeList{ nullptr }, opencl_cmdq{ nullptr }, opencl_device{ nullptr } , enable_node_level_gpu_flush{ true } @@ -3433,6 +3444,16 @@ AgoGraph::AgoGraph() } AgoGraph::~AgoGraph() { + // stop and cleanup pipelining state + if (pipelining) { + agoStopGraphPipelining(this); + delete pipelining; + pipelining = nullptr; + } + + if (ref.context) + agoRemoveEventRegistrations(ref.context, (vx_reference)this); + // decrement auto age delays for (auto it = autoAgeDelayList.begin(); it != autoAgeDelayList.end(); it++) { if ((agoIsValidData(*it, VX_TYPE_DELAY) || agoIsValidData(*it, VX_TYPE_OBJECT_ARRAY)) && (*it)->ref.internal_count > 0) @@ -3464,7 +3485,7 @@ AgoGraph::~AgoGraph() AgoContext::AgoContext() : perfNormFactor{ 0 }, dataGenerationCount{ 0 }, nextUserStructId{ VX_TYPE_USER_STRUCT_START }, nextUserKernelId{ 0 }, nextUserLibraryId{ 1 }, num_active_modules{ 0 }, num_active_references{ 0 }, callback_log{ nullptr }, callback_reentrant{ vx_false_e }, - thread_config{ CONFIG_THREAD_DEFAULT }, importing_module_index_plus1{ 0 }, graph_garbage_data{ nullptr }, graph_garbage_node{ nullptr }, graph_garbage_list{ nullptr } + thread_config{ CONFIG_THREAD_DEFAULT }, importing_module_index_plus1{ 0 }, graph_garbage_data{ nullptr }, graph_garbage_node{ nullptr }, graph_garbage_list{ nullptr }, events{ new AgoContextEventSystem() } #if ENABLE_OPENCL #if defined(CL_VERSION_2_0) , opencl_svmcaps{ 0 } @@ -3568,6 +3589,12 @@ AgoContext::~AgoContext() // remove kernel objects agoResetKernelList(&kernelList); + // cleanup event system + if (events) { + delete events; + events = nullptr; + } + #if ENABLE_OPENCL if (opencl_mem_alloc_count > 0) { agoAddLogEntry(&ref, VX_SUCCESS, "OK: OpenCL buffer usage: " VX_FMT_SIZE ", " VX_FMT_SIZE "/" VX_FMT_SIZE "\n", @@ -3583,3 +3610,24 @@ AgoContext::~AgoContext() // critical section DeleteCriticalSection(&cs); } + +AgoGraphPipeliningState::AgoGraphPipeliningState() + : schedule_mode{ VX_GRAPH_SCHEDULE_MODE_NORMAL }, timeout_ms{ VX_TIMEOUT_WAIT_FOREVER }, + event_timeout_ms{ VX_TIMEOUT_WAIT_FOREVER }, pipeline_depth{ 1 }, + streaming_enabled{ false }, trigger_node{ nullptr }, streaming_stop{ false }, + active_executions{ 0 }, executor_stop{ false } +{ +} + +AgoGraphPipeliningState::~AgoGraphPipeliningState() +{ +} + +AgoContextEventSystem::AgoContextEventSystem() + : enabled{ false }, timeout_ms{ VX_TIMEOUT_WAIT_FOREVER } +{ +} + +AgoContextEventSystem::~AgoContextEventSystem() +{ +} diff --git a/amd_openvx/openvx/api/vx_api.cpp b/amd_openvx/openvx/api/vx_api.cpp index 7b256a4c1..5288fa9be 100644 --- a/amd_openvx/openvx/api/vx_api.cpp +++ b/amd_openvx/openvx/api/vx_api.cpp @@ -325,6 +325,13 @@ VX_API_ENTRY vx_status VX_API_CALL vxQueryContext(vx_context context, vx_enum at status = VX_SUCCESS; } break; + case VX_CONTEXT_EVENT_TIMEOUT: + if (size == sizeof(vx_uint32)) { + AgoContextEventSystem * evsys = agoGetContextEventSystem(context); + *(vx_uint32 *)ptr = evsys ? evsys->timeout_ms : VX_TIMEOUT_WAIT_FOREVER; + status = VX_SUCCESS; + } + break; default: status = VX_ERROR_NOT_SUPPORTED; break; @@ -354,6 +361,17 @@ VX_API_ENTRY vx_status VX_API_CALL vxSetContextAttribute(vx_context context, vx_ CAgoLock lock(context->cs); switch (attribute) { + case VX_CONTEXT_EVENT_TIMEOUT: + if (size == sizeof(vx_uint32)) { + AgoContextEventSystem * evsys = agoGetContextEventSystem(context); + if (evsys) { + evsys->timeout_ms = *(const vx_uint32 *)ptr; + status = VX_SUCCESS; + } else { + status = VX_ERROR_NOT_SUPPORTED; + } + } + break; case VX_CONTEXT_ATTRIBUTE_IMMEDIATE_BORDER_MODE: if(!ptr) return VX_ERROR_INVALID_PARAMETERS; if (size == sizeof(vx_border_mode_t)) { @@ -2539,6 +2557,20 @@ VX_API_ENTRY vx_status VX_API_CALL vxQueryKernel(vx_kernel kernel, vx_enum attri status = VX_SUCCESS; } break; +#if OPENVX_USE_PIPELINING + case VX_KERNEL_PIPEUP_OUTPUT_DEPTH: + if (size == sizeof(vx_uint32)) { + *(vx_uint32 *)ptr = kernel->pipeup_output_depth; + status = VX_SUCCESS; + } + break; + case VX_KERNEL_PIPEUP_INPUT_DEPTH: + if (size == sizeof(vx_uint32)) { + *(vx_uint32 *)ptr = kernel->pipeup_input_depth; + status = VX_SUCCESS; + } + break; +#endif default: status = VX_ERROR_NOT_SUPPORTED; break; @@ -2658,7 +2690,7 @@ VX_API_ENTRY vx_kernel VX_API_CALL vxAddUserKernel(vx_context context, vx_kernel_deinitialize_f deinit) { vx_kernel kernel = NULL; - if (agoIsValidContext(context) && numParams > 0 && numParams <= AGO_MAX_PARAMS && func_ptr && validate) { + if (agoIsValidContext(context) && numParams > 0 && numParams <= AGO_MAX_PARAMS && func_ptr) { CAgoLock lock(context->cs); // make sure there are no kernels with the same name if (!agoFindKernelByEnum(context, enumeration) && !agoFindKernelByName(context, name)) { @@ -2782,9 +2814,9 @@ VX_API_ENTRY vx_status VX_API_CALL vxRemoveKernel(vx_kernel kernel) vx_status status = VX_ERROR_INVALID_REFERENCE; if (agoIsValidKernel(kernel)) { status = VX_ERROR_INVALID_PARAMETERS; - // release if the kernel is not finalized and not a built-in kernel or user kernel with validate_f without external references + // release if the kernel is not finalized or is an externally registered user kernel if (!kernel->finalized || - (kernel->validate_f && kernel->external_kernel && (kernel->flags & AGO_KERNEL_FLAG_GROUP_USER) /*&& + (kernel->external_kernel && (kernel->flags & AGO_KERNEL_FLAG_GROUP_USER) /*&& kernel->ref.internal_count < 2 && kernel->ref.external_count == 0*/)) { CAgoLock lock(kernel->ref.context->cs); @@ -2822,6 +2854,22 @@ VX_API_ENTRY vx_status VX_API_CALL vxSetKernelAttribute(vx_kernel kernel, vx_enu status = VX_SUCCESS; } break; +#if OPENVX_USE_PIPELINING + case VX_KERNEL_PIPEUP_OUTPUT_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_output_depth = v; status = VX_SUCCESS; } + } + break; + 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; +#endif case VX_KERNEL_ATTRIBUTE_AMD_NODE_REGEN_CALLBACK: if (size == sizeof(void *)) { if (!kernel->finalized) { @@ -3158,6 +3206,39 @@ VX_API_ENTRY vx_status VX_API_CALL vxQueryGraph(vx_graph graph, vx_enum attribut status = VX_SUCCESS; } break; +#if OPENVX_USE_PIPELINING + case VX_GRAPH_SCHEDULE_MODE: + if (size == sizeof(vx_enum)) { + *(vx_enum *)ptr = graph->pipelining ? graph->pipelining->schedule_mode : VX_GRAPH_SCHEDULE_MODE_NORMAL; + status = VX_SUCCESS; + } + break; + case VX_GRAPH_TIMEOUT: + if (size == sizeof(vx_uint32)) { + *(vx_uint32 *)ptr = graph->pipelining ? graph->pipelining->timeout_ms : VX_TIMEOUT_WAIT_FOREVER; + status = VX_SUCCESS; + } + break; + case VX_GRAPH_EVENT_TIMEOUT: + if (size == sizeof(vx_uint32)) { + *(vx_uint32 *)ptr = graph->pipelining ? graph->pipelining->event_timeout_ms : VX_TIMEOUT_WAIT_FOREVER; + status = VX_SUCCESS; + } + break; + case VX_GRAPH_PIPELINE_DEPTH: + if (size == sizeof(vx_uint32)) { + *(vx_uint32 *)ptr = graph->pipelining ? graph->pipelining->pipeline_depth : 1; + status = VX_SUCCESS; + } + break; +#else + case VX_GRAPH_SCHEDULE_MODE: + case VX_GRAPH_TIMEOUT: + case VX_GRAPH_EVENT_TIMEOUT: + case VX_GRAPH_PIPELINE_DEPTH: + status = VX_ERROR_NOT_SUPPORTED; + break; +#endif case VX_GRAPH_ATTRIBUTE_AMD_OPTIMIZER_FLAGS: if (size == sizeof(vx_uint32)) { *(vx_uint32 *)ptr = graph->optimizer_flags; @@ -3247,6 +3328,35 @@ VX_API_ENTRY vx_status VX_API_CALL vxSetGraphAttribute(vx_graph graph, vx_enum a CAgoLock lock(graph->cs); switch (attribute) { +#if OPENVX_USE_PIPELINING + case VX_GRAPH_TIMEOUT: + if (size == sizeof(vx_uint32)) { + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + pipe->timeout_ms = *(const vx_uint32 *)ptr; + status = VX_SUCCESS; + } + break; + case VX_GRAPH_EVENT_TIMEOUT: + if (size == sizeof(vx_uint32)) { + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + pipe->event_timeout_ms = *(const vx_uint32 *)ptr; + status = VX_SUCCESS; + } + break; + case VX_GRAPH_PIPELINE_DEPTH: + if (size == sizeof(vx_uint32)) { + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + pipe->pipeline_depth = *(const vx_uint32 *)ptr; + status = VX_SUCCESS; + } + break; +#else + case VX_GRAPH_TIMEOUT: + case VX_GRAPH_EVENT_TIMEOUT: + case VX_GRAPH_PIPELINE_DEPTH: + status = VX_ERROR_NOT_SUPPORTED; + break; +#endif case VX_GRAPH_ATTRIBUTE_AMD_IMPORT_FROM_TEXT: if (size == sizeof(AgoGraphImportInfo)) { status = VX_SUCCESS; @@ -3494,6 +3604,18 @@ VX_API_ENTRY vx_status VX_API_CALL vxQueryNode(vx_node node, vx_enum attribute, status = VX_SUCCESS; } break; +#if OPENVX_USE_PIPELINING + case VX_NODE_STATE: + if (size == sizeof(vx_uint32)) { + *(vx_uint32 *)ptr = node->node_state; + status = VX_SUCCESS; + } + break; +#else + case VX_NODE_STATE: + status = VX_ERROR_NOT_SUPPORTED; + break; +#endif case VX_NODE_ATTRIBUTE_AMD_AFFINITY: if (size == sizeof(AgoTargetAffinityInfo_)) { *(AgoTargetAffinityInfo_ *)ptr = node->attr_affinity; @@ -4486,6 +4608,16 @@ VX_API_ENTRY vx_status VX_API_CALL vxQueryReference(vx_reference ref, vx_enum at status = VX_SUCCESS; } break; + case VX_REFERENCE_ENQUEUE_COUNT: + if (size == sizeof(vx_uint32)) { +#if OPENVX_USE_PIPELINING + *(vx_uint32 *)ptr = agoGetReferenceEnqueueCount(ref->context, ref); +#else + *(vx_uint32 *)ptr = 0; +#endif + status = VX_SUCCESS; + } + break; case VX_REFERENCE_TYPE: if (size == sizeof(vx_enum)) { *(vx_enum *)ptr = ref->type; diff --git a/amd_openvx/openvx/api/vx_pipelining_api.cpp b/amd_openvx/openvx/api/vx_pipelining_api.cpp new file mode 100644 index 000000000..bae0ab532 --- /dev/null +++ b/amd_openvx/openvx/api/vx_pipelining_api.cpp @@ -0,0 +1,612 @@ +/* +Copyright (c) 2015 - 2026 Advanced Micro Devices, Inc. All rights reserved. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. +*/ + +#include "ago_internal.h" + +// +// OpenVX Pipelining Extension - public API implementation +// + +#if OPENVX_USE_PIPELINING + +VX_API_ENTRY vx_status VX_API_CALL vxSetGraphScheduleConfig( + vx_graph graph, + vx_enum graph_schedule_mode, + vx_uint32 graph_parameters_list_size, + const vx_graph_parameter_queue_params_t graph_parameters_queue_params_list[]) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph) && !graph->verified) { + if ((graph_schedule_mode != VX_GRAPH_SCHEDULE_MODE_NORMAL) && + (graph_schedule_mode != VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO) && + (graph_schedule_mode != VX_GRAPH_SCHEDULE_MODE_QUEUE_MANUAL)) { + return VX_ERROR_INVALID_PARAMETERS; + } + if (graph_schedule_mode == VX_GRAPH_SCHEDULE_MODE_NORMAL) { + if (graph_parameters_list_size != 0 || graph_parameters_queue_params_list != nullptr) { + return VX_ERROR_INVALID_PARAMETERS; + } + } else { + if (graph_parameters_list_size == 0 || graph_parameters_queue_params_list == nullptr) { + return VX_ERROR_INVALID_PARAMETERS; + } + } + + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + + // Stop any active executor before reconfiguring. This has to happen outside + // graph->cs because the executor runs the graph inside that section. + agoStopGraphPipelining(graph); + + CAgoLock lock(graph->cs); + + pipe->schedule_mode = graph_schedule_mode; + pipe->param_queues.clear(); + pipe->param_queues.resize(graph->parameters.size()); + for (size_t i = 0; i < pipe->param_queues.size(); i++) { if (!pipe->param_queues[i]) pipe->param_queues[i].reset(new AgoGraphParameterQueue()); pipe->param_queues[i]->index = (vx_uint32)i; } + for (vx_uint32 i = 0; i < graph_parameters_list_size; i++) { + const vx_graph_parameter_queue_params_t & p = graph_parameters_queue_params_list[i]; + vx_uint32 index = p.graph_parameter_index; + if (index >= (vx_uint32)graph->parameters.size()) + return VX_ERROR_INVALID_PARAMETERS; + if (p.refs_list_size == 0) + return VX_ERROR_INVALID_PARAMETERS; + pipe->param_queues[index].get()->max_depth = p.refs_list_size; + pipe->param_queues[index].get()->enabled = true; + if (p.refs_list) { + for (vx_uint32 j = 0; j < p.refs_list_size; j++) { + vx_reference ref = p.refs_list[j]; + if (!ref) + return VX_ERROR_INVALID_PARAMETERS; + if (!agoIsValidReference((AgoReference *)ref)) + return VX_ERROR_INVALID_REFERENCE; + pipe->param_queues[index].get()->valid_refs.push_back((AgoData *)ref); + } + } + } + + if (graph_schedule_mode == VX_GRAPH_SCHEDULE_MODE_QUEUE_AUTO) { + agoStartGraphPipeliningAutoExecutor(graph); + } + + status = VX_SUCCESS; + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxGetGraphParameterRefsList( + vx_graph graph, + vx_uint32 param, + vx_uint32 ref_list_size, + vx_reference refs_list[]) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph) && graph->verified) { + status = VX_ERROR_INVALID_PARAMETERS; + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (pipe && param < (vx_uint32)pipe->param_queues.size() && refs_list) { + AgoGraphParameterQueue * q = pipe->param_queues[param].get(); + if (ref_list_size >= (vx_uint32)q->valid_refs.size()) { + for (size_t i = 0; i < q->valid_refs.size(); i++) { + refs_list[i] = (vx_reference)q->valid_refs[i]; + } + status = VX_SUCCESS; + } + } + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxAddReferencesToGraphParameterList( + vx_graph graph, + vx_uint32 graph_parameter_index, + vx_uint32 number_to_add, + const vx_reference new_references[]) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph) && graph->verified) { + status = VX_ERROR_INVALID_PARAMETERS; + if (number_to_add == 0 || !new_references) + return status; + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (pipe && graph_parameter_index < (vx_uint32)pipe->param_queues.size()) { + AgoGraphParameterQueue * q = pipe->param_queues[graph_parameter_index].get(); + for (vx_uint32 i = 0; i < number_to_add; i++) { + if (!new_references[i] || !agoIsValidReference((AgoReference *)new_references[i])) + return VX_ERROR_INVALID_REFERENCE; + q->valid_refs.push_back((AgoData *)new_references[i]); + } + status = VX_SUCCESS; + } + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterEnqueueReadyRef( + vx_graph graph, + vx_uint32 graph_parameter_index, + const vx_reference *refs, + vx_uint32 num_refs) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph)) { + status = VX_ERROR_INVALID_PARAMETERS; + if (num_refs > 0 && !refs) + return status; + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + if (graph_parameter_index >= (vx_uint32)pipe->param_queues.size()) + return status; + + AgoGraphParameterQueue * q = pipe->param_queues[graph_parameter_index].get(); + // If the queue is not explicitly enabled by the schedule config, allow + // enqueueing as long as the corresponding graph parameter is a valid output. + if (!q->enabled) { + if (graph_parameter_index >= graph->parameters.size()) + return status; + vx_parameter param = graph->parameters[graph_parameter_index]; + if (!param || param->direction != VX_OUTPUT) + return status; + q->enabled = true; + q->max_depth = num_refs; + } + + for (vx_uint32 i = 0; i < num_refs; i++) { + if (!refs[i] || !agoIsValidReference((AgoReference *)refs[i])) + return VX_ERROR_INVALID_REFERENCE; + // If valid_refs is configured, reject refs not in the list. + if (!q->valid_refs.empty()) { + bool found = false; + for (AgoData * valid : q->valid_refs) { + if ((vx_reference)valid == refs[i]) { + found = true; + break; + } + } + if (!found) + return VX_ERROR_INVALID_PARAMETERS; + } + std::lock_guard lock(q->mtx); + q->ready_refs.push_back((AgoData *)refs[i]); + } + { + std::lock_guard lock(pipe->enqueue_mtx); + pipe->enqueue_cv.notify_all(); + } + status = VX_SUCCESS; + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterDequeueDoneRef( + vx_graph graph, + vx_uint32 graph_parameter_index, + vx_reference *refs, + vx_uint32 max_refs, + vx_uint32 *num_refs) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph)) { + if (!refs || !num_refs) + return VX_ERROR_INVALID_PARAMETERS; + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + if (graph_parameter_index >= (vx_uint32)pipe->param_queues.size()) + return VX_ERROR_INVALID_PARAMETERS; + + AgoGraphParameterQueue * q = pipe->param_queues[graph_parameter_index].get(); + if (!q->enabled) + return VX_ERROR_INVALID_PARAMETERS; + + std::unique_lock lock(q->mtx); + if (pipe->timeout_ms == VX_TIMEOUT_WAIT_FOREVER) { + q->done_cv.wait(lock, [q]() { return !q->done_refs.empty(); }); + } else { + if (!q->done_cv.wait_for(lock, std::chrono::milliseconds(pipe->timeout_ms), + [q]() { return !q->done_refs.empty(); })) + return VX_FAILURE; + } + if (q->done_refs.empty()) + return VX_FAILURE; + vx_uint32 count = 0; + while (count < max_refs && !q->done_refs.empty()) { + refs[count] = (vx_reference)q->done_refs.front(); + q->done_refs.pop_front(); + count++; + } + *num_refs = count; + status = VX_SUCCESS; + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterCheckDoneRef( + vx_graph graph, + vx_uint32 graph_parameter_index, + vx_uint32 *num_refs) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph)) { + if (!num_refs) + return VX_ERROR_INVALID_PARAMETERS; + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + if (graph_parameter_index >= (vx_uint32)pipe->param_queues.size()) + return VX_ERROR_INVALID_PARAMETERS; + + AgoGraphParameterQueue * q = pipe->param_queues[graph_parameter_index].get(); + std::lock_guard lock(q->mtx); + *num_refs = (vx_uint32)q->done_refs.size(); + status = VX_SUCCESS; + } + return status; +} + +// +// Context-level event API +// + +VX_API_ENTRY vx_status VX_API_CALL vxEnableEvents(vx_context context) +{ + if (!agoIsValidContext((AgoContext *)context)) + return VX_ERROR_INVALID_REFERENCE; + AgoContextEventSystem * evsys = agoGetContextEventSystem((AgoContext *)context); + if (!evsys) + return VX_FAILURE; + evsys->enabled = true; + return VX_SUCCESS; +} + +VX_API_ENTRY vx_status VX_API_CALL vxDisableEvents(vx_context context) +{ + if (!agoIsValidContext((AgoContext *)context)) + return VX_ERROR_INVALID_REFERENCE; + AgoContextEventSystem * evsys = agoGetContextEventSystem((AgoContext *)context); + if (!evsys) + return VX_FAILURE; + evsys->enabled = false; + return VX_SUCCESS; +} + +VX_API_ENTRY vx_status VX_API_CALL vxWaitEvent(vx_context context, vx_event_t *event, vx_bool do_not_block) +{ + if (!agoIsValidContext((AgoContext *)context) || !event) + return VX_ERROR_INVALID_REFERENCE; + AgoContextEventSystem * evsys = agoGetContextEventSystem((AgoContext *)context); + if (!evsys) + return VX_FAILURE; + + std::unique_lock lock(evsys->events_mtx); + if (do_not_block == vx_true_e) { + if (evsys->events.empty()) + return VX_FAILURE; + } else { + if (evsys->timeout_ms == VX_TIMEOUT_WAIT_FOREVER) { + evsys->events_cv.wait(lock, [&evsys]() { return !evsys->events.empty(); }); + } else { + if (!evsys->events_cv.wait_for(lock, std::chrono::milliseconds(evsys->timeout_ms), + [&evsys]() { return !evsys->events.empty(); })) + return VX_FAILURE; + } + } + if (evsys->events.empty()) + return VX_FAILURE; + + AgoEvent evt = evsys->events.front(); + evsys->events.pop_front(); + lock.unlock(); + + event->type = evt.event_type; + event->timestamp = evt.timestamp; + event->app_value = evt.app_value; + switch (evt.event_type) { + case VX_EVENT_GRAPH_PARAMETER_CONSUMED: + event->event_info.graph_parameter_consumed.graph = (vx_graph)evt.graph; + event->event_info.graph_parameter_consumed.graph_parameter_index = evt.graph_parameter_index; + break; + case VX_EVENT_GRAPH_COMPLETED: + event->event_info.graph_completed.graph = (vx_graph)evt.graph; + break; + case VX_EVENT_NODE_COMPLETED: + event->event_info.node_completed.graph = (vx_graph)evt.graph; + event->event_info.node_completed.node = (vx_node)evt.node; + break; + case VX_EVENT_NODE_ERROR: + event->event_info.node_error.graph = (vx_graph)evt.graph; + event->event_info.node_error.node = (vx_node)evt.node; + event->event_info.node_error.status = evt.status; + break; + case VX_EVENT_USER: + event->event_info.user_event.user_event_parameter = evt.user_parameter; + break; + default: + break; + } + return VX_SUCCESS; +} + +VX_API_ENTRY vx_status VX_API_CALL vxSendUserEvent(vx_context context, vx_uint32 app_value, const void *parameter) +{ + if (!agoIsValidContext((AgoContext *)context)) + return VX_ERROR_INVALID_REFERENCE; + AgoContextEventSystem * evsys = agoGetContextEventSystem((AgoContext *)context); + if (!evsys || !evsys->enabled) + return VX_FAILURE; + + AgoEvent evt; + evt.event_type = VX_EVENT_USER; + evt.timestamp = 0; // could use steady_clock + evt.app_value = app_value; + evt.graph = nullptr; + evt.node = nullptr; + evt.graph_parameter_index = 0; + evt.status = VX_SUCCESS; + evt.user_parameter = (void *)parameter; + agoPushEvent((AgoContext *)context, evt); + return VX_SUCCESS; +} + +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 lock(evsys->registrations_mtx); + evsys->registrations.push_back(reg); + return VX_SUCCESS; +} + +// +// Graph-level event API (forwarded to context-level event system for now). +// + +VX_API_ENTRY vx_status VX_API_CALL vxRegisterGraphEvent(vx_reference graph_or_node, enum vx_event_type_e type, vx_uint32 param, vx_uint32 app_value) +{ + return vxRegisterEvent(graph_or_node, type, param, app_value); +} + +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); +} + +VX_API_ENTRY vx_status VX_API_CALL vxEnableGraphEvents(vx_graph graph) +{ + if (!agoIsValidGraph((AgoGraph *)graph)) + return VX_ERROR_INVALID_REFERENCE; + AgoContext * context = ((AgoGraph *)graph)->ref.context; + return vxEnableEvents(context); +} + +VX_API_ENTRY vx_status VX_API_CALL vxDisableGraphEvents(vx_graph graph) +{ + if (!agoIsValidGraph((AgoGraph *)graph)) + return VX_ERROR_INVALID_REFERENCE; + AgoContext * context = ((AgoGraph *)graph)->ref.context; + return vxDisableEvents(context); +} + +VX_API_ENTRY vx_status VX_API_CALL vxSendUserGraphEvent(vx_graph graph, vx_uint32 app_value, const void *parameter) +{ + if (!agoIsValidGraph((AgoGraph *)graph)) + return VX_ERROR_INVALID_REFERENCE; + AgoContext * context = ((AgoGraph *)graph)->ref.context; + return vxSendUserEvent(context, app_value, parameter); +} + +// +// Streaming API +// + +VX_API_ENTRY vx_status VX_API_CALL vxEnableGraphStreaming(vx_graph graph, vx_node trigger_node) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph)) { + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + CAgoLock lock(graph->cs); + pipe->streaming_enabled = true; + if (trigger_node && agoIsValidNode((AgoNode *)trigger_node)) { + pipe->trigger_node = (AgoNode *)trigger_node; + } + status = VX_SUCCESS; + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxStartGraphStreaming(vx_graph graph) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph)) { + if (!graph->verified) + return VX_ERROR_NOT_SUFFICIENT; + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + CAgoLock lock(graph->cs); + if (!pipe->streaming_enabled) + return VX_FAILURE; + agoStartGraphStreamingThread(graph); + status = VX_SUCCESS; + } + return status; +} + +VX_API_ENTRY vx_status VX_API_CALL vxStopGraphStreaming(vx_graph graph) +{ + vx_status status = VX_ERROR_INVALID_REFERENCE; + if (agoIsValidGraph(graph)) { + AgoGraphPipeliningState * pipe = agoGetGraphPipeliningState(graph); + if (!pipe) + return VX_FAILURE; + // The streaming thread executes the graph under graph->cs, so it has to be + // joined before that section is entered. + pipe->streaming_stop.store(true); + if (pipe->streaming_thread.joinable()) { + pipe->streaming_thread.join(); + } + CAgoLock lock(graph->cs); + pipe->streaming_enabled = false; + status = VX_SUCCESS; + } + return status; +} + +// +// Additional helper API (stub) +// + +VX_API_ENTRY vx_status VX_API_CALL vxGetKernelParameterConfig(vx_kernel kernel, vx_uint32 num_params, vx_kernel_parameter_config_t parameter_config[]) +{ + return VX_ERROR_NOT_SUPPORTED; +} +#else +// Stubs when the pipelining/streaming/event extension is disabled. +VX_API_ENTRY vx_status VX_API_CALL vxSetGraphScheduleConfig( + vx_graph graph, + vx_enum graph_schedule_mode, + vx_uint32 graph_parameters_list_size, + const vx_graph_parameter_queue_params_t graph_parameters_queue_params_list[]) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxGetGraphParameterRefsList( + vx_graph graph, + vx_uint32 param, + vx_uint32 ref_list_size, + vx_reference refs_list[]) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxAddReferencesToGraphParameterList( + vx_graph graph, + vx_uint32 graph_parameter_index, + vx_uint32 number_to_add, + const vx_reference new_references[]) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterEnqueueReadyRef( + vx_graph graph, + vx_uint32 graph_parameter_index, + const vx_reference *refs, + vx_uint32 num_refs) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterDequeueDoneRef( + vx_graph graph, + vx_uint32 graph_parameter_index, + vx_reference *refs, + vx_uint32 max_refs, + vx_uint32 *num_refs) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxGraphParameterCheckDoneRef( + vx_graph graph, + vx_uint32 graph_parameter_index, + vx_uint32 *num_refs) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxEnableEvents(vx_context context) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxDisableEvents(vx_context context) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxWaitEvent(vx_context context, vx_event_t *event, vx_bool do_not_block) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxSendUserEvent(vx_context context, vx_uint32 app_value, const void *parameter) +{ + return VX_ERROR_NOT_SUPPORTED; +} +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) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxRegisterGraphEvent(vx_reference graph_or_node, enum vx_event_type_e type, vx_uint32 param, vx_uint32 app_value) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxWaitGraphEvent(vx_graph graph, vx_event_t *event, vx_bool do_not_block) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxEnableGraphEvents(vx_graph graph) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxDisableGraphEvents(vx_graph graph) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxSendUserGraphEvent(vx_graph graph, vx_uint32 app_value, const void *parameter) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxEnableGraphStreaming(vx_graph graph, vx_node trigger_node) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxStartGraphStreaming(vx_graph graph) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxStopGraphStreaming(vx_graph graph) +{ + return VX_ERROR_NOT_SUPPORTED; +} +VX_API_ENTRY vx_status VX_API_CALL vxGetKernelParameterConfig(vx_kernel kernel, vx_uint32 num_params, vx_kernel_parameter_config_t parameter_config[]) +{ + return VX_ERROR_NOT_SUPPORTED; +} +#endif