【Python / ソケット通信】マルチプロセッシングでサーバーの処理を並列化

このページでは、マルチプロセッシングを利用してソケット通信を行うサーバーの処理の並列化を実現していきたいと思います。言語としては Python を利用します。

サーバーの処理を並列化すると何が嬉しい？

まず、なぜサーバーの処理を並列化すると嬉しいのか？という点について説明していきます。

サーバーやクライアントの意味合いについては下記ページで解説していますが、サーバーはサービスを提供するプログラムであり、クライアントはサーバーに対してリクエストを送信してサービスを利用するプログラムとなります。

Pythonでのソケット通信（ポート番号・プロトコル・サーバー / クライアント）

並列化されていないとレスポンス速度が低下

サーバーはクライアントからのリクエスト（何らかの要求）を受け取ると、サーバーが提供するサービス＆リクエストの内容に応じた処理を行い、その結果をレスポンスとしてクライアントに対して返却するような作りになっていることが多いです。

また、このサーバーとクライアントの関係において、サーバーとクライアントの数的な関係は１対多、もしくは少数対多数という関係になります。要は、サーバーの数に比べてクライアントの数の方が多いということになります。

なので、サーバーは多数のクライアントから同時にリクエストを受け取る可能性があります。その時に、処理が並列化されていないサーバーの場合、基本的には同時に受け取ったリクエストを１つ１つ逐次的に処理していくことになります。つまり、この場合、サーバーは同時に１つの処理しか実行できません。

そのため、この場合は特定のクライアントからのリクエストを処理している間、他のクライアントからリクエストを受け取ることもできないですし、他のクライアントに対してレスポンスが返却できないことになります。こうなると、サーバーに対してリクエストを送信しても応答がないため、クライアントを利用しているユーザーはサーバーに対して「レスポンスが遅い」という印象を持ち、せっかく画期的なサービスを開発したとしても、ユーザーからは使いにくいサービスと評価されて人気がなくなってしまう可能性があります…。

スポンサーリンク

並列化によってレスポンス速度が向上！

この問題を解決する１つの手段が並列化となります。

並列化を取り入れれば、プログラムは同時に１つのみの処理だけでなく、複数の処理を同時に実行できるようになります。

したがって、サーバーの処理を並列化しておけば、特定のクライアントからのリクエストを受け付けて処理を行っている間も、他のクライアントからのリクエストの受け付けや、そのリクエストに対する処理を並列的に同時に実行することが可能となります。

そのため、複数のクライアントが同じサーバーに同時にリクエストを送信した場合でも、それらのクライアントが待たされることが減ります。

このようにサーバーの処理を並列化をしておくことで、クライアントへのレスポンス速度を向上させ、ノンストレスで使い勝手の良いサービスを提供することができるようになります。

処理が並列化されていないサーバーの動作確認

実際に、処理が並列化されていないサーバーの Python スクリプトを実行し、複数のクライアントからリクエストを受け付けたときにどのような動作になるのかを確認していきたいと思います。

並列化されていないサーバーのスクリプト

まず、今回動作確認に使用する「処理が並列化されていないサーバー」のスクリプトの例は下記となります。

import socket
import time

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

sock.bind(('0.0.0.0', 40001))

sock.listen(5)

# 無限ループで常駐させる
while True:

    # 接続要求の受付 / 接続の確立
    e_sock, addr = sock.accept()

    accepted_time = time.time()
    print(f'{accepted_time}:{addr}からの接続要求受付完了')

    # データの受信待ち
    data = e_sock.recv(1024)
    
    # サービス・受信データに応じた処理
    str_data = data.decode()
    upper_str_data = str_data.upper()

    # 10秒間待ち
    time.sleep(10)

    # 処理結果の送信
    send_data = upper_str_data.encode()
    e_sock.send(send_data)

    responded_time = time.time()
    print(f'{responded_time}:{addr}へのレスポンスの返却完了')
   
    e_sock.close()

sock.close()

このスクリプトは TCP 通信を行うサーバーで、クライアントから受信した文字列を大文字に変換するサービスを提供するサーバーとなっています。ちょうど、下記ページの TCP 通信のサーバー の節で紹介したものをベースとした、非常に簡単なサーバーの例となります。

