[ruby] Parsletで正規表現をパースしてvm型のエンジンを実装してみる[その2] AST〜VMまで

Parsletで正規表現

その１で、正規表現をパースしてASTを作るところまで実装しました。
ここではASTからVMの命令列を作って、実際に文字列のマッチを行えるようにします。

命令列に関しても 正規表現技術入門 を参考にしてて、ほとんどそのままです。

概要

命令列

バイトコードなどと呼ばれるものですが、今回はバイト列というより、Rubyレベルで命令オブジェクトを作って、その命令オブジェクトの配列を命令列とし、その配列のindexを操作するのを、プログラムカウンタ的なものに相当するように作っています。

今回の命令は以下の４つです。各命令は引数を一つとるか、または引数無しです。
(スタックとかジャンプの意味は後述)

命令	内容
CHAR	文字を引数にとり、その文字とマッチするための命令
PUSH	スタックに引数addrから計算したアドレスをpush
JUMP	引数addrから計算した(命令列の)indexにジャンプする
MATCH	この命令まで到達できたらマッチ成功 (引数無し)

バックトラック

正規表現(今回はDFAというタイプでなくNFAの方)でのマッチのやり方として、マッチが成功する間、命令を進め、マッチに失敗したら最後に成功したところまで戻り、別の選択肢があればマッチを続ける、選択肢がなければそこで終了、というような挙動になります。
その「失敗したら、Xに戻って再開」というような動きをバックトラックと呼ばれますが、その「戻る」というようなオペレーションをを実現するために、 PUSH という命令を用意しています。

再帰関数を使ってのバックトラックは関数呼び出しのスタックフレームでスタックを実現できるのですが、再帰の深さによってはマシンスタックが足りなくなってスタックオーバーフローになったりします。今回はスタックを自前で管理して、ループを使ってバックトラックを実現します。

簡単な例だと、 1〜4の数字の中から重複を許して 2つの数字を選ぶような処理の全パターンを取得する場合

再帰で書くと

$max_length = 2

def backtracking_by_recursive_funcion(n, acc)
  if n == $max_length # 選択する個数分になったので表示
    p acc
  else
    1.upto(4) do |i|
      backtracking_by_recursive_funcion(n + 1, acc + [i])
    end
  end
end

再帰をやめて、関数呼び出しのスタック部分を自分で管理してループにすると、下記のようになります

def backtracking_by_loop
  # 再帰のときに引数で渡していた 「今の個数(n)、 組み合わせた数字の途中経過(acc)」、をスタックに入れて管理
  stack = [[0, []]]

  until stack.empty?
    count, acc = stack.pop
    if count == $max_length # 選択する個数分になったので表示
      # 終了した場合は表示するだけで、もうstackに積まないので、ここまで進むと終了しどんどんstackが空になっていく。
      p acc
    else
      # 再帰の場合は、関数が呼ばれた順に処理が進むが、
      # stackに入れていく場合は、最後に入れたものが次に処理されるため、
      # 処理順が逆になる・・・ので、再帰の場合を同じになるように、4〜1の順でstackに積んでいく
      4.downto(1) do |i|
        stack << [count + 1, acc + [i]]
      end
    end
  end
end

微妙に数字を入れていく順場が違いますが、これはパターンの処理順が再帰と、ループで違うからなので、結果が同じになるように適当に調整しています。

これらを実行すると、同じ結果が出力されます。(２個選ぶうち、１つ目の選択肢が4パターン、2つ目の選択しも4パターンなので4*4=16パターン)

p :recursive_function
backtracking_by_recursive_funcion(0, [])
p :loop
backtracking_by_loop
=> :recursive_function
[1, 1]
[1, 2]
[1, 3]
[1, 4]
[2, 1]
[2, 2]
[2, 3]
[2, 4]
[3, 1]
[3, 2]
[3, 3]
[3, 4]
[4, 1]
[4, 2]
[4, 3]
[4, 4]
:loop
[1, 1]
[1, 2]
[1, 3]
[1, 4]
[2, 1]
[2, 2]
[2, 3]
[2, 4]
[3, 1]
[3, 2]
[3, 3]
[3, 4]
[4, 1]
[4, 2]
[4, 3]
[4, 4]

いずれの処理も、 一旦１つ目の数字を選んでみて、２つ目の数字を選べるだけ選び(4パターン)、選べなくなったら、一つ目の数字を一つ進めて、次のパターンを探す というバックトラックの処理をしてすべての組み合わせを探しています。

さて、このバックトラックの深さが深すぎると、再帰の場合スタックフレームが足りなくなってエラーになるのですが、僕の環境だと、だいたい3650個ぐらいまで深くなると下記のようにエラーになるようでした。ループの方は問題なく実行されます。

