Merge branch 'master' of http://github.com/alextkachman/jedis

src/main/java/redis/clients/util/FixedResourcePool.java

@@ -9,14 +9,13 @@ import java.util.concurrent.TimeoutException;
 
 /**
  * Abstract resource pool of type T.
- * 
+ * <p/>
  * Needs implementation for creation, validation and destruction of the
  * resources.
- * 
+ * <p/>
  * Keeps a fixed amount of resources
  *
  * @author Luis Dario Simonassi
- * 
  * @param <T>
  * The type of the resource to be managed.
  */
@@ -48,6 +47,16 @@ public abstract class FixedResourcePool<T> {
         }
     }
 
+    public abstract static class Printer {
+        public abstract void print(String str);
+    }
+
+    public static class DefaultPrinter extends Printer {
+        public void print(String str) {
+            System.out.println(str);
+        }
+    }
+
     /**
      * Generic Repair Thread
      */
@@ -105,7 +114,7 @@ public abstract class FixedResourcePool<T> {
                 }
             }
 
-	    System.out.println("Ending thread ["
+            println("Ending thread ["
                     + Thread.currentThread().getName() + "]");
         }
 
@@ -166,6 +175,8 @@ public abstract class FixedResourcePool<T> {
     private volatile long resourcesProvided = 0;
     private volatile long resourcesReturned = 0;
 
+    private Printer printer = new DefaultPrinter();
+
     /*
      * Pool metrics accessing methods.
      */
@@ -263,18 +274,27 @@ public abstract class FixedResourcePool<T> {
         return defaultPoolWait;
     }
 
+    public void setPrinter(Printer printer) {
+        this.printer = printer;
+    }
+
+    private void println(String str) {
+        if(printer != null)
+            printer.print(str);
+    }
+
     /**
      * Pool initialization & destruction
      */
     public void destroy() {
         checkInit();
 
-	System.out.println("Destroying [" + getName() + "]...");
+        println("Destroying [" + getName() + "]...");
 
         // Signal al threads to end
         finishing = true;
 
-	System.out.println("Destroying [" + getName() + "] threads");
+        println("Destroying [" + getName() + "] threads");
         // Wait for the Repair Threas
         for (int i = 0; i < repairThreads.length; i++) {
             boolean joined = false;
@@ -289,7 +309,7 @@ public abstract class FixedResourcePool<T> {
             } while (!joined);
         }
 
-	System.out.println("Waiting for [" + getName()
+        println("Waiting for [" + getName()
                 + "] resources to be returned.");
         // Wait for all resources to be returned to the pool
         synchronized (this) {
@@ -302,7 +322,9 @@ public abstract class FixedResourcePool<T> {
             }
         }
 
-	System.out.println("Destroying [" + getName() + "] resources.");
+        printStatistics();
+
+        println("Destroying [" + getName() + "] resources.");
         // Destroy resources
         for (Wrapper<T> resource : availableQueue) {
             destroyResource(resource.wrapped);
@@ -319,7 +341,7 @@ public abstract class FixedResourcePool<T> {
         repairQueue = null;
 
         // Destroy metrics timer
-	System.out.println("Shuting metrics timer for [" + getName()
+        println("Shuting metrics timer for [" + getName()
                 + "] down.");
         t.cancel();
         t = null;
@@ -335,7 +357,7 @@ public abstract class FixedResourcePool<T> {
         initializated = false;
         finishing = false;
 
-	System.out.println("Pool [" + getName() + "] successfully destroyed.");
+        println("Pool [" + getName() + "] successfully destroyed.");
     }
 
     /**
@@ -344,7 +366,7 @@ public abstract class FixedResourcePool<T> {
     @SuppressWarnings("unchecked")
     public void init() {
         if (initializated == true) {
-	    System.err.println("Warning, double initialization of [" + this
+            println("Warning, double initialization of [" + this
                     + "]");
             return;
         }
@@ -375,28 +397,25 @@ public abstract class FixedResourcePool<T> {
         t.schedule(new TimerTask() {
             @Override
             public void run() {
-		System.out.println("**********************************");
+                printStatistics();
-		System.out.println("* Pool name:[" + name + "]");
-		System.out.println("* resourcesCreated....:"
-			+ getResourcesCreated());
-		System.out.println("* failsReported.......:"
-			+ getFailsReported());
-		System.out.println("* fails...............:" + getFails());
-		System.out.println("* resourcesCreated....:"
-			+ getResourcesCreated());
-		System.out.println("* resourcesProvided...:"
-			+ getResourcesProvided());
-		System.out.println("* resourcesReturned...:"
-			+ getResourcesReturned());
-		System.out.println("* available size......:"
-			+ availableQueue.size());
-		System.out.println("* repair size.........:"
-			+ repairQueue.size());
-		System.out.println("**********************************");
             }
         }, 10000, 10000);
 
-	System.out.println("Initialized [" + name + "]");
+        println("Initialized [" + name + "]");
+    }
+
+    private void printStatistics() {
+        println("**********************************" +
+        "\n* Pool name:[" + name + "]" +
+        "\n* resourcesCreated....:"  + getResourcesCreated() +
+        "\n* failsReported.......:" + getFailsReported() +
+        "\n* fails...............:" + getFails() +
+        "\n* resourcesCreated....:" + getResourcesCreated() +
+        "\n* resourcesProvided...:" + getResourcesProvided() +
+        "\n* resourcesReturned...:" + getResourcesReturned() +
+        "\n* available size......:" + availableQueue.size() +
+        "\n* repair size.........:" + repairQueue.size() +
+        "\n**********************************");
     }
 
     protected void checkInit() {
@@ -507,8 +526,7 @@ public abstract class FixedResourcePool<T> {
     /**
      * Get a resource from the pool.
      *
-     * @param maxTime
+     * @param maxTime Max time you would like to wait for the resource
-     *            Max time you would like to wait for the resource
      * @return
      * @throws TimeoutException
      */

src/test/java/redis/clients/jedis/tests/benchmark/PoolBenchmark.java

@@ -5,6 +5,7 @@ import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisPool;
@@ -24,7 +25,7 @@ public class PoolBenchmark {
 	// withoutPool();
 	withPool();
 	long elapsed = System.currentTimeMillis() - t;
-	System.out.println(((1000 * 3 * TOTAL_OPERATIONS) / elapsed) + " ops");
+	System.out.println(((1000 * 2 * TOTAL_OPERATIONS) / elapsed) + " ops");
     }
 
     private static void withoutPool() throws InterruptedException {
@@ -60,19 +61,19 @@ public class PoolBenchmark {
 
     private static void withPool() throws InterruptedException {
 	final JedisPool pool = new JedisPool("localhost");
-	pool.setResourcesNumber(1000);
+	pool.setResourcesNumber(50);
-	pool.setDefaultPoolWait(20);
+	pool.setDefaultPoolWait(1000000);
 	pool.init();
 	List<Thread> tds = new ArrayList<Thread>();
 
-	for (int i = 0; i < TOTAL_OPERATIONS; i++) {
+    final AtomicInteger ind = new AtomicInteger();
-	    final String key = "foo" + i;
+	for (int i = 0; i < 50; i++) {
 	    Thread hj = new Thread(new Runnable() {
-		@Override
 		public void run() {
+            for(int i = 0; (i = ind.getAndIncrement()) < TOTAL_OPERATIONS; ) {
                 try {
                     Jedis j = pool.getResource();
-			j.auth("foobared");
+                    final String key = "foo" + i;
                     j.set(key, key);
                     j.get(key);
                     pool.returnResource(j);
@@ -80,10 +81,15 @@ public class PoolBenchmark {
                     e.printStackTrace();
                 }
             }
+		}
 	    });
 	    tds.add(hj);
 	    hj.start();
 	}
+
+    for(Thread t : tds)
+        t.join();
+        
 	pool.destroy();
     }
 }