Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Made the SystemUsage and associated *Usage classes IOC friendly so that they can be configured 

via xbean/spring.  Fixed the ConfigTest test case.



git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@567796 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-08-20 19:30:33 +00:00
parent 6d8e2c5b3a
commit 2b0024c269
15 changed files with 389 additions and 291 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
View File

@ -118,7 +118,7 @@ public class BrokerService implements Service {
private ObjectName brokerObjectName;
private TaskRunnerFactory taskRunnerFactory;
private TaskRunnerFactory persistenceTaskRunnerFactory;
private SystemUsage usageManager;
private SystemUsage systemUsage;
private SystemUsage producerSystemUsage;
private SystemUsage consumerSystemUsage;
private PersistenceAdapter persistenceAdapter;
@ -646,23 +646,23 @@ public class BrokerService implements Service {
this.populateJMSXUserID = populateJMSXUserID;
}
public SystemUsage getUsageManager() {
public SystemUsage getSystemUsage() {
try {
if (usageManager == null) {
usageManager = new SystemUsage("Main",getPersistenceAdapter(),getTempDataStore());
usageManager.getMemoryUsage().setLimit(1024 * 1024 * 64); // Default to 64 Meg
usageManager.getTempDiskUsage().setLimit(1024 * 1024 * 1024 * 100);//10 Gb
usageManager.getStoreUsage().setLimit(1024 * 1024 * 1024 * 100); //100 GB
}
return usageManager;
}catch(IOException e) {
LOG.fatal("Cannot create SystemUsage",e);
if (systemUsage == null) {
systemUsage = new SystemUsage("Main", getPersistenceAdapter(), getTempDataStore());
systemUsage.getMemoryUsage().setLimit(1024 * 1024 * 64); // Default 64 Meg
systemUsage.getTempUsage().setLimit(1024 * 1024 * 1024 * 100); // 10 Gb
systemUsage.getStoreUsage().setLimit(1024 * 1024 * 1024 * 100); // 100 GB
}
return systemUsage;
} catch (IOException e) {
LOG.fatal("Cannot create SystemUsage", e);
throw new RuntimeException("Fatally failed to create SystemUsage" + e.getMessage());
}
}
public void setUsageManager(SystemUsage memoryManager) {
this.usageManager = memoryManager;
public void setSystemUsage(SystemUsage memoryManager) {
this.systemUsage = memoryManager;
}
/**
@ -671,7 +671,7 @@ public class BrokerService implements Service {
*/
public SystemUsage getConsumerSystemUsage() throws IOException {
if (consumerSystemUsage == null) {
consumerSystemUsage = new SystemUsage(getUsageManager(), "Consumer");
consumerSystemUsage = new SystemUsage(getSystemUsage(), "Consumer");
consumerSystemUsage.getMemoryUsage().setUsagePortion(0.5f);
}
return consumerSystemUsage;
@ -690,7 +690,7 @@ public class BrokerService implements Service {
*/
public SystemUsage getProducerSystemUsage() throws IOException {
if (producerSystemUsage == null) {
producerSystemUsage = new SystemUsage(getUsageManager(), "Producer");
producerSystemUsage = new SystemUsage(getSystemUsage(), "Producer");
producerSystemUsage.getMemoryUsage().setUsagePortion(0.45f);
}
return producerSystemUsage;

6
activemq-core/src/main/java/org/apache/activemq/broker/jmx/BrokerView.java
View File

@ -85,15 +85,15 @@ public class BrokerView implements BrokerViewMBean {
}
public int getMemoryPercentageUsed() {
return brokerService.getUsageManager().getMemoryUsage().getPercentUsage();
return brokerService.getSystemUsage().getMemoryUsage().getPercentUsage();
}
public long getMemoryLimit() {
return brokerService.getUsageManager().getMemoryUsage().getLimit();
return brokerService.getSystemUsage().getMemoryUsage().getLimit();
}
public void setMemoryLimit(long limit) {
brokerService.getUsageManager().getMemoryUsage().setLimit(limit);
brokerService.getSystemUsage().getMemoryUsage().setLimit(limit);
}
public void resetStatistics() {

4
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
View File

@ -148,7 +148,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
} else {
flushToDisk();
node.decrementReferenceCount();
systemUsage.getTempDiskUsage().waitForSpace();
systemUsage.getTempUsage().waitForSpace();
getDiskList().addLast(node);
}
} catch (Exception e) {
@ -168,7 +168,7 @@ public class FilePendingMessageCursor extends AbstractPendingMessageCursor imple
memoryList.addFirst(node);
} else {
flushToDisk();
systemUsage.getTempDiskUsage().waitForSpace();
systemUsage.getTempUsage().waitForSpace();
node.decrementReferenceCount();
getDiskList().addFirst(node);
}

105
activemq-core/src/main/java/org/apache/activemq/usage/MemoryUsage.java
View File

@ -16,24 +16,20 @@
*/
package org.apache.activemq.usage;
/**
* Used to keep track of how much of something is being used so that a
* productive working set usage can be controlled.
*
* Main use case is manage memory usage.
* productive working set usage can be controlled. Main use case is manage
* memory usage.
*
* @org.apache.xbean.XBean
*
* @version $Revision: 1.3 $
*/
public class MemoryUsage extends Usage{
public class MemoryUsage extends Usage<MemoryUsage> {
private MemoryUsage parent;
private long usage;
public MemoryUsage(){
this(null,"default");
public MemoryUsage() {
this(null, null);
}
/**
@ -43,73 +39,72 @@ public class MemoryUsage extends Usage{
*
* @param parent
*/
public MemoryUsage(MemoryUsage parent){
this(parent,"default");
public MemoryUsage(MemoryUsage parent) {
this(parent, "default");
}
public MemoryUsage(String name){
this(null,name);
public MemoryUsage(String name) {
this(null, name);
}
public MemoryUsage(MemoryUsage parent,String name){
this(parent,name,1.0f);
public MemoryUsage(MemoryUsage parent, String name) {
this(parent, name, 1.0f);
}
public MemoryUsage(MemoryUsage parent,String name,float portion){
super(parent,name,portion);
public MemoryUsage(MemoryUsage parent, String name, float portion) {
super(parent, name, portion);
}
/**
* @throws InterruptedException
*/
public void waitForSpace() throws InterruptedException{
if(parent!=null){
public void waitForSpace() throws InterruptedException {
if (parent != null) {
parent.waitForSpace();
}
synchronized(usageMutex){
for(int i=0;percentUsage>=100;i++){
synchronized (usageMutex) {
for (int i = 0; percentUsage >= 100; i++) {
usageMutex.wait();
}
}
}
/**
* @param timeout
* @param timeout
* @throws InterruptedException
*
* @return true if space
*/
public boolean waitForSpace(long timeout) throws InterruptedException{
if(parent!=null){
if(!parent.waitForSpace(timeout)){
public boolean waitForSpace(long timeout) throws InterruptedException {
if (parent != null) {
if (!parent.waitForSpace(timeout)) {
return false;
}
}
synchronized(usageMutex){
if(percentUsage>=100){
synchronized (usageMutex) {
if (percentUsage >= 100) {
usageMutex.wait(timeout);
}
return percentUsage<100;
return percentUsage < 100;
}
}
public boolean isFull(){
if(parent!=null&&parent.isFull()){
public boolean isFull() {
if (parent != null && parent.isFull()) {
return true;
}
synchronized(usageMutex){
return percentUsage>=100;
synchronized (usageMutex) {
return percentUsage >= 100;
}
}
/**
* Tries to increase the usage by value amount but blocks if this object is
* currently full.
* @param value
*
* @param value
* @throws InterruptedException
*/
public void enqueueUsage(long value) throws InterruptedException{
public void enqueueUsage(long value) throws InterruptedException {
waitForSpace();
increaseUsage(value);
}
@ -119,17 +114,17 @@ public class MemoryUsage extends Usage{
*
* @param value
*/
public void increaseUsage(long value){
if(value==0){
public void increaseUsage(long value) {
if (value == 0) {
return;
}
if(parent!=null){
parent.increaseUsage(value);
if (parent != null) {
((MemoryUsage)parent).increaseUsage(value);
}
int percentUsage;
synchronized(usageMutex){
usage+=value;
percentUsage=caclPercentUsage();
synchronized (usageMutex) {
usage += value;
percentUsage = caclPercentUsage();
}
setPercentUsage(percentUsage);
}
@ -139,22 +134,30 @@ public class MemoryUsage extends Usage{
*
* @param value
*/
public void decreaseUsage(long value){
if(value==0){
public void decreaseUsage(long value) {
if (value == 0) {
return;
}
if(parent!=null){
if (parent != null) {
parent.decreaseUsage(value);
}
int percentUsage;
synchronized(usageMutex){
usage-=value;
percentUsage=caclPercentUsage();
synchronized (usageMutex) {
usage -= value;
percentUsage = caclPercentUsage();
}
setPercentUsage(percentUsage);
}
protected long retrieveUsage(){
protected long retrieveUsage() {
return usage;
}
public long getUsage() {
return usage;
}
public void setUsage(long usage) {
this.usage = usage;
}
}

43
activemq-core/src/main/java/org/apache/activemq/usage/StoreUsage.java
View File

@ -20,29 +20,42 @@ import org.apache.activemq.store.PersistenceAdapter;
/**
* Used to keep track of how much of something is being used so that a
* productive working set usage can be controlled.
*
* Main use case is manage memory usage.
* productive working set usage can be controlled. Main use case is manage
* memory usage.
*
* @org.apache.xbean.XBean
*
* @version $Revision: 1.3 $
*/
public class StoreUsage extends Usage{
public class StoreUsage extends Usage<StoreUsage> {
final private PersistenceAdapter store;
private PersistenceAdapter store;
public StoreUsage(String name,PersistenceAdapter store){
super(null,name,1.0f);
this.store=store;
}
public StoreUsage(StoreUsage parent,String name){
super(parent,name,1.0f);
this.store=parent.store;
public StoreUsage() {
super(null, null, 1.0f);
}
protected long retrieveUsage(){
public StoreUsage(String name, PersistenceAdapter store) {
super(null, name, 1.0f);
this.store = store;
}
public StoreUsage(StoreUsage parent, String name) {
super(parent, name, 1.0f);
this.store = parent.store;
}
protected long retrieveUsage() {
if (store == null)
return 0;
return store.size();
}
public PersistenceAdapter getStore() {
return store;
}
public void setStore(PersistenceAdapter store) {
this.store = store;
onLimitChange();
}
}

146
activemq-core/src/main/java/org/apache/activemq/usage/SystemUsage.java
View File

@ -22,102 +22,96 @@ import org.apache.activemq.Service;
import org.apache.activemq.kaha.Store;
import org.apache.activemq.store.PersistenceAdapter;
/**
* Holder for Usage instances for memory, store and temp files
*
* Main use case is manage memory usage.
* Holder for Usage instances for memory, store and temp files Main use case is
* manage memory usage.
*
* @org.apache.xbean.XBean
*
* @version $Revision: 1.3 $
*/
public class SystemUsage implements Service{
public class SystemUsage implements Service {
private SystemUsage parent;
private String name;
private MemoryUsage memoryUsage;
private StoreUsage storeUsage;
private TempUsage tempUsage;
private final SystemUsage parent;
private final String name;
private final MemoryUsage memoryUsage;
private final StoreUsage storeUsage;
private final TempUsage tempDiskUsage;
/**
* True if someone called setSendFailIfNoSpace() on this particular usage
* manager
*/
private boolean sendFailIfNoSpaceExplicitySet;
private boolean sendFailIfNoSpace;
private List<SystemUsage> children=new CopyOnWriteArrayList<SystemUsage>();
private List<SystemUsage> children = new CopyOnWriteArrayList<SystemUsage>();
public SystemUsage(){
this.parent=null;
this.name="default";
this.memoryUsage=new MemoryUsage(name+":memory");
this.storeUsage=null;
this.tempDiskUsage=null;
public SystemUsage() {
this("default", null, null);
}
public SystemUsage(String name,PersistenceAdapter adapter,Store tempStore){
this.parent=null;
this.name=name;
this.memoryUsage=new MemoryUsage(name+":memory");
this.storeUsage=new StoreUsage(name+":store",adapter);
this.tempDiskUsage=new TempUsage(name+":temp",tempStore);
public SystemUsage(String name, PersistenceAdapter adapter, Store tempStore) {
this.parent = null;
this.name = name;
this.memoryUsage = new MemoryUsage(name + ":memory");
this.storeUsage = new StoreUsage(name + ":store", adapter);
this.tempUsage = new TempUsage(name + ":temp", tempStore);
}
public SystemUsage(SystemUsage parent,String name){
this.parent=parent;
this.name=name;
this.memoryUsage=new MemoryUsage(parent.memoryUsage,name+":memory");
this.storeUsage=new StoreUsage(parent.storeUsage,name+":store");
this.tempDiskUsage=new TempUsage(parent!=null?parent.tempDiskUsage:null,name+":temp");
public SystemUsage(SystemUsage parent, String name) {
this.parent = parent;
this.name = name;
this.memoryUsage = new MemoryUsage(parent.memoryUsage, name + ":memory");
this.storeUsage = new StoreUsage(parent.storeUsage, name + ":store");
this.tempUsage = new TempUsage(parent.tempUsage, name + ":temp");
}
public String getName(){
public String getName() {
return name;
}
/**
* @return the memoryUsage
*/
public MemoryUsage getMemoryUsage(){
public MemoryUsage getMemoryUsage() {
return this.memoryUsage;
}
/**
* @return the storeUsage
*/
public StoreUsage getStoreUsage(){
public StoreUsage getStoreUsage() {
return this.storeUsage;
}
/**
* @return the tempDiskUsage
*/
public TempUsage getTempDiskUsage(){
return this.tempDiskUsage;
public TempUsage getTempUsage() {
return this.tempUsage;
}
public String toString(){
return "UsageManager("+getName()+")";
public String toString() {
return "UsageManager(" + getName() + ")";
}
public void start(){
if(parent!=null){
public void start() {
if (parent != null) {
parent.addChild(this);
}
this.memoryUsage.start();
this.storeUsage.start();
this.tempDiskUsage.start();
this.tempUsage.start();
}
public void stop(){
if(parent!=null){
public void stop() {
if (parent != null) {
parent.removeChild(this);
}
this.memoryUsage.stop();
this.storeUsage.stop();
this.tempDiskUsage.stop();
this.tempUsage.stop();
}
/**
* Sets whether or not a send() should fail if there is no space free. The
* default value is false which means to block the send() method until space
@ -136,11 +130,71 @@ public class SystemUsage implements Service{
}
}
private void addChild(SystemUsage child){
private void addChild(SystemUsage child) {
children.add(child);
}
private void removeChild(SystemUsage child){
private void removeChild(SystemUsage child) {
children.remove(child);
}
public SystemUsage getParent() {
return parent;
}
public void setParent(SystemUsage parent) {
this.parent = parent;
}
public boolean isSendFailIfNoSpaceExplicitySet() {
return sendFailIfNoSpaceExplicitySet;
}
public void setSendFailIfNoSpaceExplicitySet(boolean sendFailIfNoSpaceExplicitySet) {
this.sendFailIfNoSpaceExplicitySet = sendFailIfNoSpaceExplicitySet;
}
public void setName(String name) {
this.name = name;
this.memoryUsage.setName(name + ":memory");
this.storeUsage.setName(name + ":store");
this.tempUsage.setName(name + ":temp");
}
public void setMemoryUsage(MemoryUsage memoryUsage) {
if (memoryUsage.getName() == null) {
memoryUsage.setName(this.memoryUsage.getName());
}
if (parent != null) {
memoryUsage.setParent(parent.memoryUsage);
}
this.memoryUsage = memoryUsage;
}
public void setStoreUsage(StoreUsage storeUsage) {
if (storeUsage.getStore() == null) {
storeUsage.setStore(this.storeUsage.getStore());
}
if (storeUsage.getName() == null) {
storeUsage.setName(this.storeUsage.getName());
}
if (parent != null) {
storeUsage.setParent(parent.storeUsage);
}
this.storeUsage = storeUsage;
}
public void setTempUsage(TempUsage tempDiskUsage) {
if (tempDiskUsage.getStore() == null) {
tempDiskUsage.setStore(this.tempUsage.getStore());
}
if (tempDiskUsage.getName() == null) {
tempDiskUsage.setName(this.tempUsage.getName());
}
if (parent != null) {
tempDiskUsage.setParent(parent.tempUsage);
}
this.tempUsage = tempDiskUsage;
}
}

47
activemq-core/src/main/java/org/apache/activemq/usage/TempUsage.java
View File

@ -20,30 +20,43 @@ import org.apache.activemq.kaha.Store;
/**
* Used to keep track of how much of something is being used so that a
* productive working set usage can be controlled.
*
* Main use case is manage memory usage.
* productive working set usage can be controlled. Main use case is manage
* memory usage.
*
* @org.apache.xbean.XBean
*
* @version $Revision: 1.3 $
*/
public class TempUsage extends Usage{
public class TempUsage extends Usage<TempUsage> {
final private Store store;
private Store store;
public TempUsage(String name,Store store){
super(null,name,1.0f);
this.store=store;
}
public TempUsage(TempUsage parent,String name){
super(parent,name,1.0f);
this.store=parent.store;
public TempUsage() {
super(null, null, 1.0f);
}
protected long retrieveUsage(){
public TempUsage(String name, Store store) {
super(null, name, 1.0f);
this.store = store;
}
public TempUsage(TempUsage parent, String name) {
super(parent, name, 1.0f);
this.store = parent.store;
}
protected long retrieveUsage() {
if (store == null) {
return 0;
}
return store.size();
}
}
public Store getStore() {
return store;
}
public void setStore(Store store) {
this.store = store;
onLimitChange();
}
}

260
activemq-core/src/main/java/org/apache/activemq/usage/Usage.java
View File

@ -28,38 +28,36 @@ import org.apache.commons.logging.LogFactory;
/**
* Used to keep track of how much of something is being used so that a
* productive working set usage can be controlled.
*
* Main use case is manage memory usage.
* productive working set usage can be controlled. Main use case is manage
* memory usage.
*
* @org.apache.xbean.XBean
*
* @version $Revision: 1.3 $
*/
public abstract class Usage implements Service{
public abstract class Usage<T extends Usage> implements Service {
private static final Log LOG=LogFactory.getLog(Usage.class);
protected final Object usageMutex=new Object();
private static final Log LOG = LogFactory.getLog(Usage.class);
protected final Object usageMutex = new Object();
protected int percentUsage;
private final Usage parent;
protected T parent;
private UsageCapacity limiter = new DefaultUsageCapacity();
private int percentUsageMinDelta=1;
private final List<UsageListener> listeners=new CopyOnWriteArrayList<UsageListener>();
private final boolean debug=LOG.isDebugEnabled();
private String name="";
private float usagePortion=1.0f;
private List<Usage> children=new CopyOnWriteArrayList<Usage>();
private final List<Runnable> callbacks=new LinkedList<Runnable>();
private int percentUsageMinDelta = 1;
private final List<UsageListener> listeners = new CopyOnWriteArrayList<UsageListener>();
private final boolean debug = LOG.isDebugEnabled();
private String name;
private float usagePortion = 1.0f;
private List<T> children = new CopyOnWriteArrayList<T>();
private final List<Runnable> callbacks = new LinkedList<Runnable>();
private int pollingTime = 100;
public Usage(Usage parent,String name,float portion){
this.parent=parent;
this.usagePortion=portion;
if(parent!=null){
this.limiter.setLimit((long)(parent.getLimit()*portion));
this.name=parent.name+":";
public Usage(T parent, String name, float portion) {
this.parent = parent;
this.usagePortion = portion;
if (parent != null) {
this.limiter.setLimit((long)(parent.getLimit() * portion));
name = parent.name + ":" + name;
}
this.name+=name;
this.name = name;
}
protected abstract long retrieveUsage();
@ -67,63 +65,60 @@ public abstract class Usage implements Service{
/**
* @throws InterruptedException
*/
public void waitForSpace() throws InterruptedException{
public void waitForSpace() throws InterruptedException {
waitForSpace(0);
}
/**
* @param timeout
* @param timeout
* @throws InterruptedException
*
* @return true if space
*/
public boolean waitForSpace(long timeout) throws InterruptedException{
if(parent!=null){
if(!parent.waitForSpace(timeout)){
public boolean waitForSpace(long timeout) throws InterruptedException {
if (parent != null) {
if (!parent.waitForSpace(timeout)) {
return false;
}
}
synchronized(usageMutex){
synchronized (usageMutex) {
caclPercentUsage();
if(percentUsage>=100){
long deadline=timeout>0?System.currentTimeMillis()+timeout:Long.MAX_VALUE;
long timeleft=deadline;
while(timeleft>0){
if (percentUsage >= 100) {
long deadline = timeout > 0 ? System.currentTimeMillis() + timeout : Long.MAX_VALUE;
long timeleft = deadline;
while (timeleft > 0) {
caclPercentUsage();
if(percentUsage>=100){
if (percentUsage >= 100) {
usageMutex.wait(pollingTime);
timeleft=deadline-System.currentTimeMillis();
}else{
timeleft = deadline - System.currentTimeMillis();
} else {
break;
}
}
}
return percentUsage<100;
return percentUsage < 100;
}
}
public boolean isFull(){
if(parent!=null&&parent.isFull()){
public boolean isFull() {
if (parent != null && parent.isFull()) {
return true;
}
synchronized(usageMutex){
synchronized (usageMutex) {
caclPercentUsage();
return percentUsage>=100;
return percentUsage >= 100;
}
}
public void addUsageListener(UsageListener listener){
public void addUsageListener(UsageListener listener) {
listeners.add(listener);
}
public void removeUsageListener(UsageListener listener){
public void removeUsageListener(UsageListener listener) {
listeners.remove(listener);
}
public long getLimit(){
synchronized(usageMutex){
public long getLimit() {
synchronized (usageMutex) {
return limiter.getLimit();
}
}
@ -131,54 +126,52 @@ public abstract class Usage implements Service{
/**
* Sets the memory limit in bytes. Setting the limit in bytes will set the
* usagePortion to 0 since the UsageManager is not going to be portion based
* off the parent.
*
* When set using XBean, you can use values such as: "20 mb", "1024 kb", or
* "1 gb"
* off the parent. When set using XBean, you can use values such as: "20
* mb", "1024 kb", or "1 gb"
*
* @org.apache.xbean.Property propertyEditor="org.apache.activemq.util.MemoryPropertyEditor"
*/
public void setLimit(long limit){
if(percentUsageMinDelta<0){
public void setLimit(long limit) {
if (percentUsageMinDelta < 0) {
throw new IllegalArgumentException("percentUsageMinDelta must be greater or equal to 0");
}
synchronized(usageMutex){
synchronized (usageMutex) {
this.limiter.setLimit(limit);
this.usagePortion=0;
this.usagePortion = 0;
}
onLimitChange();
}
private void onLimitChange(){
protected void onLimitChange() {
// We may need to calculate the limit
if(usagePortion>0&&parent!=null){
synchronized(usageMutex){
this.limiter.setLimit((long)(parent.getLimit()*usagePortion));
if (usagePortion > 0 && parent != null) {
synchronized (usageMutex) {
this.limiter.setLimit((long)(parent.getLimit() * usagePortion));
}
}
// Reset the percent currently being used.
int percentUsage;
synchronized(usageMutex){
percentUsage=caclPercentUsage();
synchronized (usageMutex) {
percentUsage = caclPercentUsage();
}
setPercentUsage(percentUsage);
// Let the children know that the limit has changed. They may need to
// set
// their limits based on ours.
for(Usage child:children){
for (T child : children) {
child.onLimitChange();
}
}
public float getUsagePortion(){
synchronized(usageMutex){
public float getUsagePortion() {
synchronized (usageMutex) {
return usagePortion;
}
}
public void setUsagePortion(float usagePortion){
synchronized(usageMutex){
this.usagePortion=usagePortion;
public void setUsagePortion(float usagePortion) {
synchronized (usageMutex) {
this.usagePortion = usagePortion;
}
onLimitChange();
}
@ -187,14 +180,14 @@ public abstract class Usage implements Service{
* Sets the minimum number of percentage points the usage has to change
* before a UsageListener event is fired by the manager.
*/
public int getPercentUsage(){
synchronized(usageMutex){
public int getPercentUsage() {
synchronized (usageMutex) {
return percentUsage;
}
}
public int getPercentUsageMinDelta(){
synchronized(usageMutex){
public int getPercentUsageMinDelta() {
synchronized (usageMutex) {
return percentUsageMinDelta;
}
}
@ -205,89 +198,90 @@ public abstract class Usage implements Service{
*
* @param percentUsageMinDelta
*/
public void setPercentUsageMinDelta(int percentUsageMinDelta){
if(percentUsageMinDelta<1){
public void setPercentUsageMinDelta(int percentUsageMinDelta) {
if (percentUsageMinDelta < 1) {
throw new IllegalArgumentException("percentUsageMinDelta must be greater than 0");
}
int percentUsage;
synchronized(usageMutex){
this.percentUsageMinDelta=percentUsageMinDelta;
percentUsage=caclPercentUsage();
synchronized (usageMutex) {
this.percentUsageMinDelta = percentUsageMinDelta;
percentUsage = caclPercentUsage();
}
setPercentUsage(percentUsage);
}
public long getUsage(){
synchronized(usageMutex){
public long getUsage() {
synchronized (usageMutex) {
return retrieveUsage();
}
}
protected void setPercentUsage(int value){
synchronized(usageMutex){
int oldValue=percentUsage;
percentUsage=value;
if(oldValue!=value){
fireEvent(oldValue,value);
protected void setPercentUsage(int value) {
synchronized (usageMutex) {
int oldValue = percentUsage;
percentUsage = value;
if (oldValue != value) {
fireEvent(oldValue, value);
}
}
}
protected int caclPercentUsage(){
if(limiter.getLimit()==0){
protected int caclPercentUsage() {
if (limiter.getLimit() == 0) {
return 0;
}
return (int)((((retrieveUsage()*100)/limiter.getLimit())/percentUsageMinDelta)*percentUsageMinDelta);
return (int)((((retrieveUsage() * 100) / limiter.getLimit()) / percentUsageMinDelta) * percentUsageMinDelta);
}
private void fireEvent(int oldPercentUsage,int newPercentUsage){
if(debug){
LOG.debug("Memory usage change. from: "+oldPercentUsage+", to: "+newPercentUsage);
private void fireEvent(int oldPercentUsage, int newPercentUsage) {
if (debug) {
LOG.debug("Memory usage change. from: " + oldPercentUsage + ", to: " + newPercentUsage);
}
// Switching from being full to not being full..
if(oldPercentUsage>=100&&newPercentUsage<100){
synchronized(usageMutex){
if (oldPercentUsage >= 100 && newPercentUsage < 100) {
synchronized (usageMutex) {
usageMutex.notifyAll();
for(Iterator<Runnable> iter=new ArrayList<Runnable>(callbacks).iterator();iter.hasNext();){
Runnable callback=iter.next();
for (Iterator<Runnable> iter = new ArrayList<Runnable>(callbacks).iterator(); iter.hasNext();) {
Runnable callback = iter.next();
callback.run();
}
callbacks.clear();
}
}
// Let the listeners know
for(Iterator<UsageListener> iter=listeners.iterator();iter.hasNext();){
UsageListener l=iter.next();
l.onUsageChanged(this,oldPercentUsage,newPercentUsage);
for (Iterator<UsageListener> iter = listeners.iterator(); iter.hasNext();) {
UsageListener l = iter.next();
l.onUsageChanged(this, oldPercentUsage, newPercentUsage);
}
}
public String getName(){
public String getName() {
return name;
}
public String toString(){
return "Usage("+getName()+") percentUsage="+percentUsage+"%, usage="+retrieveUsage()+" limit="+limiter.getLimit()
+" percentUsageMinDelta="+percentUsageMinDelta+"%";
public String toString() {
return "Usage(" + getName() + ") percentUsage=" + percentUsage + "%, usage=" + retrieveUsage() + " limit=" + limiter.getLimit() + " percentUsageMinDelta=" + percentUsageMinDelta + "%";
}
public void start(){
if(parent!=null){
@SuppressWarnings("unchecked")
public void start() {
if (parent != null) {
parent.addChild(this);
}
}
public void stop(){
if(parent!=null){
@SuppressWarnings("unchecked")
public void stop() {
if (parent != null) {
parent.removeChild(this);
}
}
private void addChild(Usage child){
private void addChild(T child) {
children.add(child);
}
private void removeChild(Usage child){
private void removeChild(T child) {
children.remove(child);
}
@ -296,63 +290,71 @@ public abstract class Usage implements Service{
* @return true if the UsageManager was full. The callback will only be
* called if this method returns true.
*/
public boolean notifyCallbackWhenNotFull(final Runnable callback){
if(parent!=null){
Runnable r=new Runnable(){
public boolean notifyCallbackWhenNotFull(final Runnable callback) {
if (parent != null) {
Runnable r = new Runnable() {
public void run(){
synchronized(usageMutex){
if(percentUsage>=100){
public void run() {
synchronized (usageMutex) {
if (percentUsage >= 100) {
callbacks.add(callback);
}else{
} else {
callback.run();
}
}
}
};
if(parent.notifyCallbackWhenNotFull(r)){
if (parent.notifyCallbackWhenNotFull(r)) {
return true;
}
}
synchronized(usageMutex){
if(percentUsage>=100){
synchronized (usageMutex) {
if (percentUsage >= 100) {
callbacks.add(callback);
return true;
}else{
} else {
return false;
}
}
}
/**
* @return the limiter
*/
public UsageCapacity getLimiter(){
public UsageCapacity getLimiter() {
return this.limiter;
}
/**
* @param limiter the limiter to set
*/
public void setLimiter(UsageCapacity limiter){
this.limiter=limiter;
public void setLimiter(UsageCapacity limiter) {
this.limiter = limiter;
}
/**
* @return the pollingTime
*/
public int getPollingTime(){
public int getPollingTime() {
return this.pollingTime;
}
/**
* @param pollingTime the pollingTime to set
*/
public void setPollingTime(int pollingTime){
this.pollingTime=pollingTime;
public void setPollingTime(int pollingTime) {
this.pollingTime = pollingTime;
}
public void setName(String name) {
this.name = name;
}
public T getParent() {
return parent;
}
public void setParent(T parent) {
this.parent = parent;
}
}

2
activemq-core/src/test/java/org/apache/activemq/AMQDeadlockTest3.java
View File

@ -240,7 +240,7 @@ public class AMQDeadlockTest3 extends TestCase {
final SystemUsage memoryManager = new SystemUsage();
memoryManager.getMemoryUsage().setLimit(5000000);
brokerService.setUsageManager(memoryManager);
brokerService.setSystemUsage(memoryManager);
final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();

2
activemq-core/src/test/java/org/apache/activemq/broker/MessageExpirationTest.java
View File

@ -91,7 +91,7 @@ public class MessageExpirationTest extends BrokerTestSupport {
// Reduce the limit so that only 1 message can flow through the broker
// at a time.
broker.getUsageManager().getMemoryUsage().setLimit(1);
broker.getSystemUsage().getMemoryUsage().setLimit(1);
final Message m1 = createMessage(producerInfo, destination, deliveryMode);
final Message m2 = createMessage(producerInfo, destination, deliveryMode, 1000);

13
activemq-core/src/test/java/org/apache/activemq/config/ConfigTest.java
View File

@ -233,13 +233,16 @@ public class ConfigTest extends TestCase {
// Check usage manager
// System.out.print("Checking memory manager configurations... ");
SystemUsage memMgr = broker.getUsageManager();
assertTrue("Should have a memory manager", memMgr != null);
assertEquals("UsageManager Config Error (limit)", 200000, memMgr.getMemoryUsage().getLimit());
assertEquals("UsageManager Config Error (percentUsageMinDelta)", 20, memMgr.getMemoryUsage().getPercentUsageMinDelta());
SystemUsage systemUsage = broker.getSystemUsage();
assertTrue("Should have a SystemUsage", systemUsage != null);
assertEquals("SystemUsage Config Error (MemoryUsage.limit)", 1024 * 1024 * 10, systemUsage.getMemoryUsage().getLimit());
assertEquals("SystemUsage Config Error (MemoryUsage.percentUsageMinDelta)", 20, systemUsage.getMemoryUsage().getPercentUsageMinDelta());
assertEquals("SystemUsage Config Error (TempUsage.limit)", 1024 * 1024 * 100, systemUsage.getTempUsage().getLimit());
assertEquals("SystemUsage Config Error (StoreUsage.limit)", 1024 * 1024 * 1024, systemUsage.getStoreUsage().getLimit());
assertEquals("SystemUsage Config Error (StoreUsage.name)", "foo", systemUsage.getStoreUsage().getName());
LOG.info("Success");
LOG.info("Success");
} finally {
if (broker != null) {
broker.stop();

2
activemq-core/src/test/java/org/apache/activemq/usecases/AMQDeadlockTestW4Brokers.java
View File

@ -175,7 +175,7 @@ public class AMQDeadlockTestW4Brokers extends TestCase {
final SystemUsage memoryManager = new SystemUsage();
memoryManager.getMemoryUsage().setLimit(100000000);
brokerService.setUsageManager(memoryManager);
brokerService.setSystemUsage(memoryManager);
final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();

2
activemq-core/src/test/java/org/apache/activemq/usecases/AMQFailoverIssue.java
View File

@ -108,7 +108,7 @@ public class AMQFailoverIssue extends TestCase {
brokerService.setUseJmx(true);
final SystemUsage memoryManager = new SystemUsage();
memoryManager.getMemoryUsage().setLimit(5000000);
brokerService.setUsageManager(memoryManager);
brokerService.setSystemUsage(memoryManager);
final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();
final PolicyEntry entry = new PolicyEntry();
entry.setQueue(">");

2
activemq-core/src/test/java/org/apache/activemq/usecases/AMQStackOverFlowTest.java
View File

@ -119,7 +119,7 @@ public class AMQStackOverFlowTest extends TestCase {
final SystemUsage memoryManager = new SystemUsage();
memoryManager.getMemoryUsage().setLimit(10);
brokerService.setUsageManager(memoryManager);
brokerService.setSystemUsage(memoryManager);
final List<PolicyEntry> policyEntries = new ArrayList<PolicyEntry>();

16
activemq-core/src/test/resources/org/apache/activemq/config/example.xml
View File

@ -76,9 +76,19 @@
</amq:policyMap>
</amq:destinationPolicy>
<amq:memoryManager>
<amq:usageManager limit="200000" percentUsageMinDelta="20"/>
</amq:memoryManager>
<amq:systemUsage>
<amq:systemUsage>
<amq:memoryUsage>
<amq:memoryUsage limit="10 mb" percentUsageMinDelta="20"/>
</amq:memoryUsage>
<amq:tempUsage>
<amq:tempUsage limit="100 mb"/>
</amq:tempUsage>
<amq:storeUsage>
<amq:storeUsage limit="1 gb" name="foo"/>
</amq:storeUsage>
</amq:systemUsage>
</amq:systemUsage>
<amq:persistenceAdapter>
<amq:memoryPersistenceAdapter init-method="createTransactionStore"/>