Pythonでのソケット通信（ポート番号・プロトコル・サーバー / クライアント）

ただし、上記ページの TCP 通信のサーバー に対して、ところどころ print で時刻（time.time() の返却値）を出力するようにし、さらに send メソッドを実行する前に time.sleep(10) で 10 秒間スリープするようにしています。このスリープは、サーバーの処理に時間がかかることを模擬するために実行しています。サーバーの処理に 10 秒かかるようなケースは少ないと思いますが、極端に長い時間スリープさせた方が並列化するメリットが感じやすいと思います。

スポンサーリンク

動作確認用のクライアントのスクリプト

次に、先ほど示したサーバーの通信相手となるクライアントのスクリプト例を示します。そのスクリプトが下記になります。

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

input_str_data = input('文字列を入力してください : ')


# 接続要求の送信
address = ('127.0.0.1', 40001)
sock.connect(address)

# データの送信
data = input_str_data.encode()
sock.send(data)

# データの受信
data = sock.recv(1024)
str_data = data.decode()

print(str_data)

sock.close()

これは下記ページの TCP 通信のクライアント の節で紹介したスクリプトと同じクライアントの例となりますので、詳細な解説が読みたい方は下記ページの TCP 通信のクライアント を参照していただければと思います。

Pythonでのソケット通信（ポート番号・プロトコル・サーバー / クライアント）

並列化されていないサーバーの動作確認

次に 並列化されていないサーバーのスクリプト と 動作確認用のクライアントのスクリプト を使って処理が並列化されていないサーバーの動作の確認を行っていきます。

まず、並列化されていないサーバーのスクリプト で紹介したスクリプトを no_parallel_server.py というファイル名で、 動作確認用のクライアントのスクリプト で紹介したスクリプトを client.py というファイル名で、適当なフォルダに保存してください。

続いて、ターミナルアプリ（Windows の場合はコマンドプロンプトや Power Shell など）を５つ開き、この５つ全てのターミナルアプリで、先ほどスクリプトをファイル保存したフォルダに移動してください。

この５つのターミナルアプリのうち、１つは no_parallel_server.py 実行用で、残りの４つは client.py 実行用として利用します。４つの client.py を実行することで、複数のクライアントからサーバーがリクエストを受け付ける動作を模擬していきます。

ということで、フォルダを移動したら、１つ目のターミナルアプリで下記のコマンドを実行してサーバーを起動します。

python no_parallel_server.py

ここからは、出来るだけ素早くアプリの操作とコマンドの実行を行っていく必要があるので注意してください。何度でもやり直せるので、失敗したらやり直していただければ良いですし、ここに記載されている結果だけ確認していただくのでも別に良いです。

では、つづいて、２つ目のターミナルアプリで下記のコマンドを実行します。これによりクライアントが起動し、最初に文字列の入力が促されるので、ここで適当な英数字を入力してください。

python client.py
文字列を入力してください : a

同様に、次は３つ目のターミナルアプリでも同じコマンド実行を行います。入力する文字列は、先ほど入力したものとは異なるものにしてください。

python client.py
文字列を入力してください : b

これと同じことを、４つ目と５つ目のターミナルアプリでも行います。そして５つ目のターミナルアプリで入力した文字列を大文字に変換した結果が出力されるまでしばらく待ってください。

５つ目のターミナルアプリに大文字に変換した結果が出力されたら、１つ目のターミナルアプリに出力されている時刻や文字列、つまりサーバーの出力結果に注目してみましょう。

私の場合、サーバーが出力した文字列は下記のようになっていました。各クライアントを起動するタイミング等によって結果は異なるものの、「先に接続要求を送信してきたクライアントへのレスポンスを返却しないと次のクライアントの接続要求の受付が行われない」という点は確認できるのではないかと思います。ちなみに、括弧内の２つ目の数値がクライアント側のソケットに関連付けられたポート番号になります。

