psim言語講座(第14回)チュートリアル編(5)psim のスケジューラを使いこなす

  • HOME
  • psim言語講座(第14回)チュートリアル編(5)psim のスケジューラを使いこなす

本記事は当社が発行しているシミュレーションメールマガジンVol. 14の記事です。
シミュレーションメールマガジンの詳細・購読申込はこちら のサポートページから

はじめに

前回は、psim の Store 資源の基本的な利用方法を説明しました。

今回は、Store の応用例と、その背後にある psim のスケジューラを説明します。

ストアから取得前の確認と psim スケジューラ

前回、ストアから取得条件の指定方法の例を説明しました。

具体的には、s.get1 に条件式を加える事ができます。

以下のように書いた場合、

        yield s.get1(lambda v: v == "商品3")

ストア s の中から、「商品3」を取得しようとします。商品3がある場合は即座に取得しますが、商品3がない場合は取得できるまで待ちます。

しかしながら、現実の世界のモデルでは、あまりこのような待ち方はしないかもしれません。

例えば、実際のスーパーの商品棚などを想定すると、おそらく、商品が無い場合は、商品の入手をあきらめる事が多いのではないでしょうか。

psim の API は、待ち受け操作をアトミックにするために、少々複雑な仕様になっています。なぜなら、アトミックな操作でないと、様々な問題が生じてしまうためです。例えば、同時に 2 人のユーザが、在庫が 1 つしかない ストア s から商品を取得した場合、現実の世界では、1 人のみが商品を取得し、もう一人は商品を取得できません。

この動作はあたり前のように思うかもしれませんが、プログラミング言語の世界では、ここは注意深くなる必要があります。もし、ストア s にひとつの商品が格納されている状態で、「同時刻」に、s.get1() という待ち受け式を実行するプロセスが 2 つあるような状況を、想定してみます。もし、それらの待ち受けが、本当に「同時刻」に実行された場合、両方のプロセスで、s.get1() が成立します。しかし、ストア s にひとつの商品が格納されているのに、両方のプロセスが、商品を取得できるような事は、発生してはなりません。

そのような不整合が発生しないように、psim では、s.get1() という待ち受け式はアトミックな操作である事を保証しています。つまり、片方のプロセスのみが、商品を取得する事を保証しています。psim のルールに従ってプログラミングしている場合、このような不整合は発生しないように設計されています。

s.get1() は、もし、在庫があれば即座に成立し、在庫がない場合、取得出来るまで、待ち受けるという API になっています。更に、複数の並行した待ち受けが存在した場合、先に実行した待ち受け式の方が先に発火する事を保証しています。また、全ての操作には順序が存在し、同時刻の操作というものは存在しない事を保証しています。

psim 言語の中では、プロセスは generator を用いて記述します。複数の並行プロセスの切り替えは、psim のスケジューラが行っており、上記のようなルールが厳密にまもられるように、並行プロセスの切り替えを適切に行っています。基本的には、yield 式で指定された待ち受け式の内部のみで、この並行プロセスの切り替えが発生するようになっています。

このような背景で、「ストア s に指定された商品が無い場合は、商品の入手をあきらめる」という待ち受けはどう書くのかを考えます。

実は、以下のようにして、ストア s が格納している商品リストを得る事ができます。

    s.buffer

この結果は、Python の普通のリストオブジェクトなので、for 文でループを回したら、各要素に比較演算を行う事ができます。

しかしながら、以下のルールがあります。このリストはあくまでも「読み込み専用」です。このリストを破壊するような操作を行った場合、もし、s.get1() のような待ち受けを行っている他のプロセスがあった場合、不整合が発生してしまいます。そのため、「読み込み専用」というルールは厳密にまもる必要があります。

基本的には、読み込み操作、つまり観測であれば、自由に行う事ができます。

例えば、「ストア s の中に 製品1があるなら」という条件は以下のように書く事ができます。

    if "製品1" in s.buffer:

このような観測は、プロセス内で自由に配置する事が可能で、指定された条件によって、処理を切り替える事が可能です。

