Pausable Unittest その1 (continulet と tasklet)

概要

初めて Python 向けのライブラリを書きました。
Pausable Unittest という名前で、名前の通り unittest の亜種です。

このライブラリは、通常の unittest と異なり、途中で中断することができます。

今回は、このライブラリの目的と、利用している PyPy, Stackless Python の技術についてまとめます。

目的

PC に限らず、家電や組み込みデバイスなどの開発をする場合、再起動をはさんだ処理の挙動を確認したいことが、しばしばあります。
例えば、特定の条件下で起動時にハングアップしないかや、再起動後に種々のレジスタが正しく設定されているかをチェックする場合などです。

こういった場合、既存のテストフレームワークでは、テストを書くのが難しいことが多いです。
例えば、実際には動きませんが、 Python の unittest なら、下記のように書けると嬉しいです。

class TestSample(unittest.TestCase):
    def test_check_reg(self):
        reg1 = read_reg()  # (*1)
        self.reboot()      # (*2)
        reg2 = read_reg()  # (*3)
        self.assertEqual(reg1, reg2)  # (*4)

意図としては、(*1) でレジスタ値を読み、(*2) で再起動し、 (*3) で再度レジスタ値を読む、ということを実現しようとしています。
(*1) で reg1 変数に格納した結果を、 (*2) のシステム再起動後の (*4) でも参照できるようにしたいところがポイントです。

pausable_unittest

上記のようにテストスクリプトを書けるようにしたのが、 pausable_unittest です。
下記のように、 pausable_unittest.TestCase を継承することで、再起動をはさんでもテストを続けることができます。

import pausable_unittest
import pausable_unittest.windowspauser
 
def read_reg():
  # read some registers.
  return 0x0000
 
class TestSample(pausable_unittest.TestCase):
    def test_check_reg(self):
        reg1 = read_reg()  # (*1)
        self.reboot()      # (*2)
        reg2 = read_reg()  # (*3)
        self.assertEqual(reg1, reg2)  # (*4)
 
pausable_unittest.main(pausable_unittest.windowspauser.Pauser())  # (*5)

ただし現在のところ、 Stackless Python か PyPy でしか動作しません。
また、 (*1) でファイルなどを開いた状態で (*2) の reboot を呼ぶと、エラーで停止します。
これらは実装上の都合によるものです。後述するように pausable_unittest では、現在の状態を pickle を用いてシリアライズするので、シリアライズできないオブジェクト (ファイルなど) が変数に保持されていると、状態の保存に失敗します。

pausable_unittest の仕組み

Stackless Python と PyPy でやや異なりますが、 PyPy の continulet が基本になっていますので、そちらから説明します。

continulet

PyPy には、 continulet というオブジェクトがあります。

continulet は軽量スレッドの一種ですが、ユーザーが明示的に continulet の中断/切り替え操作をしないと、continulet (軽量スレッド) が切り替わることはありません。
この点で Ruby の Fiber とほぼ同じものです。

また、 continulet はシリアライズすることができます。
つまり、中断状態の continulet を、実行中の関数の変数の値などを含めて、すべてファイルに書き出すことができます。当然、ファイルに書き出した状態を、後で復旧させることもできます。

この機能を用いて、下記のようなことが実現できます。

from _continuation import continulet
import pickle
 
def func(con):
  for i in range(2):
    print(i)
    con.switch()
 
def main():
  c1 = continulet(func) # (*1)
  c1.switch()           # (*2)
  # => print(0)
  s = pickle.dumps(c1)  # (*3)
  c2 = pickle.loads(s)  # (*4)
  c2.switch()           # (*5)
  # => print(1)
  # ...
 
main()

このスクリプトは以下のように実行されます。

func 関数を実行する continulet を作成します。
この時点では、まだ func は実行されません。
c1.switch() を呼び、 func に制御を移します。
その結果、 func が実行されます。
func の中で con.switch() が呼ばれると func の制御を中断し、制御を main に戻します。
続いて、 pickle で中断状態を含めて continulet をシリアライズします。
シリアライズされた文字列から、 continulet を復旧させています。
復旧させた continulet に対して switch を呼び、 func に制御を移します。
このとき、前回に中断した状態から実行が再開されます。今回の例では、 for 文の中から再開されます。
ローカル変数 i の値も含めて、前回に中断した状態から再開されます。

tasklet

一方で、 Stackless Python には似たような仕組みとして tasklet があります。
PyPy の continulet とややインターフェースは異なりますが、中断状態の tasklet を保存できるなど、ほぼ同じような特徴を持っています。

pausable_unittest では、 tasklet を continulet 風に使うためのラッパーオブジェクト pausable_unittest.continulet を用意して、 PyPy でも Stackless Python でも同じように動作させられるようにしています。
ただし、あくまで pausable_unittest で使うためだけのものなので、複数の continulet がある場合などをエミュレートするようにはできていません。

この、「中断状態をシリアライズし、後で復旧させられる」という特徴を活用することで、 pausable_unittest では楽に再起動後をまたいだ処理を書くことができます。

余談ですが、私は Ruby が好きなので、 Ruby の Fiber に同様の機能があれば、おそらく Ruby で実装していただろうと思います。

次回は、これらを使ってどのように pausable_unittest を実装したかについて書く予定です。

という予定だったのですが、 PyCon JP 2016 で発表させていただくことになったので、それが終わったらまとめようと思います。

覚え書き

SyntaxHighlighter

2016年5月7日土曜日