Sanmoku(0.0.5): 原型や読みの情報取得に対応

Sanmoku(0.0.4): 辞書データサイズ縮小のコメントにて要望があったのでSanmokuを形態素の原型や読みの情報取得に対応させてみた。
Sanmoku本体のインターフェースは以前の同様*1で、原型・読み・発音の取得を行うためのFeatureExクラス(sanmoku-feature-ex-x.x.x.jar)を新しく追加した。

使用例:

import net.reduls.sanmoku.Tagger;
import net.reduls.sanmoku.Morpheme;
import net.reduls.sanmoku.FeatureEx; // 追加

for(Morpheme m : Tagger.parse("...")) {  // 形態素解析
  FeatureEx fe = new FeatureEx(m);       // 解析結果の形態素インスタンスから、追加の素性データを取得

  // 結果表示
  System.out.println(m.surface+"\t"+m.feature+","+fe.baseform+","+fe.reading+","+fe.pronunciation);
}

$ export CLASSPATH=sanmoku-0.0.5.jar:sanmoku-feature-ex-0.0.1.jar
$ echo '招待制でリリースしました' | java net.reduls.sanmoku.bin.SanmokuFeatureEx
招待	名詞,サ変接続,*,*,*,*,招待,ショウタイ,ショータイ
制	名詞,接尾,一般,*,*,*,制,セイ,セイ
で	助詞,格助詞,一般,*,*,*,で,デ,デ
リリース	名詞,サ変接続,*,*,*,*,リリース,リリース,リリース
し	動詞,自立,*,*,サ変・スル,連用形,する,シ,シ
まし	助動詞,*,*,*,特殊・マス,連用形,ます,マシ,マシ
た	助動詞,*,*,*,特殊・タ,基本形,た,タ,タ
EOS

比較

以前の表にSanmoku-0.0.5(+ FeatureEx-0.0.1)を追加したもの。

	辞書データサイズ(IPADIC)	最小所要メモリ(-Xmx)	起動(≒辞書ロード)時間*2	10MBテキストの解析時間
Igo-0.4.3	40MB	73MB	0.058秒	2.729秒
Gomoku-0.0.4	8.2MB	23MB	0.371秒	2.621秒
Sanmoku-0.0.4	4.8MB	2MB	0.057秒	5.807秒
Sanmoku-0.0.5(+ FeatureEx-0.0.1)	10MB	11MB	0.098秒	6.814秒

読み等の情報を取得した場合、性能は全体的に劣化しているけど、これくらいなら十分許容範囲内のような気がする。

*1:内部的にはMorphemeクラスが形態素IDを保持するように修正されている

*2:JVM自体の起動時間は除いた数値

UNF-0.0.4: サイズ削減

今日は久しぶりにUNF(ユニコード正規化ライブラリ)に手を加えていた。
大きな変更点は、正規化用変換テーブルを実現していたTRIEをDAWGにしたこと。
もともとは正規分解と互換分解用に、内容がほぼ等しいTRIEを別々に持っていたので、それを一つDAWGにして共有することでだいぶサイズが節約できた。

# unf-0.0.3
$ ls -lh unf-0.0.3/bin/unf
-rwxrwxr-x 1 user user 596K 2011-11-19 17:54 unf-0.0.3/bin/unf  # 596KB

# unf-0.0.4
$ ls -lh unf-0.0.4/bin/unf
-rwxrwxr-x 1 user user 411K 2011-11-19 20:20 unf-0.0.4/bin/unf  # 411KB

処理速度も、ごく僅かだけど新しいバージョンの方が速くなっているように見える。

# 17MBのテキストデータの正規化時間

# unf-0.0.3
$ unf-0.0.3/bin/unf-time < 17MB.txt 
= read: 
  == 172090 lines
  == average(line length): 99 byte
= time: 
  == NFD :  0.203354 sec
  == NFC :  0.109814 sec
  == NFKD:  0.215196 sec
  == NFKC:  0.137385 sec
DONE

# unf-0.0.4
$ unf-0.0.4/bin/unf-time < 17MB.txt 
= read: 
  == 172090 lines
  == average(line length): 99 byte
= time: 
  == NFD :  0.199866 sec
  == NFC :  0.104912 sec
  == NFKD:  0.206675 sec
  == NFKC:  0.137277 sec
DONE

Sanmoku(0.0.4): 辞書データサイズ縮小

