SkipList

예를 들어 32bit 환경에서 어떤 객체가 4바이트 단위로 정렬된다면, 그 객체의 주소 하위 2비트는 항상 00이 된다. 이처럼 aligment 등으로 인해 항상 일정한 값을 가지는 비트를 tag(플래그, 타입 정보 등) 용도로 활용하는 기법을 tagged pointer라고 한다.

Example

리눅스 커널에서 동적으로 할당된 포인터는 최소 4바이트로 정렬되므로, 하위 2비트는 항상 0이다. 리눅스의 XArray는 이 점을 활용해 포인터 값 자체에 작은 태그를 함께 인코딩한다. 사용 가능한 태그 값은 0, 1, 3이며, 2는 내부 엔트리(internal entry)용으로 예약되어 있다.

• xa_tag_pointer(): 포인터의 하위 비트에 태그를 OR 연산으로 삽입한다.
• xa_untag_pointer(): 하위 2비트를 마스킹해 원래 포인터를 복원한다.
• xa_pointer_tag(): 하위 2비트만 추출해 저장된 태그 값을 반환한다.

static inline void *xa_tag_pointer(void *p, unsigned long tag)
{
    return (void *)((unsigned long)p | tag);
}

static inline void *xa_untag_pointer(void *entry)
{
    return (void *)((unsigned long)entry & ~3UL);
}

static inline unsigned int xa_pointer_tag(void *entry)
{
    return (unsigned long)entry & 3UL;
}

단순히 컨테이너를 순서대로 띄우고 싶다면 depends_on 서비스의 연결까지 보장하고 싶다면 healthcheck와 condition: service_healthy를 조합

Depends_on

depends_on은 컨테이너 간의 실행 순서를 정의해주는 옵션이다. 특정 컨테이너가 다른 컨테이너보다 먼저 실행되어야 할 때 사용한다.

services:
  redis:
    image: redis:latest

  nginx:
    image: nginx:latest
    depends_on:
      - redis # redis 컨테이너가 먼저 실행된 후 nginx가 실행된다.

docker compose는 의존 대상인 컨테이너의 프로세스가 시작되면 즉시 다음 컨테이너를 실행시킨다. 이 점으로 인해 데이터베이스와 같이 내부 초기화(데이터 로딩, 네트워크 연결 등) 시간이 필요한 프로세스는 Connection Refused 같은 에러가 발생될 수 있다.

(정말 단순하게 실행 순서만 보장한다)

Healthcheck

healthcheck는 컨테이너 내부의 서비스가 실제로 요청을 처리할 준비가 되었는지 확인할 때 사용한다. 이러한 점으로 depends_on과 함께 사용하면, 컨테이너의 상태가 healthy로 변경된 시점을 확인하여 의존성을 완벽하게 보장할 수 있게된다.

예시

services:
  redis:
    image: redis:latest
    healthcheck: # 5초마다 redis ping 테스트를 총 5번 진행  (timeout 3초)
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5
      start_period: 30s

  app:
    build: .
    depends_on:
      redis:
        condition: service_healthy # redis가 healthy 상태가 될 때까지 실행 대기

설정 옵션

healthcheck에서 사용되는 옵션은 다음과 같다.

• test: 상태를 결정할 명렁어 (성공 시 0, 실패 시 1 반환)
• interval: 검사를 수행할 주기
• timeout: 설정한 시간 안에 응답이 없으면 실패로 간주
• retries: 연속으로 실패한 경우 unhealthy 상태로 변경
• start_period: 컨테이너 부팅 직후, 초기화를 위해 검사 실패를 무시하는 유예 기간 (이 시간동안 retries를 차감하지 않음)

condition

depends_on에서 condition으로 설정할 수 있는 옵션은 크게 3가지로 구성되어 있다.

service_started (기본값)

의존하는 컨테이너의 프로세스가 시작되면 즉시 다음 컨테이너를 실행 이 옵션은 condition을 별도로 명시하지 않았을 때의 기본 동작이며, 실제 서비스가 준비되었는지 보장하지 않는다.

service_healthy

의존하는 컨테이너의 healthcheck 결과가 healthy (준비 완료)가 될 때까지 기다린다. 이 옵션을 지정해야지만 Connection Refused와 같은 문제를 해결할 수 있다.

service_completed_successfully

의존하는 컨테이너가 실행을 마치고 정상적으로 종료 (exit code 0)될 때까지 기다린다. 일회성 작업 (초기 데이터 삽입, 빌드 스크립트 등)들이 완전히 끝난 후에 메인 앱을 실행해야 할 때 사용한다.

하지만 리눅스 기반 CLI 환경에서도 가상 디스플레이를 구성하면, headful 브라우저를 실행해 크롤링을 수행할 수 있다.

Xvfb

Xvfb는 실제 모니터가 없어도 GUI 애플리케이션을 실행할 수 있게 해준다.

apt install xvfb

가상 디스플레이 구축

Xvfb는 아래의 명령어로 가상 디스플레이를 구축할 수 있다.

xvfb :99 -screen 0 1920x1080x24 &

사용된 옵션 정보는 다음과 같다.

• :99 가상 디스플레이 번호
• -screen 0 첫 번째 스크린
• 1920x1080x24 해상도 / 색 깊이 (24bit)
• & 백그라운드 실행

DISPLAY 환경 변수 설정

export DISPLAY=99

DISPLAY 환경 변수는 리눅스가 사용하는 표준 환경 변수이다. 따라서 위에서 구축한 가상 디스플레이 번호로 환경 변수를 세팅해주면 된다.

GUI 애플리케이션 실행

python crwal.py

위에서 설정한 디스플레이 환경 변수로 인해 추가적인 명령어 없이도 크롤러는 가상 디스플레이를 통해 Headful 크롤링을 수행하게 된다.

xvfb-run

xvfb-run는 위 과정을 명령어 한 줄로 처리하는 래퍼이다.

xvfb-run -s "-screen 0 1920x1080x24" python crawl.py

• -a: 비어있는 display 번호를 탐색하여 실행
• -s: Xvfb에 전달할 인자 (-screen <screen_number> <resolution>x<depth>)

사용법

Cron은 내부적으로 cron daemon (crond)이 백그라운드에서 실행되며, crontab에 등록된 스케줄을 주기적으로 확인한다.

crontab -l   # 목록 보기
crontab -r   # 등록된 모든 crontab 삭제 
crontab -e   # 현재 사용자 crontab 편집

# ex) 매 5분마다 실행
*/5 * * * * /usr/bin/python3 /app/job.py

문법 정리

crontab은 "언제(스케줄) 무엇을(커맨드)" 실행할지 적는 설정이며, 형식은 아래와 같다.

# ┌─ 분(0-59)
# │ ┌─ 시(0-23)
# │ │ ┌─ 일(1-31)
# │ │ │ ┌─ 월(1-12)
# │ │ │ │ ┌─ 요일(0-7, 0과 7은 일요일)
# │ │ │ │ │
# * * * * *  실행할_명령어

필드값

각 시간 필드는 다음 표현을 조합해서 사용이 가능하다.

와일드카드 *

• *: 와일드카드는 모든 값을 의미한다. ex) * * * * * -> 매 분마다 실행

단일값

• <value>: value 값일 때만 실행 ex) 0 5 * * * -> 매일 05:00에 실행

