日記帳だ! with Tux on Libserver

ところで昨日の記事で最後に書いた、安定なクイックソートですが、例えばこんなの。

sub mysort{
  return @_ if(@_<2);
  if(@_<=16){
    for my $i(1..$#_){
      my $c=splice(@_,$i,1);
      my $j=$i;
      for(;$j>0;$j--){
        last if $_[$j-1]<=$c;
      }
      splice(@_,$j,0,$c);
    }
    return @_;
  }else{
    my $p=$_[rand(@_)];
    my @s=([],[],[]);
    while(@_){
      my $c=shift(@_);
      push(@{$s[($c <=> $p)+1]},$c);
    }
    return (mysort(@{$s[0]}),@{$s[1]},mysort(@{$s[2]}));
  }
}
@a=mysort(@a);

適当な値を選んで、前から順番に大きいか同じか小さいかに仕分けをすると。
($c <=> $p)+1というのは$cより$pが小さければ0、同じならば1、大きければ1という風になるようにしてあると。
これで仕分けて、小さいグループ・大きいグループをそれぞれクイックソートすると。
このクイックソートは安定ですよ。世の中にはそんなものもあるのです。
案外早いのですが、かかる時間のばらつきがひどいのでどうしたものかなと思っている。

---

ところでマージソートは並列処理に向くらしい。
というわけでいろいろ実験してたのだが、結局実験に使うことにしたのはC#ですね。
というのもPerlもRubyもマルチスレッドは無理ではないが、そんなに性能がよくない。
どちらも画面が固まってしまうのを防ぐような、そういうマルチスレッドだけなんですね。
それならと久々にLL(軽量プログラミング言語のこと)じゃない言語を使おうかなとか。
そうなりゃC#ですね。今までC#でいろんなものを作ってきたけど、.NET Frameworkの性能はかなりよろしい。
マルチスレッドだってバリバリに使えますね。
それで久しぶりに書き始めたらforeachをこんな風に書いてしまったりした。

foreach(my cur in foo){ }

myじゃなくてintですね。Perlではなんでもかんでもmyでしたからね。
しかしuse strict;していて変数宣言をする習慣があったので、変数宣言しないとかはそうもなかった。
というかuse strict;って変数宣言については結構厳しいですからね。その点ではそうも心配いらないと。
ちなみに今回使ったのはVisual C# 2008 Express、アカデミック版のVisual Studio 2005は持ってるのにね。
けどなんでこれなのかといえば、C# 3.0を使いたかったから。

---

というわけで作ってみたソースを張ろうかな。

