rik/jlibredis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Removed System.out and System.err logging. Using native java logging support

Browse Source
This commit is contained in:
Jonathan Leibiusky
2010-10-01 15:13:53 -03:00
parent 9db72b749c
commit 45bb728425
1 changed files with 423 additions and 373 deletions
Download Patch File Download Diff File Expand all files Collapse all files

278
src/main/java/redis/clients/util/FixedResourcePool.java
View File

@@ -6,20 +6,25 @@ import java.util.TimerTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.logging.Level;
import java.util.logging.Logger;
/**
* 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.
*/
public abstract class FixedResourcePool<T> {
private static Logger logger = Logger.getLogger(FixedResourcePool.class
.getName());
/*
* Generic Inner Control Classes ------- ----- ------- ------- * Wrapper *
@@ -47,16 +52,6 @@ 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
*/
@@ -79,6 +74,9 @@ public abstract class FixedResourcePool<T> {
continue;
}
} catch (InterruptedException e) {
if (logger.isLoggable(Level.SEVERE)) {
logger.severe(e.getMessage());
}
continue;
}
@@ -91,14 +89,14 @@ public abstract class FixedResourcePool<T> {
if (resource != null) {
valid = isResourceValid(resource); // Validate the resource.
if (!valid)
fails++;
metrics.fails++;
}
// If resource is invalid or null, create a new resource and
// destroy the invalid one.
if (!valid) {
T replace = createResource();
resourcesCreated++;
metrics.resourcesCreated++;
wrapper.wrapped = replace;
if (resource != null)
destroyResource(resource);
@@ -109,14 +107,18 @@ public abstract class FixedResourcePool<T> {
// Offer the resource to the available resources pool.
if (!availableQueue.offer(wrapper)) {
System.err
.println("This shouldn't happen, offering to available was rejected.");
if (logger.isLoggable(Level.SEVERE)) {
logger
.severe("This shouldn't happen, offering to available was rejected.");
}
}
}
println("Ending thread ["
if (logger.isLoggable(Level.INFO)) {
logger.info("Ending thread ["
+ Thread.currentThread().getName() + "]");
}
}
/**
* Check if resources in the idle queue needs to be repaired
@@ -153,13 +155,17 @@ public abstract class FixedResourcePool<T> {
if (repairNeeded) {
if (!repairQueue.offer(wrapper)) {
System.err
.print("FATAL: This shouldn't happen, offering to repairing was rejected.");
if (logger.isLoggable(Level.SEVERE)) {
logger
.severe("FATAL: This shouldn't happen, offering to repairing was rejected.");
}
}
} else {
if (!availableQueue.offer(wrapper)) {
System.err
.print("FATAL: This shouldn't happen, offering to available was rejected.");
if (logger.isLoggable(Level.SEVERE)) {
logger
.severe("FATAL: This shouldn't happen, offering to available was rejected.");
}
}
}
}
@@ -169,14 +175,15 @@ public abstract class FixedResourcePool<T> {
/*
* Pool metrics
*/
Metrics metrics;
public static class Metrics {
private volatile long failsReported = 0;
private volatile long fails = 0;
private volatile long resourcesCreated = 0;
private volatile long resourcesProvided = 0;
private volatile long resourcesReturned = 0;
private Printer printer = new DefaultPrinter();
/*
* Pool metrics accessing methods.
*/
@@ -200,13 +207,14 @@ public abstract class FixedResourcePool<T> {
public long getResourcesReturned() {
return resourcesReturned;
}
}
/*
* Pool status structures
*/
private LinkedBlockingQueue<Wrapper<T>> availableQueue;
private LinkedBlockingQueue<Wrapper<T>> repairQueue;
private final HashMap<T, Wrapper<T>> inUse = new HashMap<T, Wrapper<T>>();
private HashMap<T, Wrapper<T>> inUse = new HashMap<T, Wrapper<T>>();
private RepairThread[] repairThreads;
private Timer t;
private boolean initializated = false;
@@ -220,6 +228,7 @@ public abstract class FixedResourcePool<T> {
private int resourcesNumber = 10;
private int repairThreadsNumber = 3;
private long timeBetweenValidation = 150000;
private long destructionWait = 10000;
/*
* Bean pool configuration
@@ -252,6 +261,14 @@ public abstract class FixedResourcePool<T> {
this.timeBetweenValidation = timeBetweenValidation;
}
public long getDestructionWait() {
return destructionWait;
}
public void setDestructionWait(long destructionWait) {
this.destructionWait = destructionWait;
}
public void setName(String name) {
if (initializated)
throw new IllegalStateException(
@@ -260,9 +277,8 @@ public abstract class FixedResourcePool<T> {
}
public String getName() {
if (name == null || name.isEmpty()) {
name = this.getClass().getName();
}
if (name == null)
name = getClass().getName();
return name;
}
@@ -274,27 +290,98 @@ public abstract class FixedResourcePool<T> {
return defaultPoolWait;
}
public void setPrinter(Printer printer) {
this.printer = printer;
/**
* Pool initialization & destruction
*/
/**
* Initialize the pool
*/
@SuppressWarnings("unchecked")
public void init() {
if (initializated == true) {
if (logger.isLoggable(Level.WARNING)) {
logger.severe("Warning, double initialization of [" + this
+ "]");
}
return;
}
private void println(String str) {
if(printer != null)
printer.print(str);
initializated = true;
metrics = new Metrics();
// Create queues with maximum possible capacity
availableQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber);
repairQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber);
// Create and start the repair threads.
repairThreads = new FixedResourcePool.RepairThread[repairThreadsNumber];
for (int i = 0; i < repairThreads.length; i++) {
repairThreads[i] = new RepairThread();
repairThreads[i].setName("REPAIR[" + i + "]:" + name);
repairThreads[i].start();
}
// Create resource wrappers with null content.
for (int i = 0; i < resourcesNumber; i++) {
if (!repairQueue.offer(new Wrapper<T>(null)))
throw new IllegalStateException(
"What!? not enough space in the repairQueue to offer the element. This shouldn't happen!");
}
// Schedule a status report every 10 seconds.
t = new Timer();
t.schedule(new TimerTask() {
@Override
public void run() {
if (logger.isLoggable(Level.FINE)) {
StringBuilder sb = new StringBuilder();
sb.append("**********************************");
sb.append("* Pool name:[" + name + "]");
sb.append("* resourcesCreated....:"
+ metrics.getResourcesCreated());
sb.append("* failsReported.......:"
+ metrics.getFailsReported());
sb.append("* fails...............:" + metrics.getFails());
sb.append("* resourcesCreated....:"
+ metrics.getResourcesCreated());
sb.append("* resourcesProvided...:"
+ metrics.getResourcesProvided());
sb.append("* resourcesReturned...:"
+ metrics.getResourcesReturned());
sb
.append("* available size......:"
+ availableQueue.size());
sb.append("* repair size.........:" + repairQueue.size());
sb.append("**********************************");
logger.fine(sb.toString());
}
}
}, 10000, 10000);
if (logger.isLoggable(Level.INFO)) {
logger.info("Initialized [" + name + "]");
}
}
/**
* Pool initialization & destruction
* Destroy the pool. After being destroyed the pool can be initialized
* again.
*/
public void destroy() {
checkInit();
println("Destroying [" + getName() + "]...");
if (logger.isLoggable(Level.INFO)) {
logger.info("Destroying [" + getName() + "]...");
}
// Signal al threads to end
finishing = true;
println("Destroying [" + getName() + "] threads");
if (logger.isLoggable(Level.INFO)) {
logger.info("Destroying [" + getName() + "] threads");
}
// Wait for the Repair Threas
for (int i = 0; i < repairThreads.length; i++) {
boolean joined = false;
@@ -304,27 +391,44 @@ public abstract class FixedResourcePool<T> {
repairThreads[i].join();
joined = true;
} catch (InterruptedException e) {
e.printStackTrace();
if (logger.isLoggable(Level.SEVERE)) {
logger.severe(e.getMessage());
}
}
} while (!joined);
}
println("Waiting for [" + getName()
+ "] resources to be returned.");
if (logger.isLoggable(Level.INFO)) {
logger.info("Waiting at most [" + getDestructionWait()
+ "]ms for [" + getName() + "] resources to be returned.");
}
// Wait for all resources to be returned to the pool
long t1 = System.currentTimeMillis();
synchronized (this) {
while (!inUse.isEmpty()) {
try {
this.wait();
this.wait(getDestructionWait());
long t2 = System.currentTimeMillis();
if ((t2 - t1 > getDestructionWait()) & !inUse.isEmpty()) {
if (logger.isLoggable(Level.INFO)) {
logger
.info("Resource wait timeout. Forcing inUse resources destruction.");
}
for (T used : inUse.keySet()) {
destroyResource(used);
}
}
} catch (InterruptedException e) {
e.printStackTrace();
if (logger.isLoggable(Level.SEVERE)) {
logger.severe(e.getMessage());
}
}
}
}
printStatistics();
println("Destroying [" + getName() + "] resources.");
if (logger.isLoggable(Level.INFO)) {
logger.info("Destroying [" + getName() + "] resources.");
}
// Destroy resources
for (Wrapper<T> resource : availableQueue) {
destroyResource(resource.wrapped);
@@ -341,81 +445,22 @@ public abstract class FixedResourcePool<T> {
repairQueue = null;
// Destroy metrics timer
println("Shuting metrics timer for [" + getName()
+ "] down.");
if (logger.isLoggable(Level.INFO)) {
logger.info("Shuting metrics timer for [" + getName() + "] down.");
}
t.cancel();
t = null;
// Reset metrics
failsReported = 0;
fails = 0;
resourcesCreated = 0;
resourcesProvided = 0;
resourcesReturned = 0;
metrics = null;
// Set states to initial values
initializated = false;
finishing = false;
println("Pool [" + getName() + "] successfully destroyed.");
if (logger.isLoggable(Level.INFO)) {
logger.info("Pool [" + getName() + "] successfully destroyed.");
}
/**
* Initialize the pool
*/
@SuppressWarnings("unchecked")
public void init() {
if (initializated == true) {
println("Warning, double initialization of [" + this
+ "]");
return;
}
initializated = true;
// Create queues with maximum possible capacity
availableQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber);
repairQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber);
// Create and start the repair threads.
repairThreads = new FixedResourcePool.RepairThread[repairThreadsNumber];
for (int i = 0; i < repairThreads.length; i++) {
repairThreads[i] = new RepairThread();
repairThreads[i].setName("REPAIR[" + i + "]:" + getName());
repairThreads[i].start();
}
// Create resource wrappers with null content.
for (int i = 0; i < resourcesNumber; i++) {
if (!repairQueue.offer(new Wrapper<T>(null)))
throw new IllegalStateException(
"What!? not enough space in the repairQueue to offer the element. This shouldn't happen!");
}
// Schedule a status report every 10 seconds.
t = new Timer();
t.schedule(new TimerTask() {
@Override
public void run() {
printStatistics();
}
}, 10000, 10000);
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() {
@@ -471,7 +516,7 @@ public abstract class FixedResourcePool<T> {
throw new IllegalStateException(
"This shouldn't happen. Offering to available queue rejected.");
}
resourcesReturned++;
metrics.resourcesReturned++;
if (finishing) {
synchronized (this) {
@@ -503,7 +548,7 @@ public abstract class FixedResourcePool<T> {
if (!repairQueue.offer(wrapper))
throw new IllegalStateException(
"This shouldn't happen. Offering to repair queue rejected.");
resourcesReturned++;
metrics.resourcesReturned++;
if (finishing) {
synchronized (this) {
@@ -526,7 +571,8 @@ public abstract class FixedResourcePool<T> {
/**
* Get a resource from the pool.
*
* @param maxTime Max time you would like to wait for the resource
* @param maxTime
* Max time you would like to wait for the resource
* @return
* @throws TimeoutException
*/
@@ -550,11 +596,13 @@ public abstract class FixedResourcePool<T> {
synchronized (inUse) {
inUse.put(ret.wrapped, ret);
}
resourcesProvided++;
metrics.resourcesProvided++;
return ret.wrapped;
}
} catch (InterruptedException e1) {
e1.printStackTrace();
if (logger.isLoggable(Level.SEVERE)) {
logger.severe(e1.getMessage());
}
} // If the wait gets interrupted, doesn't matter but print it (just
// in case).
} while (true);
@@ -565,7 +613,9 @@ public abstract class FixedResourcePool<T> {
*/
/**
* Create a resource for the pool
* Create a resource for the pool. This method is always called from the
* REPAIR threads. If it's impossible to create the resource, this method
* should wait or hold until resource creation becomes possible.
*/
protected abstract T createResource();