プロセス、スレッド、非同期、並行、並列を理解する #Linux

プロセスやスレッド、そして並行処理や並列処理といったプログラムの処理方式について理解が浅かったため整理する。

並行処理や並列処理を技術的に正しく理解しようと思った場合、プロセスやスレッドに関する理解が不可欠であるため、一つの文脈状で整理していく。

プロセスとスレッドを理解する

参考

スレッドとプロセスの違いを完全に理解する

3.1 スレッドの仕組みと操作（マルチスレッド、並行処理、並列処理、非同期呼び出し、スレッドセーフ、同期化、デッドロック、ウェイトセット、スレッドローカルなど）～Java Advanced編 - Qiita

プロセス

プロセスとは、OSレベルで見た際の実行中のプログラムのインスタンスである。

よく、「実行中のプログラムのインスタンス」という解説がされる場合も多いが、例えばRuby等でインスタンスをnewした時に生成されるようなインスタンスとは異なるため、留意すること。

プロセスの実態としては、以下のようなリソースが挙げられる。

メモリ空間:
- コードセグメント: 実行されるプログラムの命令コードを含みます。
- データセグメント: 静的変数やグローバル変数を含みます。
- ヒープ（Heap）: 動的に割り当てられるメモリ領域で、mallocやnewなどで確保されるメモリが含まれます。
- スタック（Stack）: 関数呼び出し時のローカル変数や戻りアドレスなどを格納します。関数の呼び出しと戻りに対応してサイズが変動します。
CPUリソース:
- プログラムカウンタ: 現在実行中の命令のアドレスを保持します。
- レジスタ: プロセスの実行状態を保持するためのCPUレジスタ。汎用レジスタ、インデックスレジスタ、スタックポインタなどが含まれます。
ファイルディスクリプタ:
- オープンファイル: プロセスが開いているファイルのリスト。ファイルディスクリプタを通じてアクセスされます。
- ソケット: ネットワーク通信に使用されるリソース。プロセスが通信を行うためのソケットもファイルディスクリプタとして管理されます。
入出力リソース:
- 標準入力/出力/エラー（stdin, stdout, stderr）: デフォルトの入出力ストリーム。

もう少しわかりやすく言うと、「アプリケーションの実行単位」とか言ってもよいかもしれない。

スレッド

マルチスレッドと並行処理をわかりやすく説明します - フラミナル

スレッドは、プロセス内で処理が実行される単位を指す。

基本的には、アプリケーションレベルで実装したプログラムがスレッドによって実行されるような認識でいるとよい。

分かりやすい例として考えると、何らかのHTTPリクエストを受け付け、レスポンスを返却するようなアプリケーションをGoやRubyで実装した場合、その処理はスレッド上で実行される。

単一のスレッドに複数のリクエストが同時に到達するようなシナリオでは、同期的に実装している場合スレッドは逐次的に処理を進めていくため、FIFOのような形でリクエストをキューイングして順次処理していくような挙動になる。

コンテキストスイッチ

コンテキストスイッチは、現在実行しているプロセスやスレッドから、異なるプロセスやスレッドに実行主体を切り替えるOSの動作を指す。

では、具体的にコンテキストの中にはどのような情報が含まれるのかというと、以下のような情報が含まれる。

プログラムカウンタ
スタックポインタ
フラグレジスタ

非同期、並行、並列を理解する。

参考

その並列処理待った！　「Python 並列処理」でググったあなたに捧ぐasync, threading, multiprocessingのざっくりとした説明 - Qiita

非同期処理 / Asynchronise

非同期処理とは、**「ある処理の完了を待つ事なく、異なるタスクを実行する」**事を指す。

主にI/O系の操作を行っている際の待ち時間を活用して、別の処理を行うようなユースケースが挙げられる。

I/O操作の例

I/Oとは - IT用語辞典
1. ファイルI/O
  - ファイル読み込み、書き込み
  - ファイルの削除、移動、コピー
2. ネットワークI/O
  - HTTPリクエストとレスポンス
  - ソケット通信（TCP/UDP）
  - データのストリーミング
3. データベースI/O
  - クエリの実行
  - データの挿入、更新、削除
  - トランザクション管理
4. デバイスI/O
  - キーボード入力
  - マウスの操作
  - スキャナーやカメラからのデータ取得
  - プリンターへのデータ送信
5. メモリI/O
  - メモリマップトファイルの読み書き
  - 共有メモリの操作

何かしらの Input を与えて何かしらの Output が返却されるまでの時間が長い場合、同期的に処理を行ってしまうとOutputが返ってくるまで処理が止まってしまう。

待ち時間が秒単位で発生する場合は、I/Oの処理を待っている間に別の操作を非同期で実行することで、処理を素早く行うことが可能になる。

例えば、外部のAPIにリクエストを投げて、レスポンスが返ってくるまでの間に別の処理を進めるようなケースを再現してみる。

クライアント側プログラム

import aiohttp
import asyncio
import time

async def fetch(session, url):
    async with session.get(url) as response:
        result = await response.json()
        return result

async def main():
    async with aiohttp.ClientSession() as session:
        task1 = asyncio.create_task(fetch(session, 'http://127.0.0.1:5001/task1'))
        task2 = asyncio.create_task(fetch(session, 'http://127.0.0.1:5002/task2'))

        tasks = [task1, task2]

        start_time = time.time()
        end_time = start_time + 5  # カウントを5秒間続ける

        async def monitor_tasks():
            while tasks:
                done, _ = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
                for task in done:
                    tasks.remove(task)
                    result = await task
                    print(f"Task completed with result: {result}")

        async def count_work():
            count = 0
            while time.time() < end_time:
                print(f"Doing other work: {count}")
                count += 1
                await asyncio.sleep(0.5)

        await asyncio.gather(monitor_tasks(), count_work())

        print(f"Total time taken: {time.time() - start_time:.2f} seconds")

asyncio.run(main())

サーバー側プログラム

＃ 同じようなサーバーをポートを変えていくつか用意する。

from flask import Flask, jsonify
import time
import random

app = Flask(__name__)

@app.route('/task1')
def task1():
    delay = random.uniform(0.1, 5.0)  # 0.1秒から5.0秒の間でランダムな遅延
    print(f"Task 1 delay: {delay:.1f} seconds")  # デバッグメッセージ
    time.sleep(delay)
    return jsonify({"message": f"Task 1 completed in {delay:.1f} seconds"})

if __name__ == '__main__':
    app.run(port=5001)

上記のプログラムでは、リクエストを投げてレスポンスを待っている間に標準出力を行っているだけなのでいまいちメリットが感じづらいが、例えばこれを非同期を使わず同期的に処理してみる。

クライアントプログラム

import requests
import time

def fetch(url):
    response = requests.get(url)
    result = response.json()
    print(result)  # タスクが完了したら即座に結果を出力
    return result

def main():
    urls = [
        'http://127.0.0.1:5001/task1',
        'http://127.0.0.1:5002/task2'
    ]

    start_time = time.time()

    for url in urls:
        fetch(url)
        
    for i in range(10):
        print(f"Doing other work: {i}")
        time.sleep(0.5)

    print(f"Total time taken: {time.time() - start_time:.2f} seconds")

if __name__ == '__main__':
    main()