この一週間でSanmokuの辞書データサイズの縮小をいろいろ試していたので、その結果を載せておく。
現時点でのバージョンは 0.0.4。

やったこと

試した主なこと。

データ	内容	サイズ (Gomoku-0.0.4 => Sanmoku-0.0.4)
連接コストデータ (matrix.bin)	類似品詞の連接コストを併合*1 + コスト値を14bitで保持	3.5MB => 2.2MB
形態素辞書引きインデックス (surface-id.bin)	2バイト文字(UTF-16)DAWGから、1バイト文字(UTF-8)DAWGに変更。 かつIPADICに合わせてノードレイアウトを最適化	2.7MB => 1.5MB
形態素データ (morpheme.bin,id-morphems-map.bin)	4バイト(品詞情報:2バイト、単語コスト:2バイト)から2バイトに	1.8MB => 1.0MB

比較

IgoとGomokuとSanmokuの比較。

	辞書データサイズ(IPADIC)	最小所要メモリ(-Xmx)	起動(≒辞書ロード)時間*2	10MBテキストの解析時間
Igo-0.4.3	40MB	73MB	0.058秒	2.729秒
Gomoku-0.0.4	8.2MB	23MB	0.371秒	2.621秒
Sanmoku-0.0.4	4.8MB	2MB	0.057秒	5.807秒

Sanmokuは所要メモリや辞書ロード時間が短いが、辞書データを圧縮するためにビット演算や間接参照等を多用しているため、解析速度は他に比べて二倍以上遅くなっている。
Igoは辞書データサイズ自体は大きいが、mmap(java.nio.MappedByteBuffer)を利用しているため、ロード時間は高速となっている。

*1:このためSanmokuは、若干(Gomokuに比べ1%にも満たない程度だが)解析精度が落ちている。

*2:JVM自体の起動時間は除いた数値

Sanmoku: 省メモリな形態素解析器

GomokuをベースにしたSanmokuという形態素解析器を実装した。
Gomokuに比べて解析時に必要なメモリ量が少ないのと初期ロード時間が短いのが特徴。
将来的には解析精度を若干落として、辞書サイズ*1をさらに削減する可能性もあるけど、現状は解析結果はGomoku互換。
Android等のリソースの制限が厳しい環境での使用を想定。

最低メモリ所要量とロード時間

以下、自分の環境*2での計測結果。

## 最低メモリ所要量
# Gomoku(0.0.4)は 26MBのメモリが必要
$ java -Xmx26m -cp gomoku-0.0.4.jar net.reduls.gomoku.bin.Gomoku < /path/to/natsume-soseki.txt > /dev/null

# Sanmoku(0.0.1)は 11MBのメモリが必要
$ java -Xmx11m -cp sanmoku-0.0.1.jar net.reduls.sanmoku.bin.Sanmoku < /path/to/natsume-soseki.txt > /dev/null


## ロード時間
# Gomoku(0.0.4)は 0.633秒 (内 0.094秒はJVM起動時間)
time echo 'a' | java -Xmx26m -cp gomoku-0.0.4.jar net.reduls.gomoku.bin.Gomoku
a	名詞,固有名詞,組織,*,*,*
EOS

real	0m0.633s
user	0m0.808s
sys	0m0.044s

# Sanmoku(0.0.1)は 0.217秒 (内 0.094秒はJVM起動時間)
time echo 'a' | java -Xmx11m -cp sanmoku-0.0.1.jar net.reduls.sanmoku.bin.Sanmoku 
a	名詞,固有名詞,組織,*,*,*
EOS

real	0m0.217s
user	0m0.244s
sys	0m0.024s

Android

https://github.com/sile/sanmoku/downloads に Sanmoku-0.0.1.apk という名前でサンプルAndroidアプリを配置。
自分の環境(HTC EVO WiMAX ISW11HT)でしか動作確認していないので、他のスマートフォンで正常に動くかどうかは不明。

辞書ロード時間は、Gomokuに比べるとだいぶ短縮されてはいるが、それでも一番初めの解析が始まるまで、現状では数秒程度の時間を要する。

*1:現状はJAR展開時で7.6MB

*2:Linxu, x86-64bit

マージソート(3): 高階関数呼び出し最適化

マージソート(1)の改良版。
ソートのような高階関数では、引数で渡した比較関数の間接呼び出しのコストも実行速度にそれなりの影響を与えるので、それを(マクロをほとんど使わずに)できるだけ低く抑えるための試み。