범위 -

• <value1>-<value2>: value1부터 value2까지 실행 ex) 0 5 * * 1-5 -> 평일 05:00에 실행

리스트 ,

• <value1>,<value2>,.. -> 여러 값 지정 가능 ex) 0 5 * * 1,3,5 -> 월/수/금 05:00에 실행

간격

• */<value>: 매 value마다 실행 ex) */10 * * * * -> 매 10분마다 실행

• <value1>/<value2>: value1부터 시작해서 매 value2마다 실행 ex) 1-30/5 * * * * -> 매 시간 01,06,11,16,21,26분에 실행

pytest의 fixture는 테스트에 필요한 준비물(의존성/리소스) 을 함수에 정의해두고 테스트 함수의 인자를 주입하는 형태로 재사용하는 기능이다. (중복되는 setup/teardown 코드를 줄이기 위해 사용)

인자 주입

pytest @pytest.fixture로 등록된 함수는 테스트 함수의 파라미터 이름을 기준으로 매칭된다. 즉, pytest는 test_... 함수의 인자 목록을 보고 동일한 이름의 fixture를 찾아 실행한 뒤 반환값(또는 yield 값)을 인자로 주입한다.

import pytest

@pytest.fixture
def user():
    return {"id": 1, "name": "test_user"}

def test_user_name(user):
    assert user["name"] == "test_user"

위 코드에서 test_user_name는 user()를 직접 호출하지 않아도 pytest가 test_user_name(user)의 user 파라미터를 보고 user fixture를 실행한 결과를 주입해준다.

Fixture Scope

fixture는 scope 단위로 결과를 캐시하고, 같은 scope 안에서는 재사용한다. (즉, scope="function"이면 테스트 함수마다 새로 만들고, 그 함수 안에서는 동일 fixture를 여러 번 요청해도 재사용된다.)

@pytest.fixture(scope="function")
def fx():
    ...

스코프의 종류는 다음과 같다

• function: 테스트 함수마다 새로 생성
• class: 클래스 단위로 1번 생성
• module: 파일(모듈) 단위로 1번 생성
• package: 패키지 단위로 1번 생성
• session: 전체 테스트 실행 동안 1번 생성

teardown

teardown은 테스트에서 사용한 리소스를 정리(cleanup)하기 위한 코드이며, pytest fixture에서 yield를 기준으로 setup/teardown을 한 함수에서 구현할 수 있다.

@pytest.fixture(scope="function")
def db_session():
    session = SessionLocal() 
    yield session    # setup 완료 후 테스트에 session 전달
    session.close()  # teardown: fixture scope 종료 시 정리

def test_db_connection(db_session):
    ...

test_db_connection 함수가 호출되면 다음의 흐름으로 진행된다.

1. yield 이전: DB 세션 생성 (setup)
2. yield 지점: session을 pytest가 받아서 테스트 함수 인자로 주입
3. test_db_connection 실행 시작
4. test_db_connection 실행 종료
5. yield 이후: DB 세션 종료 (teardown)

실제 구현

pytest는 fixture 함수에 대해 generator 객체를 next()로 재개(resume) 해서 teardown을 수행하는 구조로 되어있다.

pytest 관련 코드 링크

# pytest/src/_pytest/fixtures.py
def call_fixture_func(
    fixturefunc: _FixtureFunc[FixtureValue], request: FixtureRequest, kwargs
) -> FixtureValue:
    ...
    generator = fixturefunc(**kwargs)  # 제너레이터 객체 생성
    fixture_result = next(generator)   # yield 까지 실행: setup 수행

    # scope가 끝날 때 teardown을 호출하는 콜백 등록
    finalizer = functools.partial(_teardown_yield_fixture, fixturefunc, generator)
    request.addfinalizer(finalizer)
    ...
    return fixture_result # yield 값 반환


def _teardown_yield_fixture(fixturefunc, it) -> None:
    try:
        next(it)  # teardown 수행
    except StopIteration:
        pass  # 정상 종료 
    else:     
        fs, lineno = getfslineno(fixturefunc)  # yield가 2개 이상이라면 에러 반환 
        fail(
            f"fixture function has more than one 'yield':\n\n"
            f"{Source(fixturefunc).indent()}\n"
            f"{fs}:{lineno + 1}",
            pytrace=False,
        )

pytest는 첫 next(generator)로 setup을 수행하고 yield 값을 받아 테스트에 주입한다. 그리고 teardown을 즉시 실행하지 않고 _teardown_yield_fixture(..., generator)를 finalizer로 등록한다. 이후 테스트 scope가 끝나는 시점에 next(generator)를 다시 호출하여 teardown을 진행한다. (teardown을 위한 fixture을 사용 시, yield를 정확히 한 번만 사용해야 된다는 것도 알 수 있다.)

해당 문제는 LeetCode의 Daily Question에서 2025.11.21.에 나온 문제입니다. (난이도: Medium)

참신한 풀이가 있어서 가져왔습니다.

문제

문제 정의는 간단하게 서술하겠습니다. (자세한 건 위의 링크를 타고 들어가주세요) 문자열 s가 주어졌을 때, Palindrom이 되는 3글자로 된 부분 문자열의 종류 찾는 것입니다. (중복 문자열 제외)

예시 데이터 s = "asafa" answer = 3 asa (asafa의 부분 문자열) asa (asafa의 부분 문자열) afa (asafa의 부분 문자열)

풀이

문자열의 크기는 최대 $10^5$까지 들어오기 때문에 $O(N^3)$의 풀이로는 해결하지 못한다. (단순히 반복문 3개를 이용하여 first, middle, last를 찍으며 푸는 방식)

풀이1 (내 풀이)

각 알파벳마다 처음(first)과 끝(last)에 등장하는 인덱스를 기록하고, $first < i < last$를 만족하는 $s[i] \notin temp_set$을 구하면 된다.

조금 쉽게 표현하면, 3글자로 된 팰린드롬은 1번째 글자와 3번째 글자는 같아야 하며, 그 가운데에 들어가는 알파벳의 종류만 구하면 된다는 뜻이다.

struct Postion {
    int first;
    int last;
};

class Solution {
public:
    int countPalindromicSubsequence(string s) {
        vector<Postion> letters_position(26, {-1, -1});
        for (int i = 0; i < s.size(); i++) {
            char cur = s[i] - 'a';
            if (letters_position[cur].first == -1)
                letters_position[cur].first = i;
            letters_position[cur].last = i;
        }

        int answer = 0;
        for (auto position : letters_position) {
            int left = position.first;
            int right = position.last;
            if (left == -1)
                continue;

            // 가운데에 들어가는 알파벳 중에 유니크한 것들만 뽑아내기 위해 set을 사용
            unordered_set<char> temp_set;
            for (int i = left + 1; i < right; i++) {
                temp_set.insert(s[i]);
            }
            answer += temp_set.size();
        }

        return answer;
    }
};

풀이2 (시간 복잡도 개선)

내 방식의 풀이는 $O(N + 26N)$ 이지만, 지금 서술할 풀이는 $O(N + 27^2 * log(N))$으로 조금 더 최적화 된 풀이이다. (상수값이 최적화 됨, 단 push_back으로 인한 연산은 배제)

