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

ability to provide logger instead of stdout

Browse Source
This commit is contained in:
Alex Tkachman
2010-09-14 11:50:49 +02:00
parent f9e7887e02
commit 032fe7e134
1 changed files with 342 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -9,16 +9,15 @@ import java.util.concurrent.TimeoutException;
/** /**
* Abstract resource pool of type T. * Abstract resource pool of type T.
* * <p/>
* Needs implementation for creation, validation and destruction of the * Needs implementation for creation, validation and destruction of the
* resources. * resources.
* * <p/>
* Keeps a fixed amount of resources * Keeps a fixed amount of resources
* *
* @author Luis Dario Simonassi * @author Luis Dario Simonassi
*
* @param <T> * @param <T>
* The type of the resource to be managed. * The type of the resource to be managed.
*/ */
public abstract class FixedResourcePool<T> { public abstract class FixedResourcePool<T> {
@@ -31,130 +30,140 @@ public abstract class FixedResourcePool<T> {
* Generic Resource Wrapper * Generic Resource Wrapper
*/ */
private static class Wrapper<T> { private static class Wrapper<T> {
long timestamp; long timestamp;
T wrapped; T wrapped;
public Wrapper(T wrapped) { public Wrapper(T wrapped) {
this.wrapped = wrapped; this.wrapped = wrapped;
mark(); mark();
} }
public void mark() { public void mark() {
timestamp = System.currentTimeMillis(); timestamp = System.currentTimeMillis();
} }
public long getLastMark() { public long getLastMark() {
return timestamp; return timestamp;
} }
}
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 * Generic Repair Thread
*/ */
protected class RepairThread extends Thread { protected class RepairThread extends Thread {
public void run() { public void run() {
// Contribute to the repairing and validation effort until the pool // Contribute to the repairing and validation effort until the pool
// is destroyed (finishig=true) // is destroyed (finishig=true)
while (!finishing) { while (!finishing) {
Wrapper<T> wrapper; Wrapper<T> wrapper;
try { try {
// Remove the oldest element from the repair queue. // Remove the oldest element from the repair queue.
wrapper = repairQueue.poll(timeBetweenValidation, wrapper = repairQueue.poll(timeBetweenValidation,
TimeUnit.MILLISECONDS); TimeUnit.MILLISECONDS);
if (wrapper == null) { if (wrapper == null) {
// If I've been waiting too much, i'll check the idle // If I've been waiting too much, i'll check the idle
// pool if connections need // pool if connections need
// validation and move them to the repair queue // validation and move them to the repair queue
checkIdles(); checkIdles();
continue; continue;
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
continue; continue;
} }
// Now, I have something to repair! // Now, I have something to repair!
T resource = wrapper.wrapped; T resource = wrapper.wrapped;
boolean valid = false; boolean valid = false;
// Resources are null right after initialization, it means the // Resources are null right after initialization, it means the
// same as being an invalid resource // same as being an invalid resource
if (resource != null) { if (resource != null) {
valid = isResourceValid(resource); // Validate the resource. valid = isResourceValid(resource); // Validate the resource.
if (!valid) if (!valid)
fails++; fails++;
} }
// If resource is invalid or null, create a new resource and // If resource is invalid or null, create a new resource and
// destroy the invalid one. // destroy the invalid one.
if (!valid) { if (!valid) {
T replace = createResource(); T replace = createResource();
resourcesCreated++; resourcesCreated++;
wrapper.wrapped = replace; wrapper.wrapped = replace;
if (resource != null) if (resource != null)
destroyResource(resource); destroyResource(resource);
} }
// Mark the resource as fresh! // Mark the resource as fresh!
wrapper.mark(); wrapper.mark();
// Offer the resource to the available resources pool. // Offer the resource to the available resources pool.
if (!availableQueue.offer(wrapper)) { if (!availableQueue.offer(wrapper)) {
System.err System.err
.println("This shouldn't happen, offering to available was rejected."); .println("This shouldn't happen, offering to available was rejected.");
} }
} }
System.out.println("Ending thread [" println("Ending thread ["
+ Thread.currentThread().getName() + "]"); + Thread.currentThread().getName() + "]");
} }
/** /**
* Check if resources in the idle queue needs to be repaired * Check if resources in the idle queue needs to be repaired
*/ */
private void checkIdles() { private void checkIdles() {
// Get a sample without removing it // Get a sample without removing it
Wrapper<T> wrapper = availableQueue.peek(); Wrapper<T> wrapper = availableQueue.peek();
// If no available items, nothing to repair. // If no available items, nothing to repair.
if (wrapper == null) if (wrapper == null)
return; return;
// Check if the sampled resource needs to be repaired // Check if the sampled resource needs to be repaired
boolean repairNeeded = isValidationNeeded(wrapper); boolean repairNeeded = isValidationNeeded(wrapper);
if (!repairNeeded) if (!repairNeeded)
return; return;
// Move available resources from the available queue to the repair // Move available resources from the available queue to the repair
// queue until no repair is needed. // queue until no repair is needed.
while (repairNeeded) { while (repairNeeded) {
// Get the connection from the available queue and check again. // Get the connection from the available queue and check again.
wrapper = availableQueue.poll(); wrapper = availableQueue.poll();
// No resources in the available queue, nothing to do // No resources in the available queue, nothing to do
if (wrapper == null) { if (wrapper == null) {
repairNeeded = false; repairNeeded = false;
return; return;
} }
// Add the resource to the corresponding queue, depending on // Add the resource to the corresponding queue, depending on
// weather the resource needs to be repaired or not. // weather the resource needs to be repaired or not.
repairNeeded = isValidationNeeded(wrapper); repairNeeded = isValidationNeeded(wrapper);
if (repairNeeded) { if (repairNeeded) {
if (!repairQueue.offer(wrapper)) { if (!repairQueue.offer(wrapper)) {
System.err System.err
.print("FATAL: This shouldn't happen, offering to repairing was rejected."); .print("FATAL: This shouldn't happen, offering to repairing was rejected.");
} }
} else { } else {
if (!availableQueue.offer(wrapper)) { if (!availableQueue.offer(wrapper)) {
System.err System.err
.print("FATAL: This shouldn't happen, offering to available was rejected."); .print("FATAL: This shouldn't happen, offering to available was rejected.");
} }
} }
} }
} }
} }
/* /*
@@ -166,28 +175,30 @@ public abstract class FixedResourcePool<T> {
private volatile long resourcesProvided = 0; private volatile long resourcesProvided = 0;
private volatile long resourcesReturned = 0; private volatile long resourcesReturned = 0;
private Printer printer = new DefaultPrinter();
/* /*
* Pool metrics accessing methods. * Pool metrics accessing methods.
*/ */
public long getFailsReported() { public long getFailsReported() {
return failsReported; return failsReported;
} }
public long getFails() { public long getFails() {
return fails; return fails;
} }
public long getResourcesCreated() { public long getResourcesCreated() {
return resourcesCreated; return resourcesCreated;
} }
public long getResourcesProvided() { public long getResourcesProvided() {
return resourcesProvided; return resourcesProvided;
} }
public long getResourcesReturned() { public long getResourcesReturned() {
return resourcesReturned; return resourcesReturned;
} }
/* /*
@@ -215,127 +226,138 @@ public abstract class FixedResourcePool<T> {
*/ */
public int getResourcesNumber() { public int getResourcesNumber() {
return resourcesNumber; return resourcesNumber;
} }
public void setResourcesNumber(int resourcesNumber) { public void setResourcesNumber(int resourcesNumber) {
this.resourcesNumber = resourcesNumber; this.resourcesNumber = resourcesNumber;
} }
public int getRepairThreadsNumber() { public int getRepairThreadsNumber() {
return repairThreadsNumber; return repairThreadsNumber;
} }
public void setRepairThreadsNumber(int repairThreadsNumber) { public void setRepairThreadsNumber(int repairThreadsNumber) {
if (initializated) if (initializated)
throw new IllegalStateException( throw new IllegalStateException(
"Repair threads should be setted up before init()"); "Repair threads should be setted up before init()");
this.repairThreadsNumber = repairThreadsNumber; this.repairThreadsNumber = repairThreadsNumber;
} }
public long getTimeBetweenValidation() { public long getTimeBetweenValidation() {
return timeBetweenValidation; return timeBetweenValidation;
} }
public void setTimeBetweenValidation(long timeBetweenValidation) { public void setTimeBetweenValidation(long timeBetweenValidation) {
this.timeBetweenValidation = timeBetweenValidation; this.timeBetweenValidation = timeBetweenValidation;
} }
public void setName(String name) { public void setName(String name) {
if (initializated) if (initializated)
throw new IllegalStateException( throw new IllegalStateException(
"Name should be setted up before init()"); "Name should be setted up before init()");
this.name = name; this.name = name;
} }
public String getName() { public String getName() {
if (name == null || name.isEmpty()) { if (name == null || name.isEmpty()) {
name = this.getClass().getName(); name = this.getClass().getName();
} }
return name; return name;
} }
public void setDefaultPoolWait(long defaultPoolWait) { public void setDefaultPoolWait(long defaultPoolWait) {
this.defaultPoolWait = defaultPoolWait; this.defaultPoolWait = defaultPoolWait;
} }
public long getDefaultPoolWait() { public long getDefaultPoolWait() {
return defaultPoolWait; return defaultPoolWait;
}
public void setPrinter(Printer printer) {
this.printer = printer;
}
private void println(String str) {
if(printer != null)
printer.print(str);
} }
/** /**
* Pool initialization & destruction * Pool initialization & destruction
*/ */
public void destroy() { public void destroy() {
checkInit(); checkInit();
System.out.println("Destroying [" + getName() + "]..."); println("Destroying [" + getName() + "]...");
// Signal al threads to end // Signal al threads to end
finishing = true; finishing = true;
System.out.println("Destroying [" + getName() + "] threads"); println("Destroying [" + getName() + "] threads");
// Wait for the Repair Threas // Wait for the Repair Threas
for (int i = 0; i < repairThreads.length; i++) { for (int i = 0; i < repairThreads.length; i++) {
boolean joined = false; boolean joined = false;
do { do {
try { try {
repairThreads[i].interrupt(); repairThreads[i].interrupt();
repairThreads[i].join(); repairThreads[i].join();
joined = true; joined = true;
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} }
} while (!joined); } while (!joined);
} }
System.out.println("Waiting for [" + getName() println("Waiting for [" + getName()
+ "] resources to be returned."); + "] resources to be returned.");
// Wait for all resources to be returned to the pool // Wait for all resources to be returned to the pool
synchronized (this) { synchronized (this) {
while (!inUse.isEmpty()) { while (!inUse.isEmpty()) {
try { try {
this.wait(); this.wait();
} catch (InterruptedException e) { } catch (InterruptedException e) {
e.printStackTrace(); e.printStackTrace();
} }
} }
} }
System.out.println("Destroying [" + getName() + "] resources."); printStatistics();
// Destroy resources
for (Wrapper<T> resource : availableQueue) {
destroyResource(resource.wrapped);
}
availableQueue.clear(); println("Destroying [" + getName() + "] resources.");
availableQueue = null; // Destroy resources
for (Wrapper<T> resource : availableQueue) {
destroyResource(resource.wrapped);
}
for (Wrapper<T> resource : repairQueue) { availableQueue.clear();
destroyResource(resource.wrapped); availableQueue = null;
}
repairQueue.clear(); for (Wrapper<T> resource : repairQueue) {
repairQueue = null; destroyResource(resource.wrapped);
}
// Destroy metrics timer repairQueue.clear();
System.out.println("Shuting metrics timer for [" + getName() repairQueue = null;
+ "] down.");
t.cancel();
t = null;
// Reset metrics // Destroy metrics timer
failsReported = 0; println("Shuting metrics timer for [" + getName()
fails = 0; + "] down.");
resourcesCreated = 0; t.cancel();
resourcesProvided = 0; t = null;
resourcesReturned = 0;
// Set states to initial values // Reset metrics
initializated = false; failsReported = 0;
finishing = false; fails = 0;
resourcesCreated = 0;
resourcesProvided = 0;
resourcesReturned = 0;
System.out.println("Pool [" + getName() + "] successfully destroyed."); // Set states to initial values
initializated = false;
finishing = false;
println("Pool [" + getName() + "] successfully destroyed.");
} }
/** /**
@@ -343,203 +365,199 @@ public abstract class FixedResourcePool<T> {
*/ */
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
public void init() { public void init() {
if (initializated == true) { if (initializated == true) {
System.err.println("Warning, double initialization of [" + this println("Warning, double initialization of [" + this
+ "]"); + "]");
return; return;
} }
initializated = true; initializated = true;
// Create queues with maximum possible capacity // Create queues with maximum possible capacity
availableQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber); availableQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber);
repairQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber); repairQueue = new LinkedBlockingQueue<Wrapper<T>>(resourcesNumber);
// Create and start the repair threads. // Create and start the repair threads.
repairThreads = new FixedResourcePool.RepairThread[repairThreadsNumber]; repairThreads = new FixedResourcePool.RepairThread[repairThreadsNumber];
for (int i = 0; i < repairThreads.length; i++) { for (int i = 0; i < repairThreads.length; i++) {
repairThreads[i] = new RepairThread(); repairThreads[i] = new RepairThread();
repairThreads[i].setName("REPAIR[" + i + "]:" + getName()); repairThreads[i].setName("REPAIR[" + i + "]:" + getName());
repairThreads[i].start(); repairThreads[i].start();
} }
// Create resource wrappers with null content. // Create resource wrappers with null content.
for (int i = 0; i < resourcesNumber; i++) { for (int i = 0; i < resourcesNumber; i++) {
if (!repairQueue.offer(new Wrapper<T>(null))) if (!repairQueue.offer(new Wrapper<T>(null)))
throw new IllegalStateException( throw new IllegalStateException(
"What!? not enough space in the repairQueue to offer the element. This shouldn't happen!"); "What!? not enough space in the repairQueue to offer the element. This shouldn't happen!");
} }
// Schedule a status report every 10 seconds. // Schedule a status report every 10 seconds.
t = new Timer(); t = new Timer();
t.schedule(new TimerTask() { t.schedule(new TimerTask() {
@Override @Override
public void run() { public void run() {
System.out.println("**********************************"); printStatistics();
System.out.println("* Pool name:[" + name + "]"); }
System.out.println("* resourcesCreated....:" }, 10000, 10000);
+ 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() { protected void checkInit() {
if (!initializated) if (!initializated)
throw new IllegalStateException("Call the init() method first!"); throw new IllegalStateException("Call the init() method first!");
} }
/** /**
* Returns true if wrapped resource needs validation * Returns true if wrapped resource needs validation
* *
* @param wrapper * @param wrapper
* @return * @return
*/ */
private boolean isValidationNeeded(Wrapper<T> wrapper) { private boolean isValidationNeeded(Wrapper<T> wrapper) {
// Add noise to the check times to avoid simultaneous resource checking. // Add noise to the check times to avoid simultaneous resource checking.
long noisyTimeBetweenCheck = (timeBetweenValidation - (long) ((Math long noisyTimeBetweenCheck = (timeBetweenValidation - (long) ((Math
.random() - 0.5) * (timeBetweenValidation / 10))); .random() - 0.5) * (timeBetweenValidation / 10)));
// Check if the resource need to be checked. // Check if the resource need to be checked.
return wrapper.getLastMark() + noisyTimeBetweenCheck < System return wrapper.getLastMark() + noisyTimeBetweenCheck < System
.currentTimeMillis(); .currentTimeMillis();
} }
/** /**
* Return a resource to the pool. When no longer needed. * Return a resource to the pool. When no longer needed.
* *
* @param resource * @param resource
*/ */
public void returnResource(T resource) { public void returnResource(T resource) {
checkInit(); checkInit();
Wrapper<T> wrapper; Wrapper<T> wrapper;
if (resource == null) if (resource == null)
throw new IllegalArgumentException( throw new IllegalArgumentException(
"The resource shouldn't be null."); "The resource shouldn't be null.");
// Delete the resource from the inUse list. // Delete the resource from the inUse list.
synchronized (inUse) { synchronized (inUse) {
wrapper = inUse.remove(resource); wrapper = inUse.remove(resource);
} }
if (wrapper == null) if (wrapper == null)
throw new IllegalArgumentException("The resource [" + resource throw new IllegalArgumentException("The resource [" + resource
+ "] isn't in the busy resources list."); + "] isn't in the busy resources list.");
if (isValidationNeeded(wrapper)) { if (isValidationNeeded(wrapper)) {
if (!repairQueue.offer(wrapper)) if (!repairQueue.offer(wrapper))
throw new IllegalStateException( throw new IllegalStateException(
"This shouldn't happen. Offering to repair queue rejected."); "This shouldn't happen. Offering to repair queue rejected.");
} else { } else {
if (!availableQueue.offer(wrapper)) if (!availableQueue.offer(wrapper))
throw new IllegalStateException( throw new IllegalStateException(
"This shouldn't happen. Offering to available queue rejected."); "This shouldn't happen. Offering to available queue rejected.");
} }
resourcesReturned++; resourcesReturned++;
if (finishing) { if (finishing) {
synchronized (this) { synchronized (this) {
this.notify(); this.notify();
} }
} }
} }
/** /**
* Return a broken resource to the pool. If the application detects a * Return a broken resource to the pool. If the application detects a
* malfunction of the resource. This resources will go directly to the * malfunction of the resource. This resources will go directly to the
* repair queue. * repair queue.
* *
* @param resource * @param resource
*/ */
public void returnBrokenResource(T resource) { public void returnBrokenResource(T resource) {
checkInit(); checkInit();
Wrapper<T> wrapper; Wrapper<T> wrapper;
// Delete the resource from the inUse list. // Delete the resource from the inUse list.
synchronized (inUse) { synchronized (inUse) {
wrapper = inUse.remove(resource); wrapper = inUse.remove(resource);
} }
if (wrapper == null) if (wrapper == null)
throw new IllegalArgumentException("The resource [" + resource throw new IllegalArgumentException("The resource [" + resource
+ "] isn't in the busy resources list."); + "] isn't in the busy resources list.");
if (!repairQueue.offer(wrapper)) if (!repairQueue.offer(wrapper))
throw new IllegalStateException( throw new IllegalStateException(
"This shouldn't happen. Offering to repair queue rejected."); "This shouldn't happen. Offering to repair queue rejected.");
resourcesReturned++; resourcesReturned++;
if (finishing) { if (finishing) {
synchronized (this) { synchronized (this) {
this.notify(); this.notify();
} }
} }
} }
/** /**
* Get a resource from the pool waiting the default time. * Get a resource from the pool waiting the default time.
* {@link #setDefaultPoolWait(long)} * {@link #setDefaultPoolWait(long)}
* *
* @return the resource of type T * @return the resource of type T
* @throws TimeoutException * @throws TimeoutException
*/ */
public T getResource() throws TimeoutException { public T getResource() throws TimeoutException {
return getResource(defaultPoolWait); return getResource(defaultPoolWait);
} }
/** /**
* Get a resource from the pool. * 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 * @return
* @throws TimeoutException * @throws TimeoutException
*/ */
public T getResource(long maxTime) throws TimeoutException { public T getResource(long maxTime) throws TimeoutException {
if (finishing) if (finishing)
throw new IllegalStateException("Pool [" + getName() throw new IllegalStateException("Pool [" + getName()
+ "] is currently being destroyed."); + "] is currently being destroyed.");
checkInit(); checkInit();
final long tInit = System.currentTimeMillis(); final long tInit = System.currentTimeMillis();
do { do {
try { try {
long timeSpent = System.currentTimeMillis() - tInit; long timeSpent = System.currentTimeMillis() - tInit;
long timeToSleep = maxTime - timeSpent; long timeToSleep = maxTime - timeSpent;
timeToSleep = timeToSleep > 0 ? timeToSleep : 0; timeToSleep = timeToSleep > 0 ? timeToSleep : 0;
if (timeToSleep == 0) if (timeToSleep == 0)
throw new TimeoutException("" + timeSpent + ">" + maxTime); throw new TimeoutException("" + timeSpent + ">" + maxTime);
Wrapper<T> ret = availableQueue.poll(timeToSleep, Wrapper<T> ret = availableQueue.poll(timeToSleep,
TimeUnit.MILLISECONDS); TimeUnit.MILLISECONDS);
if (ret != null) { if (ret != null) {
synchronized (inUse) { synchronized (inUse) {
inUse.put(ret.wrapped, ret); inUse.put(ret.wrapped, ret);
} }
resourcesProvided++; resourcesProvided++;
return ret.wrapped; return ret.wrapped;
} }
} catch (InterruptedException e1) { } catch (InterruptedException e1) {
e1.printStackTrace(); e1.printStackTrace();
} // If the wait gets interrupted, doesn't matter but print it (just } // If the wait gets interrupted, doesn't matter but print it (just
// in case). // in case).
} while (true); } while (true);
} }
/* /*
@@ -553,7 +571,7 @@ public abstract class FixedResourcePool<T> {
/** /**
* Check if the resource is still valid. * Check if the resource is still valid.
* *
* @param resource * @param resource
* @return * @return
*/ */
@@ -561,14 +579,14 @@ public abstract class FixedResourcePool<T> {
/** /**
* Destroy a resource. * Destroy a resource.
* *
* @param resource * @param resource
*/ */
protected abstract void destroyResource(T resource); protected abstract void destroyResource(T resource);
@Override @Override
public String toString() { public String toString() {
return getName() + "[" + super.toString() + "]"; return getName() + "[" + super.toString() + "]";
} }
/** /**