比較関数最適化

まず、比較関数自体の実行コストを下げるために、汎用的な数値比較関数ではなく、より特殊化されたものを使用するようにする。

;; fixnum用の比較関数
(define-symbol-macro fixnum< (lambda (x y) (declare (fixnum x y)) (< x y)))

;;; fixnum< を使用した場合の処理時間
;;;
;;; 大量データのソート速度
;;; 100万要素のリストが対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 1000000 COLLECT (random 10000000)))
        (d1 (copy-seq data))
        (d2 (copy-seq data))
        (r1 (time (stable-sort d1 fixnum<)))
        (r2 (time (merge-sort:sort d2 fixnum<))))
   (equal r1 r2)))

Evaluation took:
  1.366 seconds of real time  ; stable-sort# 1.366秒 (前回 2.484秒)
  1.360085 seconds of total run time (1.360085 user, 0.000000 system)
  99.56% CPU
  2,723,515,890 processor cycles
  0 bytes consed
  
Evaluation took:
  0.541 seconds of real time  ; merge-sort:sort# 0.541秒 (前回 1.662秒)
  0.540034 seconds of total run time (0.540034 user, 0.000000 system)
  99.82% CPU
  1,079,254,874 processor cycles
  0 bytes consed
  
--> T

後は、ここからどれだけ短縮できるか。

実装

今回のマージソート実装。
八割方は前と同じ。
まず、ほとんど変更がない前半部分から載せる。(変更箇所はコメントで記載)

(defpackage merge-sort
  (:use common-lisp)
  (:shadow :common-lisp sort)
  (:export sort))
(in-package :merge-sort)

;; inline-sort関数とsort-impl関数がinline宣言に加わっている。前者は今回新たに追加される関数
(declaim (inline halve merge-lists inline-sort sort-impl)  
         (optimize (speed 3) (debug 0) (safety 0)))

(defun halve (n)
  (declare (fixnum n))
  (multiple-value-bind (n1 x) (floor n 2)
    (values (+ n1 x) n1)))

(defmacro cdr! (list new-cdr)
  `(setf (cdr ,list) ,new-cdr))

(defmacro multiple-value-let* (bind-specs &body body)
  (if (null bind-specs)
      `(locally ,@body)
    (destructuring-bind ((vars exp) . rest) bind-specs
      `(multiple-value-bind ,vars ,exp
         (multiple-value-let* ,rest ,@body)))))
   
(defun merge-lists (list1 list2 test key)
  (declare (function test key))
  (labels ((less-equal-than (list1 list2)  ; 安定ソートになるように比較方法が若干修正されている
             (not (funcall test (funcall key (car list2)) (funcall key (car list1)))))
           (recur (head tail l1 l2)
             (cond ((null l1)               (cdr! tail l2) head)
                   ((null l2)               (cdr! tail l1) head)
                   ((less-equal-than l1 l2) (recur head (cdr! tail l1) (cdr l1) l2))
                   (t                       (recur head (cdr! tail l2) l1 (cdr l2))))))
    (declare (inline less-equal-than))
    (if (less-equal-than list1 list2)
        (recur list1 list1 (cdr list1) list2)
      (recur list2 list2 list1 (cdr list2)))))

次はsort-impl関数。
量は多くないけど、ここが一番重要な変更箇所。

;; 前回は、sort-impl関数自体で再帰処理を行っていたのを、
;; 再帰部分をrecur関数に括り出すように修正。
;;
;; これによって、sort-impl関数に対してinline宣言を行うことが可能になる。
;;
;; sort-impl関数がinline展開可能となると、
;; inline-sort関数(後述) => sort-impl関数 => merge-lists関数、の
;; 全てがinline展開されるようになるため、
;; コンパイラが(inline-sort関数の引数で渡され)merge-lists関数内でfuncallされている
;; testとkeyの情報を知ることができるようになり、間接呼び出しを除去する等といった
;; 最適化が可能となる(と思っている)。
(defun sort-impl (list size test key)
  (labels ((recur (list size)
             (declare (fixnum size))
             (if (= 1 size)
                 (values list (prog1 (cdr list) (cdr! list nil)))
               (multiple-value-let* (((size1 size2) (halve size))
                                     ((list1 rest) (recur list size1))
                                     ((list2 rest) (recur rest size2)))
                 (values (merge-lists list1 list2 test key) rest)))))
    (recur list size)))

最後はsort関数。
inline展開の有無を選択するための引数を追加してみた。
※ 単にsort関数をinline宣言するだけでも良いのだが、常に展開されるようになってしまうのも避けたかったの若干(無駄に)凝ってみた

;; inline引数を追加。これが真の場合は、inline展開される。
(defun sort (list test &key (key #'identity) inline)
  (declare (list list)
           (function test key)
           (ignore inline)
           (optimize (speed 3) (safety 2) (debug 2)))
  (when list
    (values (sort-impl list (length list) test key))))

;; sort関数のinline展開版。上でinline宣言されている以外は、sort関数と基本的に同様。
(defun inline-sort (list test &key (key #'identity))
  (declare (list list)
           (optimize (speed 3) (safety 0) (debug 0)))
  (when list
    (values (sort-impl list (length list) test key))))

;; sort関数のinline引数が真の場合に、(sort関数の代わりに)inline-sort関数を呼び出すためのコンパイラマクロ
(define-compiler-macro sort (&whole form list test &key (key '#'identity) inline)
  (if inline
      `(inline-sort ,list ,test :key ,key)
    form))