1722293212.764506:('127.0.0.1', 58620)からの接続要求受付完了
1722293222.7732816:('127.0.0.1', 58620)へのレスポンスの返却完了
1722293222.7732816:('127.0.0.1', 58622)からの接続要求受付完了
1722293232.7745848:('127.0.0.1', 58622)へのレスポンスの返却完了
1722293232.7948654:('127.0.0.1', 58624)からの接続要求受付完了
1722293242.80623:('127.0.0.1', 58624)へのレスポンスの返却完了
1722293242.814341:('127.0.0.1', 58626)からの接続要求受付完了
1722293252.8232408:('127.0.0.1', 58626)へのレスポンスの返却完了

つまり、先に connect メソッドによって接続要求を送信してきたクライアントに対して accept を行った後に recv メソッド〜 close メソッドを実行している間、次に接続要求を送信してきたクライアントに対する accept は実行されません。

なので、クライアントはサーバーで accept が実行されるのを待つことになりますし、それに伴い、その後のサーバーでの recv メソッド以降の処理も遅れて実行されることになります。

そのため、下の図のように、複数のクライアントから同時にリクエストが送信されてきた場合に、最初に accept した接続要求を送信してきたクライアントを除き、他のクライアントを待たせてしまうことになります。

もちろん、今回はサーバー側で sleep を 10 秒実行していて処理時間が長いので、クライアントが待たされる時間も極端に長くなってしまっています。ですが、どんなサーバーでも処理時間は 0 ではないですし、クライアント側の状況によってデータの送受信に時間がかかる可能性もあるため、あらゆるサーバーで上記のような処理が起こりうると考えて良いです。

ここからは、上記のようなクライアントが待たされる問題の解決策について解説していきます。

サーバーの処理の並列化

ここまで説明してきたように、同時に複数のクライアントがリクエストを送信してきたときにクライアントが待たされてしまう問題の要因の１つは「サーバーが逐次的にしか処理が実行できない」という点にあります。

スポンサーリンク

処理の並列化の考え方

なので、逐次的ではなく、並列的に処理を行うようにすることで、上記のような問題を解決することができます。少なくとも、この待たされる時間をを軽減することが可能です。

この並列化を行う方法はたくさん存在するのですが、このページではマルチプロセッシングという技術を使ってサーバーの処理の並列化を行っていきたいと思います。

マルチプロセッシングとは、簡単に言うとプロセスを複数に分割する技術のことを言います。通常、プログラムは単一のプロセスとして動作するのですが、プログラム内の特定の処理（関数・メソッド）を別のプロセスに分離してやれば、プログラムが複数のプロセスとして実行されることになります。さらに、思い切って細かい説明は省略しますが、複数のプロセスが存在すれば、それらの複数のプロセスは別々の CPU コアによって並列的に同時に実行されることになります。

今回は、 並列化されていないサーバーのスクリプト で紹介したスクリプトにおいて、recv メソッド～ close メソッドの処理の部分を切り離して別の関数として定義し、この関数が accept とは別のプロセスとして処理されるようにします（この関数の関数名は service であるとして説明していきます）。

これにより、特に while ループ内の処理で考えれば、accept を実行するプロセスと service を実行するプロセスが別のプロセスとして存在することになり、これらが別々のCPU コアによって並列的に、かつ同時に実行されることになります。

さらに、accept を実行するたびに service を実行するプロセスを新規で生成するようにしてやれば、もともと存在している accept を実行するプロセスに加えて “通信相手となるクライアントの数だけのプロセス” が存在することになり、各プロセスが別々のコアによって実行されるため、理想的にはクライアントが待たされることが無くなります。

ただ、結局は CPU コアによってプロセスは処理されることになり、CPU コアの数は有限であるため、いずれは CPU コアの数が足りなくなり、サーバーが相手するクライアントが多くなるとクライアントは結局待たされることになります。ですが、それでも逐次的に各クライアントに対するリクエストを処理するよりかは、トータルで考えればクライアントに対するレスポンスが向上することになります。

ということで、詳細な説明に関しては省略させていただきましたが、とりあえずマルチプロセッシングを利用してプログラムを複数プロセスで処理させることで “処理の並列化” を実現可能であることは理解していただけたのではないかと思います。

処理の並列化の手順

次は、具体的な並列化の手順を説明していきます。

