import threading
import time
from typing import List, Dict, Any, Optional

def worker(name: str, sleep_time: float = 2.0) -> None:
    """
    기본 작업자 스레드 함수
    
    Args:
        name: 작업 식별 이름
        sleep_time: 대기 시간(초)
    """
    print(f"스레드 {name} 시작")
    time.sleep(sleep_time)
    print(f"스레드 {name} 종료")

# 스레드 생성 및 시작
thread = threading.Thread(target=worker, args=("Worker-1",))
thread.daemon = True  # 메인 스레드 종료시 함께 종료
thread.start()

# 스레드 정보 확인
print(f"스레드 ID: {thread.ident}")
print(f"스레드 이름: {thread.name}")
print(f"스레드 활성 상태: {thread.is_alive()}")

thread.join()  # 스레드가 종료될 때까지 대기

import threading
import time
from typing import List, Iterable, Any

def count_numbers(name: str, numbers: Iterable[int]) -> None:
    """
    주어진 숫자 목록을 순회하면서 출력하는 스레드 함수
    
    Args:
        name: 스레드 이름
        numbers: 처리할 숫자 목록
    """
    for i in numbers:
        print(f"{name}: {i}")
        time.sleep(0.1)

class WorkerThread(threading.Thread):
    """
    Thread 클래스를 상속받는 사용자 정의 스레드 클래스
    """
    def __init__(self, name: str, numbers: Iterable[int]) -> None:
        """
        생성자
        
        Args:
            name: 스레드 이름
            numbers: 처리할 숫자 목록
        """
        super().__init__(name=name)
        self.numbers = numbers
        self.result: List[int] = []
        
    def run(self) -> None:
        """스레드 실행 시 호출되는 메서드"""
        print(f"{self.name} 시작")
        for num in self.numbers:
            self.result.append(num * 2)  # 각 숫자를 2배로 처리
            time.sleep(0.1)
        print(f"{self.name} 종료")
        
    def get_result(self) -> List[int]:
        """처리 결과 반환"""
        return self.result

# 함수 기반 다중 스레드 실행
threads: List[threading.Thread] = []
for i in range(3):
    thread = threading.Thread(
        target=count_numbers,
        args=(f"Thread-{i}", range(i*3, (i+1)*3))
    )
    threads.append(thread)
    thread.start()

# 클래스 기반 다중 스레드 실행
worker_threads: List[WorkerThread] = []
for i in range(3):
    worker = WorkerThread(f"WorkerThread-{i}", range(i*3, (i+1)*3))
    worker_threads.append(worker)
    worker.start()

# 모든 스레드 종료 대기
for thread in threads:
    thread.join()
    
# 작업 결과 수집
results: List[List[int]] = []
for worker in worker_threads:
    worker.join()
    results.append(worker.get_result())
    
print(f"처리 결과: {results}")

import threading
import time
from typing import List, Dict

class Counter:
    """스레드 안전한 카운터 클래스"""
    
    def __init__(self) -> None:
        self.value = 0
        self.lock = threading.Lock()
        
    def increment(self) -> None:
        """락을 사용한 안전한 증가 연산"""
        with self.lock:
            current = self.value
            time.sleep(0.1)  # 경쟁 조건 시뮬레이션
            self.value = current + 1
            
    def safe_update(self, func) -> None:
        """사용자 정의 업데이트 함수를 락으로 보호"""
        with self.lock:
            func(self)

# RLock 사용 예제 (재진입 가능 락)
class RecursiveCounter:
    """재진입 락을 사용하는 카운터"""
    
    def __init__(self) -> None:
        self.value = 0
        self.lock = threading.RLock()  # 재진입 가능한 락
        
    def increment(self, n: int = 1) -> None:
        """값 증가"""
        with self.lock:
            self.value += n
            
    def increment_by_10(self) -> None:
        """10씩 증가 (내부적으로 increment 호출)"""
        with self.lock:  # 같은 스레드에서 이미 획득한 락 재진입
            for _ in range(10):
                self.increment()  # 락을 이미 보유한 상태에서 호출