前回*1と比べて、本質的に異なるのは、sort-impl関数がinline展開可能になった、という点だけ。

計時

今回追加した関数(オプション)を加えて、再度計測。

;;; 大量データのソート速度
;;; 100万要素のリストが対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 1000000 COLLECT (random 10000000)))
        (d1 (copy-seq data))
        (d2 (copy-seq data))
        (d3 (copy-seq data))
        (r1 (time (stable-sort d1 fixnum<)))
        (r2 (time (merge-sort:sort d2 fixnum<)))
        (r3 (time (merge-sort:sort d3 fixnum< :inline t)))) 
   (list (equal r1 r2)
         (equal r1 r3))))

Evaluation took:
  1.336 seconds of real time  ; stable-sort# 1.336秒
  1.332083 seconds of total run time (1.332083 user, 0.000000 system)
  99.70% CPU
  2,664,840,158 processor cycles
  0 bytes consed
  
Evaluation took:
  0.555 seconds of real time  ; merge-sort:sort# 0.555秒
  0.552034 seconds of total run time (0.552034 user, 0.000000 system)
  99.46% CPU
  1,107,829,062 processor cycles
  0 bytes consed
  
Evaluation took:
  0.382 seconds of real time  ; merge-sort:sort(inline)# 0.382秒
  0.376024 seconds of total run time (0.376024 user, 0.000000 system)
  98.43% CPU
  761,537,180 processor cycles
  0 bytes consed
  
--> (T T)


;;; 少量データのソート速度
;;; 平均50要素のリスト x 1万 が対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 10000 
                    COLLECT (loop REPEAT (random 100) 
                                  COLLECT (random 10000))))
        (d1 (copy-tree data))
        (d2 (copy-tree data))
        (d3 (copy-tree data))
        (r1 (time (loop FOR d IN d1 COLLECT (stable-sort d fixnum<))))
        (r2 (time (loop FOR d IN d2 COLLECT (merge-sort:sort d fixnum<))))
        (r3 (time (loop FOR d IN d3 COLLECT (merge-sort:sort d fixnum< :inline t)))))
   (list (equal r1 r2)
         (equal r1 r3))))

Evaluation took:
  0.072 seconds of real time ; stable-sort# 0.072秒
  0.072004 seconds of total run time (0.072004 user, 0.000000 system)
  100.00% CPU
  144,958,896 processor cycles
  327,680 bytes consed
  
Evaluation took:
  0.058 seconds of real time  ; merge-sort:sort# 0.058秒
  0.056003 seconds of total run time (0.056003 user, 0.000000 system)
  96.55% CPU
  116,927,902 processor cycles
  163,840 bytes consed
  
Evaluation took:
  0.036 seconds of real time   ; merge-sort:sort(inline)# 0.036秒
  0.032002 seconds of total run time (0.032002 user, 0.000000 system)
  88.89% CPU
  72,255,454 processor cycles
  163,840 bytes consed
  
--> (T T)

今回のように比較関数自体の実行コストが低い場合だと、関数呼び出し(funcall)部分を含めてinline化するだけで、処理時間が2/3程度に削減できていることが分かる。