mutliprocessing モジュールの import

まず、このマルチプロセッシングは mutliprocessing モジュールによって提供される仕組みであるため、事前に multiprocessing モジュールを import しておく必要があります。この multiprocessing モジュールは Python の標準モジュールとなっているため、Python が利用できれば multiprocessing モジュールも利用可能なはずです。

import multiprocessing

並列化したい処理の関数化

続いて、別のプロセスで実行したい処理を関数化します。要は、元々のプログラムと並列に実行したい処理を１つの関数にまとめます。

今回は、while ループ内の処理における recv メソッド～ close メソッドの部分を別のプロセスで実行できるよう、これらの実行部分を関数化します。

そのために、まずは 並列化されていないサーバーのスクリプト で示したスクリプトの while ループ内の処理における recv メソッド～ close メソッドを実行している部分を削除し、

# 無限ループで常駐させる
while True:

    # 接続要求の受付 / 接続の確立
    e_sock, addr = sock.accept()

    accepted_time = time.time()
    print(f'{accepted_time}:{addr}からの接続要求受付完了')

続いて、service 関数を新たに定義し、先ほど削除した部分をそのまま service 関数内に記述します。recv メソッド～ close メソッドを実行するためには、accept メソッドの返却値である e_sock (接続確立後のソケット) と addr (接続確立先のクライアントの情報) が必要になるため、それらを引数として受け取るようにしています。今回の例の場合、addr に関しては print するために利用するだけなので実処理には必要ではありませんが、e_sock は recv や send を実行するために必須となります。

def service(e_sock, addr):
    # データの受信待ち
    data = e_sock.recv(1024)

    # サービス・受信データに応じた処理
    str_data = data.decode()
    upper_str_data = str_data.upper()

    # 10秒間待ち
    time.sleep(10)

    # 処理結果の送信
    send_data = upper_str_data.encode()
    e_sock.send(send_data)

    responded_time = time.time()
    print(f'{responded_time}:{addr}へのレスポンスの返却完了')

    e_sock.close()

プロセスの生成

続いて、並列化を実現するためにプロセスの生成を行います。このプロセスの生成は multiprocessing モジュールの Process クラスのインスタンスの生成および、そのインスタンスでの start メソッドの実行により実現できます。

まず、Process クラスのインスタンス生成について説明すると、このインスタンスは Process クラスのコンストラクタ、すなわち multiprocessing.Process() を実行することで生成できます。さらに、この multiprocessing.Process() の target 引数に、その生成されるプロセスで実行したい関数やメソッドのオブジェクトを指定します。さらに、args 引数や kwargs 引数で、その target 引数に指定した関数に渡したい引数を指定します。args 引数の場合はイテラブルなオブジェクト、kwargs 引数には辞書等を指定します。位置引数を指定するのであれば args、キーワード引数を指定するのであれば kwargs を指定すれば良いですし、両方指定するようなことも可能です。

process = multiprocessing.Process(
    target=関数, # 生成するプロセスで実行する関数
    args=リストやタプル, # 関数に渡す位置引数
    kwargs=辞書 # 関数に渡すキーワード引数
)

そして、Process クラスのインスタンスである multiprocessing.Process() の返却値に start メソッドを実行させれば、args 引数や kwargs 引数で指定したデータが引数として渡された状態で target 引数で指定した関数が実行されるプロセスが生成されることになります。そして、そのプロセスが CPU コアによって実行されることになります。

今回は、while ループ内の accept メソッド実行後にプロセスの生成を行います。そして、ここで生成するプロセスで実行する関数は先ほど作成した service 関数とし、さらに service 関数には accept メソッドの返却値である e_sock と addr を引数として渡すため、while ループ内の処理を下記のように変更することになります。 

# 無限ループで常駐させる
while True:

    # 接続要求の受付 / 接続の確立
    e_sock, addr = sock.accept()

    accepted_time = time.time()
    print(f'{accepted_time}:{addr}からの接続要求受付完了')

    # プロセスの生成
    process = multiprocessing.Process(
        target=service,
        kwargs={'e_sock':e_sock, 'addr':addr}
    )
    process.start()

