[Python] C Library 이용해서 성능 높이기(SIMD + 병렬처리, 3편)

Teus 24.10.22

1,623 4 0

facebook twitter kakao link

4 0

facebook twitter kakao link

DEVOTEE 요약: 이 포스팅은 C++의 Thread, OpenMP, SIMD를 사용하여 파이썬에서의 연산 성능을 높이는 방법을 다룹니다. Thread는 malloc을 통해 할당 받은 메모리를 활용하여 각 쓰레드에서 데이터 처리, OpenMP는 병렬 반복문을 활용하여 효율적으로 작업을 분할, SIMD는 256비트 연산을 통해 데이터 처리를 빠르게 합니다. 다양한 방법을 비교한 결과, 멀티 코어 활용시 성능 개선이 됐지만, SIMD 등의 최적화 효과는 예상보다 크지 않았습니다.

DEVOTEE 추천 블로그

안녕하세요 Teus입니다.

이번 포스팅은 ［Python］ C Library 이용해서 성능 높이기(Ctypes+Numpy, 2편) 의 후속 포스팅으로 찾아왔습니다!

(2년만에 후속포스팅 내는 못난놈 죄송합니다🥺)

지난번 포스팅에서 Ctypes와 Numpy + SIMD, openMP를 다 써봅시다! 라고 했었는데요

해당 내용을 이번 포스팅에서 다뤄볼 예정입니다!

이번 포스팅에서는.

c++의 thread를 사용해서 gil-free 멀티쓰레딩하기
SIMD intrinsic을 사용해서 Vector Data처리하기
OpenMP를 사용해서 Embrassingly parallel 처리해보기

를 다뤄볼 생각 입니다.

1. C++의 Thread

C에서 Thread를 사용할 경우

pthread 혹은 C++의 Thread모듈을 사용하게 됩니다.

#include <thread>

다행히도 Python에서 사용하는 Thread와 유사한 사용 패턴을 제공합니다.

Thread를 만들면서 해당 Thread에서 실행할 함수를 제공하고

다시 Thread를 Join시켜서 분리된 작업을 마무리 시킵니다.

아래 코드를 보시죠!

#include <iostream>
#include <thread>
#include <vector>
#include <stdio.h>
int thread_cnt = 8;

void func(int thread_id) {	
	printf("this thread num : %d\n", thread_id);
}

int main() {	
	std::vector<std::thread> thread_list;
	for (int i = 0; i < thread_cnt; i++) {
		thread_list.push_back(std::thread(func, i));
	}

	for (int i = 0; i < thread_cnt; i++) {
		thread_list[i].join();
	}
}

이때 malloc을 이용해서 memory를 확보하고, 이 메모리 주소 기반으로 Thread마다 나눠서 데이터를 처리하는것이 가능합니다.

#include <iostream>
#include <thread>
#include <vector>
#include <stdio.h>
#include<stdlib.h>
int data_num = 1000000;
int thread_cnt = 8;
void func(int thread_id, long long* arr1_address, long long* arr2_address, long long* ret_address) {
	printf("this thread num : %d\n", thread_id);
	for (int i = data_num * thread_id; i < data_num * (thread_id + 1); i++) {
		ret_address[i] = arr1_address[i] + arr2_address[i];
	}
}

int main() {
	long long* my_arr1 = (long long*)malloc(sizeof(long long) * thread_cnt * data_num);
	long long* my_arr2 = (long long*)malloc(sizeof(long long) * thread_cnt * data_num);
	long long* ret_arr = (long long*)malloc(sizeof(long long) * thread_cnt * data_num);
	for (int i = 0; i < thread_cnt * data_num; i++) {
		my_arr1[i] = i;
		my_arr2[i] = i;
	}
	std::vector<std::thread> thread_list;
	for (int i = 0; i < thread_cnt; i++) {
		thread_list.push_back(std::thread(func, i, my_arr1, my_arr2, ret_arr));
	}

	for (int i = 0; i < thread_cnt; i++) {
		thread_list[i].join();
	}
	for (int i = 0; i < 100; i++) {
		printf("%d\n", ret_arr[i]);
	}
}

