| package org.apache.jcs.engine; |
| |
| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| import java.io.IOException; |
| import java.io.Serializable; |
| import java.util.ArrayList; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.jcs.engine.behavior.ICacheElement; |
| import org.apache.jcs.engine.behavior.ICacheEventQueue; |
| import org.apache.jcs.engine.behavior.ICacheListener; |
| import org.apache.jcs.engine.stats.StatElement; |
| import org.apache.jcs.engine.stats.Stats; |
| import org.apache.jcs.engine.stats.behavior.IStatElement; |
| import org.apache.jcs.engine.stats.behavior.IStats; |
| import org.apache.jcs.utils.threadpool.ThreadPool; |
| import org.apache.jcs.utils.threadpool.ThreadPoolManager; |
| |
| import EDU.oswego.cs.dl.util.concurrent.BoundedBuffer; |
| |
| /** |
| * An event queue is used to propagate ordered cache events to one and only one target listener. |
| * <p> |
| * This is a modified version of the experimental version. It uses a PooledExecutor and a |
| * BoundedBuffer to queue up events and execute them as threads become available. |
| * <p> |
| * The PooledExecutor is static, because presumably these processes will be IO bound, so throwing |
| * more than a few threads at them will serve no purpose other than to saturate the IO interface. In |
| * light of this, having one thread per region seems unnecessary. This may prove to be false. |
| */ |
| public class PooledCacheEventQueue |
| implements ICacheEventQueue |
| { |
| /** The type of event queue */ |
| private static final String queueType = POOLED_QUEUE_TYPE; |
| |
| /** The logger */ |
| private static final Log log = LogFactory.getLog( PooledCacheEventQueue.class ); |
| |
| /** time to wait for an event before snuffing the background thread |
| if the queue is empty. make configurable later */ |
| private int waitToDieMillis = 10000; |
| |
| /** The listener to process events */ |
| private ICacheListener listener; |
| |
| /** The listener id */ |
| private long listenerId; |
| |
| /** The name of the cache */ |
| private String cacheName; |
| |
| /** Max failures before self destruction */ |
| private int maxFailure; |
| |
| /** in milliseconds */ |
| private int waitBeforeRetry; |
| |
| /** Has the pool been destroyed */ |
| private boolean destroyed = true; |
| |
| /** Is it working */ |
| private boolean working = true; |
| |
| /** The Thread Pool to execute events with. */ |
| private ThreadPool pool = null; |
| |
| /** |
| * Constructor for the CacheEventQueue object |
| * <p> |
| * @param listener |
| * @param listenerId |
| * @param cacheName |
| * @param maxFailure |
| * @param waitBeforeRetry |
| * @param threadPoolName |
| */ |
| public PooledCacheEventQueue( ICacheListener listener, long listenerId, String cacheName, int maxFailure, |
| int waitBeforeRetry, String threadPoolName ) |
| { |
| initialize( listener, listenerId, cacheName, maxFailure, waitBeforeRetry, threadPoolName ); |
| } |
| |
| /** |
| * Initializes the queue. |
| * <p> |
| * @param listener |
| * @param listenerId |
| * @param cacheName |
| * @param maxFailure |
| * @param waitBeforeRetry |
| * @param threadPoolName |
| */ |
| public void initialize( ICacheListener listener, long listenerId, String cacheName, int maxFailure, |
| int waitBeforeRetry, String threadPoolName ) |
| { |
| if ( listener == null ) |
| { |
| throw new IllegalArgumentException( "listener must not be null" ); |
| } |
| |
| this.listener = listener; |
| this.listenerId = listenerId; |
| this.cacheName = cacheName; |
| this.maxFailure = maxFailure <= 0 ? 3 : maxFailure; |
| this.waitBeforeRetry = waitBeforeRetry <= 0 ? 500 : waitBeforeRetry; |
| |
| // this will share the same pool with other event queues by default. |
| if ( threadPoolName == null ) |
| { |
| threadPoolName = "cache_event_queue"; |
| } |
| pool = ThreadPoolManager.getInstance().getPool( threadPoolName ); |
| |
| if ( log.isDebugEnabled() ) |
| { |
| log.debug( "Initialized: " + this ); |
| } |
| } |
| |
| /** |
| * @return the queue type |
| */ |
| public String getQueueType() |
| { |
| return queueType; |
| } |
| |
| /** |
| * Event Q is empty. |
| */ |
| public synchronized void stopProcessing() |
| { |
| destroyed = true; |
| } |
| |
| /** |
| * Returns the time to wait for events before killing the background thread. |
| * <p> |
| * @return the time to wait before shutting down in ms. |
| */ |
| public int getWaitToDieMillis() |
| { |
| return waitToDieMillis; |
| } |
| |
| /** |
| * Sets the time to wait for events before killing the background thread. |
| * <p> |
| * @param wtdm |
| */ |
| public void setWaitToDieMillis( int wtdm ) |
| { |
| waitToDieMillis = wtdm; |
| } |
| |
| /** |
| * @return String info. |
| */ |
| public String toString() |
| { |
| return "CacheEventQueue [listenerId=" + listenerId + ", cacheName=" + cacheName + "]"; |
| } |
| |
| /** |
| * @return true if not destroyed. |
| */ |
| public boolean isAlive() |
| { |
| return ( !destroyed ); |
| } |
| |
| /** |
| * @param aState |
| */ |
| public void setAlive( boolean aState ) |
| { |
| destroyed = !aState; |
| } |
| |
| /** |
| * @return The listenerId value |
| */ |
| public long getListenerId() |
| { |
| return listenerId; |
| } |
| |
| /** |
| * Destroy the queue. Interrupt all threads. |
| */ |
| public synchronized void destroy() |
| { |
| if ( !destroyed ) |
| { |
| destroyed = true; |
| // TODO decide whether to shutdown or interrupt |
| // pool.getPool().shutdownNow(); |
| pool.getPool().interruptAll(); |
| if ( log.isInfoEnabled() ) |
| { |
| log.info( "Cache event queue destroyed: " + this ); |
| } |
| } |
| } |
| |
| /** |
| * Constructs a PutEvent for the object and passes it to the event queue. |
| * <p> |
| * @param ce The feature to be added to the PutEvent attribute |
| * @exception IOException |
| */ |
| public synchronized void addPutEvent( ICacheElement ce ) |
| throws IOException |
| { |
| if ( isWorking() ) |
| { |
| put( new PutEvent( ce ) ); |
| } |
| else |
| { |
| if ( log.isWarnEnabled() ) |
| { |
| log.warn( "Not enqueuing Put Event for [" + this + "] because it's non-functional." ); |
| } |
| } |
| } |
| |
| /** |
| * @param key The feature to be added to the RemoveEvent attribute |
| * @exception IOException |
| */ |
| public synchronized void addRemoveEvent( Serializable key ) |
| throws IOException |
| { |
| if ( isWorking() ) |
| { |
| put( new RemoveEvent( key ) ); |
| } |
| else |
| { |
| if ( log.isWarnEnabled() ) |
| { |
| log.warn( "Not enqueuing Remove Event for [" + this + "] because it's non-functional." ); |
| } |
| } |
| } |
| |
| /** |
| * @exception IOException |
| */ |
| public synchronized void addRemoveAllEvent() |
| throws IOException |
| { |
| if ( isWorking() ) |
| { |
| put( new RemoveAllEvent() ); |
| } |
| else |
| { |
| if ( log.isWarnEnabled() ) |
| { |
| log.warn( "Not enqueuing RemoveAll Event for [" + this + "] because it's non-functional." ); |
| } |
| } |
| } |
| |
| /** |
| * @exception IOException |
| */ |
| public synchronized void addDisposeEvent() |
| throws IOException |
| { |
| if ( isWorking() ) |
| { |
| put( new DisposeEvent() ); |
| } |
| else |
| { |
| if ( log.isWarnEnabled() ) |
| { |
| log.warn( "Not enqueuing Dispose Event for [" + this + "] because it's non-functional." ); |
| } |
| } |
| } |
| |
| /** |
| * Adds an event to the queue. |
| * <p> |
| * @param event |
| */ |
| private void put( AbstractCacheEvent event ) |
| { |
| try |
| { |
| pool.execute( event ); |
| } |
| catch ( InterruptedException e ) |
| { |
| log.error( e ); |
| } |
| } |
| |
| /** |
| * @return Statistics info |
| */ |
| public String getStats() |
| { |
| return getStatistics().toString(); |
| } |
| |
| /** |
| * @return IStats |
| */ |
| public IStats getStatistics() |
| { |
| IStats stats = new Stats(); |
| stats.setTypeName( "Pooled Cache Event Queue" ); |
| |
| ArrayList elems = new ArrayList(); |
| |
| IStatElement se = null; |
| |
| se = new StatElement(); |
| se.setName( "Working" ); |
| se.setData( "" + this.working ); |
| elems.add( se ); |
| |
| se = new StatElement(); |
| se.setName( "Destroyed" ); |
| se.setData( "" + this.isAlive() ); |
| elems.add( se ); |
| |
| se = new StatElement(); |
| se.setName( "Empty" ); |
| se.setData( "" + this.isEmpty() ); |
| elems.add( se ); |
| |
| if ( pool.getQueue() != null ) |
| { |
| if ( pool.getQueue() instanceof BoundedBuffer ) |
| { |
| BoundedBuffer bb = (BoundedBuffer) pool.getQueue(); |
| se = new StatElement(); |
| se.setName( "Queue Size" ); |
| se.setData( "" + bb.size() ); |
| elems.add( se ); |
| |
| se = new StatElement(); |
| se.setName( "Queue Capacity" ); |
| se.setData( "" + bb.capacity() ); |
| elems.add( se ); |
| } |
| } |
| |
| se = new StatElement(); |
| se.setName( "Pool Size" ); |
| se.setData( "" + pool.getPool().getPoolSize() ); |
| elems.add( se ); |
| |
| se = new StatElement(); |
| se.setName( "Maximum Pool Size" ); |
| se.setData( "" + pool.getPool().getMaximumPoolSize() ); |
| elems.add( se ); |
| |
| // get an array and put them in the Stats object |
| IStatElement[] ses = (IStatElement[]) elems.toArray( new StatElement[elems.size()] ); |
| stats.setStatElements( ses ); |
| |
| return stats; |
| } |
| |
| // /////////////////////////// Inner classes ///////////////////////////// |
| |
| /** |
| * Retries before declaring failure. |
| * <p> |
| * @author asmuts |
| * @created January 15, 2002 |
| */ |
| private abstract class AbstractCacheEvent |
| implements Runnable |
| { |
| /** Times failed to process */ |
| int failures = 0; |
| |
| /** Has the event been processed */ |
| boolean done = false; |
| |
| /** |
| * Main processing method for the AbstractCacheEvent object. It calls the abstract doRun |
| * method that all concrete instances must implement. |
| */ |
| public void run() |
| { |
| try |
| { |
| doRun(); |
| } |
| catch ( IOException e ) |
| { |
| if ( log.isWarnEnabled() ) |
| { |
| log.warn( e ); |
| } |
| if ( ++failures >= maxFailure ) |
| { |
| if ( log.isWarnEnabled() ) |
| { |
| log.warn( "Error while running event from Queue: " + this |
| + ". Dropping Event and marking Event Queue as non-functional." ); |
| } |
| setWorking( false ); |
| setAlive( false ); |
| return; |
| } |
| if ( log.isInfoEnabled() ) |
| { |
| log.info( "Error while running event from Queue: " + this + ". Retrying..." ); |
| } |
| try |
| { |
| Thread.sleep( waitBeforeRetry ); |
| run(); |
| } |
| catch ( InterruptedException ie ) |
| { |
| if ( log.isErrorEnabled() ) |
| { |
| log.warn( "Interrupted while sleeping for retry on event " + this + "." ); |
| } |
| setWorking( false ); |
| setAlive( false ); |
| } |
| } |
| } |
| |
| /** |
| * @exception IOException |
| */ |
| protected abstract void doRun() |
| throws IOException; |
| } |
| |
| /** |
| * An event that puts an item to a ICacheListener |
| */ |
| private class PutEvent |
| extends AbstractCacheEvent |
| { |
| /** The payload */ |
| private ICacheElement ice; |
| |
| /** |
| * Constructor for the PutEvent object |
| * @param ice |
| * @exception IOException |
| */ |
| PutEvent( ICacheElement ice ) |
| throws IOException |
| { |
| this.ice = ice; |
| } |
| |
| /** |
| * Tells the ICacheListener to handle the put. |
| * <p> |
| * @exception IOException |
| */ |
| protected void doRun() |
| throws IOException |
| { |
| listener.handlePut( ice ); |
| } |
| |
| /** @return debugging info */ |
| public String toString() |
| { |
| return new StringBuffer( "PutEvent for key: " ).append( ice.getKey() ).append( " value: " ) |
| .append( ice.getVal() ).toString(); |
| } |
| } |
| |
| /** |
| * An event that knows how to call remove on an ICacheListener |
| */ |
| private class RemoveEvent |
| extends AbstractCacheEvent |
| { |
| /** The payload, the key to remove */ |
| private Serializable key; |
| |
| /** |
| * Constructor for the RemoveEvent object |
| * @param key |
| * @exception IOException |
| */ |
| RemoveEvent( Serializable key ) |
| throws IOException |
| { |
| this.key = key; |
| } |
| |
| /** |
| * Calls remove on the listener. |
| * <p> |
| * @exception IOException |
| */ |
| protected void doRun() |
| throws IOException |
| { |
| listener.handleRemove( cacheName, key ); |
| } |
| |
| /** @return debugging info */ |
| public String toString() |
| { |
| return new StringBuffer( "RemoveEvent for " ).append( key ).toString(); |
| } |
| } |
| |
| /** |
| * An event that knows how to call remove all on an ICacheListener |
| */ |
| private class RemoveAllEvent |
| extends AbstractCacheEvent |
| { |
| /** |
| * Call removeAll on the listener. |
| * <p> |
| * @exception IOException |
| */ |
| protected void doRun() |
| throws IOException |
| { |
| listener.handleRemoveAll( cacheName ); |
| } |
| |
| /** @return debugging info */ |
| public String toString() |
| { |
| return "RemoveAllEvent"; |
| } |
| } |
| |
| /** |
| * The Event put into the queue for dispose requests. |
| */ |
| private class DisposeEvent |
| extends AbstractCacheEvent |
| { |
| /** |
| * Called when gets to the end of the queue |
| * <p> |
| * @exception IOException |
| */ |
| protected void doRun() |
| throws IOException |
| { |
| listener.handleDispose( cacheName ); |
| } |
| |
| /** @return debugging info */ |
| public String toString() |
| { |
| return "DisposeEvent"; |
| } |
| } |
| |
| /** |
| * @return whether or not the queue is functional |
| */ |
| public boolean isWorking() |
| { |
| return working; |
| } |
| |
| /** |
| * @param isWorkingArg whether the queue is functional |
| */ |
| public void setWorking( boolean isWorkingArg ) |
| { |
| working = isWorkingArg; |
| } |
| |
| /** |
| * If the Queue is using a bounded channel we can determine the size. If it is zero or we can't |
| * determine the size, we return true. |
| * <p> |
| * @return whether or not there are items in the queue |
| */ |
| public boolean isEmpty() |
| { |
| if ( pool.getQueue() == null ) |
| { |
| return pool.getQueue().peek() == null; |
| } |
| else |
| { |
| if ( pool.getQueue() instanceof BoundedBuffer ) |
| { |
| BoundedBuffer bb = (BoundedBuffer) pool.getQueue(); |
| return bb.size() == 0; |
| } |
| else |
| { |
| return true; |
| } |
| } |
| } |
| |
| /** |
| * Returns the number of elements in the queue. If the queue cannot determine the size |
| * accurately it will return 1. |
| * <p> |
| * @return number of items in the queue. |
| */ |
| public int size() |
| { |
| if ( pool.getQueue() == null ) |
| { |
| return pool.getQueue().peek() == null ? 0 : 1; |
| } |
| else |
| { |
| if ( pool.getQueue() instanceof BoundedBuffer ) |
| { |
| BoundedBuffer bb = (BoundedBuffer) pool.getQueue(); |
| return bb.size(); |
| } |
| else |
| { |
| return 1; |
| } |
| } |
| } |
| } |