このように処理を変更すれば、accept メソッドによってクライアントからの接続要求を受け付けた後に、multiprocessing.Process() と start メソッドが実行されて新たなプロセスが生成されることになります。さらに、start メソッドはプロセスを生成のみを行って終了しますので、すぐに while ループ先頭の accept が実行されることになり、次のクライアントからの接続要求をすぐに受け付けることが可能となります。

並行して、start メソッドが実行されることでプロセスが生成され、service 関数が実行されることになり、プロセス生成前に接続要求を受け付けたクライアントとの間でデータの送受信が recv メソッドや send メソッドにより行われることになります。

また、service 関数の処理が終了したら、生成されたプロセスは終了して消滅することになります。

マルチプロセッシングを利用せずに単に service 関数を呼び出すようにしてしまうと、service 関数が終了するまで、すなわち recv メソッド～ close メソッドまでの間 acceptが実行されないことになってクライアントが待たされてしまいますが、マルチプロセッシングを利用すれば accept と service 関数の実行を並列に実行することが可能となります。

サーバーの処理を並列化したプログラム

先ほどは並列化を行うためにポイントとなる変更部分のみのコードを紹介しましたが、ここでプログラム全体のスクリプトを紹介しておきたいと思います。

スポンサーリンク

処理の並列化を行ったスクリプト

サーバーの処理を並列化したプログラムのスクリプト例は下記のようになります。

import socket
import time
import multiprocessing

def service(e_sock, addr):
    # データの受信待ち
    data = e_sock.recv(1024)
    
    # サービス・受信データに応じた処理
    str_data = data.decode()
    upper_str_data = str_data.upper()

    # 10秒間待ち
    time.sleep(10)

    # 処理結果の送信
    send_data = upper_str_data.encode()
    e_sock.send(send_data)

    responded_time = time.time()
    print(f'{responded_time}:{addr}へのレスポンスの返却完了')
   
    e_sock.close()

if __name__ == '__main__':
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    sock.bind(('0.0.0.0', 40001))

    sock.listen(5)

    # 無限ループで常駐させる
    while True:

        # 接続要求の受付 / 接続の確立
        e_sock, addr = sock.accept()

        accepted_time = time.time()
        print(f'{accepted_time}:{addr}からの接続要求受付完了')

        process = multiprocessing.Process(
            target=service,
            kwargs={'e_sock':e_sock, 'addr':addr}
        )
        
        process.start()

    sock.close()

スクリプトの解説

上記スクリプトのポイントを説明しておきます。

処理の並列化の手順 で説明したように、accept が終了してクライアントと接続が確立されたタイミングで新たなプロセスを起動し、そのプロセスで service 関数が実行されることになります。

また、 上記スクリプトでは、ソケットを生成や while ループ等の各処理が __name__ == '__main__' が成立する場合しか実行されないようになっている点もポイントになります。__name__ == '__main__' の意味合いに関しては下記ページで説明しているので詳しく知りたい方は下記ページを参照していただければと思いますが、__name__ == '__main__' が成立する場合しか各処理が実行されないようになっているのは import 時にそれらの処理が実行されないようにするためです。

【Python】if __name__ == ‘__main__’ とは

どうも、プロセスを生成する時には import と同様の処理が行われてスクリプトが再読み込みされるようで、その時にソケット生成や bind が行われると余分にソケットが生成されたり例外が発生することになるので、各処理は __name__ == '__main__' が成立する場合のみ、つまり、直接このスクリプトが実行された時のみ行われるようにしています。

こういった注意点等もあるのですが、基本的には multiprocessing.Process() でのインスタンス生成と start メソッドの実行を行えば良いだけなので、簡単に処理の並列化が可能であることは感じ取っていただけるのではないかと思います。

ただし、実は処理の配列化を行う際には排他制御や同期制御などが必要になることが多く、これらの制御が必要になると一気にプログラミングの難易度が上がります。今回は、これらの制御が特に必要ないので簡単に処理が並列化を行うことができていますが、こういった制御が必要になる場合は注意が必要です。