2. C++의 OMP

openmp는 예전에도 잠깐 나왔었지만

C++에서 컴파일러 한태 자동으로 해당 구문을 병렬처리하라 라고 알려주는 기능 입니다.

<omp.h>를 통해서 활용이 가능하며

최신 gcc나 msvc컴파일러에서 바로 활용이 가능합니다.

여러가지 활용 방법이 있지만

이번 포스팅에서는 단순하게 embrassingly parallel을 위한 parallel for loop만 적용할 예정입니다.

아래는 그 예시코드입니다.

#include <omp.h>
#include <stdio.h>
#include <stdlib.h>
int data_num = 1000000;
int thread_cnt = 8;

int main() {
	long long* my_arr1 = (long long*)malloc(sizeof(long long) * thread_cnt * data_num);
	long long* my_arr2 = (long long*)malloc(sizeof(long long) * thread_cnt * data_num);
	long long* ret_arr = (long long*)malloc(sizeof(long long) * thread_cnt * data_num);
	for (int i = 0; i < thread_cnt * data_num; i++) {
		my_arr1[i] = i;
		my_arr2[i] = i;
	}

	#pragma omp parallel for
	for (int i = 0; i < thread_cnt * data_num; i++) {
		ret_arr[i] = my_arr1[i] + my_arr2[i];
	}
	
	for (int i = 0; i < 100; i++) {
		printf("%d\n", ret_arr[i]);
	}
}

3. C++의 SIMD intrinsic

simd는 single instruction multi data의 약자로

하나의 명령어로 2개 이상의 Data를 한번에 처리할 수 있는 처리 방법을 의미합니다.

(다른말로 요즘은 AVX라고 많이 불리죠?)

단순하게 함수로 묶어서 코드한줄에 두개의 Data를 처리하는것이 아니라

CPU명령어 Level에서 한번에 256, 512bit의 Data를 한번에 계산할 수가 있습니다.

기본적으로 <xmminsrin.h>와 <immintrin.h>를 이용해서 256bit SIMD연산이 가능합니다.

저는 이번에 SIMD Array Add연산을 통해서 성능비교를 해볼 예정입니다.

256bit SIMD Add를 통해서 Data를 계산하는 방법은 아래를 참고해주세요!

(SIMD에 대한 내용은 추후 연재한 예정입니다! 기대해주셔요🤭)

#include <iostream>
#include <stdio.h>
#include <stdlib.h>
#include <malloc.h>
#include <xmmintrin.h>
#include <immintrin.h >
int data_num = 1000000;
int thread_cnt = 8;

int main() {
	//_aligend_malloc을 사용해서 정렬된 memory를 할당받음
	int* my_arr1 = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
	int* my_arr2 = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
	int* ret_arr = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
	for (int i = 0; i < thread_cnt * data_num; i++) {
		my_arr1[i] = i;
		my_arr2[i] = i;
	}

	__m256i* _arr1 = (__m256i*)my_arr1;
	__m256i* _arr2 = (__m256i*)my_arr2;
	__m256i* _ret = (__m256i*)ret_arr;
	
	for (int i = 0; i < ((thread_cnt * data_num)/8); i++) {		
		_ret[i] = _mm256_add_epi32(_arr1[i], _arr2[i]);
	}	
	
	for (int i = 0; i < 22222; i++) {
		printf("%d\n", ret_arr[i]);
	}
}

4. Integration

이제 위 기술들을 조합해봐야겠죠?

SIMD만 적용
OpenMP만 적용
Thread만 적용
OpenMP + SIMD
Thread + SIMD

위에서 각자 적용하는 것은 해 보았으니

SIMD와 병렬처리를 조합한 코드를 간단히 보고 넘어가겠습니다.