# 기본 락 사용 예제
counter = Counter()

def increment_task() -> None:
    """카운터 증가 작업"""
    for _ in range(3):
        counter.increment()

threads: List[threading.Thread] = [
    threading.Thread(target=increment_task) for _ in range(5)
]

for thread in threads:
    thread.start()
    
for thread in threads:
    thread.join()
    
print(f"최종 카운터 값: {counter.value}")  # 예상 결과: 15

# 재진입 락 사용 예제
recursive_counter = RecursiveCounter()
recursive_counter.increment_by_10()
print(f"재진입 락 카운터 값: {recursive_counter.value}")  # 예상 결과: 10

import threading
import time
import queue
from typing import List, Dict, Any, Optional, Deque
from collections import deque

# 조건 변수를 사용한 생산자-소비자 패턴
class ThreadSafeQueue:
    """조건 변수를 사용한 스레드 안전 큐"""
    
    def __init__(self, max_size: int = 5) -> None:
        self.queue: Deque[Any] = deque()
        self.max_size = max_size
        self.condition = threading.Condition()
        
    def put(self, item: Any) -> None:
        """아이템 추가"""
        with self.condition:
            # 큐가 가득 찼으면 대기
            while len(self.queue) >= self.max_size:
                print(f"큐가 가득 참 (크기: {len(self.queue)}), 생산자 대기 중...")
                self.condition.wait()
                
            self.queue.append(item)
            print(f"아이템 추가: {item}, 큐 크기: {len(self.queue)}")
            
            # 소비자에게 알림
            self.condition.notify()
    
    def get(self) -> Any:
        """아이템 가져오기"""
        with self.condition:
            # 큐가 비었으면 대기
            while len(self.queue) == 0:
                print("큐가 비었음, 소비자 대기 중...")
                self.condition.wait()
                
            item = self.queue.popleft()
            print(f"아이템 가져옴: {item}, 큐 크기: {len(self.queue)}")
            
            # 생산자에게 알림
            self.condition.notify()
            return item

# 이벤트를 사용한 스레드 동기화
def wait_for_event(event: threading.Event, name: str) -> None:
    """이벤트 대기 함수"""
    print(f"{name}: 이벤트 대기 중...")
    event.wait()  # 이벤트가 설정될 때까지 대기
    print(f"{name}: 이벤트 발생! 작업 실행")

# 조건 변수 사용 예제
queue = ThreadSafeQueue(max_size=3)

def producer() -> None:
    """생산자 스레드"""
    for i in range(10):
        queue.put(f"아이템-{i}")
        time.sleep(0.5)

def consumer() -> None:
    """소비자 스레드"""
    for _ in range(10):
        item = queue.get()
        # 아이템 처리 시뮬레이션
        time.sleep(1)

# 생산자-소비자 스레드 시작
producer_thread = threading.Thread(target=producer)
consumer_thread = threading.Thread(target=consumer)

producer_thread.start()
consumer_thread.start()

# 이벤트 사용 예제
shutdown_event = threading.Event()

event_threads = [
    threading.Thread(target=wait_for_event, args=(shutdown_event, f"Worker-{i}"))
    for i in range(3)
]

for t in event_threads:
    t.start()

print("메인: 3초 후 이벤트 설정...")
time.sleep(3)
shutdown_event.set()  # 모든 대기 중인 스레드에게 신호 전송

# 모든 스레드 종료 대기
producer_thread.join()
consumer_thread.join()
for t in event_threads:
    t.join()

import threading
import time
import random
from typing import List, Dict, Set, Any

