説明
本文以 OpenCV 中的圖像翻轉 cv::flip 為切入點，選取內部 ippicv 提供的圖像翻轉函數 ippiMirror（具體為 ippiMirror_8u_C1R）進行復現，以此加深對 SIMD（SSE）指令應用的理解。
ippicv（IPP for Computer Vision）是 Intel Integrated Performance Primitives（Intel IPP）中專門面向圖像處理和計算機視覺場景的一個裁剪子集，由 Intel 官方以預編譯第三方庫的形式集成到 OpenCV 中，用來為部分核心算子提供基於 SSE/AVX 等指令集的高性能實現。它通常以靜態庫和頭文件的形式出現，在開啓 WITH_IPP=ON 構建選項後，會自動被下載並鏈接到 OpenCV 中，為常見圖像操作（卷積、插值、幾何變換等）實現更好的性能。
當前復現代碼支持水平翻轉或水平+垂直翻轉。在有了這個版本後，復現完整的ippiMirror_8u_C1R也就只是時間問題。
關鍵詞
OpenCV；SIMD ；圖像處理；性能優化；圖像翻轉；流式存儲；非臨時存儲；內存帶寬瓶頸；內存受限算法；ippicv；ippiMirror；cv::flip
Keywords: OpenCV; SIMD; image processing; performance optimization; image flipping; streaming store; non-temporal store; memory bandwidth bottleneck; memory-bound algorithm; ippicv; ippiMirror; cv::flip
復現
#include <tmmintrin.h>
#include <smmintrin.h>
#include <immintrin.h>
#include <stdint.h>
#include <stddef.h>
#include <string.h>
#ifdef _MSC_VER
#define FORCE_INLINE __forceinline
#else
#define FORCE_INLINE inline __attribute__((always_inline))
#endif
#ifdef _MSC_VER
#include <intrin.h>
static void cpuid(int info[4], int func_id) { __cpuid(info, func_id); }
#else
static void cpuid(int info[4], int func_id) { __asm__ __volatile__("cpuid":"=a"(info[0]),"=b"(info[1]),"=c"(info[2]),"=d"(info[3]):"a"(func_id)); }
#endif
// 返回：0=無SSSE3，1=SSSE3，2=AVX2
int detect_simd_level(void){
int info[4] = {0};
cpuid(info, 0);
if (info[0] < 7) {
cpuid(info, 1);
int ssse3 = (info[2] >> 9) & 1; // ECX bit 9
return ssse3 ? 1 : 0;
} else {
cpuid(info, 1);
int ssse3 = (info[2] >> 9) & 1;
cpuid(info, 7);
int avx2 = (info[1] >> 5) & 1; // EBX bit 5
return avx2 ? 2 : (ssse3 ? 1 : 0);
}
}
// -------------------------------------------------------------------------
// 圖像翻轉
// -------------------------------------------------------------------------
// 16字節翻轉掩碼: [15, 14, ... 0]
static const __m128i mask_flip_128 = _mm_setr_epi8(
15, 14, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1, 0
);
// 8字節翻轉掩碼: 低8字節翻轉 [7...0]，高8字節設為 -1 (保持不變或清零，取決於具體指令)
// 在 SSE shuffle 中，高位 1 (0x80) 會將結果置零。
// 這裏我們只關心低64位的結果，高位會被 _mm_storel_epi64 / cast 忽略。
static const __m128i mask_flip_64 = _mm_setr_epi8(
7, 6, 5, 4, 3, 2, 1, 0, -1, -1, -1, -1, -1, -1, -1, -1
);
// Switch-Jump Table 處理尾部 1-7 字節
FORCE_INLINE void copy_tail_reversed_switch(const uint8_t* src, uint8_t* dst_end_ptr, int n) {
// src: 指向剩餘數據的開頭
// dst_end_ptr: 指向剩餘數據在目標內存中的尾部+1位置
// 寫入方向：從後向前。 dst_end_ptr[-1] = src[0]
switch (n) {
case 7: dst_end_ptr[-7] = src[6]; [[fallthrough]];
case 6: dst_end_ptr[-6] = src[5]; [[fallthrough]];
case 5: dst_end_ptr[-5] = src[4]; [[fallthrough]];
case 4: dst_end_ptr[-4] = src[3]; [[fallthrough]];
case 3: dst_end_ptr[-3] = src[2]; [[fallthrough]];
case 2: dst_end_ptr[-2] = src[1]; [[fallthrough]];
case 1: dst_end_ptr[-1] = src[0];
}
}
/**
* @brief 處理翻轉操作的尾部數據 (16/8/1-7 字節)
* @param srow [in/out] 源數據當前位置指針(會被修改)
* @param drow_end [in/out] 目標數據末尾位置指針(會被修改)
* @param w [in/out] 剩餘寬度(會被修改)
* @param mask_flip_128 [in] 128位翻轉掩碼
* @param mask_flip_64 [in] 64位翻轉掩碼
*/
FORCE_INLINE void process_tail_reversed(
const uint8_t*& srow,
uint8_t*& drow_end,
int& w,
const __m128i& mask_flip_128,
const __m128i& mask_flip_64)
{
// 16字節處理
if (w >= 16) {
__m128i v0 = _mm_loadu_si128((const __m128i*)(srow));
_mm_storeu_si128((__m128i*)(drow_end - 16), _mm_shuffle_epi8(v0, mask_flip_128));
srow += 16;
drow_end -= 16;
w -= 16;
}
// 8字節處理
if (w >= 8) {
__m128i v0 = _mm_loadl_epi64((const __m128i*)srow);
v0 = _mm_shuffle_epi8(v0, mask_flip_64);
*(int64_t*)(drow_end - 8) = _mm_cvtsi128_si64(v0);
srow += 8;
drow_end -= 8;
w -= 8;
}
// 1-7 字節尾部處理
if (w > 0) {
copy_tail_reversed_switch(srow, drow_end, w);
}
}
/**
* @brief 8位單通道圖像翻轉
* @param pSrc [in] 源圖像的起始地址 (左上角).
* @param srcStep [in] 源圖像的行步長 (字節).
* @param pDst [out] 目標圖像的起始地址.
* @param dstStep [in] 目標圖像的行步長 (字節).
* @param width [in] 寬度 (像素數).
* @param height [in] 高度 (像素數).
* @param reverseRows [in] 垂直翻轉標誌位.
* - 0: 僅水平鏡像 (行序不變).
* - 1: 水平鏡像 + 垂直翻轉 (即旋轉 180 度).
* @return int64_t 返回實際使用的目標行步長 (dst_stride_actual).
* - 如果 reverseRows=0, 返回 dstStep.
* - 如果 reverseRows=1, 返回 -dstStep (用於指示反向遍歷).
*/
int64_t ippiMirror_8u_C1_simd(
const uint8_t *pSrc,
int srcStep,
uint8_t *pDst,
int dstStep,
int width,
int height,
int reverseRows)
{
// 初始化步長和起始位置
int64_t dst_stride_actual = reverseRows ? -dstStep : dstStep;
int64_t dst_start_offset = reverseRows ? (int64_t)dstStep * (height - 1) : 0;
// 這裏的 dst_ptr_base 指向要處理的第一行的"起始"位置
uint8_t* dst_ptr_base = pDst + dst_start_offset;
// ---------------------------------------------------------------------
// 檢查 1: 對齊檢查
//
// 因為使用了反向寫入 (stream store backwards)，只要寫入的"起始點"(即行尾)是對齊的，
// 那麼 ptr-16, ptr-32 就都是對齊的。
//
// 這樣即使 pDst (行首) 不對齊，只要 pDst + width 對齊，也可以走優化路徑。
// 剩餘的不對齊部分會落入最後的 switch case 處理，這是安全的。
// ---------------------------------------------------------------------
// 計算當前行邏輯上的"末尾+1"地址 (也是反向寫入的起始地址)
uint8_t* const dst_row_end_addr = dst_ptr_base + width;
// 16字節對齊
bool is_aligned = (((uintptr_t)pSrc | (uintptr_t)srcStep | (uintptr_t)dst_row_end_addr | (uintptr_t)dst_stride_actual) & 0xF) == 0;
if (!is_aligned)
{
// 分支 1: 通用/未對齊路徑
// 可以在這裏根據 src/dst 對齊情況做更細緻的 if/else，
// 但核心都是 load/shuffle/store。
// 這裏使用 _mm_loadu_si128 (unaligned) 統一處理，這在現代 CPU 上性能幾乎等同於拆分處理。
const uint8_t* sptr = pSrc;
uint8_t* dptr_end = dst_row_end_addr; // 指向當前行尾+1
for (int i = 0; i < height; ++i) {
int w = width;
const uint8_t* srow = sptr;
uint8_t* drow_end = dptr_end;
// 32字節循環 (Loop Unrolling x2)
while (w >= 32) {
__m128i v0 = _mm_loadu_si128((const __m128i*)(srow));
__m128i v1 = _mm_loadu_si128((const __m128i*)(srow + 16));
v0 = _mm_shuffle_epi8(v0, mask_flip_128);
v1 = _mm_shuffle_epi8(v1, mask_flip_128);
// 注意：drow_end 是向後退的
_mm_storeu_si128((__m128i*)(drow_end - 16), v0);
_mm_storeu_si128((__m128i*)(drow_end - 32), v1);
srow += 32;
drow_end -= 32;
w -= 32;
}
// 尾部處理
process_tail_reversed(srow, drow_end, w, mask_flip_128, mask_flip_64);
sptr += srcStep;
dptr_end += dst_stride_actual; // 移動到下一行
}
return dst_stride_actual;
}
// ---------------------------------------------------------------------
// 檢查 2: 大數據量 Streaming 檢查
// ---------------------------------------------------------------------
int64_t total_size = (int64_t)(srcStep + dstStep) * height;
if (total_size > 0x100000)
{
// 分支 2: 流式寫入路徑 (Streaming Stores)
const uint8_t* sptr = pSrc;
uint8_t* dptr_end = dst_row_end_addr;
for (int i = 0; i < height; ++i) {
int w = width;
const uint8_t* srow = sptr;
uint8_t* drow_end = dptr_end;
// 128字節循環 (8x16)
while (w >= 128) {
// 流水線加載
__m128i v0 = _mm_load_si128((const __m128i*)(srow + 0));
__m128i v1 = _mm_load_si128((const __m128i*)(srow + 16));
__m128i v2 = _mm_load_si128((const __m128i*)(srow + 32));
__m128i v3 = _mm_load_si128((const __m128i*)(srow + 48));
__m128i v4 = _mm_load_si128((const __m128i*)(srow + 64));
__m128i v5 = _mm_load_si128((const __m128i*)(srow + 80));
__m128i v6 = _mm_load_si128((const __m128i*)(srow + 96));
__m128i v7 = _mm_load_si128((const __m128i*)(srow + 112));
// 翻轉
v0 = _mm_shuffle_epi8(v0, mask_flip_128);
v1 = _mm_shuffle_epi8(v1, mask_flip_128);
v2 = _mm_shuffle_epi8(v2, mask_flip_128);
v3 = _mm_shuffle_epi8(v3, mask_flip_128);
v4 = _mm_shuffle_epi8(v4, mask_flip_128);
v5 = _mm_shuffle_epi8(v5, mask_flip_128);
v6 = _mm_shuffle_epi8(v6, mask_flip_128);
v7 = _mm_shuffle_epi8(v7, mask_flip_128);
// 流式寫入 (繞過緩存)
_mm_stream_si128((__m128i*)(drow_end - 16), v0);
_mm_stream_si128((__m128i*)(drow_end - 32), v1);
_mm_stream_si128((__m128i*)(drow_end - 48), v2);
_mm_stream_si128((__m128i*)(drow_end - 64), v3);
_mm_stream_si128((__m128i*)(drow_end - 80), v4);
_mm_stream_si128((__m128i*)(drow_end - 96), v5);
_mm_stream_si128((__m128i*)(drow_end - 112), v6);
_mm_stream_si128((__m128i*)(drow_end - 128), v7);
srow += 128;
drow_end -= 128;
w -= 128;
}
// 32字節循環
while (w >= 32) {
__m128i v0 = _mm_load_si128((const __m128i*)(srow));
__m128i v1 = _mm_load_si128((const __m128i*)(srow + 16));
v0 = _mm_shuffle_epi8(v0, mask_flip_128);
v1 = _mm_shuffle_epi8(v1, mask_flip_128);
_mm_stream_si128((__m128i*)(drow_end - 16), v0);
_mm_stream_si128((__m128i*)(drow_end - 32), v1);
srow += 32;
drow_end -= 32;
w -= 32;
}
// 尾部處理
process_tail_reversed(srow, drow_end, w, mask_flip_128, mask_flip_64);
sptr += srcStep;
dptr_end += dst_stride_actual;
}
_mm_sfence(); // 保證 stream store 完成
return dst_stride_actual;
}
// ---------------------------------------------------------------------
// 分支 3: 已對齊且小數據量 (Aligned Cache-Friendly)
// ---------------------------------------------------------------------
if (width > 31)
{
const uint8_t* sptr = pSrc;
uint8_t* dptr_end = dst_row_end_addr;
for (int i = 0; i < height; ++i) {
int w = width;
const uint8_t* srow = sptr;
uint8_t* drow_end = dptr_end;
// 96 字節循環 (3x32)
while (w >= 96) {
// Load 32 * 3
__m128i v0 = _mm_load_si128((const __m128i*)(srow));
__m128i v1 = _mm_load_si128((const __m128i*)(srow + 16));
__m128i v2 = _mm_load_si128((const __m128i*)(srow + 32));
__m128i v3 = _mm_load_si128((const __m128i*)(srow + 48));
__m128i v4 = _mm_load_si128((const __m128i*)(srow + 64));
__m128i v5 = _mm_load_si128((const __m128i*)(srow + 80));
// Store 32 * 3
_mm_store_si128((__m128i*)(drow_end - 16), _mm_shuffle_epi8(v0, mask_flip_128));
_mm_store_si128((__m128i*)(drow_end - 32), _mm_shuffle_epi8(v1, mask_flip_128));
_mm_store_si128((__m128i*)(drow_end - 48), _mm_shuffle_epi8(v2, mask_flip_128));
_mm_store_si128((__m128i*)(drow_end - 64), _mm_shuffle_epi8(v3, mask_flip_128));
_mm_store_si128((__m128i*)(drow_end - 80), _mm_shuffle_epi8(v4, mask_flip_128));
_mm_store_si128((__m128i*)(drow_end - 96), _mm_shuffle_epi8(v5, mask_flip_128));
srow += 96;
drow_end -= 96;
w -= 96;
}
// 32 字節循環
while (w >= 32) {
__m128i v0 = _mm_load_si128((const __m128i*)(srow));
__m128i v1 = _mm_load_si128((const __m128i*)(srow + 16));
_mm_store_si128((__m128i*)(drow_end - 16), _mm_shuffle_epi8(v0, mask_flip_128));
_mm_store_si128((__m128i*)(drow_end - 32), _mm_shuffle_epi8(v1, mask_flip_128));
srow += 32; drow_end -= 32; w -= 32;
}
// 尾部處理
process_tail_reversed(srow, drow_end, w, mask_flip_128, mask_flip_64);
sptr += srcStep;
dptr_end += dst_stride_actual;
}
return dst_stride_actual;
}
// 最後的 fallback (如果 aligned 但 width 很小 < 32)
{
const uint8_t* sptr = pSrc;
uint8_t* dptr_end = dst_row_end_addr;
for (int i = 0; i < height; ++i) {
int w = width;
const uint8_t* srow = sptr;
uint8_t* drow_end = dptr_end;
// 尾部處理
process_tail_reversed(srow, drow_end, w, mask_flip_128, mask_flip_64);
sptr += srcStep;
dptr_end += dst_stride_actual;
}
}
return dst_stride_actual;
}功能測試和性能測試
功能測試代碼如下
#include <memory>
#include <vector>
#include <opencv2/opencv.hpp>
#include "ippiflip.h"
// 輔助函數
static bool equalMat(const cv::Mat& a, const cv::Mat& b) {
cv::Mat diff;
cv::absdiff(a, b, diff);
return cv::countNonZero(diff) == 0;
}
// 核心測試函數 - 支持src和dst都可以有padding
static bool run_case(int rows, int cols,
int src_pad_bytes, int dst_pad_bytes,
bool hv,
const char* desc = nullptr) {
// 創建帶padding的源圖像
cv::Mat src_full(rows, cols + src_pad_bytes, CV_8UC1);
cv::RNG rng(12345u + rows * 131 + cols * 17 + src_pad_bytes * 7 + dst_pad_bytes * 3);
rng.fill(src_full, cv::RNG::UNIFORM, 0, 256);
cv::Mat src = src_full(cv::Rect(0, 0, cols, rows));
// 創建帶padding的目標圖像
cv::Mat dst_full(rows, cols + dst_pad_bytes, CV_8UC1, cv::Scalar::all(0));
cv::Mat dst = dst_full(cv::Rect(0, 0, cols, rows));
// OpenCV參考實現
cv::Mat cvref;
const int reverseRows = hv ? 1 : 0;
if (hv)
cv::flip(src, cvref, -1); // H+V
else
cv::flip(src, cvref, 1); // 僅H
// 調用SIMD實現
int64_t usedStep = ippiMirror_8u_C1_simd(
src.data, static_cast<int>(src.step),
dst.data, static_cast<int>(dst.step),
cols, rows, reverseRows);
(void)usedStep;
// 驗證結果
if (!equalMat(dst, cvref)) {
fprintf(stderr, "FAILED: %s\n", desc ? desc : "");
fprintf(stderr, " Size: %dx%d, src_pad=%d, dst_pad=%d, mode=%s\n",
rows, cols, src_pad_bytes, dst_pad_bytes, hv ? "H+V" : "H");
return false;
}
return true;
}
// 測試1: 分辨率測試
static int test_typical_resolutions() {
printf("\n=== Test Suite 1: Typical Resolutions ===\n");
struct Case {
int r, c, src_pad, dst_pad;
const char* desc;
};
std::vector<Case> cases = {
{1080, 1920, 0, 0, "1080p no padding"},
{1080, 1920, 7, 0, "1080p src padded"},
{1080, 1920, 0, 11, "1080p dst padded"},
{1080, 1920, 7, 11, "1080p both padded"},
{480, 641, 3, 5, "480p odd width, both padded"},
{720, 1281, 11, 13, "720p odd width, both padded"},
};
int passed = 0, total = 0;
for (const auto& cs : cases) {
// 測試僅H翻轉
total++;
if (run_case(cs.r, cs.c, cs.src_pad, cs.dst_pad, false, cs.desc)) {
passed++;
}
// 測試H+V翻轉
total++;
if (run_case(cs.r, cs.c, cs.src_pad, cs.dst_pad, true, cs.desc)) {
passed++;
}
}
printf("Result: %d/%d passed\n", passed, total);
return passed == total ? 0 : 1;
}
// 測試2: 小尺寸窮舉測試 (width <= 64)
static int test_small_sizes() {
printf("\n=== Test Suite 2: Small Size Exhaustive (w<=64, h<=8) ===\n");
int passed = 0, total = 0;
// 測試小寬度 1-64, 小高度 1-8
for (int h = 1; h <= 8; ++h) {
for (int w = 1; w <= 64; ++w) {
// 每種尺寸測試幾種padding組合
for (int pad_combo = 0; pad_combo < 3; ++pad_combo) {
int src_pad = (pad_combo == 0) ? 0 : (w % 7 + 1);
int dst_pad = (pad_combo == 1) ? 0 : (w % 5 + 1);
// 測試H翻轉
total++;
char desc[128];
snprintf(desc, sizeof(desc), "Small %dx%d", h, w);
if (run_case(h, w, src_pad, dst_pad, false, desc)) {
passed++;
}
// 測試H+V翻轉
total++;
if (run_case(h, w, src_pad, dst_pad, true, desc)) {
passed++;
}
}
}
}
printf("Result: %d/%d passed\n", passed, total);
return passed == total ? 0 : 1;
}
// 測試3: 對齊邊界測試 (測試關鍵寬度邊界)
static int test_alignment_boundaries() {
printf("\n=== Test Suite 3: Alignment Boundaries ===\n");
int passed = 0, total = 0;
// 關鍵邊界: 16, 31, 32, 63, 64, 127, 128等SIMD相關邊界
std::vector<int> critical_widths = {
15, 16, 17, // 16字節邊界
31, 32, 33, // 32字節邊界
63, 64, 65, // 64字節邊界
127, 128, 129, // 128字節邊界
255, 256, 257, // 256字節邊界
};
std::vector<int> heights = {1, 2, 7, 8, 15, 16, 100};
for (int w : critical_widths) {
for (int h : heights) {
// 測試不同padding組合
for (int src_pad = 0; src_pad <= 3; src_pad += 3) {
for (int dst_pad = 0; dst_pad <= 5; dst_pad += 5) {
total++;
char desc[128];
snprintf(desc, sizeof(desc), "Boundary %dx%d", h, w);
if (run_case(h, w, src_pad, dst_pad, false, desc)) {
passed++;
}
total++;
if (run_case(h, w, src_pad, dst_pad, true, desc)) {
passed++;
}
}
}
}
}
printf("Result: %d/%d passed\n", passed, total);
return passed == total ? 0 : 1;
}
// 測試4: 大尺寸測試 (包含4K)
static int test_large_sizes() {
printf("\n=== Test Suite 4: Large Sizes (Including 4K) ===\n");
struct LargeCase {
int r, c;
const char* desc;
};
std::vector<LargeCase> cases = {
{1080, 1920, "1080p (Full HD)"},
{1440, 2560, "1440p (2K)"},
{2160, 3840, "2160p (4K)"},
};
int passed = 0, total = 0;
for (const auto& cs : cases) {
cv::Mat big(cs.r, cs.c, CV_8UC1);
cv::randu(big, 0, 256);
// 測試H翻轉
{
cv::Mat out_icv(cs.r, cs.c, CV_8UC1);
cv::Mat out_cv;
ippiMirror_8u_C1_simd(
big.data, static_cast<int>(big.step),
out_icv.data, static_cast<int>(out_icv.step),
cs.c, cs.r, 0);
cv::flip(big, out_cv, 1);
total++;
if (equalMat(out_icv, out_cv)) {
passed++;
} else {
fprintf(stderr, "FAILED: %s H-flip\n", cs.desc);
}
}
// 測試H+V翻轉
{
cv::Mat out_icv(cs.r, cs.c, CV_8UC1);
cv::Mat out_cv;
ippiMirror_8u_C1_simd(
big.data, static_cast<int>(big.step),
out_icv.data, static_cast<int>(out_icv.step),
cs.c, cs.r, 1);
cv::flip(big, out_cv, -1);
total++;
if (equalMat(out_icv, out_cv)) {
passed++;
} else {
fprintf(stderr, "FAILED: %s H+V-flip\n", cs.desc);
}
}
}
printf("Result: %d/%d passed\n", passed, total);
return passed == total ? 0 : 1;
}
// 測試5: 隨機壓力測試
static int test_random_stress() {
printf("\n=== Test Suite 5: Random Stress Test ===\n");
int passed = 0, total = 0;
const int num_tests = 100;
cv::RNG rng(0xDEADBEEF);
for (int i = 0; i < num_tests; ++i) {
// 隨機生成測試參數
int w = rng.uniform(1, 2048);
int h = rng.uniform(1, 1024);
int src_pad = rng.uniform(0, 32);
int dst_pad = rng.uniform(0, 32);
bool hv = (rng.uniform(0, 2) == 1);
total++;
char desc[128];
snprintf(desc, sizeof(desc), "Random #%d: %dx%d", i, h, w);
if (run_case(h, w, src_pad, dst_pad, hv, desc)) {
passed++;
}
}
printf("Result: %d/%d passed\n", passed, total);
return passed == total ? 0 : 1;
}
int main() {
printf("SIMD Level: %d (0=none, 1=SSSE3, 2=AVX2)\n", detect_simd_level());
int failed_suites = 0;
// 運行所有測試
failed_suites += test_typical_resolutions();
failed_suites += test_small_sizes();
failed_suites += test_alignment_boundaries();
failed_suites += test_large_sizes();
failed_suites += test_random_stress();
// 總結
printf("\n");
if (failed_suites == 0) {
printf("ALL TESTS PASSED\n");
} else {
printf("%d TEST SUITE(S) FAILED\n", failed_suites);
}
return failed_suites == 0 ? 0 : 1;
}功能測試全部通過
性能測試部分代碼如下
static void BM_ICV_1024_Horizontal(benchmark::State& state) {
cv::Mat src = createBenchmarkImage(1024);
cv::Mat dst = src.clone();
for (auto _ : state) {
ippiMirror_8u_C1_simd(
src.data,
static_cast<int>(src.step),
dst.data,
static_cast<int>(dst.step),
src.cols,
src.rows,
0 // reverseRows=0: 僅水平翻轉
);
benchmark::DoNotOptimize(dst.data);
}
state.SetBytesProcessed(static_cast<int64_t>(state.iterations()) *
src.total() * src.elemSize());
}
static void BM_OpenCV_1024_Horizontal(benchmark::State& state) {
cv::Mat src = createBenchmarkImage(1024);
cv::Mat dst;
for (auto _ : state) {
cv::flip(src, dst, 1); // flipCode=1: 水平翻轉
benchmark::DoNotOptimize(dst.data);
}
state.SetBytesProcessed(static_cast<int64_t>(state.iterations()) *
src.total() * src.elemSize());
}性能測試結果如下
BM_ICV_1024_Horizontal 28117 ns 27832 ns 23579 bytes_per_second=35.0878Gi/s
BM_OpenCV_1024_Horizontal 28629 ns 29157 ns 23579 bytes_per_second=33.4929Gi/s
BM_ICV_1024_Both 30761 ns 30762 ns 24889 bytes_per_second=31.7462Gi/s
BM_OpenCV_1024_Both 29216 ns 29157 ns 23579 bytes_per_second=33.4929Gi/s
BM_ICV_2048_Horizontal 129590 ns 131138 ns 5600 bytes_per_second=29.7872Gi/s
BM_OpenCV_2048_Horizontal 127283 ns 125558 ns 5600 bytes_per_second=31.1111Gi/s
BM_ICV_2048_Both 127280 ns 128348 ns 5600 bytes_per_second=30.4348Gi/s
BM_OpenCV_2048_Both 131635 ns 131138 ns 5600 bytes_per_second=29.7872Gi/s這裏挑選了一個測試結果。另外在正式執行BM_ICV_1024_Horizontal之前先進行了BM_OpenCV_1024_Horizontal熱身，因為發現第一項測試總是隻有20+Gi/s。
實際性能測試中互有勝負，且差距很小。由此可見，可以認為成功復現了該算法。
總結
通過對 ippicv 中 owniFlipCopy_8u_C1 函數的復現，我們深入探索了 SIMD 在圖像處理優化方面的技術實踐。
真正的高性能代碼需要綜合考慮 CPU 架構、緩存層次、內存帶寬、指令延遲等多個因素。
Streaming Stores 流式存儲
主要作用如下：
- 避免 RFO，減少無意義的目的緩衝區讀取，直接省帶寬
在之前自己實現過一版圖像翻轉SSE，但發現被內存帶寬所限制。這次復現後，對這個問題也有了更清晰的瞭解。
對於圖像翻轉這種操作，計算密度極低（只做簡單的 shuffle），內存讀寫密度極高。CPU 的運算速度遠快於內存供數速度，因此它是一個典型的 Memory-Bound（內存受限） 算法。
內存帶寬瓶頸正是代碼中引入 Streaming Stores（流式存儲）的最主要原因，在這裏，減少數據訪存延時至關重要。
在普通的內存寫入中，CPU 遵循 RFO (Read-For-Ownership) ，可以簡單概括為：
CPU 想要寫一個 cache line。
CPU 先從內存把這塊舊數據讀進緩存（佔用帶寬）（實際可以讀其他緩存，但這裏重點是內存）。
修改緩存中的數據。
稍後將髒數據寫回內存。而 Streaming Store (Non-temporal Store)：
CPU 告訴內存控制器：“我要覆蓋這整塊數據，別管舊數據了”。
CPU 跳過讀取步驟，直接將 Write Combining Buffer 中的數據刷入內存。省掉了一次無效的內存讀取帶寬。
要真正避免 RFO、發揮帶寬優勢，通常需要滿足：
- 按 cache line 對齊（比如 64B 對齊）；
- 寫入模式是密集連續的，讓 write-combining buffer 可以湊滿一整行再寫出；
- 不要混雜頻繁訪問同一行的小寫/讀，不然硬件可能仍然需要獲取這行數據。
- 減少緩存污染，把 cache 留給真正需要複用的數據
普通 store 會把寫入的數據放進 L1/L2 cache，但對這段代碼來説，dst 寫完就行了，後面不會再讀同一塊 dst
再次讀取一般是後面的處理 pipeline，而不是這一次函數調用
- 利用寫合併緩衝區，讓連續的大塊寫更高效，提升內存吞吐
- 多次 _mm_stream_si128（16B）寫向同一 cache line 區域
- 硬件在 WC buffer 裏先合成完整/半完整的 cache line
- 再一次性寫出
最終減少總線事務數量、提高有效帶寬
特點
反向寫入設計：行尾向行首寫入，只要行尾對齊，就能保證連續的 SIMD 寫入是對齊的，從而巧妙規避了目標首地址（pDst）不對齊帶來的性能損耗，最大化了 Aligned Store 指令的覆蓋率。
核心算子優化：利用 SSSE3 的 _mm_shuffle_epi8 替代位運算，實現寄存器內字節亂序/翻轉。
指令級並行：通過 128/32 字節的多級循環展開（Loop Unrolling）和批量加載，掩蓋指令延遲，充分利用 CPU 流水線性能。
流式寫入：針對大數據量場景，使用 _mm_stream_si128 繞過緩存直接寫入內存，減少了數據訪存延時，提升了內存帶寬利用率，顯著提升吞吐量。
Switch-Fallthrough 模式：利用 Switch-Case 的穿透特性（Fallthrough）處理 1~7 字節的尾部數據，避免了複雜的條件判斷分支。
本文章為轉載內容，我們尊重原作者對文章享有的著作權。如有內容錯誤或侵權問題，歡迎原作者聯繫我們進行內容更正或刪除文章。