EntryProcessorを使ってみた

目的

CoherenceのEntryProcessorによる並行処理の検証を行う。

注：個人的な検証の結果です。

検証環境

Intel Core i7 S860(2.53GHz)
DDR3 16GB
Windows7 Ultimate
Java 6.0 update 18
Oracle Coherence 3.6.1

検証

　CoherenceのEntryProcessorを使い、QuadCore+HyperThreadingのCPUで並行処理のスケールアウトを確認してみます。

横軸の数値は起動したCoherenceの数、縦軸は処理秒になります。
グラフ上の数字は処理秒を表します。

　グラフを見るとCoherenceの起動数が4つまでは処理時間が1/2、1/3、1/4と減っていきました。しかし、5つ以降は4つの場合とほぼ変わりませんでした。検証を行ったCPUの実コアは4つでありその数まではスケールアウトしますが、HyperThreadingによる効果は低いようです。

　また、検証を行っての感想として実装が非常に簡単でした。これまでに研究などでマルチスレッドやMPIを用いたプログラミングを行ってきましたが、一部制約はありますがそれを考慮するのみでプログラム上に他のノードを気にするようなコードを記述する必要もなくCoherenceを追加起動するだけで処理可能量を増やせるのには驚きでした。

検証方法

　今回はこれといった計算は行っておりません。処理時間を稼ぐためのforループを大量に回しているのみです。本ソースからjarファイルを作成しcache-server.batでクラスパスを指定することにより起動時に取り込まれます。

cache-server.bat(:launchのみ抜粋)

:launch

set java_opts="-Xms%memory% -Xmx%memory% -Dtangosol.coherence.localhost=127.0.0.1 -Dtangosol.coherence.ttl=0"

"%java_exec%" -server -showversion "%java_opts%" -cp "%coherence_home%\lib\coherence.jar;%coherence_home%\lib\faceted.jar" com.tangosol.net.DefaultCacheServer %1

goto exit

オプションとクラスパスを適宜変更して下さい。

EntryProcessorFunc.java

package com.chirokings;

import java.io.IOException;

import com.tangosol.io.pof.PofReader;
import com.tangosol.io.pof.PofWriter;
import com.tangosol.io.pof.PortableObject;
import com.tangosol.util.InvocableMap.Entry;
import com.tangosol.util.processor.AbstractProcessor;

public class EntryProcessorFunc extends AbstractProcessor implements
  PortableObject {

 @Override
 public Object process(Entry entry) {
  double ret = 0;
  for(long i = 0 ; i< 500000000L; i++){
   ret=5;
  }
  return ret;
 }

 @Override
 public void readExternal(PofReader arg0) throws IOException {
 }

 @Override
 public void writeExternal(PofWriter arg0) throws IOException {
 }
}

EntryProcessorRun.java

package com.chirokings;

import java.util.Map;

import com.tangosol.net.CacheFactory;
import com.tangosol.net.NamedCache;
import com.tangosol.util.filter.AlwaysFilter;

public class EntryProcessorRun {

 public static void main(String[] args) {
  NamedCache cache = CacheFactory.getCache("VirtualCache");
  for(int i = 0 ; i<128 ; i++){
   cache.put(i, i);
  }
  System.out.println(System.currentTimeMillis());
  Map ret = cache.invokeAll(new AlwaysFilter(), new EntryProcessorFunc());
  System.out.println(System.currentTimeMillis());  
 }
}

参考

Oracle Coherence入門