# 세마포어로 동시 접근 제한
class ConnectionPool:
    """세마포어를 사용한 데이터베이스 연결 풀"""
    
    def __init__(self, max_connections: int = 3) -> None:
        self.semaphore = threading.Semaphore(max_connections)
        self.connections: Set[int] = set()
        self.lock = threading.Lock()
        self.connection_count = 0
        
    def acquire_connection(self) -> int:
        """연결 획득"""
        with self.semaphore:
            # 임계 영역 내에서 연결 생성
            with self.lock:
                self.connection_count += 1
                connection_id = self.connection_count
                self.connections.add(connection_id)
                print(f"연결 획득: {connection_id}, 활성 연결: {len(self.connections)}")
            
            return connection_id
            
    def release_connection(self, connection_id: int) -> None:
        """연결 반환"""
        with self.lock:
            if connection_id in self.connections:
                self.connections.remove(connection_id)
                print(f"연결 반환: {connection_id}, 활성 연결: {len(self.connections)}")

# BoundedSemaphore 사용 예제
class ResourceManager:
    """정확한 리소스 관리를 위한 바운디드 세마포어"""
    
    def __init__(self, resources: int = 5) -> None:
        # BoundedSemaphore는 최대 카운트를 초과하여 release하면 ValueError 발생
        self.semaphore = threading.BoundedSemaphore(resources)
        self.resource_count = resources
        
    def use_resource(self, worker_id: int) -> None:
        """리소스 사용"""
        with self.semaphore:
            print(f"워커 {worker_id}: 리소스 획득")
            # 리소스 사용 시뮬레이션
            time.sleep(random.uniform(0.5, 1.5))
            print(f"워커 {worker_id}: 리소스 반환")

# 세마포어 사용 예제
pool = ConnectionPool(max_connections=3)

def worker(worker_id: int) -> None:
    """연결 풀을 사용하는 작업자"""
    connection_id = pool.acquire_connection()
    try:
        # 연결 사용 시뮬레이션
        print(f"워커 {worker_id}: 연결 {connection_id} 사용 중")
        time.sleep(random.uniform(1, 3))
    finally:
        # 항상 연결 반환
        pool.release_connection(connection_id)
        print(f"워커 {worker_id}: 작업 완료")

# 바운디드 세마포어 사용 예제
resource_mgr = ResourceManager(resources=2)

def resource_worker(worker_id: int) -> None:
    """리소스 관리자를 사용하는 작업자"""
    for _ in range(3):
        resource_mgr.use_resource(worker_id)
        time.sleep(random.uniform(0.1, 0.5))

# 스레드 생성 및 실행
semaphore_threads = [threading.Thread(target=worker, args=(i,)) for i in range(5)]
resource_threads = [threading.Thread(target=resource_worker, args=(i,)) for i in range(3)]

for thread in semaphore_threads + resource_threads:
    thread.start()

for thread in semaphore_threads + resource_threads:
    thread.join()

import concurrent.futures
import threading
import requests
import time
from typing import List, Dict, Any, Callable, TypeVar, Generic

T = TypeVar('T')  # 제네릭 타입 변수

def download_url(url: str) -> Dict[str, Any]:
    """URL에서 데이터 다운로드"""
    print(f"다운로드 시작: {url} (스레드: {threading.current_thread().name})")
    start_time = time.time()
    response = requests.get(url)
    data = response.text
    elapsed = time.time() - start_time
    return {
        'url': url,
        'status': response.status_code,
        'size': len(data),
        'time': elapsed,
        'thread': threading.current_thread().name
    }

def process_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """다운로드한 데이터 처리"""
    print(f"데이터 처리 시작: {data['url']} (스레드: {threading.current_thread().name})")
    # 처리 시간 시뮬레이션
    time.sleep(0.5)
    data['processed'] = True
    return data

