ソート済みのリストに対する破壊的マージソートの改良

以前に載せたマージソート(をベースとしたもの)をSBCL(1.0.58)にコミットしてくれたPaul Khuongさんが、こんな記事を書いていて、なるほどなー、と思ったので、表題に関係する部分を参考にさせて貰って変更前後での比較を行ったメモ。

オリジナルのマージソート

まず、SBCL(1.0.58)のリストに対する破壊的マージソートの実装*1:

;; 二つのソート済みリストのマージ関数
(declaim (inline merge-lists*))
(defun merge-lists* (head list1 list2 test key &aux (tail head))
  (declare (type cons head list1 list2)
           (type function test key)
           (optimize speed))
  (macrolet ((merge-one (l1 l2)
               `(progn
                  (setf (cdr tail) ,l1
                        tail       ,l1)
                  (let ((rest (cdr ,l1)))
                    (cond (rest
                           (setf ,l1 rest))
                          (t
                           (setf (cdr ,l1) ,l2)
                           (return (cdr head))))))))
    (loop
     (if (funcall test (funcall key (car list2))  ; this way, equivalent
                       (funcall key (car list1))) ; values are first popped
         (merge-one list2 list1)                  ; from list1
         (merge-one list1 list2)))))

;; 実行
(merge-lists* '(:head) '(1 3 5) '(2 4 6) #'< #'identity))
=> (1 2 3 4 5 6)
;; リストのマージソート関数
(declaim (inline stable-sort-list))
(defun stable-sort-list (list test key &aux (head (cons :head list)))
  (declare (type list list)
           (type function test key)
           (dynamic-extent head))
  (labels ((recur (list size)
             (declare (optimize speed)
                      (type cons list)
                      (type (and fixnum unsigned-byte) size))
             (if (= 1 size)
                 (values list (shiftf (cdr list) nil))
                 (let ((half (ash size -1)))
                   (multiple-value-bind (list1 rest)
                       (recur list half)
                     (multiple-value-bind (list2 rest)
                         (recur rest (- size half))
                       (values (merge-lists* head list1 list2 test key)
                               rest)))))))
    (when list
      (values (recur list (length list))))))

;; 実行
(stable-sort-list '(8 73 2 40 0 3) #'< #'identity)
=> (0 2 3 8 40 73)

何種類かデータを用意して実行時間を計測:

;;; 計測用データ
;; 1] 400万要素のソート済みリスト
(defparameter *sorted-list* (loop FOR i FROM 0 BELOW 4000000 COLLECT i))

;; 2] 400万要素の逆順ソート済みリスト
(defparameter *reverse-sorted-list* (reverse *sorted-list*))

;; 3] 400万要素のほぼソート済みリスト1  ※ 千要素に一つがランダムな値
(defparameter *nearly-sorted-list1* (loop FOR i FROM 0 BELOW 4000000
                                         COLLECT (if (zerop (random 1000))
                                                     (random 4000000)
                                                   i)))

;; 4] 400万要素のほぼソート済みリスト2  ※ 複数のソート済みリストが連結
(defparameter *nearly-sorted-list2* (loop REPEAT 4 APPEND (loop FOR i FROM 0 BELOW 1000000 COLLECT i)))

;; 5] 400万要素のランダムなリスト
(defparameter *random-list* (loop REPEAT 4000000 COLLECT (random most-positive-fixnum)))


;;; 計測用マクロ
(defmacro sort-time (sort-fn-name list)
  `(let ((list~ (copy-list ,list)))
     (declare (optimize (speed 3) (safety 0)))
     (time (progn (,sort-fn-name list~ #'< #'identity)
                  t))))


;;; 計測
;; 1] ソート済みリスト
(sort-time stable-sort-list *sorted-list*)
Evaluation took:
  0.254 seconds of real time  ; 0.254秒
  0.252017 seconds of total run time (0.248016 user, 0.004001 system)
  99.21% CPU
  508,247,464 processor cycles
  0 bytes consed
=> T

;; 2] 逆順ソート済みリスト
(sort-time stable-sort-list *reverse-sorted-list*)
Evaluation took:
  0.235 seconds of real time  ; 0.235秒
  0.232015 seconds of total run time (0.232015 user, 0.000000 system)
  98.72% CPU
  468,869,834 processor cycles
  0 bytes consed
=> T

;; 3] ほぼソート済みリスト1  ※ 千要素に一つがランダムな値
(sort-time stable-sort-list *nearly-sorted-list1*)
Evaluation took:
  0.348 seconds of real time  ; 0.348秒
  0.348023 seconds of total run time (0.344022 user, 0.004001 system)
  100.00% CPU
  694,968,622 processor cycles
  0 bytes consed
=> T

;; 4] ほぼソート済みリスト2  ※ 複数のソート済みリストが連結
(sort-time stable-sort-list *nearly-sorted-list2*)
Evaluation took:
  0.271 seconds of real time  ; 0.271秒
  0.272017 seconds of total run time (0.272017 user, 0.000000 system)
  100.37% CPU
  538,952,732 processor cycles
  0 bytes consed
=> T

;; 5] ランダムリスト
(sort-time stable-sort-list *random-list*)
Evaluation took:
  2.171 seconds of real time  ; 2.171秒
  2.168135 seconds of total run time (2.160135 user, 0.008000 system)
  99.86% CPU
  4,332,215,938 processor cycles
  0 bytes consed