먼저 각 알파벳의 등장 인덱스를 오름차순으로 저장한다.

그리고 팰린드롬의 가운데에 들어갈 알파벳(두 번째 글자)이 첫 번째 글자와 세 번째 글자의 인덱스 사이에 존재하는지를upper_bound를 이용해 검사하는 방식이다. (upper_bound의 사용으로 상수값이 최적화됨)

class Solution {
public:
    int countPalindromicSubsequence(string s) {
        vector<vector<int>> pos(26);
        for (int i = 0; i < s.size(); i++) {
            pos[s[i] - 'a'].push_back(i);
        }

        int ans = 0;

        // 문자 X 선택
        for (int x = 0; x < 26; x++) {
            auto &vx = pos[x];
            // 팰린드롬이 만들어질 수 없는 케이스
            if (vx.size() < 2) continue;   

            int left = vx.front();
            int right = vx.back();

            // 중간에 들어갈 문자 Y 검사
            for (int y = 0; y < 26; y++) {
                auto &vy = pos[y];

                // index1 보다 큰 첫 위치 찾기
                auto it = upper_bound(vy.begin(), vy.end(), left);

                // 중간 영역에 있는지 확인
                if (it != vy.end() && *it < right)
                    ans++;
            }
        }

        return ans;
    }
};

미리보는 최적화 결과 노드 감소량: 1,410,449개 감소 노드 감소율: 59.32% 감소 메모리 절감량: 0.897 GB 메모리 절감율: 57.5%

해당 블로그 글에서 사용된 코드는 해당 Github Repository에서 확인하실 수 있습니다. (코드까지 포함하면 글이 너무 길어져 Github 링크를 참조하게 된 점 양해 부탁드립니다.)

자료구조 설명

Trie

Compressed Trie를 보기 전에 아주 간단하게 Trie에 대해서 복기하자면 이렇다.

자료 출처: WIKIPEDIA Trie는 문자열을 저장하고 효율적으로 탐색하기 위한 트리 형태의 자료구조이다.

각 노드마다 한 글자가 들어가며, 트리를 탐색하여 내려갈 때마다 글자가 조합되어 leaf node에 도달하면 단어가 완성이 되는 구조로 되어있다. (사진에서 왼쪽으로 쭉 타고 내려가면 t 노드, o 노드를 지나 최종적으로 to 단어가 완성된다.)

Compressed Trie

Trie의 단점은 한 노드 당 하나의 글자만 저장하기 때문에 메모리 사용량이 크다는 단점이 있다. 이 때 단어들 간의 공통된 prefix는 하나의 노드로 묶고, 나머지 부분들은 분기하여 각각 하나의 노드로 묶는 식으로 구조를 변경하면 전체 노드 수를 크게 줄일 수 있다.

두 단어 APPLE과 APT가 주어졌다고 해보자. 두 단어의 공통된 부분(AP)을 하나의 노드로 합치고, 그 외의 겹치지 않는 글자(PLE, T)들을 각각 하나의 노드로 구성하면 Compressed Trie가 된다.

Trie에서는 노드 6개 였던 것이 Compressed Trie 에서는 노드 3개로 압축된 것을 확인 할 수 있다.

최적화 결과

데이터셋은 우리말샘 사전에서 가져왔으며, 총 단어는 1,190,944개로 테스트 데이터를 구축하였다.

Trie 구축 후

• 총 노드의 개수: 2378541
• 노드 insert 시간: 6.08초
• 노드 search 시간: 평균 0.002초
• 메모리 사용량: 1.559 GB

Compressed Trie 구축 후

• 총 노드의 개수: 968092
• 노드 insert 시간: 1분 16초
• 노드 search 시간: 평균 0.003초
• 메모리 사용량: 662 MB

정리

사전 데이터를 토대로 Compressed Trie를 구축했을 때 결과는 다음과 같다

총 결과 노드 감소량: 1,410,449개 감소 노드 감소율: 59.32% 감소 메모리 절감량: 0.897 GB 메모리 절감율: 57.5%

초기 데이터 로드에 12.5배 정도 늘어났지만 search 시간에는 큰 변동이 없으며, 메모리 절감율 57.5%면 Compressed Trie의 압축률은 훌륭하다고 볼 수 있다.

Insert 시간이 증가한 이유는 Compressed Trie에서는 단어 삽입 시 노드를 분할(split)하거나 병합(merge)해야 하는 경우가 생기기 때문이다. (Trie는 Insert시 추가 연산이 없음)

블로그 정리를 위해 Compressed Trie 자료를 다시 찾아보다보니 한국어 자료가 부족하다는 걸 느꼈습니다. 그럴리는 없겠지만 반응이 좋다면, Compressed Trie의 Insert, Search, Delete 연산에 대해서 코드와 함께 블로그 정리하겠습니다.

미리보는 결론 단일 DB의 환경이라면, 인덱스 크기를 고려하여 AUTO_INCREMENT를 사용하는 것이 좋다. 그러나 보안을 신경써야 하며 여러 노드가 동시에 ID를 생성하는 환경이라면, UUID가 좋다. 또한, UUID 버전 중 순수 랜덤이 필요 없는 경우에는 성능을 고려하여 v6 또는 v7을 사용하는 것이 효율적이다.

UUID란?

UUID(Universally Unique IDentifier)는 간단하게 겹치지 않는 ID라고 보면 된다. 일반적으로 32개의 16진수(HEX)로 구성되어있으며, 하이폰으로 구분해서 총 36글자로 되어있다.

UUID 예시) 
d6e74418-2081-4256-8c44-d8922d26dece

UUID 장점

AUTO_INCREMENT를 사용하는 ID 값과 대비하여 UUID가 가지는 장점이 있다.

1. 보안적 이점: 순차 ID(1,2,..)는 쉽게 추측되어 예측이 가능하지만, UUID는 랜덤/비예측적이라 위험도가 낮다.
2. 동시성 확장성 향상: AUTO_INCREMENT는 DB가 시퀀스를 관리해 병목이 생기고 샤딩이 어렵다. UUID는 애플리케이션에서 독립 생성이 가능해 여러 서버나 리전에서 충돌없이 생성 가능하다.

UUID 버전별 차이점

버전별 사용처

이름 기반

버전 v3, v5는 입력값에 따라 항상 동일한 UUID가 생성된다. 또한, v5는 MD5 대비 충돌 확률이 낮다.

즉, 입력값에 따른 고정 ID값이 필요한 경우에 v5 사용이 권장된다.

랜덤 기반

버전 v4는 완전히 랜덤하게 생성된다. 따라서 충돌 가능성이 거의 없으며, 보안적으로도 안전하다.

범용적으로 사용되는 버전이며, 유니크한 ID 값이 필요할 때 사용이 권장된다.

시간 기반

버전 v1, v6, v7은 생성 시, 시간 정보를 포함하므로 순차적으로 생성된다. 덕분에 DB B-Tree 인덱스 삽입 시 성능이 최적화된다.

그 중에서 v1은 완벽한 순차적 정렬이 보장되지 않으며, MAC 주소로 인해 개인정보 노출의 위험이 있어 최근에는 사용되는 기법이 아니라고 한다. 즉, DB의 효율을 생각한다면 v6, v7이 권장된다.

근거