class ThreadPoolManager(Generic[T]):
    """ThreadPoolExecutor 관리 클래스"""
    
    def __init__(self, max_workers: int = None) -> None:
        self.max_workers = max_workers
        
    def map_tasks(self, func: Callable[..., T], tasks: List[Any]) -> List[T]:
        """작업 매핑 및 병렬 실행"""
        results: List[T] = []
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # submit 메서드로 작업 제출 및 Future 객체 수집
            future_to_task = {executor.submit(func, task): task for task in tasks}
            
            # as_completed로 완료된 작업부터 결과 수집
            for future in concurrent.futures.as_completed(future_to_task):
                try:
                    result = future.result()
                    results.append(result)
                except Exception as exc:
                    task = future_to_task[future]
                    print(f"작업 {task} 실행 중 오류 발생: {exc}")
        
        return results
    
    def process_tasks(self, tasks: List[Any], task_func: Callable[[Any], T]) -> List[T]:
        """작업 처리 with progress callback"""
        results: List[T] = []
        completed = 0
        total = len(tasks)
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            # submit 메서드로 작업 제출
            futures = [executor.submit(task_func, task) for task in tasks]
            
            # as_completed로 완료된 작업부터 결과 수집
            for future in concurrent.futures.as_completed(futures):
                try:
                    result = future.result()
                    results.append(result)
                    
                    # 진행 상황 업데이트
                    completed += 1
                    print(f"진행 상황: {completed}/{total} ({completed/total*100:.1f}%)")
                    
                except Exception as exc:
                    print(f"작업 실행 중 오류 발생: {exc}")
        
        return results
    
    def chain_tasks(self, initial_tasks: List[Any], 
                   first_func: Callable[[Any], Any], 
                   second_func: Callable[[Any], T]) -> List[T]:
        """연쇄 작업 처리 (첫 작업 결과를 두 번째 작업의 입력으로 사용)"""
        intermediate_results = []
        final_results: List[T] = []
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as first_executor:
            # 첫 번째 단계 작업 제출
            first_futures = [first_executor.submit(first_func, task) for task in initial_tasks]
            
            # 첫 번째 작업 결과 수집
            for future in concurrent.futures.as_completed(first_futures):
                try:
                    result = future.result()
                    intermediate_results.append(result)
                except Exception as exc:
                    print(f"첫 번째 작업 실행 중 오류 발생: {exc}")
        
        with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as second_executor:
            # 두 번째 단계 작업 제출
            second_futures = [second_executor.submit(second_func, result) for result in intermediate_results]
            
            # 두 번째 작업 결과 수집
            for future in concurrent.futures.as_completed(second_futures):
                try:
                    result = future.result()
                    final_results.append(result)
                except Exception as exc:
                    print(f"두 번째 작업 실행 중 오류 발생: {exc}")
        
        return final_results

# ThreadPoolExecutor 사용 예제
urls = [
    "https://www.example.com",
    "https://www.google.com",
    "https://www.github.com",
    "https://www.python.org",
    "https://www.wikipedia.org"
]

# 1. 기본 ThreadPoolExecutor 사용
print("1. 기본 ThreadPoolExecutor 실행")
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # map 메서드로 간단하게 병렬 처리
    results = list(executor.map(download_url, urls))
    for result in results:
        print(f"다운로드 완료: {result['url']}, 크기: {result['size']}, 시간: {result['time']:.2f}초")

# 2. ThreadPoolManager 클래스 사용
print("\n2. ThreadPoolManager 클래스 실행")
pool_manager = ThreadPoolManager[Dict[str, Any]](max_workers=3)

# 연쇄 작업 - 다운로드 후 처리
print("\n2.1 연쇄 작업 실행")
final_results = pool_manager.chain_tasks(urls, download_url, process_data)
for result in final_results:
    print(f"처리 완료: {result['url']}, 처리됨: {result.get('processed', False)}")

print("\n2.2 단일 작업 진행률 추적")
download_results = pool_manager.process_tasks(urls, download_url)
print(f"모든 다운로드 완료. 총 {len(download_results)}개 URL 처리됨.")