psim言語講座（第14回）チュートリアル編(5)

HOME
psim言語講座（第14回）チュートリアル編(5)

2024年4月15日 11:13

本記事は当社が発行しているシミュレーションメールマガジン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 廃棄
...