//Thread + SIMD
#include <utility>
#include <limits.h>
#include <stdlib.h>
#include <thread>
#include <vector>
#include <stdio.h>
#include <immintrin.h>
int data_num = 1000000;
int thread_cnt = 8;

void func_simd(int* _ret, int* _a, int* _b, int th_idx) { 
    //Pointer의 위치를 옮긴다음 type casting을 통해서
    //각 thread에서 다른 위치에서 시작점을 갖는 Array생성
    __m256i* a = (__m256i*)&_a[th_idx * ((data_num*thread_cnt) / 8)];
    __m256i* b = (__m256i*)&_b[th_idx * ((data_num*thread_cnt) / 8)];
    __m256i* result = (__m256i*)&_ret[th_idx * ((data_num*thread_cnt) / 8)];
    for (int i = 0; i < (data_num*thread_cnt) / ((256/32)*thread_cnt); i++) {
    //256bit / (4byte*8bit/byte) => 한번에 8개의 integer를 처리
    //근데 thread만큼 나눴으므로 thread_cnt만큼 적은 반복문 처리
        result[i] = _mm256_add_epi32(a[i], b[i]);
    };
}

int main() {
    int* a = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
    int* b = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);    
    int* ret = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
    std::vector<std::thread> v;
    for (int i = 0; i < thread_cnt; i++) {
        v.push_back(std::thread(func_simd, ret, a, b, i));        
    }

    for (auto& thread : v) {
        thread.join();
    }
    return ret;
}

Thread한태 malloc을 통해서 할당받은 pointer를 넘겨주고

해당 pointer를 thread job내부에서 typecasting을 통해서 계산하는 구조를 가지고 있습니다.

그럼 이번에 OpenMP를 사용한 경우를 보실까요?

//Thread + SIMD
#include <utility>
#include <limits.h>
#include <stdlib.h>
#include <thread>
#include <vector>
#include <stdio.h>
#include <immintrin.h>
int data_num = 1000000;
int thread_cnt = 8;

int main() {
    int* a = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
    int* b = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);    
    int* ret = (int*)_aligned_malloc(sizeof(int) * data_num * thread_cnt, 32);
    __m256i* _a = (__m256i*)a;
    __m256i* _b = (__m256i*)b;
    __m256i* _ret = (__m256i*)ret;
    #pragma omp parallel for    
    for (int i = 0; i < ((thread_cnt * data_num)/8); i++) {		
        _ret[i] = _mm256_add_epi32(_a[i], _b[i]);
    }	
    return ret;
}

OpenMP를 사용할 경우, 위처럼 편리하게 반복문을 병렬화 시켜서 계산 할 수가 있습니다.

최종 벤치마킹을 위해 사용된 C함수들의 소스코드 입니다.

//mydll.c
#include "pch.h" // use stdafx.h in Visual Studio 2017 and earlier
#include <utility>
#include <limits.h>
#include <stdlib.h>
#include <thread>
#include <vector>
#include <stdio.h>
#include "mydll.h"
#include <omp.h>
#include <immintrin.h>

void* list_add_serial(int* a, int* b, int* ret, int data_cnt) {    
    for (int i = 0; i < (data_cnt); i++) {
        ret[i] = a[i] + b[i];
    }    
}

void* list_add_parallel_omp(int* a, int* b, int* ret, int data_cnt) {    
    #pragma omp parallel for    
    for (int i = 0; i < (data_cnt); i++) {
        ret[i] = a[i] + b[i];
    }    
}

void func(int* _ret, int* _a, int* _b, int tot_len, int th_idx) {
    int partition = tot_len / 8;
    for (int i = (th_idx)*partition; i < (th_idx + 1) * partition; i++) {
        _ret[i] = _a[i] + _b[i];
    }
}

void* list_add_parallel_thread(int* a, int* b, int* ret, int data_cnt) {
    int thread_cnt = 8;    
    std::vector<std::thread> v;
    for (int i = 0; i < thread_cnt; i++) {
        v.push_back(std::thread(func, ret, a, b, data_cnt, i));
    }
    for (auto& thread : v) {
        thread.join();
    }    
}


