プロセス間通信を非同期化する

１．はじめに

　Google Assistantから制御できるジュークボックスサーバーを製作しました。事前に用意されたリストに従って楽曲再生をするシンプルなものですが、作成にあたってはあんまり単純ではありませんでした。ただ、オーソドックスな方法を使っています、という話を続けていて、前回は子プロセスの非同期実行管理でしたが、今回は非同期プロセス間通信の話になります。
　下図では「Class IPC」が担う機能となります。

　非同期通信を行う必要性については前回の子プロセスの稿でも触れていますが、サーバーの応答を待つ（同期通信を行う）とGoogleAssistantとの中継ぎをするフルフィルメント（上図では「Smart Device Butler」）の応答が遅くなり、GoogleAssistant側でタイムアウトするためです。

２．概要

「Class IPC」の概要を下図にまとめます。古典的な機能しか持たせていませんが、簡単に説明しておきます。

作成したクラスには以下の機能があります。

1. セマフォ
2. シングルオーダー
3. オーダーアレイ

２．１．セマフォ

　①は「セマフォ」と呼ばれるものです。英語ではsemaphoreという、初見では読み間違えそうな名前です。今回はサーバーとクライアント（複数）が同時に読み書きすることによるトラブルを避けるためのフラグみたいな使い方をしています。変数を使ったフラグと違い、セマフォへのアクセスは排他的に行われ、誰かがセマフォを確保した時は、他の人は取れないように使っています（微妙な書き方をしているのは、セマフォには他の使い方もできるからです）。

２．２．シンプルな送信

②は本稿では「シングルオーダー」と呼ぶことにします。ルールとして、誰かが先に書き込みをしていたら他の人は書けず、誰かが読み取ったら消し込むようにします。これも微妙な書き方になりますが、ＰＨＰの関数としてメッセージング関数が似たような機能を提供しているのですがちょっとした制約があるため「Class IPC」では使っていません。理由については２．４．２．で後述します。

２．３．バッファリング

③は「オーダーアレイ」と呼ぶことにします。要するに書き込みバッファです。構造としては「シングルオーダー」（以下、「スロット」と呼ぶことにします）が幾つか並んでいるのですが、書き込みポインターと読み出しポインター（「ポインター」っていう言い方は古いんでしょうかね）を持っていて、書き込み側は空きスロットがあれば書き込み、読み出し側は書き込みポインタが進んでいれば読み出すようにします。要はＦＩＦＯバッファですが、スロットは１次元配列なので、最後まで書きこんでしまうとポインタはそれより先に進めません。ただ、読み込み済みのスロットであれば上書きしてしまって問題ないので、読み出しポインタを追い越さないようにして頭からまた書き込むようにします。いわゆる「リングバッファ」です。

２．４．ＰＨＰでの実装

　以下、ＰＨＰの関数でどう使ったかという説明になります。システムの造りとしてはプロセス間通信に必要なリソースはサーバープロセスで取得し、クライアントがそのリソースを利用する、という構造にしています。そのため、以下の説明では最初にリソースを確保するのは｛明記なくても）サーバー側になります。

２．４．１．ＰＨＰのセマフォ関数

　ＰＨＰでセマフォを利用するには以下の流れになります。

1. セマフォＩＤを取得する(sem_get())
2. セマフォを取得する(sem_acquire())
3. （「シングルオーダー」や「オーダーアレイ」になにやら書き込む／読み出す）
4. セマフォを離す(sem_release())
5. セマフォを削除する(sem_remove())

　最後のsewm_remove()は意外と大事で（いや、意外ではないか）、これを忘れると取得したセマフォがいつまでもＯＳの共用メモリ空間に残り続けます。うっかりした作りや使い方をしていると、あとでipcsコマンドを実行してぎょっとなったりします。

２．４．２．ＰＨＰの共有メモリ関数

　ＯＳの機能の中にはメッセージング機能があり、ＰＨＰもmsg_****という関数群で取り扱うことができるのですが、今回製作したシステムでは使用していません。メッセージング機能はＦＩＦＯバッファなので用途的には問題ないのですが、メッセージング用に確保されている領域が（比較的）大きくないことと、ＰＨＰで利用できる型に制約があり、メッセージに構造を持たせようとすると途端に複雑化してしまうことが難点です。例えば楽曲再生では、アーティスト名、アルバム名、楽曲名を配列構造で渡したい場合に配列をそのまま送ることができません。シリアライズ関数(serialize()/unserialize())を通して、なおかつ戻り値のバイトコードを文字列として扱えるようにさらに関数を挟まざるを得なくなったりします。
　そのため、「Class IPC」では共有メモリ関数群(shm_****)を使いました。取得した共有メモリに対してＰＨＰは配列やオブジェクトであってもシリアライズ操作なしに直接読み書きできます。
共有メモリを取り扱う流れはだいたい以下のようになります。

1. 共有メモリセグメントを作成する（あるいは作成済みのセグメントに接続する）(shm_attach())
2. 接続した共有メモリセグメントに変数を書き込む(shm_put_var())
3. 接続した共有メモリセグメントから変数を読み出す(shm_get_var())
4. 共有メモリセグメントを切り離す(shm_detach())
5. 共有メモリセグメントを削除する(shm_remove())

