gistfile1.java

import java.util.Random;

import org.infinispan.Cache;
import org.infinispan.configuration.cache.CacheMode;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.configuration.cache.ConfigurationBuilder;
import org.infinispan.configuration.global.GlobalConfiguration;
import org.infinispan.configuration.global.GlobalConfigurationBuilder;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;

public class InfinispanCacheTest {
    
    private static final String DEFAULT_CACHE_NAME = "default";
    
    private Cache<String,byte[]> cache;
    
    private EmbeddedCacheManager cacheManager;
    
    private int count  = 0;
    
    private int port = 1025;
    
    private String keyPrefix = null;
    
    
    public InfinispanCacheTest(int count,int port, String keyIdx) {
        this.count = count;
        this.port = port;
        this.keyPrefix = keyIdx;
    }

    public Cache<String,byte[]> createCache() {
        
        System.setProperty("jgroups.tcp.address", "10.0.0.1");
        System.setProperty("jgroups.tcpping.initial_hosts", "10.0.0.1[6701],10.0.0.1[6702],10.0.0.1[6703],10.0.0.1[6704]");
        System.setProperty("jgroups.tcpping.num_initial_members", "2");
        System.setProperty("jgroups.tcp.port", String.valueOf(port));
  
        GlobalConfiguration globalConfig = new GlobalConfigurationBuilder().transport()
            .defaultTransport()
            .clusterName("test-cluster")
            .addProperty("configurationFile", "jgroups-tcpdiscovery-tcpreplication.xml")
            .build();
        
        cacheManager = new DefaultCacheManager(globalConfig);
        
        Configuration configuration = new ConfigurationBuilder().clustering().cacheMode(CacheMode.DIST_SYNC)
            .sync().hash().numOwners(2).build();
        
        cacheManager.defineConfiguration(DEFAULT_CACHE_NAME, configuration);
        cache = cacheManager.getCache(DEFAULT_CACHE_NAME);
        
        return cache;


    }
    
    
    public void putEntries(int entrySize) {
        
        System.out.println("Starting inserting...");
        int aTenth = count / 10;
        for (int idx=0;idx<count;idx++) {
            
            String key = keyPrefix+idx;
            if (idx > aTenth && (idx % aTenth == 0)) {
                System.out.println("Inserted "+idx);
            }
            cache.put(key, createByteArray(entrySize, true) );

        }
        
        System.out.println("Done inserting ");
        
        
    }
    
    
    public void getEntries() {
        
        System.out.println(cache.size());
        long totalAccessTime = 0;
        for (int idx=0;idx<count;idx++) {
            long startTime = System.currentTimeMillis();
            String key = keyPrefix +String.valueOf(idx);
            if (null==cache.get(key)) {
                System.out.println("No value found for key "+key+", stopping....");
                break;
            };
            totalAccessTime += (System.currentTimeMillis()-startTime);
            
            if (idx==(count-1)) {
                System.out.println("All entries found");
            }
        }
        
        System.out.println("Mean access time "+(totalAccessTime / count));
    }
    
    public static void main(String[] args) throws InterruptedException {
        
        int numEntries = 100000;
        
        String op = args[0];
        int port = Integer.parseInt(args[1]);
        String keyIdx = args[2];

        
        final InfinispanCacheTest infiniAndBeyondCacheTest = new InfinispanCacheTest(numEntries,port,keyIdx);
        final Cache<String,byte[]> cache = infiniAndBeyondCacheTest.createCache();
        new Thread(new CacheMonitor(cache)).start();
        
        switch (op) {
        case "put":
            
            infiniAndBeyondCacheTest.putEntries(4*1024);
            break;

        case "get" :
            Thread t = new Thread(new Runnable() {
                
                @Override
                public void run() {
                    while(true) {
                        System.out.println("Getting entries");
                        infiniAndBeyondCacheTest.getEntries();
                        try {
                            Thread.currentThread().sleep(15000);
                        }
                        catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        
                    }
                    
                }
            }) ;
            t.start();
        default:
            break;
        }

    }
    
    
    private static class CacheMonitor implements Runnable {
        
        private Cache<String,byte[]> aCache;
        
        public CacheMonitor(Cache<String,byte[]> aCache) {
            this.aCache = aCache;
        }

        @Override
        public void run() {

            while (true) {
                System.out.println("Cache size "+aCache.size());
                try {
                    Thread.currentThread().sleep(15000);
                }
                catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
                
            }
            
        }
        
    }
    
    public static byte[] createByteArray(int size, boolean random) {
        
        Random rd = new Random();
        byte[] abyte = new byte[size];
        for (int idx=0;idx<size;idx++) {
            if (random) {
                abyte[idx]=(byte)rd.nextInt();
            }
            else {
                abyte[idx]=0x41;
            }
        }
        return abyte;
        
    }

}