스레드와 동시성 Threads & Concurrency

struct tcb {
    tid_t        tid;             /* 스레드별 */
    void        *kstack;          /* 스레드별: 커널 스택 */
    cpu_context  ctx;             /* 스레드별: 저장된 레지스터·SP·PC */
    int          state;           /* RUNNING / READY / BLOCKED ... */
    int          prio;            /* 스케줄링 우선순위 */
    void        *tls;             /* 스레드 지역 저장소 베이스 */

    struct mm   *mm;              /* 공유: 주소 공간(페이지 테이블) — 같은 프로세스면 동일 포인터 */
    struct files *files;          /* 공유: 열린 파일 테이블 */
    struct signal *sig;           /* 공유: 시그널 핸들러 */
    struct tcb  *next_in_proc;    /* 같은 프로세스의 스레드 리스트 */
};

; void switch_to(cpu_context *old, cpu_context *new)
switch_to:
    ; --- 현재(old) 스레드 상태 저장 ---
    mov [rdi + CTX_RSP], rsp     ; 스택 포인터 저장
    mov [rdi + CTX_RBX], rbx     ; 콜리-세이브드(callee-saved) 레지스터들 저장
    mov [rdi + CTX_RBP], rbp
    ; ... r12~r15 등 ...

    ; --- 다음(new) 스레드 상태 복원 ---
    mov rsp, [rsi + CTX_RSP]     ; 스택을 B의 것으로 교체 (= 실행 흐름이 바뀜)
    mov rbx, [rsi + CTX_RBX]
    mov rbp, [rsi + CTX_RBP]
    ; ... 복원 ...
    ret                          ; B의 스택에 쌓인 반환 주소로 점프 → B가 이어서 실행

            1                                   1
speedup ≤ ─────────────        lim  speedup =  ───   (N → ∞)
          S + (1−S)/N           N→∞              S

speedup(N) = N − S·(N − 1)         (S = 직렬 비율, N = 코어 수)
            = S + N·(1 − S)

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

long sum;                      /* 전역 → 모든 스레드가 공유 */

void *runner(void *arg) {
    long upper = atol(arg);
    sum = 0;
    for (long i = 1; i <= upper; i++) sum += i;
    pthread_exit(NULL);        /* return 으로도 종료 가능 */
}

int main(int argc, char *argv[]) {
    pthread_t       tid;
    pthread_attr_t  attr;
    pthread_attr_init(&attr);                 /* 기본 속성: 스택 크기·스케줄링 등 */
    /* pthread_attr_setstacksize(&attr, 1<<20);  // 스택 1MB로 지정 가능 */

    pthread_create(&tid, &attr, runner, argv[1]);
    pthread_join(tid, NULL);                  /* 종료까지 블록, 자원 회수 */
    printf("sum = %ld\n", sum);
    pthread_attr_destroy(&attr);
}

#define _GNU_SOURCE
#include <sched.h>
#include <sys/mman.h>
#include <unistd.h>
#include <stdio.h>

static int worker(void *arg) {
    /* CLONE_THREAD이므로 getpid()는 부모와 같은 TGID(=PID) 반환. 개별 TID는 gettid() */
    printf("child task: 주소공간 공유, pid(TGID)=%d\n", (int)getpid());
    return 0;
}

int main(void) {
    const int STACK = 1 << 20;
    /* 자식 스택은 호출자가 마련한다 (스택은 아래로 자라므로 top을 넘김) */
    char *stack = mmap(NULL, STACK, PROT_READ|PROT_WRITE,
                       MAP_PRIVATE|MAP_ANONYMOUS|MAP_STACK, -1, 0);
    /* pthread_create가 내부적으로 켜는 플래그들: */
    int flags = CLONE_VM      /* 메모리 공간 공유  → "스레드"의 핵심 */
              | CLONE_FS       /* 파일시스템 정보 공유 */
              | CLONE_FILES    /* 열린 파일 테이블 공유 */
              | CLONE_SIGHAND  /* 시그널 핸들러 공유 */
              | CLONE_THREAD;  /* 같은 스레드 그룹(TGID) → 같은 PID로 보임 */
    clone(worker, stack + STACK, flags, NULL);
    /* (실제로는 여기서 futex 등으로 자식 종료를 기다린다) */
    sleep(1);
}