実際に、商品が無い場合は、商品の入手をあきらめるようなシミュレーションは以下のように記述可能です。

    initialize()
    s = Store(False, monitor = True)

    def supply(item):
        print(f"時間 {now()} {item} 追加待ち")
        printStoreStatus()
        yield s.put1(item)
        print(f"時間 {now()} {item} 追加終了")
        printStoreStatus()

    def supplier():
        for item in ["商品1", "商品2", "商品3", "商品4", "商品5",
                     "商品6", "商品7"]:
            yield pause(1)
            yield subactivate(supply)(item)
        yield alwaysFalse()

    def consume(item):
        print(f"時間 {now()} {item} 取得待ち")
        printStoreStatus()
        if item in s.buffer:
            result = yield s.get1(lambda v: v == item, name = "get1")
            item = result["get1"]
            print(f"時間 {now()} {item} 取得終了")
            printStoreStatus()
        else:
            print(f"時間 {now()} {item} 取得断念")
            printStoreStatus()

    def consumer():
        for item in ["商品4", "商品1", "商品2", "商品7", "商品5",
                     "商品6", "商品3"]:
            yield pause(2)
            yield subactivate(consume)(item)
        yield alwaysFalse()

    activate(supplier)()
    activate(consumer)()
    start(until = 10)

ポイントは以下の条件が成立した場合に、

    if item in s.buffer:

s.get1 を呼出している所です。

         result = yield s.get1(lambda v: v == item, name = "get1")

実行結果は以下のようになります。

時間 1.0 商品1 追加待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 1.0 商品1 追加終了
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品4 取得待ち
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品4 取得断念
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加待ち
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
...

pause(0) を利用したストアから取得前の確認

商品が無い場合は、商品の入手をあきらめるようなシミュレーションを記述する方法には、別のアプローチも存在します。

少々テクニカルな手法になりますが、以下のようなコードでも、同じ結果が得られます。

    initialize()
    s = Store(False, monitor = True)

    def supply(item):
        print(f"時間 {now()} {item} 追加待ち")
        printStoreStatus()
        yield s.put1(item)
        print(f"時間 {now()} {item} 追加終了")
        printStoreStatus()

    def supplier():
        for item in ["商品1", "商品2", "商品3", "商品4", "商品5",
                     "商品6", "商品7"]:
            yield pause(1)
            yield subactivate(supply)(item)
        yield alwaysFalse()

    def consume(item):
        print(f"時間 {now()} {item} 取得待ち")
        printStoreStatus()
        result = yield s.get1(lambda v: v == item, name = "get1") | pause(0, name = "giveup")
        if "giveup" in result:
            print(f"時間 {now()} {item} 取得断念")
            printStoreStatus()
        else:
            print(f"時間 {now()} {item} 取得終了")
            printStoreStatus()

    def consumer():
        for item in ["商品4", "商品1", "商品2", "商品7", "商品5",
                     "商品6", "商品3"]:
            yield pause(2)
            yield subactivate(consume)(item)
        yield alwaysFalse()

    activate(supplier)()
    activate(consumer)()
    start(until = 10)

ポイントは以下のような待ち受け式です。

        result = yield s.get1(lambda v: v == item, name = "get1") | pause(0, name = "giveup")

この複合待ち受け式は、s.get1() と、pause(0) を同時に実行して、先に成立した方が成立します。

もし、ストア s に、指定した製品の在庫がある場合、即座に成立します。一方で、指定した製品の在庫がなかった場合、pause(0)の待ち受け式を実行します。この待ち受け式は、0 秒後に発火する待ち受け式です。つまり、実行直後に成立する事になります。結果として、指定した製品の在庫がなかった場合、即座に成立する事になります。

よって、これは、どのような状態であれ、即座に成立する待ち受け式です。また、

        if "giveup" in result:

が成立する場合は、指定した製品の在庫がなかった事を示し、成立しなかった場合は、指定した製品の在庫取得できた事を、ひとつの待ち受け式で確認できます。

これは、先の、

    if item in s.buffer:

による確認と、

        result = yield s.get1(lambda v: v == item, name = "get1")

の取得操作を、同時に行っている事になります。

実行結果は以下のように、先の例と同じ結果が得られます。

時間 1.0 商品1 追加待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 1.0 商品1 追加終了
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品4 取得待ち
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加待ち
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品4 取得断念
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加終了

廃棄処理

今までの例では、ストアの中に保存するデータは、単純に文字列としていましたが、他にも複数の属性を含める事ができます。