v4는 완전 랜덤이기 때문에 데이터를 삽입할수록 인덱스 조각화가 심해진다. 이는 I/O 비용이 증가하고, 인덱스 조회 성능이 저하된다는 것을 의미한다. 사진 출처: toomanyafterthoughts 사이트 위의 자료는 데이터 삽입 대비 디스크 사용량을 나타낸 자료이다. 시간 기반인 v6, v7은 데이터가 많이 삽입되도 디스크 페이지를 순차적으로 사용하므로 디스크 쓰기량이 최소화됨을 확인할 수 있다. (v1도 시간 기반이지만 정확한 시간 순서로 정렬되지 않아 v6, v7에 비해 디스크 사용 최적화가 덜 효율적임도 볼 수 있다.)

게시글 작성 순서를 '결론 -> 배경 -> 설명 -> 근거' 순으로 작성해 보았는데, 독자 입장에서 어떤 방식이 더 친화적인지 고민된다..

참고자료

• https://www.toomanyafterthoughts.com/uuids-are-bad-for-database-index-performance-uuid7/

찾아보고 고민해본 결과 총 3가지 방식으로 이미지 업로드 기능을 구현가능해 보인다.

1. Delayed image upload
2. 임시 이미지 업로드
3. 이미지 업로드 후 스케줄링 기반 정리

해당 글에서는 delayed image upload에 대해서 작성한다. 구현 방식에는 정답이 없으므로 장단점을 고려해 선택하는 것을 추천합니다.

Delayed image upload

흐름

1. 사용자가 이미지를 입력
2. 웹에서는 해당 이미지를 BASE64로 보여줌 (서버로 저장 X)
3. 게시글 저장 시, 3-1. 서버에 이미지들을 업로드한 뒤, URL을 반환 받음 3-2. BASE64 -> URL 치환 -> 게시글 HTML 저장

장점

1. 서버 트래픽 최소화: 게시글 업로드 전까지 트래픽을 사용하지 않음
2. 저장 공간 최소화: 게시글에 저장될 이미지만 서버에 저장됨 (임시 이미지가 저장되지 않음)

단점

1. 브라우저 메모리 부담 증가: BASE64는 원본보다 용량이 약 33% 증가
2. 업로드 시 트래픽 집중: 이미지가 많은 게시글에서는 서버로 한 번에 트래픽이 몰릴 수 있음

구현

1) 이미지 미리보기 및 pendingImage

해당 코드에서는 서버로 업로드 될 이미지를 담는 pendingImages와BASE64로 된 이미지 출력을 메인으로 보면 된다. 또한, 이후 BASE64에서 치환할 file을 트래킹하기 위해 data-index에는 pendingImages의 인덱스도 같이 저장한다.

const pendingImages = []; // 서버 업로드 예정 이미지 저장
const previewContainer = document.getElementById('preview');
const input = document.getElementById('imageInput');

input.addEventListener('change', (e) => {
    const files = Array.from(e.target.files);

    files.forEach((file) => {
        // 1) pendingImages 배열에 저장
        const index = pendingImages.length;
        pendingImages.push(file);

        // 2) FileReader로 Base64 변환
        const reader = new FileReader();
        reader.onload = function(event) {
            // 3) 이미지 요소 생성 및 미리보기
            const img = document.createElement('img');
            img.src = event.target.result; // Base64
            // BASE64에서 치환할 file을 트래킹하기 위한 index 저장
            img.setAttribute('data-index', index);

            previewContainer.appendChild(img);
        };
        reader.readAsDataURL(file);
    });
});

2) pendingImages에 저장된 이미지 서버에 저장

해당 코드에서는 pendingImages에 쌓인 이미지들을 서버에 저장하고, 반환받은 URL로 HTML에 치환하여 게시글을 저장하는 방식이다. (간결한 코드를 위해 따로 예외처리는 되어있지 않습니다.)

// 게시글 제출 시 서버 업로드 + BASE64 → URL 치환
submitBtn.addEventListener('click', async () => {
    // 1) FormData로 서버 업로드
    const formData = new FormData();
    pendingImages.forEach(file => formData.append('files', file));

    // 서버에 이미지 업로드 (예: /api/upload-images)
    const res = await fetch('/api/upload-images', {
        method: 'POST',
        body: formData
    });
    const uploadedUrls = await res.json(); // ["image_url_1.png", "image_url_2.png", ...]

    // 2) Base64 → 서버 URL 치환
    const imgs = previewContainer.querySelectorAll('img');
    imgs.forEach(img => {
        const index = parseInt(img.getAttribute('data-index'));
        img.src = uploadedUrls[index];
    });

    // 3) 게시글 내용 생성 (HTML 포함)
    const title = document.getElementById('title').value;
    const contentHTML = previewContainer.innerHTML;

    // 4) 게시글 서버 전송
    await fetch('/api/post', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
            title,
            content: contentHTML
        })
    });

    alert('게시글 작성 완료!');
});

해당 게시글에서는 fastapi_mail을 이용하여 이메일 인증을 구현해볼 것이다.

방식

이메일 인증의 전체적인 흐름은 다음과 같다.

1. 유저가 회원가입을 진행 → FastAPI가 인증 토큰 생성
2. FastAPI가 Redis에 토큰 저장 (만료시간 포함) key: email_verify:{token}, value: user_id
3. FastAPI가 인증 링크을 포함한 메일 보내기 인증 링크에는 get parameter로 token 값을 저장한다.
4. 유저가 메일 링크 클릭 → FastAPI가 토큰 검증 → 인증 완료 처리

필자는 사용자 편의성을 고려하여 "인증 링크"를 발송하도록 하였지만, "인증 번호" 방식도 이와 비슷하게 진행하면 된다.

이메일 인증 구현

아래는 이해를 돕기 위한 코드로 redis value에 user 정보를 모두 넣어 작성하였다. 실제 프로젝트에서는 DB에 유저 정보를 저장 (email_verified=false)하고, redis value에 user_id를 넣어 메일 인증이 완료되었을 때 email_verified=true를 해주면 된다. (로그인 과정에서는 email_verified=false면, 로그인 실패 + 메일 인증 요구를 해주면 된다)

import json
import secrets
import uvicorn
from typing import Optional
from pydantic import BaseModel, EmailStr

import redis
from fastapi import FastAPI, HTTPException
from fastapi_mail import FastMail, MessageSchema, ConnectionConfig

APP_BASE_URL = "http://127.0.0.1:5000"
VERIFY_TTL_SEC = 60 * 10  # 10분
conf = ConnectionConfig(
    MAIL_USERNAME="<USERNAME>",
    MAIL_PASSWORD="<APP_PASSWORD>", # Gamil의 경우, App 패스워드를 발급받아서 사용하면 된다.
    MAIL_FROM="<FROM_EMAIL>",
    MAIL_SERVER="smtp.gmail.com",
    MAIL_PORT=587,
    MAIL_STARTTLS=True,
    MAIL_SSL_TLS=False,
)

app = FastAPI()
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)

class User(BaseModel):
    name: Optional[str] = None
    email: EmailStr
    password: str


def redis_verify_key(token: str) -> str:
    return f"email_verify:{token}"


async def send_email(title: str, email: str, context: str):
    message = MessageSchema(
        subject=title,
        recipients=[email],
        body=context,
        subtype="plain"
    )
    fm = FastMail(conf)
    await fm.send_message(message)


