DONGBUM KIM

Beat Yesterday

3 분 소요

2026년 수정 안내 이 글에 처음 실었던 측정 코드에 버그가 있었다. 두 번째 측정 블록이 vec2가 아니라 첫 번째 블록에서 이미 정렬을 끝낸 vec1을 다시 정렬하고 있었다. 즉 병렬 정렬 쪽 수치는 “이미 정렬된 배열을 다시 정렬한 시간”이라 비교 대상이 되지 못한다. 코드를 바로잡고, 신뢰할 수 없는 결과표는 내렸다. 아래는 수정된 코드와, 재측정 전까지 유효한 판단 기준이다.

서버 코드를 점검하다가 작은 알고리즘부터 하나씩 개선하고 있다.

병렬 프로그래밍을 조사하던 중 정렬 알고리즘에서도 최적화 여지가 있다는 것을 알게 되어 테스트해보았다.

기존 코드는 std::sort를 사용하고 있었다. 이를 concurrency::parallel_sort로 바꿀 수 있었지만, PPL은 비표준인 데다 Windows에서만 동작하는 코드였다. ChatGPT는 concurrency::parallel_sort 대신 std::sort에 std::execution::par를 조합하는 방식을 추천해주었다. (C++17 이상에서만 유효하다.)

테스트 코드

기존의 std::sort와 std::sort + std::execution::par를 비교하는 코드를 작성했다.

#define WIN32_LEAN_AND_MEAN
#include <Windows.h>
#include <vector>
#include <algorithm>
#include <string>
#include <random>
#include <chrono>
#include <execution>
#include <iostream>
#include <locale>

int main( void )
{
	std::vector<int> vec;

	for ( auto i = 0; i < 1000000; ++i )
		vec.push_back( i );

	std::random_device rd;
	std::mt19937 engine( rd() );
	std::shuffle( vec.begin(), vec.end(), engine );

	std::vector<int> vec1 = vec;
	std::vector<int> vec2 = vec;

	{
		// 시간 측정 시작
		auto start = std::chrono::high_resolution_clock::now();

		std::sort( vec1.begin(), vec1.end(), []( auto lhs, auto rhs )
		{
			return lhs < rhs;
		} );

		// 시간 측정 끝
		auto end = std::chrono::high_resolution_clock::now();

		// 경과 시간 계산 (단위: 마이크로초)
		auto duration = std::chrono::duration_cast<std::chrono::microseconds>( end - start );

		std::cout << "실행 시간: " << duration.count() << " 마이크로초" << std::endl;
	}

	{
		// 시간 측정 시작
		auto start = std::chrono::high_resolution_clock::now();

		std::sort( std::execution::par, vec2.begin(), vec2.end(), []( auto lhs, auto rhs )
		{
			return lhs < rhs;
		} );

		// 시간 측정 끝
		auto end = std::chrono::high_resolution_clock::now();

		// 경과 시간 계산 (단위: 마이크로초)
		auto duration = std::chrono::duration_cast<std::chrono::microseconds>( end - start );

		std::cout << "실행 시간: " << duration.count() << " 마이크로초" << std::endl;
	}

	return 0;
}

배열에 int 값을 여러 개 넣고 섞은 후 다시 정렬하는 데 걸리는 시간을 측정하는 코드이다.

측정할 때 주의할 점

원래 이 자리에는 원소 개수별 측정 표가 있었지만, 위에서 밝힌 버그 때문에 병렬 쪽 수치가 무의미해 내렸다. 같은 실수를 반복하지 않으려면 다음을 확인해야 한다.

  • 매 측정마다 입력 상태를 동일하게 맞출 것. 정렬은 입력이 이미 정렬되어 있는지에 따라 성능이 크게 달라진다. vec1, vec2처럼 원본에서 복사한 별도의 벡터를 각각 써야 한다.
  • 한 번만 재지 말 것. 여러 번 반복해 중앙값을 쓰는 편이 안정적이다. 첫 실행은 페이지 폴트와 캐시 워밍업 때문에 느리게 나온다.
  • 릴리즈 빌드로 측정할 것. 디버그 빌드의 STL은 반복자 검사 때문에 몇 배 느리다.
  • 정렬 대상의 타입을 고려할 것. int처럼 비교와 이동이 싼 타입은 병렬화 이득이 작고, 문자열이나 큰 구조체는 이득이 크다.

임계값을 어떻게 잡을 것인가

std::execution::par는 스레드 풀에 작업을 나눠주고 결과를 합치는 고정 비용이 있다. 원소가 적으면 이 비용이 정렬 자체보다 커서 오히려 느려진다. 따라서 임계값을 두고 갈라주는 접근 자체는 타당하다.

다만 정확한 임계값은 하드웨어·타입·컴파일러마다 다르므로 각자 측정해서 정해야 한다. 수천 개 단위에서 갈리는 경우가 많지만, 이건 출발점으로 삼을 값이지 그대로 믿을 값이 아니다.

아래와 같은 템플릿 함수를 만들어 사용 중이다. std::sort와 동일한 인터페이스를 유지하면서 원소 개수에 따라 정렬 전략을 자동으로 선택한다.

// 임계값은 프로젝트 환경에서 직접 측정해 조정할 것.
constexpr std::ptrdiff_t kParallelSortThreshold = 1000;

template<class RandomIt, class Compare>
void RzSort( RandomIt first, RandomIt last, Compare comp )
{
	const auto size = std::distance( first, last );
	if ( size <= 1 )
		return;

	if ( size < kParallelSortThreshold )
		std::sort( first, last, comp );
	else
		std::sort( std::execution::par, first, last, comp );
}

그 밖에 알아둘 것

  • std::execution::par는 C++17부터 쓸 수 있다. MSVC는 별도 설정 없이 동작하지만, GCC/Clang의 libstdc++에서는 기본적으로 Intel TBB가 필요하다. 링크하지 않으면 undefined reference 링크 에러가 난다. (libstdc++를 _GLIBCXX_USE_TBB_PAR_BACKEND=0으로 빌드한 환경이라면 TBB 없이도 컴파일은 되지만 병렬 실행 없이 순차로 동작한다 — 배포판 기본 빌드에서는 흔치 않은 설정이다.) CMake라면 find_package(TBB REQUIRED) 후 TBB::tbb를 링크한다.
  • 비교 함수가 스레드 안전해야 한다. 위 예제처럼 상태 없는 람다라면 문제없지만, 외부 변수를 캡처해 수정하는 비교자는 데이터 레이스가 된다.
  • std::execution::par_unseq는 벡터화까지 허용하므로 더 빠를 수 있지만, 비교 함수 안에서 메모리 할당이나 락을 쓰면 안 된다는 제약이 추가된다.

댓글 남기기

참고