セマフォでタスクの同時実行数制限

Transcript

1. セマフォで(並列)タスクの同時実行数制限 Hank Ehly (ハンク・イーリー) ENECHANGE株式会社 @hankehly (GitHub) qiita.com/hankehly
2. 問題 • 並列タスクの同時実行数を制御したい場合がある ◦ スクレイピング ◦ レート制限がかかっている API ◦ CPU/メモリーの負荷を抑えたい • ThreadPoolExecutor / ProcessPoolExecutor • 同時実行数を制御するオプションがない ◦ “一度に最大 N タスクまで実行したい ”
3. どんな工夫があるか 1. time.sleep while True: run_job() time.sleep(3) current_jobs = 0 max_jobs = 3 while True: if current_jobs < max_jobs: run_job() current_jobs -= 1 while True: if acquire_lock(): run_job() release_lock() 2. グローバルなカウンター変数 3. ロックオブジェクト ◦ redis ◦ multiprocessing.Lock ◦ 空っぽのファイル
4. セマフォとは（Wiki） セマフォとは、並列プログラミング環境での複数の実行単位が共有する資源にアクセスするのを制御する 際の、便利な抽象化を提供する変数または抽象データ型である。 つまり… • セマフォは、限られた容量のリソースへのアクセスを制御するために使う • “リソース”は「獲得」した後に「解放」しないといけないもの • セマフォは「獲得/解放の回数」を監視する “Semaphore” (セマフォ) を知っていただきたい
5. Semaphore と BoundedSemaphore - Semaphore - BoundedSemaphore sem = threading.Semaphore(1) print(sem._value) # 1 sem.acquire() # Trueを返す print(sem._value) # 0 sem.acquire(timeout=5) # 5秒経過後にFalseを返す sem.release() print(sem._value) # 1 sem.release() print(sem._value) # 2 セマフォが「獲得回数と解放回数」を見てくれる
6. Semaphore と BoundedSemaphore - Semaphore - BoundedSemaphore - 有限セマフォとも sem = threading.BoundedSemaphore(1) print(sem._value) # 1 sem.acquire() # Trueを返す print(sem._value) # 0 sem.acquire(timeout=5) # 5秒経過後にFalseを返す sem.release() print(sem._value) # 1 sem.release() Traceback (most recent call last): File "<stdin>", line 1, in <module> File "/ (省略) /threading.py", line 504, in release raise ValueError("Semaphore released too many times") ValueError: Semaphore released too many times 下限値に加えて上限値もある。 “解放”し過ぎないようにしたい時に使う
7. 【実例】スクレイピング def scrape_url(url): res = requests.get(url) s3_bucket.put(res.content, key=”...”) urls = [“http://netkeiba.com”, ...] sem = BoundedSemaphore(3) with ThreadPoolExecutor() as pool: while urls: if sem.acquire(): try: urls = url.pop() task = pool.submit(scrape_url, url) task.add_done_callback(lambda _: sem.release()) except: sem.release() else: logging.debug("他のタスク待ち..") コード例：qiita.com/hankehly
8. 【実例】スクレイピング def scrape_url(url): res = requests.get(url) s3_bucket.put(res.content, key=”...”) urls = [“http://netkeiba.com”, ...] sem = BoundedSemaphore(3) with ThreadPoolExecutor() as pool: while urls: if sem.acquire(): try: urls = url.pop() task = pool.submit(scrape_url, url) task.add_done_callback(lambda _: sem.release()) except: sem.release() else: logging.debug("他のタスク待ち..") コード例：qiita.com/hankehly
9. 【実例】スクレイピング def scrape_url(url): res = requests.get(url) s3_bucket.put(res.content, key=”...”) urls = [“http://netkeiba.com”, ...] sem = BoundedSemaphore(3) with ThreadPoolExecutor() as pool: while urls: if sem.acquire(): try: urls = url.pop() task = pool.submit(scrape_url, url) task.add_done_callback(lambda _: sem.release()) except: sem.release() else: logging.debug("他のタスク待ち..") コード例：qiita.com/hankehly
10. 終わりに 並列プログラミングで とあるタスクを “同時に N 個まで” 制限する必要があった場合 セマフォでエレガントに解決できるかも！