また、このループ形式のサンプルは、後でつくるVMのバックトラックの構造と全く同じになります。

:recursive_function
Traceback (most recent call last):
        10920: from main.rb:34:in `<main>'
        10919: from main.rb:8:in `backtracking_by_recursive_funcion'
        10918: from main.rb:8:in `upto'
        10917: from main.rb:9:in `block in backtracking_by_recursive_funcion'
        10916: from main.rb:8:in `backtracking_by_recursive_funcion'
        10915: from main.rb:8:in `upto'
        10914: from main.rb:9:in `block in backtracking_by_recursive_funcion'
        10913: from main.rb:8:in `backtracking_by_recursive_funcion'
         ... 10908 levels...
            4: from main.rb:8:in `upto'
            3: from main.rb:9:in `block in backtracking_by_recursive_funcion'
            2: from main.rb:8:in `backtracking_by_recursive_funcion'
            1: from main.rb:8:in `upto'
main.rb:9:in `block in backtracking_by_recursive_funcion': stack level too deep (SystemStackError)

正規表現のバックトラック

バックトラックの具体例

正規表現のバックトラックに関しての前提として、 今対象文字列のどの位置でマッチしようとしているかを表す変数を sp とします。

a*b

という正規表現を aab にマッチさせる場合、

• sp=0 の位置で、 a を試す

sp
|
v
aab

a がマッチするので、 sp を進めて次の a のマッチへ

• sp=1の位置で、 a を試す

 sp
 |
 v
aab

a がマッチするので、 sp を進めて次の a のマッチへ

• sp=2の位置で、 a を試す

  sp
  |
  v
aab

sp=2 は b なので a のマッチが失敗する。

するとここまでで a* のパターンが、２つの a にマッチし、３個めの a のマッチが失敗したため、a* のマッチを成功(0回以上aがあったため)とし、次のパターン b を試すようにバックトラックします。

• sp=2の位置で、 b を試す

  sp
  |
  v
aab

b がマッチし、正規表現の最後まで到達したので全体としてマッチします。

バックトラックまとめ

上記でみたように、失敗したとき用に、「次にどの状態で、どこからマッチを再開するか？」 というのをスタックにどんどん保存しておいて、失敗したときに、スタックの先頭から(＝直前の保存状態)から復元してマッチを続けるという動きでバックトラックを実現します。

このときに具体的に、スタックに保存されるものは、

• マッチ対象文字列のどの位置のマッチか？
• 正規表現の命令のどの位置のマッチか？

前者は先程の sp で、単純に検索文字列のどの部分か？の位置です。
後者の「命令のどの位置」というのは、 正規表現の字面の位置ではなく、先程出てきた 「命令列のどの場所か？」 の位置で、今後 pc (プログラムカウンタ) と呼びます。

命令列の構造

各命令を見る前に、命令列と、その実行に関しての概要です。

命令は冒頭に挙げた４種類あり、それが配列で並んでいます。
VMはその命令を上から順に実行していき、その命令としてJUMPがあったりすると、どこかにJUMPしたりしますが、それも含めて、「単純に上から順番に命令を実行しているだけ」という動きです。

例) abcにマッチする、正規表現の命令列

index(アドレス)	命令
0	char a
1	char b
2	char c
3	match

上から順に、 aにマッチ、bにマッチ、cにマッチ、マッチ終了という4状態進めるだけです。

これに対して、 文字列 abc をマッチさせてpc, spの動きを見てみます。

初期状態は、 sp=0, pc=0 で始まります。

pcの値が左列の index(アドレス) のどこを実行するかに当たります。

index(アドレス)	命令	命令終了後のpc, spへの操作
0	char a	マッチしたので、pc,spともに１進めるので、次はpc=1の命令に進み、spは1になる
1	char b	マッチしたので、pc,spともに１進めるので、次はpc=2の命令に進み、spは2になる
2	char c	マッチしたので、pc,spともに１進めるので、次はpc=3の命令に進み、spは3になる
3	match	pc=3はmatch命令なので、ここに到達すればマッチ成功として終了する

各命令の詳細

• CHAR x
  現在のspの位置の文字が x かどうか調べて同じであれば成功として、 spを1進める。 同じでなければ、マッチ失敗とする

• PUSH addr
  このPUSHの次の命令のアドレスからみてaddrだけ離れたアドレスと、今のspをスタックに積む。

  例)

index	命令
0	char a
1	push 1
2	jump -3
3	char b
4	match

このような命令列(ちなみに a+b に相当する命令列です) があったとして、 今、 pc=1, sp=2とします。

pc=1は push命令なので、「次の命令( jump -3 )のあるアドレス( 2 ) からみて addr だけはずれた( push 1 なのでaddr=1だけ離れたアドレス)と今の sp=2 をスタックに積みます。

つまり、次の命令アドレス(2) + addr(=1) = 3 と、 sp=2をスタックに積みます。
つまり、失敗したときは、 「pc=3の位置の命令を sp=2の位置の文字に対してマッチするところから再開してね」 ということです(つまり a+ のマッチを終了して、 bのマッチにいってね、ということです)

ということで、これのPUSH命令が、バックトラックして再開する位置をスタックに積む処理です。

• JUMP addr

  このJUMP命令の次の命令のアドレスからみてaddrだけ離れた位置にpcを更新します。
  つまり、この命令で別の命令にジャンプすることができます。

  push命令の説明でつかった、命令列でいうと、pc=2の処理になります。
  「次の命令( char b ) のあるアドレス( 3 )からみて addr だけ外れた( jump -3 なのでadd=-3だけ離れたアドレス)にジャンプします。

  つまり、 次の命令のアドレス(3) + addr(=-3) = 0 の位置にジャンプします
  つまり、 また char a の命令に戻って a+ の繰り返し処理を実現する処理です。

• MATCH

  pcがこの命令まで到達したらマッチ成功で終了します。

命令の実装

以上で、命令レベルの詳細はみたので、各正規表現に対応した命令列を組み立てて行きます。
具体的にはパースして作ったASTのノードの種類単位に命令列を定義し、それらを組み合わせて構文木全体の命令列==その正規表現のバイトコードを作っていきます。

まず命令関連のclassを定義しておきます。

module MyRegexp
  class Ir
    OP_CHAR = 0
    OP_PUSH = 1
    OP_JUMP = 2
    OP_MATCH = 3

    # for debug
    OP_NAME = [
      :char,
      :push,
      :jump,
      :match
    ]

    attr_reader :op, :arg1

    def initialize(op, arg1)
      @op = op
      @arg1 = arg1
    end

    def inspect
      [OP_NAME[op], arg1].inspect
    end

    class << self
      def char(char)
        Ir.new(Ir::OP_CHAR, char)
      end

      def push(addr)
        Ir.new(Ir::OP_PUSH, addr)
      end

      def jump(addr)
        Ir.new(Ir::OP_JUMP, addr)
      end

      def match
        Ir.new(Ir::OP_MATCH, nil)
      end

      def compile(ast)
        ast.compile + [Ir.match]
      end
    end
  end
end

Irクラスは、命令一つを表すクラスで、各命令毎にその命令を作成すhelperメソッドを定義しているだけです。(char, push メソッド等)
各ノードでは、自分の正規表現の種類に応じた命令列を生成し、それら全体を再帰的に繋げて、最後に 「マッチした！」 という情報を付与する、 compile メソッドも用意しています。

    def compile(ast)
      ast.compile + [Ir.match]
    end

注: 以降の各ASTのノードクラスは、以前定義したものですが、そこにcompileメソッドを追加していっています。ここではそのcompileメソッドだけ記載してます。

• Charノードの命令列

これは単純で、自分の文字にマッチする命令を生成します。

  class Char < Node
    def compile
      [
        Ir.char(char)
      ]
    end
  end

• Listノードの命令列

これも単純で、連結した各命令列を順番に繋げていくだけです。

  class List < Node
    def compile
      car.compile + cdr.compile
    end
  end

• Branchノードの命令列

ここで初めてバックトラック用の push が発生します。
Branchは a|b のような正規表現で、 a は文字のときもありますが、もっと大きな正規表現の場合もあります。 (ab)|c など。
なので、ここで、 | を挟んだ左の正規表現の命令列、右の正規表現の命令列をそれぞれ、先に left_ir , right_ir として生成しておきます。

で、 メインの | 自体の命令列はというと、

1. まず左の正規表現から試す
2. だが、左の正規表現のマッチが失敗したとき用に、右の正規表現から再開するための情報をスタックに積む
3. その後左の正規表現が成功したら、右をスキップするために、そのマッチ用のコードをまるまるジャンプする命令を実行する

という感じで構築します。

まず、 ステップ1, 2ですが、スタックに再開用の情報を積むので、
push 命令をおきます。引数のaddrは、 次の命令のアドレスから、右の命令までの相対アドレスをセットするので、
ステップ3のジャンプの1命令を考慮して、 左の命令列の命令数＋1(jump分)だけ移動した先をスタックにプッシュします。

その後、左の命令列をそのまま埋め込み、次にステップ３のjump命令を生成します。
jumpは右の正規表現をスキップするjumpなので、右の正規表現の命令列の命令数分jumpします。

これらを考慮して、Branchのcompileは下記のようになります。

  class Branch < Node
    def compile
      left_ir = left.compile
      right_ir = right.compile
      [
        Ir.push(left_ir.size + 1),
        *left_ir,
        Ir.jump(right_ir.size),
        *right_ir
      ]
    end
  end

• Starノードの命令列

Starは exp の正規表現の0回以上の繰り返しになりますが、命令列としては、
繰り返しのたびに、その時点までの状態をスタックに積んでいくのがポイントです（任意の繰り返し後に失敗してもその前から再開できるように)

手順としては

1. まず、0回以上の繰り返しなので、0回の場合（＝マッチしなかった場合)を考慮して、まず、 exp の正規表現をスキップした先から再開する状態をスタックにプッシュ
2. つぎに、 exp の正規表現をそのまま埋め込む
3. 繰り返しなので、またステップ1のpushまで戻るjump命令を生成

という感じで生成します。

ステップ1として、 exp の命令列と ステップ3でのjumpの1命令を考慮して、 expの命令列の命令数+1(jump分)だけ先のアドレスをスタックに積みます。
ステップ2に関しては、上記の説明そのままで下記のようになります。
ステップ3としては、もう一度、 exp にマッチするために上にjumpする命令になるのですが、
jumpの引数が、 exp の命令列の命令数 + 2(push＋jump)命令分戻るようにjumpする感じです。

  class Star < Node
    def compile
      exp_ir = exp.compile
      [
        Ir.push(exp_ir.size + 1),
        *exp_ir,
        Ir.jump(-(exp_ir.size + 2))
      ]
    end
  end

• Plusノードの命令列

PlusもStarと同じく繰り返しなので、似た命令列になるのですが、Starと違い、最低1回はマッチしないといけないので、失敗したときのpushの位置が変わってくるだけです。
具体的には、まず、 exp の正規表現にマッチさせたうえで、その後に pushしていくという感じです。これで最低１回は exp にマッチする、というのが実現できます。

具体的には、 Starと違って、 pushと exp の順番が逆になるだけです。

  class Plus < Node
    def compile
      exp_ir = exp.compile
      [
        *exp_ir,
        Ir.push(1),
        Ir.jump(-(exp_ir.size + 2))
      ]
    end
  end

これで、命令列の生成は終わりです。

VMの構造

さて、やっとVMまで来ました。

上記までで生成した命令を ir と呼ぶとすると、 Vm クラスは、 ir を受け取って、 任意の文字列 str とのマッチを実行し、マッチしたかどうかを返す処理になります。なので概要としては下記のようになります。

module MyRegexp
  class Vm
    attr_reader :ir, :str, :stack
    def initialize(ir)
      # 生成した命令列を受け取る
      @ir = ir
    end

    def match(str)
      # strにirがマッチするか確認します。
      # 本来はここで、sp=0以外からもチェックを実行すると、部分一致の確認ができますが、
      # 今回は未実装で、とりあえず sp=0 ・・・ 文字列の先頭からマッチさせるだけとします。
      sp = 0
      match_at(0, sp, str)
    end
  end
end

Vmの各種スタック操作

スタックの操作である、 push, popを作っていきます。
まずスタックに積むためのデータ構造を一応作っておきます。

といっても、先程見たように、スタックに積むのは、 その時の命令列の位置である pc と、 今マッチしようとしている文字列の位置 sp だけです。一応構造体にして保持しておきます。

VmThread = Struct.new(:pc, :sp)

ちなみに VmThread と名付けているのは、並行処理とかのThreadではなく、　正規表現技術入門 で、分岐一つに対して、 thread という呼び方をしてたのでそれに合わせています。

• スタックにpush

pc, spを受け取り、スタック(ここではただのrubyの配列)に追加します。

    def push(pc, sp)
      stack.push(VmThread.new(pc, sp))
    end

• スタックからpop

今スタックにある一番上のVmThreadを取り出して、 pc, sp を返します。

    def pop
      th = stack.pop
      [th.pc, th.sp]
    end

• スタックの初期化

先程までの説明には出てきませんでしたが、失敗したときようだけでなく、一番先頭のマッチ自体も一つのthreadとみなして、まず最初にpc=0, sp=0(文字列の先頭の場合。先頭でない場合はその場所を渡す)をスタックに積んでおいて、それを最初にpopすることでマッチを実行します。

    def init(sp)
      @stack = []
      push(0, sp)
    end

Vmのメイン処理

さて、ここがVmのメイン処理です。命令列を順に処理しながら、分岐する場所では、スタックに保存、失敗したら(分岐のもう片方に戻るなら)スタックから復元、という処理を記述していきます。

形としては、最初にサンプルで書いた文字の組み合わせ処理のループ版と同じで、一般にバックトラックをループで書く場合下記のような形になります。

def do_process
  スタック初期化
  初期状態をスタックに積む

  while スタックに何か値がある間
    状態x = スタックからpop

    if 状態xが終了状態か？
      成功したので、return
    else
      次の状態を生成してスタックに積む
    end
  end
  失敗
end

さて、これを元にVmのマッチを書いてしまうと、下記のようになります。

    def match_at(sp, pc, str)
      # 上で見たように、ここで、スタックの初期化と、初期状態のpushまで行う
      init(sp)

      # スタックに値が残っている間は、試すべき状態が残っているのでループする
      until stack.empty?
        # スタックの先頭から状態を復元
        pc, sp = pop

        # 今の状態からマッチを続けられるだけ続ける
        loop do
          # pcに該当する命令取得
          code = ir[pc]
          # 合わせてpcも一つ進める
          pc += 1

          # 各命令の個別処理
          case code.op
          when Ir::OP_CHAR
            # OP_CHARの対象文字と、今のspの位置の文字が異なればマッチ失敗
            break if code.arg1 != str[sp]

            # マッチすれば、次の文字とマッチするためにspを進める
            sp += 1
          when Ir::OP_PUSH
            # ここは先程までに説明したとおり、再開するための情報をスタックに積む
            push(pc + code.arg1, sp)
          when Ir::OP_JUMP
            # ここは先程までに説明したとおり、pcからの相対位置にジャンプ(次に実行するアドレスで、pcを上書き)します。
            pc += code.arg1
          when Ir::OP_MATCH
            # ここに到達すればマッチ成功なのでtrueを返します。
            return true
          end
        end
      end
    end

さて、細かくコメントを書いたのでマッチ処理としては以上です。試しに適当な正規表現をコンパイルしてなにかの文字列にマッチしてみます。

今まで作った、parser, transformer, vmを組み合わせるフロントエンドとして、下記のようなクラスを定義します。
astや、irも後から取得できるようにしています。

module MyRegexp
  class Regexp
    attr_reader :ast, :ir
    def initialize(pattern)
      parsed = RegexpParser.new.parse(pattern)
      @ast = RegexpTransformer.new.apply(parsed)
      @ir = Ir.compile(@ast)
    end

    def match(str)
      vm = Vm.new(ir)
      vm.match(str)
    end
  end
end

これを使って下記のように

• ASTの情報
• 生成した命令列
• 実際にマッチさせた結果

# patternの正規表現を元に、test_stringsの各文字列に関してのマッチ結果をmatched, unmatchedでdumpする
# また、patternのAST,生成した命令列もdumpする
def check(pattern, test_strings)
  p [:pattern, pattern]
  reg = MyRegexp::Regexp.new(pattern)
  pp [:ast, reg.ast]

  pp :ir
  puts reg.ir.map.with_index { |e, i|
    no = format("%2d", i)
    [no, e.inspect].join(" ")
  }.join("\n")

  test_strings.each do |str|
    p [:match, str, reg.match(str) ? 'matched' : 'unmatched']
  end

  puts
end

check('a*b', %w[b ab aab cb])

これを実行すると、下記のように出力されます。

# 元の正規表現文字列
[:pattern, "a*b"]

# パース結果のAST
[:ast, [(* a) b]]

# 生成した命令列
:ir
 0 [:push, 2]
 1 [:char, "a"]
 2 [:jump, -3]
 3 [:char, "b"]
 4 [:match, nil]

#各文字列に先頭からマッチさせた結果
[:match, "b", "matched"]
[:match, "ab", "matched"]
[:match, "aab", "matched"]
[:match, "cb", "unmatched"]

ちゃんと、 aがゼロ回以上繰り返し、その後にbがくる文字列 だけがマッチして、それ以外の(cb)はunmatchedになっていますね。

最後に

今まで名前だけは聞いたことがあった、バイトコードインタープリターとかなんとなくイメージだけはありましたが、実際に作ってみると理解が深まってよかったです。
また、命令列の生成や、Vmの処理自体は、実際書いているときは、個別にみたらあってそうやけど、つなげてうまく動くんやろかと疑問に思いながら書いていましたが、実際実行してみると、なんか魔法のようにうまく組み合わさって動いてて、結構感動しました。

ここまでに書いたコードはgithubに置いています。