@app.post("/signup")
async def signup_submit(user: User):
    # 1) 토큰 생성
    token = secrets.token_urlsafe(32)

    # 2) Redis에 저장
    #    key: email_verify:{token}
    #    value: 회원가입 정보
    payload = user.model_dump_json()
    r.setex(redis_verify_key(token), VERIFY_TTL_SEC, payload)

    # 3) 인증 링크 생성
    verify_link = f"{APP_BASE_URL}/verify-email?token={token}"

    # 4) 메일 발송
    title = "[TEST] 이메일 인증"
    content = f"아래 링크를 클릭해서 이메일 인증을 완료하세요:\n\n{verify_link}\n\n(만료: {VERIFY_TTL_SEC//60}분)"
    await send_email(title, user.email, content)
    return {"ok": True, "message": "verification email sent"}


@app.get("/verify-email")
async def verify_email(token: str):
    key = redis_verify_key(token)
    verify_value = r.get(key)
    if not verify_value:
        raise HTTPException(status_code=400, detail="invalid_or_expired_token")

    r.delete(key)
    user = json.loads(verify_value)
    return {"ok": True, "verified_email": user["email"]}


if __name__ == '__main__':
    uvicorn.run(app, host="0.0.0.0", port=5000)

테스트 코드

원래는 회원가입 페이지를 만들어야 하지만, 간단하게 테스트 코드만 작성해두었다.

import requests

user = {
    "email": "zhemfpdlf@gmail.com",
    "password": "<PASSWORD>",
    "name": "<name>"
}

r = requests.post("http://127.0.0.1:5000/signup", json=user)
r.raise_for_status()
print(r.text)

위의 코드를 실행시켰을 때, 다음과 같이 메일이 온다.

또한, 해당 링크를 클릭하면, 메일 인증까지 완료된다. (인증 완료 페이지도 만들고, login 페이지로 리다이렉션까지 진행하면 완벽할 것이다.)

가끔 사이트 이용 중에 한국어 입력 시, 마지막 글자가 2번 입력이 되는 모습을 본 적이 있을 것이다. (아래 짤은 한국어 입력 후 Enter를 눌렀을 때, 마지막 글자가 한번 더 입력되는 모습)

해결법

해당 문제는 조합 중인 글자에서 발생하는 IME 문제이므로 compositionstart (조합시작), compositionend (조합 완성) 이벤트를 이용하여 '글자 조합 확정 이벤트'를 예외처리 해주면 된다.

IME에 대해 좀 더 알아보고 싶다면, 아래의 코드를 실행하여 영어로도 입력해보길 추천 (영어는 조합형 글자가 아니기 때문에 composition evnet가 발생하지 않음)

<!DOCTYPE html>
<html lang="ko">
<head>
    <meta charset="UTF-8">
    <title>IME 이벤트 디버깅</title>
    <style>
        body {
            font-family: sans-serif;
            padding: 20px;
        }

        #input {
            font-size: 1.2em;
            padding: 8px 16px;
            width: 300px;
        }

        #stack {
            margin-top: 20px;
            border: 1px solid #ccc;
            padding: 10px;
            overflow-y: auto;
            background-color: #f9f9f9;
            font-family: monospace;
        }

        .event-line {
            padding: 4px;
            border-bottom: 1px dotted #ddd;
        }

        #list {
            margin-top: 30px;
            padding-left: 20px;
        }

        #list li {
            font-size: 1.1em;
            margin-bottom: 4px;
        }
    </style>
</head>
<body>

<h2>IME 이벤트 디버깅</h2>

<input type="text" id="input"/>
<div id="stack"></div>

<h3>📋 Todo List</h3>
<ul id="list"></ul>

<script>
    const input = document.getElementById("input");
    const stack = document.getElementById("stack");
    const list = document.getElementById("list");

    function pushToStack(message) {
        const time = new Date().toLocaleTimeString();
        const div = document.createElement("div");
        div.className = "event-line";
        div.textContent = `[${time}] ${message}`;
        stack.prepend(div); // 최신 이벤트를 위로 쌓기
        console.log(`[${time}] ${message}`);
    }

    function addTodo() {
        const value = input.value;
        const li = document.createElement("li");
        li.textContent = value;
        list.appendChild(li);
        input.value = "";
        pushToStack(`[addTodo] '${value}' 추가됨`);
    }

    let isComposing = false;
    input.addEventListener("compositionstart", () => {
        isComposing = true;
        pushToStack("compositionstart");
    });

    input.addEventListener("compositionend", (e) => {
        isComposing = false;
        pushToStack(`compositionend → data: '${e.data}'`);
    });

    input.addEventListener("compositionupdate", (e) => {
        pushToStack(`compositionupdate → data: '${e.data}'`);
    })


    input.addEventListener("keydown", (e) => {
        if (e.key === "Enter") {
            if (isComposing === true) {
                pushToStack(`OS가 조합 중인 글자를 확정 짓는 이벤트는 예외처리: isComposing = ${isComposing}`);
                return;
            }
            pushToStack(`조합 완료된 글자 (${input.value})만 Todo에 반영: isComposing = ${isComposing}`);
            addTodo();
        }
    });
</script>
</body>
</html>

문제 원인 자세히 살펴보기

IME 문제

한글 (정확히는 조합형 글자)을 입력하고 Enter 키를 누르면 마지막 글자가 한 번 더 입력되는 문제가 발생한다.

Enter 키로 입력을 완료하는 기능 구현 시, 글자 조합 과정과 맞물려 2가지의 이벤트가 수행된다.

1. Keydown 이벤트: js로 등록해둔 keydown 이벤트가 수행된다.
2. 글자 조합 확정 이벤트: IME는 확정된 문자열을 커밋(commit)한다.

"사용자가 입력한 문자열"과 "확정된 문자열 커밋"이라는 두 이벤트로 인해 위 예시처럼 (한국어) + (어)가 발생하게 된 것이다.

조합 중인 글자?

한국어는 자모 (字母)를 조합해서 하나의 글자로 만드는 "조합형 글자"이다.

조합 중인 글자는 사용자의 키 입력 순서대로 compositionstart -> compositionupdate -> compositionend 순으로 이벤트가 발생한다.

그리고 조합된 글자는 일반적으로 텍스트 커서(Text Cursor)가 다음 글자로 넘어가는 순간이라고 이해하면 된다 (위 표에서 4-1,4-2). 위 상황 외에도 Enter키를 누르면 OS가 내부적으로 해당 문자를 확정 짓는다 (이로인해 IME 문제가 발생하게 된다).

실제 웹에서 확인해보기

reference

• MDN Web Docs: CompositionEvent 설명 (Link)
• 중국어 IME 조합: 조합 과정에서 onChange가 발생하는 문제 (Link)

기존 REST API의 문제점

오버패치 (over-fetching)

오버 패치는 영어 문장 그대로 불필요한 데이터까지 같이 가져오는 문제를 말한다. 어느 부분에서는 name, email 값만 필요하더라도 /users/1/을 요청하면 "주소, 생일, 가입일" 등의 불필요한 데이터까지 가져오게 된다.

문제점

