局所トレーサビリティビュー グループ: BGIO

キャッシュの保存処理をバックグラウンドで非同期に実行し、関数の応答レイテンシに影響を与えないモードを提供すること。

コンポーネント構成

graph TD
  A[core.Spot] -->|非同期タスク投入| B[_BackgroundLoop]
  B -->|Daemon Thread| C[asyncio.AbstractEventLoop]
  C -->|I/O| D[TaskDB / BlobStorage]
  A -->|ライフサイクル制御| E[atexit / ContextManager]

コンポーネント責務

データフロー

sequenceDiagram
  participant Main as メインスレッド
  participant BGLoop as _BackgroundLoop
  participant Store as ストレージ

  Main->>BGLoop: submit(save_coro)
  Note over Main: 関数は即座に結果を返す (save_sync=False)
  BGLoop->>Store: 書き込み実行 (I/O)
  alt 成功
      Store-->>BGLoop: OK
  else 失敗
      BGLoop->>Main: on_background_error(error)
  end

技術選定

非機能要件方針

• 透過性: 保存失敗時もユーザーコードの例外にはせず、コールバック経由で通知

• 一貫性: flush() により、クリティカルな処理の直後での保存完了を保証

• 安全性: シャットダウンシーケンス中に新規タスクを受け付けない排他制御を導入

インターフェース

class _BackgroundLoop:
    def submit(self, coro: Coroutine) -> None: ...
    def stop(self) -> None: ...
    def is_running(self) -> bool: ...

振る舞い

1. 起動: 初回の submit 時にデーモンスレッドを作成し、asyncio.run() を開始する
2. 投入: run_coroutine_threadsafe を使用して、メインスレッドからコルーチンをイベントループへ投入
3. 待機: 投入されたタスクの完了は待たず、制御を即座に呼び出し元へ返す

パラメータ詳細

エラーハンドリング

• 投入失敗: ループ停止中にタスクが投入された場合、エラーをログ出力し無視する
• タスク内エラー: コルーチン内で発生した例外は、on_background_error コールバックで処理される

エッジケース

• 多重起動: すでにスレッドが実行中の場合は、既存のスレッドを再利用する

インターフェース

def flush(timeout: float | None = None) -> bool: ...

@spot.mark(save_sync=False)
def my_func(): ...

振る舞い

同期保存 (save_sync=True)

• 関数終了後、DBおよびBlobへの保存が完了するまで呼び出し元をブロックする。

非同期保存 (save_sync=False)

• 保存処理を BackgroundLoop に委譲し、即座に関数の戻り値を返す。

フラッシュ (flush)

• _inflight タスク（保存中）のリストを確認し、それら全てが完了するまで、またはタイムアウトまで待機する。

パラメータ詳細

エラーハンドリング

• タイムアウト: flush がタイムアウトした場合、False を返し、一部のデータが未保存である可能性を通知する。

エッジケース

• コンテキスト終了: with spot: 終了時に自動的に flush が呼ばれ、プロセス終了前のデータ保存を確実にする。

目的

_BackgroundLoop はデーモンスレッドで asyncio イベントループを駆動し、非同期キャッシュ保存を処理する。スレッド間のコルーチン受け渡しは競合状態やデッドロックが発生しやすく、シャットダウン時のタスク消失はデータロスに直結する。

検証観点

• 保存の正確性: submit(coro) で投入したコルーチンが正しく実行され、結果がDBとストレージに反映されること
• 直列化保証: 同一キーへの保存が投入順に処理され、最終状態が最後に投入されたデータであること
• ゾンビスレッド防止: シャットダウン後にアクティブタスクが drain_timeout 秒以内に完了すること。タイムアウト後は強制終了されること
• atexit 安全網: プロセス終了時に atexit ハンドラが発火し、未完了タスクの drain が試行されること
• 新規受付停止: シャットダウン後の submit() 呼び出しが受け付けられないこと

目的

save_sync パラメータと flush / drain メカニズムは、キャッシュ保存のレイテンシとデータ安全性のトレードオフをユーザーが制御する手段である。save_sync=False 使用時にデータが失われないことと、コンテキストマネージャによる確実なフラッシュを保証する。

検証観点

• save_sync=True（デフォルト）: キャッシュ保存が呼び出しスレッドで同期的に完了し、関数リターン時点でデータが永続化されていること
• save_sync=False: キャッシュ保存がバックグラウンドに委譲され、関数がレイテンシなしでリターンすること。保存が最終的に完了すること
• flush(): spot.flush(timeout) 呼び出しで全ペンディング保存が完了するまで待機すること。タイムアウト時の挙動が明確であること
• コンテキスト exit での drain: with spot: ブロック終了時に未完了の保存が自動的に drain されること
• コンテキストマネージャの再利用: 同一 spot インスタンスが複数の with ブロックで再利用でき、各ブロック終了時に適切に drain されること
• バックグラウンド保存失敗: 非同期保存が例外を送出した場合に on_background_error コールバックが呼ばれ、メインスレッドには影響しないこと

実装概要

_BackgroundLoop クラスが非同期タスクのバックグラウンド実行を管理する。 初期化時に asyncio.new_event_loop() で新しいイベントループを作成し、 _thread（デーモンスレッド）上で loop.run_forever() を実行する。 外部からは submit() を通じてコルーチンを投入でき、run_coroutine_threadsafe で スレッドセーフにループへ渡される。

設計判断

デーモンスレッドと明示的なシャットダウン

スレッドは daemon=True とし、メインスレッド終了時にプロセスがブロックされないようにしている。 しかし、安全なリソース解放のために drain() メソッドを提供し、 投入された全タスクの完了を drain_timeout の範囲で待機する。

独立したイベントループ

メインスレッドのイベントループと競合しないよう、専用のイベントループを 作成してバックグラウンドスレッドで駆動する。これにより save_sync=False 時の 保存処理などが、呼び出し元の asyncio 環境から完全に隔離される。

実装メモ

• submit() は投入されたタスクを _tasks 集合に登録し、add_done_callback で 完了時に自身を削除することで、未完了タスクの追跡を可能にしている
• シャットダウン時の競合を防ぐため、_lock による排他制御を行っている
• atexit ハンドラとしてもシャットダウンが登録される

実装概要

Spot クラスにおいて、バックグラウンド書き込みの同期と完了待機を制御する。 save_sync=False の場合、キャッシュへの保存処理は _bg_loop.submit() で バックグラウンドに投入される。これら未完了のタスクやDBキューを同期するため、 flush() およびコンテキストマネージャによる drain が実装されている。

設計判断

flush と drain の使い分け

• flush(): DBライタースレッドのキュー (self.db.flush()) を空になるまで待機する
• __exit__: コンテキストマネージャ終了時に flush() を呼び出した上で、 さらに _bg_loop.drain() を呼び出し、非同期タスクの完了も待つ これにより、スクリプト終了時やバッチ処理の区切りで、データの完全な永続化を保証する。

バックグラウンドエラーの隔離

バックグラウンド保存時のエラーは呼び出し元のメインフローを妨げないよう、 on_background_error コールバックに通知され、ログ出力のみ行う。 例外の再送出は行わない。

実装メモ

• flush() にはタイムアウトを設定でき、ハングアップを防止している
• on_background_error は SaveErrorContext を引数に取り、失敗時の キーや関数の詳細情報を提供する