また、生成するプロセスの数にも注意が必要です。上記の場合、accept が終了するたびにプロセスが生成されることになるため、クライアントからの接続要求が同時に大量に送信されてくると、その分生成されるプロセスの数も多くなります。そして、PC の CPU コアの数を上回る数のプロセスが生成されると処理量に対して CPU の仕事が間に合わなくなり、PC の処理が極端に重くなることになります。今回は、listen メソッドの引数で同時に受信可能な接続要求数を制限しているので問題ないと思いますが、実戦で処理の並列化を行う際はこのあたりも頭に入れて設計等を行うようにしましょう！

処理を並列化したサーバーの動作確認

最後に、処理を並列化したサーバーの動作確認を行っていきます。

ここでは、サーバーとして 処理の並列化を行ったスクリプト で紹介したスクリプトを利用し、クライアントとしては 並列化されていないサーバーの動作確認 での動作確認時と同様に、 動作確認用のクライアントのスクリプト で紹介スクリプトを利用します。

動作確認手順は 並列化されていないサーバーの動作確認 と同様で、まずサーバーを起動し、手際よく複数のクライアント（４つのクライアント）を起動して各クライアントに対して文字列を入力し、連続的に複数のクライアントからサーバーに対してリクエストを送信します。

それにより、サーバーを起動しているターミナルに時刻と文字列が出力されるので、それを確認してみましょう！私が試したところ、下記のような出力結果が得られました。

1722504235.7295523:('127.0.0.1', 63494)からの接続要求受付完了
1722504237.7655413:('127.0.0.1', 63497)からの接続要求受付完了
1722504241.0729225:('127.0.0.1', 63499)からの接続要求受付完了
1722504242.8379776:('127.0.0.1', 63502)からの接続要求受付完了
1722504245.9958396:('127.0.0.1', 63494)へのレスポンスの返却完了
1722504247.9743876:('127.0.0.1', 63497)へのレスポンスの返却完了
1722504251.2884297:('127.0.0.1', 63499)へのレスポンスの返却完了
1722504253.0687761:('127.0.0.1', 63502)へのレスポンスの返却完了

並列化されていないサーバーの動作確認 でのサーバーの出力結果と見比べていただければ並列化の効果が確認できると思います。

また、今回のサーバーやクライアントの処理の流れをシーケンス図で表すと下の図のようになります。

並列化されていないサーバーの動作確認 で利用したサーバーは処理を逐次的にしか実行できないため、特定のクライアントからの接続要求を受け付けてからレスポンスを返却するまでの間、他のクライアントからの接続要求の受付を行うことができず、その間クライアントが待たされてしまうという問題がありました。なので、複数のクライアントから接続要求が連続的に送信されてきたとしても、必ず特定のクライアントに対する「レスポンスの返却完了」が出力されてから次のクライアントに対する「接続要求受付完了」が出力されるようになっています。

ですが、今回利用したサーバーの場合、前のクライアントに対する処理を実行中でも次のクライアントの接続要求を受け取り、そのクライアントに対する処理を並列的に実行できるようになっています。そのため、複数のクライアントから接続要求が連続的に送信されてきた場合でも、クライアントが待たされることなくすぐに接続要求の受付が行われるようになっています。この結果、クライアントのユーザーからはレスポンスの早いサーバーという印象を持たせることができます。

スポンサーリンク

まとめ

このページでは、マルチプロセッシングを利用したサーバーの処理の並列化について解説しました！

サーバーは複数のクライアントを相手にするため、複数のクライアントからのリクエストを並列的に処理できるようにしておくことが重要です。でなければ、複数のクライアントから同時にリクエストを受け取ったような場合にクライアントが待たされることになり、レスポンス性能の悪いサーバーであるとユーザーに印象を持たれてしまう可能性が高いです。

で、処理の並列化自体はマルチプロセッシングを利用すれば簡単に実現可能です。処理の並列化を行うことで、プログラムの処理速度を向上させたり、同時に異なる複数の処理を実行させるようなことも可能になり、これを使いこなせば今まで実現できなかったこともプログラムで実現できるようになります！

もし、多数のクライアントと同時に通信を行うようなサーバーを開発するのであれば、今回説明した処理の並列化の採用を是非検討してみてください！