1. 네트워크 낭비: 불필요한 데이터까지 전송하기 때문에 데이터 양이 커지게 되고, 이로 인해 네트워크 성능이 느려질 가능성이 있다
2. 메모리 낭비: 전송 받은 데이터는 우선 메모리에 저장되기 때문에 메모리를 차지한다.
3. 클라이언트 리소스 낭비: 받아온 데이터 중 필요한 데이터를 추출하는 과정에서의 렌더링 비용 (CPU 연산, 메모리 사용) 증가 가능성이 있다.

받아온 데이터 중 필요한 것만 추려야 해서 메모리 사용량 증가, 렌더링 비용 증가 가능성이 있음

언더패치 (under-fetching)

언더 패치도 문장 그대로 여러 종류의 데이터가 필요한데, /users/1, /users/1/friends처럼 여러 번 요청 해야 필요한 정보를 다 채울 수 있는 문제를 말한다.

문제점

1. 다수의 HTTP 요청: 하나의 기능을 위해 여러 번의 API 요청을 보내야 한다. 이로 인해 지연 (latency) 발생 가능성이 있음.
2. 요청 순서 의존성 발생: API 요청 시, 요청 순서에 의존된다면 비동기 흐름의 복잡성이 증가될 수 있다.
   • /posts/123 (123번 게시글 가져오기)
   • /users/45 (45번 작성자 가져오기) 123번 게시글이 로드되어야 45번 작성자 정보를 사용할 수 있는 상황과 같이 순서 의존이 발생할 수 있음

Graph QL

쿼리

기존 RestAPI 방식과 다르게 GraphQL은 하나의 엔드포인트에서 요청마다 원하는 필드만 선택적으로 받아올 수 있다.

qeury {  # User 타입에서 name만 가져옴
    user {
        name
    }
}

query {  # User 타입에서 name과 friends 필드의 name 필드를 가져옴
    user {
        name
          friends {
            name
        }
    }
}

리졸버

리졸버는 GraphQL 스키마의 각 필드에 데이터를 채워주는 함수이다. (default resolver가 사전에 정의되어 있으며, 가공이 필요한 경우엔 커스텀 리졸버를 제작하여 사용 가능)

// 커스텀 리졸버
const resolvers = {
    Query: {
        numberSix() {
            return 6;
        }
    }
}

// 쿼리
query {
    numberSix
}

// 반환 결과
{
    "data": {
        "numberSix": 6
    }
}

위의 예시는 numberSix 필드에 대해 요청이 들어온 경우, 커스텀 리졸버를 통해 6을 반환받도록 한 예시이다.

새 창에 데이터를 보내는 로직은 아래와 같다 (get 파라미터, chrome storage 사용 X)

1. 새 창을 만든 후, 해당 탭의 id값을 가져온다. (chrome.tabs.create)
2. 1번에서 가져온 탭의 id 값을 통해 웹 페이지가 로드 되었는지 확인한다. (chrome.tabs.onUpdated)
3. 웹 페이지가 로드 되었다면, 해당 탭으로 데이터를 보낸다. (chrome.tabs.sendMessage)
4. onMessage 함수를 통해 데이터를 받아온다. (chrome.runtime.onMessage)

2번은 꼭 필요한 과정은 아니지만, 타이밍 이슈가 있을 수 있기 때문에 해주면 안정성이 좋아진다.

예시 코드

// background.js

// 1. 새창 열기
chrome.tabs.create({ url: "https://www.naver.com/" }, (tab) => {  
    const tabId = tab.id;  

    // 2. 웹 페이지 로드 확인
    chrome.tabs.onUpdated.addListener(function listener(updatedTabId, info) {  
        if (updatedTabId === tabId && info.status === "complete") {  
            // 3. 데이터 전송
            chrome.tabs.sendMessage(tabId, {  
                data: "데이터 전송",
                payload: "payload 전송",  
            });  

            // 업데이트를 계속 감지하지 않도록 방지
            chrome.tabs.onUpdated.removeListener(listener);  
        }  
    });  
});

// content script

// 4. 데이터 받아오기
chrome.runtime.onMessage.addListener((msg, sender, sendResponse) => {  
    console.log(msg.data)
    console.log(msg.payload)
});

#!/bin/bash

SESSION_NAME="<session_name>"

compare_version="<kernel_version>"
kernel_version=$(uname -r)

result_path="dstat"
file_name="<result_filename>"

generate_unique_filename() {
        local base_name="$1"
        local extension="csv"
        local count=1

        local new_filename="${result_path}/${base_name}_${count}.${extension}"
        while [ -e "$new_filename" ]; do
                count=$((count + 1))
                new_filename="${result_path}/${base_name}_${count}.${extension}"
        done

        echo $new_filename
}

if [[ "$kernel_version" == *"$compare_version"* ]]; then
        result_path+="/proposed"
else
        result_path+="/original"
fi
mkdir -p ${result_path}

unique_filename=$(generate_unique_filename $file_name)
echo $unique_filename


cur_path=$(pwd)
tmux new-session -d -s $SESSION_NAME "dstat --output ${cur_path}/${unique_filename}"

# test code
# do something
sleep 2

tmux send-keys -t $SESSION_NAME C-c
sleep 1

tmux kill-session -t $SESSION_NAME # 혹시 모르니 한번 더 종료

command

환경 구축

환경 구축을 위한 command 목록은 아래와 같다.

# docker ubuntu 22.04 만들기
$ docker image pull ubuntu:22.04

# docker 실행
$ docker run -dit --name <container_name> <docker_image>

# 파일 전송
$ docker cp <host_path> <container_name>:/<container_path>

# container를 이미지로 만들기
$ docker commit <container_name> <repository:tag>

# 외부에서 명령어 실행
$ docker exec -it <container_name> <command>

container 상태 정보 보기

# docker image 확인
$ docker images

# container 확인
$ docker ps -a

# container 내부 접속
$ docker exec -it /bin/bash

script

create_containers.sh

아래의 스크립트는 docker container를 원하는 개수만큼 만들어주는 스크립트이다.

use case는 다음과 같다.

1. container가 이미 만들어져 있으며 running 중인 경우, 무시
2. container가 이미 만들어져 있으나 running이 아닌 경우, container start
3. conatiner가 만들어져 있지 않은 경우, <docker_image>에 대해 container를 만든 뒤 start

usgae: ./create_containers <container_num>

#!/bin/bash

# 제작할 container 개수
MAX_INDEX=$1
IMAGE_NAME="<docker_image>"  # 사용하려는 Docker 이미지 이름 설정

declare -A pid_array # 백그라운드 프로세스의 pid를 담는 배열

for i in $(seq 1 "$MAX_INDEX"); do
    CONTAINER_NAME="<container_name>$i" # $i는 지우지 말 것

    # 컨테이너가 존재하는지 확인
    CONTAINER_ID=$(docker ps -a --filter "name=^/${CONTAINER_NAME}$" --format "{{.ID}}")

    if [ -n "$CONTAINER_ID" ]; then
        if [ "$(docker ps --filter "name=^/${CONTAINER_NAME}$" --format "{{.ID}}")" ]; then
            # case1. 컨테이너가 이미 실행 중
            echo "컨테이너 $CONTAINER_NAME 가 이미 실행 중입니다."
        else
               # case2. 컨테이너가 존재하지만 실행 중이지 않다면 시작
            echo "컨테이너 $CONTAINER_NAME 이 존재하지만 실행 중이 아닙니다. 실행 시키겠습니다."
            docker start "$CONTAINER_NAME" &                                                                                                                                                                                                                                             
        fi  
    else
        # case3. 컨테이너가 존재하지 않으므로 새로 생성
        # background 프로세스로 진행되게 하여 제작 속도를 높임
        echo "컨테이너 $CONTAINER_NAME 이 없습니다. 새로 생성합니다."
        docker run -dit --name "$CONTAINER_NAME" "$IMAGE_NAME" /bin/bash &
        pid_array[$CONTAINER_NAME]=$!
    fi  