=> T

ソート済みのリストに対する改良を加えたマージソート

変更後のマージソート関数: ※ 変更内容はコメントを参照

;; 改良版マージソート関数
;; - fast-merge-lists*関数が追加されたこと以外は、もともとの関数とほとんど同様
;; - fast-merge-lists*関数は要素の範囲が重複しない二つのリストをO(1)でマージ可能
(declaim (inline stable-sort-list2))
(defun stable-sort-list2 (list test key &aux (head (cons :head list)))
  (declare (type list list)
           (type function test key)
           (dynamic-extent head))
        
           ;; マージ対象の二つのリスト内の片方が、もう片方に完全に先行している場合は、
           ;; 各要素の比較などは省略して、末尾のcdrの更新のみを行う。
  (labels ((fast-merge-lists* (try-fast-merge? list1 tail1 list2 tail2 rest)
             (when try-fast-merge?
                      ;; list1がlist2に完全に先行: (list1 .. tail1) <= (list2 .. tail2)
               (cond ((not (funcall test (funcall key (car list2))
                                         (funcall key (car tail1))))
                      (setf (cdr tail1) list2)
                      (return-from fast-merge-lists* (values list1 tail2 rest)))

                      ;; list2がlist1に完全に先行: (list2 .. tail2) < (list1 .. tail1)
                     ((funcall test (funcall key (car tail2))
                                    (funcall key (car list1)))
                      (setf (cdr tail2) list1)
                      (return-from fast-merge-lists* (values list2 tail1 rest)))))
             
             ;; その他: 通常のマージ
             (values (merge-lists* head list1 list2 test key)
                     (if (null (cdr tail1))
                         tail1
                       tail2)
                     rest))
                  
            ;; トップダウンマージリスト関数: リストの末尾を管理するようになったのとfast-merge-lists*関数を使うようになったこと以外は変更なし            
            (recur (list size)
             (declare (optimize speed)
                      (type cons list)
                      (type (and fixnum unsigned-byte) size))
             (if (= 1 size)
                 (values list list (shiftf (cdr list) nil))
                 (let ((half (ash size -1)))
                   (multiple-value-bind (list1 tail1 rest)
                       (recur list half)
                     (multiple-value-bind (list2 tail2 rest)
                         (recur rest (- size half))
                       (fast-merge-lists* (>= size 8)  ; オーバヘッドを少なくするために、一定サイズ以上のリストに対してのみ適用を試みる
                                          list1 tail1 list2 tail2 rest)))))))
    (when list
      (values (recur list (length list))))))

;; 実行
(stable-sort-list2 '(8 73 2 40 0 3) #'< #'identity)
=> (0 2 3 8 40 73)

処理時間計測:

;; 1] ソート済みリスト
(sort-time stable-sort-list2 *sorted-list*)
Evaluation took:
  0.086 seconds of real time  ; 0.086秒  (変更前: 0.254秒)
  0.088005 seconds of total run time (0.088005 user, 0.000000 system)
  102.33% CPU
  171,845,432 processor cycles
  0 bytes consed
=> T

;; 2] 逆順ソート済みリスト
(sort-time stable-sort-list2 *reverse-sorted-list*)
Evaluation took:
  0.087 seconds of real time  ; 0.0.87秒  (変更前: 0.235秒)
  0.088006 seconds of total run time (0.088006 user, 0.000000 system)
  101.15% CPU
  173,196,084 processor cycles
  0 bytes consed
=> T

;; 3] ほぼソート済みリスト1  ※ 千要素に一つがランダムな値
(sort-time stable-sort-list2 *nearly-sorted-list1*)
Evaluation took:
  0.293 seconds of real time  ; 0.293秒  (変更前: 0.348秒)
  0.292019 seconds of total run time (0.292019 user, 0.000000 system)
  99.66% CPU
  585,393,530 processor cycles
  0 bytes consed
=> T

;; 4] ほぼソート済みリスト2  ※ 複数のソート済みリストが連結
(sort-time stable-sort-list2 *nearly-sorted-list2*)
Evaluation took:
  0.122 seconds of real time  ; 0.122秒  (変更前: 0.271秒)
  0.120007 seconds of total run time (0.116007 user, 0.004000 system)
  98.36% CPU
  242,403,024 processor cycles
  0 bytes consed
=> T

;; 5] ランダムリスト
(sort-time stable-sort-list2 *random-list*)
Evaluation took:
  2.193 seconds of real time  ; 2.193秒  (変更前: 2.171秒)
  2.192138 seconds of total run time (2.164136 user, 0.028002 system)
  99.95% CPU
  4,376,336,316 processor cycles
  0 bytes consed
=> T

完全にランダムなリストに対するソートは心なしか改良版の方が(ごく若干)遅くなっているように思うが、入力リストにソート済みの部分が多ければ多いほど、確実に改良版の方が速くなっている。
確かに、二つのリストをマージする場合、それぞれの領域が独立しているなら、片方の先頭要素ともう片方の末尾要素を比較するだけで、リスト全体を完全に順序づけ可能なんだけど、自分が実装方法を考えている時には、そのことに思い至らなかった。
なるほどなー。

*1:sbcl-1.0.58/src/code/sort.lisp より引用