　最後のshm_remove()は意外と大事で（以下略）。

２．４．３．実装

　以下、「class IPC」の実装の説明に入りますが、なにぶんサイズが大きいので分割して説明します。

２．４．３．１．コンストラクタとデストラクタとおまじない

　先に説明したように、プロセス間通信のリソースはサーバー側で取得するように作りました。サーバーなのか、クライアントなのかは引数で明示的に指定するようにします。コンストラクタの中ではサーバー動作とクライアント動作に分かれて、利用できるリソースがあるかとか、サーバーが二重起動していないかとかチェックしています。
　このクラスファイルはサーバーとクライアントの双方で物理的に同じファイルを共有して使います。

　そして冒頭に変なdefine()があります。これは共有リソースにアクセスするためのキーを得るために、このクラスを定義したファイルそのものを使うための定義です。サーバー側とクライアント側が同じキーを取る（そして他のアプリと重複しない）ためのおまじないです。

define("gpd_keyGplayModule", ftok(__FILE__, 'A'));
class IPC{
  private $semGpd;
  private $shmGpd;
  private $mode;
  private $receivedCommand;
  public  $enable;

  public function __construct($mode=GPD_MSERVER){
    if( ($this->shmGpd = shm_attach(gpd_keyGplayModule, 40960, 0600)) == false){
      return null;
    }
    if( ($this->semGpd = sem_get(gpd_keyGplayModule, 1, 0666, true)) == false){
      return null;
    }
    sem_acquire($this->semGpd);
    
    $this->mode = $mode;
    if($mode == GPD_MCLIENT){
      if(shm_has_var($this->shmGpd, GPD_ENABLE) === false){
        sem_release($this->semGpd);
        shm_remove($this->shmGpd);
        sem_remove($this->semGpd);
        $this->enable = false;
        return;
      }
    } else {
      shm_put_var($this->shmGpd, GPD_ENABLE, false);
      $ringList = array();
      for($i=0;$i<gpdRingSize;$i++){
        $ringList[] = array("artist"=>"", "album"=>"", "title"=>"", "dst"=>"");
      }
      $ringStat = array('SIZE'=>gpdRingSize, 'W'=> -1, 'R'=> -1);
      shm_put_var($this->shmGpd, GPD_KEY, $ringStat);
      shm_put_var($this->shmGpd, GPD_RING, $ringList);
      shm_put_var($this->shmGpd, GPD_ENABLE, true);
      shm_put_var($this->shmGpd, GPD_CMD, null);
    }
    sem_release($this->semGpd);
    $this->enable = true;
  }
  
  public function __destruct(){
    if($this->mode == GPD_MSERVER){
      shm_remove($this->shmGpd);
      sem_remove($this->semGpd);
    }
    shm_detach($this->shmGpd);
  }
}

　コンストラクタのサーバー側動作としては、セマフォを確保してからリングバッファの初期設定や、「シングルオーダー」の領域を作り、セマフォを解放する、といったことをしています。

２．４．３．２．オーダーを送ったり受け取ったり

「シングルオーダー」まわりは以下のメソッドで実装しています。

  public function sendOrder($order){
    if(shm_get_var($this->shmGpd, GPD_CMD) != null) return false;
    sem_acquire($this->semGpd);
    shm_put_var($this->shmGpd, GPD_CMD, $order);
    sem_release($this->semGpd);
    return true;
  }

  
  public function receivedOrder(){
    sem_acquire($this->semGpd);
    if($this->receivedCommand != null){
      $command = shm_get_var($this->shmGpd, GPD_CMD);
    } else {
      $command = $this->receivedCommand;
    }
    $this->receivedCommand = null;
    shm_put_var($this->shmGpd, GPD_CMD, null);
    sem_release($this->semGpd);
    return $command;
  }
  
  public function is_ordered(){
    sem_acquire($this->semGpd);
    $this->receivedCommand = shm_get_var($this->shmGpd, GPD_CMD);
    sem_release($this->semGpd);
    return ($this->receivedCommand != null);
  }

• IPC::sendOrder()
  引数で渡された変数をそのまま共有メモリセグメントに書き込みます。変数が持つ属性、構造は維持されます。
• IPC::receivedOrder()
  　共有メモリセグメントから変数を取り出します。
• IPC::is_orderd()
  　「シングルオーダー」に変数が書き込まれているかどうかを判定します。ＰＨＰによくあるis_****()に似せるためのメソッドです。

　IPC::receivedOrder()が妙なことをやっていますが、IPC::is_orderd()で一回読みだしてたら再度読み込みする場合は共有メモリへアクセスに行かないようにしています。が、そこまでやる必要はないようにも思います。

２．４．３．３．リングバッファ

　リングバッファは先に説明したようにリングバッファとして扱える構造を丸ごと共有メモリセグメントにshm_put_var()したりshm_get_var()しているだけなので本稿での説明は割愛します。

３．おわりに

　こんな感じのクラスを作成してプロセス間通信の非同期化を図りました。もともとプロセスが分かれている時点で端から非同期になってしまうので、正確に言うと「非同期状態でも通信できる機能」、と呼んだ方がいいのかもしれません。ともあれ、タイミングの問題でおかしなエラーに見舞われるようなこともなく、ジュークボックスサーバーは使えています。