done

for CONTAINER_NAME in "${!pid_array[@]}"; do
        PID=${pid_array[$CONTAINER_NAME]}

        wait "$PID"
        echo "컨테이너 $CONTAINER_NAME의 생성이 완료되었습니다. "
        docker start "$CONTAINER_NAME" &
done

wait

exec_containers.sh

해당 script는 만들어진 container들에 대해 명령을 순차적으로 보내는 script이다.

tmux는 dstat 결과를 저장하기 위해 추가한 것으로 dstat 결과가 필요하지 않은 경우엔 tmux 관련된 부분을 지우면 된다.

usage: ./exec_containers.sh <container_num>

#!/bin/bash

SESSION_NAME="docker_stress"
RUN_CONTAINER=$1

echo "== ${RUN_CONTAINER} container =="

        tmux new-session -d -s $SESSION_NAME "dstat -D total --output <dstat_result_path>"
        for i in $(seq 1 $RUN_CONTAINER); do
                docker exec <container_name>$i <some_work> & 
        done
        wait

        tmux send-keys -t $SESSION_NAME C-c 
        sleep 1

        tmux kill-session -t $SESSION_NAME
echo "========= [done] =========="

분기 예측 (branch prediction)은 CPU의 성능을 높이기 위해 등장한 것으로 조건문의 결과를 예측하는 식으로 진행된다.

이해를 돕기위해 아래와 같은 코드가 있다고 해보자

if (condition) {
    A() // condition의 결과가 True일 경우, A() 실행
} else {
    B() // condition의 결과가 False일 경우, B() 실행
}

이러한 코드에 대해 분기 예측이 적용된 CPU는 condition의 결과를 예측하여 A() 또는 B()를 실행시킨다. 만약 예측이 맞으면 성능은 크게 향상될 것이다.

그러나 예측이 틀린 경우엔 해당 결과를 되돌리고 (undo), 올바른 경로를 다시 실행하게 된다.

스펙터 버그

스펙터 버그는 이러한 분기 예측의 취약점을 이용한 공격이다. 등장 배경에서 예측이 틀린 경우엔 결과를 되돌린다고 했었지만, 실제 코드가 실행된 데이터는 캐시에 남게 된다.

이에 대해 공격자는 아래와 같은 방식으로 취약점 공격을 할 수 있다. (스펙터 버그)

스펙터 취약점 공격 방법

1. 조건문의 결과를 참으로 예측하도록 유도한다. (조건이 참인 상황을 여러 번 반복한다.)
2. 공격자가 알아내고자 하는 값에 접근한다.
3. 분기 예측으로 해당 코드가 실행이 되고, 조건이 false라면 명령이 취소된다.
4. 명령이 취소가 되었을지라도 그 데이터는 캐시에 남게 된다.
5. 공격자는 여러 데이터에 접근하면서 접근 시간을 확인한다. 그 중 접근 시간이 짧은 데이터는 캐시에 저장되어 있다는 뜻이므로 공격자가 알아내고자 한 데이터임을 알 수 있다.

reference

• https://parksb.github.io/article/31.html

CPU가 매번 메모리에 접근하여 데이터를 가져오는 것은 느리기 때문에 지역성의 특성을 이용해 CPU 캐시에 데이터를 추가로 가져온다. (추가로 가져오는 데이터의 단위는 캐시 라인이며, 대부분의 CPU는 64 bytes로 구성되어 있다.)

이와 관련한 그림으로 아래의 그림을 많이 보았을 것이다.

그러나 병렬 처리 환경에서는 CPU cache으로 인해 오히려 성능이 낮아질 때도 있다. (false sharing 문제)

False sharing

False sharing은 "거짓 공유"의 문제로 실제로 쓰레드간 공유되지 않은 데이터이지만, 동일한 캐시 라인의 데이터를 마치 공유하는 것처럼 인식하여 성능 저하를 일으키는 문제를 말한다.

이에 대하여 병렬처리 환경에서 false sharing 문제가 어떻게 발생되는지 그림으로 확인하면 이해가 쉽다.

False sharing 문제는 메모리가 연속적 (정확히는 한 캐시 라인에 두 데이터가 포함) 일 때, 나타나는 문제이기 때문에 배열, 구조체 등에서 자주 발생된다.

False sharing으로 인한 성능 저하 테스트

#include <iostream>
#include <thread>
#include <chrono>

#define FALSE_SHARING
// #define RNAD_TEST
// #define __DEBUG

#define TEST_CNT 100
#define ITER_CNT 1000000

#ifdef FALSE_SHARING
struct Info {
    volatile int num1;
    volatile int num2;
} info;
volatile long long num3 = 0;
#else
struct Info {
    volatile long long num1 = 0;
    alignas(64) volatile long long num2 = 0; // cache line의 범위를 벗어나도록 64 bytes 만큼 padding을 추가함 (align)
} info;
alignas(64) volatile long long num3 = 0;
#endif

void fun1() {
    for (long long i = 0; i < ITER_CNT / 2; i++)
#ifndef RNAD_TEST
        info.num1 += 1;
#else
        info.num1 += rand();
#endif
}

void fun2() {
    for (long long i = 0; i < ITER_CNT / 2; i++)
#ifndef RNAD_TEST
        info.num2 += 1;
#else
        info.num2 += rand();
#endif
}

void fun3() {
    for (long long i = 0; i < ITER_CNT; i++)
#ifndef RNAD_TEST
        num3 += 1;
#else
        num3 += rand();
#endif
}

std::chrono::duration<double> test(bool is_multi_test) {
    auto beginTime = std::chrono::high_resolution_clock::now();

    if (is_multi_test) {
        std::thread t1(fun1);
        std::thread t2(fun2);

        t1.join(); t2.join();
    }
    else {
        fun3(); //Single Thread 실행
    }

    auto endTime = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> resultTime = endTime - beginTime;

#ifdef __DEBUG
    std::cout << "-------[single]-------\n";
    std::cout << "total value: " << num3 << std::endl;
    std::cout << "excution time: " << resultTime.count() << std::endl;
#endif
    return resultTime;
}

int main() {
    double single_total = 0, multi_total = 0;

    for (int i = 0; i < TEST_CNT; i++) {
        info.num1 = 0;
        info.num2 = 0;
        num3 = 0;
        single_total += test(false).count() / TEST_CNT;
        multi_total += test(true).count() / TEST_CNT;
#ifdef __DEBUG
        std::cout << "---------------------\n";
        std::cout << std::endl;;
#endif
    }

    std::cout << "single test excution time: " << single_total << std::endl;
    std::cout << "multi test excution time:  " << multi_total << std::endl;
}

