Python multiprocessing 멈춤? daemonic·pickle 에러 해결

import multiprocessing as mp
import time


def producer(q: mp.Queue) -> None:
    for i in range(10_000):
        q.put(i)  # 소비가 느리면 여기서 블로킹
    q.put(None)


def consumer(q: mp.Queue) -> None:
    while True:
        item = q.get()
        if item is None:
            return
        time.sleep(0.01)  # 일부러 느리게


if __name__ == "__main__":
    q = mp.Queue(maxsize=100)  # 작게 잡으면 쉽게 재현
    p1 = mp.Process(target=producer, args=(q,))
    p2 = mp.Process(target=consumer, args=(q,))
    p1.start(); p2.start()
    p1.join(); p2.join()

import multiprocessing as mp


def worker(q: mp.Queue) -> None:
    for item in iter(q.get, None):
        # do work
        pass


if __name__ == "__main__":
    n_workers = 4
    q = mp.Queue()
    procs = [mp.Process(target=worker, args=(q,)) for _ in range(n_workers)]
    for p in procs:
        p.start()

    for i in range(1000):
        q.put(i)

    for _ in range(n_workers):
        q.put(None)  # 워커 수만큼

    for p in procs:
        p.join()

import multiprocessing as mp


def inner_task(x: int) -> int:
    return x * x


def outer_task(x: int) -> int:
    # (문제) 워커 프로세스 안에서 또 프로세스를 만들려 함
    with mp.Pool(2) as p:
        return sum(p.map(inner_task, [x, x + 1]))


if __name__ == "__main__":
    with mp.Pool(2) as p:
        print(p.map(outer_task, [1, 2, 3]))

import multiprocessing as mp
from concurrent.futures import ThreadPoolExecutor


def inner_io(x: int) -> int:
    return x + 1


def outer_cpu(x: int) -> int:
    with ThreadPoolExecutor(max_workers=8) as ex:
        return sum(ex.map(inner_io, [x] * 100))


if __name__ == "__main__":
    with mp.Pool(4) as p:
        print(p.map(outer_cpu, [1, 2, 3, 4]))

import multiprocessing as mp


if __name__ == "__main__":
    mp.set_start_method("spawn", force=True)
    # 이후 프로세스 생성

import multiprocessing as mp


def main() -> None:
    def f(x: int) -> int:  # 로컬 함수
        return x * 2

    with mp.Pool(2) as p:
        print(p.map(f, [1, 2, 3]))


if __name__ == "__main__":
    main()

import multiprocessing as mp


def f(x: int) -> int:
    return x * 2


def main() -> None:
    with mp.Pool(2) as p:
        print(p.map(f, [1, 2, 3]))


if __name__ == "__main__":
    main()

import multiprocessing as mp
from typing import Optional

_client: Optional[object] = None


def init_worker() -> None:
    global _client
    _client = object()  # 예: DB 연결, HTTP 세션 등을 여기서 생성


def task(x: int) -> int:
    # _client 사용
    return x * x


if __name__ == "__main__":
    with mp.Pool(processes=4, initializer=init_worker) as p:
        print(p.map(task, range(10)))

import multiprocessing as mp


def work(x: int) -> int:
    return x + 1


if __name__ == "__main__":
    with mp.Pool(2) as p:
        print(p.map(work, [1, 2, 3]))

import multiprocessing as mp


def init_worker() -> None:
    # 여기서 로거/세션/커넥션 초기화
    pass


def task(x: int) -> int:
    return x * 2


if __name__ == "__main__":
    mp.set_start_method("spawn", force=True)
    with mp.Pool(4, initializer=init_worker) as p:
        print(p.map(task, range(100)))

from queue import Empty
import multiprocessing as mp


def worker(q: mp.Queue) -> None:
    while True:
        try:
            item = q.get(timeout=5)
        except Empty:
            # 상태 로깅 후 continue/break
            continue
        if item is None:
            return

import multiprocessing as mp


def task(x: int) -> int:
    if x == 3:
        raise RuntimeError("boom")
    return x


def on_error(e: BaseException) -> None:
    print("worker error:", repr(e))


if __name__ == "__main__":
    with mp.Pool(2) as p:
        for i in range(5):
            p.apply_async(task, (i,), error_callback=on_error)
        p.close()
        p.join()

import multiprocessing as mp
from dataclasses import dataclass


@dataclass(frozen=True)
class Job:
    id: int
    payload: str


def init_worker() -> None:
    # 프로세스별 초기화(세션/커넥션/로거 등)
    pass


def run_job(job: Job) -> tuple[int, int]:
    # 순수 함수에 가깝게: 입력을 받아 결과만 반환
    return (job.id, len(job.payload))


def main() -> None:
    jobs = [Job(i, "x" * (i % 10)) for i in range(1000)]

    with mp.Pool(processes=4, initializer=init_worker) as p:
        for job_id, size in p.imap_unordered(run_job, jobs, chunksize=50):
            # 결과 처리
            pass


if __name__ == "__main__":
    mp.set_start_method("spawn", force=True)
    main()