ループ処理を関数型っぽく書いてみる(1)

今週は、common lispでループ処理を関数型っぽく、かつ効率良く実装できるかどうかを試していたので、その結果を載せておく。
結論から云えば、処理系の十分な最適化を期待できれば、関数型っぽく書いても、手続き型的に書いた場合と比肩しえる性能が得られそうな感じだった。

成果物

作ったもの: loop
今回はこれの使用例とベンチマーク結果を、次回は実装方法を載せる予定。

使用例

シーケンス(or 無限シーケンス)とmapとかfilterとかを組み合わせてループを表現する。

;; 1から5までの数値を表示する
> (loop:each (lambda (n) (print n))
             (loop:from 1 :to 5))
1
2
3
4
5
=> NIL

> (loop:from 1 :to 5)
=> #<CLOSURE (LAMBDA () :IN LOOP:FROM) {10030E0E0B}> ; 実態はクロージャー

;; マッピング
> (loop:collect (loop:map (lambda (c) (list c (char-code c)))
                          (loop:for-string "mapping")))
=> ((#\m 109) (#\a 97) (#\p 112) (#\p 112) (#\i 105) (#\n 110) (#\g 103))

;; フィルター
> (loop:collect (loop:filter #'oddp (loop:from 1 :to 10)))
-> (1 3 5 7 9)

;;; fizzbuzz
> (defun fizzbuzz-seq ()
    (loop:filter #'consp
      (loop:map (lambda (n) (cond ((zerop (mod n 15)) (cons n :fizzbuzz))
                                  ((zerop (mod n  5)) (cons n :buzz))
                                  ((zerop (mod n  3)) (cons n :fizz))
                                  (t nil)))
                (loop:from 1))))

;; 先頭三つ
> (loop:collect (loop:take 3 (fizzbuzz-seq)))
=> ((3 . :FIZZ) (5 . :BUZZ) (6 . :FIZZ))

;; 10から12番目
> (loop:collect (loop:take 2 (loop:drop 10 (fizzbuzz-seq))))
=> ((24 . :FIZZ) (25 . :BUZZ))

;; 100以下かつ偶数のものだけ
> (loop:collect 
   (loop:take-while (lambda (x) (<= (car x) 100)) 
                    (loop:filter (lambda (x) (evenp (car x)))
                                 (fizzbuzz-seq))))
=> ((6 . :FIZZ) (10 . :BUZZ) (12 . :FIZZ) (18 . :FIZZ) (20 . :BUZZ) (24 . :FIZZ)
    (30 . :FIZZBUZZ) (36 . :FIZZ) (40 . :BUZZ) (42 . :FIZZ) (48 . :FIZZ)
    (50 . :BUZZ) (54 . :FIZZ) (60 . :FIZZBUZZ) (66 . :FIZZ) (70 . :BUZZ)
    (72 . :FIZZ) (78 . :FIZZ) (80 . :BUZZ) (84 . :FIZZ) (90 . :FIZZBUZZ)

;; zip
> (loop:collect 
    (loop:take 3
      (loop:map-n 3 (lambda (x y z) (list x y z))  ; zipと組み合わせる場合は XXX-n 系のマクロを使用して、引数の数を指定する
        (loop:zip (loop:filter #'oddp (loop:from 1))
                  (loop:down-from 100 :by 3)
                  (loop:map #'sqrt (loop:repeat (lambda () (random 1000))))))))
=> ((1 100 19.078785) 
    (3 97 19.052559) 
    (5 94 16.309507))

;; フィボナッチ数列を定義
(defun fib-seq () 
  (let ((n+1 1))
    (declare (fixnum n+1))
    (loop:make-generator 
     :init (lambda () 0)                             ; 初期値生成関数
     :next (lambda (n) (prog1 n+1 (incf n+1 n)))     ; 値更新関数
     :end? (lambda (n) (declare (ignore n)) nil))))  ; 終端判定関数

> (loop:collect (loop:take 20 (fib-seq)))
=> (0 1 1 2 3 5 8 13 21 34 55 89 144 233 377 610 987 1597 2584 4181)

速度比較

loopマクロやdoを使って書いた場合との速度比較。
処理系はSBCL(1.0.54)。

比較1: sum関数 (単純なループ)

;; 準備
(defparameter *fastest* '(optimize (speed 3) (safety 0) (debug 0) (compilation-speed 0)))
(defparameter *note* '(sb-ext:unmuffle-conditions sb-ext:compiler-note))

;; loopパッケージ版
(defun sum1 (start end)
  (declare (fixnum start end) #.*fastest* #.*note*)
  (loop:reduce (lambda (acc x) (the fixnum (+ acc x)))
               0
               (loop:from start :to end)))

;; loopマクロ版
(defun sum2 (start end)
  (declare (fixnum start end) #.*fastest* #.*note*)
  (loop WITH total fixnum = 0
        FOR i FROM start TO end
        DO (incf total i)
        FINALLY (return total)))  ; (loop ... SUM i) では最適化されない部分があるので、少し長くなるけどこちらを採用

;; do版
(defun sum3 (start end)
  (declare (fixnum start end) #.*fastest* #.*note*)
  (let ((total 0))
    (declare (fixnum total))
    (do ((i start (1+ i)))
        ((> i end) total)
      (incf total i))))

;; 実行
> (time (sum1 1 100000000))  ; loopパッケージ版: 0.084秒
Evaluation took:
  0.084 seconds of real time
  0.084006 seconds of total run time (0.084006 user, 0.000000 system)
  100.00% CPU
  167,231,593 processor cycles
  0 bytes consed
=> 5000000050000000

> (time (sum2 1 100000000)) ; loopマクロ版: 0.086秒
Evaluation took:
  0.086 seconds of real time
  0.088005 seconds of total run time (0.088005 user, 0.000000 system)
  102.33% CPU
  171,410,077 processor cycles
  0 bytes consed
=> 5000000050000000

> (time (sum3 1 100000000)) ; do版: 0.083秒
Evaluation took:
  0.083 seconds of real time
  0.084005 seconds of total run time (0.084005 user, 0.000000 system)
  101.20% CPU
  166,793,649 processor cycles
  0 bytes consed
=> 5000000050000000

単純なループ処理なら、どれも速度は同じくらい。

比較2: 数値リストの奇数番目の要素の平均値を求める (zipを使ったループ)

;; データ準備: 1000万要素のリスト
(defparameter *list* (loop REPEAT 10000000 COLLECT (random 100000)))

;; loopパッケージ版
(defun avg1 (list)
  (declare #.*fastest* #.*note*)
  (flet ((average (sequence)  ; シーケンスの生成と平均値を求める処理を分離することが可能
           (let ((total 0)
                 (count 0))
             (declare (fixnum total count))
             (loop:each (lambda (n)
                          (incf total n)
                          (incf count))
                        sequence)
             (float (/ total count)))))

    (let ((seq (loop:map-n 2 (lambda (_ n) n)
                 (loop:filter-n 2 (lambda (i _) (oddp i))
                   (loop:zip (loop:from 0 :to most-positive-fixnum)
                             (loop:for-list list :element-type fixnum))))))
      (average seq))))

;; loopマクロ版
(defun avg2 (list)
  (declare #.*fastest* #.*note*)
  (loop WITH total fixnum = 0
        WITH count fixnum = 0
        FOR i fixnum FROM 0
        FOR n fixnum IN list
        WHEN (oddp i)
    DO
    (incf total n)
    (incf count)
    FINALLY
    (return (float (/ total count)))))

;; do版
(defun avg3 (list)
  (declare #.*fastest* #.*note*)
  (let ((total 0)
        (count 0))
    (declare (fixnum total count))
    (do ((i 0 (1+ i))
         (head list (cdr head)))
        ((endp head))
      (declare (fixnum i))
      (when (oddp i)
        (incf count)
        (incf total (the fixnum (car head)))))
    (float (/ total count))))

;; 実行
> (time (avg1 *list*))  ; loopパッケージ版: 0.084秒
Evaluation took:
  0.084 seconds of real time
  0.084005 seconds of total run time (0.084005 user, 0.000000 system)
  100.00% CPU
  166,739,958 processor cycles
  0 bytes consed
=> 50003.64

> (time (avg2 *list*))  ; loopマクロ版: 0.036秒
Evaluation took:
  0.036 seconds of real time
  0.036002 seconds of total run time (0.036002 user, 0.000000 system)
  100.00% CPU
  72,645,764 processor cycles
  0 bytes consed
=> 50003.64

> (time (avg3 *list*))  ; do版: 0.037秒
Evaluation took:
  0.037 seconds of real time
  0.040003 seconds of total run time (0.040003 user, 0.000000 system)
  108.11% CPU
  75,246,648 processor cycles
  0 bytes consed 
=> 50003.64

loopパッケージ版はzipを通すと、loopマクロやdoを使った場合の半分以下になってしまう。

比較3: 比較2での、平均値算出部分と奇数番要素のフィルタ部分を、別関数に分けた場合

;; loopマクロで、平均値算出部分と奇数番要素のフィルタ部分を、別関数に分けた場合
(declaim (inline average-list))
(defun average-list (list)
  (declare #.*fastest* #.*note*)
  (loop WITH total fixnum = 0
        WITH count fixnum = 0
        FOR n fixnum IN list
    DO
    (incf total n)
    (incf count)
    FINALLY
    (return (float (/ total count)))))

(declaim (inline filter-list))
(defun filter-list (list)
  (declare #.*fastest* #.*note*)
  (loop FOR i fixnum FROM 0
        FOR x IN list
        WHEN (oddp i)
        COLLECT x))

> (time (average-list (filter-list *list*)))
Evaluation took:
  0.122 seconds of real time              ; 全体で0.122秒、GC抜きなら0.081秒
  0.120008 seconds of total run time (0.120008 user, 0.000000 system)
  [ Run times consist of 0.040 seconds GC time, and 0.081 seconds non-GC time. ]
  98.36% CPU
  244,986,221 processor cycles
  79,986,688 bytes consed
=> 50003.64

;; do版
;; ※ loopマクロ版とほとんど変わらないので省略

;; loopパッケージで、平均値算出部分と奇数番要素のフィルタ部分を、別関数に分けた場合
(declaim (inline average-loop))
(defun average-loop (sequence)
  (declare #.*fastest* #.*note*)
  (let ((total 0)
        (count 0))
    (declare (fixnum total count))
    (loop:each (lambda (n)
                 (incf total (the fixnum n))
                 (incf count))
               sequence)
    (float (/ total count))))

(declaim (inline filter-loop))
(defun filter-loop (list)
  (declare #.*fastest* #.*note*)
  (loop:map-n 2 (lambda (_ n) n)
    (loop:filter-n 2 (lambda (i _) (oddp i))
      (loop:zip (loop:from 0 :to most-positive-fixnum)
                (loop:for-list list :element-type fixnum)))))

> (time (average-loop (filter-loop *list*)))
Evaluation took:
  0.070 seconds of real time   ; 0.070秒
  0.072005 seconds of total run time (0.072005 user, 0.000000 system)
  102.86% CPU
  139,856,631 processor cycles
  0 bytes consed
=> 50003.64

;; loopパッケージでinline宣言を外した場合
(declaim (notinline average-loop))
(declaim (notinline filter-loop))
> (time (average-loop (filter-loop *list*)))
Evaluation took:
  0.378 seconds of real time  ; 0.378秒
  0.376024 seconds of total run time (0.376024 user, 0.000000 system)
  99.47% CPU
  754,498,049 processor cycles
  0 bytes consed
=> 50003.64

loopマクロやdoマクロでは、ループ処理の一部を自然な形で効率よく外だしするのが困難なので、そういった用途ではloopパッケージの方が性能が良い。
ただし、現状のloopパッケージは、inline展開による最適化に過度に依存しているので、展開が効かないケースでは、いっきに処理速度が遅くなってしまう。