=== [False sharing test] ===
single test excution time: 0.00161374
multi test excution time:  0.00260395 (false sharing으로 인해 병렬 처리 환경임에도 더 오래걸림)

=== [ False sharing Mitigated ] ===
single test excution time: 0.00177046 (false sharing 문제를 해결하면 single thread가 더 느림을 확인 할 수 있음)
multi test excution time:  0.00119301

example

in linux

리눅스에서도 false sharing 문제를 회피하기 위한 코드들을 확인할 수 있다.

in structure

struct page_counter {
    /*
     * Make sure 'usage' does not share cacheline with any other field. The
     * memcg->memory.usage is a hot member of struct mem_cgroup.
     */
    atomic_long_t usage;
    CACHELINE_PADDING(_pad1_);

    /* effective memory.min and memory.min usage tracking */
    unsigned long emin;
    atomic_long_t min_usage;
    atomic_long_t children_min_usage;

    /* effective memory.low and memory.low usage tracking */
    unsigned long elow;
    atomic_long_t low_usage;
    atomic_long_t children_low_usage;

    unsigned long watermark;
    unsigned long failcnt;

    /* Keep all the read most fields in a separete cacheline. */
    CACHELINE_PADDING(_pad2_);

    unsigned long min;
    unsigned long low;
    unsigned long high;
    unsigned long max;
    struct page_counter *parent;
} ____cacheline_internodealigned_in_smp;

위는 리눅스에서 사용되는 page_counter 구조체이다. 주석을 보면 page_counter 구조체의 usage 멤버 변수가 hot member (자주 사용되는 변수)라고 설명되어 있다. 이 usage 변수가 다른 멤버 변수들과 같은 캐시 라인에 위치하게 되면, false sharing 문제로 인해 성능이 심각하게 저하될 수 있다. 그래서 이를 방지하기 위해 CACHELINE_PADDING을 추가하여 다른 멤버 변수들과 같은 캐시라인에 들어오지 않도록 하고 있다.

reference

• https://smallake.kr/?p=2271
• https://elky84.github.io/2010/08/19/false_sharing/
• https://ypangtrouble.tistory.com/entry/%EC%BA%90%EC%8B%9C-%EC%9D%BC%EA%B4%80%EC%84%B1%EA%B3%BC-%EA%B1%B0%EC%A7%93-%EA%B3%B5%EC%9C%A0

📚 해설 및 코드

✏️ 문제 접근

해당 문제는 테스트 케이스마다 주어지는 입력값의 크기가 크지 않으므로 (최대 자릿수: 6, 최대 교환 횟수: 10) DFS를 이용한 브루트 포스로 해결할 수 있다.

하지만 문제를 그대로 해석하여 정렬로도 풀 수 있다.

정렬 풀이

해당 문제를 정렬로 풀 때에는 정렬의 한 step의 결과가 최대가 되도록 하는 것이 중요하다

정렬 풀이 예외 케이스

정렬 풀이로 풀게 되면 예외 케이스가 발생하게 된다. (output를 보면 알겠지만, 문제의 답이 정렬의 결과랑 다르기 때문이다)

1) 숫자 카드에 중복된 값이 있을 경우

해당 예외 케이스는 정렬의 한 step 당 결과가 최대가 되도록 하는 것에 집중한다면 문제가 되지 않을 수 있다.

문제의 예제를 보도록 하자

input 숫자 카드: 3, 2, 8, 8, 8 &nbsp&nbsp | &nbsp&nbsp교환 횟수: 2

step1: 8, 2, 8, 8, 3 step2: 8, 8, 8, 2, 3 answer: 8, 8, 8, 3, 2 (wrong!)

위의 예제처럼 그리디 시점으로 봤을 때 정렬의 각 step이 최대가 되기 위해선 8 중에서도 제일 마지막에 위치한 값과 맨 앞의 카드를 바꾸면 된다. 그러나 answer의 값을 보면 그렇지 않을 것을 확인할 수 있다.

그렇기 때문에 정렬 풀이로 진행할 때는 다음 문장의 차이를 주의해야 한다.

각 step에 대해 최선을 고르는 것이 아닌, step의 결과가 최대가 나오도록 해야 한다.

2) 정렬을 수행하고도 교환 횟수가 남았을 경우

해당 예외 케이스는 정렬 수행 후의 남은 교환 횟수에 대해 어쩔 수 없이 끝의 두 자리를 swapping을 해야 하기 때문에 발생하는 예외 케이스이다.

남은 교환 횟수가 홀수라면: 끝의 두 자리를 교환해야만 한다. 남은 교환 횟수가 짝수라면: 정렬 수행 결과를 반환한다.

추가로 주의해야 할 점이 있다. 숫자 카드 중에 중복된 값이 있다면, 중복된 값끼리 교환하면 되므로 해당 예외를 고려하지 않아도 된다.

📑 코드

#include <iostream>
#include <vector>
#include <string>
#include <set>

using namespace std;


int main() {
    int T;
    cin >> T;

    for (int i = 1; i <= T; i++) {
        string num;
        int max_change_cnt;
        cin >> num >> max_change_cnt;

        // 1. 선택 정렬 수행
        int cur_change_cnt = 0;
        bool exist_same_val = false;
        vector<int> is_changed(num.length()); // 어느 숫자 카드로 인해 스왑되었는지를 저장
        for (int i = 0; i < num.length() - 1; i++) {
            int max_idx = i;

            for (int j = i + 1; j < num.length(); j++) {
                if (num[max_idx] <= num[j]) {
                    max_idx = j;
                }

                if (num[i] == num[j]) {
                    exist_same_val = true;
                }
            }

            // 현재 step에서 이미 최대값인 경우
            if (max_idx == i)
                continue;

            cur_change_cnt++;
            is_changed[max_idx] = num[max_idx]; // swap 된 결과를 기준으로 max_idx번 째는 num[i]의 값에 의해 swap 되었음을 의미
            swap(num[max_idx], num[i]);

            // 정렬 한 step의 결과가 최대가 되도록 하기 위해 추가 정렬 수행
            //  -> 추가 정렬시 is_changed 고려
            for (int j = i; j < num.length() - 1; j++) {
                if (!is_changed[j]) // swap 된 적이 없는 숫자 카드는, 추가 정렬을 수행해선 안됨
                    continue;

                max_idx = j;
                for (int k = i + 1; k < num.length(); k++) {
                    // 예외 1) 숫자 카드에 같은 값이 있을 경우에 문제가 됨
                    //   => 이를 판단하기 위해 is_changed를 사용
                    if (is_changed[j] == is_changed[k] && num[max_idx] < num[k]) {
                        max_idx = k;
                    }
                }

                if (max_idx == j)
                    continue;
                swap(num[max_idx], num[j]);
            }

            // 정렬 횟수를 모두 소모함
            if (cur_change_cnt == max_change_cnt)
                break;
        }

        // 2. 숫자카드 중 중복된 값이 없다면,
        //  2-1. 남은 횟수가 홀수라면 맨 끝 2개의 숫자 swap
        //  2-2. 남은 횟수가 짝수라면 무시
        if (!exist_same_val && (max_change_cnt - cur_change_cnt) % 2) {
            swap(num[num.length() - 1], num[num.length() - 2]);
        }
        cout << "#" << i << " " << stoi(num) << "\n";
    }

    return 0;
}