static class SortWorld {
  static void Main(string[] args) {
    float[] rnd=new float[50000000];
    System.Diagnostics.Stopwatch watch=new System.Diagnostics.Stopwatch();
    SortWorld.MultiThreadMode=1;
    for (int i=0; i    Console.WriteLine("Begin MargeSort"); watch.Reset(); watch.Start();
    rnd.MargeSortDestroy();
    watch.Stop(); Console.WriteLine("Finish MargeSort Time:{0}[ms]", watch.ElapsedMilliseconds);
    for (int i=0; i    Console.WriteLine("Begin Array.Sort"); watch.Reset(); watch.Start();
    Array.Sort(rnd);
    watch.Stop(); Console.WriteLine("Finish Array.Sort Time:{0}[ms]", watch.ElapsedMilliseconds);
    Console.ReadKey(true);
  }
  static private Random _ourrnd;
  static public int MultiThreadMode;
  static SortWorld() {
    _ourrnd=new Random();
    MultiThreadMode=0;
  }
  static public T[] MargeSortDestroy(this T[] self, params Comparison[] fs) where T : IComparable {
    self._MargeSortDestroy(MultiThreadMode, fs.Length==0?null:fs[0]);
    return self;
  }
  static private void _MargeSortDestroy(this T[] self, int mode, Comparison f) where T : IComparable {
    if (self.Length<2) return;
    if (self.Length<=10) {
      self._InsertSortDestroy(self.Length, f); return;
    }
    T[] left=new T[self.Length/2];
    Array.Copy(self, 0, left, 0, left.Length);
    T[] right=new T[self.Length-left.Length];
    Array.Copy(self, left.Length, right, 0, right.Length);
    if (mode>0) {
      System.Threading.ThreadStart ld=() => { left._MargeSortDestroy(mode-1, f); };
      System.Threading.Thread lt=new System.Threading.Thread(ld); lt.Start();
      System.Threading.ThreadStart rd=() => { right._MargeSortDestroy(mode-1, f); };
      System.Threading.Thread rt=new System.Threading.Thread(rd); rt.Start();
      lt.Join(); rt.Join();
    } else {
      left ._MargeSortDestroy(mode, f);
      right._MargeSortDestroy(mode, f);
    }
    int lp=0, rp=0, ap=0;
    while (lp      int c=(f==null?left[lp].CompareTo(right[rp]):f(left[lp], right[rp]));
      if (c>0) self[ap++]=right[rp++];
      else     self[ap++]=left[lp++];
    }
    if (lp    else Array.Copy(right, rp, self, ap, right.Length-rp);
  }
  static private void _InsertSortDestroy(this T[] self, int len, Comparison f) where T : IComparable {
    for (int i=1; i      for (int j=i; j>0; j--) {
        if ((f==null?self[j-1].CompareTo(self[j]):f(self[j-1], self[j]))<=0) break;
        T tmp=self[j];
        self[j]=self[j-1];
        self[j-1]=tmp;
      }
    }
  }
}

なぜか長い。Perlのソースコードよりも明らかにかさばるんですね。なんでだろ。
まぁいいや、ポイントは_MargeSortDestroyメソッドですね。
ところでC# 3.0から始まったことだけど、第一引数のところにthisをつけると、拡張メソッドというメソッドにできるんですね。
どういうことかといいますと、Foo(bar)とするべきところを、拡張メソッドならばbar.Foo()とかけると。
それをちょっと使っています。
10個以下の場合は後ろに書いてある_InsertSortDestroyメソッドで挿入ソートでソートしてもらう。
それ以外の場合は左右に分けて、それぞれを再帰的にマージソート、
それで左右をくっつけていくと。それでできあがりと。
ところでいろんなメソッドにやたらDestroyとついてるけど、破壊的メソッドってことね。
元の配列の要素を半分ずつleftとrightにコピーして、これをそれぞれソート、
そしてそのleftとrightの要素を取り出していって、元の配列に上書きしてしまう。
これこそが破壊的である理由。Rubyなら破壊的メソッドには!つけるんだから、それにならったわけ。
それで、ちょっと工夫してあるのがマルチスレッドの活用ですね。
ただ何でもかんでもスレッドに切り分けると重くなるので、leftとrightで並行処理を行うのは、始め1段のみとしました。
デュアルコアのCPUだとこれが一番効率よさそうです。
そのためにmodeという引数を取ってるけど、これは1段だけ並行処理するときは始めに1を与えて、
並行処理させるたびに1減らしていって、0になったら並行処理しなくするということね。
その並行処理ですが、System.Threading.Threadクラスを使ってる。
デリゲートが作りやすいのはC# 3.0のいいところですね。(x,y)=>{ x.foo(y); }のように作れますからね。
それが_MargeSortDestroyメソッド、実際に使うときはMargeSortDestro
yメソッドを使う。
ここでちょっとおもしろい工夫をしてある。
C# 4.0からは省略可能な引数ができるのですが、C# 3.0まではありません、
けど最後の引数をparamsつけて可変長引数にすれば、最後の引数はなくてもいいんですね。
それで省略できるようにしてある。最後の引数は比較用デリゲートですね。(x,y)=>((int)x).CompareTo((int)y)とか与えれば動きますよ。
まぁこんなところです。
それで実行結果ですが5000万個の配列のソートを38秒で行う。かなり速いですね。
これはマルチスレッド化の成果で、そうしなければ60秒ぐらいかかる。
ちなみに標準のSystem.Array.Sortメソッドは16秒でやる。ただしこちらは安定でないソート。

---

安定なクイックソートをがんばって実装してみたのですが、これが結構困る。
というのも配列を3つに仕分けするためにメモリをすごく食いそうなんですね。
そこでこんな風にして配列を節約することに。
まず元の配列と同じ長さの配列を用意する。そしてこの配列を頭から読み取って、
ある値よりも小さければ元の配列の読み取り済みのスペースに入れる。
大きければ新しく用意した配列に前から入れる。等しければ新しく用意した配列に後ろから入れる。
仕分けが終わったら、等しいグループを小さいグループの後ろに移動する。
そして、元の配列の小さいグループの部分だけと、新しく用意した配列の大きいグループの部分だけをクイックソート。
完了したら新しく用意した配列のその部分を元の配列の後ろに戻す。
というメモリの節約のためにかなり苦しいことをしています。なので読みにくいったらありゃしない。

  static public T[] StableQuickSortDestory(this T[] self, params Comparison[] fs) where T : IComparable {
    self._StableQuickSortDestory(self.Length, MultiThreadMode,fs.Length==0?null:fs[0]);
    return self;
  }
  static private void _StableQuickSortDestory(this T[] self, int len, int mode, Comparison f) where T : IComparable {
    if (len<2) return;
    if (len<=50) {
      self._InsertSortDestroy(len, f);
      return;
    }
    T[] buf=new T[len];
    T p=self[_ourrnd.Next(len)];
    int lp=0, rp=0, cp=buf.Length-1;
    for (int sp=0; sp      int m=(f==null?p.CompareTo(self[sp]):f(p, self[sp]));
      if (m>0)       self[lp++]=self[sp];
      else if (m==0) buf[cp--]=self[sp];
      else           buf[rp++]=self[sp];
    }
    int llen=lp;
    for (int i=buf.Length-1; i>cp; i--) self[lp++]=buf[i];
    if (mode>0) {
      System.Threading.ThreadStart rd=() => {
        buf._StableQuickSortDestory(rp, mode-1, f);
        Array.Copy(buf, 0, self, lp, rp);
      };
      System.Threading.Thread rt=new System.Threading.Thread(rd); rt.Start();
      System.Threading.ThreadStart ld=() => { self._StableQuickSortDestory(llen, mode-1, f); };
      System.Threading.Thread lt=new System.Threading.Thread(ld); lt.Start();
      lt.Join(); rt.Join();
    } else {
      buf._StableQuickSortDestory(rp, mode, f);
      Array.Copy(buf, 0, self, lp, rp);
      self._StableQuickSortDestory(llen, mode, f);
    }
  }

クイックソートは数が少ないと本当に調子が悪いので50個以下で挿入ソートに切り替えています。
それでさっきのMainメソッドのMargeSortDestroyをStableQuickSortDestoryにすればいいのですが、
MultiThreadMode=3にセットしといた方がいいですね。
というのもクイックソートは左右の分割が均等でなくなることがよくあります。
そうなると左はとうに処理終わってるのに、右はまだまだということがあります。
なのでスレッド数を多めに用意しておいて、手が空いたら右のを分担してくれと。
そんな感じです。するとCPUはずっと100%で動いてくれます。
これで40秒ぐらいでソートします。実はマージソートと性能はそうも変わらないんですね。
というのもマージソートは並列処理にすると本当に性能が上がるんですけど、
クイックソートはさっき言ったような事情で並列処理がなかなかうまくいかないと。
それでスレッド数を増やすと遅くなってくる。
それとクイックソート一般の問題ですが、ある値の選び方によって、時間がかかりすぎてしまうという問題がある。
安定ソートなクイックソートですが、残念ながらかかる時間は不安定です。
というわけでさっきのマージソートの方がおすすめですね。

---

いろいろ試していたけど、やっぱりマージソートは安定していいですね。
これはPerlがマージソートに切り替えたのも納得いくことだ。
ところで久しぶりにC#使ってたら、配列自体にはpushもshiftもないと。
だから始めSystem.Collections.Generic.Queueクラスを使ったんですね。
しかしあまりに遅いし、OutOfMemoryException投げるし。これで楽できればなぁという気持ちはあったのですが無理らしい。
というわけで配列をガリガリ操作するプログラムを久々に書きましたね。
やっぱりPerlやらの配列操作関数は偉大ですね。
けどそういうのに対応できない.NET Frameworkの配列は動作が速いですからね。そういう意味ではありがたいかなと。
適材適所ということですが、あまりに巨大なデータを扱うときはひたすら配列を使うのがよさそうです。
そうじゃなければSystem.Collectionの下にある数々のクラスを選んで使っても便利かなと思いますよ。