void* list_add_simd_serial(int* a, int* b, int* ret, int data_cnt) {
    __m256i* _a = (__m256i*)a;
    __m256i* _b = (__m256i*)b;
    __m256i* _ret = (__m256i*)ret;
    for (int i = 0; i < data_cnt / (8); i++) {
        _ret[i] = _mm256_add_epi32(_a[i], _b[i]);
    }
}

void* list_add_simd_parallel_omp(int* a, int* b, int* ret, int data_cnt) {
    __m256i* _a = (__m256i*)a;
    __m256i* _b = (__m256i*)b;
    __m256i* _ret = (__m256i*)ret;
    #pragma omp parallel for
    for (int i = 0; i < data_cnt / (8); i++) {
        _ret[i] = _mm256_add_epi32(_a[i], _b[i]);
    }
}


void func_simd(int* _ret, int* _a, int* _b, int tot_len, int th_idx) {
    __m256i* a = (__m256i*) & _a[th_idx * (tot_len / 8)];
    __m256i* b = (__m256i*) & _b[th_idx * (tot_len / 8)];
    __m256i* result = (__m256i*) & _ret[th_idx * (tot_len / 8)];
    for (int i = 0; i < tot_len / (8 * 8); i++) {
        result[i] = _mm256_add_epi32(a[i], b[i]);
    };
}


void* list_add_simd_parallel_thread(int* a, int* b, int* ret, int data_cnt) {    
    std::vector<std::thread> v;
    for (int i = 0; i < 8; i++) {
        v.push_back(std::thread(func_simd, ret, a, b, data_cnt, i));
    }
    for (auto& thread : v) {
        thread.join();
    }    
}

5. Benchmark

import ctypes
import numpy as np
import timeit
lib = ctypes.CDLL(r'mydll.dll')
func_list = [
        "list_add_serial",
        "list_add_parallel_omp",
        "list_add_parallel_thread",
        "list_add_simd_serial",
        "list_add_simd_parallel_omp",
        "list_add_simd_parallel_thread"
        ]
ret = {func : [] for func in func_list}
ret["numpy_add"] = []

#data개수를 10^5~10^9까지 테스트
for data_cnt in (10**(i+5) for i in range(5)):
    dt1 = np.ones(data_cnt, dtype = np.int32)
    dt2 = np.ones(data_cnt, dtype = np.int32)
    
        
    def np_add_2d(_a, _b, _lib):
        global data_cnt        
        _ret = np.empty_like(_b, dtype = np.int32)
        temp_a = np.ctypeslib.as_ctypes(_a)
        temp_b = np.ctypeslib.as_ctypes(_b)    
        temp_ret = np.ctypeslib.as_ctypes(_ret)
        _lib(temp_a, temp_b, temp_ret, len(_ret))
        return _ret    
    
    def np_add_normal(_a, _b):    
        return np.add(_a, _b)
            
    
    ret["numpy_add"].append(timeit.timeit('np_add_normal(dt1, dt2)', setup = "from __main__ import dt1, dt2, np_add_normal", number = 10))
    for func in func_list:    
        ret[func].append(timeit.timeit(f'np_add_2d(dt1, dt2, lib.{func})', setup = "from __main__ import dt1, dt2, np_add_2d, lib", number = 10))
        
#벤치마킹 결과 시각화
import matplotlib.pyplot as plt
import seaborn as sns
plt.figure(figsize = (10, 5))
for func in ret.keys():
    sns.lineplot(x = [10**(i+5) for i in range(5)],y = ret[func], label = func, markers = True, dashes = False)
plt.show()

(x축 : data_count, y축 : 10회반복 실행 소요시간)

함수이름	실행시간
numpy_add	22.04
list_add_serial	19.61
list_add_parallel_omp	19.62
list_add_parallel_thread	11.68
list_add_simd_serial	18.97
list_add_simd_parallel_omp	18.90
list_add_simd_parallel_thread	11.63