一覧CL-13-1 · サーバーレスとイベント

サーバーレス関数の実行環境と制限

同じ処理結果でも実行環境の寿命と並行数が異なることを、世代・呼出し回数・最大同時数で確認します。

説明

イベントを受けて動く関数では、実行環境の初期化と、各イベントを処理する段階が分かれます。同じ環境が再利用されると、メモリ内の値が次の処理でも残る場合があります。一方、再利用を前提に業務データを保存すると、環境が入れ替わったときに失われます。

Pythonの子プロセスを実行環境として、値を2倍にする関数へイベントを渡します。新規起動・再利用・期限超過・再起動の順に状態を確認し、最後に4件の並行実行へ上限を設けます。coldはこの例でプロセスを新しく起動したかを表します。

同じ処理結果でも実行環境の寿命と並行数が異なることを、世代・呼出し回数・最大同時数で確認します。

1実行環境を用意する

Runtimeは必要になった時点で子プロセスを起動します。初期化直後はまだプロセスがありません。

from functions import Runtime, concurrent実行環境と並行処理の試料を読み込む


import asyncio

runtime = Runtime()まだ起動していない実行環境を用意する

print起動状態を確認する(runtime.process, runtime.generation)

None、世代0です。環境を管理する側の状態と、イベント処理を行う子プロセスは別です。

2最初のイベントで起動する

e1のvalue=7を2倍にします。worker内では処理した回数も加算します。

runtime.invoke新しい実行環境でe1を処理する({"id": "e1", "value": 7})

世代1、cold=True、結果14、呼出し回数1です。プロセスの初期化後にイベントを処理しました。

3同じ環境を再利用する

続けてe2を渡します。Runtimeを作り直していないため、同じプロセスを使います。

runtime.invoke同じ実行環境へe2を渡す({"id": "e2", "value": 9})

世代1のままcold=False、結果18、呼出し回数2です。回数はプロセスのメモリに残っていた値です。

4期限を超える処理

イベントに0.2秒の待機を入れ、応答期限を0.05秒にします。この試料のRuntimeは期限超過時に子プロセスを終了させます。

runtime.invoke待機する処理を短い期限で呼ぶ({"id": "slow", "value": 5, "delay": 0.2}, timeout=0.05応答期限を0.05秒にする)

timeoutとなり、呼出し元へ結果は返りません。呼出し元の期限と、外部へ行った副作用が取り消されるかは別の問題です。

5次の処理は新しい環境になる

期限超過後にe3を渡します。

runtime.invoke終了後の新しい実行環境へe3を渡す({"id": "e3", "value": 4})

世代2、cold=True、呼出し回数1になります。前のメモリ内カウンターは引き継がれていません。

64件を同時に処理する

asyncioで0から3までの4件を処理します。最初は同時実行数を4にします。処理中のカウンターから最大同時数を測ります。

unlimited = asyncio.run(concurrent(4))4件を最大4並行で実行する

結果は0、2、4、6で、最大同時数は4です。4件が処理中になる区間がありました。

7同時実行数を2へ制限する

同じ4件を、Semaphoreで最大2件ずつ処理します。残りの処理は空きを待ちます。

limited = asyncio.run(concurrent(2))同じ4件を最大2並行で実行する

結果は同じですが最大同時数は2です。並行数の制限は処理を待たせますが、この例では件数を捨てていません。

8環境を終了する

子プロセスを終了し、管理側の参照を確認します。

runtime.close()子プロセスを終了する

print(runtime.process)実行環境が残っていないことを確認する

Noneになります。関数の業務状態をプロセス内だけに置くと、こうした環境の終了で失われます。

まとめ

環境の再利用中はメモリ内の値が残り、環境の終了後は失われます。

期限超過と外部の副作用の取消しは同じではありません。

並行数の上限は結果だけでなく待機と負荷にも関わります。

操作履歴と解説を振り返れます。

Python 3.15 / SQLite / 固定時計のイベント処理モデル

from functions import Runtime, concurrent
import asyncio
runtime = Runtime()
print(runtime.process, runtime.generation)

None 0

runtime.invoke({"id": "e1", "value": 7})

generation=1 cold=True result={"id": "e1", "invocations": 1, "value": 14}

runtime.invoke({"id": "e2", "value": 9})

generation=1 cold=False result={"id": "e2", "invocations": 2, "value": 18}

runtime.invoke({"id": "slow", "value": 5, "delay": 0.2}, timeout=0.05)

generation=1 cold=False timeout

runtime.invoke({"id": "e3", "value": 4})

generation=2 cold=True result={"id": "e3", "invocations": 1, "value": 8}

unlimited = asyncio.run(concurrent(4))

results=[0, 2, 4, 6] peak=4

limited = asyncio.run(concurrent(2))

results=[0, 2, 4, 6] peak=2

runtime.close()
print(runtime.process)

None

演習完了