/* 0=unlocked, 1=locked, 2=locked&waiters */
void lock(atomic_int *m) {
    int c;
    if ((c = cmpxchg(m, 0, 1)) != 0) {          /* fast path: 경합 없으면 즉시 획득 */
        if (c != 2) c = xchg(m, 2);             /* 대기자 있음 표시 */
        while (c != 0) {
            futex(m, FUTEX_WAIT, 2, NULL);      /* slow path: 커널에서 잠듦 */
            c = xchg(m, 2);
        }
    }
}
void unlock(atomic_int *m) {
    /* fetch_sub는 '감소 전' 값을 반환. 1이었다면(대기자 없음) 이미 0이 되어 끝 */
    if (atomic_fetch_sub(m, 1) != 1) {          /* 1이 아니었음 = 2였음 → 대기자 있음 */
        store(m, 0);                            /* 0으로 만들고 */
        futex(m, FUTEX_WAKE, 1, NULL);          /* 대기자 하나 깨움 */
    }
}

#include <windows.h>
DWORD Sum;
DWORD WINAPI Summation(LPVOID p) {
    DWORD upper = *(DWORD*)p;
    for (DWORD i = 1; i <= upper; i++) Sum += i;
    return 0;
}
int main(int argc, char *argv[]) {
    DWORD id, param = atoi(argv[1]);
    HANDLE h = CreateThread(NULL, 0, Summation, &param, 0, &id);
    WaitForSingleObject(h, INFINITE);   /* 여러 개면 WaitForMultipleObjects */
    CloseHandle(h);
}

import java.util.concurrent.*;

class Summation implements Callable<Long> {
    private final long upper;
    Summation(long upper) { this.upper = upper; }
    public Long call() {                 // 별도 스레드에서 실행
        long s = 0;
        for (long i = 1; i <= upper; i++) s += i;
        return s;                        // Runnable과 달리 결과 반환 가능
    }
}
public class Driver {
    public static void main(String[] a) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(4);
        Future<Long> f = pool.submit(new Summation(Long.parseLong(a[0])));
        System.out.println("sum = " + f.get());   // 결과 준비될 때까지 블록
        pool.shutdown();
    }
}

import java.util.concurrent.*;

class SumTask extends RecursiveTask<Long> {
    static final int THRESHOLD = 10_000;
    final int[] a; final int lo, hi;
    SumTask(int[] a, int lo, int hi){ this.a=a; this.lo=lo; this.hi=hi; }
    protected Long compute() {
        if (hi - lo <= THRESHOLD) {              // 충분히 작으면 직접
            long s = 0; for (int i=lo;i<hi;i++) s += a[i]; return s;
        }
        int mid = (lo + hi) >>> 1;                // 아니면 분할 후 fork
        SumTask L = new SumTask(a, lo, mid);
        SumTask R = new SumTask(a, mid, hi);
        L.fork();                                // 비동기 제출 (다른 워커가 훔쳐갈 수 있음)
        long r = R.compute();                    // 현재 스레드가 직접 (work-first)
        return r + L.join();                     // L 결과 합류
    }
}
// ForkJoinPool.commonPool().invoke(new SumTask(arr, 0, arr.length));

#include <thread>
#include <future>
#include <numeric>
#include <vector>
#include <print>   // C++23

