diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6b3807f..d5b2c95 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -56,7 +56,7 @@ jobs: name: Build (CUDA 11.8) runs-on: ubuntu-latest container: - image: nvidia/cuda:11.8-devel-ubuntu22.04 + image: nvidia/cuda:11.8.0-devel-ubuntu22.04 steps: - name: Checkout uses: actions/checkout@v6 @@ -64,21 +64,25 @@ jobs: - name: Install dependencies run: | apt-get update - apt-get install -y cmake + apt-get install -y cmake git - - name: Configure (minimal preset) - run: cmake --preset minimal -DMINI_IMAGE_PIPE_WITH_CVCUDA=OFF -DMINI_IMAGE_PIPE_WITH_TENSORRT=OFF -DMINI_IMAGE_PIPE_WITH_GSTREAMER=OFF - env: - CMAKE_CUDA_ARCHITECTURES: "80" # Single arch for faster CI build + - name: Configure + run: > + cmake -S . -B build-minimal + -DCMAKE_BUILD_TYPE=Debug + -DCMAKE_CUDA_ARCHITECTURES=80 + -DMINI_IMAGE_PIPE_WITH_CVCUDA=OFF + -DMINI_IMAGE_PIPE_WITH_TENSORRT=OFF + -DMINI_IMAGE_PIPE_WITH_GSTREAMER=OFF - name: Build - run: cmake --build --preset minimal + run: cmake --build build-minimal - name: Test (CPU-only environment) run: | echo "Note: CUDA tests require GPU hardware and will be skipped on CPU-only runners." echo "Running test executable to verify build integrity..." - ./build/mini_image_pipe_tests --gtest_filter=* || echo "Tests require GPU - skipped" + ctest --test-dir build-minimal --output-on-failure || echo "Tests require GPU - skipped" continue-on-error: true gpu-validation: @@ -89,18 +93,19 @@ jobs: - name: Checkout uses: actions/checkout@v6 - - name: Configure (minimal preset) - run: cmake --preset minimal - env: - CMAKE_CUDA_ARCHITECTURES: "80" + - name: Configure + run: > + cmake -S . -B build-minimal + -DCMAKE_BUILD_TYPE=Debug + -DCMAKE_CUDA_ARCHITECTURES=80 - name: Build tests and benchmark run: | - cmake --build --preset minimal --target mini_image_pipe_tests - cmake --build --preset minimal --target benchmark_pipeline + cmake --build build-minimal --target mini_image_pipe_tests + cmake --build build-minimal --target benchmark_pipeline - name: Run tests - run: ctest --preset minimal --output-on-failure + run: ctest --test-dir build-minimal --output-on-failure - name: Run benchmark run: ./build-minimal/benchmark_pipeline --iterations 5 --batch 2 diff --git a/CMakePresets.json b/CMakePresets.json index 44b2c28..26cc102 100644 --- a/CMakePresets.json +++ b/CMakePresets.json @@ -1,5 +1,5 @@ { - "version": 6, + "version": 3, "configurePresets": [ { "name": "default", diff --git a/PERFORMANCE.md b/PERFORMANCE.md index 3b767f0..34276c5 100644 --- a/PERFORMANCE.md +++ b/PERFORMANCE.md @@ -47,10 +47,17 @@ for (int i = 0; i < 1000; i++) { ```cpp // Process multiple frames in one call -std::vector inputs = {frame1, frame2, frame3}; -std::vector outputs; - -pipeline.executeBatch(inputs, outputs, width, height, channels); +std::vector inputs = { + {frame1, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame2, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame3, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, +}; +std::vector outputs; + +pipeline.executeBatch(inputs, outputs); ``` ## Profiling diff --git a/docs/api/pipeline.md b/docs/api/pipeline.md index 34cdaa5..8a2662b 100644 --- a/docs/api/pipeline.md +++ b/docs/api/pipeline.md @@ -19,9 +19,8 @@ public: void* getOutput(int nodeId); cudaError_t execute(); - cudaError_t executeBatch(const std::vector& inputs, - std::vector& outputs, - int width, int height, int channels); + cudaError_t executeBatch(const std::vector& inputs, + std::vector& outputs); TaskGraph& getTaskGraph(); const TaskGraph& getTaskGraph() const; @@ -91,10 +90,21 @@ void* out = pipeline.getOutput(n3); ### Batch execution ```cpp -std::vector inputs = {frame0, frame1, frame2}; -std::vector outputs; +std::vector inputs = { + {frame0, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame1, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame2, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, +}; +std::vector outputs; -cudaError_t err = pipeline.executeBatch(inputs, outputs, width, height, channels); +cudaError_t err = pipeline.executeBatch(inputs, outputs); +for (const auto& output : outputs) { + std::cout << "Sink node " << output.nodeId << " produced " + << output.frames.size() << " frames" << std::endl; +} ``` ## Error handling diff --git a/docs/blog/tutorials/video-pipeline.md b/docs/blog/tutorials/video-pipeline.md index 7fdc712..9a5b91b 100644 --- a/docs/blog/tutorials/video-pipeline.md +++ b/docs/blog/tutorials/video-pipeline.md @@ -91,13 +91,15 @@ cudaStreamDestroy(stream); ### Batch Processing ```cpp -std::vector frames; +std::vector frames; for (int i = 0; i < batchSize; i++) { - frames.push_back(getNextFrame()); + frames.push_back({getNextFrame(), width, height, channels, width * channels, + sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}); } -std::vector outputs; -pipeline.executeBatch(frames, outputs, width, height, channels); +std::vector outputs; +pipeline.executeBatch(frames, outputs); ``` ### Zero-Copy Output diff --git a/docs/guide/usage.md b/docs/guide/usage.md index 9996741..e981fdf 100644 --- a/docs/guide/usage.md +++ b/docs/guide/usage.md @@ -49,18 +49,24 @@ int main() { For processing multiple frames efficiently: ```cpp -std::vector inputs = {...}; // Array of device pointers -std::vector outputs; +std::vector inputs = { + {frame0, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame1, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, +}; +std::vector outputs; Pipeline pipeline; // ... setup pipeline ... -cudaError_t err = pipeline.executeBatch(inputs, outputs, width, height, channels); +cudaError_t err = pipeline.executeBatch(inputs, outputs); ``` The batch executor: - Processes frames concurrently across multiple streams -- Reuses allocated buffers between frames +- Validates that every frame has identical device-memory shape metadata +- Returns one `BatchOutput` per sink node instead of silently picking one - Synchronizes only at the end of each batch ## Runtime Parameter Updates diff --git a/docs/zh/api/pipeline.md b/docs/zh/api/pipeline.md index 55cf9d2..77a3589 100644 --- a/docs/zh/api/pipeline.md +++ b/docs/zh/api/pipeline.md @@ -19,9 +19,8 @@ public: void* getOutput(int nodeId); cudaError_t execute(); - cudaError_t executeBatch(const std::vector& inputs, - std::vector& outputs, - int width, int height, int channels); + cudaError_t executeBatch(const std::vector& inputs, + std::vector& outputs); TaskGraph& getTaskGraph(); const TaskGraph& getTaskGraph() const; @@ -91,10 +90,21 @@ void* out = pipeline.getOutput(n3); ### 批量执行 ```cpp -std::vector inputs = {frame0, frame1, frame2}; -std::vector outputs; +std::vector inputs = { + {frame0, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame1, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame2, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, +}; +std::vector outputs; -cudaError_t err = pipeline.executeBatch(inputs, outputs, width, height, channels); +cudaError_t err = pipeline.executeBatch(inputs, outputs); +for (const auto& output : outputs) { + std::cout << "Sink node " << output.nodeId << " produced " + << output.frames.size() << " frames" << std::endl; +} ``` ## 错误处理 diff --git a/docs/zh/blog/tutorials/video-pipeline.md b/docs/zh/blog/tutorials/video-pipeline.md index 80c2ceb..8bb5543 100644 --- a/docs/zh/blog/tutorials/video-pipeline.md +++ b/docs/zh/blog/tutorials/video-pipeline.md @@ -91,13 +91,15 @@ cudaStreamDestroy(stream); ### 批处理 ```cpp -std::vector frames; +std::vector frames; for (int i = 0; i < batchSize; i++) { - frames.push_back(getNextFrame()); + frames.push_back({getNextFrame(), width, height, channels, width * channels, + sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}); } -std::vector outputs; -pipeline.executeBatch(frames, outputs, width, height, channels); +std::vector outputs; +pipeline.executeBatch(frames, outputs); ``` ### 零拷贝输出 diff --git a/docs/zh/guide/usage.md b/docs/zh/guide/usage.md index 644c8c9..dd31b7f 100644 --- a/docs/zh/guide/usage.md +++ b/docs/zh/guide/usage.md @@ -49,18 +49,24 @@ int main() { 高效处理多帧图像: ```cpp -std::vector inputs = {...}; // 设备指针数组 -std::vector outputs; +std::vector inputs = { + {frame0, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, + {frame1, width, height, channels, width * channels, sizeof(uint8_t), 1, + static_cast(width) * height * channels, true, false}, +}; +std::vector outputs; Pipeline pipeline; // ... 配置流水线 ... -cudaError_t err = pipeline.executeBatch(inputs, outputs, width, height, channels); +cudaError_t err = pipeline.executeBatch(inputs, outputs); ``` 批量执行器特性: - 跨多个流并发处理帧 -- 帧间复用已分配缓冲区 +- 校验每帧的设备内存形状元数据必须一致 +- 每个 sink 节点返回一个 `BatchOutput`，不再静默丢弃额外输出 - 仅在每个批次结束时同步 ## 运行时参数更新 diff --git a/examples/benchmark_pipeline.cpp b/examples/benchmark_pipeline.cpp index b37f3cb..0deb7a2 100644 --- a/examples/benchmark_pipeline.cpp +++ b/examples/benchmark_pipeline.cpp @@ -23,6 +23,20 @@ int parseIntArg(char** begin, char** end, const std::string& flag, int defaultVa return defaultValue; } +mini_image_pipe::ImageBuffer makeFrameBuffer(void* data, int width, int height, int channels) { + mini_image_pipe::ImageBuffer buffer; + buffer.data = data; + buffer.width = width; + buffer.height = height; + buffer.channels = channels; + buffer.stride = width * channels; + buffer.elementSize = sizeof(uint8_t); + buffer.batchSize = 1; + buffer.batchStride = static_cast(buffer.stride) * height; + buffer.isDeviceMemory = true; + return buffer; +} + } // namespace int main(int argc, char** argv) { @@ -64,9 +78,11 @@ int main(int argc, char** argv) { pipeline.connect(n2, n3); pipeline.connect(n3, n4); - std::vector inputs(batchSize, nullptr); - std::vector outputs; - for (void*& input : inputs) { + std::vector inputs; + inputs.reserve(batchSize); + std::vector outputs; + for (int i = 0; i < batchSize; ++i) { + void* input = nullptr; cudaError_t allocErr = cudaMalloc(&input, imageBytes); if (allocErr != cudaSuccess) { std::cerr << "cudaMalloc failed: " << cudaGetErrorString(allocErr) << std::endl; @@ -77,18 +93,19 @@ int main(int argc, char** argv) { std::cerr << "cudaMemset failed: " << cudaGetErrorString(memsetErr) << std::endl; return 1; } + inputs.push_back(makeFrameBuffer(input, width, height, channels)); } const int warmupIterations = 3; for (int i = 0; i < warmupIterations; ++i) { if (batchSize == 1) { - pipeline.setInput(n1, inputs.front(), width, height, channels); + pipeline.setInput(n1, inputs.front().data, width, height, channels); if (pipeline.execute() != cudaSuccess) { std::cerr << "Warmup execute failed" << std::endl; return 1; } } else { - if (pipeline.executeBatch(inputs, outputs, width, height, channels) != cudaSuccess) { + if (pipeline.executeBatch(inputs, outputs) != cudaSuccess) { std::cerr << "Warmup executeBatch failed" << std::endl; return 1; } @@ -98,13 +115,13 @@ int main(int argc, char** argv) { auto start = std::chrono::steady_clock::now(); for (int i = 0; i < iterations; ++i) { if (batchSize == 1) { - pipeline.setInput(n1, inputs.front(), width, height, channels); + pipeline.setInput(n1, inputs.front().data, width, height, channels); if (pipeline.execute() != cudaSuccess) { std::cerr << "Execute failed" << std::endl; return 1; } } else { - if (pipeline.executeBatch(inputs, outputs, width, height, channels) != cudaSuccess) { + if (pipeline.executeBatch(inputs, outputs) != cudaSuccess) { std::cerr << "ExecuteBatch failed" << std::endl; return 1; } @@ -125,8 +142,8 @@ int main(int argc, char** argv) { std::cout << " total_ms: " << totalMs << std::endl; std::cout << " fps: " << fps << std::endl; - for (void* input : inputs) { - cudaFree(input); + for (const auto& input : inputs) { + cudaFree(input.data); } return 0; diff --git a/include/memory_manager.h b/include/memory_manager.h index ab4f24b..b1cfcd4 100644 --- a/include/memory_manager.h +++ b/include/memory_manager.h @@ -12,7 +12,19 @@ namespace mini_image_pipe { -class MemoryManager { +class IRuntimeAllocator { +public: + virtual ~IRuntimeAllocator() = default; + + virtual void setDeviceAllocatorMode(DeviceAllocatorMode mode) = 0; + virtual void* allocateDevice(size_t size, cudaStream_t stream) = 0; + virtual void freeDevice(void* ptr, cudaStream_t stream) = 0; + virtual OperatorWorkspace allocateWorkspace(const WorkspaceRequirements& requirements, + cudaStream_t stream) = 0; + virtual void freeWorkspace(const OperatorWorkspace& workspace, cudaStream_t stream) = 0; +}; + +class MemoryManager : public IRuntimeAllocator { public: static MemoryManager& getInstance(); @@ -30,16 +42,16 @@ class MemoryManager { // Allocate device memory void* allocateDevice(size_t size); - void* allocateDevice(size_t size, cudaStream_t stream); + void* allocateDevice(size_t size, cudaStream_t stream) override; // Free device memory void freeDevice(void* ptr); - void freeDevice(void* ptr, cudaStream_t stream); + void freeDevice(void* ptr, cudaStream_t stream) override; // Allocate and free workspace bundles OperatorWorkspace allocateWorkspace(const WorkspaceRequirements& requirements, - cudaStream_t stream); - void freeWorkspace(const OperatorWorkspace& workspace, cudaStream_t stream); + cudaStream_t stream) override; + void freeWorkspace(const OperatorWorkspace& workspace, cudaStream_t stream) override; // Async copy host to device cudaError_t copyToDeviceAsync(void* dst, const void* src, size_t size, cudaStream_t stream); @@ -58,9 +70,13 @@ class MemoryManager { size_t getPinnedReuseCount() const { return pinnedReuseCount_; } size_t getActiveAllocations() const; - void setDeviceAllocatorMode(DeviceAllocatorMode mode); - DeviceAllocatorMode getRequestedDeviceAllocatorMode() const { return requestedDeviceAllocatorMode_; } - DeviceAllocatorMode getEffectiveDeviceAllocatorMode() const { return effectiveDeviceAllocatorMode_; } + void setDeviceAllocatorMode(DeviceAllocatorMode mode) override; + DeviceAllocatorMode getRequestedDeviceAllocatorMode() const { + return requestedDeviceAllocatorMode_; + } + DeviceAllocatorMode getEffectiveDeviceAllocatorMode() const { + return effectiveDeviceAllocatorMode_; + } bool supportsAsyncDeviceAllocator() const; private: diff --git a/include/operator.h b/include/operator.h index 7b242a5..fcaf971 100644 --- a/include/operator.h +++ b/include/operator.h @@ -50,9 +50,8 @@ class IOperator { const void* inputPtr = static_cast(input.data) + input.batchStride * batchIndex; void* outputPtr = static_cast(output.data) + output.batchStride * batchIndex; - cudaError_t err = - execute(inputPtr, outputPtr, input.width, input.height, input.channels, - context.stream); + cudaError_t err = execute(inputPtr, outputPtr, input.width, input.height, + input.channels, context.stream); if (err != cudaSuccess) { return err; } diff --git a/include/operators/gaussian_blur.h b/include/operators/gaussian_blur.h index 260ff29..fd39f1c 100644 --- a/include/operators/gaussian_blur.h +++ b/include/operators/gaussian_blur.h @@ -1,5 +1,6 @@ #pragma once +#include "memory_manager.h" #include "operator.h" #include "types.h" @@ -7,7 +8,8 @@ namespace mini_image_pipe { class GaussianBlurOperator : public IOperator { public: - explicit GaussianBlurOperator(GaussianKernelSize size, float sigma = 0.0f); + explicit GaussianBlurOperator(GaussianKernelSize size, float sigma = 0.0f, + IRuntimeAllocator& allocator = MemoryManager::getInstance()); ~GaussianBlurOperator() override; cudaError_t execute(const void* input, void* output, int width, int height, int channels, @@ -38,6 +40,7 @@ class GaussianBlurOperator : public IOperator { float* d_kernel_ = nullptr; // 1D kernel on device void* d_intermediate_ = nullptr; // Intermediate buffer for separable filter size_t intermediateSize_ = 0; + IRuntimeAllocator& allocator_; void generateKernel(); void freeResources(); diff --git a/include/operators/sobel.h b/include/operators/sobel.h index a139ce6..66cac6f 100644 --- a/include/operators/sobel.h +++ b/include/operators/sobel.h @@ -13,6 +13,8 @@ class SobelOperator : public IOperator { cudaError_t execute(const void* input, void* output, int width, int height, int channels, cudaStream_t stream) override; + KernelConfig getKernelConfig(int width, int height, int channels) const; + void getOutputDimensions(int inputWidth, int inputHeight, int inputChannels, int& outputWidth, int& outputHeight, int& outputChannels) const override; diff --git a/include/pipeline.h b/include/pipeline.h index d9d631f..70ad2c5 100644 --- a/include/pipeline.h +++ b/include/pipeline.h @@ -15,7 +15,8 @@ namespace mini_image_pipe { class Pipeline { public: - explicit Pipeline(const PipelineConfig& config = PipelineConfig()); + explicit Pipeline(const PipelineConfig& config = PipelineConfig(), + IRuntimeAllocator& allocator = MemoryManager::getInstance()); ~Pipeline(); // Add operator to pipeline, returns node ID @@ -34,8 +35,8 @@ class Pipeline { cudaError_t execute(); // Execute batch of frames - cudaError_t executeBatch(const std::vector& inputs, std::vector& outputs, - int width, int height, int channels); + cudaError_t executeBatch(const std::vector& inputs, + std::vector& outputs); // Update operator parameters at runtime template @@ -55,7 +56,7 @@ class Pipeline { PipelineConfig config_; TaskGraph graph_; DAGScheduler scheduler_; - MemoryManager& memMgr_; + IRuntimeAllocator& memMgr_; std::unordered_map intermediateBuffers_; std::unordered_map bufferSizes_; @@ -66,9 +67,11 @@ class Pipeline { std::unordered_map> parameters_; cudaError_t allocateIntermediateBuffers(); + cudaError_t validateBatchInputs(const std::vector& inputs) const; + cudaError_t collectBatchOutputs(size_t batchSize, std::vector& outputs) const; + void clearBatchInputs(); void freeIntermediateBuffers(); void freeOperatorResources(); - void setupBufferConnections(); void findInputOutputNodes(); }; diff --git a/include/scheduler.h b/include/scheduler.h index d0cf2c0..854e45d 100644 --- a/include/scheduler.h +++ b/include/scheduler.h @@ -20,6 +20,21 @@ struct TaskProfileRecord { cudaError_t status = cudaSuccess; }; +struct SchedulerExecutionTrace { + std::unordered_map taskStreamMap; + std::vector> synchronizations; + std::vector profileRecords; +}; + +struct SchedulerGraphCaptureState { + bool enabled = false; + bool replayedLastGraph = false; + bool signatureValid = false; + std::string lastGraphSignature; + cudaGraph_t graph = nullptr; + cudaGraphExec_t graphExec = nullptr; +}; + class DAGScheduler { public: explicit DAGScheduler(int numStreams = 4); @@ -41,12 +56,14 @@ class DAGScheduler { bool hasSynchronization(int fromTask, int toTask) const; void setGraphExecutionEnabled(bool enabled); - bool isGraphExecutionEnabled() const { return graphExecutionEnabled_; } - bool didReplayLastGraph() const { return replayedLastGraph_; } - bool hasCapturedGraph() const { return graphExec_ != nullptr; } - const std::string& getLastGraphSignatureForTesting() const { return lastGraphSignature_; } + bool isGraphExecutionEnabled() const { return graphCapture_.enabled; } + bool didReplayLastGraph() const { return graphCapture_.replayedLastGraph; } + bool hasCapturedGraph() const { return graphCapture_.graphExec != nullptr; } + const std::string& getLastGraphSignatureForTesting() const { + return graphCapture_.lastGraphSignature; + } const std::vector& getLastProfileRecords() const { - return lastProfileRecords_; + return executionTrace_.profileRecords; } private: @@ -54,16 +71,8 @@ class DAGScheduler { std::vector streams_; std::vector taskEvents_; // One event per task std::function errorCallback_; - - std::unordered_map taskStreamMap_; - std::vector> synchronizations_; // (from, to) pairs - bool graphExecutionEnabled_ = false; - bool replayedLastGraph_ = false; - bool graphSignatureValid_ = false; - std::string lastGraphSignature_; - std::vector lastProfileRecords_; - cudaGraph_t graph_ = nullptr; - cudaGraphExec_t graphExec_ = nullptr; + SchedulerExecutionTrace executionTrace_; + SchedulerGraphCaptureState graphCapture_; // Assign stream to task based on dependencies int assignStream(TaskNode& task, const TaskGraph& graph); diff --git a/include/task_graph.h b/include/task_graph.h index 7f2c1ca..8523f02 100644 --- a/include/task_graph.h +++ b/include/task_graph.h @@ -12,6 +12,40 @@ namespace mini_image_pipe { // Represents a single task in the DAG +struct TaskRuntimeState { + std::atomic state{TaskState::PENDING}; + int assignedStream = -1; // CUDA stream index + bool initialized = false; + bool profilingEnabled = false; + std::vector inputImages; + ImageBuffer outputImage; + OperatorWorkspace workspace; + + TaskRuntimeState() = default; + + TaskRuntimeState(const TaskRuntimeState& other) + : state(other.state.load()), + assignedStream(other.assignedStream), + initialized(other.initialized), + profilingEnabled(other.profilingEnabled), + inputImages(other.inputImages), + outputImage(other.outputImage), + workspace(other.workspace) {} + + TaskRuntimeState& operator=(const TaskRuntimeState& other) { + if (this != &other) { + state.store(other.state.load()); + assignedStream = other.assignedStream; + initialized = other.initialized; + profilingEnabled = other.profilingEnabled; + inputImages = other.inputImages; + outputImage = other.outputImage; + workspace = other.workspace; + } + return *this; + } +}; + struct TaskNode { int id = -1; std::string name; @@ -19,8 +53,6 @@ struct TaskNode { std::vector dependencies; // IDs of upstream tasks std::vector dependents; // IDs of downstream tasks - std::atomic state{TaskState::PENDING}; - void* inputBuffer = nullptr; void* outputBuffer = nullptr; int width = 0; @@ -29,13 +61,7 @@ struct TaskNode { int outputWidth = 0; int outputHeight = 0; int outputChannels = 0; - - int assignedStream = -1; // CUDA stream index - bool initialized = false; - bool profilingEnabled = false; - std::vector inputImages; - ImageBuffer outputImage; - OperatorWorkspace workspace; + TaskRuntimeState runtime; TaskNode() = default; TaskNode(int id, const std::string& name, OperatorPtr op) : id(id), name(name), op(op) {} @@ -47,7 +73,6 @@ struct TaskNode { op(other.op), dependencies(other.dependencies), dependents(other.dependents), - state(other.state.load()), inputBuffer(other.inputBuffer), outputBuffer(other.outputBuffer), width(other.width), @@ -56,12 +81,7 @@ struct TaskNode { outputWidth(other.outputWidth), outputHeight(other.outputHeight), outputChannels(other.outputChannels), - assignedStream(other.assignedStream), - initialized(other.initialized), - profilingEnabled(other.profilingEnabled), - inputImages(other.inputImages), - outputImage(other.outputImage), - workspace(other.workspace) {} + runtime(other.runtime) {} // Copy assignment operator for atomic member TaskNode& operator=(const TaskNode& other) { @@ -71,7 +91,6 @@ struct TaskNode { op = other.op; dependencies = other.dependencies; dependents = other.dependents; - state.store(other.state.load()); inputBuffer = other.inputBuffer; outputBuffer = other.outputBuffer; width = other.width; @@ -80,12 +99,7 @@ struct TaskNode { outputWidth = other.outputWidth; outputHeight = other.outputHeight; outputChannels = other.outputChannels; - assignedStream = other.assignedStream; - initialized = other.initialized; - profilingEnabled = other.profilingEnabled; - inputImages = other.inputImages; - outputImage = other.outputImage; - workspace = other.workspace; + runtime = other.runtime; } return *this; } diff --git a/include/types.h b/include/types.h index 9c6c483..0227dd3 100644 --- a/include/types.h +++ b/include/types.h @@ -4,6 +4,7 @@ #include #include +#include namespace mini_image_pipe { @@ -89,6 +90,12 @@ struct PipelineConfig { bool preferAsyncAllocator = false; // Prefer stream-ordered async device allocation }; +struct BatchOutput { + int nodeId = -1; + ImageBuffer image; + std::vector frames; +}; + enum class DeviceAllocatorMode { UNKNOWN, LEGACY_POOL, ASYNC_STREAM_ORDERED }; // Color conversion types diff --git a/scripts/docs-quality/check-links.mjs b/scripts/docs-quality/check-links.mjs index 8e5d6d8..b6854d2 100644 --- a/scripts/docs-quality/check-links.mjs +++ b/scripts/docs-quality/check-links.mjs @@ -4,17 +4,11 @@ * Failures are blocking by default. */ import { readFileSync, readdirSync, statSync } from 'node:fs'; -import { basename, dirname, join, resolve, relative } from 'node:path'; +import { dirname, join, resolve, relative } from 'node:path'; import { fileURLToPath } from 'node:url'; const __dirname = dirname(fileURLToPath(import.meta.url)); const root = resolve(__dirname, '..', '..', 'docs'); -const SITE_BASE = resolve(__dirname, '..', '..', 'docs', '.vitepress', 'dist'); - -const BILATERAL_WHITELIST = new Set([ - 'whitepaper-prompts.md', - '.vitepress/dist', -]); function* walk(dir) { for (const entry of readdirSync(dir, { withFileTypes: true })) { @@ -27,35 +21,48 @@ function* walk(dir) { } } -function getDistPath(relMdPath) { - // markdown relative to docs -> dist relative - let distRel = relMdPath.replace(/\.md$/, '.html'); - // index.md -> index.html (VitePress cleanUrls: false) - if (distRel.endsWith('/index.html')) { - distRel = distRel.slice(0, -'index.html'.length) + 'index.html'; - } - return join(SITE_BASE, distRel); +function fileExists(path) { + return statSync(path, { throwIfNoEntry: false })?.isFile() ?? false; +} + +function pathCandidates(basePath, rawPath) { + const candidates = []; + const seen = new Set(); + + const addCandidate = (candidate) => { + if (!candidate || seen.has(candidate)) return; + seen.add(candidate); + candidates.push(candidate); + }; + + addCandidate(basePath); + addCandidate(basePath.replace(/\.html$/, '.md')); + addCandidate(basePath + '.md'); + addCandidate(join(basePath, 'index.md')); + + return candidates; } function resolveLink(srcPath, raw) { if (raw.startsWith('http://') || raw.startsWith('https://') || raw.startsWith('#')) { return null; // external/hash links not checked } - if (raw.startsWith('/')) { - // site-root link - const clean = raw.replace(/\.html$/, ''); - const target = join(SITE_BASE, clean.endsWith('/') ? clean + 'index.html' : clean + '.html'); - return target; + const clean = raw.split('#', 1)[0].split('?', 1)[0]; + if (!clean) { + return null; + } + + const basePath = raw.startsWith('/') + ? resolve(root, '.' + clean) + : resolve(dirname(srcPath), clean); + + for (const candidate of pathCandidates(basePath, clean)) { + if (fileExists(candidate)) { + return candidate; + } } - // relative to source file - const baseDir = dirname(srcPath); - const target = resolve(baseDir, raw); - // VitePress cleanUrls: false -> .html extension - if (target.endsWith('.md')) return target; - if (statSync(target, { throwIfNoEntry: false })?.isFile()) return target; - const withHtml = target + '.html'; - if (statSync(withHtml, { throwIfNoEntry: false })?.isFile()) return withHtml; - return target; + + return pathCandidates(basePath, clean)[0]; } const issues = []; diff --git a/src/memory_manager.cu b/src/memory_manager.cu index a78bedb..94501f4 100644 --- a/src/memory_manager.cu +++ b/src/memory_manager.cu @@ -235,7 +235,8 @@ OperatorWorkspace MemoryManager::allocateWorkspace(const WorkspaceRequirements& if (requirements.hostBytes > 0) { workspace.host = std::malloc(requirements.hostBytes); if (!workspace.host) { - std::cerr << "Error: malloc failed for workspace host bytes " << requirements.hostBytes << std::endl; + std::cerr << "Error: malloc failed for workspace host bytes " << requirements.hostBytes + << std::endl; freeWorkspace(workspace, stream); return {}; } diff --git a/src/operators/gaussian_blur.cu b/src/operators/gaussian_blur.cu index 367457f..44a157f 100644 --- a/src/operators/gaussian_blur.cu +++ b/src/operators/gaussian_blur.cu @@ -121,8 +121,9 @@ __global__ void gaussianVerticalKernel(const float* input, uint8_t* output, int } } -GaussianBlurOperator::GaussianBlurOperator(GaussianKernelSize size, float sigma) - : kernelSize_(size), sigma_(sigma) { +GaussianBlurOperator::GaussianBlurOperator(GaussianKernelSize size, float sigma, + IRuntimeAllocator& allocator) + : kernelSize_(size), sigma_(sigma), allocator_(allocator) { generateKernel(); } @@ -180,7 +181,7 @@ void GaussianBlurOperator::freeResources() { d_kernel_ = nullptr; } if (d_intermediate_) { - MemoryManager::getInstance().freeDevice(d_intermediate_); + allocator_.freeDevice(d_intermediate_, nullptr); d_intermediate_ = nullptr; intermediateSize_ = 0; } @@ -213,8 +214,8 @@ WorkspaceRequirements GaussianBlurOperator::getWorkspaceRequirements( WorkspaceRequirements requirements; requirements.deviceBytes = static_cast(inputs.front().width) * inputs.front().height * - inputs.front().channels * - std::max(inputs.front().batchSize, 1) * sizeof(float); + inputs.front().channels * std::max(inputs.front().batchSize, 1) * + sizeof(float); return requirements; } @@ -231,19 +232,20 @@ cudaError_t GaussianBlurOperator::executeBuffers(const std::vector& input.channels, context.stream, context.workspace.device); } - size_t workspaceStride = static_cast(input.width) * input.height * input.channels * - sizeof(float); + size_t workspaceStride = + static_cast(input.width) * input.height * input.channels * sizeof(float); if (input.batchStride == 0 || output.batchStride == 0) { return cudaErrorInvalidValue; } for (int batchIndex = 0; batchIndex < input.batchSize; ++batchIndex) { - const void* inputPtr = static_cast(input.data) + input.batchStride * batchIndex; + const void* inputPtr = + static_cast(input.data) + input.batchStride * batchIndex; void* outputPtr = static_cast(output.data) + output.batchStride * batchIndex; - void* workspacePtr = context.workspace.device - ? static_cast(context.workspace.device) + - workspaceStride * batchIndex - : nullptr; + void* workspacePtr = + context.workspace.device + ? static_cast(context.workspace.device) + workspaceStride * batchIndex + : nullptr; cudaError_t err = executeWithWorkspace(inputPtr, outputPtr, input.width, input.height, input.channels, context.stream, workspacePtr); if (err != cudaSuccess) { @@ -275,11 +277,10 @@ cudaError_t GaussianBlurOperator::executeWithWorkspace(const void* input, void* // Legacy direct-execute path keeps the previous fallback allocation behavior. size_t requiredSize = static_cast(width) * height * channels * sizeof(float); if (intermediateSize_ < requiredSize) { - MemoryManager& mgr = MemoryManager::getInstance(); if (d_intermediate_) { - mgr.freeDevice(d_intermediate_); + allocator_.freeDevice(d_intermediate_, nullptr); } - d_intermediate_ = mgr.allocateDevice(requiredSize); + d_intermediate_ = allocator_.allocateDevice(requiredSize, nullptr); if (!d_intermediate_) { intermediateSize_ = 0; return cudaErrorMemoryAllocation; diff --git a/src/operators/merge_average.cu b/src/operators/merge_average.cu index fe89d1a..87f4472 100644 --- a/src/operators/merge_average.cu +++ b/src/operators/merge_average.cu @@ -11,7 +11,8 @@ __global__ void mergeAverageKernel(const uint8_t* const* inputs, uint8_t* output int x = blockIdx.x * blockDim.x + threadIdx.x; int y = blockIdx.y * blockDim.y + threadIdx.y; - if (x >= width || y >= height) return; + if (x >= width || y >= height) + return; int pixelIdx = (y * width + x) * channels; for (int c = 0; c < channels; ++c) { @@ -49,7 +50,8 @@ cudaError_t MergeAverageOperator::executeBuffers(const std::vector& ImageBuffer& output, const OperatorExecutionContext& context) { cudaError_t err = validateInputs(inputs); - if (err != cudaSuccess) return err; + if (err != cudaSuccess) + return err; if (!output.data || !output.isValid()) { return cudaErrorInvalidValue; @@ -69,7 +71,8 @@ cudaError_t MergeAverageOperator::executeBuffers(const std::vector& const uint8_t** dPointers = nullptr; err = cudaMalloc(&dPointers, numInputs * sizeof(const uint8_t*)); - if (err != cudaSuccess) return err; + if (err != cudaSuccess) + return err; err = cudaMemcpy(dPointers, hPointers.data(), numInputs * sizeof(const uint8_t*), cudaMemcpyHostToDevice); if (err != cudaSuccess) { @@ -96,8 +99,7 @@ cudaError_t MergeAverageOperator::validateInputs(const std::vector& const auto& first = inputs.front(); for (size_t i = 1; i < inputs.size(); ++i) { const auto& in = inputs[i]; - if (in.width != first.width || in.height != first.height || - in.channels != first.channels) { + if (in.width != first.width || in.height != first.height || in.channels != first.channels) { return cudaErrorInvalidValue; } if (!in.isValid()) { diff --git a/src/operators/sobel.cu b/src/operators/sobel.cu index f7f9912..337cd5e 100644 --- a/src/operators/sobel.cu +++ b/src/operators/sobel.cu @@ -10,9 +10,45 @@ namespace mini_image_pipe { __constant__ int c_sobelGx[9] = {-1, 0, 1, -2, 0, 2, -1, 0, 1}; __constant__ int c_sobelGy[9] = {-1, -2, -1, 0, 0, 0, 1, 2, 1}; +constexpr int kSobelBlockSize = 16; +constexpr int kSobelHalo = 1; + +__device__ float readLuminance(const uint8_t* input, int x, int y, int width, int channels) { + if (channels == 1) { + return static_cast(input[y * width + x]); + } + if (channels >= 3) { + int idx = (y * width + x) * channels; + float r = static_cast(input[idx]); + float g = static_cast(input[idx + 1]); + float b = static_cast(input[idx + 2]); + return 0.299f * r + 0.587f * g + 0.114f * b; + } + + return static_cast(input[(y * width + x) * channels]); +} + // CUDA kernel for Sobel edge detection __global__ void sobelKernel(const uint8_t* input, uint8_t* output, int width, int height, int channels) { + extern __shared__ float tile[]; + + const int tileWidth = blockDim.x + 2 * kSobelHalo; + const int tileHeight = blockDim.y + 2 * kSobelHalo; + const int blockOriginX = blockIdx.x * blockDim.x; + const int blockOriginY = blockIdx.y * blockDim.y; + + for (int sharedY = threadIdx.y; sharedY < tileHeight; sharedY += blockDim.y) { + for (int sharedX = threadIdx.x; sharedX < tileWidth; sharedX += blockDim.x) { + int globalX = min(max(blockOriginX + sharedX - kSobelHalo, 0), width - 1); + int globalY = min(max(blockOriginY + sharedY - kSobelHalo, 0), height - 1); + tile[sharedY * tileWidth + sharedX] = + readLuminance(input, globalX, globalY, width, channels); + } + } + + __syncthreads(); + int x = blockIdx.x * blockDim.x + threadIdx.x; int y = blockIdx.y * blockDim.y + threadIdx.y; @@ -22,28 +58,12 @@ __global__ void sobelKernel(const uint8_t* input, uint8_t* output, int width, in float gx = 0.0f; float gy = 0.0f; - // Sobel Gx kernel: [-1 0 1; -2 0 2; -1 0 1] - // Sobel Gy kernel: [-1 -2 -1; 0 0 0; 1 2 1] for (int ky = -1; ky <= 1; ky++) { for (int kx = -1; kx <= 1; kx++) { - int nx = min(max(x + kx, 0), width - 1); - int ny = min(max(y + ky, 0), height - 1); - - float pixelValue; - if (channels == 1) { - pixelValue = static_cast(input[ny * width + nx]); - } else if (channels >= 3) { - int idx = (ny * width + nx) * channels; - float r = static_cast(input[idx]); - float g = static_cast(input[idx + 1]); - float b = static_cast(input[idx + 2]); - pixelValue = 0.299f * r + 0.587f * g + 0.114f * b; - } else { - // channels == 2: use first channel only (grayscale with alpha) - pixelValue = static_cast(input[(ny * width + nx) * channels]); - } - int kidx = (ky + 1) * 3 + (kx + 1); + int tileX = threadIdx.x + kx + kSobelHalo; + int tileY = threadIdx.y + ky + kSobelHalo; + float pixelValue = tile[tileY * tileWidth + tileX]; gx += pixelValue * c_sobelGx[kidx]; gy += pixelValue * c_sobelGy[kidx]; } @@ -58,6 +78,17 @@ __global__ void sobelKernel(const uint8_t* input, uint8_t* output, int width, in SobelOperator::SobelOperator() = default; +KernelConfig SobelOperator::getKernelConfig(int width, int height, int channels) const { + (void)channels; + KernelConfig config; + config.blockSize = dim3(kSobelBlockSize, kSobelBlockSize); + config.gridSize = dim3((width + config.blockSize.x - 1) / config.blockSize.x, + (height + config.blockSize.y - 1) / config.blockSize.y); + config.sharedMem = static_cast(config.blockSize.x + 2 * kSobelHalo) * + static_cast(config.blockSize.y + 2 * kSobelHalo) * sizeof(float); + return config; +} + cudaError_t SobelOperator::execute(const void* input, void* output, int width, int height, int channels, cudaStream_t stream) { if (!input || !output || width <= 0 || height <= 0 || channels <= 0) { @@ -67,11 +98,10 @@ cudaError_t SobelOperator::execute(const void* input, void* output, int width, i const uint8_t* inputPtr = static_cast(input); uint8_t* outputPtr = static_cast(output); - dim3 blockSize(16, 16); - dim3 gridSize((width + blockSize.x - 1) / blockSize.x, - (height + blockSize.y - 1) / blockSize.y); + KernelConfig config = getKernelConfig(width, height, channels); - sobelKernel<<>>(inputPtr, outputPtr, width, height, channels); + sobelKernel<<>>( + inputPtr, outputPtr, width, height, channels); return cudaGetLastError(); } diff --git a/src/pipeline.cpp b/src/pipeline.cpp index e9f4ac8..d53c8b8 100644 --- a/src/pipeline.cpp +++ b/src/pipeline.cpp @@ -2,6 +2,7 @@ #include #include +#include namespace mini_image_pipe { @@ -30,10 +31,38 @@ bool needsWorkspaceReallocation(const OperatorWorkspace& workspace, workspace.requirements.deviceBytes < requirements.deviceBytes; } +bool hasContiguousFrameLayout(const ImageBuffer& buffer) { + return buffer.stride == buffer.width * buffer.channels * static_cast(buffer.elementSize); +} + +bool matchesBatchShape(const ImageBuffer& lhs, const ImageBuffer& rhs) { + return lhs.width == rhs.width && lhs.height == rhs.height && lhs.channels == rhs.channels && + lhs.stride == rhs.stride && lhs.elementSize == rhs.elementSize; +} + +bool isDeviceAccessiblePointer(void* ptr) { + if (!ptr) { + return false; + } + + cudaPointerAttributes attributes{}; + cudaError_t attrErr = cudaPointerGetAttributes(&attributes, ptr); + if (attrErr != cudaSuccess) { + cudaGetLastError(); + return false; + } + +#if CUDART_VERSION >= 10000 + return attributes.type == cudaMemoryTypeDevice || attributes.type == cudaMemoryTypeManaged; +#else + return attributes.memoryType == cudaMemoryTypeDevice; +#endif +} + } // namespace -Pipeline::Pipeline(const PipelineConfig& config) - : config_(config), scheduler_(config.numStreams), memMgr_(MemoryManager::getInstance()) { +Pipeline::Pipeline(const PipelineConfig& config, IRuntimeAllocator& allocator) + : config_(config), scheduler_(config.numStreams), memMgr_(allocator) { scheduler_.setGraphExecutionEnabled(config_.enableCudaGraphs); memMgr_.setDeviceAllocatorMode(config_.preferAsyncAllocator ? DeviceAllocatorMode::ASYNC_STREAM_ORDERED @@ -62,8 +91,8 @@ void Pipeline::setInput(int nodeId, void* data, int width, int height, int chann task->width = width; task->height = height; task->channels = channels; - task->inputImages = {makeRuntimeImageBuffer(data, width, height, channels)}; - task->profilingEnabled = config_.enableProfiling; + task->runtime.inputImages = {makeRuntimeImageBuffer(data, width, height, channels)}; + task->runtime.profilingEnabled = config_.enableProfiling; // Calculate output dimensions if (task->op) { @@ -128,16 +157,16 @@ cudaError_t Pipeline::allocateIntermediateBuffers() { if (!task) continue; - std::vector presetInputs = task->inputImages; - task->inputImages.clear(); + std::vector presetInputs = task->runtime.inputImages; + task->runtime.inputImages.clear(); if (task->dependencies.empty()) { if (!presetInputs.empty()) { - task->inputImages = presetInputs; + task->runtime.inputImages = presetInputs; } else if (task->inputBuffer && task->width > 0 && task->height > 0 && task->channels > 0) { - task->inputImages.push_back(makeRuntimeImageBuffer(task->inputBuffer, task->width, - task->height, task->channels)); + task->runtime.inputImages.push_back(makeRuntimeImageBuffer( + task->inputBuffer, task->width, task->height, task->channels)); } } else { for (int depId : task->dependencies) { @@ -145,37 +174,37 @@ cudaError_t Pipeline::allocateIntermediateBuffers() { if (!dep) continue; - if (dep->outputImage.isValid()) { - task->inputImages.push_back(dep->outputImage); + if (dep->runtime.outputImage.isValid()) { + task->runtime.inputImages.push_back(dep->runtime.outputImage); } else if (dep->outputBuffer && dep->outputWidth > 0 && dep->outputHeight > 0 && dep->outputChannels > 0) { - task->inputImages.push_back( + task->runtime.inputImages.push_back( makeRuntimeImageBuffer(dep->outputBuffer, dep->outputWidth, dep->outputHeight, dep->outputChannels)); } } } - if (task->inputImages.empty()) { + if (task->runtime.inputImages.empty()) { return cudaErrorInvalidValue; } - const ImageBuffer& primaryInput = task->inputImages.front(); + const ImageBuffer& primaryInput = task->runtime.inputImages.front(); task->inputBuffer = primaryInput.data; task->width = primaryInput.width; task->height = primaryInput.height; task->channels = primaryInput.channels; - if (!task->op->supportsInputCount(task->inputImages.size())) { + if (!task->op->supportsInputCount(task->runtime.inputImages.size())) { return cudaErrorInvalidValue; } - if (!task->initialized) { + if (!task->runtime.initialized) { cudaError_t initErr = task->op->initialize(); if (initErr != cudaSuccess) { return initErr; } - task->initialized = true; + task->runtime.initialized = true; } if (task->op && task->width > 0 && task->height > 0) { @@ -216,29 +245,59 @@ cudaError_t Pipeline::allocateIntermediateBuffers() { size_t outputBatchStride = static_cast(task->outputWidth) * task->outputHeight * task->outputChannels * primaryInput.elementSize; - task->outputImage = - makeRuntimeImageBuffer(buffer, task->outputWidth, task->outputHeight, - task->outputChannels, true, primaryInput.batchSize, - outputBatchStride); + task->runtime.outputImage = makeRuntimeImageBuffer( + buffer, task->outputWidth, task->outputHeight, task->outputChannels, true, + primaryInput.batchSize, outputBatchStride); } else { return cudaErrorMemoryAllocation; } } - WorkspaceRequirements requirements = task->op->getWorkspaceRequirements(task->inputImages); - if (needsWorkspaceReallocation(task->workspace, requirements)) { - memMgr_.freeWorkspace(task->workspace, nullptr); - task->workspace = memMgr_.allocateWorkspace(requirements, nullptr); - if ((requirements.hostBytes > 0 && !task->workspace.host) || - (requirements.pinnedBytes > 0 && !task->workspace.pinned) || - (requirements.deviceBytes > 0 && !task->workspace.device)) { + WorkspaceRequirements requirements = + task->op->getWorkspaceRequirements(task->runtime.inputImages); + if (needsWorkspaceReallocation(task->runtime.workspace, requirements)) { + memMgr_.freeWorkspace(task->runtime.workspace, nullptr); + task->runtime.workspace = memMgr_.allocateWorkspace(requirements, nullptr); + if ((requirements.hostBytes > 0 && !task->runtime.workspace.host) || + (requirements.pinnedBytes > 0 && !task->runtime.workspace.pinned) || + (requirements.deviceBytes > 0 && !task->runtime.workspace.device)) { return cudaErrorMemoryAllocation; } } else { - task->workspace.requirements = requirements; + task->runtime.workspace.requirements = requirements; } - task->profilingEnabled = config_.enableProfiling; + task->runtime.profilingEnabled = config_.enableProfiling; + } + + return cudaSuccess; +} + +cudaError_t Pipeline::validateBatchInputs(const std::vector& inputs) const { + if (inputs.empty()) { + return cudaErrorInvalidValue; + } + + if (config_.maxBatchSize > 0 && inputs.size() > static_cast(config_.maxBatchSize)) { + return cudaErrorInvalidValue; + } + + const ImageBuffer& reference = inputs.front(); + if (!reference.isValid() || !reference.isDeviceMemory || reference.batchSize != 1 || + !hasContiguousFrameLayout(reference) || reference.elementSize != sizeof(uint8_t)) { + return cudaErrorInvalidValue; + } + + for (const auto& input : inputs) { + if (!input.isValid() || !input.isDeviceMemory || input.batchSize != 1 || + !hasContiguousFrameLayout(input) || input.elementSize != sizeof(uint8_t) || + !matchesBatchShape(reference, input)) { + return cudaErrorInvalidValue; + } + + if (!isDeviceAccessiblePointer(input.data)) { + return cudaErrorInvalidValue; + } } return cudaSuccess; @@ -254,27 +313,50 @@ void Pipeline::freeIntermediateBuffers() { void Pipeline::freeOperatorResources() { for (auto& task : graph_.getTasks()) { - memMgr_.freeWorkspace(task.workspace, nullptr); - task.workspace = {}; - if (task.initialized && task.op) { + memMgr_.freeWorkspace(task.runtime.workspace, nullptr); + task.runtime.workspace = {}; + if (task.runtime.initialized && task.op) { task.op->shutdown(); - task.initialized = false; + task.runtime.initialized = false; } } } -void Pipeline::setupBufferConnections() { - for (auto& task : graph_.getTasks()) { - if (!task.dependencies.empty()) { - int depId = task.dependencies[0]; - TaskNode* dep = graph_.getTask(depId); - if (dep) { - task.inputBuffer = dep->outputBuffer; - task.width = dep->outputWidth; - task.height = dep->outputHeight; - task.channels = dep->outputChannels; - } +cudaError_t Pipeline::collectBatchOutputs(size_t batchSize, + std::vector& outputs) const { + outputs.clear(); + outputs.reserve(outputNodes_.size()); + + for (int nodeId : outputNodes_) { + const TaskNode* outputTask = graph_.getTask(nodeId); + if (!outputTask || !outputTask->runtime.outputImage.isValid() || + outputTask->runtime.outputImage.batchStride == 0) { + outputs.clear(); + return cudaErrorInvalidValue; } + + BatchOutput output; + output.nodeId = nodeId; + output.image = outputTask->runtime.outputImage; + output.frames.reserve(batchSize); + for (size_t i = 0; i < batchSize; ++i) { + output.frames.push_back(static_cast(output.image.data) + + output.image.batchStride * i); + } + outputs.push_back(std::move(output)); + } + + return cudaSuccess; +} + +void Pipeline::clearBatchInputs() { + for (int nodeId : inputNodes_) { + TaskNode* task = graph_.getTask(nodeId); + if (!task) { + continue; + } + task->inputBuffer = nullptr; + task->runtime.inputImages.clear(); } } @@ -306,13 +388,13 @@ cudaError_t Pipeline::execute() { } for (const auto& task : graph_.getTasks()) { - if (task.inputImages.empty()) { + if (task.runtime.inputImages.empty()) { return cudaErrorInvalidValue; } if (task.outputWidth <= 0 || task.outputHeight <= 0 || task.outputChannels <= 0) { return cudaErrorInvalidValue; } - if (!task.outputImage.isValid()) { + if (!task.runtime.outputImage.isValid()) { return cudaErrorMemoryAllocation; } } @@ -322,16 +404,8 @@ cudaError_t Pipeline::execute() { return err; } -cudaError_t Pipeline::executeBatch(const std::vector& inputs, std::vector& outputs, - int width, int height, int channels) { - if (inputs.empty()) { - return cudaErrorInvalidValue; - } - - if (config_.maxBatchSize > 0 && inputs.size() > static_cast(config_.maxBatchSize)) { - return cudaErrorInvalidValue; - } - +cudaError_t Pipeline::executeBatch(const std::vector& inputs, + std::vector& outputs) { // Pre-check: identify input nodes (execute() will re-discover, but we need // to validate that at least one input node exists before entering the loop) findInputOutputNodes(); @@ -339,13 +413,19 @@ cudaError_t Pipeline::executeBatch(const std::vector& inputs, std::vector return cudaErrorInvalidValue; } + cudaError_t validationErr = validateBatchInputs(inputs); + if (validationErr != cudaSuccess) { + return validationErr; + } + cudaStream_t stagingStream = nullptr; cudaError_t streamErr = cudaStreamCreate(&stagingStream); if (streamErr != cudaSuccess) { return streamErr; } - size_t singleImageBytes = static_cast(width) * height * channels * sizeof(uint8_t); + const ImageBuffer& referenceInput = inputs.front(); + size_t singleImageBytes = referenceInput.sizeInBytes(); size_t batchStride = singleImageBytes; size_t totalBatchBytes = batchStride * inputs.size(); void* batchInput = memMgr_.allocateDevice(totalBatchBytes, stagingStream); @@ -356,7 +436,7 @@ cudaError_t Pipeline::executeBatch(const std::vector& inputs, std::vector for (size_t i = 0; i < inputs.size(); ++i) { cudaError_t copyErr = - cudaMemcpyAsync(static_cast(batchInput) + batchStride * i, inputs[i], + cudaMemcpyAsync(static_cast(batchInput) + batchStride * i, inputs[i].data, singleImageBytes, cudaMemcpyDeviceToDevice, stagingStream); if (copyErr != cudaSuccess) { memMgr_.freeDevice(batchInput, stagingStream); @@ -377,41 +457,35 @@ cudaError_t Pipeline::executeBatch(const std::vector& inputs, std::vector continue; } task->inputBuffer = batchInput; - task->width = width; - task->height = height; - task->channels = channels; - task->inputImages = {makeRuntimeImageBuffer(batchInput, width, height, channels, true, - static_cast(inputs.size()), batchStride)}; - task->profilingEnabled = config_.enableProfiling; + task->width = referenceInput.width; + task->height = referenceInput.height; + task->channels = referenceInput.channels; + task->runtime.inputImages = {makeRuntimeImageBuffer( + batchInput, referenceInput.width, referenceInput.height, referenceInput.channels, true, + static_cast(inputs.size()), batchStride)}; + task->runtime.profilingEnabled = config_.enableProfiling; } - outputs.assign(inputs.size(), nullptr); + outputs.clear(); cudaError_t err = execute(); if (err != cudaSuccess) { + outputs.clear(); + clearBatchInputs(); memMgr_.freeDevice(batchInput, stagingStream); cudaStreamDestroy(stagingStream); return err; } - if (!outputNodes_.empty()) { - TaskNode* outputTask = graph_.getTask(outputNodes_[0]); - if (outputTask && outputTask->outputImage.isValid() && outputTask->outputImage.batchStride > 0) { - for (size_t i = 0; i < inputs.size(); ++i) { - outputs[i] = static_cast(outputTask->outputImage.data) + - outputTask->outputImage.batchStride * i; - } - } - } - - for (int nodeId : inputNodes_) { - TaskNode* task = graph_.getTask(nodeId); - if (!task) { - continue; - } - task->inputBuffer = nullptr; - task->inputImages.clear(); + cudaError_t collectErr = collectBatchOutputs(inputs.size(), outputs); + if (collectErr != cudaSuccess) { + outputs.clear(); + clearBatchInputs(); + memMgr_.freeDevice(batchInput, stagingStream); + cudaStreamDestroy(stagingStream); + return collectErr; } + clearBatchInputs(); memMgr_.freeDevice(batchInput, stagingStream); cudaStreamDestroy(stagingStream); return cudaSuccess; diff --git a/src/scheduler.cu b/src/scheduler.cu index 52de5a4..e0e0078 100644 --- a/src/scheduler.cu +++ b/src/scheduler.cu @@ -28,8 +28,8 @@ ImageBuffer makeRuntimeImageBuffer(void* data, int width, int height, int channe } std::vector resolveInputImages(TaskNode& task) { - if (!task.inputImages.empty()) { - return task.inputImages; + if (!task.runtime.inputImages.empty()) { + return task.runtime.inputImages; } if (task.inputBuffer && task.width > 0 && task.height > 0 && task.channels > 0) { return {makeRuntimeImageBuffer(task.inputBuffer, task.width, task.height, task.channels)}; @@ -38,8 +38,8 @@ std::vector resolveInputImages(TaskNode& task) { } ImageBuffer resolveOutputImage(TaskNode& task) { - if (task.outputImage.isValid()) { - return task.outputImage; + if (task.runtime.outputImage.isValid()) { + return task.runtime.outputImage; } if (task.outputBuffer && task.outputWidth > 0 && task.outputHeight > 0 && task.outputChannels > 0) { @@ -95,11 +95,12 @@ void DAGScheduler::setErrorCallback(std::function } void DAGScheduler::setGraphExecutionEnabled(bool enabled) { - graphExecutionEnabled_ = enabled; + graphCapture_.enabled = enabled; if (!enabled) { invalidateCapturedGraph(); - graphSignatureValid_ = false; - lastGraphSignature_.clear(); + graphCapture_.signatureValid = false; + graphCapture_.replayedLastGraph = false; + graphCapture_.lastGraphSignature.clear(); } } @@ -112,8 +113,8 @@ int DAGScheduler::assignStream(TaskNode& task, const TaskGraph& graph) { // Try to find a stream different from dependencies for parallelism std::vector depStreams; for (int depId : task.dependencies) { - auto it = taskStreamMap_.find(depId); - if (it != taskStreamMap_.end()) { + auto it = executionTrace_.taskStreamMap.find(depId); + if (it != executionTrace_.taskStreamMap.end()) { depStreams.push_back(it->second); } } @@ -130,10 +131,11 @@ int DAGScheduler::assignStream(TaskNode& task, const TaskGraph& graph) { } void DAGScheduler::insertSynchronization(int fromTask, int toTask, TaskGraph& graph) { - auto fromIt = taskStreamMap_.find(fromTask); - auto toIt = taskStreamMap_.find(toTask); + auto fromIt = executionTrace_.taskStreamMap.find(fromTask); + auto toIt = executionTrace_.taskStreamMap.find(toTask); - if (fromIt == taskStreamMap_.end() || toIt == taskStreamMap_.end()) { + if (fromIt == executionTrace_.taskStreamMap.end() || + toIt == executionTrace_.taskStreamMap.end()) { return; } @@ -146,7 +148,7 @@ void DAGScheduler::insertSynchronization(int fromTask, int toTask, TaskGraph& gr if (fromTask < static_cast(taskEvents_.size()) && streams_[toStream] && taskEvents_[fromTask]) { cudaStreamWaitEvent(streams_[toStream], taskEvents_[fromTask], 0); - synchronizations_.push_back({fromTask, toTask}); + executionTrace_.synchronizations.push_back({fromTask, toTask}); } } } @@ -164,8 +166,8 @@ cudaError_t DAGScheduler::executeTask(TaskNode& task, cudaStream_t stream) { OperatorExecutionContext context; context.stream = stream; - context.profilingEnabled = task.profilingEnabled; - context.workspace = task.workspace; + context.profilingEnabled = task.runtime.profilingEnabled; + context.workspace = task.runtime.workspace; maybeRunProfilingSeam(context); const bool recordProfile = kProfilingInstrumentationCompiled && context.profilingEnabled; @@ -177,12 +179,12 @@ cudaError_t DAGScheduler::executeTask(TaskNode& task, cudaStream_t stream) { TaskProfileRecord record; record.taskId = task.id; record.taskName = task.name; - record.streamIndex = task.assignedStream; + record.streamIndex = task.runtime.assignedStream; record.durationMs = std::chrono::duration_cast>(end - start) .count(); record.status = status; - lastProfileRecords_.push_back(record); + executionTrace_.profileRecords.push_back(record); } return status; @@ -195,8 +197,8 @@ void DAGScheduler::propagateFailure(int taskId, TaskGraph& graph) { for (int depId : task->dependents) { TaskNode* dep = graph.getTask(depId); - if (dep && dep->state.load() == TaskState::PENDING) { - dep->state.store(TaskState::FAILED); + if (dep && dep->runtime.state.load() == TaskState::PENDING) { + dep->runtime.state.store(TaskState::FAILED); propagateFailure(depId, graph); } } @@ -246,14 +248,14 @@ cudaError_t DAGScheduler::executeDirect(TaskGraph& graph, const std::vector if (depFailed) { failed[taskId] = true; if (commitTaskState) { - task->state.store(TaskState::FAILED); + task->runtime.state.store(TaskState::FAILED); } continue; } int streamIdx = assignStream(*task, graph); - taskStreamMap_[taskId] = streamIdx; - task->assignedStream = streamIdx; + executionTrace_.taskStreamMap[taskId] = streamIdx; + task->runtime.assignedStream = streamIdx; if (recordEvents) { for (int depId : task->dependencies) { @@ -262,7 +264,7 @@ cudaError_t DAGScheduler::executeDirect(TaskGraph& graph, const std::vector } if (commitTaskState) { - task->state.store(TaskState::RUNNING); + task->runtime.state.store(TaskState::RUNNING); graph.incrementExecutionCount(taskId); } @@ -278,14 +280,14 @@ cudaError_t DAGScheduler::executeDirect(TaskGraph& graph, const std::vector lastError = err; if (commitTaskState) { - task->state.store(TaskState::FAILED); + task->runtime.state.store(TaskState::FAILED); if (errorCallback_) { errorCallback_(taskId, err); } propagateFailure(taskId, graph); } } else if (commitTaskState) { - task->state.store(TaskState::COMPLETED); + task->runtime.state.store(TaskState::COMPLETED); } } @@ -305,7 +307,7 @@ bool DAGScheduler::canCaptureGraph(const TaskGraph& graph) const { (void)graph; return false; #else - if (!graphExecutionEnabled_ || graph.size() == 0 || numStreams_ != 1) { + if (!graphCapture_.enabled || graph.size() == 0 || numStreams_ != 1) { return false; } @@ -319,7 +321,7 @@ bool DAGScheduler::isProfilingEnabled(const TaskGraph& graph) const { } for (const auto& task : graph.getTasks()) { - if (task.profilingEnabled) { + if (task.runtime.profilingEnabled) { return true; } } @@ -339,22 +341,22 @@ void DAGScheduler::recordGraphEvent(const char* eventName, cudaError_t status, d record.streamIndex = 0; record.durationMs = durationMs; record.status = status; - lastProfileRecords_.push_back(record); + executionTrace_.profileRecords.push_back(record); } void DAGScheduler::invalidateCapturedGraph() { - if (graphExec_) { - cudaGraphExecDestroy(graphExec_); - graphExec_ = nullptr; + if (graphCapture_.graphExec) { + cudaGraphExecDestroy(graphCapture_.graphExec); + graphCapture_.graphExec = nullptr; } - if (graph_) { - cudaGraphDestroy(graph_); - graph_ = nullptr; + if (graphCapture_.graph) { + cudaGraphDestroy(graphCapture_.graph); + graphCapture_.graph = nullptr; } } cudaError_t DAGScheduler::launchCapturedGraph(TaskGraph& graph, const std::vector& order) { - if (!graphExec_ || streams_.empty() || !streams_.front()) { + if (!graphCapture_.graphExec || streams_.empty() || !streams_.front()) { return cudaErrorInvalidResourceHandle; } @@ -363,21 +365,21 @@ cudaError_t DAGScheduler::launchCapturedGraph(TaskGraph& graph, const std::vecto if (!task) { continue; } - task->assignedStream = 0; - taskStreamMap_[taskId] = 0; - task->state.store(TaskState::RUNNING); + task->runtime.assignedStream = 0; + executionTrace_.taskStreamMap[taskId] = 0; + task->runtime.state.store(TaskState::RUNNING); graph.incrementExecutionCount(taskId); } const bool profilingEnabled = isProfilingEnabled(graph); auto start = std::chrono::steady_clock::now(); - cudaError_t launchErr = cudaGraphLaunch(graphExec_, streams_.front()); + cudaError_t launchErr = cudaGraphLaunch(graphCapture_.graphExec, streams_.front()); if (launchErr == cudaSuccess) { launchErr = cudaStreamSynchronize(streams_.front()); } auto end = std::chrono::steady_clock::now(); recordGraphEvent( - replayedLastGraph_ ? "graph-replay" : "graph-launch", launchErr, + graphCapture_.replayedLastGraph ? "graph-replay" : "graph-launch", launchErr, std::chrono::duration_cast>(end - start).count(), profilingEnabled); @@ -386,7 +388,8 @@ cudaError_t DAGScheduler::launchCapturedGraph(TaskGraph& graph, const std::vecto if (!task) { continue; } - task->state.store(launchErr == cudaSuccess ? TaskState::COMPLETED : TaskState::FAILED); + task->runtime.state.store(launchErr == cudaSuccess ? TaskState::COMPLETED + : TaskState::FAILED); } return launchErr; @@ -434,19 +437,19 @@ cudaError_t DAGScheduler::captureGraph(TaskGraph& graph, const std::vector& return instantiateErr; } - graph_ = capturedGraph; - graphExec_ = graphExec; - lastGraphSignature_ = signature; - graphSignatureValid_ = true; + graphCapture_.graph = capturedGraph; + graphCapture_.graphExec = graphExec; + graphCapture_.lastGraphSignature = signature; + graphCapture_.signatureValid = true; recordGraphEvent("graph-capture", cudaSuccess, durationMs, profilingEnabled); return cudaSuccess; } cudaError_t DAGScheduler::execute(TaskGraph& graph) { - taskStreamMap_.clear(); - synchronizations_.clear(); - replayedLastGraph_ = false; - lastProfileRecords_.clear(); + executionTrace_.taskStreamMap.clear(); + executionTrace_.synchronizations.clear(); + graphCapture_.replayedLastGraph = false; + executionTrace_.profileRecords.clear(); const size_t numTasks = graph.size(); std::vector order = graph.getTopologicalOrder(); @@ -455,42 +458,44 @@ cudaError_t DAGScheduler::execute(TaskGraph& graph) { } const bool profilingEnabled = isProfilingEnabled(graph); - if (!graphExecutionEnabled_) { + if (!graphCapture_.enabled) { invalidateCapturedGraph(); - graphSignatureValid_ = false; - lastGraphSignature_.clear(); + graphCapture_.signatureValid = false; + graphCapture_.lastGraphSignature.clear(); } const std::string signature = - graphExecutionEnabled_ ? buildGraphSignature(graph) : std::string(); + graphCapture_.enabled ? buildGraphSignature(graph) : std::string(); const bool signatureChanged = - graphExecutionEnabled_ && (!graphSignatureValid_ || signature != lastGraphSignature_); - if (graphExecutionEnabled_ && signatureChanged && hasCapturedGraph()) { + graphCapture_.enabled && + (!graphCapture_.signatureValid || signature != graphCapture_.lastGraphSignature); + if (graphCapture_.enabled && signatureChanged && hasCapturedGraph()) { recordGraphEvent("graph-invalidate", cudaSuccess, 0.0, profilingEnabled); invalidateCapturedGraph(); } - if (graphExecutionEnabled_ && numTasks == 0) { - replayedLastGraph_ = graphSignatureValid_ && signature == lastGraphSignature_; - lastGraphSignature_ = signature; - graphSignatureValid_ = true; + if (graphCapture_.enabled && numTasks == 0) { + graphCapture_.replayedLastGraph = + graphCapture_.signatureValid && signature == graphCapture_.lastGraphSignature; + graphCapture_.lastGraphSignature = signature; + graphCapture_.signatureValid = true; return cudaSuccess; } - if (graphExecutionEnabled_ && !signatureChanged && hasCapturedGraph()) { - replayedLastGraph_ = true; + if (graphCapture_.enabled && !signatureChanged && hasCapturedGraph()) { + graphCapture_.replayedLastGraph = true; return launchCapturedGraph(graph, order); } - if (graphExecutionEnabled_ && canCaptureGraph(graph)) { + if (graphCapture_.enabled && canCaptureGraph(graph)) { cudaError_t captureErr = captureGraph(graph, order, signature); if (captureErr == cudaSuccess && hasCapturedGraph()) { return launchCapturedGraph(graph, order); } invalidateCapturedGraph(); - } else if (graphExecutionEnabled_) { - lastGraphSignature_ = signature; - graphSignatureValid_ = true; + } else if (graphCapture_.enabled) { + graphCapture_.lastGraphSignature = signature; + graphCapture_.signatureValid = true; } cudaError_t eventErr = ensureTaskEvents(numTasks); @@ -502,12 +507,12 @@ cudaError_t DAGScheduler::execute(TaskGraph& graph) { } int DAGScheduler::getTaskStream(int taskId) const { - auto it = taskStreamMap_.find(taskId); - return (it != taskStreamMap_.end()) ? it->second : -1; + auto it = executionTrace_.taskStreamMap.find(taskId); + return (it != executionTrace_.taskStreamMap.end()) ? it->second : -1; } bool DAGScheduler::hasSynchronization(int fromTask, int toTask) const { - for (const auto& sync : synchronizations_) { + for (const auto& sync : executionTrace_.synchronizations) { if (sync.first == fromTask && sync.second == toTask) { return true; } @@ -522,15 +527,16 @@ std::string DAGScheduler::buildGraphSignature(const TaskGraph& graph) const { signature << task.id << ':' << task.width << 'x' << task.height << 'x' << task.channels << ':' << task.outputWidth << 'x' << task.outputHeight << 'x' << task.outputChannels << ':' << task.dependencies.size() << ':' - << task.inputImages.size() << ':' - << (task.inputImages.empty() ? 1 : task.inputImages.front().batchSize) << ':' - << reinterpret_cast(task.inputBuffer) << ':' + << task.runtime.inputImages.size() << ':' + << (task.runtime.inputImages.empty() ? 1 + : task.runtime.inputImages.front().batchSize) + << ':' << reinterpret_cast(task.inputBuffer) << ':' << reinterpret_cast(task.outputBuffer) << ':' - << reinterpret_cast(task.outputImage.data) << ';'; + << reinterpret_cast(task.runtime.outputImage.data) << ';'; for (int dep : task.dependencies) { signature << dep << ','; } - for (const auto& input : task.inputImages) { + for (const auto& input : task.runtime.inputImages) { signature << reinterpret_cast(input.data) << '@' << input.width << 'x' << input.height << 'x' << input.channels << '@' << input.batchSize << ','; } diff --git a/src/task_graph.cpp b/src/task_graph.cpp index 3160231..8b9f21d 100644 --- a/src/task_graph.cpp +++ b/src/task_graph.cpp @@ -124,13 +124,13 @@ std::vector TaskGraph::getReadyTasks() const { std::vector ready; for (const auto& node : nodes_) { - if (node.state.load() != TaskState::PENDING) { + if (node.runtime.state.load() != TaskState::PENDING) { continue; } bool allDepsCompleted = true; for (int depId : node.dependencies) { - if (nodes_[depId].state.load() != TaskState::COMPLETED) { + if (nodes_[depId].runtime.state.load() != TaskState::COMPLETED) { allDepsCompleted = false; break; } @@ -160,7 +160,8 @@ const TaskNode* TaskGraph::getTask(int id) const { void TaskGraph::reset() { for (auto& node : nodes_) { - node.state.store(TaskState::PENDING); + node.runtime.state.store(TaskState::PENDING); + node.runtime.assignedStream = -1; } resetExecutionCounts(); } diff --git a/tests/test_merge_average.cpp b/tests/test_merge_average.cpp index 0027b75..f773856 100644 --- a/tests/test_merge_average.cpp +++ b/tests/test_merge_average.cpp @@ -135,8 +135,7 @@ TEST(MergeAveragePropertyTest, ThreeInputsAverage) { cudaStreamSynchronize(stream); for (size_t i = 0; i < bytes; ++i) { - uint8_t expected = - static_cast((static_cast(hA[i]) + hB[i] + hC[i]) / 3); + uint8_t expected = static_cast((static_cast(hA[i]) + hB[i] + hC[i]) / 3); EXPECT_EQ(hOut[i], expected) << "Mismatch at byte " << i; } @@ -210,7 +209,9 @@ TEST(MergeAveragePropertyTest, PipelineForkJoinIntegration) { return cudaMemcpyAsync(output, input, size, cudaMemcpyDeviceToDevice, stream); } void getOutputDimensions(int iw, int ih, int ic, int& ow, int& oh, int& oc) const override { - ow = iw; oh = ih; oc = ic; + ow = iw; + oh = ih; + oc = ic; } const char* getName() const override { return "Identity"; } }; diff --git a/tests/test_pipeline.cpp b/tests/test_pipeline.cpp index 6289fe3..dedf733 100644 --- a/tests/test_pipeline.cpp +++ b/tests/test_pipeline.cpp @@ -12,6 +12,24 @@ using namespace mini_image_pipe; +namespace { + +ImageBuffer makeBatchFrame(void* data, int width, int height, int channels) { + ImageBuffer buffer; + buffer.data = data; + buffer.width = width; + buffer.height = height; + buffer.channels = channels; + buffer.stride = width * channels; + buffer.elementSize = sizeof(uint8_t); + buffer.batchSize = 1; + buffer.batchStride = buffer.sizeInBytes(); + buffer.isDeviceMemory = true; + return buffer; +} + +} // namespace + // Feature: mini-image-pipe, Property 19: Pipeline Topology and Buffer Management // Validates: Requirements 8.1, 8.2 TEST(PipelinePropertyTest, TopologyAndBufferManagement) { @@ -225,8 +243,8 @@ TEST(PipelinePropertyTest, BatchProcessing) { // Allocate batch of input buffers size_t inputSize = width * height * channels; - std::vector inputs(batchSize); - std::vector outputs; + std::vector inputs(batchSize); + std::vector outputs; for (int i = 0; i < batchSize; i++) { uint8_t* h_input = static_cast(mgr.allocatePinned(inputSize)); @@ -238,26 +256,25 @@ TEST(PipelinePropertyTest, BatchProcessing) { } mgr.copyToDeviceAsync(d_input, h_input, inputSize, stream); - inputs[i] = d_input; + inputs[i] = makeBatchFrame(d_input, width, height, channels); mgr.freePinned(h_input); } cudaStreamSynchronize(stream); - // Set input for first frame to establish dimensions - pipeline.setInput(color, inputs[0], width, height, channels); - // Execute batch - cudaError_t err = pipeline.executeBatch(inputs, outputs, width, height, channels); + cudaError_t err = pipeline.executeBatch(inputs, outputs); EXPECT_EQ(err, cudaSuccess); - // Verify we got the right number of outputs - EXPECT_EQ(outputs.size(), batchSize); + ASSERT_EQ(outputs.size(), 1u); + EXPECT_EQ(outputs.front().nodeId, color); + EXPECT_EQ(outputs.front().image.batchSize, batchSize); + EXPECT_EQ(outputs.front().frames.size(), static_cast(batchSize)); // Cleanup - for (void* input : inputs) { - mgr.freeDevice(input); + for (const auto& input : inputs) { + mgr.freeDevice(input.data); } } diff --git a/tests/test_runtime_foundation.cpp b/tests/test_runtime_foundation.cpp index e1198a8..4b3695c 100644 --- a/tests/test_runtime_foundation.cpp +++ b/tests/test_runtime_foundation.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include using namespace mini_image_pipe; @@ -108,6 +109,48 @@ class RuntimeProbeOperator : public IOperator { size_t workspaceBytes_ = 0; }; +class TrackingAllocator : public IRuntimeAllocator { +public: + explicit TrackingAllocator(MemoryManager& inner) : inner_(inner) {} + + void setDeviceAllocatorMode(DeviceAllocatorMode mode) override { + observedModes.push_back(mode); + inner_.setDeviceAllocatorMode(mode); + } + + void* allocateDevice(size_t size, cudaStream_t stream) override { + ++deviceAllocations; + return inner_.allocateDevice(size, stream); + } + + void freeDevice(void* ptr, cudaStream_t stream) override { + ++deviceFrees; + inner_.freeDevice(ptr, stream); + } + + OperatorWorkspace allocateWorkspace(const WorkspaceRequirements& requirements, + cudaStream_t stream) override { + ++workspaceAllocations; + lastWorkspaceRequirements = requirements; + return inner_.allocateWorkspace(requirements, stream); + } + + void freeWorkspace(const OperatorWorkspace& workspace, cudaStream_t stream) override { + ++workspaceFrees; + inner_.freeWorkspace(workspace, stream); + } + + std::vector observedModes; + int deviceAllocations = 0; + int deviceFrees = 0; + int workspaceAllocations = 0; + int workspaceFrees = 0; + WorkspaceRequirements lastWorkspaceRequirements{}; + +private: + MemoryManager& inner_; +}; + } // namespace TEST(RuntimeFoundationTest, MultiInputOperatorReceivesAllResolvedInputs) { @@ -195,6 +238,150 @@ TEST(RuntimeFoundationTest, ProfilingFlagFlowsThroughExecutionContext) { mgr.freeDevice(deviceInput); } +TEST(RuntimeFoundationTest, ExecuteBatchRejectsHostFramesAtInterfaceSeam) { + Pipeline pipeline; + int nodeId = pipeline.addOperator("Identity", std::make_shared()); + (void)nodeId; + + unsigned char hostPixels[16] = {}; + std::vector inputs = {makeImageBuffer(hostPixels, 4, 4, 1)}; + inputs.front().isDeviceMemory = false; + + std::vector outputs; + EXPECT_EQ(pipeline.executeBatch(inputs, outputs), cudaErrorInvalidValue); +} + +TEST(RuntimeFoundationTest, ExecuteBatchRejectsMixedFrameShapes) { + Pipeline pipeline; + int nodeId = pipeline.addOperator("Identity", std::make_shared()); + (void)nodeId; + + std::vector inputs = {makeImageBuffer(reinterpret_cast(0x1), 4, 4, 1), + makeImageBuffer(reinterpret_cast(0x2), 8, 4, 1)}; + + std::vector outputs; + EXPECT_EQ(pipeline.executeBatch(inputs, outputs), cudaErrorInvalidValue); +} + +TEST(RuntimeFoundationTest, ExecuteBatchReturnsAllSinkOutputs) { + REQUIRE_CUDA_DEVICE(); + MemoryManager& mgr = MemoryManager::getInstance(); + Pipeline pipeline; + + int sourceId = pipeline.addOperator("Source", std::make_shared()); + int leftId = pipeline.addOperator("Left", std::make_shared()); + int rightId = pipeline.addOperator("Right", std::make_shared()); + ASSERT_TRUE(pipeline.connect(sourceId, leftId)); + ASSERT_TRUE(pipeline.connect(sourceId, rightId)); + + constexpr int kWidth = 8; + constexpr int kHeight = 8; + constexpr int kChannels = 1; + constexpr int kBatchSize = 2; + constexpr size_t kBytes = kWidth * kHeight * kChannels; + + std::vector inputs; + inputs.reserve(kBatchSize); + for (int i = 0; i < kBatchSize; ++i) { + void* deviceInput = mgr.allocateDevice(kBytes); + ASSERT_NE(deviceInput, nullptr); + inputs.push_back(makeImageBuffer(deviceInput, kWidth, kHeight, kChannels)); + } + + std::vector outputs; + ASSERT_EQ(pipeline.executeBatch(inputs, outputs), cudaSuccess); + ASSERT_EQ(outputs.size(), 2u); + + std::unordered_set nodeIds; + for (const auto& output : outputs) { + nodeIds.insert(output.nodeId); + EXPECT_TRUE(output.image.isValid()); + EXPECT_EQ(output.image.batchSize, kBatchSize); + ASSERT_EQ(output.frames.size(), static_cast(kBatchSize)); + for (void* frame : output.frames) { + EXPECT_NE(frame, nullptr); + } + } + + EXPECT_EQ(nodeIds.count(leftId), 1u); + EXPECT_EQ(nodeIds.count(rightId), 1u); + + for (const auto& input : inputs) { + mgr.freeDevice(input.data); + } +} + +TEST(RuntimeFoundationTest, PipelineConfiguresInjectedAllocatorMode) { + MemoryManager& mgr = MemoryManager::getInstance(); + TrackingAllocator allocator(mgr); + + PipelineConfig config; + config.preferAsyncAllocator = true; + + Pipeline pipeline(config, allocator); + + ASSERT_EQ(allocator.observedModes.size(), 1u); + EXPECT_EQ(allocator.observedModes.front(), DeviceAllocatorMode::ASYNC_STREAM_ORDERED); +} + +TEST(RuntimeFoundationTest, PipelineUsesInjectedAllocatorForRuntimeResources) { + REQUIRE_CUDA_DEVICE(); + MemoryManager& mgr = MemoryManager::getInstance(); + TrackingAllocator allocator(mgr); + auto probe = std::make_shared(1024); + + constexpr int kWidth = 16; + constexpr int kHeight = 16; + constexpr int kChannels = 1; + constexpr size_t kBytes = kWidth * kHeight * kChannels; + + void* deviceInput = mgr.allocateDevice(kBytes); + ASSERT_NE(deviceInput, nullptr); + + { + Pipeline pipeline(PipelineConfig(), allocator); + int probeId = pipeline.addOperator("Probe", probe); + pipeline.setInput(probeId, deviceInput, kWidth, kHeight, kChannels); + ASSERT_EQ(pipeline.execute(), cudaSuccess); + } + + EXPECT_GT(allocator.deviceAllocations, 0); + EXPECT_GT(allocator.workspaceAllocations, 0); + EXPECT_GT(allocator.deviceFrees, 0); + EXPECT_GT(allocator.workspaceFrees, 0); + EXPECT_EQ(allocator.lastWorkspaceRequirements.deviceBytes, 1024u); + + mgr.freeDevice(deviceInput); +} + +TEST(RuntimeFoundationTest, GaussianBlurFallbackUsesInjectedAllocator) { + REQUIRE_CUDA_DEVICE(); + MemoryManager& mgr = MemoryManager::getInstance(); + TrackingAllocator allocator(mgr); + + constexpr int kWidth = 8; + constexpr int kHeight = 8; + constexpr int kChannels = 1; + constexpr size_t kBytes = kWidth * kHeight * kChannels; + + void* deviceInput = mgr.allocateDevice(kBytes); + void* deviceOutput = mgr.allocateDevice(kBytes); + ASSERT_NE(deviceInput, nullptr); + ASSERT_NE(deviceOutput, nullptr); + + { + GaussianBlurOperator op(GaussianKernelSize::KERNEL_3x3, 0.0f, allocator); + ASSERT_EQ(op.execute(deviceInput, deviceOutput, kWidth, kHeight, kChannels, nullptr), + cudaSuccess); + } + + EXPECT_GT(allocator.deviceAllocations, 0); + EXPECT_GT(allocator.deviceFrees, 0); + + mgr.freeDevice(deviceInput); + mgr.freeDevice(deviceOutput); +} + TEST(RuntimeFoundationTest, OperatorLifecycleHooksAreCalledByPipeline) { REQUIRE_CUDA_DEVICE(); MemoryManager& mgr = MemoryManager::getInstance(); diff --git a/tests/test_scheduler.cpp b/tests/test_scheduler.cpp index a49ffee..e484b59 100644 --- a/tests/test_scheduler.cpp +++ b/tests/test_scheduler.cpp @@ -116,10 +116,10 @@ TEST(SchedulerPropertyTest, ErrorPropagation) { EXPECT_TRUE(errorCallbackCalled); // A should complete, B should fail, C and D should be failed due to propagation - EXPECT_EQ(taskA->state.load(), TaskState::COMPLETED); - EXPECT_EQ(taskB->state.load(), TaskState::FAILED); - EXPECT_EQ(taskC->state.load(), TaskState::FAILED); - EXPECT_EQ(taskD->state.load(), TaskState::FAILED); + EXPECT_EQ(taskA->runtime.state.load(), TaskState::COMPLETED); + EXPECT_EQ(taskB->runtime.state.load(), TaskState::FAILED); + EXPECT_EQ(taskC->runtime.state.load(), TaskState::FAILED); + EXPECT_EQ(taskD->runtime.state.load(), TaskState::FAILED); mgr.freeDevice(buffer); } @@ -254,7 +254,7 @@ TEST(SchedulerPropertyTest, StreamSynchronizationOnCompletion) { // After execute returns, all tasks should be completed for (int id : taskIds) { TaskNode* task = graph.getTask(id); - EXPECT_EQ(task->state.load(), TaskState::COMPLETED) + EXPECT_EQ(task->runtime.state.load(), TaskState::COMPLETED) << "Task " << id << " not completed after execute()"; } diff --git a/tests/test_sobel.cpp b/tests/test_sobel.cpp index dd16321..c3f8423 100644 --- a/tests/test_sobel.cpp +++ b/tests/test_sobel.cpp @@ -160,3 +160,15 @@ TEST(SobelPropertyTest, SingleChannelOutput) { cudaStreamDestroy(stream); } + +TEST(SobelPropertyTest, SharedMemoryKernelConfig) { + SobelOperator op; + + KernelConfig config = op.getKernelConfig(63, 35, 3); + + EXPECT_EQ(config.blockSize.x, 16u); + EXPECT_EQ(config.blockSize.y, 16u); + EXPECT_EQ(config.gridSize.x, 4u); + EXPECT_EQ(config.gridSize.y, 3u); + EXPECT_GT(config.sharedMem, 0u); +} diff --git a/tests/test_task_graph.cpp b/tests/test_task_graph.cpp index a2b2f15..f9969d4 100644 --- a/tests/test_task_graph.cpp +++ b/tests/test_task_graph.cpp @@ -141,3 +141,27 @@ TEST(TaskGraphPropertyTest, IndependentTaskDetection) { EXPECT_FALSE(graph.areIndependent(a, c)); } } + +TEST(TaskGraphPropertyTest, ResetClearsRuntimeStateButKeepsTopology) { + TaskGraph graph; + int a = graph.addTask("A", nullptr); + int b = graph.addTask("B", nullptr); + + ASSERT_TRUE(graph.addDependency(a, b)); + + TaskNode* taskA = graph.getTask(a); + TaskNode* taskB = graph.getTask(b); + ASSERT_NE(taskA, nullptr); + ASSERT_NE(taskB, nullptr); + + taskA->runtime.state.store(TaskState::RUNNING); + taskA->runtime.assignedStream = 2; + + graph.reset(); + + EXPECT_EQ(taskA->runtime.state.load(), TaskState::PENDING); + EXPECT_EQ(taskA->runtime.assignedStream, -1); + EXPECT_EQ(taskA->dependencies.size(), 0u); + ASSERT_EQ(taskB->dependencies.size(), 1u); + EXPECT_EQ(taskB->dependencies.front(), a); +} diff --git a/tests/test_throughput_engine.cpp b/tests/test_throughput_engine.cpp index 7676514..721c7e9 100644 --- a/tests/test_throughput_engine.cpp +++ b/tests/test_throughput_engine.cpp @@ -177,6 +177,45 @@ TEST(ThroughputEngineTest, SchedulerCapturesAndInvalidatesStableSingleStreamGrap mgr.freeDevice(output); } +TEST(ThroughputEngineTest, DisablingGraphExecutionClearsCaptureState) { + REQUIRE_CUDA_DEVICE(); + + MemoryManager& mgr = MemoryManager::getInstance(); + DAGScheduler scheduler(1); + scheduler.setGraphExecutionEnabled(true); + + TaskGraph graph; + auto op = std::make_shared(); + int taskId = graph.addTask("Copy", op); + + constexpr size_t kBytes = 64 * 64; + void* input = mgr.allocateDevice(kBytes); + void* output = mgr.allocateDevice(kBytes); + ASSERT_NE(input, nullptr); + ASSERT_NE(output, nullptr); + + TaskNode* task = graph.getTask(taskId); + ASSERT_NE(task, nullptr); + task->inputBuffer = input; + task->outputBuffer = output; + task->width = 64; + task->height = 64; + task->channels = 1; + + ASSERT_EQ(scheduler.execute(graph), cudaSuccess); + ASSERT_EQ(scheduler.execute(graph), cudaSuccess); + ASSERT_TRUE(scheduler.hasCapturedGraph()); + ASSERT_TRUE(scheduler.didReplayLastGraph()); + + scheduler.setGraphExecutionEnabled(false); + + EXPECT_FALSE(scheduler.hasCapturedGraph()); + EXPECT_FALSE(scheduler.didReplayLastGraph()); + + mgr.freeDevice(input); + mgr.freeDevice(output); +} + TEST(ThroughputEngineTest, SchedulerRecordsProfilingForEnabledTasks) { DAGScheduler scheduler(1); TaskGraph graph; @@ -192,7 +231,7 @@ TEST(ThroughputEngineTest, SchedulerRecordsProfilingForEnabledTasks) { task->width = 1; task->height = 1; task->channels = 1; - task->profilingEnabled = true; + task->runtime.profilingEnabled = true; EXPECT_EQ(scheduler.execute(graph), cudaSuccess); EXPECT_TRUE(probe->sawProfilingFlag); @@ -253,20 +292,25 @@ TEST(ThroughputEngineTest, BatchExecutionInvokesNodeOncePerBatchContext) { constexpr int kChannels = 1; constexpr size_t kBytes = kWidth * kHeight * kChannels; - std::vector inputs(3, nullptr); - std::vector outputs; - for (void*& input : inputs) { - input = mgr.allocateDevice(kBytes); + std::vector inputs; + inputs.reserve(3); + std::vector outputs; + for (int i = 0; i < 3; ++i) { + void* input = mgr.allocateDevice(kBytes); ASSERT_NE(input, nullptr); + inputs.push_back({input, kWidth, kHeight, kChannels, kWidth * kChannels, sizeof(uint8_t), 1, + kBytes, true, false}); } - EXPECT_EQ(pipeline.executeBatch(inputs, outputs, kWidth, kHeight, kChannels), cudaSuccess); + EXPECT_EQ(pipeline.executeBatch(inputs, outputs), cudaSuccess); EXPECT_EQ(probe->executeBuffersCount, 1); EXPECT_EQ(probe->lastInputBatchSize, 3); EXPECT_EQ(probe->lastOutputBatchSize, 3); - ASSERT_EQ(outputs.size(), inputs.size()); + ASSERT_EQ(outputs.size(), 1u); + EXPECT_EQ(outputs.front().nodeId, probeId); + ASSERT_EQ(outputs.front().frames.size(), inputs.size()); - for (void* input : inputs) { - mgr.freeDevice(input); + for (const auto& input : inputs) { + mgr.freeDevice(input.data); } }