複数の属性を含めるには、たとえば、Python のオブジェクトをストアに格納する方法と、Python の辞書をストアに格納する方法があります。

ここでは、辞書を格納する方法の例を示します。

シミュレーション対象としては、前述のスーパーの商品棚の例で、商品の供給と消費が自立的に発生する中、毎時間棚卸しを行う例を考えます。毎時間の棚卸しでは、消費期限切れの商品を廃棄するとします。

まず、各商品に、廃棄時間という概念が必要になります。そのために、商品を追加する時に、以下の様に辞書を指定するようにします。

    yield s.put1({"商品": "商品名" "廃棄時間": now() + 廃棄時間})

こうした場合、s.buffer の各要素は辞書になりますし、s.get1() で取得したアイテムも、辞書になります。

商品取得時の条件式も、注意深く書く必要があります。例えば、指定した商品を取得するには、以下のように記述する必要があります。

        result = yield s.get1(lambda v: v["商品"] == item, name = "get1") | pause(0, name = "giveup")

実際のコードは以下のように記述できます。

    initialize()
    s = Store(False, monitor = True)

    def supply(item, disposalTime):
        print(f"時間 {now()} {item} 追加待ち")
        printStoreStatus()
        yield s.put1({"商品": item, "廃棄時間": now() + disposalTime})
        print(f"時間 {now()} {item} 追加終了")
        printStoreStatus()

    def supplier():
        for item, disposalTime in [("商品1", 3),
                                   ("商品2", 1),
                                   ("商品3", 2),
                                   ("商品4", 1),
                                   ("商品5", 2),
                                   ("商品6", 3),
                                   ("商品7", 2)]:
            yield pause(1)
            yield subactivate(supply)(item, disposalTime)
        yield alwaysFalse()

    def consume(item):
        print(f"時間 {now()} {item} 取得待ち")
        printStoreStatus()
        result = yield s.get1(lambda v: v["商品"] == item, name = "get1") | pause(0, name = "giveup")
        if "giveup" in result:
            print(f"時間 {now()} {item} 取得断念")
            printStoreStatus()
        else:
            print(f"時間 {now()} {item} 取得終了")
            printStoreStatus()

    def consumer():
        for item in ["商品4", "商品1", "商品2", "商品7", "商品5",
                     "商品6", "商品3"]:
            yield pause(2)
            yield subactivate(consume)(item)
        yield alwaysFalse()

    def disposer():
        while True:
            yield pause(1)
            while True:
                result = yield s.get1(lambda v: v["廃棄時間"] >= now(),
                                     name = "get1") | pause(0, name = "giveup")
                if "giveup" in result:
                    break
                else:
                    item = result["get1"]["商品"]
                    print(f"時間 {now()} {item} 廃棄")
            printStoreStatus()

    activate(supplier)()
    activate(consumer)()
    activate(disposer)()
    start(until = 10)

disposer が、廃棄処理プロセスです。

    v["廃棄時間"] >= now()

が成立する商品を除去します。この除去ループを回して、ひとつも除去できなかったら、その時刻の除去処理は完了します。

実行結果は以下のようになります。

バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 1.0 商品1 追加待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 1.0 商品1 追加終了
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品1 廃棄
時間 2.0 商品4 取得待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加待ち
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品4 取得断念
バッファ: 0/False, 取得待ち: 0, 追加待ち: 0
時間 2.0 商品2 追加終了
バッファ: 1/False, 取得待ち: 0, 追加待ち: 0
時間 3.0 商品2 廃棄
...

まとめ

今回は、Store の応用例として、その背後にある psim のスケジューラを説明しました。

psim のスケジューラは、競合する複数の待ち受け式を適切に処理される事を保証してくれます。そのため、s.buffer によってストアの中身を確認する事ができますが、その中身を破壊してはならない事を説明しました。また、複合待ち受け式を用いて、同様の待ち受けを記述可能である事も説明しました。

次回は、離散イベントシミュレーションでよく発生するような、より具体的なストアの使い方について説明しようと思います。

監修:株式会社NTTデータ数理システム 機械学習、統計解析、数理計画、シミュレーションなどの数理科学を 背景とした技術を活用し、業種・テーマを問わず幅広く仕事をしています。
http://www.msi.co.jp NTTデータ数理システムができること
「数理科学の基礎知識」e-book無料ダウンロードはこちら

関連記事