long partial_sum(long lo, long hi) {
    long s = 0; for (long i = lo; i <= hi; ++i) s += i; return s;
}
int main() {
    // (1) std::async: 결과를 future로 받는 가장 간단한 작업 병렬
    std::future<long> f = std::async(std::launch::async, partial_sum, 1, 500'000);
    long here = partial_sum(500'001, 1'000'000);
    std::println("sum = {}", here + f.get());

    // (2) std::jthread (C++20): 소멸 시 자동 join + 협력적 취소 토큰
    std::jthread t([](std::stop_token st){
        while (!st.stop_requested()) { /* work */ }   // 깔끔한 취소
    });
    // t 소멸 시 request_stop() + join() 자동 호출 → RAII 안전
}

#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>

class ThreadPool {
    std::vector<std::thread>            workers;
    std::queue<std::function<void()>>   tasks;
    std::mutex                         m;
    std::condition_variable            cv;
    bool                               stop = false;
public:
    explicit ThreadPool(size_t n) {
        for (size_t i = 0; i < n; ++i)
            workers.emplace_back([this]{
                for (;;) {
                    std::function<void()> job;
                    {
                        std::unique_lock lk(m);
                        cv.wait(lk, [this]{ return stop || !tasks.empty(); });
                        if (stop && tasks.empty()) return;     // 종료
                        job = std::move(tasks.front()); tasks.pop();
                    }
                    job();                                     // 락 밖에서 실행 (중요)
                }
            });
    }
    void submit(std::function<void()> job) {
        { std::lock_guard lk(m); tasks.push(std::move(job)); }
        cv.notify_one();                                       // 워커 하나 깨움
    }
    ~ThreadPool() {
        { std::lock_guard lk(m); stop = true; }
        cv.notify_all();
        for (auto &w : workers) w.join();
    }
};

// 소유자 전용 (bottom 끝, 단일 스레드) — 락 없음
push(task):  buf[bottom] = task; bottom++          // (메모리 배리어 필요)
pop():       bottom--;
             if (bottom < top) { bottom = top; return EMPTY }
             t = buf[bottom];
             if (bottom == top)                      // 마지막 1개: 도둑과 경쟁
                 if (!CAS(&top, top, top+1)) t = EMPTY
             return t
// 도둑들 (top 끝, 여러 스레드) — CAS로 경쟁
steal():     t = top;
             if (top >= bottom) return EMPTY
             task = buf[t];
             if (!CAS(&top, t, t+1)) return ABORT     // 다른 도둑이 이김 → 재시도
             return task

#include <omp.h>
double dot(const double *a, const double *b, int n) {
    double sum = 0.0;
    /* 루프를 스레드들에 분할 + 부분합을 안전하게 reduction */
    #pragma omp parallel for reduction(+:sum) schedule(static)
    for (int i = 0; i < n; i++)
        sum += a[i] * b[i];
    return sum;
}
/* #pragma omp parallel  → 코어 수만큼 스레드 생성
   reduction(+:sum)       → 스레드별 사적 sum을 마지막에 합산(레이스 없음)
   schedule(static|dynamic|guided) → 반복 분배 정책 */

// 동시 큐(userInitiated)에 작업을 던지고, 끝나면 메인 큐에서 UI 갱신
DispatchQueue.global(qos: .userInitiated).async {
    let result = heavyComputation()
    DispatchQueue.main.async {           // 직렬(메인) 큐 → UI 스레드 안전
        updateUI(result)
    }
}
// 직렬 큐: 순차 실행 보장 (락 없이 상태 보호 용도로도 씀)
let serial = DispatchQueue(label: "com.app.state")
serial.async { mutate(&sharedState) }

#include <tbb/parallel_for.h>
// 직렬: for (size_t i = 0; i < n; ++i) apply(v[i]);
// 병렬: range(0..n) × body(람다) 만 선언 — 청크 분할·스레드 매핑은 TBB가
tbb::parallel_for(size_t(0), n, [=](size_t i) { apply(v[i]); });

// 합산처럼 결합이 필요하면 parallel_reduce
double sum = tbb::parallel_reduce(
    tbb::blocked_range<size_t>(0, n), 0.0,
    [&](auto r, double acc){ for (size_t i=r.begin(); i<r.end(); ++i) acc += v[i]; return acc; },
    std::plus<double>());

// counter++ 는 사실 세 단계: load → add → store
//
//   Thread A           Thread B
//   r = load(counter)
//                      r = load(counter)   // 둘 다 같은 값을 읽음
//   store(counter, r+1)
//                      store(counter, r+1) // B가 A의 증가를 덮어씀 → 하나 유실
//
// 두 스레드가 1,000,000번씩 ++ 해도 결과는 2,000,000이 아니다(lost update).

#include <atomic>
std::atomic<long> counter{0};
counter.fetch_add(1, std::memory_order_relaxed);   // 원자적 RMW, 카운터엔 relaxed로 충분

// 또는 뮤텍스
std::mutex m;
{ std::lock_guard lk(m); ++plain_counter; }   // 임계 구역(critical section): 한 번에 한 스레드만 진입

// 초기: x = 0, y = 0
//
//   Thread 1          Thread 2
//   x = 1;            y = 1;
//   r1 = y;           r2 = x;
//
// SC라면 r1==0 && r2==0 은 불가능.
// 그러나 x86(TSO)에서도 store buffer 때문에 Store→Load가 재배열되어
//   r1==0 && r2==0  이 실제로 관측된다!  (각 코어가 자기 store를
//   버퍼에 담아두고, 상대의 store가 아직 안 보이는 채로 load)
//
// 해결: 두 store 뒤에 메모리 배리어(MFENCE / atomic_thread_fence(seq_cst))

std::atomic<bool> ready{false};
int data = 0;                                   // 일반 변수

// Producer
data = 42;                                      // (1) 평범한 쓰기
ready.store(true, std::memory_order_release);   // (2) release: (1)이 (2)보다 먼저 보이도록 못박음

// Consumer
while (!ready.load(std::memory_order_acquire))  // (3) acquire
    ;
assert(data == 42);                             // 보장됨! (2)→(3) 동기화로 (1)이 보임

// 나쁨: 두 카운터가 같은 라인 → 코어 간 라인 핑퐁
struct Bad { std::atomic<long> a; std::atomic<long> b; };   // 16B, 한 라인

// 좋음: 각 카운터를 캐시 라인 경계로 정렬/패딩 → 독립
struct Good {
    alignas(64) std::atomic<long> a;
    alignas(64) std::atomic<long> b;     // 다른 라인 → 핑퐁 없음
};
// C++17: std::hardware_destructive_interference_size 로 라인 크기 질의 가능

// prepare: 부모가 fork 직전에 모든 락 획득
// parent : fork 직후 부모가 락 해제
// child  : fork 직후 자식이 락 (재)초기화
pthread_atfork(prepare_acquire_locks,
               parent_release_locks,
               child_reinit_locks);

sigset_t set;
sigemptyset(&set); sigaddset(&set, SIGINT); sigaddset(&set, SIGTERM);
pthread_sigmask(SIG_BLOCK, &set, NULL);   // main과 모든 자식이 블록(상속됨)

void *sig_thread(void *arg) {
    int sig;
    for (;;) {
        sigwait(&set, &sig);               // 동기 수신 — 핸들러 제약 없음
        printf("received signal %d, shutting down\n", sig);
        graceful_shutdown();
    }
}

void *worker(void *arg) {
    pthread_cleanup_push(free, buf);          // 취소 시 자원 해제 보장
    while (running) {
        do_chunk();
        pthread_testcancel();                 // 취소 요청 확인 = 취소 지점
    }
    pthread_cleanup_pop(1);                    // 정상 종료 시에도 정리
    return NULL;
}
// 취소 요청: pthread_cancel(tid);  (요청일 뿐, 실제 종료는 대상이 결정)

_Thread_local int tx_id;          // C11 표준 키워드
__thread       int tx_id;          // gcc/clang 확장 (동일)

// Pthreads 동적 TLS (라이브러리 코드에서 키 생성)
pthread_key_t key;
pthread_key_create(&key, destructor);   // 스레드 종료 시 destructor 호출
pthread_setspecific(key, ptr);
void *p = pthread_getspecific(key);

// Java:  static final ThreadLocal<Integer> txId = ThreadLocal.withInitial(() -> 0);
// C#:    [ThreadStatic] static int txId;   /  ThreadLocal<int>

func worker(id int, jobs <-chan int, results chan<- int) {
    for j := range jobs {                 // 채널이 닫힐 때까지
        results <- j * j                   // 결과를 채널로 통신 (락 불필요)
    }
}
func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    for w := 1; w <= 4; w++ { go worker(w, jobs, results) }   // 4 goroutine
    for j := 1; j <= 100; j++ { jobs <- j }
    close(jobs)
    for a := 1; a <= 100; a++ { <-results }
}

// 요청마다 가상 스레드 하나 — 100만 개도 OK (각 ~수백 바이트)
try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
    for (int i = 0; i < 1_000_000; i++) {
        executor.submit(() -> {
            var data = socket.read();    // 블로킹처럼 보이지만 OS 스레드는 안 막음
            return process(data);
        });
    }
}   // 동기 코드의 모습, 비동기의 확장성

// Rust — async fn은 Future(상태 기계)를 반환, .await로 합성
async fn handle(sock: TcpStream) -> io::Result<()> {
    let req = sock.read().await?;       // 여기서 yield 가능
    let resp = process(req).await?;
    sock.write(resp).await
}
#[tokio::main]                          // 멀티스레드 work-stealing executor
async fn main() { /* tokio::spawn(handle(...)) ... */ }

# Python — asyncio: 단일 스레드 이벤트 루프 + 코루틴
async def handle(reader, writer):
    data = await reader.read(1024)      # 이벤트 루프에 제어 양보
    writer.write(process(data))
    await writer.drain()