サーバーとマルチスレッド処理

シングルスレッドの問題点

前掲のEchoサーバーの実装は、単一のスレッドで全てのクライアントの処理を行っていました。 すなわち、クライアントからのメッセージを一件ずつシーケンシャルに処理していることになります。

もし、メッセージ受信時の処理の一つ一つに長い処理時間がかかるとするとどうなるでしょうか。 試しに、Echoサーバーのプログラムを以下のように修正し、レスポンスを送信する前に delay 関数で3秒間の待ち時間を加えてみます。

import processing.net.*;

Server myServer;

void setup() {
  myServer = new Server(this, 7);
}

void draw() {
  Client client = myServer.available();
  if (client != null) {
    String message = client.readString();
    delay(3000);
    client.write(message);
  }
}

1本のコネクションで接続する分には、レスポンスにかかる時間が3秒間伸びただけに思います。 しかし、サーバーは複数のコネクションを同時に処理していることに注意しましょう。 以下のようなクライアントプログラムを作成します。

import processing.net.*;

Client[] clients = new Client[3];

void setup() {
  clients[0] = new Client(this, "127.0.0.1", 7);
  clients[1] = new Client(this, "127.0.0.1", 7);
  clients[2] = new Client(this, "127.0.0.1", 7);

  for (int i = 0; i < clients.length; ++i) {
    clients[i].write("hello");
  }
}

void draw() {
  for (int i = 0; i < clients.length; ++i) {
    if (clients[i].available() > 0) {
      println("receive " + i + ": " +  clients[i].readString() + " (" + millis() + ")");
    }
  }
}

このプログラムでは、サーバーに対して3本のコネクションを確立し、メッセージの送受信を行います。 レスポンスの受信時には、受信したメッセージと同時にコネクションの番号と受信時間を表示します。 上記の3秒間の遅延を行うサーバープログラムを実行した後に、このクライアントプログラムを実行してみましょう。

実行結果は以下のようになるでしょう。

receive 0: hello (3283)
receive 1: hello (6301)
receive 2: hello (9301)

クライアントからサーバーへのメッセージ送信は setup 関数内でほぼ同時に行われますが、サーバーでの処理は以下のような順序で行われます。

1. コネクション0からメッセージを受信する
2. 3秒間待つ
3. コネクション0へレスポンスを送信する
4. コネクション1からメッセージを受信する
5. 3秒間待つ
6. コネクション1へレスポンスを送信する
7. コネクション2からメッセージを受信する
8. 3秒間待つ
9. コネクション2へレスポンスを送信する

そのため、各コネクションのレスポンスの間に3秒間の間隔が開きます。

今回は、何もせずに3秒間待つだけのプログラムですが、レスポンスを返すために何らかの重い処理が必要なサーバープログラムがあるとしたとき、一つのコネクションの処理が終わるまで次のコネクションを処理できないのは場合によっては非効率です。 マルチスレッドを用いた並列処理によってこの問題を解消します。

マルチスレッドの利用

Processingでのマルチスレッド処理にはいくつかの方法がありますが、ここでは Thread クラスのサブクラスを利用することにします。 Thread を継承したクラスを作成し、 run メソッドのオーバーライドします。

サーバープログラムは、コネクションが確立されるとそのコネクションを処理する専用のスレッドを立ち上げるようにしてみます。 コネクションを処理するクラスを Handler とし、コンストラクタでコネクションを受け取ります。 serverEvent ハンドラーで、コネクションが確立されると Handler クラスのインスタンスを生成し、別スレッドとして実行します。 Handler の処理は、これまでのサーバー処理と同様に、クライアントからのメッセージを受け取ったら3秒間待った後にレスポンスを返します。

import processing.net.*;

Server myServer;

void setup() {
  myServer = new Server(this, 7);
}

void draw() {
}

void serverEvent(Server server, Client client) {
  Handler handler = new Handler(client);
  handler.start();
}

class Handler extends Thread {
  Client client;

  Handler(Client client) {
    this.client = client;
  }

  void run() {
    while (this.client.active()) {
      if (this.client.available() > 0) {
        String message = this.client.readString();
        delay(3000);
        this.client.write(message);
      }
    }
  }
}

サーバープログラムを実行した後に、クライアントプログラムを実行しましょう。 実行結果は以下のようになります。

receive 0: hello (3304)
receive 1: hello (3304)
receive 2: hello (3304)

全てのコネクションがほぼ同時にレスポンスの受信をできています。

実用的なサーバープログラムでは、並列処理を用いたコネクション処理を取り入れると良いでしょう。 また、スレッド化することで、一つのコネクションに対する処理を分離しやすくなります。