*1:マージソート(1) - sileの日記

マージソート(2): 要素数が少ない部分リストの特別扱い

昨日に作成したマージソートに手を加えたもの。
要素数が少ない部分リスト*1には、(再帰的な)マージソートではなく、ソーティングネットワーク的なソートを適用することで高速化を図った。
けど、結果的にはほとんど効果がなかった。

計時

まず計測結果から載せる。

;;; 大量データのソート速度
;;; 100万要素のリストが対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 1000000 COLLECT (random 10000000)))
        (d1 (copy-seq data))
        (d2 (copy-seq data))
        (r1 (time (stable-sort d1 #'<)))
        (r2 (time (merge-sort:sort d2 #'<))))  ; 更新版: 実装は末尾に掲載
   (equal r1 r2)))

Evaluation took:
  2.542 seconds of real time  ; stable-sort# 2.542秒 (前回 2.484秒)
  2.536158 seconds of total run time (2.536158 user, 0.000000 system)
  99.76% CPU
  5,071,126,128 processor cycles
  0 bytes consed
  
Evaluation took:
  1.691 seconds of real time   ; merge-sort:sort# 1.691秒 (前回 1.662秒)
  1.688106 seconds of total run time (1.688106 user, 0.000000 system)
  99.82% CPU
  3,373,722,509 processor cycles
  0 bytes consed
  
--> T


;;; 少量データのソート速度
;;; 平均50要素のリスト x 1万 が対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 10000 
                    COLLECT (loop REPEAT (random 100) 
                                  COLLECT (random 10000))))
        (d1 (copy-tree data))
        (d2 (copy-tree data))
        (r1 (time (loop FOR d IN d1 COLLECT (stable-sort d #'<))))
        (r2 (time (loop FOR d IN d2 COLLECT (merge-sort:sort d #'<)))))
   (equal r1 r2)))

Evaluation took:
  0.207 seconds of real time  ; stable-sort# 0.207秒 (前回 0.204秒)
  0.204013 seconds of total run time (0.204013 user, 0.000000 system)
  98.55% CPU
  414,010,874 processor cycles
  327,680 bytes consed
  
Evaluation took:
  0.174 seconds of real time   ; merge-sort:sort# 0.174秒 (前回 0.176秒)
  0.172011 seconds of total run time (0.172011 user, 0.000000 system)
  98.85% CPU
  346,667,396 processor cycles
  163,840 bytes consed
  
--> T

見ての通り、全くと云って良いほど(前回と)結果に差異がない。
少しくらいは速くなるかと期待していたのだけれど・・・。

ソースコード

今回の実装のソースコード。
特に何かが改善されたということでもないので、コメントはいつも以上に手抜き。

(defpackage merge-sort
  (:use common-lisp)
  (:shadow :common-lisp sort)
  (:export sort))
(in-package :merge-sort)

(declaim (inline halve last! merge-lists less-equal-than
                 sort2 sort3 sort4 sort5)
         (optimize (speed 3) (debug 0) (safety 0)))

(defun halve (n)
  (declare (fixnum n))
  (multiple-value-bind (n1 x) (floor n 2)
    (values (+ n1 x) n1)))

(defmacro cdr! (list new-cdr)
  `(setf (cdr ,list) ,new-cdr))

(defun last! (list)
  (prog1 (cdr list) (cdr! list nil)))

(defmacro multiple-value-let* (bind-specs &body body)
  (if (null bind-specs)
      `(locally ,@body)
    (destructuring-bind ((vars exp) . rest) bind-specs
      `(multiple-value-bind ,vars ,exp
         (multiple-value-let* ,rest ,@body)))))
   
(defun less-equal-than (list1 list2 test key)
  (declare (function test key))
  (not (funcall test (funcall key (car list2)) (funcall key (car list1)))))

(defun merge-lists (list1 list2 test key)
  (declare (function test key))
  (labels ((recur (head tail l1 l2)
             (cond ((null l1) (cdr! tail l2) head)
                   ((null l2) (cdr! tail l1) head)
                   ((less-equal-than l1 l2 test key) 
                    (recur head (cdr! tail l1) (cdr l1) l2))
                   (t                 
                    (recur head (cdr! tail l2) l1 (cdr l2))))))
    (if (less-equal-than list1 list2 test key)
        (recur list1 list1 (cdr list1) list2)
      (recur list2 list2 list1 (cdr list2)))))

(eval-when (:compile-toplevel :load-toplevel :execute)
  (defun symb (&rest args)
    (intern (format nil "~{~a~}" args))))

(defun sort2 (list test key &aux (l1 list) (l2 (cdr list)))
  (unless (less-equal-than l1 l2 test key)
    (rotatef (car l1) (car l2)))
  (values l1 (last! l2)))


          ;; (vars (a b c) (list key)
          ;;   body)
          ;;
          ;; =>
          ;; (let* ((a list)
          ;;        (b (cdr a))
          ;;        (c (cdr b)))
          ;;   (let ((_a (funcall key (car a)))
          ;;         (_b (funcall key (car b)))
          ;;         (_c (funcall key (car c))))
          ;;     body))
(macrolet ((vars (vars (list key) &body body)
             `(let* ,(loop FOR prev = nil THEN var
                           FOR var IN vars
                           FOR i fixnum FROM 0
                           COLLECT (if prev 
                                       `(,var (cdr ,prev))
                                     `(,var ,list)))
                (declare (function ,key))
                (let ,(loop FOR var IN vars
                            COLLECT `(,(symb '_ var) (funcall ,key (car ,var))))
                  ,@body)))
           (swap-if-greater-than (x y test)
             `(unless (less-equal-than ,x ,y ,test #'identity)
                (rotatef (car ,x) (car ,y))
                (rotatef ,(symb '_ x) ,(symb '_ y)))))

  (defun sort3 (list test key)
    (vars (a b c) (list key)
      (swap-if-greater-than a c test)
      (swap-if-greater-than a b test)
      (swap-if-greater-than b c test)
      (values a (last! c))))
  
  (defun sort4 (list test key)
    (vars (a b c d) (list key)
      (swap-if-greater-than a c test)
      (swap-if-greater-than b d test)
      (swap-if-greater-than a b test)
      (swap-if-greater-than c d test)
      (swap-if-greater-than b c test)
      (values a (last! d))))

  (defun sort5 (list test key)
    (vars (a b c d e) (list key)
      (swap-if-greater-than a b test)
      (swap-if-greater-than d e test)
      (swap-if-greater-than a c test)
      (swap-if-greater-than b c test)
      (swap-if-greater-than a d test)
      (swap-if-greater-than c d test)
      (swap-if-greater-than b e test)
      (swap-if-greater-than b c test)
      (swap-if-greater-than d e test)
      (values a (last! e)))))

(defun sort-impl (list size test key)
  (declare (fixnum size))
  (case size
    (5 (sort5 list test key))
    (4 (sort4 list test key))
    (3 (sort3 list test key))
    (otherwise
     (multiple-value-let* (((size1 size2) (halve size))
                           ((list1 rest) (sort-impl list size1 test key))
                           ((list2 rest) (sort-impl rest size2 test key)))
       (values (merge-lists list1 list2 test key) rest)))))

(defun sort (list test &key (key #'identity) &aux (size (length list)))
  (declare (list list)
           (function test key)
           (optimize (speed 3) (safety 2) (debug 2)))
  (case size
    ((0 1) list)
    (2 
     (values (sort2 list test key)))
    (otherwise
     (values (sort-impl list size test key)))))

*1:具体的には要素数が5以下の部分リスト

マージソート(1)

久々にマージソートを実装してみたら、結構良いものができたので載せておく。
まずはパッケージ定義とグルーバルな宣言。

;;;; SBCL-1.0.51 (x86-64)
(defpackage merge-sort
  (:use common-lisp)
  (:shadow :common-lisp sort)
  (:export sort))
(in-package :merge-sort)

(declaim (inline halve merge-lists)
         (optimize (speed 3) (debug 0) (safety 0)))

ユーティリティ関数とマクロ。

;; 整数nを二分割する関数
;; => (values n1 n2)
;;
;; nが奇数の場合は (= (1+ n1) n2) となる
(defun halve (n)
  (declare (fixnum n))
  (multiple-value-bind (n1 x) (floor n 2)
    (values (+ n1 x) n1)))

;; listのcdr部を更新するマクロ
(defmacro cdr! (list new-cdr)
  `(setf (cdr ,list) ,new-cdr))

;; 複数のmultiple-value-bindの使用を簡略化するためのマクロ
;;
;; (multiple-value-let* (((a b) (values 1 2))
;;                       ((c d) (values 3 4)))
;;   (list a b c d))
;; =>
;; (multiple-value-bind (a b) (values 1 2)
;;    (multiple-value-bind (c d) (values 3 4)
;;        (list a b c d)))
(defmacro multiple-value-let* (bind-specs &body body)
  (if (null bind-specs)
      `(locally ,@body)
    (destructuring-bind ((vars exp) . rest) bind-specs
      `(multiple-value-bind ,vars ,exp
         (multiple-value-let* ,rest ,@body)))))

マージソート実装。

;; list1とlist2をマージしたリストを返す
(defun merge-lists (list1 list2 test key)
  (declare (function test key))
  (labels ((not-less-than (l1 l2)
             (not (funcall test (funcall key (car l1)) (funcall key (car l2))))) ; XXX: これでは安定ソートにならない！
           (recur (head tail l1 l2)
             (cond ((null l1)             (cdr! tail l2) head)
                   ((null l2)             (cdr! tail l1) head)
                   ((not-less-than l1 l2) (recur head (cdr! tail l2) l1 (cdr l2)))
                   (t                     (recur head (cdr! tail l1) (cdr l1) l2)))))
    (declare (inline not-less-than))
    (if (not-less-than list1 list2)
        (recur list2 list2 list1 (cdr list2))
      (recur list1 list1 (cdr list1) list2))))

;; マージソート
;; => (values ソート済みリスト=(subseq list 0 size) 
;;            未ソートリスト=(subseq list size))
(defun sort-impl (list size test key)
  (declare (fixnum size))
  (if (= 1 size)
      (values list (prog1 (cdr list) (cdr! list nil)))
    (multiple-value-let* (((size1 size2) (halve size))
                          ((list1 rest) (sort-impl list size1 test key))
                          ((list2 rest) (sort-impl rest size2 test key)))
      (values (merge-lists list1 list2 test key) rest))))

;; 公開用の関数: listが空でなければ、sort-implに処理を任せる
(defun sort (list test &key (key #'identity))
  (declare (list list)
           (function test key)
           (optimize (speed 3) (safety 2) (debug 2)))
  (when list
    (values (sort-impl list (length list) test key))))

以上。

計時

SBCLの組み込みのマージソート(stable-sort関数)との比較。

;;; 大量データのソート速度
;;; 100万要素のリストが対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 1000000 COLLECT (random 10000000)))
        (d1 (copy-seq data))
        (d2 (copy-seq data))
        (r1 (time (stable-sort d1 #'<)))
        (r2 (time (merge-sort:sort d2 #'<))))
   (equal r1 r2))) ; ソート結果が正しいかの確認

Evaluation took:
  2.484 seconds of real time   ; stable-sort# 2.484秒
  2.476154 seconds of total run time (2.368148 user, 0.108006 system)
  99.68% CPU
  4,955,234,522 processor cycles
  0 bytes consed
  
Evaluation took:
  1.662 seconds of real time   ; merge-sort:sort# 1.662秒
  1.652103 seconds of total run time (1.592099 user, 0.060004 system)
  99.40% CPU
  3,315,393,750 processor cycles
  0 bytes consed
 
--> T  ; ソート結果は正しい


;;; 少量データのソート速度
;;; 平均50要素のリスト x 1万 が対象
(sb-sys:without-gcing
 (let* ((data (loop REPEAT 10000 
                    COLLECT (loop REPEAT (random 100) 
                                  COLLECT (random 10000))))
        (d1 (copy-tree data))
        (d2 (copy-tree data))
        (r1 (time (loop FOR d IN d1 COLLECT (stable-sort d #'<))))
        (r2 (time (loop FOR d IN d2 COLLECT (merge-sort:sort d #'<)))))
   (equal r1 r2)))

Evaluation took:
  0.204 seconds of real time  ; stable-sort# 0.204秒
  0.204012 seconds of total run time (0.204012 user, 0.000000 system)
  100.00% CPU
  407,272,146 processor cycles
  294,912 bytes consed
  
Evaluation took:
  0.176 seconds of real time  ; merge-sort:sort# 0.176秒
  0.172010 seconds of total run time (0.172010 user, 0.000000 system)
  97.73% CPU
  351,409,803 processor cycles
  163,840 bytes consed
  
--> T

自分の環境で試した限りでは、なかなか良好な結果となった。